🐥

【AWS】CloudWatchLogs の Lmabda サブスクリプションフィルターからエラーの直前のログを含めてアラームをあげる

2024/08/18に公開

AWS

背景

CloudWatch を使用してログ監視を行う場合、マネージドな機能がないので、Lambda サブスクリプションフィルターを使用して特定の文字列が出力された場合に SNS トピックをトリガーする作りこみをしている方は多いと思います。
ただ、普通にやるとエラー内容しかメールに添付できません。
エラーの直前のログはマネコンにログインして、ロググループ>ログストリームを確認する必要があります。
メールにタイムスタンプやログストリーム名を記載してても毎回やるのは結構めんどくさいです。
メールにエラー直前のログまで添付すれば解析の時間を短縮できるため、今回はそれを実現していきます。

構成

使用するサービス・機能は以下の通り

CloudWatch Logs
Lambda サブスクリプションフィルター
Lambda
SNS

Lambda のコード

python を採用しました。

import base64
import logging
import json
import zlib
import datetime
import os
from datetime import datetime, timezone, timedelta
import boto3
from botocore.exceptions import ClientError


# logger初期化
logger = logging.getLogger()
logger.setLevel(logging.INFO)


#ハンドラー
def lambda_handler(event, context):
    logger.info("LOAD Function: " + context.function_name)

    #CloudWatchLogsから渡される値は圧縮エンコードされているためデコードして解凍
    cwlogs_raw_data = zlib.decompress(base64.b64decode(event['awslogs']['data']), 16+zlib.MAX_WBITS)
        
    #JSON読み込み
    cwlogs_raw_data_json = json.loads(cwlogs_raw_data)

    #Lambdaサブスクリプションフィルターから連携された情報からデータ取得
    account_id = cwlogs_raw_data_json['owner']
    log_group_name = cwlogs_raw_data_json['logGroup']
    log_stream_name = cwlogs_raw_data_json['logStream']
    log_timestamp = convert_time(int(cwlogs_raw_data_json['logEvents'][0]['timestamp']))
    log_message = cwlogs_raw_data_json['logEvents'][0]['message']
    target_timestamp = cwlogs_raw_data_json['logEvents'][0]['timestamp']

    #ログストリーム内のログデータを取得する
    logs_with_timestamp = get_logs_around_with_ids(log_group_name, log_stream_name)

    #timestampを基に前後のログを取得する
    before = os.environ['GET_LOG_BEFORE']
    context_logs = get_logs_by_timestamp(logs_with_timestamp, target_timestamp, before)

    #SNS連携&メール送信
    send_mail(account_id, log_group_name, log_stream_name, log_timestamp, log_message, context_logs)

    logger.info("END Function: " + context.function_name)


# CloudWatchLogs LogsStream内の全てのtimestampとmessageのペアを返す関数
def get_logs_around_with_ids(log_group_name, log_stream_name):
    logs_client = boto3.client('logs')
    response = logs_client.get_log_events(
        logGroupName=log_group_name,
        logStreamName=log_stream_name,
        limit=1000
    )
    
    logs_with_timestamp = [{'message': event['message'], 'timestamp': event['timestamp']} for event in response['events']]
    return logs_with_timestamp


# 取得したログのリストから指定したtimestampを持つログを探して、その前後のログを抽出してリストに格納する関数
def get_logs_by_timestamp(logs_with_timestamp, target_timestamp, before):
    for i, log in enumerate(logs_with_timestamp):
        if log['timestamp'] == target_timestamp:
            # 負の値は考えられないためmaxで抑制する
            start_index = max(0, i - int(before))
            end_index = i + 1
            return [log['message'] for log in logs_with_timestamp[start_index:end_index]]
    return []


# タイムスタンプ時刻変換用関数
def convert_time(raw_timestamp):
    # UNIXミリ秒から秒への変換
    timestamp_sec = raw_timestamp / 1000
    dt = datetime.fromtimestamp(timestamp_sec, timezone(timedelta(hours=9)))
    # ISO:8601形式に変換
    timestamp_jst = dt.isoformat()

    return timestamp_jst


# SNS連携用関数
def send_mail(owner, loggroup, logstream, logtimestamp, message, expand_mesages):
    sns_client = boto3.client('sns')
    sns_topic_arn = os.environ['SNS_TOPIC_ARN']
    messages = expand_mesages

    sns_client.publish (
        TopicArn = sns_topic_arn,
        Subject = "ログ監視アラーム",
        Message =(
            "ログ グループ名: " + loggroup + "\n" +
            "ログ ストリーム名: " + logstream + "\n" +
            "ログ 検知日時(JST): " + logtimestamp + "\n" +
            "ログ 内容: " + message + "\n" +
            "エラーログの直前のログ:\n" + 
            "\n".join(messages) + "\n"
        )
    )

工夫したところ

ログストリームからエラーログの直前のログを取得するところ。
get_log_eventsメソッドでとってきていますが、response 内に eventId とかがなかったので、しょうがなく timestamp でエラーログを特定して、その直前〇個のログを取得して、配列に入れるようにしました。
本当は eventId みたいな値でとらないと重複の可能性がありますが、CloudWatchLogsに関しては、ミリ秒単位のUNIX時間で出るので、そこまで重複は気にしていません・・・('ω')
調べたところ、boto3 の当該メソッドでは、ログをクエリするのにeventIdとかは使えないようです。
また、Reponse Syntax には以下の情報が出力されると記載がありました。

timestamp
message
ingestionTime
nextForwardToken
nextBackwardToken

参考

生データ

cwlogs_raw_data:b'
{
    "messageType": "DATA_MESSAGE",
    "owner": "XXXXXXXX",
    "logGroup": "failure-notification-logs",
    "logStream": "test",
    "subscriptionFilters": [
        "failure-notification"
    ],
    "logEvents": [
        {
            "id": "38445998629021071516356264961554827036108240146371706880",
            "timestamp": 1723978202825,
            "message": "error"
        }
    ]
}

get_event_logメソッドで取得した場合の生データ

events: [
{
timestamp: 1723970143845,
message: test,
ingestionTime: 1723970144794
},
{
timestamp: 1723970166214,
message: [ERROR]This is a test!!!,
ingestionTime: 1723970167276
},
{
timestamp: 1723973829766,
message: hogehoge,
ingestionTime: 1723973830883
},
{
timestamp: 1723973841634,
message: [ERROR]This is a test!!!,
ingestionTime: 1723973842752
},
{
timestamp: 1723973852027,
message: fugafuga,
ingestionTime: 1723973853137
},
{
timestamp: 1723975568373,
message: [ERROR]This is a test!!!,
ingestionTime: 1723975569421
},
{
timestamp: 1723975645329,
message: chokuchokuchokuchoku,
ingestionTime: 1723975646547
},
{
timestamp: 1723975650272,
message: [ERROR]This is a test!!!,
ingestionTime: 1723975651316
}
]

背景

構成

Lambda のコード

工夫したところ

Discussion