获取新启动EC2实例公网IP并通过SNS发送的代码修改需求

核心修改点： #

• 新增EC2客户端查询实例信息、SNS客户端发送通知
• 调整ASG容量后，等待新实例完成初始化并进入running状态
• 通过指定标签筛选目标实例，提取其公网IP
• 调用SNS主题发送包含公网IP的通知

完整修改代码： #

import boto3
import time

session = boto3.session.Session()
asg_client = session.client('autoscaling')
ec2_client = session.client('ec2')
sns_client = session.client('sns')

# 需根据实际情况修改以下参数
ASG_NAME = "test"
DESIRED_CAPACITY = 1
TARGET_TAG_KEY = "你的标签键"  # 替换为启动模板自动添加的标签键
TARGET_TAG_VALUE = "你的标签值"  # 替换为对应标签值
SNS_TOPIC_ARN = "arn:aws:sns:区域:账号ID:主题名称"  # 替换为你的SNS主题ARN
WAIT_INTERVAL = 10  # 每次查询实例状态的间隔（秒）
MAX_RETRIES = 12  # 最多等待120秒（12*10）

def set_desired_capacity(asg_name, desired_capacity):
    response = asg_client.set_desired_capacity(
        AutoScalingGroupName=asg_name,
        DesiredCapacity=desired_capacity,
    )
    return response

def get_asg_instances(asg_name):
    # 获取ASG下的所有实例ID
    response = asg_client.describe_auto_scaling_groups(AutoScalingGroupNames=[asg_name])
    if not response['AutoScalingGroups']:
        return []
    return [instance['InstanceId'] for instance in response['AutoScalingGroups'][0]['Instances']]

def get_running_instance_public_ip(instance_ids, tag_key, tag_value):
    # 筛选符合标签且处于运行状态的实例公网IP
    filters = [
        {'Name': 'instance-id', 'Values': instance_ids},
        {'Name': f'tag:{tag_key}', 'Values': [tag_value]},
        {'Name': 'instance-state-name', 'Values': ['running']}
    ]
    response = ec2_client.describe_instances(Filters=filters)
    
    public_ips = []
    for reservation in response['Reservations']:
        for instance in reservation['Instances']:
            if 'PublicIpAddress' in instance:
                public_ips.append(instance['PublicIpAddress'])
    return public_ips

def send_sns_notification(topic_arn, message):
    sns_client.publish(
        TopicArn=topic_arn,
        Message=message,
        Subject="Auto Scaling新实例公网IP通知"
    )

def lambda_handler(event, context):
    # 调整ASG期望容量
    set_desired_capacity(ASG_NAME, DESIRED_CAPACITY)
    
    # 等待新实例启动并运行
    retries = 0
    public_ips = []
    while retries < MAX_RETRIES:
        instance_ids = get_asg_instances(ASG_NAME)
        public_ips = get_running_instance_public_ip(instance_ids, TARGET_TAG_KEY, TARGET_TAG_VALUE)
        if len(public_ips) >= DESIRED_CAPACITY:
            break
        time.sleep(WAIT_INTERVAL)
        retries += 1
    
    if not public_ips:
        send_sns_notification(SNS_TOPIC_ARN, "未找到符合条件的运行中实例或公网IP")
        return {"status": "error", "message": "未获取到实例公网IP"}
    
    # 发送SNS通知
    ip_message = f"Auto Scaling组{ASG_NAME}新实例公网IP：\n" + "\n".join(public_ips)
    send_sns_notification(SNS_TOPIC_ARN, ip_message)
    
    return {"status": "success", "public_ips": public_ips}

if __name__ == '__main__':
    print(lambda_handler("", ""))

关键说明： #

1. 参数配置：必须替换代码中TARGET_TAG_KEY、TARGET_TAG_VALUE、SNS_TOPIC_ARN为实际的标签信息和SNS主题ARN
2. 等待逻辑：通过循环查询实例状态，避免因实例未启动完成导致无法获取IP，可根据实例启动速度调整WAIT_INTERVAL和MAX_RETRIES
3. 标签筛选：利用EC2的describe_instances接口结合标签和实例状态筛选，确保只获取目标实例的公网IP
4. 权限要求：Lambda执行角色需要具备autoscaling:SetDesiredCapacity、autoscaling:DescribeAutoScalingGroups、ec2:DescribeInstances、sns:Publish的权限

内容的提问来源于stack exchange，提问作者charles uneze