基于Prometheus监控Python批处理作业的指标选型与状态处理咨询

一、布尔型状态指标的处理（Prometheus无布尔类型） #

Prometheus不支持原生布尔值，推荐用带标签的Counter或Gauge实现，按需选择：

1. 统计历史运行次数（推荐） #

用Counter类型，给每个状态指标添加status标签（如success/failure、pass/fail），每次作业结束后对应递增：

• file_push_total{status="success"}：文件推送成功总次数
• file_push_total{status="failure"}：文件推送失败总次数
• 同理处理first_test_total、second_test_total

这种方式能统计全量历史数据，方便计算成功率（如sum(file_push_total{status="success"}) / sum(file_push_total)）。

2. 记录最新运行状态 #

如果仅需追踪最近一次作业结果，用Gauge类型，每次作业结束后设置为1（成功/通过）或0（失败/未通过）：

• file_push_status 1（成功）或0（失败）
• 同理first_test_status、second_test_status

但该方式仅能查看最新状态，无法统计历史次数，适合只关心当前结果的场景。

二、耗时指标：Summary vs Histogram选择 #

核心差异 #

• Summary：在Python脚本（客户端）计算固定分位数（默认50%、90%、99%），但不支持跨实例聚合，分位数无法自定义（除非提前配置）。
• Histogram：在Prometheus（服务端）计算分位数，需预先定义耗时桶（如0.5s、1s、2s），支持跨实例聚合，且能灵活计算任意分位数。

推荐方案 #

优先选择Histogram：

1. 若后续作业扩展多实例，Histogram的聚合能力更实用；
2. 可根据业务场景自定义耗时桶（比如测试耗时集中在1-10秒，就设置桶为[0.5, 1, 2, 5, 10, 30]）；
3. Prometheus对Histogram的查询支持更灵活，比如计算95%分位耗时。

如果作业永远单实例运行，且仅需固定分位数，Summary也可满足需求，但Histogram扩展性更强。

三、Python实现示例（结合Node Exporter Textfile Collector） #

Node Exporter的textfile collector可读取本地文件中的Prometheus格式指标，只需在Python作业结束后生成符合格式的metrics文件即可。

步骤1：安装依赖 #

pip install prometheus-client

步骤2：示例代码 #

from prometheus_client import CollectorRegistry, Counter, Histogram, write_to_textfile
import time
import os

# 初始化指标注册表
registry = CollectorRegistry()

# 1. 状态统计指标（Counter）
file_push_counter = Counter(
    'file_push_total',
    'Total count of file push operations',
    ['status'],
    registry=registry
)
first_test_counter = Counter(
    'first_test_total',
    'Total count of first test executions',
    ['status'],
    registry=registry
)
second_test_counter = Counter(
    'second_test_total',
    'Total count of second test executions',
    ['status'],
    registry=registry
)

# 2. 耗时统计指标（Histogram）
first_test_histogram = Histogram(
    'first_test_time_taken_seconds',
    'Time consumed by first test (seconds)',
    buckets=[0.5, 1, 2, 5, 10, 30],  # 自定义耗时桶
    registry=registry
)
second_test_histogram = Histogram(
    'second_test_time_taken_seconds',
    'Time consumed by second test (seconds)',
    buckets=[0.5, 1, 2, 5, 10, 30],
    registry=registry
)

def main():
    # 模拟文件推送逻辑（替换为实际代码）
    file_push_success = True
    if file_push_success:
        file_push_counter.labels(status='success').inc()
    else:
        file_push_counter.labels(status='failure').inc()

    # 模拟第一个测试逻辑
    start = time.time()
    first_test_pass = True
    duration = time.time() - start
    first_test_histogram.observe(duration)
    first_test_counter.labels(status='pass' if first_test_pass else 'fail').inc()

    # 模拟第二个测试逻辑
    start = time.time()
    second_test_pass = False
    duration = time.time() - start
    second_test_histogram.observe(duration)
    second_test_counter.labels(status='pass' if second_test_pass else 'fail').inc()

    # 写入到Node Exporter指定的textfile目录（需提前创建目录）
    write_to_textfile('/var/lib/node_exporter/textfile_collector/job_metrics.prom', registry)

if __name__ == '__main__':
    main()

步骤3：配置Node Exporter #

启动Node Exporter时添加参数，指定textfile目录：

node_exporter --collector.textfile.directory=/var/lib/node_exporter/textfile_collector/

四、常用查询示例 #

• 文件推送成功率：sum(file_push_total{status="success"}) / sum(file_push_total)
• 第一个测试的95%分位耗时：histogram_quantile(0.95, sum(rate(first_test_time_taken_seconds_bucket[1h])) by (le))
• 最近1小时第一个测试的失败次数：sum(rate(first_test_total{status="fail"}[1h]))

内容的提问来源于stack exchange，提问作者Chel MS