当前位置：首页>python>【Python 监控】6个脚本打造Prometheus自定义Exporter

【Python 监控】6个脚本打造Prometheus自定义Exporter

2026-09-02 17:04:19

适用环境：Python 3.8+ / Prometheus 2.x / Grafana 9.x。当现成的 Exporter 覆盖不了你的业务指标，自己动手写一个才是最快的路。

为什么你需要自定义 Exporter？

Prometheus 生态里有几百个现成的 Exporter：Node Exporter 采系统指标，MySQL Exporter 采数据库，Windows Exporter 采 Windows 性能计数器。大多数场景下，装一个就能用。

但运维现场总有这些情况：老板要看业务订单量随时间变化的曲线，DBA 想监控慢查询 TOP10 的实时趋势，网络组需要统计每台交换机的端口错误包。这些指标没有任何现成 Exporter 能采集——因为它们是你们公司独有的业务逻辑。

这时候，自定义 Exporter 就是唯一解。

好消息是：用 Python 写一个 Prometheus Exporter 非常简单。核心就是一个 HTTP 服务，在 /metrics 端点返回符合 Prometheus 格式的纯文本。整个流程用一张表就能说清楚：

步骤	做什么	关键工具
1. 装依赖	安装 prometheus_client	pip install prometheus_client
2. 定义指标	选择 Gauge/Counter/Histogram	Gauge / Counter / Summary
3. 采集数据	调用 API / 读文件 / 执行命令	requests / subprocess / psutil
4. 暴露端点	启动 HTTP 服务返回 /metrics	start_http_server(9100)
5. 配置抓取	Prometheus 添加 scrape target	prometheus.yml
6. 可视化	Grafana 建仪表盘 + 告警	Grafana Dashboard + AlertManager

下面6个脚本，从最基础的系统指标采集到业务级别的自定义监控，逐步递进。每个脚本都可以直接拿去用，改改参数就能跑。

脚本一：最简 Exporter — 30 行代码跑起来

先不管什么复杂场景，把一个能跑的 Exporter 搭起来再说。这个脚本采集当前进程的 CPU 和内存使用率，暴露到 9100 端口。

# exporter_basic.py — 最简 Prometheus Exporter
import time, psutil, os
from prometheus_client import Gauge, start_http_server

# 定义两个 Gauge 指标
CPU_GAUGE = Gauge("process_cpu_percent", "当前进程CPU使用率")
MEM_GAUGE = Gauge("process_memory_mb", "当前进程内存占用MB")

def collect():
    proc = psutil.Process(os.getpid())
    CPU_GAUGE.set(proc.cpu_percent(interval=1))
    MEM_GAUGE.set(proc.memory_info().rss / 1024 / 1024)

if __name__ == "__main__":
    start_http_server(9100) # 暴露 /metrics 端点
    print("Exporter running on :9100/metrics")
    while True:
        collect()
        time.sleep(10) # 每10秒采集一次

运行后用浏览器打开 http://localhost:9100/metrics，你会看到这样的输出：

# HELP process_cpu_percent 当前进程CPU使用率
# TYPE process_cpu_percent gauge
process_cpu_percent 2.3
# HELP process_memory_mb 当前进程内存占用MB
# TYPE process_memory_mb gauge
process_memory_mb 45.7

就这么简单。Prometheus 会定期来拉这个端点，数据就进 TSDB 了。接下来我们在这个框架上加各种采集逻辑就行。

⚠️ 指标命名规范：Prometheus 指标名只能用小写字母、数字和下划线，格式建议为 namespace_subsystem_name_unit，比如 myapp_http_requests_total。别用中划线，别用大写。

脚本二：Windows 性能计数器 Exporter

Linux 有 Node Exporter，但 Windows 很多关键指标藏在性能计数器（Performance Counter）里。比如 IIS 并发连接数、.NET CLR GC 次数、SQL Server 批处理速率——这些数据 Windows Exporter 不一定暴露得全。

用 Python 的 win32pdh 模块可以直接读性能计数器，配合 Prometheus 暴露出来：

# exporter_win_perfmon.py — Windows性能计数器Exporter
import time, win32pdh
from prometheus_client import Gauge, start_http_server

# 定义要采集的计数器路径
COUNTERS = {
    "win_iis_current_connections": "\\Web Service(_Total)\\Current Connections",
    "win_sql_batch_requests_sec": "\\SQLServer:SQL Statistics\\Batch Requests/sec",
    "win_net_bytes_total_sec": "\\Network Interface(*)\\Bytes Total/sec",
}

GAUGES = {}
for name in COUNTERS:
    GAUGES[name] = Gauge(name, f"Windows计数器: {name}")

def query_counter(path):
    hq = win32pdh.OpenQuery()
    hc = win32pdh.AddCounter(hq, path)
    win32pdh.CollectQueryData(hq)
    time.sleep(1) # 部分计数器需要两次采样
    win32pdh.CollectQueryData(hq)
    _, val = win32pdh.GetFormattedCounterValue(hc, win32pdh.PDH_FMT_DOUBLE)
    win32pdh.CloseQuery(hq)
    return val

