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

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

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

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

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

前提環境

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関数で使用する環境変数です。

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

 # 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