如何通过AWS Lambda批量启动带不同初始化脚本的Windows EC2实例

核心实现逻辑 #

因为AWS EC2的单个run_instancesAPI调用中，传入的UserData会统一应用到本次启动的所有实例，因此要实现每台实例使用独立UserData，需要按单实例维度单独调用run_instances，配合并发调用保证100台实例的启动效率。

具体实现步骤 #

• 第一步：提前生成所有实例对应的独立用户名、密码凭据列表，可直接在Lambda中随机生成，也可以从事件参数、DynamoDB等存储中拉取提前生成好的凭据
• 第二步：封装单实例启动函数，每调用一次生成对应实例的专属UserData，调用run_instances启动单台实例（设置MinCount=1、MaxCount=1）
• 第三步：使用线程池并发调用单实例启动函数，避免串行调用耗时过长超出Lambda超时限制
• 第四步：添加API调用重试逻辑，规避EC2 API默认速率限制的影响

代码示例 #

#!/usr/bin/python
# -*- coding: utf-8 -*-
import os
import boto3
import random
import string
from concurrent.futures import ThreadPoolExecutor, as_completed

AMI = os.environ["AMI"]
INSTANCE_TYPE = os.environ["INSTANCE_TYPE"]
KEY_NAME = os.environ["KEY_NAME"]
SUBNET_ID = os.environ["SUBNET_ID"]
REGION = os.environ["REGION"]
# 配置启动实例数量
INSTANCE_COUNT = 100
# 配置线程池大小，可根据EC2 API限流调整
MAX_WORKERS = 10

ec2 = boto3.client("ec2", region_name=REGION, config=boto3.session.Config(retries={'max_attempts': 10, 'mode': 'standard'}))

def generate_random_password(length=12):
    """生成符合Windows密码复杂度要求的随机密码"""
    chars = string.ascii_letters + string.digits + "!@#$%^&*"
    while True:
        password = ''.join(random.choice(chars) for _ in range(length))
        # 校验复杂度：至少包含大写、小写、数字、特殊字符各一个
        if (any(c.isupper() for c in password) 
            and any(c.islower() for c in password) 
            and any(c.isdigit() for c in password) 
            and any(c in "!@#$%^&*" for c in password)):
            return password

def launch_single_instance(username, password):
    """启动单台带专属UserData的EC2实例"""
    init_script = f'''<powershell>$USERNAME="{username}"
$PASSWORD="{password}"
net user /add $USERNAME $PASSWORD
net localgroup administrators $USERNAME /add # 可选：给用户加管理员权限
</powershell>'''
    response = ec2.run_instances(
        ImageId=AMI,
        InstanceType=INSTANCE_TYPE,
        KeyName=KEY_NAME,
        SubnetId=SUBNET_ID,
        MaxCount=1,
        MinCount=1,
        InstanceInitiatedShutdownBehavior="terminate",
        UserData=init_script,
    )
    instance_id = response['Instances'][0]['InstanceId']
    # 可在这里将instance_id、username、password的映射关系写入DynamoDB持久化存储
    return instance_id, username, password

def lambda_handler(event, context):
    # 生成100台实例的凭据
    credentials = []
    for i in range(INSTANCE_COUNT):
        username = f"latchu_{i+1}"
        password = generate_random_password()
        credentials.append((username, password))
    
    # 并发启动实例
    launch_results = []
    with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
        futures = [executor.submit(launch_single_instance, uname, pwd) for uname, pwd in credentials]
        for future in as_completed(futures):
            try:
                res = future.result()
                launch_results.append(res)
            except Exception as e:
                print(f"实例启动失败: {str(e)}")
    
    print(f"成功启动{len(launch_results)}台实例")
    return launch_results

注意事项 #

• Lambda执行超时时间建议调整到30秒以上，100台实例并发启动通常10秒内即可完成
• 生成的账号密码建议持久化存储到DynamoDB等组件中，避免丢失无法登录实例
• 若需要启动的实例数量超过1000台，可配合Step Functions分批次执行，避免超出Lambda最大运行时长限制

内容的提问来源于stack exchange，提问作者Lakshminarayanan S