概要

CloudWatch Logsにて収集しているログを、リアルタイムで別システムに転送するための仕組み及びセットアップ手順を示します。

構成

ログを転送するための仕組みを構成図にすると次のようになります。

CloudWatch Logsから直接別のシステムにログを転送することは不可能なため、中継サービスが必要になりますが、今回はそれをLambdaとしています。
CloudWatch Logsのサブスクリプションフィルターという機能を利用することで、CloudWatch Logsから次のAWSサービスにログを転送することができます。
　・Amazon OpenSearch Service
　・Kinesis Data Streams
　・Amazon Data Firehose
　・Lambda
この中でAmazon Data FirehoseとLambdaが外部へHTTP POSTすることができます。
ただしサブスクリプションフィルターを利用すると、ログが"base64 でエンコードされ、gzip 形式で圧縮される"という仕様になっています。
最終的な転送先システムでは圧縮及びエンコードされていないログを受け取りたく、Amazon Data Firehoseでは実現不可能のため、Lambdaを利用してデータ変換処理することにしました。

前提

・CloudWatch Logsのロググループが作成済みであること
・ロググループのログストリームにてログが受信できていること

手順

Lambdaセットアップ

Lambda関数はPythonで記述します。

HTTP POST送信するのにrequestsモジュールを使いたいですが、Lambdaに標準搭載されていないので使えるようにする必要があります。レイヤーと呼ばれる機能でモジュールをアップロードします。

レイヤー作成

こちらのページが非常に参考になると思います。
zipファイルを作成し、レイヤー作成時にzipファイルをアップロードしてください。

関数作成

Lambda関数を作成します。
作成時のパラメータは次の設定にて動作確認しています。基本的にデフォルト設定です。

関数名：任意の名前
Pythonランタイム：3.13
アーキテクチャ：x86_64
アクセス権限：新規でロール作成
Additional Configurations：すべてOFF

関数のコードを記述してください。こちらのコードはサンプルになります。
<URL>と<Token>の箇所は転送先システムの情報を入力して下さい。

import json
import requests
import os
import base64
import gzip
from io import BytesIO

def lambda_handler(event, context):
    # 転送先システムのエンドポイント
    url = '<URL>'

    # 認証トークン
    auth_token = '<Token>'

    # ヘッダーに認証トークンを追加
    headers = {
        'Content-Type': 'application/json',
        'Authorization': f'Bearer {auth_token}'
    }
    
    # デコード処理
    decoded_data = base64.b64decode(event['awslogs']['data'])
    
    # gzipを解凍
    json_data = json.loads(gzip.decompress(decoded_data))

    # ログを1件ずつループして外部システムに送信
    for log in json_data['logEvents']:
        # HTTP POSTリクエストを送信
        response = requests.post(url, json=log, headers=headers)

        # レスポンスステータスコードが200（HTTP OK）でなければエラーをログに出力
        if response.status_code != 200:
            print(f"Failed to send POST request. Status code: {response.status_code}")

記述出来たらデプロイして下さい。
また、先ほど作成したレイヤーを関数に追加してください。
Lambdaのセットアップは以上で完了です。

サブスクリプションフィルター設定

CloudWatch Logsのロググループ画面に移動し、Lambdaサブスクリプションフィルターを作成します。
Lambda関数には先ほど作成したものを指定してください。
その他はデフォルト設定で構いません。

動作確認

転送先システムにて、次のようにログが受信できることが確認できればセットアップ完了です。

{
   "id": "38628348948545520451557491645803902238867625204600930304",
   "timestamp": 1732155073952,
   "message": "<ログメッセージ>"
}

ご参考情報ですが、ログが生成されてから転送先システムにて受信するまでの時間を複数回測定したところ、1秒～10秒ほどで受信できたことを確認しています。

参考

https://docs.aws.amazon.com/ja_jp/AmazonCloudWatch/latest/logs/SubscriptionFilters.html
https://sebenkyo.com/2021/05/21/post-1979/
https://bbh.bz/2019/09/14/link-cloudwatchlogs-to-lambda/