EC2 Auto Scaling のライフサイクルフック通知が Lambda 関数へ直接ルーティング可能になっていたので試してみた

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "logs:CreateLogGroup",
                "logs:CreateLogStream",
                "logs:PutLogEvents"
            ],
            "Resource": "arn:aws:logs:*:*:*"
        },
        {
            "Sid": "AllowCompleteLifecycleAction",
            "Effect": "Allow",
            "Action": [
                "autoscaling:CompleteLifecycleAction",
                "autoscaling:RecordLifecycleActionHeartbeat"
            ],
            "Resource": "*"
        }
    ]
}

import json
import boto3

autoscaling = boto3.client("autoscaling")

def lambda_handler(event, context):
    print("Received event:")
    print(json.dumps(event))

    if event.get("Event") == "autoscaling:TEST_NOTIFICATION":
        print("Test notification received. No action required.")
        return {"status": "ok"}

    instance_id = event.get("EC2InstanceId")
    group_name = event.get("AutoScalingGroupName")
    hook_name = event.get("LifecycleHookName")
    token = event.get("LifecycleActionToken")
    transition = event.get("LifecycleTransition")

    print(f"Instance ID: {instance_id}")
    print(f"Lifecycle Transition: {transition}")

    if transition == "autoscaling:EC2_INSTANCE_LAUNCHING":
        print(f"Custom launch task for instance {instance_id}")
    elif transition == "autoscaling:EC2_INSTANCE_TERMINATING":
        print(f"Custom termination task for instance {instance_id}")
    else:
        print("Unknown transition type:", transition)

    resp = autoscaling.complete_lifecycle_action(
        LifecycleHookName=hook_name,
        AutoScalingGroupName=group_name,
        LifecycleActionToken=token,
        LifecycleActionResult="CONTINUE"
    )

    print("Lifecycle action completed:", resp)