以下の内容はhttps://toranoana-lab.hatenablog.com/entry/2026/03/30/100000より取得しました。

CloudWatchのエラーログをSlackにリアルタイム転送する方法

AWS Lambda Slack Python

こんにちは！虎の穴ラボの鷺山です。

AWSでアプリケーションを運用すると、ログをCloudWatch Logsに集約・管理できて便利です。

ただ、定期的なログ監視を人力で行うのは大変です。AWSのコンソールに都度アクセスしてチェックするのも手間がかかります。とはいえ、エラーログや不穏なログにはすぐに気付けるようにしておきたいです。

そこで今回は、CloudWatchのエラーログをSlackに転送する仕組みの作り方をご紹介します。

普段使っているSlackなどのビジネスチャットツールにログを流すことで、負担なくログ監視ができるようになります。

前提環境

項目 バージョン・値
Python 3.14
AWSリージョン ap-northeast-1

SlackのWebhook URLの作成

Slackにログを転送する上で必要なWebhook URLを作成します。

  1. Slackアプリの管理ページ https://api.slack.com/apps に移動します。
  2. Create New AppまたはCreate an Appを押し、From scratchを選択します。
  3. App Nameにアプリ名を入力し、Pick a workspace to develop your app in: に作成先のワークスペースを選択します。
    • 今回はERRORログ通知というアプリ名とします。
    • Create Appを押すとアプリが作成されます。
  4. アプリが作成されたらIncoming Webhooksメニューを開き、Activate Incoming WebhooksをOnにします。
  5. 次にApp Homeメニューを開き、Your App's Presence in SlackのApp Display Nameを設定します。
    • Display Name (Bot Name): ボットの表示名を入力します。
    • Default username: ボットのユーザー名を入力します。
  6. もう一度Incoming Webhooksメニューを開き、Webhook URLs for Your WorkspaceのAdd New Webhookを押します。
  7. アクセス権の設定画面で投稿先のチャンネルを選択して許可するを押します。これでWebhook URLが生成されます。

「ERRORログ用」と「WARNログ用」でApp (Webhook URL)を分けて作成すると、Slack上でログの種別が分かりやすくなるのでおすすめです。

Lambda関数の作成

CloudWatchのログをSlackに転送するLambda関数を作成します。
Slackでの視認性を高めるため、ログを見やすい形式に整形して転送します。

今回はsendLogsToSlackという関数名にしました。ランタイムはPythonを使用します。

コードは以下のようになります。

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}"
    )

💡ポイント

上記のコードの各ポイントを解説します。

# 環境変数から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"]

このLambda関数で使用する環境変数です。

キー 値
SLACK_WEBHOOK_URL_FOR_ERROR 前のセクションで生成したERRORログ用のWebhook URL
SLACK_WEBHOOK_URL_FOR_WARN 前のセクションで生成したWARNログ用のWebhook URL

「設定」タブから事前に設定してください。

    # Base64からデコードし、gzipを解凍
    raw_data = event.get("awslogs", {}).get("data")
    decoded_data = base64.b64decode(raw_data)
    unzipped_data = gzip.decompress(decoded_data)

CloudWatchから転送されるデータにはBase64エンコードとgzip圧縮が施されているため、デコードと解凍を行っています。

    webhook_url = get_webhook_url(messages)

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

ログのメッセージに含まれる文字列からエラーレベルを判定し、ERRORとWARNのどちらのWebhook URLに転送するかを判定しています。
アプリケーションのログフォーマットに応じて変更してください。
ここでは、前後に半角スペースを含む「 ERROR 」という文字列が含まれる場合にERRORレベルと判定しています。

    # 各ログメッセージをSlackに送信
    for i, message in enumerate(messages):
        header = build_header(log_group, log_stream) if i == 0 else ""
        ...

def build_header(log_group, log_stream):
    ...

def build_cloudwatch_url(log_group, log_stream):
    ...

ロググループ・ログストリームの情報から、ヘッダー文字列を作成しています。
出力元のログストリーム名を含めることで「どのシステムからの出力か」を分かりやすくしています。
さらにCloudWatchのURLを含めることで、AWSのコンソールに移動しやすくしています。

        # 長いメッセージはカットし、改行を復元
        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()

最後に、長いメッセージを300文字で切り出したり、改行を復元 (\\n → \n) などの加工をした後、Slackへメッセージを送信しています。

サブスクリプションフィルターの作成

上記で作成したLambda関数に向けて、CloudWatchからログが転送されるように設定します。

対象のロググループの画面からLambda サブスクリプションフィルターを作成を選択します。

上記で作成したLambda関数 (sendLogsToSlack) を選択します。

ログ形式にその他を選択し、フィルターパターン、フィルター名を入力します。

例えば、前後に半角スペースを含む「 ERROR 」や「 WARN 」という文字列が含まれていた場合にログを転送させるには、以下のようにフィルターパターンを設定します。

?" ERROR " ?" WARN "

"..."で条件のグルーピング、?でOR条件を表現することができます。

実行結果

以上の設定で、ERRORやWARNのログメッセージがSlackに転送されるようになります。

▼ERRORログ

▼WARNログ

各メッセージのヘッダーには対象のログストリームへのリンクも含んでいるため、AWSのコンソールに移動してログの全容をすぐに確認できます。

まとめ

CloudWatchのエラーログを、Slack WebhookとLambdaを組み合わせてSlackに自動転送する仕組みをご紹介しました。
サブスクリプションフィルターでERRORやWARNを絞り込むことで、Slack上で必要なログだけを手軽に確認できます。
AWSでアプリケーションを運用している方はぜひ試してみてください！

採用情報

虎の穴ラボでは一緒に働く仲間を募集中です！
この記事を読んで、興味を持っていただけた方はぜひ弊社の採用情報をご覧ください。
toranoana-lab.co.jp



以上の内容はhttps://toranoana-lab.hatenablog.com/entry/2026/03/30/100000より取得しました。
このページはhttp://font.textar.tv/のウェブフォントを使用してます
不具合報告/要望等はこちらへお願いします。
モバイルやる夫Viewer Ver0.14