{
  "$type": "site.standard.document",
  "bskyPostRef": {
    "cid": "bafyreicp4gkuw2av5yugxnkwp63kqnwo7vvkpjrh2sm5pdsiqcezmegsge",
    "uri": "at://did:plc:ctnrfotecqqhvo7kl5cgtpwt/app.bsky.feed.post/3miasdbfk5n72"
  },
  "coverImage": {
    "$type": "blob",
    "ref": {
      "$link": "bafkreih2neo5vcziuzon2jgaejt4pclo33zlkqhuzyplzbt6ohp522xtxy"
    },
    "mimeType": "image/png",
    "size": 759187
  },
  "path": "/entry/2026/03/30/100000",
  "publishedAt": "2026-03-30T01:00:00.000Z",
  "site": "https://toranoana-lab.hatenablog.com",
  "tags": [
    "https://api.slack.com/apps",
    "toranoana-lab.co.jp"
  ],
  "textContent": "こんにちは！虎の穴ラボの鷺山です。\n\nAWSでアプリケーションを運用すると、ログをCloudWatch Logsに集約・管理できて便利です。\n\nただ、定期的なログ監視を人力で行うのは大変です。AWSのコンソールに都度アクセスしてチェックするのも手間がかかります。とはいえ、エラーログや不穏なログにはすぐに気付けるようにしておきたいです。\n\nそこで今回は、**CloudWatchのエラーログをSlackに転送する仕組み** の作り方をご紹介します。\n\n普段使っているSlackなどのビジネスチャットツールにログを流すことで、負担なくログ監視ができるようになります。\n\n## 前提環境\n\n項目  |  バージョン・値\n---|---\nPython  |  3.14\nAWSリージョン  |  `ap-northeast-1`\n\n## SlackのWebhook URLの作成\n\nSlackにログを転送する上で必要なWebhook URLを作成します。\n\n  1. Slackアプリの管理ページ https://api.slack.com/apps に移動します。\n  2. **Create New App** または**Create an App** を押し、**From scratch** を選択します。\n  3. **App Name** にアプリ名を入力し、**Pick a workspace to develop your app in:** に作成先のワークスペースを選択します。\n     * 今回は**ERRORログ通知** というアプリ名とします。\n     * **Create App** を押すとアプリが作成されます。\n  4. アプリが作成されたら**Incoming Webhooks** メニューを開き、**Activate Incoming Webhooks** を**On** にします。\n  5. 次に**App Home** メニューを開き、**Your App's Presence in Slack** の**App Display Name** を設定します。\n     * **Display Name (Bot Name):** ボットの表示名を入力します。\n     * **Default username:** ボットのユーザー名を入力します。\n  6. もう一度**Incoming Webhooks** メニューを開き、**Webhook URLs for Your Workspace** の**Add New Webhook** を押します。\n  7. アクセス権の設定画面で投稿先のチャンネルを選択して**許可する** を押します。これでWebhook URLが生成されます。\n\n\n\n「ERRORログ用」と「WARNログ用」でApp (Webhook URL)を分けて作成すると、Slack上でログの種別が分かりやすくなるのでおすすめです。\n\n## Lambda関数の作成\n\nCloudWatchのログをSlackに転送するLambda関数を作成します。\nSlackでの視認性を高めるため、ログを見やすい形式に整形して転送します。\n\n今回は**sendLogsToSlack** という関数名にしました。ランタイムは**Python** を使用します。\n\nコードは以下のようになります。\n\n\n    import base64 import gzip import json import os from urllib.parse import quote from urllib.request import Request, urlopen # 環境変数からSlack Webhook URLを取得 SLACK_WEBHOOK_URL_FOR_ERROR = os.environ[\"SLACK_WEBHOOK_URL_FOR_ERROR\"] SLACK_WEBHOOK_URL_FOR_WARN = os.environ[\"SLACK_WEBHOOK_URL_FOR_WARN\"] def lambda_handler(event, context): # Base64からデコードし、gzipを解凍 raw_data = event.get(\"awslogs\", {}).get(\"data\") decoded_data = base64.b64decode(raw_data) unzipped_data = gzip.decompress(decoded_data) # JSON形式のデータをロードし、ログメッセージを抽出 json_data = json.loads(unzipped_data.decode(\"utf-8\")) log_group = json_data.get(\"logGroup\") log_stream = json_data.get(\"logStream\") log_events = json_data.get(\"logEvents\") messages = [log_event[\"message\"] for log_event in log_events] print(f\"Incoming data: {json.dumps(json_data)}\") webhook_url = get_webhook_url(messages) # 各ログメッセージをSlackに送信 for i, message in enumerate(messages): header = build_header(log_group, log_stream) if i == 0 else \"\" # 長いメッセージはカットし、改行を復元 message = message[:300].replace(\"\\\\n\", \"\\n\").rstrip() message = f\"{header}```\\n{message}\\n```\" # Slackへのリクエストを作成し送信 payload = json.dumps({\"text\": message}).encode(\"utf-8\") req = Request(webhook_url, payload) urlopen(req).read() return {\"statusCode\": 200} def get_webhook_url(messages): message = messages[0] if \" ERROR \" in message.upper(): # 👈 ログフォーマットに応じて変更してください return SLACK_WEBHOOK_URL_FOR_ERROR else: return SLACK_WEBHOOK_URL_FOR_WARN def build_header(log_group, log_stream): app_name = log_group.split(\"/\")[-1] cloudwatch_url = build_cloudwatch_url(log_group, log_stream) return f\"• <{cloudwatch_url}|`{app_name}`>:\\n\" def build_cloudwatch_url(log_group, log_stream): escaped_log_group = quote(log_group, safe=\"\").replace(\"%2F\", \"%252F\") escaped_log_stream = quote(log_stream, safe=\"\").replace(\"%2F\", \"%252F\") return ( f\"https://ap-northeast-1.console.aws.amazon.com\" f\"/cloudwatch/home?region=ap-northeast-1\" f\"#logsV2:log-groups/log-group/{escaped_log_group}\" f\"/log-events/{escaped_log_stream}\" )\n\n### 💡ポイント\n\n上記のコードの各ポイントを解説します。\n\n\n    # 環境変数からSlack Webhook URLを取得 SLACK_WEBHOOK_URL_FOR_ERROR = os.environ[\"SLACK_WEBHOOK_URL_FOR_ERROR\"] SLACK_WEBHOOK_URL_FOR_WARN = os.environ[\"SLACK_WEBHOOK_URL_FOR_WARN\"]\n\nこのLambda関数で使用する環境変数です。\n\nキー  |  値\n---|---\n**SLACK_WEBHOOK_URL_FOR_ERROR** |  前のセクションで生成したERRORログ用のWebhook URL\n**SLACK_WEBHOOK_URL_FOR_WARN** |  前のセクションで生成したWARNログ用のWebhook URL\n\n「設定」タブから事前に設定してください。\n\n\n     # Base64からデコードし、gzipを解凍 raw_data = event.get(\"awslogs\", {}).get(\"data\") decoded_data = base64.b64decode(raw_data) unzipped_data = gzip.decompress(decoded_data)\n\nCloudWatchから転送されるデータにはBase64エンコードとgzip圧縮が施されているため、デコードと解凍を行っています。\n\n\n     webhook_url = get_webhook_url(messages)\n\n\n    def get_webhook_url(messages): message = messages[0] if \" ERROR \" in message.upper(): # 👈 ログフォーマットに応じて変更してください return SLACK_WEBHOOK_URL_FOR_ERROR else: return SLACK_WEBHOOK_URL_FOR_WARN\n\nログのメッセージに含まれる文字列からエラーレベルを判定し、ERRORとWARNのどちらのWebhook URLに転送するかを判定しています。\nアプリケーションのログフォーマットに応じて変更してください。\nここでは、前後に半角スペースを含む**「 ERROR 」** という文字列が含まれる場合にERRORレベルと判定しています。\n\n\n     # 各ログメッセージをSlackに送信 for i, message in enumerate(messages): header = build_header(log_group, log_stream) if i == 0 else \"\" ...\n\n\n    def build_header(log_group, log_stream): ... def build_cloudwatch_url(log_group, log_stream): ...\n\nロググループ・ログストリームの情報から、ヘッダー文字列を作成しています。\n出力元のログストリーム名を含めることで「どのシステムからの出力か」を分かりやすくしています。\nさらにCloudWatchのURLを含めることで、AWSのコンソールに移動しやすくしています。\n\n\n     # 長いメッセージはカットし、改行を復元 message = message[:300].replace(\"\\\\n\", \"\\n\").rstrip() message = f\"{header}```\\n{message}\\n```\" # Slackへのリクエストを作成し送信 payload = json.dumps({\"text\": message}).encode(\"utf-8\") req = Request(webhook_url, payload) urlopen(req).read()\n\n最後に、長いメッセージを300文字で切り出したり、改行を復元 (`\\\\n` → `\\n`) などの加工をした後、Slackへメッセージを送信しています。\n\n## サブスクリプションフィルターの作成\n\n上記で作成したLambda関数に向けて、CloudWatchからログが転送されるように設定します。\n\n対象のロググループの画面から**Lambda サブスクリプションフィルターを作成** を選択します。\n\n上記で作成したLambda関数 (`sendLogsToSlack`) を選択します。\n\nログ形式に**その他** を選択し、フィルターパターン、フィルター名を入力します。\n\n例えば、前後に半角スペースを含む**「 ERROR 」** や**「 WARN 」** という文字列が含まれていた場合にログを転送させるには、以下のようにフィルターパターンを設定します。\n\n\n    ?\" ERROR \" ?\" WARN \"\n\n`\"...\"`で条件のグルーピング、`?`でOR条件を表現することができます。\n\n## 実行結果\n\n以上の設定で、ERRORやWARNのログメッセージがSlackに転送されるようになります。\n\n▼ERRORログ\n\n▼WARNログ\n\n各メッセージのヘッダーには対象のログストリームへのリンクも含んでいるため、AWSのコンソールに移動してログの全容をすぐに確認できます。\n\n## まとめ\n\nCloudWatchのエラーログを、Slack WebhookとLambdaを組み合わせてSlackに自動転送する仕組みをご紹介しました。\nサブスクリプションフィルターでERRORやWARNを絞り込むことで、Slack上で必要なログだけを手軽に確認できます。\nAWSでアプリケーションを運用している方はぜひ試してみてください！\n\n# 採用情報\n\n虎の穴ラボでは一緒に働く仲間を募集中です！\nこの記事を読んで、興味を持っていただけた方はぜひ弊社の採用情報をご覧ください。\ntoranoana-lab.co.jp",
  "title": "CloudWatchのエラーログをSlackにリアルタイム転送する方法",
  "updatedAt": "2026-03-30T01:00:01.000Z"
}