功能说明
该示例生成 Prometheus 抓取端点所需的文本格式。每个指标在首次出现时输出 HELP 与 TYPE 行,随后逐条输出“指标名{标签} 值”的样本行。标签值按规范转义反斜杠、换行与双引号,否则含这些字符的标签会让抓取端解析失败。输出以排序遍历保证同一份数据每次生成同样的文本。
关键代码
标签以排序后的键值对作为指标的唯一身份,保证同一组标签始终映射到同一条时间序列。escape_label_value 依次替换反斜杠、换行、双引号,顺序很关键:必须最先替换反斜杠,否则后续插入的转义反斜杠会被再次转义。record 用排序后的标签元组作键递增计数,与暴露阶段的键一致。
使用注意
标签基数必须受控,把用户 id 或请求 id 放进标签会让时间序列数量爆炸并拖垮监控系统。TYPE 只可为 counter、gauge、histogram、summary 之一,直方图还需输出 _bucket、_sum、_count 系列,本例以计数器为例演示格式,扩展直方图时需补齐这些约定名称。文本格式要求数值为浮点或整数字面量,特殊值用 +Inf、NaN 表达,避免输出 Python 的字符串化结果。
代码示例
from collections import defaultdict
from typing import Iterable
def escape_label_value(value: str) -> str:
return value.replace('\\', '\\\\').replace('\n', '\\n').replace('"', '\\"')
def exposition(counters: dict[tuple[str, tuple[tuple[str, str], ...]], float]) -> str:
lines: list[str] = []
seen = set()
for (name, labels), value in sorted(counters.items()):
if name not in seen:
lines.append(f'# HELP {name} Application counter.')
lines.append(f'# TYPE {name} counter')
seen.add(name)
rendered = ','.join(f'{k}="{escape_label_value(v)}"' for k, v in labels)
lines.append(f'{name}{{{rendered}}} {value}')
return '\n'.join(lines) + '\n'
def record(counters, name: str, value: float, **labels: str) -> None:
key = (name, tuple(sorted(labels.items())))
counters[key] += value