ERROR行をリアルタイム検知／数十秒でSlack通知／ローカルテストサンプル付き

1. はじめに

CloudWatch Logs の ERROR を監視したい時、まず思いつく構成は
メトリクスフィルター → アラーム → SNS → Slack だと思います。

しかしこの方式には以下のようなデメリットがあります：

通知が数分遅れる
ERROR が 10件出ても「10件超えました」しか通知されない

本番運用では、

「1エラー = 1通知」
「リアルタイム性」

を求められるケースが多いです。

そこで今回は、
CloudWatch サブスクリプションフィルター + Lambda → Slack Webhook
の構成を使って、
CloudWatch Logs に ERROR が出た瞬間にほぼリアルタイムで通知する仕組み
を、実際に動いたコード・テストデータサンプル付きで公開します！

この記事のゴール：

✔ ERRORを1件ずつリアルタイム通知
✔ SlackにログURL + ログ本文を表示
✔ ローカルテストできるテンプレ付き
　（データ生成コード、テストデータサンプル、ローカルテストコード）

2. 結果

✔ Slack に実際に飛んだ通知

↑ CloudWatch ERROR が発生してから約15秒後に Slack に届いた通知
※一部マスキング済

例：

@channel 📢 システムA で ERROR が発生しました！
ログURL: https://console.aws.amazon.com/cloudwatch/...
ERROR: {...}
↓リアクションお願いします（確認開始: 👀／完了: ✅）

3. アーキテクチャ（図）

4. CloudWatch Logs が Lambda に渡す “実際のイベント形式”

※テストデータです。

この data は base64 + gzip でエンコードされたペイロードなので、
Lambda 内でデコード（base64 + gzip）する必要があります。
テストデータの作成方法は後述します。

5. ソースコード

完全版

全文は上記 GitHub に掲載し、この記事では “重要部分のみ” を解説します。

解説

環境変数の仕様

Key	Value
HOOK_URL_{n}	Slack Webhook URL
PATH_{n}	検知対象サービスのパス
PROJECT_NAME_{n}	対象アプリ・サービス名
MAX_LINES	Slackへ出力する最大行数

下記「webhook/project_name のマッピング」の話に繋がりますが、
今回の仕組みでは、複数のSlack Webhook URL・アプリやサービスの設定が可能です。

例：

HOOK_URL_1=https://hooks.slack...
MAX_LINES=3
PATH_1=/prod/items
PROJECT_NAME_1=ServiceA

webhook/project_name のマッピング

同じLambdaでいろいろなSlackチャンネルに + いろいろなサービス（プロジェクト）の検知内容を通知できるよう、
今回はマッピングを採用。

lambda_handler.py

def _build_maps() -> tuple[dict[str, str], dict[str, str]]:
    """
    環境変数 PATH_{n}, HOOK_URL_{n}, PROJECT_NAME_{n} から
    logGroup → webhook , project_name のマップを構築する
    """
    webhook_map: dict[str, str] = {}
    project_name_map: dict[str, str] = {}

    for key, value in os.environ.items():
        if not key.startswith("PATH_"):
            continue
        suffix = key.split("_", 1)[1]
        path = value
        if not path:
            continue

        hook_key = f"HOOK_URL_{suffix}"
        proj_key = f"PROJECT_NAME_{suffix}"

        hook = os.environ.get(hook_key)
        proj = os.environ.get(proj_key)

        if hook:
            webhook_map[path] = hook
        if proj:
            project_name_map[path] = proj

    return webhook_map, project_name_map

WEBHOOK_MAP, PROJECT_NAME_MAP = _build_maps()

🔧 CloudWatch Logs のデコード処理（base64 + gzip）

lambda_handler.py

decoded_data = base64.b64decode(event["awslogs"]["data"])
json_data = json.loads(gzip.decompress(decoded_data))

🔧 ログ抽出・行数制限

通知先（Slack）に載せるログは、他の情報が埋もれないように + エラーに気づいた非エンジニアでも読みやすいよう、短行で。

lambda_handler.py

messages = [e.get("message", "").strip() for e in log_events if e.get("message")]
if not messages:
    logger.warning("No message field found in any logEvent")
    return {"ok": False, "reason": "no_messages"}

error_message = "\n".join(messages[:MAX_LINES])

🔧 Slack Webhook 送信処理 (HTTPクライアントを使用)

HTTPクライアントで送ることが可能です。

lambda_handler.py

request = urllib.request.Request(
    hook_url,
    data=data,
    headers={"Content-Type": "application/json"},
    method="POST",
)

