PRTG 通过阿里云 API 获取云产品运行状况：自定义传感器与 Python 脚本实践

1. 项目缘起与整体设计思路

1.1 为什么要把云产品运行状况拉进本地监控体系

很多团队在云上跑业务，本地机房或者办公网里已经有一套成熟的监控系统，比如 PRTG。它负责盯着物理服务器、交换机、专线、UPS 这些设备，值班同事每天看的就是那一个仪表盘。但业务上云之后，云产品自带的监控面板和本地 PRTG 是割裂的：云上的 RDS 连接数飙了、SLB 后端健康检查挂了、ECS 的 CPU 打满了，本地监控大屏上完全看不出来。值班同事得先看 PRTG，再切到云控制台，来回切换，故障响应时间被拉长。

这个项目的核心目标，就是让 PRTG 通过云厂商开放的 API 接口，把云产品的运行指标拉回来，统一在 PRTG 里展示和告警。标题里提到的“通过阿里云 API 获取云产品运行状况”，本质上就是做一层数据搬运和格式转换：用脚本调用阿里云的 OpenAPI，拿到监控数据，再以 PRTG 能识别的格式喂给它。

适合谁来参考？如果你手里有 PRTG 的使用经验，懂一点 Python 或者 PowerShell，对云产品的监控指标有基本概念，那这篇内容可以直接抄作业。如果你完全没接触过 PRTG，也没关系，我会把关键概念用生活化的方式讲清楚，你照着步骤走也能跑通。

1.2 整体架构：PRTG 自定义传感器 + 中间脚本 + 云 API

PRTG 原生支持很多传感器类型，比如 SNMP、WMI、HTTP 等。要接云 API，最稳妥的方式是用PRTG 的“EXE/Script Advanced”传感器，或者“REST Custom Sensor”。前者是让 PRTG 定时执行一个本地脚本，脚本去调云 API，然后把结果按 PRTG 规定的 JSON 或 XML 格式输出；后者是让 PRTG 直接去请求一个 HTTP 接口，由中间服务返回数据。

考虑到云 API 的鉴权比较复杂，签名计算、Token 刷新这些逻辑放在脚本里更灵活，所以我选择的是EXE/Script Advanced 传感器 + Python 脚本的方案。整体数据流是这样的：

1. PRTG 核心服务按照设定的扫描间隔（比如 5 分钟）触发传感器。
2. 传感器执行指定的 Python 脚本，并把参数（如云产品实例 ID、指标名称）传进去。
3. Python 脚本调用阿里云 SDK，完成鉴权、请求、解析。
4. 脚本把拿到的数值按照 PRTG 要求的 JSON 格式打印到标准输出。
5. PRTG 解析输出，生成图表、触发告警。

这个架构的好处是解耦：PRTG 只管调度和展示，脚本只管取数和转换。后面要加新的云产品指标，只需要改脚本，不用动 PRTG 的传感器配置。

1.3 方案选型背后的几个关键考量

为什么不用 PRTG 自带的云传感器？PRTG 确实有一些云厂商的传感器，但覆盖的云产品有限，而且更新频率往往跟不上云厂商 API 的变化。自己写脚本，灵活性最高，想拉什么指标就拉什么指标。

为什么用 Python 而不是 PowerShell？阿里云的 SDK 对 Python 支持最好，文档全，示例多。PowerShell 虽然也能调 API，但签名计算那块写起来比较啰嗦。当然，如果你团队只熟悉 PowerShell，用 PowerShell 也能实现，逻辑是一样的。

为什么不让脚本直接写数据库？有些方案是脚本把数据写到中间数据库，PRTG 再去查库。这样多了一层，故障点也多了。直接让脚本输出给 PRTG，链路最短，排查问题也简单。

注意：PRTG 的 EXE/Script Advanced 传感器对脚本执行时间有限制，默认是 60 秒。如果你的脚本调 API 比较慢，需要在传感器设置里把超时时间调大，但也不建议超过 120 秒，否则会影响其他传感器的调度。

2. 核心细节解析与实操要点

2.1 阿里云 API 鉴权：AccessKey 与签名机制

