はじめに
CloudWatchログに出力されるログの中で、特定の条件（例えば、通知するフラグ=true）を満たした場合のみ、SNSを通じて担当者にメール通知を送る仕組みをやってみました。
目次
今回の構成
CloudwatchLogに出力されたログの中から、条件に一致したログを取得して、SNSを通じてメール通知する。
やってみよう！
では、やってみましょう！
前提条件
- AWS CLIが設定済み
- 適切なIAM権限を持つユーザー
- メール通知を受信するメールアドレス
Step 1: SNSトピックの作成
まず、メール通知用のSNSトピックを作成します。
|
1 2 3 4 5 6 7 8 |
# SNSトピック作成 aws sns create-topic --name log-notification-topic # メールアドレスをサブスクライブ aws sns subscribe \ --topic-arn arn:aws:sns:ap-northeast-1:YOUR_ACCOUNT_ID:log-notification-topic \ --protocol email \ --notification-endpoint your-email@example.com |
注意: サブスクライブ後、メールで確認リンクが送信されるので、必ずクリックして確認しましょう。
SNSトピック確認
Step 2: IAMロールの作成
Lambda関数用のIAMロール:log-notification-role を作成します。
このロールには、下記ポリシーをアタッチします。
権限ポリシー (log-notification-policy.json)
- logs:CreateLogGroup：CloudWatchにロググループを作成する権限
- logs:CreateLogStream：ロググループ内にログストリームを作成する権限
- logs:PutLogEvents：ログストリームにログイベント（実際のログメッセージ）を書き込む権限
- sns:Publish：指定されたSNSトピックにメッセージを発行する権限
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 |
{ "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "logs:CreateLogGroup", "logs:CreateLogStream", "logs:PutLogEvents" ], "Resource": "arn:aws:logs:*:*:*" }, { "Effect": "Allow", "Action": [ "sns:Publish" ], "Resource": "arn:aws:sns:ap-northeast-1:YOUR_ACCOUNT_ID:log-notification-topic" } ] } |
環境変数から、通知メールのSUBJECT、SNSトピックを取得しています。
Step 3: 通知機能Lambda関数の作成
ログを解析してSNSを通じてメール通知を送信するLambda関数を作成します。
Lambda関数コード (Python)
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 |
import base64 import json import gzip import os import boto3 import re def lambda_handler(event, context): # テスト実行時のエラーハンドリング if 'awslogs' not in event: return {'statusCode': 200, 'body': 'Test execution - no awslogs data'} try: # CloudWatch Logsからの圧縮データをデコード compressed_payload = base64.b64decode(event['awslogs']['data']) uncompressed_payload = gzip.decompress(compressed_payload) data_json = json.loads(uncompressed_payload) except Exception as e: print(f"データデコードエラー: {e}") return {'statusCode': 400, 'body': f'Decode error: {str(e)}'} # ログイベント全体を取得 log_events = data_json["logEvents"] # 各ログイベントに対して処理を実行 for log_event in log_events: try: sns = boto3.client('sns') # メッセージを取得 message_content = log_event['message'] # JSONパターンを抽出するための正規表現 json_pattern = r'\{.*\}' match = re.search(json_pattern, message_content) if match: # JSON部分を抽出 json_str = match.group(0) try: # JSONをパース json_obj = json.loads(json_str) # 日本語を正しく表示するために再度JSONに変換 formatted_json = json.dumps(json_obj, ensure_ascii=False, indent=2) # 元のメッセージからJSON部分を置換 message_content = message_content.replace(json_str, formatted_json) except json.JSONDecodeError: # JSONのパースに失敗した場合は元のまま pass message = "担当者様" + "\n" + "\n" message = message + message_content + "\n" + "\n" print(message) # SNSトピックに通知を発行 publishResponse = sns.publish( TopicArn = os.environ['SNS_TOPIC_ARN'], Message = message, Subject = os.environ['ALARM_SUBJECT'] ) print(publishResponse) except Exception as e: # エラーが発生した場合はログに出力 print(e) return {'statusCode': 200, 'body': 'Processing completed'} |
Step 4: CloudWatchログの準備
テスト用のログを出力するサンプルアプリケーションを作成します。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 |
import json import logging import time from datetime import datetime # Lambda環境用のロガー設定 logger = logging.getLogger() logger.setLevel(logging.INFO) def output_log(message, notification_flag=False, log_level="INFO"): log_data = { "logLevel": log_level, "timestamp": datetime.now().isoformat(), "message": { "message": message }, "notification_flag": "true" if notification_flag else "false" } log_json = json.dumps(log_data, ensure_ascii=False, indent=4) if log_level == "ERROR": logger.error(log_json) else: logger.info(log_json) def lambda_handler(event, context): print("Lambda関数が開始されました") # 通常ログ（検知なし） output_log("システム正常稼働中", False, "INFO") # エラーログ（検知あり） output_log("データベースの接続に失敗しました", True, "ERROR") # 通常ログ（検知なし） output_log("バッチ処理完了", False, "INFO") print("Lambda関数が完了しました") return {'statusCode': 200} |
Step 5: サブスクリプションフィルターの設定
Lambda関数にCloudWatch Logsからの実行権限を付与
|
1 2 3 4 5 6 |
aws lambda add-permission \ --function-name log-notification-processor \ --statement-id allow-cloudwatch-logs \ --action lambda:InvokeFunction \ --principal logs.amazonaws.com \ --source-arn "arn:aws:logs:ap-northeast-1:YOUR_ACCOUNT_ID:log-group:/aws/lambda/log-sample-app:*" |
Lambda関数のlog-notification-processorにCloudWatch Logsのトリガーが設定されます。
サブスクリプションフィルター作成
|
1 2 3 4 5 |
aws logs put-subscription-filter \ --log-group-name "/aws/lambda/log-sample-app" \ --filter-name "notification-filter" \ --filter-pattern '{ $.notification_flag = true }' \ --destination-arn "arn:aws:lambda:ap-northeast-1:YOUR_ACCOUNT_ID:function:log-notification-processor" |
ロググループ：/aws/lambda/log-sample-app に条件「notification_flag = true」に一致するログのみをLambda関数に送信するフィルターが設定されます。
|
1 2 3 4 5 6 7 8 |
{ $.notification_flag = true } は、ログメッセージをJSONとして解析し、notification_flagフィールドがtrueの場合にマッチします。 // ✅ マッチする（Lambda関数が実行される） {"message": "エラー発生", "notification_flag": true} // ❌ マッチしない（Lambda関数は実行されない） {"message": "正常処理", "notification_flag": false} {"message": "通常ログ"} |
Step 6: 動作テスト
サンプルアプリケーションを実行して動作確認！
- AWSコンソールでCloudWatch Logsを開く
- ロググループ
/aws/lambda/sample-appを確認 - ログが出力されていることを確認
123452025-11-28T23:41:10.032Z Lambda関数が開始されました2025-11-28T23:41:10.032Z [INFO] 2025-11-28T23:41:10.032Z 7985da0d-2f59-46ac-8e58-e2f3b5f6b4b1 { "logLevel": "INFO", "timestamp": "2025-11-28T23:41:10.032627", "message": { "message": "システム正常稼働中" }, "notification_flag": "false" }2025-11-28T23:41:10.032Z [ERROR] 2025-11-28T23:41:10.032Z 7985da0d-2f59-46ac-8e58-e2f3b5f6b4b1 { "logLevel": "ERROR", "timestamp": "2025-11-28T23:41:10.032771", "message": { "message": "データベースの接続に失敗しました" }, "notification_flag": "true" }2025-11-28T23:41:10.032Z [INFO] 2025-11-28T23:41:10.032Z 7985da0d-2f59-46ac-8e58-e2f3b5f6b4b1 { "logLevel": "INFO", "timestamp": "2025-11-28T23:41:10.032828", "message": { "message": "バッチ処理完了" }, "notification_flag": "false" }2025-11-28T23:41:10.032Z Lambda関数が完了しました
メール通知の確認
notification_flag が true のログのみメールに記載されて通知されました。
サブスクリプションフィルター vs CloudWatchアラーム
ログ監視には2つのアプローチがあります。要件に応じて適切な方式を選択することが必要です。
アーキテクチャ比較
サブスクリプションフィルター方式:
|
1 |
CloudWatch Logs → Subscription Filter → Lambda → SNS → Email |
CloudWatchアラーム方式:
|
1 |
CloudWatch Logs → Metric Filter → CloudWatch Alarm → SNS → Email |
機能比較表
| 項目 | サブスクリプションフィルター | CloudWatchアラーム |
|---|---|---|
| リアルタイム性 | 即座（秒単位） | 遅延あり（分単位） |
| 通知内容 | 個別ログの詳細情報 | 集計情報のみ |
| メール本文 | カスタマイズ可能 | 定型メッセージのみ |
| ログ詳細取得 | ○（ロググループ、ストリーム、メッセージ等） | ×（アラーム情報のみ） |
| 設定の複雑さ | 高（Lambda関数が必要） | 低（設定のみ） |
| コスト | Lambda実行料金 + SNS | メトリクスフィルター + アラーム + SNS |
| カスタマイズ性 | 高（任意の処理可能） | 低（閾値ベースのみ） |
| 通知頻度制御 | Lambda内で実装可能 | アラーム設定で制御 |
| 複数条件判定 | Lambda内で柔軟に実装 | 複数アラームが必要 |
| 障害時の影響 | Lambda障害で通知停止 | CloudWatchサービス依存 |
まとめ
CloudWatchサブスクリプションフィルターを使用することで、以下が出来ました。
- ログの条件付き監視（notification_flag=trueのみ）
- リアルタイムでのメール通知
- 日本時間での通知
- 構造化されたメール本文
この仕組みにより、重要なログのみを効率的に監視し、迅速な対応が可能になります。
応用例
- エラーレベル別の通知先変更
- Slackなど他の通知チャネルとの連携
- ログの集計・分析機能の追加
- 通知頻度の制御（同じエラーの重複通知防止）
注意点
- Lambda関数の実行時間とコストを考慮
- SNSの送信制限に注意
- ログの量が多い場合はフィルターパターンの最適化が重要
- セキュリティ上、機密情報をログに含めないよう注意
投稿者プロフィール
-
2023/1にスカイアーチネットワークスにJoin
AWSを日々勉強中
