CloudWatchのログをboto3でフィルタリングする
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 です。大量のログデータの中から、必要な情報を取得するのに適しています。