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)