lambdaのロググループに対して特定ログが出たらアラーム起動でsns経由でlambdaがキックされる処理を作ってしまったら、ループ処理して料金で恐ろしいことに発展しました。

すぐにリソースを削除してawsに問い合わせをしましたが、興味本位でもこのような設定はしないように要注意です!

さあ、例文コードを見ましょうか

import json
import logging
import os
import requests

URL = os.environ['WEBHOOK_URL']

def lambda_handler(event, context):
    # ロギングの設定
    logger = logging.getLogger()
    logger.setLevel(logging.INFO)

    # SNSメッセージの取得
    sns_message = event['Records'][0]['Sns']['Message']
    logger.info(f"Received SNS message: {sns_message}")

    # SNSメッセージをJSONオブジェクトにパース
    alarm_message = json.loads(sns_message)

    # アラーム情報の取得
    alarm_name = alarm_message.get('AlarmName')
    if not alarm_name or alarm_name == 'Unknown Alarm':
        logger.info("No valid alarm name found. Skipping message sending.")
        return {"statusCode": 200, "body": "No valid alarm name, message skipped"}

    state = alarm_message.get('NewStateValue', 'Unknown State')
    reason = alarm_message.get('NewStateReason', 'No reason provided')

    # アラーム情報をメッセージに組み込む
    alert_message = f"CloudWatch Alert: {alarm_name}\n" \
                    f"State: {state}\n" \
                    f"Reason: {reason}"

    bot_message = {'text': alert_message}

    message_headers = {'Content-Type': 'application/json; charset=UTF-8'}

    # WebhookへPOSTリクエストを送信
    response = requests.post(
        URL,
        headers=message_headers,
        data=json.dumps(bot_message),
    )

    logger.info(f"Webhook response: {response.status_code}, {response.text}")

    return {"statusCode": 200, "body": "Message sent"}

# ローカルテストのためのコード（必要に応じて）
if __name__ == '__main__':
    test_event = {
        'Records': [
            {
                'Sns': {
                    'Message': json.dumps({
                        'AlarmName': 'Example Alarm Name',
                        'NewStateValue': 'ALARM',
                        'NewStateReason': 'Threshold Crossed: 1 datapoint [30.0] was greater than or equal to the threshold [25.0].'
                    })
                }
            }
        ]
    }
    lambda_handler(test_event, None)