def collect():
    for name, path in COUNTERS.items():
        try:
            GAUGES[name].set(query_counter(path))
        except Exception as e:
            print(f"采集失败 {name}: {e}")

if __name__ == "__main__":
    start_http_server(9101)
    while True:
        collect()
        time.sleep(15)

这个脚本的关键在于 COUNTERS 字典——你想监控什么计数器，加一条路径就行。Windows 性能监视器里能看到的所有计数器，这里都能采。

如果服务器没装 pywin32，可以用 subprocess 调 typeperf 命令来替代，不过效率会低一些。

脚本三：HTTP 端点健康检查 Exporter

运维最基础的监控之一：你的服务还活着吗？响应时间正常吗？这个脚本批量检查一组 URL 的状态码和响应时间，用 Label 区分不同的服务：

# exporter_http_health.py — HTTP健康检查Exporter
import time, requests
from prometheus_client import Gauge, start_http_server

# 要监控的服务列表
TARGETS = [
    {"name": "web_frontend", "url": "https://www.example.com/health"},
    {"name": "api_backend", "url": "https://api.example.com/status"},
    {"name": "auth_service", "url": "https://auth.example.com/ping"},
]

STATUS = Gauge("http_up", "服务是否存活(1=正常)", ["service"])
LATENCY = Gauge("http_response_time_ms", "响应时间毫秒", ["service"])
STATUS_CODE = Gauge("http_status_code", "HTTP状态码", ["service"])

def check_endpoint(target):
    try:
        r = requests.get(target["url"], timeout=5)
        STATUS.labels(service=target["name"]).set(1 if r.status_code == 200 else 0)
        LATENCY.labels(service=target["name"]).set(r.elapsed.total_seconds() * 1000)
        STATUS_CODE.labels(service=target["name"]).set(r.status_code)
    except requests.RequestException:
        STATUS.labels(service=target["name"]).set(0)
        LATENCY.labels(service=target["name"]).set(-1)

if __name__ == "__main__":
    start_http_server(9102)
    while True:
        for t in TARGETS:
            check_endpoint(t)
        time.sleep(30)

注意这里用了 Label。Label 是 Prometheus 的核心能力——同一个指标名，通过不同的 Label 值区分不同的实例。在 Grafana 里你可以用 http_up{service="api_backend"} 精确过滤某一个服务。

输出的 metrics 长这样：

http_up{service="web_frontend"} 1.0
http_up{service="api_backend"} 1.0
http_up{service="auth_service"} 0.0
http_response_time_ms{service="web_frontend"} 123.5
http_response_time_ms{service="api_backend"} 45.2

脚本四：数据库慢查询监控 Exporter

DBA 最关心的事情之一：最近有没有慢查询？这个脚本连接 MySQL，统计过去 5 分钟内执行时间超过阈值的查询数量，并把 TOP3 慢查询的指纹暴露为指标：

# exporter_mysql_slow.py — MySQL慢查询Exporter
import time, pymysql
from prometheus_client import Gauge, Counter, start_http_server

DB_CONF = {"host": "127.0.0.1", "user": "monitor", "password": "secret"}
SLOW_THRESHOLD = 1.0 # 1秒以上算慢查询

SLOW_COUNT = Counter("mysql_slow_queries_total", "慢查询累计数")
SLOW_GAUGE = Gauge("mysql_slow_queries_5m", "过去5分钟慢查询数")

def collect_slow():
    conn = pymysql.connect(**DB_CONF, database="information_schema")
    cur = conn.cursor()
    cur.execute("""
        SELECT COUNT(*) FROM events_statements_summary_by_digest
        WHERE AVG_TIMER_WAIT/1000000000000 > %s
        AND LAST_SEEN > NOW() - INTERVAL 5 MINUTE
    """, (SLOW_THRESHOLD,))
    count = cur.fetchone()[0]
    SLOW_GAUGE.set(count)
    SLOW_COUNT.inc(count)
    conn.close()

if __name__ == "__main__":
    start_http_server(9103)
    while True:
        collect_slow()
        time.sleep(60)

这个脚本用 information_schema.events_statements_summary_by_digest 表来统计慢查询。它按 SQL 指纹聚合，不会因为同一条慢 SQL 执行了 100 次就报 100 个不同的指标。

如果你用的是 PostgreSQL，把查询换成 pg_stat_statements 视图就行，逻辑完全一样。

脚本五：Exporter 自监控 — 你的监控器还活着吗？

监控系统的监控，是运维的经典套娃问题。如果你的 Exporter 自己挂了，Prometheus 只会显示 target down，但你不会知道是 Exporter 进程死了还是网络不通。

这个脚本做两件事：暴露 Exporter 自身的健康指标，同时批量检查所有 Exporter 的 /metrics 端点是否正常响应：

