"""In-process metrics registry — no dependencies, single event loop. Counters, gauges and histogram summaries (count/sum/min/max), labeled. Exposed as JSON via GET /api/admin/metrics; a Prometheus text exporter would sit on top of this registry rather than replace it. """ from collections import defaultdict from dataclasses import dataclass from typing import Any LabelKey = tuple[tuple[str, str], ...] def _key(labels: dict[str, str] | None) -> LabelKey: return tuple(sorted((labels or {}).items())) @dataclass class HistogramData: count: int = 0 total: float = 0.0 minimum: float | None = None maximum: float | None = None class MetricsRegistry: def __init__(self) -> None: self._counters: dict[str, dict[LabelKey, float]] = defaultdict( lambda: defaultdict(float) ) self._gauges: dict[str, dict[LabelKey, float]] = defaultdict(dict) self._histograms: dict[str, dict[LabelKey, HistogramData]] = defaultdict(dict) def inc( self, name: str, labels: dict[str, str] | None = None, value: float = 1.0 ) -> None: self._counters[name][_key(labels)] += value def set_gauge( self, name: str, value: float, labels: dict[str, str] | None = None ) -> None: self._gauges[name][_key(labels)] = value def observe( self, name: str, value: float, labels: dict[str, str] | None = None ) -> None: data = self._histograms[name].setdefault(_key(labels), HistogramData()) data.count += 1 data.total += value data.minimum = value if data.minimum is None else min(data.minimum, value) data.maximum = value if data.maximum is None else max(data.maximum, value) def snapshot(self) -> dict[str, Any]: return { "counters": { name: [ {"labels": dict(key), "value": value} for key, value in sorted(series.items()) ] for name, series in sorted(self._counters.items()) }, "gauges": { name: [ {"labels": dict(key), "value": value} for key, value in sorted(series.items()) ] for name, series in sorted(self._gauges.items()) }, "histograms": { name: [ { "labels": dict(key), "count": data.count, "sum": data.total, "min": data.minimum, "max": data.maximum, "avg": data.total / data.count if data.count else None, } for key, data in sorted(series.items()) ] for name, series in sorted(self._histograms.items()) }, } def reset(self) -> None: self._counters.clear() self._gauges.clear() self._histograms.clear() metrics = MetricsRegistry()