「CloudWatch Logs」の「サブスクリプションフィルター」を使って、Slackにほぼリアルタイムでエラーを通知する：AWSチートシート

　上記の設定が完了したら、「関数の作成」をクリックします。

　作成したLambda関数の詳細ページが表示されたら、ページの中央にある「コードソース」に記載されているデフォルトのコードを下記のように書き換えて「Deploy」をクリックします。なお、コード内の「https://hooks.slack.com/services/xxx/xxx/xxx」の部分は、前の手順で取得したWebhook URLに修正してください。

import json
import urllib.request
import base64
import gzip
 
from datetime import datetime
 
def lambda_handler(event, context):
    # CloudWatchLogsデータをデコード
    decoded_data = base64.b64decode(event['awslogs']['data'])
    json_data = json.loads(gzip.decompress(decoded_data))
 
    messages = []
    
    # ロググループ名取得
    messages.append('ロググループ名：' + json_data['logGroup'])
 
    messages.append('検知メッセージ：')
    # サブスクリプションフィルターで転送されたメッセージの取得 ※1
    for log_event in json_data['logEvents']:
        # ログ出力時刻取得
        int_timestamp = log_event['timestamp']
        datetime_timestamp = datetime.fromtimestamp((int_timestamp/1000))
        str_timestamp = datetime_timestamp.strftime('%Y/%m/%d %H:%M:%S.%f')
 
        # 時刻とメッセージを連結
        messages.append(str_timestamp + ' ' + log_event['message'])
 
    # @channelメンションと、コードブロック付与
    req = "<!channel>\n ```" + '\n'.join(messages) + "```"
    
    post_slack(req)
    return
 
def post_slack(argStr):
    message = argStr
    send_data = {
        "text": message,
    }
    send_text = json.dumps(send_data)
    request = urllib.request.Request(
        "https://hooks.slack.com/services/xxx/xxx/xxx", 
        data=send_text.encode('utf-8'), 
        method="POST"
    )
    with urllib.request.urlopen(request) as response:
        response_body = response.read().decode('utf-8')

　作成したLambda関数をテスト実行します。「テスト」ボタンをクリックして、下記表の通りにテストイベントを設定して「保存」をクリックします。

図3　テストイベントの作成画面

　テストの作成が完了したら、再度「テスト」ボタンをクリックします。Lambda関数が問題なく実行されてSlackにメッセージが送られれば、テストは成功です。

図4　Slackへのメッセージ

3.監視対象となるLambda関数の作成

　次に、監視対象となるLambda関数を作成します。今回はLambda関数を作成していますが、「cloudwatch logs」のロググループにログ出力できれば何でも構いません。

図5　監視対象となるLambda関数の作成画面

　上記の設定が完了したら、「関数の作成」をクリックします。作成したLambda関数の詳細ページが表示されたら、ページの中央にある「コードソース」の欄に記載されているデフォルトのコードを下記のように書き換えて「Deploy」をクリックします。

def lambda_handler(event, context):
    
    print(event['key1'])
    print(event['key2'])
    print(event['key3'])

　本例では、デフォルトテストテンプレートの「hello-world」実行で送信される、key1〜key3をログ出力するだけです。

　作成したLambda関数をテスト実行します。「テスト」ボタンをクリックし、下記表の通りにテストイベントを設定して「保存」をクリックします。

　テストの作成が完了したら、再度「テスト」ボタンをクリックします。図6の通りに実行結果が出力されたらテストは成功です。

図6　監視対象となるLambda関数のテスト結果

4.ロググループへのサブスクリプションフィルターの設定

　設定はこれで最後です。監視対象となるLambda関数のロググループに、サブスクリプションフィルターを設定します。「CloudWatchコンソール」にアクセスして監視対象となるロググループ詳細画面に進みます。

図7　監視対象となるLambda関数のロググループ詳細画面

　「サブスクリプションフィルター」タブをクリックし、「作成」→「Lambdaサブスクリプションフィルターを作成」の順にクリックします。

図8　監視対象となるLambda関数のサブスクリプションフィルター追加手順

　サブスクリプションフィルターの作成画面では、下記表の項目通りに設定し、「パターンをテスト」をクリックします。

　今回の設定では「value2」という文字列がログに出力されたときに、Slack通知するLambda関数を実行するように設定します。

図9　サブスクリプションフィルター設定画面（その1）

図10　サブスクリプションフィルター設定画面（その2）

　「テスト結果を表示」以下が上記の通りなら、「ストリーミングを開始」をクリックします。以下の完了画面が表示されたら設定は完了です。

図11　サブスクリプションフィルター設定完了画面

5.動作確認

　設定が完了したので、監視対象となるLambda関数の画面に戻り、テストを実行します。Slackに図12のように通知が来ていれば成功です。

図12　動作を確認した結果、Slackに届いたメッセージ

注意点

　サブスクリプションフィルターを実運用していて気付いた点を以下にまとめておきます。

1.サブスクリプションフィルターの設定は1つのロググループに対して2種類

　これはあくまでも「呼び出せる機能が2種類」という意味です。フィルターパターンについては、下記リンクを見ていただければ分かるように、複数条件でマッチさせることが可能です。

• AWSユーザーガイド -フィルターとパターンの構文-

2.サブスクリプションフィルターは編集が不可能

　図11を見て分かる通り、「削除」「作成」しかアクションが存在しません。つまり、フィルターパターンのみを変更したい場合、削除して作成するしか手段がありません。

3.サブスクリプションフィルターの起動単位

　起動はログストリーム単位です。つまり、Lambda関数などが多重起動して、いずれもフィルターパターンにマッチするログが出力された場合、別々にログが転送されSlack通知が届きます。

4.メッセージ転送の集約

　サブスクリプションフィルターは、マッチしたログを検知して即時に転送するわけではなく、最初にマッチしたログが見つかった時点から、おおむね8秒間程度待機時間が発生します。その間に再びマッチしたログが出力されたら、まとめて転送しています。

　本稿の例では、複数転送されてくることも考慮して作成してあります（ソース内※1部分のループが必要）。

5.サブスクリプションフィルター自体は無料だが

　設定そのものは無料ですが、後続処理の利用料金はかかるので注意が必要です。