多维指标监控体系搭建PrometheusGrafana的落地实践一、从服务活着吗到服务健康吗监控有三个层次存活性监控进程是否 alive→技术指标监控CPU/内存/QPS/延迟→业务指标监控评测成功率/用户留存/推荐点击率。多数团队停留在第一层少部分到了第二层极少数能把第三层落地。这套多维指标监控体系的目标是同一套架构覆盖三个层次让每种角色都能在 Grafana 上找到自己需要的看板。flowchart TB A[应用服务] --|埋点 SDK| B[Metrics Exporter] A --|健康探测| C[Health Prober] B -- D[Prometheus 拉取/推送] C -- D D -- E[AlertManager 告警规则] E -- F{告警类型} F --|Critical| G[PagerDuty/电话] F --|Warning| H[企业微信/钉钉] F --|Info| I[记录到日志] D -- J[Grafana 可视化] J -- K[运维看板: 集群健康] J -- L[开发看板: 接口延迟] J -- M[产品看板: 业务指标] style D fill:#ccf style J fill:#cfc二、四个黄金信号与指标设计Google SRE 提出的四个黄金信号是监控的最小可行指标信号含义关键指标Latency请求延迟P50/P90/P99 延迟Traffic请求量QPS/RPSErrors错误率5xx 比例、业务错误率Saturation资源饱和度CPU/内存/连接池/队列深度对于刷题系统分级指标设计如下L1 存活性up{jobjudge-worker}、process_resident_memory_bytesL2 技术指标http_request_duration_seconds、db_connection_active、gc_pause_secondsL3 业务指标judge_success_rate、recommend_click_rate、avg_solving_time_minutes三、监控体系的核心实现 监控指标埋点 SDK 实现 Prometheus 风格的 Counter/Gauge/Histogram import time import threading from typing import Dict, List, Optional, Callable from dataclasses import dataclass, field from collections import defaultdict class MetricType: COUNTER counter GAUGE gauge HISTOGRAM histogram dataclass class MetricLabel: 指标标签 name: str value: str class Counter: 计数器只增不减的累计值 def __init__(self, name: str, help: str, labels: Optional[Dict[str, str]] None): self.name name self.help help self.labels labels or {} self._value: float 0.0 self._lock threading.Lock() def inc(self, amount: float 1.0): 增加计数值 with self._lock: self._value amount def get(self) - float: with self._lock: return self._value class Gauge: 仪表盘可增可减的瞬时值 def __init__(self, name: str, help: str, labels: Optional[Dict[str, str]] None): self.name name self.help help self.labels labels or {} self._value: float 0.0 self._lock threading.Lock() def set(self, value: float): 设置当前值 with self._lock: self._value value def inc(self, amount: float 1.0): with self._lock: self._value amount def dec(self, amount: float 1.0): with self._lock: self._value - amount def get(self) - float: with self._lock: return self._value class Histogram: 直方图记录值的分布 def __init__(self, name: str, help: str, buckets: List[float], labels: Optional[Dict[str, str]] None): self.name name self.help help self.buckets sorted(buckets) self.labels labels or {} self._bucket_counts: Dict[float, int] defaultdict(int) self._sum: float 0.0 self._count: int 0 self._lock threading.Lock() def observe(self, value: float): 观察一个值 with self._lock: self._sum value self._count 1 for bound in self.buckets: if value bound: self._bucket_counts[bound] 1 break def get_stats(self) - Dict: 获取统计信息 with self._lock: return { count: self._count, sum: self._sum, buckets: dict(self._bucket_counts), } class MetricsRegistry: 指标注册中心单例 _instance None _lock threading.Lock() def __new__(cls): if cls._instance is None: with cls._lock: if cls._instance is None: cls._instance super().__new__(cls) cls._instance._metrics: Dict[str, object] {} return cls._instance def register(self, metric): 注册指标 self._metrics[metric.name] metric def get(self, name: str): return self._metrics.get(name) def export_prometheus_format(self) - str: 导出为 Prometheus 文本格式 lines [] for name, metric in self._metrics.items(): # HELP 和 TYPE 行 lines.append(f# HELP {name} {getattr(metric, help, )}) if isinstance(metric, Counter): lines.append(f# TYPE {name} counter) lines.append(f{name}{_format_labels(metric.labels)} f{metric.get()}) elif isinstance(metric, Gauge): lines.append(f# TYPE {name} gauge) lines.append(f{name}{_format_labels(metric.labels)} f{metric.get()}) elif isinstance(metric, Histogram): lines.append(f# TYPE {name} histogram) stats metric.get_stats() for bound, count in sorted(stats[buckets].items()): lines.append( f{name}_bucket{{le{bound}}} f {count} ) lines.append(f{name}_bucket{{le\Inf\}} f{stats[count]}) lines.append(f{name}_sum {stats[sum]}) lines.append(f{name}_count {stats[count]}) return \n.join(lines) \n def _format_labels(labels: Dict[str, str]) - str: 格式化 Prometheus labels if not labels: return parts [f{k}{v} for k, v in labels.items()] return { ,.join(parts) } # 全局注册中心 registry MetricsRegistry() # 预定义指标 http_requests_total Counter( http_requests_total, HTTP 请求总数, labels{service: api-gateway}, ) registry.register(http_requests_total) active_db_connections Gauge( db_connections_active, 活跃数据库连接数, ) registry.register(active_db_connections) request_duration_seconds Histogram( http_request_duration_seconds, HTTP 请求耗时分布, buckets[0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0], ) registry.register(request_duration_seconds) judge_success_rate Gauge( judge_success_rate, 评测成功率, ) registry.register(judge_success_rate) # 常用的上下文管理器自动记录请求耗时 class RequestTimer: 自动计时并记录到 Histogram def __init__(self, histogram: Histogram): self.histogram histogram self._start 0.0 def __enter__(self): self._start time.time() return self def __exit__(self, *args): duration time.time() - self._start self.histogram.observe(duration) # AlertManager 告警规则YAML 格式示例 ALERT_RULES groups: - name: api_alerts rules: - alert: HighErrorRate expr: rate(http_requests_total{status500}[5m]) 0.05 for: 2m labels: severity: critical annotations: summary: API 错误率超过 5% - alert: HighLatency expr: histogram_quantile(0.99, rate(http_request_duration_seconds_bucket[5m])) 1.0 for: 5m labels: severity: warning annotations: summary: P99 延迟超过 1 秒 - alert: LowSuccessRate expr: judge_success_rate 0.9 for: 5m labels: severity: critical annotations: summary: 评测成功率低于 90% if __name__ __main__: with RequestTimer(request_duration_seconds): time.sleep(0.05) http_requests_total.inc() active_db_connections.set(8) judge_success_rate.set(0.95) print(registry.export_prometheus_format()) print(ALERT_RULES)四、生产级监控的工程决策指标基数控制每个接口 × 每个状态码 × 每个服务 爆炸的指标数量。用relabel_config在 Prometheus 侧过滤低价值标签组合。存储成本优化30 天的高精度数据15s 采集间隔 1 年的聚合数据5min 采集间隔。用 Thanos/Cortex 做长期存储降采样。告警收敛同一个故障可能触发 10 条告警。设置告警分组规则按服务/按时间窗口避免告警风暴。五、总结四个黄金信号是最小可行集Latency/Traffic/Errors/Saturation先覆盖这四个再扩展。分层次监控L1 存活性、L2 技术指标、L3 业务指标各层看板独立但数据互通。Histogram 优于 SummaryHistogram 可以在 Prometheus 侧做任意分位计算Summary 做不到。告警规则要审慎告警太敏感→告警疲劳太迟钝→故障发现延迟。本文实现了一个 Prometheus 兼容的指标 SDKCounter/Gauge/Histogram包含 Prometheus 导出格式和 AlertManager 告警规则模板可直接集成到后端服务中。
多维指标监控体系搭建：Prometheus+Grafana的落地实践
多维指标监控体系搭建PrometheusGrafana的落地实践一、从服务活着吗到服务健康吗监控有三个层次存活性监控进程是否 alive→技术指标监控CPU/内存/QPS/延迟→业务指标监控评测成功率/用户留存/推荐点击率。多数团队停留在第一层少部分到了第二层极少数能把第三层落地。这套多维指标监控体系的目标是同一套架构覆盖三个层次让每种角色都能在 Grafana 上找到自己需要的看板。flowchart TB A[应用服务] --|埋点 SDK| B[Metrics Exporter] A --|健康探测| C[Health Prober] B -- D[Prometheus 拉取/推送] C -- D D -- E[AlertManager 告警规则] E -- F{告警类型} F --|Critical| G[PagerDuty/电话] F --|Warning| H[企业微信/钉钉] F --|Info| I[记录到日志] D -- J[Grafana 可视化] J -- K[运维看板: 集群健康] J -- L[开发看板: 接口延迟] J -- M[产品看板: 业务指标] style D fill:#ccf style J fill:#cfc二、四个黄金信号与指标设计Google SRE 提出的四个黄金信号是监控的最小可行指标信号含义关键指标Latency请求延迟P50/P90/P99 延迟Traffic请求量QPS/RPSErrors错误率5xx 比例、业务错误率Saturation资源饱和度CPU/内存/连接池/队列深度对于刷题系统分级指标设计如下L1 存活性up{jobjudge-worker}、process_resident_memory_bytesL2 技术指标http_request_duration_seconds、db_connection_active、gc_pause_secondsL3 业务指标judge_success_rate、recommend_click_rate、avg_solving_time_minutes三、监控体系的核心实现 监控指标埋点 SDK 实现 Prometheus 风格的 Counter/Gauge/Histogram import time import threading from typing import Dict, List, Optional, Callable from dataclasses import dataclass, field from collections import defaultdict class MetricType: COUNTER counter GAUGE gauge HISTOGRAM histogram dataclass class MetricLabel: 指标标签 name: str value: str class Counter: 计数器只增不减的累计值 def __init__(self, name: str, help: str, labels: Optional[Dict[str, str]] None): self.name name self.help help self.labels labels or {} self._value: float 0.0 self._lock threading.Lock() def inc(self, amount: float 1.0): 增加计数值 with self._lock: self._value amount def get(self) - float: with self._lock: return self._value class Gauge: 仪表盘可增可减的瞬时值 def __init__(self, name: str, help: str, labels: Optional[Dict[str, str]] None): self.name name self.help help self.labels labels or {} self._value: float 0.0 self._lock threading.Lock() def set(self, value: float): 设置当前值 with self._lock: self._value value def inc(self, amount: float 1.0): with self._lock: self._value amount def dec(self, amount: float 1.0): with self._lock: self._value - amount def get(self) - float: with self._lock: return self._value class Histogram: 直方图记录值的分布 def __init__(self, name: str, help: str, buckets: List[float], labels: Optional[Dict[str, str]] None): self.name name self.help help self.buckets sorted(buckets) self.labels labels or {} self._bucket_counts: Dict[float, int] defaultdict(int) self._sum: float 0.0 self._count: int 0 self._lock threading.Lock() def observe(self, value: float): 观察一个值 with self._lock: self._sum value self._count 1 for bound in self.buckets: if value bound: self._bucket_counts[bound] 1 break def get_stats(self) - Dict: 获取统计信息 with self._lock: return { count: self._count, sum: self._sum, buckets: dict(self._bucket_counts), } class MetricsRegistry: 指标注册中心单例 _instance None _lock threading.Lock() def __new__(cls): if cls._instance is None: with cls._lock: if cls._instance is None: cls._instance super().__new__(cls) cls._instance._metrics: Dict[str, object] {} return cls._instance def register(self, metric): 注册指标 self._metrics[metric.name] metric def get(self, name: str): return self._metrics.get(name) def export_prometheus_format(self) - str: 导出为 Prometheus 文本格式 lines [] for name, metric in self._metrics.items(): # HELP 和 TYPE 行 lines.append(f# HELP {name} {getattr(metric, help, )}) if isinstance(metric, Counter): lines.append(f# TYPE {name} counter) lines.append(f{name}{_format_labels(metric.labels)} f{metric.get()}) elif isinstance(metric, Gauge): lines.append(f# TYPE {name} gauge) lines.append(f{name}{_format_labels(metric.labels)} f{metric.get()}) elif isinstance(metric, Histogram): lines.append(f# TYPE {name} histogram) stats metric.get_stats() for bound, count in sorted(stats[buckets].items()): lines.append( f{name}_bucket{{le{bound}}} f {count} ) lines.append(f{name}_bucket{{le\Inf\}} f{stats[count]}) lines.append(f{name}_sum {stats[sum]}) lines.append(f{name}_count {stats[count]}) return \n.join(lines) \n def _format_labels(labels: Dict[str, str]) - str: 格式化 Prometheus labels if not labels: return parts [f{k}{v} for k, v in labels.items()] return { ,.join(parts) } # 全局注册中心 registry MetricsRegistry() # 预定义指标 http_requests_total Counter( http_requests_total, HTTP 请求总数, labels{service: api-gateway}, ) registry.register(http_requests_total) active_db_connections Gauge( db_connections_active, 活跃数据库连接数, ) registry.register(active_db_connections) request_duration_seconds Histogram( http_request_duration_seconds, HTTP 请求耗时分布, buckets[0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0], ) registry.register(request_duration_seconds) judge_success_rate Gauge( judge_success_rate, 评测成功率, ) registry.register(judge_success_rate) # 常用的上下文管理器自动记录请求耗时 class RequestTimer: 自动计时并记录到 Histogram def __init__(self, histogram: Histogram): self.histogram histogram self._start 0.0 def __enter__(self): self._start time.time() return self def __exit__(self, *args): duration time.time() - self._start self.histogram.observe(duration) # AlertManager 告警规则YAML 格式示例 ALERT_RULES groups: - name: api_alerts rules: - alert: HighErrorRate expr: rate(http_requests_total{status500}[5m]) 0.05 for: 2m labels: severity: critical annotations: summary: API 错误率超过 5% - alert: HighLatency expr: histogram_quantile(0.99, rate(http_request_duration_seconds_bucket[5m])) 1.0 for: 5m labels: severity: warning annotations: summary: P99 延迟超过 1 秒 - alert: LowSuccessRate expr: judge_success_rate 0.9 for: 5m labels: severity: critical annotations: summary: 评测成功率低于 90% if __name__ __main__: with RequestTimer(request_duration_seconds): time.sleep(0.05) http_requests_total.inc() active_db_connections.set(8) judge_success_rate.set(0.95) print(registry.export_prometheus_format()) print(ALERT_RULES)四、生产级监控的工程决策指标基数控制每个接口 × 每个状态码 × 每个服务 爆炸的指标数量。用relabel_config在 Prometheus 侧过滤低价值标签组合。存储成本优化30 天的高精度数据15s 采集间隔 1 年的聚合数据5min 采集间隔。用 Thanos/Cortex 做长期存储降采样。告警收敛同一个故障可能触发 10 条告警。设置告警分组规则按服务/按时间窗口避免告警风暴。五、总结四个黄金信号是最小可行集Latency/Traffic/Errors/Saturation先覆盖这四个再扩展。分层次监控L1 存活性、L2 技术指标、L3 业务指标各层看板独立但数据互通。Histogram 优于 SummaryHistogram 可以在 Prometheus 侧做任意分位计算Summary 做不到。告警规则要审慎告警太敏感→告警疲劳太迟钝→故障发现延迟。本文实现了一个 Prometheus 兼容的指标 SDKCounter/Gauge/Histogram包含 Prometheus 导出格式和 AlertManager 告警规则模板可直接集成到后端服务中。
