Coverage for gco / services / service_metrics.py: 100.00%

1"""Prometheus metrics for the in-cluster GCO services. 

2 

3Exposes a Prometheus ``/metrics`` surface for the long-running GCO services so 

4the self-hosted cluster Prometheus can scrape them: 

5 

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, *, host)`` starts a standalone 

12 metrics HTTP server for the loop-based inference monitor (which has no HTTP 

13 server of its own) and registers a collector that reflects the monitor's live 

14 counters at scrape time. The deployed monitor binds it to loopback and a TLS 

15 sidecar serves it to Prometheus over verified HTTPS. 

16 

17``prometheus-client`` is already a project dependency, so instrumenting these 

18services pulls no new package into their container images. All series carry a 

19``service`` label so one Prometheus deployment can scrape all four GCO services 

20without metric-name collisions. 

21""" 

22 

23from __future__ import annotations 

24 

25import time 

26from collections.abc import Awaitable, Callable, Iterator, Mapping 

27from typing import TYPE_CHECKING, Any 

28 

29from prometheus_client import CONTENT_TYPE_LATEST, Counter, Gauge, Histogram, generate_latest 

30from prometheus_client.core import GaugeMetricFamily 

31from prometheus_client.registry import Collector 

32 

33if TYPE_CHECKING: # pragma: no cover - typing only 

34 from fastapi import FastAPI 

35 from starlette.requests import Request 

36 from starlette.responses import Response 

37 

38# Module-level metrics live in the default registry (what ``generate_latest`` 

39# and ``start_http_server`` serve). Created once per process; ``mount_metrics`` 

40# is safe to call more than once because it never re-declares them. 

41_REQUESTS = Counter( 

42 "gco_http_requests_total", 

43 "Total HTTP requests handled by a GCO service, by method and status.", 

44 ["service", "method", "status"], 

45) 

46_LATENCY = Histogram( 

47 "gco_http_request_duration_seconds", 

48 "HTTP request duration handled by a GCO service, in seconds.", 

49 ["service", "method"], 

50) 

51_INFO = Gauge( 

52 "gco_service_info", 

53 "GCO service liveness marker (always 1 while the process is up).", 

54 ["service"], 

55) 

56 

57_METRICS_PATH = "/metrics" 

58 

59 

60def _render_latest() -> tuple[bytes, str]: 

61 """Return the current Prometheus exposition payload and its content type.""" 

62 return generate_latest(), CONTENT_TYPE_LATEST 

63 

64 

65def mount_metrics(app: FastAPI, service_name: str) -> None: 

66 """Add a ``/metrics`` endpoint and request instrumentation to a FastAPI app. 

67 

68 Records request count and latency for every non-``/metrics`` request and 

69 exposes the default Prometheus registry at ``/metrics``. Cardinality is kept 

70 low: series are labelled by service, HTTP method, and status code only, not 

71 by raw request path. 

72 """ 

73 from starlette.responses import Response 

74 

75 _INFO.labels(service=service_name).set(1) 

76 

77 @app.middleware("http") 

78 async def _instrument( 

79 request: Request, call_next: Callable[[Request], Awaitable[Response]] 

80 ) -> Response: 

81 if request.url.path == _METRICS_PATH: 

82 return await call_next(request) 

83 start = time.perf_counter() 

84 response = await call_next(request) 

85 elapsed = time.perf_counter() - start 

86 _LATENCY.labels(service=service_name, method=request.method).observe(elapsed) 

87 _REQUESTS.labels( 

88 service=service_name, method=request.method, status=str(response.status_code) 

89 ).inc() 

90 return response 

91 

92 async def _metrics_endpoint(_request: Request) -> Response: 

93 payload, content_type = _render_latest() 

94 return Response(content=payload, media_type=content_type) 

95 

96 app.add_route(_METRICS_PATH, _metrics_endpoint, methods=["GET"]) 

97 

98 

99class _CallableCollector(Collector): 

100 """Prometheus collector that reflects a service's live counters at scrape time. 

101 

102 Calls ``metrics_fn`` on every scrape so the exported values are always 

103 current, without the service having to push updates from its control loop. 

104 Numeric values become ``gco_<service>_metric{name=...}`` samples; booleans 

105 are exported as 1/0; non-numeric values (ids, regions) are skipped. 

106 """ 

107 

108 def __init__(self, service_name: str, metrics_fn: Callable[[], Mapping[str, Any]]) -> None: 

109 self._service = service_name 

110 self._metrics_fn = metrics_fn 

111 self._metric_name = f"gco_{service_name.replace('-', '_')}_metric" 

112 

113 def collect(self) -> Iterator[GaugeMetricFamily]: 

114 info = GaugeMetricFamily( 

115 "gco_service_info", 

116 "GCO service liveness marker (always 1 while the process is up).", 

117 labels=["service"], 

118 ) 

119 info.add_metric([self._service], 1.0) 

120 yield info 

121 

122 family = GaugeMetricFamily( 

123 self._metric_name, 

124 f"Numeric counters reported by the {self._service} control loop.", 

125 labels=["name"], 

126 ) 

127 try: 

128 reported = self._metrics_fn() or {} 

129 except Exception: # a scrape must never crash the service 

130 reported = {} 

131 for key, value in reported.items(): 

132 if isinstance(value, bool): 

133 family.add_metric([key], 1.0 if value else 0.0) 

134 elif isinstance(value, (int, float)): 

135 family.add_metric([key], float(value)) 

136 yield family 

137 

138 

139def start_metrics_server( 

140 port: int, 

141 service_name: str, 

142 metrics_fn: Callable[[], Mapping[str, Any]], 

143 *, 

144 host: str = "0.0.0.0", # every interface unless the caller narrows it 

145) -> None: 

146 """Start a standalone Prometheus metrics HTTP server for a loop-based service. 

147 

148 Serves a scrape-time collector backed by ``metrics_fn`` (e.g. the inference 

149 monitor's ``get_metrics``) on ``host:port``. 

150 

151 ``host`` defaults to every interface so a bare ``python -m`` run (local 

152 development, the functional container test) is reachable. The deployed 

153 inference monitor passes ``127.0.0.1``: its plaintext listener is then 

154 reachable only inside the pod, and Prometheus scrapes it through the TLS 

155 sidecar that terminates HTTPS on the pod network. 

156 

157 The collector is registered on its own ``CollectorRegistry`` rather than the 

158 default one. The collector emits its own ``gco_service_info`` liveness series 

159 (so a loop service that never calls ``mount_metrics`` still reports 

160 liveness), which would collide with the module-level ``_INFO`` gauge if both 

161 lived in the default registry. A dedicated registry keeps the loop service's 

162 exposition to exactly its own series and lets the function be called more 

163 than once per process (e.g. across tests) without tripping the default 

164 registry's duplicate-name guard. 

165 """ 

166 from prometheus_client import CollectorRegistry, start_http_server 

167 

168 registry = CollectorRegistry() 

169 registry.register(_CallableCollector(service_name, metrics_fn)) 

170 start_http_server(port, addr=host, registry=registry)