调用阿里云 OpenAPI，第一步是鉴权。最常用的方式是AccessKey ID + AccessKey Secret。这对密钥相当于你的账号密码，但它是给程序用的。你可以在云控制台的访问控制里创建 RAM 用户，然后给这个用户授予只读监控数据的权限，比如AliyunCloudMonitorReadOnlyAccess。千万不要用主账号的 AccessKey，权限太大，一旦泄露风险极高。

有了 AccessKey 之后，每次请求都要做签名。阿里云的签名机制叫HMAC-SHA1，具体步骤是：

1. 把请求参数按参数名排序。
2. 拼接成规范化的查询字符串。
3. 构造待签名字符串，包括 HTTP 方法、路径、参数等。
4. 用 AccessKey Secret 作为密钥，做 HMAC-SHA1 计算。
5. 把签名结果 Base64 编码，作为 Signature 参数附加到请求里。

听起来复杂，但用官方 SDK 的话，这些步骤都是自动完成的。你只需要配置好 AccessKey，调用 SDK 的方法就行。如果你非要手写签名，也不是不行，但容易出错，调试起来很痛苦。我建议直接用 SDK。

提示：AccessKey Secret 不要硬编码在脚本里。可以放在环境变量里，或者用一个单独的配置文件，并且把文件权限设为只有运行 PRTG 服务的账号可读。PRTG 服务通常以本地系统账号运行，你需要确保这个账号能读到配置文件。

2.2 云监控 API 的关键参数：Namespace、MetricName、Dimensions

阿里云云监控（CloudMonitor）的 API 里，拉取指标的核心参数有三个：

• Namespace：命名空间，用来区分云产品。比如 ECS 是acs_ecs_dashboard，RDS 是acs_rds_dashboard，SLB 是acs_slb_dashboard。
• MetricName：指标名称。比如 ECS 的 CPU 使用率是cpu_total，内存使用率是memory_usedutilization。
• Dimensions：维度，用来指定具体的实例。格式是一个 JSON 字符串，比如[{"instanceId":"i-xxxxxxxx"}]。

这三个参数决定了你拉的是哪个产品的哪个实例的哪个指标。实际写脚本的时候，我建议把这三个参数做成脚本的入参，这样同一个脚本可以复用于多个传感器。比如：

python get_cloud_metric.py --namespace acs_ecs_dashboard --metric cpu_total --dimensions '[{"instanceId":"i-xxxxxxxx"}]'

PRTG 的 EXE/Script Advanced 传感器支持传递参数，你可以在传感器设置里把参数填进去。这样每个传感器对应一个指标，配置清晰，后期维护也方便。

2.3 PRTG 传感器的输出格式：JSON 还是 XML

PRTG 的 EXE/Script Advanced 传感器支持两种输出格式：JSON 和 XML。JSON 更简洁，我一般用 JSON。格式要求是这样的：

{ "prtg": { "result": [ { "channel": "CPU Usage", "value": 45.6, "unit": "Percent", "limitmaxwarning": 80, "limitmaxerror": 90 } ] } }

关键字段说明：

• channel：通道名称，会显示在 PRTG 的图表里。可以多个通道，比如同时返回 CPU 和内存。
• value：数值。必须是数字，不能是字符串。
• unit：单位。常用的有Percent、BytesBandwidth、Count、TimeSeconds等。
• limitmaxwarning和limitmaxerror：告警阈值。超过 warning 会显示黄色，超过 error 会显示红色。

脚本最后要把这个 JSON 打印到标准输出。注意，只能打印这个 JSON，不能有其他调试信息，否则 PRTG 解析会失败。调试信息可以写到标准错误里，或者写到日志文件。

注意：PRTG 对 JSON 的格式要求很严格。如果 value 是浮点数，不要用科学计数法。如果某个指标暂时取不到值，可以返回 0，但最好在脚本里做好异常处理，避免 PRTG 传感器变成“未知”状态。

3. 实操过程与核心环节实现

3.1 环境准备：Python 环境与阿里云 SDK 安装

首先，在 PRTG 服务器上安装 Python。建议用 Python 3.8 或以上版本。安装的时候记得勾选“Add Python to PATH”，这样在命令行里可以直接用python命令。

