"""Metrics history store for time-series metric snapshots. Records per-mcp_server metric snapshots every 60 seconds into SQLite, supports querying by mcp_server, metric name, and time range, and prunes old data based on a configurable retention window. """ import time from collections.abc import Generator from contextlib import contextmanager from ...domain.contracts.metrics_history import IMetricsHistoryStore, MetricPoint from ...logging_config import get_logger from .database_common import MigrationRunner, SQLiteConfig, SQLiteConnectionFactory logger = get_logger(__name__) # --------------------------------------------------------------------------- # Schema # --------------------------------------------------------------------------- _MIGRATIONS: list[dict] = [ { "version": 1, "name": "create_metric_snapshots", "sql": """ CREATE TABLE IF NOT EXISTS metric_snapshots ( id INTEGER PRIMARY KEY AUTOINCREMENT, mcp_server_id TEXT NOT NULL, metric_name TEXT NOT NULL, value REAL NOT NULL, recorded_at REAL NOT NULL ); CREATE INDEX IF NOT EXISTS idx_metric_snapshots_lookup ON metric_snapshots (mcp_server_id, metric_name, recorded_at); """, }, ] # --------------------------------------------------------------------------- # Data object # --------------------------------------------------------------------------- #: Re-exported so `from ...metrics_history_store import MetricPoint` keeps #: working. The definition moved to the port: it is the record both backends #: speak in, and a port that reaches into one adapter for its own vocabulary #: inverts the hexagon -- which the import contract rejected. __all__ = ["MetricPoint", "MetricsHistoryStore"] # --------------------------------------------------------------------------- # Store # --------------------------------------------------------------------------- class MetricsHistoryStore(IMetricsHistoryStore): """SQLite-backed store for per-mcp_server metric snapshots. Provides: - :meth:`record_snapshot` — persist a batch of :class:`MetricPoint` rows. - :meth:`query` — retrieve points filtered by mcp_server, metric, and time range. - :meth:`prune` — delete rows older than the retention window. Thread-safety: each method does all of its work -- execute, fetch every row, commit -- inside one :meth:`SQLiteConnectionFactory.get_connection` block. For the default in-memory database the factory has a single connection that every thread shares, and it holds its lock for the whole block, so calls from different threads, reads included, run one at a time. A file-backed store gets one connection per thread, with no lock, and SQLite's own file locking (WAL, ``busy_timeout``) orders the writes. Args: config: SQLite configuration. Defaults to in-memory (useful for tests). retention_days: How many days of history to keep. Prune is not run automatically — callers (or the background worker) must call :meth:`prune` periodically. """ def __init__( self, config: SQLiteConfig | None = None, retention_days: int = 7, ) -> None: self._config = config or SQLiteConfig(path=":memory:") self._retention_days = retention_days self._factory = SQLiteConnectionFactory(self._config) self._runner = MigrationRunner( connection_factory=self._factory, migrations=_MIGRATIONS, ) self._runner.run() @contextmanager def _conn(self) -> Generator: with self._factory.get_connection() as conn: yield conn conn.commit() def record_snapshot(self, points: list[MetricPoint]) -> None: """Persist a batch of metric data points. Args: points: List of :class:`MetricPoint` instances to store. """ if not points: return rows = [(p.mcp_server_id, p.metric_name, p.value, p.recorded_at) for p in points] with self._conn() as conn: conn.executemany( "INSERT INTO metric_snapshots (mcp_server_id, metric_name, value, recorded_at) VALUES (?, ?, ?, ?)", rows, ) def query( self, mcp_server_id: str | None = None, metric_name: str | None = None, from_ts: float | None = None, to_ts: float | None = None, limit: int = 1000, ) -> list[MetricPoint]: """Query stored metric history. Args: mcp_server_id: Filter by mcp_server. ``None`` returns all mcp_servers. metric_name: Filter by metric name. ``None`` returns all metrics. from_ts: Start of time range (unix timestamp, inclusive). to_ts: End of time range (unix timestamp, inclusive). limit: Maximum number of rows to return (capped at 10 000). Returns: List of :class:`MetricPoint` ordered by ``recorded_at`` ascending. """ limit = min(max(1, limit), 10_000) conditions: list[str] = [] params: list = [] if mcp_server_id is not None: conditions.append("mcp_server_id = ?") params.append(mcp_server_id) if metric_name is not None: conditions.append("metric_name = ?") params.append(metric_name) if from_ts is not None: conditions.append("recorded_at >= ?") params.append(from_ts) if to_ts is not None: conditions.append("recorded_at <= ?") params.append(to_ts) where = ("WHERE " + " AND ".join(conditions)) if conditions else "" sql = ( f"SELECT mcp_server_id, metric_name, value, recorded_at " f"FROM metric_snapshots {where} " f"ORDER BY recorded_at ASC " f"LIMIT ?" ) params.append(limit) with self._conn() as conn: cursor = conn.execute(sql, params) rows = cursor.fetchall() return [MetricPoint(mcp_server_id=r[0], metric_name=r[1], value=r[2], recorded_at=r[3]) for r in rows] def prune(self) -> int: """Delete metric snapshots older than the retention window. Returns: Number of rows deleted. """ cutoff = time.time() - self._retention_days * 86_400 with self._conn() as conn: cursor = conn.execute( "DELETE FROM metric_snapshots WHERE recorded_at < ?", (cutoff,), ) deleted = cursor.rowcount if deleted: logger.info("metrics_history_pruned", deleted=deleted, retention_days=self._retention_days) return deleted or 0 # --------------------------------------------------------------------------- # Singleton # --------------------------------------------------------------------------- _metrics_history_store: MetricsHistoryStore | None = None def get_metrics_history_store() -> MetricsHistoryStore: """Get the global :class:`MetricsHistoryStore` instance. **Metric history is per-process unless a storage backend is selected.** With ``persistence.backend`` configured, bootstrap installs the backend's store and history survives a restart. Without it -- the documented no-change default, and every configuration carried forward from 2.4.0 and earlier -- the first call here creates ``MetricsHistoryStore()``, which is ``SQLiteConfig(path=":memory:")``. History then lives in process memory and is gone on restart, while ``/metrics/history`` reads like a durable series. This docstring used to instruct production bootstrap to call :func:`set_metrics_history_store` with a file-backed instance. Bootstrap does exactly that (``server/bootstrap/__init__.py``) -- but only inside the ``if _backend:`` branch, so for half the deployments the advice described what the code does and for the other half it described what nobody does (#785). Not data loss in the serious sense: Prometheus scrapes the live metrics and is the system of record. What is per-process is the in-gateway history surface. """ global _metrics_history_store if _metrics_history_store is None: _metrics_history_store = MetricsHistoryStore() return _metrics_history_store def set_metrics_history_store(store: MetricsHistoryStore) -> None: """Override the global :class:`MetricsHistoryStore` (bootstrap / testing).""" global _metrics_history_store _metrics_history_store = store