# exporter_health.py — Exporter自监控
import time, requests, os, psutil
from prometheus_client import Gauge, Counter, start_http_server

# 所有Exporter的地址
EXPORTERS = [
    {"name": "basic", "url": "http://localhost:9100/metrics"},
    {"name": "win_perfmon", "url": "http://localhost:9101/metrics"},
    {"name": "http_health", "url": "http://localhost:9102/metrics"},
    {"name": "mysql_slow", "url": "http://localhost:9103/metrics"},
]

UP = Gauge("exporter_up", "Exporter是否存活", ["exporter"])
SCRAPE_DURATION = Gauge("exporter_scrape_duration_ms", "抓取耗时", ["exporter"])
SCRAPE_ERRORS = Counter("exporter_scrape_errors_total", "抓取失败累计", ["exporter"])
SELF_MEM = Gauge("exporter_self_memory_mb", "本进程内存MB")
SELF_CPU = Gauge("exporter_self_cpu_percent", "本进程CPU%")

def check_exporters():
    for exp in EXPORTERS:
        try:
            r = requests.get(exp["url"], timeout=3)
            UP.labels(exporter=exp["name"]).set(1 if r.status_code == 200 else 0)
            SCRAPE_DURATION.labels(exporter=exp["name"]).set(r.elapsed.total_seconds() * 1000)
        except Exception:
            UP.labels(exporter=exp["name"]).set(0)
            SCRAPE_ERRORS.labels(exporter=exp["name"]).inc()

def self_metrics():
    proc = psutil.Process(os.getpid())
    SELF_MEM.set(proc.memory_info().rss / 1024 / 1024)
    SELF_CPU.set(proc.cpu_percent(interval=0.5))

if __name__ == "__main__":
    start_http_server(9104)
    while True:
        check_exporters()
        self_metrics()
        time.sleep(15)

这个脚本的思路很直接：用一个"管家"Exporter 去检查所有其他 Exporter 的健康状态，同时暴露自己的资源使用情况。如果某个 Exporter 挂了，exporter_up{name="xxx"} 会变成 0，Grafana 就能立即告警。

⚠️ 生产环境注意：这个"管家"Exporter 本身也需要被监控。建议用 systemd 管理所有 Exporter 进程，并配置 Restart=always 自动重启。同时把管家 Exporter 的端口也加到 Prometheus 的抓取目标里。

脚本六：Prometheus 注册 + Grafana 仪表盘

Exporter 都跑起来了，最后一步是让 Prometheus 来抓取它们，然后在 Grafana 里建仪表盘。

编辑 Prometheus 配置文件 prometheus.yml，添加抓取目标：

# prometheus.yml 追加内容
scrape_configs:
  - job_name: "custom_python_exporters"
    scrape_interval: 15s
    static_configs:
      - targets:
          - "server01:9100" # 基础指标
          - "server01:9101" # Windows性能计数器
          - "server01:9102" # HTTP健康检查
          - "server01:9103" # MySQL慢查询
          - "server01:9104" # Exporter自监控

重启 Prometheus 后，打开 http://prometheus:9090/targets，确认所有 target 状态是 UP。

接下来在 Grafana 里建仪表盘。核心面板推荐这几个：

面板名称	PromQL 查询	可视化类型
服务存活状态	http_up	Stat (绿/红灯)
响应时间趋势	http_response_time_ms	Time Series
慢查询趋势	rate(mysql_slow_queries_total[5m])	Time Series + Threshold
Exporter健康	exporter_up	Stat (绿/红灯)

告警规则建议在 Prometheus 端配置，而不是在 Grafana 里。原因是 Prometheus 的告警规则可以复用 recording rules，性能更好：

groups:
  - name: custom_exporter_alerts
    rules:
      - alert: ServiceDown
        expr: http_up == 0
        for: 2m
        labels:
          severity: critical
        annotations:
          summary: "服务 {{ $labels.service }} 不可用"

配合 AlertManager，告警可以推送到钉钉、飞书、Slack 或者企业微信。具体的 AlertManager 配置不是本文重点，但核心思路就是：Exporter 产生指标 → Prometheus 抓取 → 告警规则匹配 → AlertManager 路由通知。

总结

6 个脚本，从零搭建了一套完整的自定义监控体系：基础指标、Windows 性能计数器、HTTP 健康检查、数据库慢查询、Exporter 自监控、Prometheus 注册 + Grafana 可视化。每个脚本都是独立可运行的，按需组合就行。

关键记住三点：指标命名要规范（namespace_subsystem_name_unit），采集间隔别太短（10-30 秒足够），Exporter 进程要用 systemd 管理并配置自动重启。做到这三点，你的自定义监控就能稳定跑在生产环境。

— END —

岩学晓林技术栈

本文来自网友投稿或网络内容，如有侵犯您的权益请联系我们删除，联系邮箱：596563534@qq.com 。

【Python 监控】6个脚本打造Prometheus自定义Exporter