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
を使ってお手軽に実装した👏 便利〜 \( 'ω')/
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 コンテキスト依存を避ける実装を紹介していたりする💡
動作確認
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"}