年底做资源盘点这件事我估计不少运维和研发同行都干过一台台登录控制台点开实例详情抄CPU、内存、IP、到期时间几十上百台搞下来半天时间搭进去还特别容易抄错行。后来我换了个思路直接用Python SDK批量提取实例信息脚本一跑CSV一导出几秒钟全部搞定还能顺便按标签筛出业务归属、统计各区域资源数。这篇文章不绕弯子把整个脚本怎么写、参数怎么调、坑在哪里一次性完整过一遍需要的话可以直接把代码拿走改改就能用。1. 先理清需求批量提取到底要解决什么1.1 高频场景资源盘点、成本分析、巡检报告批量化获取云实例信息听起来是一个很技术的需求但落到实际场景里其实大多时候是业务驱动的。最常见的就是资源盘点。公司账面上有多少台服务器、分布在哪几个地域、分别跑什么业务、每个实例是什么规格这些信息如果靠人工去控制台翻不仅慢而且容易漏。尤其到了年底预算复盘或者内部审计的时候运维得交一份完整的资源清单这时候批量提取脚本就是救命稻草。其次是成本分析和资源治理。超卖、浪费、闲置资源是云上普遍存在的问题但前提是你得先有一份完整准确的实例清单。脚本拉下来的数据里包含实例规格、CPU、内存、带宽、创建时间、到期时间把这些字段跟计费数据一关联哪些实例快到期了、哪些实例规格过大利用率低一眼就能看出来。还有一类是自动化巡检。比如每周定期跑一遍脚本检查是不是有实例处于Running状态但超过多少天没有登录或者有没有实例快到期了还没有设置自动续费。数据一旦能批量拿到后续的自动化判断就顺理成章了。我做这套脚本的时候手头正好是跨3个地域、两百多台ECS实例的存量项目前后对比非常明显人工操作光汇总就要大半天脚本跑下来不到1分钟。这里面的核心思路其实不复杂云平台本身提供了标准API不管你有几十台还是几千台实例本质都是把分页接口拿出来的数据拼成一张表。1.2 方案选型为什么选择Python SDK而不是控制台导出先别急着写代码方案选型这一步很值得说几句。其实阿里云控制台本身是提供导出功能的在ECS实例列表页也能点按钮导出CSV。那为什么还要写脚本原因有三个。第一控制台导出有数量上限和字段限制。实例数量一多控制台导出往往只导当前页或者导出的字段只有固定几个关键信息比如Tags、VPC私网IP不一定都带。而SDK拿的是API原始数据字段全、数量不做截断。第二控制台导出的数据不方便二次处理。导出来是扁平表格实际上一个实例的私网IP、公网IP、Tags都是嵌套结构需要自己拼装。脚本可以直接在拉取的过程中清洗格式化导出一版就是你想要的样子。第三无法自动化。控制台得靠人点脚本却可以挂在定时任务上。配合RAM子账号的最小授权让脚本定期跑数据落库形成长期趋势记录这才能真正体现批量提取的价值。再对比一下CLI工具和Python SDK。云平台的CLI比如阿里云CLI确实也能拿到实例列表而且写一条命令就能跑适合临时查一下。但如果你要做复杂的字段解析、多地域遍历、把结果写进Excel、对接内部门户CLI的二次处理能力就弱了Python脚本可以灵活操作数据结构配合pandas、openpyxl这类库做起来非常顺手。所以我最终选了Python SDK具体来说是aliyun-python-sdk-core加上aliyun-python-sdk-ecs这套组合。代码直白、资料多、排错容易适合绝大多数团队成员接手维护。2. 环境准备装SDK、配权限、初始化客户端2.1 安装Python SDK开发环境建议用Python 3.6以上的版本我用的是3.9目前跑得很稳。安装命令很简单直接pip装两个包就行pip install aliyun-python-sdk-core pip install aliyun-python-sdk-ecs如果你想用新版SDK方式阿里云现在主推的是alibabacloud_ecs20140526底层走的是RPC风格API代码结构更贴近OpenAPI规范pip install alibabacloud_ecs20140526两套SDK都能用我这个脚本基于老版SDK写原因主要是存量代码多、网上排错案例丰富。新版SDK的API响应类型注解更完整适合新项目。如果你只是想要一个能快速跑的脚本老版完全够用后面我会把关键的区别点也提一下。2.2 创建RAM子账号与最小权限授权这一步很多人会跳过直接拿主账号的AccessKey写在脚本里。我强烈不建议这么干尤其是脚本要固化在服务器上长期跑的时候主账号Key泄露等于整个云账号被脱管。正确做法是在RAM控制台建一个子账号只授予它“只读”级别的ECS权限。具体操作流程是这样的在RAM访问控制台创建用户比如叫cloudecs_reader选择“编程访问”勾选AccessKey生成方式。给这个子账号附加权限策略系统策略里找AliyunECSReadOnlyAccess这个策略刚好覆盖ECS实例相关的只读操作权限。如果脚本还要读取其他产品信息比如磁盘、快照再按需添加对应的只读策略原则是最小够用。有个点要提醒RAM权限变更往往不是即时生效的有时候刚授权就调用API会出现Forbidden等一两分钟再重试就好了。别问我怎么知道的第一次跑脚本遇到这个错我还以为是密钥配错了。AccessKey生成后建议把ID和Secret放到环境变量里不要硬编码在脚本文件中尤其是脚本如果提交到Git仓库密钥一旦被扫描工具发现后果很严重。我自己的习惯是读~/.env或者直接用os.getenv()。2.3 初始化AcsClient配置Region老版SDK的核心入口是AcsClient初始化的时候需要三个参数AccessKey ID、AccessKey Secret、默认地域ID。from aliyunsdkcore.client import AcsClient client AcsClient( os.getenv(ALIBABA_CLOUD_ACCESS_KEY_ID), os.getenv(ALIBABA_CLOUD_ACCESS_KEY_SECRET), cn-hangzhou )这里有一个容易忽略的细节AcsClient初始化时指定的地域是“默认地域”但实际调用ECS接口时请求的地域由DescribeInstancesRequest中的RegionId参数决定。所以如果你的资源分布在多个地域单独写死一个Region是不够的需要在每个请求里显式指定RegionId这一点在后面的多Region汇总部分还会用到。另外SDK底层默认使用HTTP协议如果你对传输安全有更高要求可以开启HTTPS老版SDK在请求对象里有set_protocol_type(https)方法。公网环境里我建议开启传输过程中AccessKey不会被明文抓包。3. 核心实现流程从单页请求到批量导出3.1 最小脚本先拿到第一页数据先把最小可用的脚本写出来目标很简单请求指定地域的ECS实例列表打印返回的JSON。这一步跑通了后面就顺了。from aliyunsdkcore.client import AcsClient from aliyunsdkecs.request.v20140526.DescribeInstancesRequest import DescribeInstancesRequest import json client AcsClient( os.getenv(ALIBABA_CLOUD_ACCESS_KEY_ID), os.getenv(ALIBABA_CLOUD_ACCESS_KEY_SECRET), cn-hangzhou ) request DescribeInstancesRequest() request.set_accept_format(json) request.set_RegionId(cn-hangzhou) request.set_PageSize(10) request.set_PageNumber(1) response client.do_action_with_exception(request) data json.loads(response) print(json.dumps(data, ensure_asciiFalse, indent2))有几个点解释一下。set_accept_format(json)很关键如果不设置默认返回的是XML字符串解析起来更麻烦。do_action_with_exception是同步调用方法如果请求出错SDK会直接抛出异常我们可以捕获异常做重试或者输出错误日志。响应里的TotalCount是实例总数Instances字段下有一个Instance列表每个元素就是一台实例的所有属性。这里我建议第一次跑的时候把完整的JSON打出来看一眼你会发现字段比控制台上展示的多得多包括HostName、SerialNumber、VpcAttributes、EipAddress、Tags等这些都是控制台列表页不一定展示的数据。3.2 翻页逻辑PageNumber/PageSize 与 NextToken实例数量超过一页之后就涉及到翻页逻辑。先说一个最直观的翻页方式——PageNumber/PageSizePageSize每页返回条数范围是1~100默认值是10。PageNumber页码从1开始计数。翻页的终止条件有两种判断思路。一种是拿当前已拉取的数量跟TotalCount比较小于TotalCount就继续拉下一页另一种是判断当前返回的列表是否为空。我更推荐第一种因为哪怕最后一页恰好是空列表也能退出循环但用数量判断更加直白def fetch_all_instances(region_id, page_size100): instances [] page_number 1 while True: request DescribeInstancesRequest() request.set_accept_format(json) request.set_RegionId(region_id) request.set_PageSize(page_size) request.set_PageNumber(page_number) response client.do_action_with_exception(request) data json.loads(response) page_instances data.get(Instances, {}).get(Instance, []) instances.extend(page_instances) total int(data.get(TotalCount, 0)) if len(instances) total: break page_number 1 return instances这里有一个高频坑PageNumber从1开始而不是从0开始。如果你按照编程里数组下标从0开始的习惯写第一页数据和第二页数据会重复而且最后一页数据永远拉不到。我第一版脚本就栽在这里当时排查了很久后来打印请求参数才发现页码从一开始传错了。另外新版ECS API其实还支持NextToken配合MaxResults的分页方式。我简单说下区别NextToken方式适合深度翻页场景大数据量下性能更好不用每次从头算偏移量。老版PageNumber方式代码更好懂脚本体量不大的话完全足够。实际业务里几百台实例用PageNumber完全没压力。3.3 字段解析与实例信息清洗拿到原始数据后重点是把嵌套结构解析成扁平化的记录。我最常用的字段包括InstanceId实例ID唯一标识InstanceName实例名称Status运行状态比如Running、StoppedInstanceType实例规格Cpu/MemoryCPU核数和内存大小单位分别是核和MiBOSName操作系统名称PublicIpAddress公网IP数组VpcAttributes.PrivateIpAddress私网IP数组CreationTime/ExpiredTime创建时间和到期时间Tags.Tag标签数组这里面的嵌套结构是重点。PublicIpAddress在API响应里是一个对象对象下是IpAddress数组私网IP藏在VpcAttributes下面。解析的时候要格外小心如果某个实例是纯VPC网络且没有绑定公网IP这些字段可能是空对象直接索引取值会报KeyError。我的处理方式是统一用.get()加默认值并且把多个IP用逗号拼接成一个字符串def extract_instance_info(instance, region_id): public_ips instance.get(PublicIpAddress, {}).get(IpAddress, []) private_ips instance.get(VpcAttributes, {}).get(PrivateIpAddress, {}).get(IpAddress, []) tags [] for tag in instance.get(Tags, {}).get(Tag, []): tags.append(f{tag.get(TagKey)}:{tag.get(TagValue)}) return { region_id: region_id, instance_id: instance.get(InstanceId), instance_name: instance.get(InstanceName), status: instance.get(Status), instance_type: instance.get(InstanceType), cpu: instance.get(Cpu), memory_mib: instance.get(Memory), os_name: instance.get(OSName), public_ip: ,.join(public_ips), private_ip: ,.join(private_ips), creation_time: instance.get(CreationTime), expired_time: instance.get(ExpiredTime), tags: ;.join(tags) }这个函数是整个脚本里改动最多的部分。第一次跑完我发现导出的CSV里IP列很多是空的排查后才发现VpcAttributes里的私网IP字段路径不对。所以建议你写完解析函数后先拿一台实例完整打印JSON对照字段路径逐层确认不要靠猜。3.4 导出CSV报表一份可直接用的脚本模板数据解析完成后导出CSV是性价比最高的方案。注意Windows下用Excel打开CSV经常会出现中文乱码解决方法是写入时加encodingutf-8-sig也就是在文件开头加BOM。这一个小细节能省很多不必要的麻烦。import csv def export_to_csv(instances, filename): if not instances: print(没有拉取到任何实例不生成文件) return fieldnames list(instances[0].keys()) with open(filename, w, newline, encodingutf-8-sig) as f: writer csv.DictWriter(f, fieldnamesfieldnames) writer.writeheader() writer.writerows(instances)把上面这些函数拼起来就是一个完整的批量提取脚本import os import json import csv import time from aliyunsdkcore.client import AcsClient from aliyunsdkecs.request.v20140526.DescribeInstancesRequest import DescribeInstancesRequest ACCESS_KEY_ID os.getenv(ALIBABA_CLOUD_ACCESS_KEY_ID) ACCESS_KEY_SECRET os.getenv(ALIBABA_CLOUD_ACCESS_KEY_SECRET) client AcsClient(ACCESS_KEY_ID, ACCESS_KEY_SECRET, cn-hangzhou) def fetch_all_instances(region_id, page_size100): instances [] page_number 1 while True: request DescribeInstancesRequest() request.set_accept_format(json) request.set_RegionId(region_id) request.set_PageSize(page_size) request.set_PageNumber(page_number) response client.do_action_with_exception(request) data json.loads(response) page_instances data.get(Instances, {}).get(Instance, []) instances.extend(page_instances) total int(data.get(TotalCount, 0)) if len(instances) total: break page_number 1 return instances def extract_instance_info(instance, region_id): public_ips instance.get(PublicIpAddress, {}).get(IpAddress, []) private_ips instance.get(VpcAttributes, {}).get(PrivateIpAddress, {}).get(IpAddress, []) tags [] for tag in instance.get(Tags, {}).get(Tag, []): tags.append(f{tag.get(TagKey)}:{tag.get(TagValue)}) return { region_id: region_id, instance_id: instance.get(InstanceId), instance_name: instance.get(InstanceName), status: instance.get(Status), instance_type: instance.get(InstanceType), cpu: instance.get(Cpu), memory_mib: instance.get(Memory), os_name: instance.get(OSName), public_ip: ,.join(public_ips), private_ip: ,.join(private_ips), creation_time: instance.get(CreationTime), expired_time: instance.get(ExpiredTime), tags: ;.join(tags) } def export_to_csv(instances, filename): if not instances: print(没有拉取到任何实例不生成文件) return fieldnames list(instances[0].keys()) with open(filename, w, newline, encodingutf-8-sig) as f: writer csv.DictWriter(f, fieldnamesfieldnames) writer.writeheader() writer.writerows(instances) if __name__ __main__: region cn-hangzhou raw_instances fetch_all_instances(region) rows [extract_instance_info(ins, region) for ins in raw_instances] export_to_csv(rows, fecs_instances_{region}.csv) print(f共拉取 {len(rows)} 台实例已导出到 ecs_instances_{region}.csv)这段代码放到服务器上配好环境变量直接python fetch_ecs.py就能跑。核心逻辑不复杂但每一步都验证过实用优先。4. 实战中的坑分页遗漏、权限异常、限流重试4.1 常见问题速查表我在实际跑这个脚本的过程中遇到过不少典型问题整理成表格方便你直接对照排查。报错信息或异常现象原因分析解决办法InvalidAccessKeyId.NotFoundAccessKey ID配置错误或者子账号被禁用检查环境变量是否读取正确去RAM控制台确认Key状态ForbiddenRAM子账号没有ECS只读权限给子账号附加AliyunECSReadOnlyAccess策略等待策略生效InvalidRegionId.NotFound请求里的RegionId不存在或拼写错误对照阿里云地域列表检查比如cn-hangzhou别写成cn-hangzhou-1CSV打开中文显示乱码导出文件用了UTF-8编码Windows Excel默认按ANSI读取写入时使用utf-8-sig带BOM拉到的实例数量比实际少分页循环提前退出或者PageSize设置超过100被截断检查终止条件确认len(instances) totalPageSize最大100页码从0开始导致重复数据PageNumber应该从1开始把page_number初值改为1请求频率过高触发限流脚本循环速度快尤其是多地域场景在每次循环末尾加time.sleep(0.2)或者做指数退避重试这个表里的问题我基本都踩过。其中最有迷惑性的是分页遗漏表面上看脚本没有报错CSV也生成了数量却对不上。如果不拿TotalCount跟实际拉取数做核对很难发现。所以我建议脚本运行完以后把这个数字打印出来跟控制台实例列表页的数量做一次人工比对确认无误后再长期使用。4.2 三个实操避坑心得再说三个比较隐蔽的坑这些都是实际项目里反反复复验证过的。第一个坑是不同地域的默认VPC与交换机信息差异。有些老实例的网络类型是经典网络VpcAttributes字段的结构跟VPC实例不完全一样直接取PrivateIpAddress可能得到空值。比较稳的做法是解析的时候同时兼容经典网络和VPC网络。不过现在新购实例基本都是VPC网络这个坑主要存在于存量资源很老的环境里。第二个坑是Tags解析。有些实例没打任何标签响应里的Tags可能是空对象也可能是缺失字段直接遍历会报TypeError。写解析函数时一定要先判空再遍历不要偷懒直接取。第三个坑是RAM策略同步延迟。新创建的子账号即使立刻绑定了授权策略第一次调用API也可能报Forbidden。这不是代码问题等一两分钟再跑往往就正常了。如果等到五分钟后还报错再去排查策略有没有绑定到正确的主体。4.3 加上异常重试脚本更皮实云平台API偶尔会出现瞬时网络抖动或者后端超时尤其是拉几百上千台实例、连续发很多请求的时候。所以我在实际项目里给请求封装了重试逻辑。def do_request_with_retry(request, retry_times3, sleep_seconds1): for attempt in range(retry_times): try: response client.do_action_with_exception(request) return json.loads(response) except ServerException as e: if attempt retry_times - 1: print(f第 {attempt1} 次请求失败: {e}{sleep_seconds} 秒后重试...) time.sleep(sleep_seconds) else: raise这里是基于异常类型分类的ServerException表示服务端错误可以重试ClientException表示客户端参数错误比如AccessKey不对、RegionId不存在这类问题重试多少次都没用直接抛出去反而方便定位。如果是限流错误比如Throttling建议重试间隔从1秒逐步拉长避免雪上加霜。5. 再进一步多Region汇总、标签筛选与定时巡检5.1 多Region遍历汇总如果你的资源分布在多个地域单地域脚本就不够用了。思路很简单维护一个地域列表循环调用同一个拉取函数最后把结果合并进同一个CSV。regions [cn-hangzhou, cn-shanghai, cn-beijing, cn-shenzhen, us-west-1] all_rows [] for region in regions: try: raw_instances fetch_all_instances(region) rows [extract_instance_info(ins, region) for ins in raw_instances] all_rows.extend(rows) print(f{region}: {len(rows)} 台实例) time.sleep(0.3) except Exception as e: print(f{region} 拉取失败: {e}) continue注意循环里每次请求后加了time.sleep(0.3)控制请求频率避免触发限流。另外不同地域的实例可能数量差异非常大我见过有些地域只有两三台有些地域上百台循环里做好逐地域日志输出很有必要。万一某个地域的Key权限没配置好也能快速定位是哪一步挂掉了。5.2 按标签筛选目标实例批量提取脚本跑熟了之后下一步基本就是做精细化筛选。最常用的筛选条件是标签比如只提取Department:Marketing这个标签的实例或者只提取Environment:Production的实例。在DescribeInstancesRequest里可以直接传标签条件过滤不用在本地二次过滤request.set_TagKey(Environment) request.set_TagValue(Production)这个方式的好处是服务端直接把结果过滤掉拉取数据量和传输开销都小很多。但要注意的是SDK里一次最多只能传一组TagKey/TagValue条件如果你需要“多标签AND过滤”从服务端拉全量数据后在本地用Python过滤可能更灵活。我通常的做法是全量拉取到本地然后用列表推导式筛选这样标签多了也不怕。还有一个比较常用的本地筛选是按到期时间。把ExpiredTime字符串转成日期跟当前日期做对比定位90天内到期的实例这个信息对续费决策非常有用from datetime import datetime threshold datetime.now().replace(yeardatetime.now().year 1) expiring_soon [] for row in all_rows: try: expired_at datetime.strptime(row[expired_time], %Y-%m-%dT%H:%M:%SZ) if expired_at threshold: expiring_soon.append(row) except (ValueError, TypeError): continueExpiredTime的格式是UTC时间字符串转换的时候记得加时区偏移不要直接把UTC时间当本地时间用否则到期判断可能偏几个小时。5.3 接入定时任务与到期提醒脚本稳定后可以挂到定时任务上实现真正意义上的自动化巡检。Linux服务器上用crontab最简单比如每周一早上9点跑一次0 9 * * 1 cd /opt/ecs-inventory /usr/bin/python3 fetch_ecs.py logs/ecs_inventory.log 21跑完以后如果你希望把结果推送到内部协作工具可以在脚本里加一个发送消息的逻辑。比如统计出即将到期的实例清单后往企业微信群里发一条机器人消息或者把生成的CSV作为附件发送。这一步纯粹是业务定制但脚本的基础能力已经提供了。除了定时任务还有一个更优雅的载体是云上的定时触发函数比如函数计算配合定时触发器每个月自动跑一次资源盘点结果直接写回到对象存储连维护服务器都省了。不过这个方案涉及的产品会多一些我这里就不展开了感兴趣的可以从定时触发这个思路入手把脚本迁移到函数计算的代码包里。最后说一点个人心得这套脚本我从最早的单地域简单版一路改到现在的多地域汇总、标签筛选、到期预警版本最大的体会是云平台的API本身不难真正花时间的是理解接口的返回结构和边界条件。一次把字段路径摸透、把分页逻辑写对后面所有基于这份数据的分析都会省力很多。如果你也是第一次上手建议先拿一个小规模账号跑通单地域版本确认字段没问题再逐步扩展不要一上来就写多地域加上各种筛选调优的时候容易找不到问题出在哪个环节。另外AccessKey的保密始终是底线脚本里的密钥一律走环境变量能不用主账号Key就不用。这套东西跑起来之后你会发现资源盘点从一件苦力活变成了一件自动化的事后面再接成本分析、闲置资源识别都顺理成章。