AWS Lambdaを使って特定ログを通知する

Index

1. 挨拶＆背景の話
2. 構成
3. 実装
4. 感想

1. 挨拶 & 背景の話

こんちには、アプリケーションエンジニアで働いてます。キムでございます。
この前、運用してるアプリケーション内で発生するとあるエラーを検知する処理を入れたことがあります。
そこまで素晴らしい構成ではないが、この構成を作るときに、なかなか私と似てる状況の例文がなくてここに少しでも似てる悩みがある方に役に立てれば…と思って記録して残します。

2. 構成

前提条件

• Cloudwatch Logsを使う
• FILTER機能はすでに別処理で動いてる
• 必ずリアルタイムでキャッチアップする必要はない
  • １日１回のアラームぐらいでも構わない

要するに、Filterがすでに他のもので機能してるので、追加でなにかしたくて今回の構成を作りました。
正直Filterをもう少し汎用的に改善して運用することもいい案になったかもしれませんが、このような使い買ってもどうかなと思います。

かなり簡単な構成になります。
やりたいことがそのまま見えると思います。

3. 実装

ベース

• Python 3.9
• cloudwatch logs insight
• boto3
  • boto3 start_query()
  • boto3 get_query_result()
• request post

この構成で使うのは主にこんな者たちがあります。
boto3で提供してくれる関数の中、Cloudwatch logs Insights のクエイルを投げ て、結果を受けてメッセージを作る感じになります。
ざっくり整理すると

1. cloudwatch logs insightのクエリを投げる
2. 投げたクエリIDを返却
3. 受けたクエリIDを元に、結果を取得し、メッセージを作成

の処理になります。
さあ、例文コードを見ましょうか

import boto3
from datetime import datetime, timedelta, timezone
import time
import json
import urllib

client = boto3.client('logs')
query = "cloudwatch logs insight query string"
log_group = 'cloudwatch logs group name'

# slack webhook
URL = 'slack incomming webhook url'
# 呼ばれた時点の日時を作成
now = datetime.now()
start = datetime(now.year, now.month, now.day, 0, 0, 0) 
end = datetime(now.year, now.month, now.day, 23, 59, 59)

def lambda_handler(event, context):
    start_query_response = client.start_query(
        logGroupName=log_group,
        startTime=int((start - timedelta(days=1)).timestamp()), # ここで、１日前を指定
        endTime=int((end - timedelta(days=1)).timestamp()), # ここで、１日前を指定
        queryString=query
    )
    
    query_id = start_query_response['queryId']
    response = None
    while response == None or response['status'] == 'Running':
        print('Waiting for query to complete ...')
        time.sleep(1)
        response = client.get_query_results(
            queryId=query_id
        )
    
    # message
    if len(response['results']) > 0:
        sendMessage = ''
        # 返却されたQueryデータを持って、メッセージを作成
        # 今回作成したSQLの結果はだいたいこんな感じ
        # response['result'][0][0]['value'] = '500 error'
        # response['result'][0][1]['value'] = '4'
        # response['result'][1][0]['value'] = '400 error'
        # response['result'][1][1]['value'] = '100'
        for items in response['results']:
            sendMessage += "Error Name" + items[0]['value'] + ": " + str(items[1]['value']) + " 件 \n"

    else:
        sendMessage = '該当エラーなし'
    
    notiMsg = {
        'text': "前日発生したアプリケーションエラーの集計 \n ```"+ sendMessage +  "``` "
    }
    send_text = json.dumps(notiMsg)
    
    # push slack テストしたいときはここをブロック
    request = urllib.request.Request(
        URL, 
        data=send_text.encode('utf-8'), 
        method="POST"
    )
    with urllib.request.urlopen(request) as response:
        response_body = response.read().decode('utf-8')
   

日付指定は、コード上は起動下地店からの前日を指定してます。
細かい時間帯を指定したい場合は、コメントを参考して調整してください。

4. 感想

いかがでしょうか？
そこまで、大変なものではないが、背景でも話したように、自分もなかなか例文が見えなくて、こんなふうにも使えますよ的な感じでの共有になります。
より良い案があれば、コメントで色々共有していただければ嬉しいです。
たまたま2023年の個人的には初記事になりました。
今年も色々勉強していきましょう。