然后安装阿里云 SDK。阿里云的 Python SDK 叫aliyun-python-sdk-core，但云监控有单独的 SDK 包，叫aliyun-python-sdk-cms。用 pip 安装：

pip install aliyun-python-sdk-core pip install aliyun-python-sdk-cms

如果你需要拉取其他云产品的指标，可能还需要安装对应的 SDK 包。比如拉 ECS 的指标，云监控的 SDK 就够了，因为指标数据都是从云监控走的。

安装完成后，可以写一个简单的测试脚本，验证 SDK 是否能正常工作：

from aliyunsdkcore.client import AcsClient from aliyunsdkcms.request.v20190101 import DescribeMetricListRequest import json client = AcsClient('your-access-key-id', 'your-access-key-secret', 'cn-hangzhou') request = DescribeMetricListRequest.DescribeMetricListRequest() request.set_Namespace('acs_ecs_dashboard') request.set_MetricName('cpu_total') request.set_Dimensions('[{"instanceId":"i-xxxxxxxx"}]') request.set_StartTime('2024-01-01 00:00:00') request.set_EndTime('2024-01-01 00:05:00') request.set_Period('60') response = client.do_action_with_exception(request) print(response.decode('utf-8'))

如果能看到返回的 JSON 数据，说明环境没问题。

3.2 脚本编写：从 API 请求到 PRTG 输出

完整的脚本需要做几件事：解析命令行参数、调用 API、处理返回数据、格式化输出。下面是一个简化版的脚本框架：

import argparse import json import sys from aliyunsdkcore.client import AcsClient from aliyunsdkcms.request.v20190101 import DescribeMetricListRequest def get_metric(access_key_id, access_key_secret, region, namespace, metric_name, dimensions): client = AcsClient(access_key_id, access_key_secret, region) request = DescribeMetricListRequest.DescribeMetricListRequest() request.set_Namespace(namespace) request.set_MetricName(metric_name) request.set_Dimensions(dimensions) request.set_StartTime('2024-01-01 00:00:00') request.set_EndTime('2024-01-01 00:05:00') request.set_Period('60') response = client.do_action_with_exception(request) data = json.loads(response.decode('utf-8')) # 解析 Datapoints，取最后一个值 datapoints = json.loads(data['Datapoints']) if not datapoints: return None return datapoints[-1]['Average'] def main(): parser = argparse.ArgumentParser() parser.add_argument('--namespace', required=True) parser.add_argument('--metric', required=True) parser.add_argument('--dimensions', required=True) parser.add_argument('--channel', required=True) parser.add_argument('--unit', default='Percent') args = parser.parse_args() # 从环境变量读取 AccessKey access_key_id = os.environ.get('ALIYUN_ACCESS_KEY_ID') access_key_secret = os.environ.get('ALIYUN_ACCESS_KEY_SECRET') region = os.environ.get('ALIYUN_REGION', 'cn-hangzhou') value = get_metric(access_key_id, access_key_secret, region, args.namespace, args.metric, args.dimensions) if value is None: value = 0 result = { "prtg": { "result": [ { "channel": args.channel, "value": value, "unit": args.unit } ] } } print(json.dumps(result)) if __name__ == '__main__': main()

这个脚本里，AccessKey 是从环境变量读的，这样不会硬编码在代码里。PRTG 服务运行的时候，需要确保这些环境变量已经设置好。你可以在系统环境变量里设置，或者在 PRTG 传感器的“环境变量”选项里设置。

3.3 PRTG 传感器配置：参数传递与扫描间隔

在 PRTG 里添加传感器，选择“EXE/Script Advanced”。然后配置以下内容：

• Script：选择你写好的 Python 脚本。PRTG 默认的脚本目录是C:\Program Files (x86)\PRTG Network Monitor\Custom Sensors\EXEXML。把脚本放在这个目录下，PRTG 就能识别。
• Parameters：填写脚本的参数，比如--namespace acs_ecs_dashboard --metric cpu_total --dimensions [{"instanceId":"i-xxxxxxxx"}] --channel "CPU Usage" --unit Percent。
• Scanning Interval：扫描间隔。云监控的数据粒度通常是 1 分钟，所以 PRTG 的扫描间隔建议设为 5 分钟，没必要太频繁，否则 API 调用次数会很多。
• Timeout：超时时间。如果脚本执行比较慢，可以调到 120 秒。

