Coverage for gco/services/service_metrics.py: 100.00%
55 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-30 21:22 +0000
1"""Prometheus metrics for the in-cluster GCO services.
3Exposes a Prometheus ``/metrics`` surface for the long-running GCO services so
4the self-hosted cluster Prometheus can scrape them:
6- ``mount_metrics(app, name)`` adds a ``GET /metrics`` endpoint and request
7 instrumentation (request count + latency) to a FastAPI app. Used by the
8 health-monitor, manifest-processor, and inference-proxy API services. Their auth
9 middleware already treats ``/metrics`` as unauthenticated, so the in-cluster
10 Prometheus scrapes it over the existing service port without credentials.
11- ``start_metrics_server(port, name, metrics_fn)`` starts a standalone metrics
12 HTTP server for the loop-based inference monitor (which has no HTTP server of
13 its own) and registers a collector that reflects the monitor's live counters
14 at scrape time.
16``prometheus-client`` is already a project dependency, so instrumenting these
17services pulls no new package into their container images. All series carry a
18``service`` label so one Prometheus deployment can scrape all four GCO services
19without metric-name collisions.
20"""
22from __future__ import annotations
24import time
25from collections.abc import Awaitable, Callable, Iterator, Mapping
26from typing import TYPE_CHECKING, Any
28from prometheus_client import CONTENT_TYPE_LATEST, Counter, Gauge, Histogram, generate_latest
29from prometheus_client.core import GaugeMetricFamily
30from prometheus_client.registry import Collector
32if TYPE_CHECKING: # pragma: no cover - typing only
33 from fastapi import FastAPI
34 from starlette.requests import Request
35 from starlette.responses import Response
37# Module-level metrics live in the default registry (what ``generate_latest``
38# and ``start_http_server`` serve). Created once per process; ``mount_metrics``
39# is safe to call more than once because it never re-declares them.
40_REQUESTS = Counter(
41 "gco_http_requests_total",
42 "Total HTTP requests handled by a GCO service, by method and status.",
43 ["service", "method", "status"],
44)
45_LATENCY = Histogram(
46 "gco_http_request_duration_seconds",
47 "HTTP request duration handled by a GCO service, in seconds.",
48 ["service", "method"],
49)
50_INFO = Gauge(
51 "gco_service_info",
52 "GCO service liveness marker (always 1 while the process is up).",
53 ["service"],
54)
56_METRICS_PATH = "/metrics"
59def _render_latest() -> tuple[bytes, str]:
60 """Return the current Prometheus exposition payload and its content type."""
61 return generate_latest(), CONTENT_TYPE_LATEST
64def mount_metrics(app: FastAPI, service_name: str) -> None:
65 """Add a ``/metrics`` endpoint and request instrumentation to a FastAPI app.
67 Records request count and latency for every non-``/metrics`` request and
68 exposes the default Prometheus registry at ``/metrics``. Cardinality is kept
69 low: series are labelled by service, HTTP method, and status code only, not
70 by raw request path.
71 """
72 from starlette.responses import Response
74 _INFO.labels(service=service_name).set(1)
76 @app.middleware("http")
77 async def _instrument(
78 request: Request, call_next: Callable[[Request], Awaitable[Response]]
79 ) -> Response:
80 if request.url.path == _METRICS_PATH:
81 return await call_next(request)
82 start = time.perf_counter()
83 response = await call_next(request)
84 elapsed = time.perf_counter() - start
85 _LATENCY.labels(service=service_name, method=request.method).observe(elapsed)
86 _REQUESTS.labels(
87 service=service_name, method=request.method, status=str(response.status_code)
88 ).inc()
89 return response
91 async def _metrics_endpoint(_request: Request) -> Response:
92 payload, content_type = _render_latest()
93 return Response(content=payload, media_type=content_type)
95 app.add_route(_METRICS_PATH, _metrics_endpoint, methods=["GET"])
98class _CallableCollector(Collector):
99 """Prometheus collector that reflects a service's live counters at scrape time.
101 Calls ``metrics_fn`` on every scrape so the exported values are always
102 current, without the service having to push updates from its control loop.
103 Numeric values become ``gco_<service>_metric{name=...}`` samples; booleans
104 are exported as 1/0; non-numeric values (ids, regions) are skipped.
105 """
107 def __init__(self, service_name: str, metrics_fn: Callable[[], Mapping[str, Any]]) -> None:
108 self._service = service_name
109 self._metrics_fn = metrics_fn
110 self._metric_name = f"gco_{service_name.replace('-', '_')}_metric"
112 def collect(self) -> Iterator[GaugeMetricFamily]:
113 info = GaugeMetricFamily(
114 "gco_service_info",
115 "GCO service liveness marker (always 1 while the process is up).",
116 labels=["service"],
117 )
118 info.add_metric([self._service], 1.0)
119 yield info
121 family = GaugeMetricFamily(
122 self._metric_name,
123 f"Numeric counters reported by the {self._service} control loop.",
124 labels=["name"],
125 )
126 try:
127 reported = self._metrics_fn() or {}
128 except Exception: # noqa: BLE001 - a scrape must never crash the service
129 reported = {}
130 for key, value in reported.items():
131 if isinstance(value, bool):
132 family.add_metric([key], 1.0 if value else 0.0)
133 elif isinstance(value, (int, float)):
134 family.add_metric([key], float(value))
135 yield family
138def start_metrics_server(
139 port: int,
140 service_name: str,
141 metrics_fn: Callable[[], Mapping[str, Any]],
142) -> None:
143 """Start a standalone Prometheus metrics HTTP server for a loop-based service.
145 Serves a scrape-time collector backed by ``metrics_fn`` (e.g. the inference
146 monitor's ``get_metrics``) on ``port``.
148 The collector is registered on its own ``CollectorRegistry`` rather than the
149 default one. The collector emits its own ``gco_service_info`` liveness series
150 (so a loop service that never calls ``mount_metrics`` still reports
151 liveness), which would collide with the module-level ``_INFO`` gauge if both
152 lived in the default registry. A dedicated registry keeps the loop service's
153 exposition to exactly its own series and lets the function be called more
154 than once per process (e.g. across tests) without tripping the default
155 registry's duplicate-name guard.
156 """
157 from prometheus_client import CollectorRegistry, start_http_server
159 registry = CollectorRegistry()
160 registry.register(_CallableCollector(service_name, metrics_fn))
161 start_http_server(port, registry=registry)