You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

Python boto3 EC2标签校验按联系人发告警邮件实现问题

问题修复方案

1. 以实例ID为键的嵌套异常存储字典实现

你触发字符串索引列表报错的核心原因是初始存储结构选错了类型,不需要复杂逻辑,直接构建两层嵌套字典即可:第一层键为EC2实例ID,第二层键为异常分类(缺失标签、空值、unknown值、格式非法),对应值为存储具体异常条目的列表。参考可直接复用的实现逻辑:

import boto3
from collections import defaultdict

ec2 = boto3.client('ec2')

def lambda_handler(event, context):
    # 初始化实例维度的异常存储字典,无需手动判断键是否存在
    instance_tag_issues = defaultdict(lambda: {
        "missing_tags": [],
        "empty_tag_values": [],
        "unknown_tag_values": [],
        "invalid_format_tags": []
    })
    # 初始化按联系人分组的异常存储字典
    contact_grouped_issues = defaultdict(list)

    # 拉取全量EC2实例,实例数超过1000台记得加分页逻辑
    resp = ec2.describe_instances()
    for reservation in resp['Reservations']:
        for instance in reservation['Instances']:
            instance_id = instance['InstanceId']
            # 提前将实例标签转为{标签键:标签值}的字典,简化后续取值逻辑
            tags = {t['Key']: t['Value'].strip() for t in instance.get('Tags', [])}
            # 提取contact标签值,未配置的统一归为未分组
            contact = tags.get('contact', 'unassigned')

            # 检测builder标签缺失,自动补全值为unknown
            if 'builder' not in tags:
                ec2.create_tags(
                    Resources=[instance_id],
                    Tags=[{'Key': 'builder', 'Value': 'unknown'}]
                )
                instance_tag_issues[instance_id]['missing_tags'].append({
                    "tag_key": "builder",
                    "desc": "builder标签缺失,已自动补全值为unknown"
                })
            
            # 扩展其他标签校验逻辑
            required_tags = ['builder', 'contact', 'env']
            for tag_key in required_tags:
                if tag_key not in tags:
                    instance_tag_issues[instance_id]['missing_tags'].append({
                        "tag_key": tag_key,
                        "desc": f"缺失必填标签{tag_key}"
                    })
                    continue
                tag_val = tags[tag_key]
                if not tag_val:
                    instance_tag_issues[instance_id]['empty_tag_values'].append({
                        "tag_key": tag_key,
                        "desc": f"标签{tag_key}值为空"
                    })
                elif tag_val.lower() == 'unknown':
                    instance_tag_issues[instance_id]['unknown_tag_values'].append({
                        "tag_key": tag_key,
                        "desc": f"标签{tag_key}值为unknown"
                    })
                # 可自行添加格式校验规则,例如env必须属于prod/test/dev集合
                # elif tag_key == 'env' and tag_val not in ['prod', 'test', 'dev']:
                #     instance_tag_issues[instance_id]['invalid_format_tags'].append(xxx)

            # 将当前实例的异常归入对应联系人分组
            if instance_tag_issues[instance_id]:
                contact_grouped_issues[contact].append({
                    "instance_id": instance_id,
                    "issue_detail": instance_tag_issues[instance_id]
                })

    # 后续遍历contact_grouped_issues即可给对应联系人发定制化通知
    return {
        "instance_issues": dict(instance_tag_issues),
        "contact_issues": dict(contact_grouped_issues)
    }

如果你不想引入defaultdict,手动判断键是否存在也能实现相同效果:

if instance_id not in instance_tag_issues:
    instance_tag_issues[instance_id] = {
        "missing_tags": [],
        "empty_tag_values": [],
        "unknown_tag_values": [],
        "invalid_format_tags": []
    }

2. contact标签提取与分组逻辑说明

  • 不要循环遍历实例的Tags列表匹配contact键,这种写法容易因为判断条件写错、键名大小写不匹配(AWS标签键大小写敏感)导致取值失败。提前把Tags列表转成键值对字典后,直接用get方法就能稳定拿到contact值,未配置标签的实例统一归到unassigned分组,避免漏发通知。
  • 每个实例的标签校验完成后,直接把实例ID和对应异常详情追加到对应contact键的列表中,最终生成的contact_grouped_issues结构完全匹配按联系人分组的需求,遍历该字典即可给每个联系人单独发送名下实例的异常汇总邮件。

注意事项

  • 确保Lambda执行角色绑定了ec2:DescribeInstances、ec2:CreateTags权限,否则拉取实例列表、自动补标签的逻辑会报错
  • 标签值校验时记得加strip()处理首尾空格,避免值为全空格的场景被漏判
  • 单账号实例数超过1000台时,需要处理describe_instances接口的分页返回,否则会拉不全实例数据

内容的提问来源于stack exchange,提问作者skyfire123

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.08.26 13:30:56