如何修改Python Lambda函数，排除带特定标签的EC2实例安装安全代理

问题背景 #

我配置了一个Python Lambda函数，用于EC2实例启动时自动安装安全代理，确保所有服务器具备安全监控能力。此前EC2基本持续运行，未出现问题。但为节约成本，其他管理员配置了InstanceScheduler，在非工作时间关停部分EC2并于次日重启。

不幸的是，一台OpenVPN连接器实例重启后被强制安装了安全代理，导致无法访问该实例进行更新等操作。现在需要修改Lambda函数，给这类连接器（每个VPC部署一个，Dev和QA环境的会夜间关停）添加标签，使其重启时不触发代理安装。

现有代码如下：

import json
from botocore.exceptions import ClientError

DOCUMENT_PARAMETER = '/R7Insight/Installer/DocumentName'
CLOUDWATCH_LOG_GROUP_PARAMETER = '/R7Insight/Installer/CloudWatchLogGroup'

DOCUMENT_PARAMETERS = {
   '/R7Insight/Installer/Token': 'token',
   '/R7Insight/Installer/Windows/Source': 'windowsSource',
   '/R7Insight/Installer/Linux/Source': 'linuxSource',
   '/R7Insight/Installer/Windows/WorkingDirectory': 'windowsWorkingDirectory',
   '/R7Insight/Installer/Linux/WorkingDirectory': 'linuxWorkingDirectory'
}


def lambda_handler(event, context):
   print("Received event: " + json.dumps(event, indent=2))
   instance = event['detail']['instance-id']

   print("Installing Insight Agent on Instance: {}".format(instance))

   ssm_client = boto3.client('ssm')

   result = install_insight_agent(instance, ssm_client)

   return result


def get_parameters(client, params):
   try:
       response = client.get_parameters(
           Names=params,
           WithDecryption=True
       )
   except ClientError as e:
       raise e
   else:
       print("Retrieved parameters")
       parameters = response['Parameters']
   return parameters


def get_parameter(client, param):
   try:
       response = client.get_parameter(
           Name=param,
           WithDecryption=False
       )
   except ClientError as e:
       raise e
   else:
       print("Retrieved parameter")
       parameter = response['Parameter']
   return parameter


def install_insight_agent(instance, client):
   print("Get parameters for the SSM document")
   result = get_parameters(client, list(DOCUMENT_PARAMETERS.keys()))
   parameters = {}
   for parameter in result:
       parameters[DOCUMENT_PARAMETERS[parameter['Name']]] = [parameter['Value']]

   print("Get parameter for Document name")
   result = get_parameter(client, DOCUMENT_PARAMETER)
   document = result['Value']

   print("Get parameter for CloudWatch Log Group name")
   result = get_parameter(client, CLOUDWATCH_LOG_GROUP_PARAMETER)
   cloudwatch_log_group = result['Value']

   print('Run SendCommand to document: {} for instance: {}'.format(document, instance))
   try:
       client.send_command(InstanceIds=[instance],
                           DocumentName=document,
                           Parameters=parameters,
                           MaxConcurrency='1',
                           CloudWatchOutputConfig={
                               'CloudWatchLogGroupName': cloudwatch_log_group,
                               'CloudWatchOutputEnabled': True
                           },
                           TimeoutSeconds=900
                           )
   except ClientError as e:
       raise e
   else:
       print('Issued {} for instance: {}'.format(document, instance))
   return True

解决方案 #

我们可以通过检查EC2实例标签来控制是否执行代理安装，具体修改如下：

修改后的完整代码 #

import json
import boto3
from botocore.exceptions import ClientError

DOCUMENT_PARAMETER = '/R7Insight/Installer/DocumentName'
CLOUDWATCH_LOG_GROUP_PARAMETER = '/R7Insight/Installer/CloudWatchLogGroup'

