如何使用Boto3动态获取EC2 Linux实例的设备名变量?
动态获取Linux EC2实例设备名用于CloudWatch告警
针对CloudWatch告警中device参数硬编码的问题,提供两种动态获取设备名的方案:
方案一:通过EC2 API获取块设备映射
直接调用EC2的DescribeInstances接口,从实例的块设备映射中匹配对应挂载路径的设备名,再根据实例类型转换为Linux系统识别的设备名。
代码实现
import boto3 ec2 = boto3.client('ec2') def get_device_from_ec2_api(instance_id, target_mount_path): # 获取实例块设备映射信息 resp = ec2.describe_instances(InstanceIds=[instance_id]) block_devices = resp['Reservations'][0]['Instances'][0]['BlockDeviceMappings'] for dev in block_devices: if dev.get('MountPoint') == target_mount_path: ec2_dev_name = dev['DeviceName'] # 转换EC2虚拟设备名为Linux实际设备名 if ec2_dev_name.startswith('/dev/sd'): # PV实例:/dev/sdh → /dev/xvdh,提取不带/dev/的后缀 return ec2_dev_name.replace('/dev/sd', '/dev/xvd').split('/')[-1] elif ec2_dev_name.startswith('/dev/nvme'): # NVMe实例返回原始设备名后缀 return ec2_dev_name.split('/')[-1] else: return ec2_dev_name.split('/')[-1] return None # 在原告警代码中调用 instance_id = instance["InstanceId"] target_mount = "/mnt/elasticsearch" dynamic_device = get_device_from_ec2_api(instance_id, target_mount) if dynamic_device: cloudwatch.put_metric_alarm( AlarmName=prefix + ' - Elastic Search 磁盘可用率低于阈值 - 请扩容EBS卷', ComparisonOperator='LowerThanThreshold', EvaluationPeriods=3, MetricName='disk_used_percent', Namespace='CWAgent', Period=300, Statistic='Average', Threshold=75, AlarmActions=[snstopicarn], AlarmDescription='标准磁盘告警 - Elastic Search磁盘可用率低于配置阈值 - 请扩容EBS卷', Dimensions=[ {'Name': 'InstanceId', 'Value': instance_id}, {'Name': 'path', 'Value': target_mount}, {'Name': 'device', 'Value': dynamic_device}, # 替换为动态设备名 {'Name': 'fstype', 'Value': "ext4"} ] ) else: print(f"未找到实例{instance_id}挂载路径{target_mount}对应的设备")
方案二:通过SSM在实例上直接查询
利用AWS Systems Manager的Run Command功能,在实例上执行shell命令获取指定挂载路径对应的实际设备名,适用于所有实例类型(包括NVMe)。
代码实现
import boto3 import time ssm = boto3.client('ssm') def get_device_from_ssm(instance_id, target_mount_path): try: # 发送shell命令查询挂载设备 cmd_resp = ssm.send_command( InstanceIds=[instance_id], DocumentName='AWS-RunShellScript', Parameters={'commands': [f"df -P {target_mount_path} | awk 'NR==2 {{print $1}}'"]} ) cmd_id = cmd_resp['Command']['CommandId'] # 等待命令执行完成 time.sleep(5) # 获取命令输出 output_resp = ssm.get_command_invocation( CommandId=cmd_id, InstanceId=instance_id ) if output_resp['Status'] == 'Success': # 提取不带/dev/前缀的设备名 return output_resp['StandardOutputContent'].strip().split('/')[-1] else: print(f"命令执行失败: {output_resp['StandardErrorContent']}") return None except Exception as e: print(f"获取设备名出错: {str(e)}") return None # 在原告警代码中调用 instance_id = instance["InstanceId"] target_mount = "/mnt/elasticsearch" dynamic_device = get_device_from_ssm(instance_id, target_mount) if dynamic_device: # 替换原硬编码的device参数 cloudwatch.put_metric_alarm( AlarmName=prefix + ' - Elastic Search 磁盘可用率低于阈值 - 请扩容EBS卷', ComparisonOperator='LowerThanThreshold', EvaluationPeriods=3, MetricName='disk_used_percent', Namespace='CWAgent', Period=300, Statistic='Average', Threshold=75, AlarmActions=[snstopicarn], AlarmDescription='标准磁盘告警 - Elastic Search磁盘可用率低于配置阈值 - 请扩容EBS卷', Dimensions=[ {'Name': 'InstanceId', 'Value': instance_id}, {'Name': 'path', 'Value': target_mount}, {'Name': 'device', 'Value': dynamic_device}, {'Name': 'fstype', 'Value': "ext4"} ] ) else: print(f"无法获取实例{instance_id}挂载路径{target_mount}的设备名")
注意事项
- 方案一无需实例在线,但仅能处理非NVMe实例的设备名转换;NVMe实例的实际设备名(如
nvme0n1p1)无法通过EC2 API直接对应,需结合方案二。 - 方案二要求实例已安装SSM Agent并处于在线状态,且执行代码的IAM角色需具备
ssm:SendCommand和ssm:GetCommandInvocation权限。
内容的提问来源于stack exchange,提问作者Damian Chelverajan
相关产品推荐
相关产品推荐