6. Subscription Filter の設定（ERRORだけを拾う）

サブスクリプションフィルターパターンは、正規表現 (regex)を採用しています。
例えば以下のように記述することで
複数件、OR条件で検知ワードを設定できます。

例:

?ERROR ?error

注意：
サブスクリプションフィルターは1つのログにつき2つまでしか設定できません。
例えば「ERROR用」と「監査ログ用」の2つを既に設定している場合、それ以上フィルターを増やせないため、
用途を絞る or ログ設計を整理する or 上記のようにパターンの書き方を工夫する必要があります。

7. Slack 側の設定（Workflow or Incoming Webhook）

変数：`text`, `log_url`, `project_name`

変数	型	説明
`text`	テキスト	通知に載せるログ
`log_url`	テキスト	該当ログのURL
`project_name`	テキスト	アプリ・サービス名

投稿テンプレ例

@channel 📢 {}project_name でエラーが発生しました！
エラー対応方法概要はコチラ: ＜マニュアルのURL＞
下記ログのURL: {}log_url
``` 
{}text

```
↓リアクションしてください！（確認開始: 👀, 対応完了: ✅）

Workflow Builderのスクショ

↑ Slack上でワークフローを組み終えたとき、このような画面になっているはず
※一部マスキング済

8. ローカルテスト

▪ テストデータの作成

上記のコードを実行すると、下記テストデータが作成されます。

▪ ローカルテストコード

環境変数を読み込んでから、lambda_handler をインポート するのがポイントです。

▪ 実行コマンド

テストデータの生成

generate_sample_event.py の配置されているディレクトリで以下を実行すると、
テストデータ sample_event.json が生成されます。

python generate_sample_event.py

ローカルテストを実施

テストデータが生成されたら、
local_test.py が配置されているディレクトリに移動しましょう。

Slack には投げず、中身だけ確認したい ときは以下を実行してください。

python local_test.py --dry-run

Slack に投げるところ含めて確認したいときは以下を実行してください。

python local_test.py

テスト結果例 (DRY RUN)

=== Local ENV summary ===
HOOK_URL_1 = .../sample***
PATH_1 = /aws/lambda/sample-log-group
PROJECT_NAME_1 = SampleService
MAX_LINES = 3
=========================

=== DRY RUN: urlopen called ===
URL: https://hooks.slack.com/services/XXXX/YYYY/ZZZZ
BODY:
{"log_url": "...", "project_name": "...", "text": "ERROR: ..."}
=== /DRY RUN ===
=== Result ===
{'ok': True}

9. よくあるハマりポイント

✔ 作成・編集した Slack Workflow が反映されない

→ ワークフローは作成・編集後に「公開」する必要があります（おそらく画面右上あたりに緑色ボタン「公開」があるはず）

✔ 通知が2回出る

→ ログ側で2行出ている可能性が高いです。監視対象ログの件数（それでもダメなら設計）を見直してみましょう。

✔ サブスクリプションフィルターではなく、メトリクスフィルターではダメ？

→ はじめに記載の通り、メトリクスフィルターの場合、集約しての通知 になってしまい、
　リアルタイム用途には不向き
→ ダッシュボード用途なら、併用は有効です

✔ デコードできない（decode_error）

→ sample_event.json の base64/gzip が壊れている可能性が高いです
（サンプルデータを作り直してみましょう。上記で紹介した generate_sample_event.py の使用が有効です）

✔ 検知対象の Lambda 自身のログには紐づけられない

→ CloudWatch の仕様です。もしLambdaのログを検知したい場合でも、
　今回ご紹介したような通知用Lambda を作成しましょう。

10. コスト（月額費用目安）

cloudWatch サブスクリプションフィルター：完全無料
Lambda：月数万〜数十万回の通知でも、ほぼ無料枠の範囲
→ 標準的なシステムなら “0円運用” が現実的
追加で課金されるのは CloudWatch Logs の保存費用のみ
（今回の仕組みとは関係なく、通常のログ運用コスト）

⇒ 全体としては 月0〜数十円、ほぼ無料で運用できます。

11. まとめ

CloudWatch Logs サブスクリプションフィルター + Lambda の組み合わせは、
「1つのエラーを確実に一つずつSlackに通知したい」場合に最適です。

今回の記事では、CloudWatch Logs の decode → 整形 → Slack POST まで
実際に動いたコードをすべて公開しました！必要に応じて自由にカスタマイズしてください◎

12. GitHub リンク