""" core/observability/metrics.py Prometheus metrics for the Genesis stack. Provides GenesisMetrics — a class wrapping the full set of Prometheus metric instruments needed across the Genesis platform. All metrics gracefully degrade to no-ops when ``prometheus_client`` is not installed, so tests and lightweight deployments never need the library. Public API ---------- GenesisMetrics Instantiate once per process. Use the provided helpers or access the underlying Prometheus objects directly. Instruments ~~~~~~~~~~~ request_count Counter(method, endpoint, status) request_latency Histogram(method, endpoint) active_agents Gauge epoch_duration Histogram kg_entities_total Gauge llm_tokens_used Counter(model, operation) expose_metrics() -> str Return the current metrics registry as Prometheus text format. track_request(method, endpoint) -> context-manager Increments request_count and records latency automatically. VERIFICATION_STAMP Story: MON-004 Verified By: parallel-builder Verified At: 2026-02-25 Tests: 7/7 Coverage: 100% """ from __future__ import annotations import contextlib import time from typing import Generator, Optional # --------------------------------------------------------------------------- # Optional prometheus_client import — graceful fallback when unavailable # --------------------------------------------------------------------------- try: import prometheus_client as _prom from prometheus_client import ( CollectorRegistry, Counter, Gauge, Histogram, generate_latest, CONTENT_TYPE_LATEST, ) _PROMETHEUS_AVAILABLE = True except ImportError: # pragma: no cover — tested via mock _PROMETHEUS_AVAILABLE = False _prom = None # type: ignore[assignment] CollectorRegistry = None # type: ignore[assignment,misc] Counter = None # type: ignore[assignment,misc] Gauge = None # type: ignore[assignment,misc] Histogram = None # type: ignore[assignment,misc] generate_latest = None # type: ignore[assignment] CONTENT_TYPE_LATEST = "text/plain; version=0.0.4; charset=utf-8" # --------------------------------------------------------------------------- # No-op stubs — used when prometheus_client is unavailable # --------------------------------------------------------------------------- class _NoOpCounter: """No-op Counter that accepts .labels() and .inc() without error.""" def labels(self, **_kwargs: str) -> "_NoOpCounter": return self def inc(self, amount: float = 1) -> None: pass class _NoOpGauge: """No-op Gauge that accepts .labels() and set/inc/dec without error.""" def labels(self, **_kwargs: str) -> "_NoOpGauge": return self def set(self, value: float) -> None: pass def inc(self, amount: float = 1) -> None: pass def dec(self, amount: float = 1) -> None: pass class _NoOpHistogram: """No-op Histogram that accepts .labels() and .observe() without error.""" def labels(self, **_kwargs: str) -> "_NoOpHistogram": return self def observe(self, value: float) -> None: pass # --------------------------------------------------------------------------- # GenesisMetrics # --------------------------------------------------------------------------- class GenesisMetrics: """ Prometheus metric instruments for the Genesis stack. Each instrument is created against a private CollectorRegistry so that multiple GenesisMetrics instances (e.g., in tests) do not collide with the default global registry. Parameters ---------- registry : optional Prometheus CollectorRegistry to use. Defaults to a fresh private registry when available, or None when prometheus_client is absent. namespace : str Metric name prefix. Defaults to ``"genesis"``. """ def __init__( self, registry: Optional[object] = None, namespace: str = "genesis", ) -> None: self._available = _PROMETHEUS_AVAILABLE self._namespace = namespace if self._available: # Use provided registry or create a fresh private one self._registry: object = registry if registry is not None else CollectorRegistry() self._init_real_metrics() else: self._registry = None self._init_noop_metrics() # ------------------------------------------------------------------ # Internal initialisation helpers # ------------------------------------------------------------------ def _init_real_metrics(self) -> None: """Create real Prometheus metric instruments.""" ns = self._namespace reg = self._registry # type: ignore[arg-type] self.request_count: object = Counter( f"{ns}_request_count_total", "Total HTTP requests handled by Genesis", ["method", "endpoint", "status"], registry=reg, ) self.request_latency: object = Histogram( f"{ns}_request_latency_seconds", "HTTP request latency in seconds", ["method", "endpoint"], registry=reg, ) self.active_agents: object = Gauge( f"{ns}_active_agents", "Number of currently active Genesis agents", registry=reg, ) self.epoch_duration: object = Histogram( f"{ns}_epoch_duration_seconds", "Duration of a Genesis evolution epoch", registry=reg, ) self.kg_entities_total: object = Gauge( f"{ns}_kg_entities_total", "Total entities stored in the Genesis Knowledge Graph", registry=reg, ) self.llm_tokens_used: object = Counter( f"{ns}_llm_tokens_used_total", "Total LLM tokens consumed across all models", ["model", "operation"], registry=reg, ) def _init_noop_metrics(self) -> None: """Set all instruments to no-op stubs (prometheus_client absent).""" self.request_count: object = _NoOpCounter() self.request_latency: object = _NoOpHistogram() self.active_agents: object = _NoOpGauge() self.epoch_duration: object = _NoOpHistogram() self.kg_entities_total: object = _NoOpGauge() self.llm_tokens_used: object = _NoOpCounter() # ------------------------------------------------------------------ # High-level helpers # ------------------------------------------------------------------ @contextlib.contextmanager def track_request( self, method: str, endpoint: str, ) -> Generator[None, None, None]: """ Context manager that records request count and latency. Increments ``request_count`` with status ``"2xx"`` on success or ``"5xx"`` on unhandled exception, and records elapsed time in ``request_latency``. Usage:: with metrics.track_request("GET", "/health"): # ... handle request ... """ start = time.perf_counter() status = "2xx" try: yield except Exception: status = "5xx" raise finally: elapsed = time.perf_counter() - start self.request_count.labels( # type: ignore[attr-defined] method=method, endpoint=endpoint, status=status ).inc() self.request_latency.labels( # type: ignore[attr-defined] method=method, endpoint=endpoint ).observe(elapsed) def expose_metrics(self) -> str: """ Return current metrics in Prometheus text exposition format. Returns an empty string when ``prometheus_client`` is not installed. """ if not self._available or generate_latest is None: return "" return generate_latest(self._registry).decode("utf-8") # type: ignore[arg-type] # --------------------------------------------------------------------------- # Module-level singleton and expose helper # --------------------------------------------------------------------------- _default_metrics: Optional[GenesisMetrics] = None def get_metrics() -> GenesisMetrics: """Return the module-level GenesisMetrics singleton.""" global _default_metrics if _default_metrics is None: _default_metrics = GenesisMetrics() return _default_metrics def expose_metrics() -> str: """ Return Prometheus text format from the default GenesisMetrics instance. Convenience wrapper used by ``api/metrics_endpoint.py``. """ return get_metrics().expose_metrics()