はじめに

みなさん、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のサービスいろいろあるから使っていきたいな。
少しでもみなさんの参考になりますように！