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
相关产品推荐
相关产品推荐