配置完成后，传感器会开始执行。如果一切正常，你会看到通道里出现数值，并且随着时间推移生成图表。

提示：PRTG 的 EXE/Script Advanced 传感器默认会用 PRTG 服务账号执行脚本。如果脚本需要访问网络，确保这个账号有网络权限。另外，Python 的路径最好用绝对路径，避免因为环境变量问题找不到 Python。

4. 常见问题与排查技巧实录

4.1 传感器显示“未知”或“无数据”

这是最常见的问题。排查思路是这样的：

1. 先手动执行脚本。在命令行里用同样的参数执行脚本，看是否能正常输出 JSON。如果手动执行都失败，那问题在脚本本身。
2. 检查 JSON 格式。把脚本输出复制到 JSON 校验工具里，看是否合法。常见错误包括：value 是字符串而不是数字、有多余的逗号、中文引号等。
3. 检查 PRTG 的日志。PRTG 的日志在C:\ProgramData\Paessler\PRTG Network Monitor\Logs目录下，里面有传感器的执行日志，会记录脚本的标准输出和标准错误。
4. 检查权限。确保 PRTG 服务账号能读取脚本文件，能执行 Python，能访问网络。

我踩过的一个坑是：脚本里用了print输出调试信息，结果 PRTG 解析 JSON 失败。后来把调试信息改成写到sys.stderr，问题就解决了。

4.2 API 调用返回“Throttling”或“LimitExceeded”

阿里云的 API 有调用频率限制。如果你在 PRTG 里加了很多传感器，每个都调 API，很容易触发限流。解决办法有几个：

• 加大扫描间隔。比如从 1 分钟改成 5 分钟。
• 合并传感器。同一个实例的多个指标，可以放在一个脚本里一次拉取，然后输出多个通道。这样 API 调用次数就少了。
• 使用云监控的批量接口。阿里云云监控有DescribeMetricList接口，可以一次拉多个指标，但需要稍微改一下脚本逻辑。

下面是一个合并多个指标的脚本示例：

def get_multiple_metrics(access_key_id, access_key_secret, region, namespace, metric_names, dimensions): client = AcsClient(access_key_id, access_key_secret, region) results = {} for metric_name in metric_names: request = DescribeMetricListRequest.DescribeMetricListRequest() request.set_Namespace(namespace) request.set_MetricName(metric_name) request.set_Dimensions(dimensions) request.set_StartTime('2024-01-01 00:00:00') request.set_EndTime('2024-01-01 00:05:00') request.set_Period('60') response = client.do_action_with_exception(request) data = json.loads(response.decode('utf-8')) datapoints = json.loads(data['Datapoints']) if datapoints: results[metric_name] = datapoints[-1]['Average'] else: results[metric_name] = 0 return results

然后输出的时候，把多个指标放到 result 数组里：

result = { "prtg": { "result": [ {"channel": "CPU Usage", "value": results['cpu_total'], "unit": "Percent"}, {"channel": "Memory Usage", "value": results['memory_usedutilization'], "unit": "Percent"} ] } }

这样一次脚本执行，就能返回多个通道，PRTG 里也只需要一个传感器。

4.3 时间戳与时区问题

云监控 API 返回的数据点带有时间戳。如果你拉的是最近 5 分钟的数据，但服务器时区不对，可能会拉到空数据。阿里云 API 的时间参数默认是 UTC 时间，但你在脚本里写2024-01-01 00:00:00的时候，SDK 会按你本地时区解释。所以最好在脚本里明确指定时区，或者直接用 UTC 时间。

我一般是这样处理的：用 Python 的datetime模块获取当前 UTC 时间，然后往前推 5 分钟，格式化成字符串传给 API。这样不管服务器在哪个时区，拉到的都是最近的数据。

from datetime import datetime, timedelta end_time = datetime.utcnow() start_time = end_time - timedelta(minutes=5) request.set_StartTime(start_time.strftime('%Y-%m-%d %H:%M:%S')) request.set_EndTime(end_time.strftime('%Y-%m-%d %H:%M:%S'))

