はじめに
みなさん、CloudWatch Logsって見にくくないですか？
例えば、Lambdaを実行してそのログを確認するときに、ロググループの中にログストリームがいくつも出力され、、、
どの時間に実行したのかを覚えておいて、その時間帯のログストリームを開き、、、
ERRORがあれば修正し、、、
毎回これを繰り返す。CloudWatch Logsのどのログストリームを見たらよいのか、、、。
これ、まとめて確認できるようになりませんかね？
CloudWatch Logs まとめて表示！
構成
構成はこちらです。
シンプルに、クイックにできるような構成です。
- Amazon CloudWatch Logs
Amazon EC2 インスタンス、Route 53、およびその他のソースからログファイルをモニタリング、保存、およびアクセスできます。
今回は、CloudWatch Logsにログ出力されていることを前提とします。
- AWS Lambda
サーバーをプロビジョニングまたは管理することなくコードを実行できます。
今回は、CloudWatch Logsから特定のロググループのログをまとめて取得します。
さらに取得したログをHTMLに出力します。
やり方
Lambdaコンソールから
１．関数を新規に作成します。
ランタイムは「Python 3.13」を選択します。
２．関数URLを有効にします。
シンプルに、クイックに作るため、認証タイプは「NONE」を選択します。
３．アクセス権限で実行ロールに「CloudWatchLogsReadOnlyAccess」を追加します。
LambdaがCloudWatch Logsからログ取得するために必要となります。
４．以下のソースコードをデプロイします。
ソースコード
lambda_function.py
import json
import boto3
from datetime import datetime, timezone, timedelta
logs_client = boto3.client('logs')
def lambda_handler(event, context):
log_group_name = 'YOUR-LOG-GROUP-NAME'
# 日本時間で今日の日付の開始と終了のタイムスタンプを計算
jst = timezone(timedelta(hours=9))
now = datetime.now(jst)
start_of_day = datetime(now.year, now.month, now.day, 0, 0, 0, tzinfo=jst)
end_of_day = start_of_day + timedelta(days=1)
start_timestamp = int(start_of_day.timestamp() * 1000)
end_timestamp = int(end_of_day.timestamp() * 1000)
response = logs_client.describe_log_streams(
logGroupName=log_group_name,
orderBy='LastEventTime',
descending=True
)
log_streams = response['logStreams']
events = []
for stream in log_streams:
log_stream_name = stream['logStreamName']
log_events = logs_client.get_log_events(
logGroupName=log_group_name,
logStreamName=log_stream_name,
startTime=start_timestamp,
endTime=end_timestamp,
startFromHead=False
)
for event in log_events['events']:
events.append(event)
# タイムスタンプでソート
events.sort(key=lambda x: x['timestamp'])
html_content = generate_html(events, log_group_name)
return {
'statusCode': 200,
'headers': {
'Content-Type': 'text/html'
},
'body': html_content
}
def generate_html(events, log_group_name):
jst = timezone(timedelta(hours=9))
html = f"""
<!DOCTYPE html>
<html lang="ja">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>CloudWatch Logs</title>
<style>
body {{
font-family: Arial, sans-serif;
margin: 0;
padding: 0;
background-color: #f4f4f4;
}}
.container {{
margin: 20px auto;
padding: 20px;
width: 90%;
background-color: #fff;
box-shadow: 0 2px 4px rgba(0, 0, 0, 0.1);
border-radius: 8px;
position: relative;
}}
.header {{
background-color: #fff;
position: sticky;
top: 0;
z-index: 3;
padding: 10px 0;
border-bottom: 2px solid #e7e7e7;
}}
h1 {{
font-size: 24px;
color: #333;
margin: 0;
padding: 10px 0;
}}
h2 {{
font-size: 20px;
color: #666;
margin: 0;
padding: 10px 0;
}}
table {{
width: 100%;
border-collapse: collapse;
margin-top: 60px;
}}
th, td {{
border: 1px solid #ddd;
padding: 8px;
text-align: left;
}}
th {{
background-color: #f8f8f8;
font-weight: bold;
position: sticky;
top: 80px;
z-index: 2;
}}
tr:nth-child(even) {{
background-color: #f9f9f9;
}}
.error {{
color: red;
font-weight: bold;
}}
</style>
</head>
<body>
<div class="container">
<div class="header">
<h1>CloudWatch Logs</h1>
<h2>Log Group: {log_group_name}</h2>
</div>
<table>
<thead>
<tr>
<th>Timestamp</th>
<th>Message</th>
</tr>
</thead>
<tbody>
"""
for event in events:
timestamp = datetime.fromtimestamp(event['timestamp'] / 1000, tz=timezone.utc).astimezone(jst).strftime('%Y-%m-%dT%H:%M:%S')
message = event['message']
if 'ERROR' in message:
message = f'<span class="error">{message}</span>'
html += f"""
<tr>
<td>{timestamp}</td>
<td>{message}</td>
</tr>
"""
html += """
</tbody>
</table>
</div>
</body>
</html>
"""
return html
以下をポイントとしています。
- CloudWatch Logsっぽいレイアウトにする
- ヘッダを固定表示にする
- Timestampを日本時間にする
- Timestampでソート（昇順）する
- 「ERROR」がある場合は 赤文字 で表示する
- （大量にログがあると困るので）当日のログのみを出力するようにする
ソースコード説明
まとめて出力したいロググループ名を指定します。
lambda_function.py
log_group_name = 'YOUR-LOG-GROUP-NAME'
日本時間でTimestampを出力します。
lambda_function.py
# 日本時間で今日の日付の開始と終了のタイムスタンプを計算
jst = timezone(timedelta(hours=9))
now = datetime.now(jst)
当日のログを出力するようにします。
lambda_function.py
start_of_day = datetime(now.year, now.month, now.day, 0, 0, 0, tzinfo=jst)
end_of_day = start_of_day + timedelta(days=1)
start_timestamp = int(start_of_day.timestamp() * 1000)
end_timestamp = int(end_of_day.timestamp() * 1000)
response = logs_client.describe_log_streams(
logGroupName=log_group_name,
orderBy='LastEventTime',
descending=True
)
log_streams = response['logStreams']
events = []
for stream in log_streams:
log_stream_name = stream['logStreamName']
log_events = logs_client.get_log_events(
logGroupName=log_group_name,
logStreamName=log_stream_name,
startTime=start_timestamp,
endTime=end_timestamp,
startFromHead=False
)
for event in log_events['events']:
events.append(event)
HTMLを生成します。HTMLの詳細はgenerate_html関数を参照ください。
lambda_function.py
html_content = generate_html(events, log_group_name)
出力結果
ブラウザを起動し、Lambdaの関数URLを実行します。
CloudWatch Logsがまとめて見れるようになった！
おわりに
- シンプルに、クイックにCloudWatch Logsをまとめて出力することができました。
- CloudWatchのサービスいろいろあるから使っていきたいな。
- 少しでもみなさんの参考になりますように！