自定义 Exporter 开发从业务指标到 Prometheus 格式的转换一、你的业务不是 HTTP 请求也不是 CPU 使用率Prometheus 自带 Exporter 全都看不懂Prometheus 自带的 Node Exporter、Blackbox Exporter、cAdvisor 能覆盖基础设施层和网络层但覆盖面到不了你这层。你想监控当前有多少 Agent 会话正在执行平均每笔推理耗时分布工具调用失败率按工具类型分组这些业务指标没有现成的 Exporter 能提供。自定义 Exporter 开发的本质是两件事定义指标模型Metric Model和暴露指标端点Metrics Endpoint。指标模型决定你监控什么——Counter 还是 Gauge带哪些 Label端点决定你怎么暴露——是 pull 模式Prometheus 主动抓取还是 push 模式Pushgateway。多数团队在写自定义 Exporter 时犯的错误不是代码写错了是模型设计错了——Label 的基数太高如把 user_id 作为 Label、指标语义不准确用 Counter 记录瞬时值、Counter 值未做单调递增保证。二、底层机制与原理剖析四种指标类型的正确使用场景Counter计数器只增不减。用于累计值——请求总数、错误总数、处理的请求字节数。关键约束Counter 值必须单调递增。如果你的服务重启了值归零Prometheus 的rate()函数会自动处理这个重置。但你不能在代码里手动把 Counter 设成 0——那会被解释为一次重置rate()会算出异常的负值。Gauge仪表可增可减。用于瞬时值——当前并发数、内存使用量、队列长度。关键区别Gauge 不累积历史信息只记录当前值。Histogram直方图把观测值分到预定义的桶中。关键桶的分界点需要提前设定且桶的数量直接影响 Prometheus 的时间序列数量每个桶 3 个 summary 指标。桶设计不好要么精度不够跨度太大要么成本太高时间序列太多。Summary摘要在客户端直接计算分位值。与 Histogram 的区别Summary 分位值可以在查询时直接取不需要histogram_quantile()函数计算。但 Summary 的 P99 不能在服务端聚合——因为分位值没有可加性。三、生产级代码实现 Agent 业务指标自定义 Exporter 设计原则 1. 使用 prometheus_client 官方库不轮子重造 2. Label 基数严格控制——不做 user_id / session_id 作为 Label 3. /metrics 端点处理 scrape 请求避免阻塞 4. Counter 只用 inc()永远不用 set()——保证单调递增 import time import threading from typing import Dict, Optional from functools import wraps from prometheus_client import ( Counter, Gauge, Histogram, Summary, CollectorRegistry, generate_latest, CONTENT_TYPE_LATEST, ) from prometheus_client.exposition import make_wsgi_app from wsgiref.simple_server import make_server # --------------------------------------------------------------------------- # 指标定义 # --------------------------------------------------------------------------- # 为什么用统一的 registry多个 registry 会各自导出 /metrics # Prometheus 只能抓取其中一个。统一 registry 确保一次 scrape 拿全全部指标。 REGISTRY CollectorRegistry(auto_describeTrue) # --- Counter: 指标 --- # 工具调用总次数按工具类型 模型分组 # Bucket 名称后缀 _total 是 Prometheus 规范不遵守会导致 Grafana 模板不兼容 AGENT_TOOL_CALLS Counter( agent_tool_calls_total, Total number of tool calls made by agents, labelnames[tool_type, model], registryREGISTRY, ) # Agent 错误总数按错误类型 AGENT_ERRORS Counter( agent_errors_total, Total number of agent execution errors, labelnames[error_type, agent_type], registryREGISTRY, ) # --- Histogram 指标 --- # 推理耗时分布 # 桶的设定根据业务观测95% 的推理在 30s 以内。 # 桶不设太细太多 time series也不设太粗分位值不准。 AGENT_INFERENCE_DURATION Histogram( agent_inference_duration_seconds, Agent inference duration distribution, labelnames[model, step], buckets[0.1, 0.5, 1.0, 2.0, 5.0, 10.0, 20.0, 30.0, 60.0, 120.0], registryREGISTRY, ) # --- Gauge 指标 --- # 当前活跃会话数 # 为什么不用 Counter rate活跃数是一个瞬时值Gauge 最准确。 AGENT_ACTIVE_SESSIONS Gauge( agent_active_sessions, Current number of active agent sessions, labelnames[agent_type], registryREGISTRY, ) # 令牌桶当前可用数用于监控限流系统 TOKEN_BUCKET_AVAILABLE Gauge( agent_token_bucket_available, Available tokens in rate limiter bucket, labelnames[agent_type, bucket_name], registryREGISTRY, ) # --------------------------------------------------------------------------- # 便捷的装饰器自动记录指标 # --------------------------------------------------------------------------- def observe_inference(model: str): 装饰器自动记录推理耗时 为什么用装饰器业务代码不需要关心 metrics 收集逻辑 加一个 observe_inference 就行——关注点分离 def decorator(func): wraps(func) def wrapper(*args, **kwargs): start time.monotonic() try: result func(*args, **kwargs) step success return result except Exception: step error raise finally: duration time.monotonic() - start AGENT_INFERENCE_DURATION.labels(modelmodel, stepstep).observe(duration) return wrapper return decorator def track_tool_call(tool_type: str, model: str): 记录工具调用 AGENT_TOOL_CALLS.labels(tool_typetool_type, modelmodel).inc() # --------------------------------------------------------------------------- # /metrics 端点WSGI 应用 # --------------------------------------------------------------------------- class MetricsServer: 轻量级 HTTP Server 暴露 /metrics 端点 为什么不依赖 Flask/FastAPI Exporter 只暴露一个端点引入 Web 框架是过度设计。 Python 标准库的 wsgiref prometheus_client 内置 WSGI App 足以胜任。 def __init__(self, port: int 9100): self.port port self._server: Optional[make_server] None self._thread: Optional[threading.Thread] None def start(self): 在独立线程中启动 HTTP Server 为什么用独立线程 主线程负责业务逻辑和指标采集不能因为 HTTP Server 的 accept() 被阻塞 app MetricsApp(REGISTRY) self._server make_server(0.0.0.0, self.port, app) self._thread threading.Thread(targetself._server.serve_forever, daemonTrue) self._thread.start() print(f[MetricsServer] Listening on :{self.port}/metrics) def stop(self): if self._server: self._server.shutdown() print([MetricsServer] Shutdown) class MetricsApp: 自定义 WSGI 应用 处理 /metrics 路径和非法请求 为什么要在应用层做路径检查 prometheus_client 内置的 make_wsgi_app 不检查路径 /anything 都会返回 metrics 数据——这样不安全且不规范 def __init__(self, registry: CollectorRegistry): self._metrics_app make_wsgi_app(registry) def __call__(self, environ, start_response): path environ.get(PATH_INFO, /) # 只处理 /metrics 路径 if path ! /metrics: status 404 Not Found headers [(Content-Type, text/plain)] start_response(status, headers) return [bNot Found. Available endpoints: /metrics] # 健康检查端点可选 if environ.get(QUERY_STRING) health: status 200 OK headers [(Content-Type, text/plain)] start_response(status, headers) return [bOK] try: return self._metrics_app(environ, start_response) except Exception as e: status 500 Internal Server Error headers [(Content-Type, text/plain)] start_response(status, headers) return [fInternal Error: {e}.encode(utf-8)] # --------------------------------------------------------------------------- # 业务代码中使用示例 # --------------------------------------------------------------------------- class AgentService: 模拟 Agent 业务逻辑展示指标埋点方式 def __init__(self): self._active_sessions 0 self._lock threading.Lock() observe_inference(modelgpt-4-turbo) def run_inference(self, prompt: str) - str: 执行推理 # 模拟推理耗时 time.sleep(0.5) return 推理结果 def start_session(self, agent_type: str): 开始一个会话——更新活跃会话 Gauge with self._lock: self._active_sessions 1 AGENT_ACTIVE_SESSIONS.labels(agent_typeagent_type).inc() def end_session(self, agent_type: str): 结束一个会话——更新活跃会话 Gauge with self._lock: if self._active_sessions 0: self._active_sessions - 1 AGENT_ACTIVE_SESSIONS.labels(agent_typeagent_type).dec() def call_tool(self, tool_type: str, model: str): 调用工具——增加 Counter try: # 工具调用的实际逻辑... track_tool_call(tool_type, model) return {status: ok} except Exception as e: AGENT_ERRORS.labels( error_typetype(e).__name__, agent_typetool_type ).inc() raise # --------------------------------------------------------------------------- # 启动入口 # --------------------------------------------------------------------------- if __name__ __main__: import signal import sys # 启动 Metrics Server server MetricsServer(port9100) server.start() # 模拟业务服务运行 agent AgentService() try: while True: agent.start_session(planner) agent.run_inference(分析数据) agent.call_tool(search, gpt-4-turbo) agent.call_tool(calculator, gpt-4-turbo) agent.end_session(planner) time.sleep(5) except KeyboardInterrupt: server.stop() sys.exit(0)四、边界分析与架构权衡Label 基数的致命陷阱每条唯一的 Label 组合创建一条新的时间序列。如果tool_type只有 10 种值影响不大但如果加了user_id这种高基数 Label在活跃用户 10 万的情况下会创建 10 万条时间序列Prometheus 单实例的时间序列上限大约在 1000 万条取决于内存大量高基数 Label 会迅速吃光内存补救用日志系统ELK/Loki存储高基数维度的数据Prometheus 只保留低基数聚合指标Push vs Pull 模式选择PullPrometheus 主动抓取适合长期运行的服务——自动发现、自动抓取PushPushgateway适合短生命周期任务——CronJob、Batch Job 完成后推送一次指标Pushgateway 的一个坑推上去的指标会永久存在直到手动删除。如果你的 Job 名字会变带时间戳会留下大量孤儿指标Exporter 高可用与性能Exporter 的 /metrics 端点应该是无状态、幂等的——每次 scrape 返回的是当时的指标状态/metrics 端点的响应时间应该在 5ms 以内。如果响应太慢如需要查数据库才能返回指标值Prometheus 的 scrape 可能超时。解决方案指标值在业务逻辑中更新inc/observe/metrics 端点只负责序列化和输出五、总结自定义 Exporter 开发的重心不在怎么写代码——prometheus_client 库已经做了 90% 的工作——而在于怎么设计指标模型。Counter 只增不减Gauge 可增可减Label 要低基数桶要精确且数量克制。指标模型设计错了后续的 Grafana 面板、告警规则、查询都得跟着改。一次把模型设计好比后续返工省至少十倍的精力。
自定义 Exporter 开发：从业务指标到 Prometheus 格式的转换
发布时间：2026/7/22 11:41:39
自定义 Exporter 开发从业务指标到 Prometheus 格式的转换一、你的业务不是 HTTP 请求也不是 CPU 使用率Prometheus 自带 Exporter 全都看不懂Prometheus 自带的 Node Exporter、Blackbox Exporter、cAdvisor 能覆盖基础设施层和网络层但覆盖面到不了你这层。你想监控当前有多少 Agent 会话正在执行平均每笔推理耗时分布工具调用失败率按工具类型分组这些业务指标没有现成的 Exporter 能提供。自定义 Exporter 开发的本质是两件事定义指标模型Metric Model和暴露指标端点Metrics Endpoint。指标模型决定你监控什么——Counter 还是 Gauge带哪些 Label端点决定你怎么暴露——是 pull 模式Prometheus 主动抓取还是 push 模式Pushgateway。多数团队在写自定义 Exporter 时犯的错误不是代码写错了是模型设计错了——Label 的基数太高如把 user_id 作为 Label、指标语义不准确用 Counter 记录瞬时值、Counter 值未做单调递增保证。二、底层机制与原理剖析四种指标类型的正确使用场景Counter计数器只增不减。用于累计值——请求总数、错误总数、处理的请求字节数。关键约束Counter 值必须单调递增。如果你的服务重启了值归零Prometheus 的rate()函数会自动处理这个重置。但你不能在代码里手动把 Counter 设成 0——那会被解释为一次重置rate()会算出异常的负值。Gauge仪表可增可减。用于瞬时值——当前并发数、内存使用量、队列长度。关键区别Gauge 不累积历史信息只记录当前值。Histogram直方图把观测值分到预定义的桶中。关键桶的分界点需要提前设定且桶的数量直接影响 Prometheus 的时间序列数量每个桶 3 个 summary 指标。桶设计不好要么精度不够跨度太大要么成本太高时间序列太多。Summary摘要在客户端直接计算分位值。与 Histogram 的区别Summary 分位值可以在查询时直接取不需要histogram_quantile()函数计算。但 Summary 的 P99 不能在服务端聚合——因为分位值没有可加性。三、生产级代码实现 Agent 业务指标自定义 Exporter 设计原则 1. 使用 prometheus_client 官方库不轮子重造 2. Label 基数严格控制——不做 user_id / session_id 作为 Label 3. /metrics 端点处理 scrape 请求避免阻塞 4. Counter 只用 inc()永远不用 set()——保证单调递增 import time import threading from typing import Dict, Optional from functools import wraps from prometheus_client import ( Counter, Gauge, Histogram, Summary, CollectorRegistry, generate_latest, CONTENT_TYPE_LATEST, ) from prometheus_client.exposition import make_wsgi_app from wsgiref.simple_server import make_server # --------------------------------------------------------------------------- # 指标定义 # --------------------------------------------------------------------------- # 为什么用统一的 registry多个 registry 会各自导出 /metrics # Prometheus 只能抓取其中一个。统一 registry 确保一次 scrape 拿全全部指标。 REGISTRY CollectorRegistry(auto_describeTrue) # --- Counter: 指标 --- # 工具调用总次数按工具类型 模型分组 # Bucket 名称后缀 _total 是 Prometheus 规范不遵守会导致 Grafana 模板不兼容 AGENT_TOOL_CALLS Counter( agent_tool_calls_total, Total number of tool calls made by agents, labelnames[tool_type, model], registryREGISTRY, ) # Agent 错误总数按错误类型 AGENT_ERRORS Counter( agent_errors_total, Total number of agent execution errors, labelnames[error_type, agent_type], registryREGISTRY, ) # --- Histogram 指标 --- # 推理耗时分布 # 桶的设定根据业务观测95% 的推理在 30s 以内。 # 桶不设太细太多 time series也不设太粗分位值不准。 AGENT_INFERENCE_DURATION Histogram( agent_inference_duration_seconds, Agent inference duration distribution, labelnames[model, step], buckets[0.1, 0.5, 1.0, 2.0, 5.0, 10.0, 20.0, 30.0, 60.0, 120.0], registryREGISTRY, ) # --- Gauge 指标 --- # 当前活跃会话数 # 为什么不用 Counter rate活跃数是一个瞬时值Gauge 最准确。 AGENT_ACTIVE_SESSIONS Gauge( agent_active_sessions, Current number of active agent sessions, labelnames[agent_type], registryREGISTRY, ) # 令牌桶当前可用数用于监控限流系统 TOKEN_BUCKET_AVAILABLE Gauge( agent_token_bucket_available, Available tokens in rate limiter bucket, labelnames[agent_type, bucket_name], registryREGISTRY, ) # --------------------------------------------------------------------------- # 便捷的装饰器自动记录指标 # --------------------------------------------------------------------------- def observe_inference(model: str): 装饰器自动记录推理耗时 为什么用装饰器业务代码不需要关心 metrics 收集逻辑 加一个 observe_inference 就行——关注点分离 def decorator(func): wraps(func) def wrapper(*args, **kwargs): start time.monotonic() try: result func(*args, **kwargs) step success return result except Exception: step error raise finally: duration time.monotonic() - start AGENT_INFERENCE_DURATION.labels(modelmodel, stepstep).observe(duration) return wrapper return decorator def track_tool_call(tool_type: str, model: str): 记录工具调用 AGENT_TOOL_CALLS.labels(tool_typetool_type, modelmodel).inc() # --------------------------------------------------------------------------- # /metrics 端点WSGI 应用 # --------------------------------------------------------------------------- class MetricsServer: 轻量级 HTTP Server 暴露 /metrics 端点 为什么不依赖 Flask/FastAPI Exporter 只暴露一个端点引入 Web 框架是过度设计。 Python 标准库的 wsgiref prometheus_client 内置 WSGI App 足以胜任。 def __init__(self, port: int 9100): self.port port self._server: Optional[make_server] None self._thread: Optional[threading.Thread] None def start(self): 在独立线程中启动 HTTP Server 为什么用独立线程 主线程负责业务逻辑和指标采集不能因为 HTTP Server 的 accept() 被阻塞 app MetricsApp(REGISTRY) self._server make_server(0.0.0.0, self.port, app) self._thread threading.Thread(targetself._server.serve_forever, daemonTrue) self._thread.start() print(f[MetricsServer] Listening on :{self.port}/metrics) def stop(self): if self._server: self._server.shutdown() print([MetricsServer] Shutdown) class MetricsApp: 自定义 WSGI 应用 处理 /metrics 路径和非法请求 为什么要在应用层做路径检查 prometheus_client 内置的 make_wsgi_app 不检查路径 /anything 都会返回 metrics 数据——这样不安全且不规范 def __init__(self, registry: CollectorRegistry): self._metrics_app make_wsgi_app(registry) def __call__(self, environ, start_response): path environ.get(PATH_INFO, /) # 只处理 /metrics 路径 if path ! /metrics: status 404 Not Found headers [(Content-Type, text/plain)] start_response(status, headers) return [bNot Found. Available endpoints: /metrics] # 健康检查端点可选 if environ.get(QUERY_STRING) health: status 200 OK headers [(Content-Type, text/plain)] start_response(status, headers) return [bOK] try: return self._metrics_app(environ, start_response) except Exception as e: status 500 Internal Server Error headers [(Content-Type, text/plain)] start_response(status, headers) return [fInternal Error: {e}.encode(utf-8)] # --------------------------------------------------------------------------- # 业务代码中使用示例 # --------------------------------------------------------------------------- class AgentService: 模拟 Agent 业务逻辑展示指标埋点方式 def __init__(self): self._active_sessions 0 self._lock threading.Lock() observe_inference(modelgpt-4-turbo) def run_inference(self, prompt: str) - str: 执行推理 # 模拟推理耗时 time.sleep(0.5) return 推理结果 def start_session(self, agent_type: str): 开始一个会话——更新活跃会话 Gauge with self._lock: self._active_sessions 1 AGENT_ACTIVE_SESSIONS.labels(agent_typeagent_type).inc() def end_session(self, agent_type: str): 结束一个会话——更新活跃会话 Gauge with self._lock: if self._active_sessions 0: self._active_sessions - 1 AGENT_ACTIVE_SESSIONS.labels(agent_typeagent_type).dec() def call_tool(self, tool_type: str, model: str): 调用工具——增加 Counter try: # 工具调用的实际逻辑... track_tool_call(tool_type, model) return {status: ok} except Exception as e: AGENT_ERRORS.labels( error_typetype(e).__name__, agent_typetool_type ).inc() raise # --------------------------------------------------------------------------- # 启动入口 # --------------------------------------------------------------------------- if __name__ __main__: import signal import sys # 启动 Metrics Server server MetricsServer(port9100) server.start() # 模拟业务服务运行 agent AgentService() try: while True: agent.start_session(planner) agent.run_inference(分析数据) agent.call_tool(search, gpt-4-turbo) agent.call_tool(calculator, gpt-4-turbo) agent.end_session(planner) time.sleep(5) except KeyboardInterrupt: server.stop() sys.exit(0)四、边界分析与架构权衡Label 基数的致命陷阱每条唯一的 Label 组合创建一条新的时间序列。如果tool_type只有 10 种值影响不大但如果加了user_id这种高基数 Label在活跃用户 10 万的情况下会创建 10 万条时间序列Prometheus 单实例的时间序列上限大约在 1000 万条取决于内存大量高基数 Label 会迅速吃光内存补救用日志系统ELK/Loki存储高基数维度的数据Prometheus 只保留低基数聚合指标Push vs Pull 模式选择PullPrometheus 主动抓取适合长期运行的服务——自动发现、自动抓取PushPushgateway适合短生命周期任务——CronJob、Batch Job 完成后推送一次指标Pushgateway 的一个坑推上去的指标会永久存在直到手动删除。如果你的 Job 名字会变带时间戳会留下大量孤儿指标Exporter 高可用与性能Exporter 的 /metrics 端点应该是无状态、幂等的——每次 scrape 返回的是当时的指标状态/metrics 端点的响应时间应该在 5ms 以内。如果响应太慢如需要查数据库才能返回指标值Prometheus 的 scrape 可能超时。解决方案指标值在业务逻辑中更新inc/observe/metrics 端点只负责序列化和输出五、总结自定义 Exporter 开发的重心不在怎么写代码——prometheus_client 库已经做了 90% 的工作——而在于怎么设计指标模型。Counter 只增不减Gauge 可增可减Label 要低基数桶要精确且数量克制。指标模型设计错了后续的 Grafana 面板、告警规则、查询都得跟着改。一次把模型设计好比后续返工省至少十倍的精力。
详解与实战配置)
