Lambda 関数を利用して EC2 Auto Scaling グループの、最小の希望する容量が 0 になったことをトリガーに SNS 通知するように実装してみた

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Principal": {
                "Service": "lambda.amazonaws.com"
            },
            "Action": "sts:AssumeRole"
        }
    ]
}

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": "logs:CreateLogGroup",
            "Resource": "arn:aws:logs:${AWS_REGION}:${ACCOUNT_ID}:*"
        },
        {
            "Effect": "Allow",
            "Action": [
                "logs:CreateLogStream",
                "logs:PutLogEvents"
            ],
            "Resource": [
                "arn:aws:logs:${AWS_REGION}:${ACCOUNT_ID}:log-group:/aws/lambda/${FUNCTION_NAME}:*"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "autoscaling:DescribeAutoScalingGroups"
            ],
            "Resource": "*"
        },
        {
            "Effect": "Allow",
            "Action": [
                "sns:Publish"
            ],
            "Resource": "arn:aws:sns:${AWS_REGION}:${ACCOUNT_ID}:*"
        }
    ]
}

import os
import boto3
import json
from datetime import datetime

def lambda_handler(event, context):
    # 環境変数から設定を取得
    asg_name = os.environ['ASG_NAME']
    sns_topic_arn = os.environ['SNS_TOPIC_ARN']

    try:
        # Auto Scalingクライアントの初期化
        autoscaling = boto3.client('autoscaling')

        # Auto Scalingグループの情報を取得
        response = autoscaling.describe_auto_scaling_groups(
            AutoScalingGroupNames=[asg_name]
        )

        # Auto Scalingグループが存在しない場合のエラーハンドリング
        if not response['AutoScalingGroups']:
            raise Exception(f"Auto Scaling Group '{asg_name}' not found")

        asg = response['AutoScalingGroups'][0]
        min_size = asg['MinSize']

        # MinSizeが0の場合、SNS通知を送信
        if min_size == 0:
            sns = boto3.client('sns')

            # 通知メッセージの作成
            message = {
                'timestamp': datetime.now().isoformat(),
                'asg_name': asg_name,
                'alert': 'Auto Scaling Group MinSize is set to 0',
                'current_min_size': min_size,
                'current_desired_capacity': asg['DesiredCapacity'],
                'current_max_size': asg['MaxSize']
            }

            # SNS通知の送信
            sns.publish(
                TopicArn=sns_topic_arn,
                Subject=f"Alert: {asg_name} MinSize is 0",
                Message=json.dumps(message, indent=2)
            )

            return {
                'statusCode': 200,
                'body': json.dumps({
                    'message': 'Alert sent successfully',
                    'asg_name': asg_name,
                    'min_size': min_size
                })
            }

        # MinSizeが0でない場合
        return {
            'statusCode': 200,
            'body': json.dumps({
                'message': 'ASG MinSize is not 0',
                'asg_name': asg_name,
                'min_size': min_size
            })
        }

    except Exception as e:
        # エラーハンドリング
        error_message = str(e)
        print(f"Error: {error_message}")

        return {
            'statusCode': 500,
            'body': json.dumps({
                'error': error_message
            })
        }