diff --git a/metrics.py b/metrics.py index f772c19..0054fd4 100644 --- a/metrics.py +++ b/metrics.py @@ -1,15 +1,22 @@ -"""In-process traffic counters for the ops dashboard. +"""Traffic counters for the ops dashboard. -Cheap, thread-safe, best-effort counters for **inbound** HTTP requests (bucketed by -category) and **outbound** calls to external data sources (bucketed by source). -Everything here is since-process-start; durable daily history lives in the -audit/error JSONL logs (see ``audit.py``). Recording never raises into the request -path. +Cheap, best-effort counters for **inbound** HTTP requests (bucketed by category) and +**outbound** calls to external data sources (bucketed by source). Everything here is +since-start; durable daily history lives in the audit/error JSONL logs (see +``audit.py``). Recording never raises into the request path. Read over the gated ``GET {BASE}/api/v2/metrics`` route (see ``app.py``) and rendered by ``scripts/dashboard.py`` over SSH. + +**Multi-worker.** Under a single uvicorn worker the counters live in process memory. +When the app runs several workers (``--workers N``), per-process memory would split the +tally across workers, so the dashboard — which polls one random worker — would see only +a fraction of the traffic. Set ``THERMOGRAPH_METRICS_DB`` to a file path and every worker +shares one SQLite store, so the dashboard reports the whole picture. Unset (dev, tests, +single worker) keeps the zero-dependency in-memory store. """ import os +import sqlite3 import threading import time @@ -32,42 +39,9 @@ SOURCES = sorted(set(_PHASE_SOURCE.values())) _OUTCOMES = ("ok", "retry", "error", "rate_limited") _STATUS_BUCKETS = ("2xx", "3xx", "4xx", "5xx", "other") -_lock = threading.Lock() -_started_at = time.time() -_inbound: "dict[str, dict[str, int]]" = {} -_outbound: "dict[str, dict[str, int]]" = { - s: {o: 0 for o in _OUTCOMES} for s in SOURCES -} - -# Rolling short-window view: per-minute buckets (epoch-minute -> {key: count}) for -# inbound categories and outbound sources, summed over the last WINDOW minutes on -# snapshot(). Cheap and bounded — at most WINDOW+1 tiny dicts, pruned as minutes roll. +# Rolling short-window view: per-minute buckets summed over the last WINDOW minutes on +# snapshot(). Cheap and bounded — minutes are pruned as they roll out of the window. WINDOW_MINUTES = 10 -_recent_in: "dict[int, dict[str, int]]" = {} -_recent_out: "dict[int, dict[str, int]]" = {} - - -def _bump_recent(buckets: "dict[int, dict[str, int]]", key: str) -> None: - """Count one event for ``key`` in the current-minute bucket (caller holds _lock).""" - minute = int(time.time() // 60) - row = buckets.get(minute) - if row is None: - row = buckets[minute] = {} - cutoff = minute - WINDOW_MINUTES - for old in [m for m in buckets if m <= cutoff]: - del buckets[old] # drop minutes that have rolled out of the window - row[key] = row.get(key, 0) + 1 - - -def _recent_totals(buckets: "dict[int, dict[str, int]]") -> "dict[str, int]": - """Sum each key's count over the last WINDOW minutes (caller holds _lock).""" - cutoff = int(time.time() // 60) - WINDOW_MINUTES + 1 - totals: "dict[str, int]" = {} - for minute, row in buckets.items(): - if minute >= cutoff: - for key, count in row.items(): - totals[key] = totals.get(key, 0) + count - return totals def source_for_phase(phase: str) -> str: @@ -130,6 +104,207 @@ def classify_inbound(path: str, base: str = "") -> str: return "other" +def _now_minute() -> int: + """Current epoch-minute. Uses ``time.time()`` so tests can monkeypatch the clock.""" + return int(time.time() // 60) + + +def _empty_inbound_row() -> "dict[str, int]": + return {"total": 0, **{b: 0 for b in _STATUS_BUCKETS}} + + +class _MemStore: + """In-process counters. One per worker — fine for a single-worker deployment.""" + + def __init__(self) -> None: + self._lock = threading.Lock() + self.started_at = time.time() + self._inbound: "dict[str, dict[str, int]]" = {} + self._outbound: "dict[str, dict[str, int]]" = { + s: {o: 0 for o in _OUTCOMES} for s in SOURCES + } + # epoch-minute -> {key: count}, pruned as minutes roll out of the window. + self._recent_in: "dict[int, dict[str, int]]" = {} + self._recent_out: "dict[int, dict[str, int]]" = {} + + def _bump_recent(self, buckets, key) -> None: + """Count one event for ``key`` in the current-minute bucket (caller holds lock).""" + minute = _now_minute() + row = buckets.get(minute) + if row is None: + row = buckets[minute] = {} + cutoff = minute - WINDOW_MINUTES + for old in [m for m in buckets if m <= cutoff]: + del buckets[old] # drop minutes that have rolled out of the window + row[key] = row.get(key, 0) + 1 + + @staticmethod + def _recent_totals(buckets) -> "dict[str, int]": + """Sum each key's count over the last WINDOW minutes (caller holds lock).""" + cutoff = _now_minute() - WINDOW_MINUTES + 1 + totals: "dict[str, int]" = {} + for minute, row in buckets.items(): + if minute >= cutoff: + for key, count in row.items(): + totals[key] = totals.get(key, 0) + count + return totals + + def record_inbound(self, category, bucket) -> None: + with self._lock: + row = self._inbound.get(category) + if row is None: + row = self._inbound[category] = _empty_inbound_row() + row["total"] += 1 + row[bucket] = row.get(bucket, 0) + 1 + self._bump_recent(self._recent_in, category) + + def record_outbound(self, source, outcome) -> None: + with self._lock: + row = self._outbound.get(source) + if row is None: + row = self._outbound[source] = {o: 0 for o in _OUTCOMES} + row[outcome] = row.get(outcome, 0) + 1 + self._bump_recent(self._recent_out, source) + + def read(self) -> dict: + with self._lock: + inbound = {k: dict(v) for k, v in self._inbound.items()} + outbound = {k: dict(v) for k, v in self._outbound.items()} + return { + "started_at": self.started_at, + "inbound": inbound, + "outbound": outbound, + "inbound_recent": self._recent_totals(self._recent_in), + "outbound_recent": self._recent_totals(self._recent_out), + } + + +class _SqliteStore: + """Cross-process counters in a shared SQLite file, so every uvicorn worker tallies + into one place and the dashboard sees the whole deployment. WAL mode lets the workers + write concurrently; the counts are tiny UPSERTs well within SQLite's throughput. + """ + + def __init__(self, path: str) -> None: + self._lock = threading.Lock() + # check_same_thread=False: the connection is shared across a worker's threads, + # serialized by self._lock. isolation_level=None: autocommit each statement. + self._db = sqlite3.connect(path, check_same_thread=False, isolation_level=None) + self._db.execute("PRAGMA journal_mode=WAL") + self._db.execute("PRAGMA synchronous=NORMAL") + self._db.execute("PRAGMA busy_timeout=3000") + self._db.executescript( + """ + CREATE TABLE IF NOT EXISTS meta(key TEXT PRIMARY KEY, value REAL); + CREATE TABLE IF NOT EXISTS inbound_cume( + category TEXT, bucket TEXT, count INTEGER, PRIMARY KEY(category, bucket)); + CREATE TABLE IF NOT EXISTS outbound_cume( + source TEXT, outcome TEXT, count INTEGER, PRIMARY KEY(source, outcome)); + CREATE TABLE IF NOT EXISTS inbound_minute( + minute INTEGER, category TEXT, count INTEGER, PRIMARY KEY(minute, category)); + CREATE TABLE IF NOT EXISTS outbound_minute( + minute INTEGER, source TEXT, count INTEGER, PRIMARY KEY(minute, source)); + """ + ) + # First worker to touch the DB stamps the shared start time; the rest inherit it. + self._db.execute( + "INSERT OR IGNORE INTO meta(key, value) VALUES('started_at', ?)", (time.time(),) + ) + self._last_pruned_minute = -1 + + @property + def started_at(self) -> float: + row = self._db.execute( + "SELECT value FROM meta WHERE key='started_at'").fetchone() + return row[0] if row else time.time() + + def _prune(self, minute: int) -> None: + """Drop per-minute rows that have rolled out of the window (idempotent across + workers). Cheap, and only run when the epoch-minute advances.""" + if minute == self._last_pruned_minute: + return + self._last_pruned_minute = minute + cutoff = minute - WINDOW_MINUTES + self._db.execute("DELETE FROM inbound_minute WHERE minute <= ?", (cutoff,)) + self._db.execute("DELETE FROM outbound_minute WHERE minute <= ?", (cutoff,)) + + def record_inbound(self, category, bucket) -> None: + minute = _now_minute() + with self._lock: + self._db.execute( + "INSERT INTO inbound_cume(category, bucket, count) VALUES(?, ?, 1) " + "ON CONFLICT(category, bucket) DO UPDATE SET count = count + 1", + (category, bucket), + ) + self._db.execute( + "INSERT INTO inbound_minute(minute, category, count) VALUES(?, ?, 1) " + "ON CONFLICT(minute, category) DO UPDATE SET count = count + 1", + (minute, category), + ) + self._prune(minute) + + def record_outbound(self, source, outcome) -> None: + minute = _now_minute() + with self._lock: + self._db.execute( + "INSERT INTO outbound_cume(source, outcome, count) VALUES(?, ?, 1) " + "ON CONFLICT(source, outcome) DO UPDATE SET count = count + 1", + (source, outcome), + ) + self._db.execute( + "INSERT INTO outbound_minute(minute, source, count) VALUES(?, ?, 1) " + "ON CONFLICT(minute, source) DO UPDATE SET count = count + 1", + (minute, source), + ) + self._prune(minute) + + def read(self) -> dict: + cutoff = _now_minute() - WINDOW_MINUTES + 1 + with self._lock: + inbound: "dict[str, dict[str, int]]" = {} + for category, bucket, count in self._db.execute( + "SELECT category, bucket, count FROM inbound_cume"): + row = inbound.setdefault(category, _empty_inbound_row()) + row[bucket] = row.get(bucket, 0) + count + row["total"] += count + # Start from the stable source rows so the dashboard shows them even at zero. + outbound: "dict[str, dict[str, int]]" = { + s: {o: 0 for o in _OUTCOMES} for s in SOURCES + } + for source, outcome, count in self._db.execute( + "SELECT source, outcome, count FROM outbound_cume"): + outbound.setdefault(source, {o: 0 for o in _OUTCOMES})[outcome] = count + recent_in = dict(self._db.execute( + "SELECT category, SUM(count) FROM inbound_minute WHERE minute >= ? " + "GROUP BY category", (cutoff,))) + recent_out = dict(self._db.execute( + "SELECT source, SUM(count) FROM outbound_minute WHERE minute >= ? " + "GROUP BY source", (cutoff,))) + return { + "started_at": self.started_at, + "inbound": inbound, + "outbound": outbound, + "inbound_recent": recent_in, + "outbound_recent": recent_out, + } + + +def _make_store(): + """Pick the store from the environment. A path in ``THERMOGRAPH_METRICS_DB`` selects + the shared SQLite store (needed when running multiple workers); otherwise in-memory. + A store that fails to open falls back to memory — metrics must never break boot.""" + path = os.environ.get("THERMOGRAPH_METRICS_DB", "").strip() + if path: + try: + return _SqliteStore(path) + except Exception: # noqa: BLE001 - never let instrumentation break startup + pass + return _MemStore() + + +_store = _make_store() + + def record_inbound(category: str, status) -> None: """Count one inbound request in ``category``, bucketed by status class.""" # The metrics endpoint is polled by the dashboard itself — counting it would just @@ -137,14 +312,7 @@ def record_inbound(category: str, status) -> None: if category == "metrics": return try: - bucket = _status_bucket(status) - with _lock: - row = _inbound.get(category) - if row is None: - row = _inbound[category] = {"total": 0, **{b: 0 for b in _STATUS_BUCKETS}} - row["total"] += 1 - row[bucket] = row.get(bucket, 0) + 1 - _bump_recent(_recent_in, category) + _store.record_inbound(category, _status_bucket(status)) except Exception: # noqa: BLE001 - instrumentation must never break a request pass @@ -152,34 +320,31 @@ def record_inbound(category: str, status) -> None: def record_outbound(phase: str, outcome: str) -> None: """Count one outbound call. ``outcome`` in {ok, retry, error, rate_limited}.""" try: - source = source_for_phase(phase) - with _lock: - row = _outbound.get(source) - if row is None: - row = _outbound[source] = {o: 0 for o in _OUTCOMES} - row[outcome] = row.get(outcome, 0) + 1 - _bump_recent(_recent_out, source) + _store.record_outbound(source_for_phase(phase), outcome) except Exception: # noqa: BLE001 pass def snapshot() -> dict: """A JSON-serializable copy of all counters plus process facts.""" - with _lock: - inbound = {k: dict(v) for k, v in _inbound.items()} - outbound = {k: dict(v) for k, v in _outbound.items()} - recent_in = _recent_totals(_recent_in) - recent_out = _recent_totals(_recent_out) + try: + data = _store.read() + except Exception: # noqa: BLE001 - a locked/broken store must not 500 the dashboard + data = {"started_at": time.time(), "inbound": {}, "outbound": {}, + "inbound_recent": {}, "outbound_recent": {}} + inbound = data["inbound"] + outbound = data["outbound"] + started_at = data["started_at"] return { "pid": os.getpid(), - "started_at": _started_at, - "uptime_s": round(time.time() - _started_at, 1), + "started_at": started_at, + "uptime_s": round(time.time() - started_at, 1), "inbound": inbound, "outbound": outbound, "inbound_total": sum(v.get("total", 0) for v in inbound.values()), "outbound_total": sum(sum(v.values()) for v in outbound.values()), # Traffic over the last WINDOW_MINUTES, per category / source. "window_minutes": WINDOW_MINUTES, - "inbound_recent": recent_in, - "outbound_recent": recent_out, + "inbound_recent": data["inbound_recent"], + "outbound_recent": data["outbound_recent"], } diff --git a/tests/test_metrics.py b/tests/test_metrics.py index e15aad3..20781ea 100644 --- a/tests/test_metrics.py +++ b/tests/test_metrics.py @@ -132,3 +132,70 @@ def test_recent_excludes_self_polling(): snap = m.snapshot() assert "metrics" not in snap["inbound_recent"] assert snap["inbound_recent"]["page"] == 1 + + +# --- shared SQLite store (multi-worker) ------------------------------------------- +# When the app runs several uvicorn workers, per-process memory would split the tally +# across workers; THERMOGRAPH_METRICS_DB points them at one shared SQLite file instead. + +def test_sqlite_store_matches_snapshot_shape(tmp_path): + m = _fresh() + store = m._SqliteStore(str(tmp_path / "metrics.db")) + store.record_inbound("api:place", "2xx") + store.record_inbound("api:place", "2xx") + store.record_inbound("api:place", "5xx") + store.record_outbound("open-meteo-archive", "ok") + data = store.read() + assert data["inbound"]["api:place"]["total"] == 3 + assert data["inbound"]["api:place"]["2xx"] == 2 + assert data["inbound"]["api:place"]["5xx"] == 1 + # stable source rows are present even at zero traffic, like the in-memory store + assert data["outbound"]["open-meteo-archive"]["ok"] == 1 + assert "nasa-power" in data["outbound"] + assert data["inbound_recent"]["api:place"] == 3 + assert data["outbound_recent"]["open-meteo-archive"] == 1 + + +def test_sqlite_store_is_shared_across_workers(tmp_path): + # Two store instances on the same file stand in for two worker processes: what one + # records, the other must see — that's the whole point of the shared store. + m = _fresh() + db = str(tmp_path / "metrics.db") + worker_a = m._SqliteStore(db) + worker_b = m._SqliteStore(db) + worker_a.record_inbound("page", "2xx") + worker_b.record_inbound("page", "2xx") + worker_b.record_inbound("api:grade", "2xx") + # Either worker, polled by the dashboard, reports the combined tally. + for worker in (worker_a, worker_b): + data = worker.read() + assert data["inbound"]["page"]["total"] == 2 + assert data["inbound"]["api:grade"]["total"] == 1 + assert data["inbound_recent"]["page"] == 2 + # Both workers share one start time (first writer wins), not one-per-process. + assert worker_a.started_at == worker_b.started_at + + +def test_sqlite_recent_window_rolls_off(tmp_path, monkeypatch): + m = _fresh() + clock = {"now": 1_000_000.0} + monkeypatch.setattr(m.time, "time", lambda: clock["now"]) + store = m._SqliteStore(str(tmp_path / "metrics.db")) + + store.record_inbound("page", "2xx") + assert store.read()["inbound_recent"]["page"] == 1 + # +11 min: the hit has rolled out of the 10-minute window and is pruned... + clock["now"] += 11 * 60 + store.record_inbound("page", "2xx") # advances the minute -> triggers prune + data = store.read() + assert data["inbound_recent"]["page"] == 1 # only the fresh hit remains in-window + assert data["inbound"]["page"]["total"] == 2 # ...but the cumulative total is intact + + +def test_env_selects_sqlite_store(tmp_path, monkeypatch): + # Setting the env var makes the module boot on the shared store, not in-memory. + monkeypatch.setenv("THERMOGRAPH_METRICS_DB", str(tmp_path / "metrics.db")) + m = _fresh() + assert isinstance(m._store, m._SqliteStore) + m.record_inbound("page", 200) + assert m.snapshot()["inbound"]["page"]["total"] == 1