Amazon CloudWatch Logs サブスクリプションフィルタを使ってログを AWS Lambda に流す構成を AWS アカウントにデプロイする前に LocalStack にデプロイして確認してみた❗

LocalStack は Amazon CloudWatch Logs サブスクリプションフィルタもサポートしている👌しかし注意点はあって，サブスクリプションフィルタに設定するフィルタパターン（JSON フィルタ・正規表現フィルタなど）は LocalStack Pro でサポートされている💡よって，今回はフィルタなし（すべてのログを流す）で試す．

あくまでサンプルとして以下の構成を LocalStack を使ってローカル環境にデプロイする．

サンプルコード

👾 template.yaml（AWS SAM テンプレート）

今回は以下のように AWS SAM テンプレートを実装した📝AWS Lambda 関数は log-sender と log-receiver の2つを準備して，log-sender の Amazon CloudWatch Logs ロググループにサブスクリプションフィルタを設定してある👌

AWSTemplateFormatVersion: 2010-09-09
Transform: AWS::Serverless-2016-10-31

Resources:
  LogSenderFunction:
    Type: AWS::Serverless::Function
    Properties:
      FunctionName: log-sender
      CodeUri: src/
      Handler: log-sender.lambda_handler
      Runtime: python3.12
      Architectures:
        - x86_64
  LogSenderLogGroup:
    Type: AWS::Logs::LogGroup
    Properties:
      LogGroupName: /aws/lambda/log-sender
  LogSenderSubscriptionFilter:
    Type: AWS::Logs::SubscriptionFilter
    Properties:
      LogGroupName: !Ref LogSenderLogGroup
      FilterPattern: ""
      DestinationArn: !GetAtt LogReceiverFunction.Arn
  LogReceiverFunction:
    Type: AWS::Serverless::Function
    Properties:
      FunctionName: log-receiver
      CodeUri: src/
      Handler: log-receiver.lambda_handler
      Runtime: python3.12
      Architectures:
        - x86_64

👾 log-sender.py

log-sender ではシンプルに {"id": "27a1cc30-b4c4-4192-9db9-d19962fe8f33", "message": "sample message"} のように UUID と固定メッセージを含んだ構造化ログ (JSON) を出力する👌

import json
import uuid


def main():
    print(
        json.dumps(
            {
                'id': str(uuid.uuid4()),
                'message': 'sample message',
            }
        )
    )


def lambda_handler(event, context):
    main()


if __name__ == '__main__':
    main()

👾 log-receiver.py

log-receiver はサブスクリプションフィルタから流れてきたログをそのままログに出力する．サブスクリプションフィルタ経由だとログは GZIP 圧縮と Base64 エンコードで変換された状態になるけど，今回は Powertools for AWS Lambda (Python) の Event Source Data Classes で CloudWatchLogsEvent と CloudWatchLogsDecodedData を使ってお手軽に実装した👏 便利〜＼( 'ω')／

docs.powertools.aws.dev

import json

from aws_lambda_powertools.utilities.data_classes import CloudWatchLogsEvent
from aws_lambda_powertools.utilities.data_classes.cloud_watch_logs_event import CloudWatchLogsDecodedData


def main(event):
    event = CloudWatchLogsEvent(event)
    decompressed: CloudWatchLogsDecodedData = event.parse_logs_data()
    logs = decompressed.log_events
    for log in logs:
        print(log.message)


def lambda_handler(event, context):
    main(event)


if __name__ == '__main__':
    with open('./events/event.json', 'r') as f:
        event = json.load(f)

    main(event)

ちなみに @event_source(data_class=CloudWatchLogsEvent) のようにデコレータを使えば event = CloudWatchLogsEvent(event) という値の詰め直しは不要になるけど，Lambda コンテキストに依存していてローカル開発がしにくく採用しなかった．前に紹介した Powertools for AWS Lambda (Python) の Validation でも Lambda コンテキスト依存を避ける実装を紹介していたりする💡

kakakakakku.hatenablog.com

動作確認

samlocal コマンドでビルド・デプロイをして，awslocal コマンドで AWS Lambda 関数を実行した．

$ samlocal build
$ samlocal deploy

$ awslocal lambda invoke --function-name log-sender outfile

LocalStack の Amazon CloudWatch Logs を確認すると，期待通りに log-sender と log-receiver どちらにも同じログが出ていた👌

log-sender

{"id": "27a1cc30-b4c4-4192-9db9-d19962fe8f33", "message": "sample message"}

log-receiver

{"id": "27a1cc30-b4c4-4192-9db9-d19962fe8f33", "message": "sample message"}

kakakakakku blog

Weekly Tech Blog: Keep on Learning!

LocalStack を使って CloudWatch Logs サブスクリプションフィルタをローカル環境で試す