DOCUMENT_PARAMETERS = {
   '/R7Insight/Installer/Token': 'token',
   '/R7Insight/Installer/Windows/Source': 'windowsSource',
   '/R7Insight/Installer/Linux/Source': 'linuxSource',
   '/R7Insight/Installer/Windows/WorkingDirectory': 'windowsWorkingDirectory',
   '/R7Insight/Installer/Linux/WorkingDirectory': 'linuxWorkingDirectory'
}

# 定义需要跳过安装的标签键值对
SKIP_TAG_KEY = 'SkipAgentInstall'
SKIP_TAG_VALUE = 'true'


def get_instance_tags(instance_id):
    """获取指定EC2实例的标签"""
    ec2_client = boto3.client('ec2')
    try:
        response = ec2_client.describe_tags(
            Filters=[
                {'Name': 'resource-id', 'Values': [instance_id]},
                {'Name': 'key', 'Values': [SKIP_TAG_KEY]}
            ]
        )
    except ClientError as e:
        raise e
    # 检查是否存在匹配的标签
    for tag in response['Tags']:
        if tag['Value'].lower() == SKIP_TAG_VALUE.lower():
            return True
    return False


def lambda_handler(event, context):
    print("Received event: " + json.dumps(event, indent=2))
    instance = event['detail']['instance-id']

    # 先检查实例是否需要跳过安装
    if get_instance_tags(instance):
        print(f"Skipping Insight Agent installation for Instance: {instance} (matched skip tag)")
        return True

    print("Installing Insight Agent on Instance: {}".format(instance))

    ssm_client = boto3.client('ssm')
    result = install_insight_agent(instance, ssm_client)
    return result


def get_parameters(client, params):
    try:
        response = client.get_parameters(
            Names=params,
            WithDecryption=True
        )
    except ClientError as e:
        raise e
    else:
        print("Retrieved parameters")
        parameters = response['Parameters']
    return parameters


def get_parameter(client, param):
    try:
        response = client.get_parameter(
            Name=param,
            WithDecryption=False
        )
    except ClientError as e:
        raise e
    else:
        print("Retrieved parameter")
        parameter = response['Parameter']
    return parameter


def install_insight_agent(instance, client):
    print("Get parameters for the SSM document")
    result = get_parameters(client, list(DOCUMENT_PARAMETERS.keys()))
    parameters = {}
    for parameter in result:
        parameters[DOCUMENT_PARAMETERS[parameter['Name']]] = [parameter['Value']]

    print("Get parameter for Document name")
    result = get_parameter(client, DOCUMENT_PARAMETER)
    document = result['Value']

    print("Get parameter for CloudWatch Log Group name")
    result = get_parameter(client, CLOUDWATCH_LOG_GROUP_PARAMETER)
    cloudwatch_log_group = result['Value']

    print('Run SendCommand to document: {} for instance: {}'.format(document, instance))
    try:
        client.send_command(InstanceIds=[instance],
                            DocumentName=document,
                            Parameters=parameters,
                            MaxConcurrency='1',
                            CloudWatchOutputConfig={
                                'CloudWatchLogGroupName': cloudwatch_log_group,
                                'CloudWatchOutputEnabled': True
                            },
                            TimeoutSeconds=900
                            )
    except ClientError as e:
        raise e
    else:
        print('Issued {} for instance: {}'.format(document, instance))
    return True

关键修改说明 #

1. 补充boto3导入：原代码遗漏import boto3，补充后才能正常调用AWS服务客户端。
2. 新增标签检查函数：get_instance_tags通过EC2 API获取实例标签，判断是否存在SkipAgentInstall: true的匹配标签。
3. 调整执行逻辑：在lambda_handler中先执行标签检查，若符合跳过条件则直接返回，不执行后续安装流程。
4. 配置实例标签：给需要跳过安装的OpenVPN实例添加标签SkipAgentInstall=true（大小写不敏感）。

权限注意事项 #

确保Lambda函数的IAM角色具备ec2:DescribeTags权限，否则无法获取实例标签信息。

内容的提问来源于stack exchange，提问作者Tyler Adams