1. はじめに

CloudWatch Logsに出力されるログによってプログラムの処理フローを分岐する必要がある要件がありました。

これまでCloudWatch Logsを操作する機会がなかったため、boto3を用いたログのフィルタリング方法について備忘録として記事を残します。

2. CloudWatch Logsとは

CloudWatch Logsは、Lambda、EC2、CloudTrail、Route53などのAWSリソースから出力されるログをモニタリング、保存、検索するためのサービスです。

AWSマネジメントコンソールからログの閲覧や文字列検索が可能なほか、例えば ERROR という文字列を検出したタイミングでCloudWatch Alarmをトリガーし、SNSで通知を行うこともできます。

3. 環境

• AWS Lambda（Python 3.13）
• boto3
• リージョン：us-west-2

4. boto3でCloudWatch Logsをフィルタリングする

4.1 boto3とは

boto3は、PythonからAWSリソースを操作するための公式SDK（Software Development Kit）です。

※ Lambdaを利用している場合、boto3はデフォルトで利用することができます。

今回はCloudWatch Logsのfilter_log_events APIを使い、特定のパターンにマッチするログを抽出します。

filter_log_eventsを利用して、Pythonでログの解析、分析に役立てることができます。

4.2 サンプルソースコード

以下のプログラムは、CloudWatch Logs に出力された過去1時間分のログから “START_DATE” と “SUCCESS” を含む JSON 形式のログメッセージを抽出し、START_DATE の中で最新の日付を1件だけ取得・表示する Lambda 関数です。

import boto3
import time
from datetime import datetime, timedelta
import json
import logging

logger = logging.getLogger("my_lambda_logger")
logger.setLevel(logging.INFO)

if not logger.handlers:
    handler = logging.StreamHandler()
    formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
    handler.setFormatter(formatter)
    logger.addHandler(handler)

client = boto3.client('logs', region_name='us-west-2')
log_group_name = '/aws/lambda/log-group-name'

def lambda_handler(event, context):
    end_time = int(time.time() * 1000)
    start_time = int((datetime.utcnow() - timedelta(minutes=60)).timestamp() * 1000)

    start_dates = []
    next_token = None

    while True:
        kwargs = {
            "logGroupName": log_group_name,
            "startTime": start_time,
            "endTime": end_time,
            "filterPattern": '?START_DATE ?SUCCESS'
        }

        if next_token:
            kwargs["nextToken"] = next_token

        response = client.filter_log_events(**kwargs)

        for e in response.get("events", []):
            try:
                message = json.loads(e["message"])
                if "START_DATE" in message:
                    start_dates.append(message["START_DATE"])
            except Exception as ex:
                logger.warning(f"Failed to parse message: {e['message']} -- {ex}")

        next_token = response.get("nextToken")
        if not next_token:
            break

    logger.info(f"Collected START_DATEs: {start_dates}")

    if start_dates:
        dt_dates = [datetime.strptime(d, "%Y-%m-%d") for d in start_dates]
        dt_dates.sort(reverse=True)
        latest_one = dt_dates[:1]
        latest_one_str = [d.strftime("%Y-%m-%d") for d in latest_one]
    else:
        latest_one_str = []

4.3 filter_log_eventsの解説

filter_log_events は、指定したロググループの中から特定の条件にマッチするログイベントを抽出するための boto3 の API です。大量のログデータの中から、必要な情報を取得するのに適しています。

4.4 filterPatternの使い方

このパターンは、ログメッセージ内に START_DATE と SUCCESS の両方の文字列が含まれている場合にマッチします。 JSON形式で出力されたログに対しても文字列として判定されるため、構造化ログであってもこのような検索が可能です。

"filterPattern": "?START_DATE ?SUCCESS"

フィルターパターン構文はこちらを参照

5. まとめ

本記事では、CloudWatch Logs に出力されたログを boto3 を使ってフィルタリングし、特定の条件（“START_DATE” と “SUCCESS” を含む）に一致するログを抽出する方法を紹介しました。

このように filter_log_events と filterPattern を活用することで、構造化ログの中から必要な情報だけを効率よく抽出し、Lambda 関数内で動的に処理を分岐させることが可能になります。

特に、バッチ処理やジョブ管理において「直近の成功した処理日時」を取得するようなユースケースでは、有効なアプローチです。

CloudWatch Logs のログ構造やフィルタパターンに慣れることで、より柔軟な監視や処理制御が行えるようになるため、業務の信頼性向上にもつながります。

6. 参考ページ

boto3

Filter pattern syntax for metric filters, subscription filters, filter log events, and Live Tail

filter_log_events