4.4 常见问题速查表

提示：如果你在脚本里用了第三方库，比如requests，记得在 PRTG 服务器上也安装。PRTG 服务账号可能和你的登录账号不是同一个，环境变量和 Python 包路径可能不一样。最好在脚本里用绝对路径指定 Python 解释器，比如C:\Python39\python.exe。

5. 扩展思路：从单指标到多产品统一监控

5.1 把 RDS、SLB、Redis 都拉进来

ECS 的指标拉通之后，其他云产品也是同样的套路。只需要改 Namespace、MetricName 和 Dimensions。比如：

• RDS：Namespace 是acs_rds_dashboard，常用指标有CpuUsage、MemoryUsage、ConnectionUsage。
• SLB：Namespace 是acs_slb_dashboard，常用指标有HeathyServerCount、UnhealthyServerCount、Qps。
• Redis：Namespace 是acs_kvstore，常用指标有CpuUsage、MemoryUsage、ConnectionUsage。

你可以写一个通用的脚本，把 Namespace、MetricName、Dimensions 都做成参数，然后为每个云产品实例创建一个 PRTG 传感器。这样 PRTG 里就能看到一个统一的视图，所有云产品的运行状况一目了然。

5.2 告警阈值与通知策略

PRTG 的告警机制很灵活。你可以在传感器级别设置阈值，也可以在通道级别设置。比如 CPU 使用率超过 80% 触发警告，超过 90% 触发错误。触发后，PRTG 可以通过邮件、短信、Webhook 等方式通知。

我建议把告警阈值和云监控里的告警规则保持一致。比如云监控里 RDS 的 CPU 告警阈值是 80%，那 PRTG 里也设 80%。这样两边不会出现一个告警一个不告警的尴尬情况。

另外，PRTG 支持“依赖关系”，比如某个传感器挂了，可以暂停其他相关传感器的告警，避免告警风暴。这个功能在云产品监控里很有用，比如 SLB 挂了，后端 ECS 的告警就可以暂时屏蔽。

5.3 性能优化：缓存与批量拉取

如果你的云产品实例很多，每个实例都创建一个传感器，PRTG 的调度压力会比较大。这时候可以考虑批量拉取：用一个脚本一次拉取多个实例的指标，然后输出多个通道。但 PRTG 的通道数量也有限制，太多通道图表会很难看。

另一个思路是引入缓存。比如脚本每次执行时，先把数据写到一个本地缓存文件里，PRTG 读取缓存文件。这样即使 API 暂时不可用，PRTG 也能显示最近一次的数据。但缓存的有效期要设置好，太旧的数据会误导判断。

我个人的经验是：对于核心业务实例，每个实例单独一个传感器，扫描间隔 5 分钟；对于非核心实例，可以合并到一个传感器里，扫描间隔 10 分钟。这样既能保证核心业务的监控精度，又不会给 PRTG 和云 API 太大压力。

5.4 安全加固：密钥管理与权限最小化

最后再强调一下安全。AccessKey 是敏感信息，一旦泄露，别人就能操作你的云资源。所以：

• 用 RAM 用户，不要用主账号。
• 给 RAM 用户授予最小权限，只读监控数据就够了。
• AccessKey 不要硬编码在脚本里，用环境变量或配置文件。
• 配置文件权限设为只有 PRTG 服务账号可读。
• 定期轮换 AccessKey，比如每 90 天换一次。

如果你觉得管理 AccessKey 太麻烦，也可以考虑用云厂商提供的临时凭证服务，但那个复杂度更高，适合大规模部署的场景。对于中小团队来说，RAM 用户 + 环境变量已经足够安全了。

我在实际使用中发现，PRTG 的 EXE/Script Advanced 传感器虽然灵活，但调试起来确实有点麻烦。每次改脚本都要等下一个扫描周期才能看到效果。后来我养成了一个习惯：先在命令行里把脚本调通，确保输出格式正确，再放到 PRTG 里。这样能省很多时间。另外，PRTG 的传感器日志一定要看，里面会记录脚本的标准错误输出，很多问题一看日志就明白了。