CloudWatchのエラーログをSlackに通知する #Python

いちいちLaravelのログを見るのがめんどくさいと思いCloudWatch Logsに転送しました。しかし、Slackに通知させた方がより便利だと思ったので今回は下記のような構成で行きます。

＊ AWS SNSが抜けているのでいつか変更します(2022年08月08日)

通知したいロググループに対してメトリクスフィルターを作成する

CloudWatchにて通知したいEC2ののロググループを選択し、「アクション」「メトリクスフィルターを作成する」をクリック

フィルターパターンはERRORログを出力したいのでERRORとします。そして、メトリクスフィルターを作成しましょう。

SNSのトピックを作成しましょう

CloudWatch LogsのログをSNSのメッセージとして発行し、Lambda(サブスクリプション)がSlackに通知するようにします。

任意の名前でさくっと作りましょう。

Lambdaの処理

Lambdaの構成はこんな感じになります。

SNSには作成したトピックを割り当てます。

lambda_function.py


import json
import urllib.request
import logging
from collections import OrderedDict
import os
import datetime
import calendar
import boto3

logger = logging.getLogger()
logger.setLevel(logging.INFO)

#抽出するログデータの最大件数
OUTPUT_LIMIT=5
#何分前までを抽出対象期間とするか
TIME_FROM_MIN=5

def lambda_handler(event, context):
    
    logger.info("Event: " + str(event))
    message = json.loads(event['Records'][0]['Sns']['Message'])
    
    logs = boto3.client('logs')

    # MetricNameとNamespaceをキーにメトリクスフィルタの情報を取得する。
    metricfilters = logs.describe_metric_filters(
        metricName = message['Trigger']['MetricName'] ,
        metricNamespace = message['Trigger']['Namespace']
    )

    logger.info("Metricfilters: " + str(metricfilters))
    #ログストリームの抽出対象時刻をUNIXタイムに変換（取得期間は TIME_FROM_MIN 分前以降）
    #終了時刻はアラーム発生時刻の1分後
    timeto = datetime.datetime.strptime(message['StateChangeTime'][:19] ,'%Y-%m-%dT%H:%M:%S') + datetime.timedelta(minutes=1)
    u_to = calendar.timegm(timeto.utctimetuple()) * 1000
    #開始時刻は終了時刻のTIME_FROM_MIN分前
    timefrom = timeto - datetime.timedelta(minutes=TIME_FROM_MIN)
    u_from = calendar.timegm(timefrom.utctimetuple()) * 1000

    # ログストリームからログデータを取得
    response = logs.filter_log_events(
        logGroupName = metricfilters['metricFilters'][0]['logGroupName'] ,
        filterPattern = metricfilters['metricFilters'][0]['filterPattern'],
        startTime = u_from,
        endTime = u_to,
        limit = OUTPUT_LIMIT
    )

    # メッセージを整形しつつslackに通知
    for event in response['events']:
        postText = '''
        {logStreamName}
        {message}
        '''.format( logStreamName=str(event['logStreamName']),
                    message=str(event['message'])).strip()

        logger.info("Response: " + str(response))
        
        send_data = {
            "text": postText,
        }
        
        send_text = json.dumps(send_data)

        request = urllib.request.Request(
            "https://hooks.slack.com/services/xxxxxxxxx/xxxxxxxxx/xxxxxxx", 
            data=send_text.encode('utf-8'), 
            method="POST"
        )
        
        with urllib.request.urlopen(request) as response:
            
            response_body = response.read().decode('utf-8')

IAMロールの設定

IAMロールにて先ほど作成したLambda関数にポリシーをアタッチします。

そして、AWSLambdaBasicExecutionRoleをクリックしてにポリシーの編集を行います。

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": "logs:CreateLogGroup",
            "Resource": "------------------------------------------"
        },
        {
            "Effect": "Allow",
            "Action": [
                "logs:CreateLogStream",
                "logs:PutLogEvents",
                "logs:DescribeMetricFilters"
            ],
            "Resource": [
                "-------------------------------------------"
            ]
        }
    ]
}

　CloudWatch Alarmを設定します

　CloudWatch Alarmにてアラームを作成します。設定は以下の通りです。しきい値が１以上を超えたらSNSトピックに送信するように設定します。

　送信するSNSトピックは先ほど作成したものを選択

　Slackの設定

　https://w1618578007-hxz964946.slack.com/intl/ja-jp/apps
を開き、「マストアプリ」をクリックし、検索窓で「 Incoming Webhook 」と検索

　そしてSlackに追加して通知させたいチャンネルを選択します。その後、生成されたWebhook URLをコピーしましょう。のちに使います。

　最後に確認します

意図的にログを吐くようにして、このような通知が来れば完了です。アイコンやアプリ名はカスタマイズすることも可能ですよ。

　参考文献

・ https://qiita.com/yahagin/items/2248287c5285cd1e9201
・ https://blog.chakimar.net/aws-chatbot-cloudwatch-logs/