From 059f459e8ff012a08c69bd3f2eea97128bca512f Mon Sep 17 00:00:00 2001 From: Emi Griffith Date: Thu, 16 Jul 2026 19:43:22 -0700 Subject: [PATCH] Sync main into dev (multi-worker prod + shared metrics store) (#160) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * Serve the app on thermograph.org at root; redirect emigriffith.dev/thermograph Move Thermograph to its own domain. thermograph.org serves the app at its root, and emigriffith.dev/thermograph* permanently redirects there (prefix stripped, so deep links map straight across). - app.py: BASE now supports an empty root prefix. THERMOGRAPH_BASE=/ (or empty) yields BASE="" so routes sit at "/", "/calendar", "/api/v2/…" with no double slashes; the bare-base redirect is skipped and the static mount falls back to "/". Non-empty values keep the existing "/thermograph" sub-path behavior unchanged (backward compatible). - deploy/Caddyfile: add a thermograph.org site block reverse-proxying to the uvicorn on 127.0.0.1:8137; turn emigriffith.dev's /thermograph handler into a permanent redirect to thermograph.org. - deploy/thermograph.env.example: default THERMOGRAPH_BASE=/. - DEPLOY.md: document the two-domain layout and that deploy.sh does not ship the Caddyfile or env — those are applied on the VPS by hand. The frontend already uses base-relative URLs, so it follows the root base with no changes. Verified both modes locally (root serves /, /calendar, /api/v2/*; the /thermograph sub-path still redirects bare→slash and 404s at root). * Add dashboard (#144) * Auto-submit IndexNow on deploy when the URL set changes (#130) Hook the production deploy (deploy/deploy.sh) to ping IndexNow after the health check passes, so new/changed pages reach Bing/DuckDuckGo/Yandex without a manual run. Best-effort — wrapped so it can never fail a deploy — and it sources /etc/thermograph.env so its key matches the one the running service serves. To avoid re-blasting ~14k URLs on every code push, indexnow.py gains a --if-changed mode gated by a signature of the URL set (persisted to data/indexnow_state.txt): code-only deploys skip; a deploy that adds or removes a city submits. `make indexnow` still forces a full submit. * Add ops metrics endpoint + SSH/Termux dashboard (#131) Adds observability for the running app, viewable over SSH (e.g. Termux): - backend/metrics.py: thread-safe, best-effort in-process counters for inbound requests (per category) and outbound calls (per external source), plus a phase->source map and snapshot(). Hooked into climate._request (outbound, all six weather/geocode sources) and the existing revalidate_static middleware (inbound). Never raises into the request path. - GET /api/v2/metrics: token-gated JSON of the counters + warm-queue depth + thread names/uptime/pid. Returns 404 unless a valid THERMOGRAPH_METRICS_TOKEN is presented or the request is direct loopback with no proxy X-Forwarded-* header, so ops data is never exposed through Caddy on the public domain. - scripts/dashboard.py (+ scripts/dash, make dashboard): stdlib-only terminal dashboard. Reads cache (parquet + SQLite), accounts (users + active subscriptions), and system resources (/proc + systemd) directly, and pulls traffic from the metrics endpoint over loopback. Live refresh, --once, --json; mobile-terminal width. Paths resolve relative to the script so the same tool works on the LAN dev and prod VPS checkouts. - Monitoring sections in DEPLOY.md / DEPLOY-DEV.md; unit tests for the counters. * Dashboard: run under the app venv, not the box's default python3 (#132) The box's default python3 can be a build without _sqlite3 (pyenv 3.10 here), so ./scripts/dash crashed on import. Prefer the repo's .venv interpreter (the Python the server itself runs on, which always has sqlite3); fall back to python3/python. Also make the sqlite3 import in dashboard.py optional so a fallback interpreter degrades to empty cache/accounts panels instead of crashing, and point the 'make dashboard' target at the wrapper. * Dashboard: redraw live view in place so it stops scrolling to the bottom (#133) The live loop did a full clear-and-reprint each refresh; when the output is taller than the terminal (common on a phone) that scrolls the view to the bottom every tick. Redraw from the top instead: hide the cursor, clear once, then each frame move to home and rewrite line-by-line (clear-to-EOL + clear-below), capping the body to the window height so it never prints more lines than fit and never scrolls. Restore the cursor on exit; piped (non-TTY) output prints plain frames. * Dashboard: make the live view a scrollable in-place pager (#134) The live mode is now an alt-screen pager (like top/less): it shows only the latest snapshot, you scroll it with swipe/arrows/jk/PgUp-PgDn/g/G, and each refresh replaces the page in place while keeping your scroll position — nothing gets appended below and the view is never yanked to the top or bottom. On exit it restores the cursor, leaves the alternate screen, and resets the terminal mode. --once still prints a single static snapshot (also used automatically when stdout/stdin aren't a terminal, e.g. piping); --json unchanged. * Metrics: don't count the dashboard's own /api/v2/metrics polling (#135) The ops dashboard polls the metrics endpoint every few seconds; counting those hits just shows the monitor watching itself and inflates the inbound totals. Skip the 'metrics' category in record_inbound (so it's excluded from inbound_total too), and hide it in the dashboard's inbound list defensively. * Dashboard: add a STORAGE section (archive + DB sizes) (#136) Show on-disk storage: the parquet archives (with history/recent split), the derived SQLite DB, and the accounts SQLite DB individually, plus a total. Archive bytes are summed during the existing TTL-cached cache scan; DB sizes include the -wal/-shm sidecars (WAL holds recent writes, so the base file alone understates the footprint). Drop the now-redundant 'payload db' line from CACHE. * Dashboard: show STORAGE sizes in fixed GB (0.01 GB floor) (#137) The auto-scaling formatter only switched to GB once a value passed 1 GB, so the archives read as MB and the DBs as KB. Format the STORAGE section in fixed GB with two decimals and a 0.01 GB floor (so the small databases stay visible instead of rounding to 0.00). Memory readouts keep the auto-scaling formatter. * Dashboard: show STORAGE under 50 MB in MB, GB above (#138) Values below 50 MB now render as MB with one decimal (no KB), so the databases read naturally (derived 7.7MB, accounts 0.4MB) instead of 0.01GB; 50 MB and up stay in GB with two decimals. Renames the helper _gb -> _storage_size to match. * Dashboard: move STORAGE MB->GB cutoff to 500 MB (#139) Values below 500 MB show in MB (one decimal); 500 MB and up in GB. * Dashboard: accounts totals only; log client IPs per request (#140) Dashboard ACCOUNTS section now shows only totals (users, active/total subscriptions, notifications, push subs) — the per-user and per-subscription lists are gone, and the dashboard no longer queries that PII at all. Backend: add a per-request access log (logs/access/access-.jsonl) recording the client IP (real IP via the left-most X-Forwarded-For hop when proxied behind Caddy, else the peer address), method, path, status, and category. Static assets and the dashboard's own metrics polling are skipped. Best-effort, never raises into the request path — retained for later traffic/IP analysis. * Dashboard: wrap-aware pager so the header stops scrolling off (#141) The pager counted one screen row per logical line, but on a narrow phone many lines wrap to two rows — so a page printed more physical rows than the terminal had, scrolling the top (title + CACHE header) off-screen with no way to scroll back to it. Measure each line's wrapped height and fit a page to the real physical-row budget; page/End step by what's actually visible; read the true tty size (not shutil, which honors stale COLUMNS/LINES). Status line is clipped to one row. * Add log storage size to the ops dashboard STORAGE section (#142) The STORAGE section reported parquet archives and both SQLite DBs but omitted the logs/ footprint, so the "total" understated real disk use. Add a recursive _tree_size helper and a "logs" row measuring the whole logs/ tree (audit + errors + access JSONL plus stray *.log files), broken out by stream, and fold it into the STORAGE total. Flows through --json automatically via read_storage. * Dashboard: show last-10-minute traffic per category (#143) Add a rolling short-window view alongside the since-start totals. metrics.py keeps per-minute buckets (epoch-minute -> {key: count}) for inbound categories and outbound sources, summed over the last 10 minutes on snapshot() and exposed as inbound_recent / outbound_recent / window_minutes. Cheap and bounded — at most WINDOW+1 tiny dicts, pruned as minutes roll off; self-polling still excluded. The dashboard's TRAFFIC section now renders that count as a dim column next to each row's total (header reads "total · last 10m"); idle sources show "-". * Add dashboard #2 (#145) * Auto-submit IndexNow on deploy when the URL set changes (#130) Hook the production deploy (deploy/deploy.sh) to ping IndexNow after the health check passes, so new/changed pages reach Bing/DuckDuckGo/Yandex without a manual run. Best-effort — wrapped so it can never fail a deploy — and it sources /etc/thermograph.env so its key matches the one the running service serves. To avoid re-blasting ~14k URLs on every code push, indexnow.py gains a --if-changed mode gated by a signature of the URL set (persisted to data/indexnow_state.txt): code-only deploys skip; a deploy that adds or removes a city submits. `make indexnow` still forces a full submit. * Add ops metrics endpoint + SSH/Termux dashboard (#131) Adds observability for the running app, viewable over SSH (e.g. Termux): - backend/metrics.py: thread-safe, best-effort in-process counters for inbound requests (per category) and outbound calls (per external source), plus a phase->source map and snapshot(). Hooked into climate._request (outbound, all six weather/geocode sources) and the existing revalidate_static middleware (inbound). Never raises into the request path. - GET /api/v2/metrics: token-gated JSON of the counters + warm-queue depth + thread names/uptime/pid. Returns 404 unless a valid THERMOGRAPH_METRICS_TOKEN is presented or the request is direct loopback with no proxy X-Forwarded-* header, so ops data is never exposed through Caddy on the public domain. - scripts/dashboard.py (+ scripts/dash, make dashboard): stdlib-only terminal dashboard. Reads cache (parquet + SQLite), accounts (users + active subscriptions), and system resources (/proc + systemd) directly, and pulls traffic from the metrics endpoint over loopback. Live refresh, --once, --json; mobile-terminal width. Paths resolve relative to the script so the same tool works on the LAN dev and prod VPS checkouts. - Monitoring sections in DEPLOY.md / DEPLOY-DEV.md; unit tests for the counters. * Dashboard: run under the app venv, not the box's default python3 (#132) The box's default python3 can be a build without _sqlite3 (pyenv 3.10 here), so ./scripts/dash crashed on import. Prefer the repo's .venv interpreter (the Python the server itself runs on, which always has sqlite3); fall back to python3/python. Also make the sqlite3 import in dashboard.py optional so a fallback interpreter degrades to empty cache/accounts panels instead of crashing, and point the 'make dashboard' target at the wrapper. * Dashboard: redraw live view in place so it stops scrolling to the bottom (#133) The live loop did a full clear-and-reprint each refresh; when the output is taller than the terminal (common on a phone) that scrolls the view to the bottom every tick. Redraw from the top instead: hide the cursor, clear once, then each frame move to home and rewrite line-by-line (clear-to-EOL + clear-below), capping the body to the window height so it never prints more lines than fit and never scrolls. Restore the cursor on exit; piped (non-TTY) output prints plain frames. * Dashboard: make the live view a scrollable in-place pager (#134) The live mode is now an alt-screen pager (like top/less): it shows only the latest snapshot, you scroll it with swipe/arrows/jk/PgUp-PgDn/g/G, and each refresh replaces the page in place while keeping your scroll position — nothing gets appended below and the view is never yanked to the top or bottom. On exit it restores the cursor, leaves the alternate screen, and resets the terminal mode. --once still prints a single static snapshot (also used automatically when stdout/stdin aren't a terminal, e.g. piping); --json unchanged. * Metrics: don't count the dashboard's own /api/v2/metrics polling (#135) The ops dashboard polls the metrics endpoint every few seconds; counting those hits just shows the monitor watching itself and inflates the inbound totals. Skip the 'metrics' category in record_inbound (so it's excluded from inbound_total too), and hide it in the dashboard's inbound list defensively. * Dashboard: add a STORAGE section (archive + DB sizes) (#136) Show on-disk storage: the parquet archives (with history/recent split), the derived SQLite DB, and the accounts SQLite DB individually, plus a total. Archive bytes are summed during the existing TTL-cached cache scan; DB sizes include the -wal/-shm sidecars (WAL holds recent writes, so the base file alone understates the footprint). Drop the now-redundant 'payload db' line from CACHE. * Dashboard: show STORAGE sizes in fixed GB (0.01 GB floor) (#137) The auto-scaling formatter only switched to GB once a value passed 1 GB, so the archives read as MB and the DBs as KB. Format the STORAGE section in fixed GB with two decimals and a 0.01 GB floor (so the small databases stay visible instead of rounding to 0.00). Memory readouts keep the auto-scaling formatter. * Dashboard: show STORAGE under 50 MB in MB, GB above (#138) Values below 50 MB now render as MB with one decimal (no KB), so the databases read naturally (derived 7.7MB, accounts 0.4MB) instead of 0.01GB; 50 MB and up stay in GB with two decimals. Renames the helper _gb -> _storage_size to match. * Dashboard: move STORAGE MB->GB cutoff to 500 MB (#139) Values below 500 MB show in MB (one decimal); 500 MB and up in GB. * Dashboard: accounts totals only; log client IPs per request (#140) Dashboard ACCOUNTS section now shows only totals (users, active/total subscriptions, notifications, push subs) — the per-user and per-subscription lists are gone, and the dashboard no longer queries that PII at all. Backend: add a per-request access log (logs/access/access-.jsonl) recording the client IP (real IP via the left-most X-Forwarded-For hop when proxied behind Caddy, else the peer address), method, path, status, and category. Static assets and the dashboard's own metrics polling are skipped. Best-effort, never raises into the request path — retained for later traffic/IP analysis. * Dashboard: wrap-aware pager so the header stops scrolling off (#141) The pager counted one screen row per logical line, but on a narrow phone many lines wrap to two rows — so a page printed more physical rows than the terminal had, scrolling the top (title + CACHE header) off-screen with no way to scroll back to it. Measure each line's wrapped height and fit a page to the real physical-row budget; page/End step by what's actually visible; read the true tty size (not shutil, which honors stale COLUMNS/LINES). Status line is clipped to one row. * Add log storage size to the ops dashboard STORAGE section (#142) The STORAGE section reported parquet archives and both SQLite DBs but omitted the logs/ footprint, so the "total" understated real disk use. Add a recursive _tree_size helper and a "logs" row measuring the whole logs/ tree (audit + errors + access JSONL plus stray *.log files), broken out by stream, and fold it into the STORAGE total. Flows through --json automatically via read_storage. * Dashboard: show last-10-minute traffic per category (#143) Add a rolling short-window view alongside the since-start totals. metrics.py keeps per-minute buckets (epoch-minute -> {key: count}) for inbound categories and outbound sources, summed over the last 10 minutes on snapshot() and exposed as inbound_recent / outbound_recent / window_minutes. Cheap and bounded — at most WINDOW+1 tiny dicts, pruned as minutes roll off; self-polling still excluded. The dashboard's TRAFFIC section now renders that count as a dim column next to each row's total (header reads "total · last 10m"); idle sources show "-". * Run prod on 3 uvicorn workers with a shared metrics store A single slow upstream weather fetch (cache-miss during an open-meteo/NASA degradation) blocked the one uvicorn worker and took the whole app down for ~2 min — real users got aborted connections. Give prod concurrency headroom so one slow request can't freeze the rest. - deploy/thermograph.service: worker count is env-driven (`--workers ${WORKERS}`, default 1); prod sets WORKERS=3 in /etc/thermograph.env. Dev's separate --user unit is untouched (stays single-worker). - metrics.py: with multiple workers, per-process counters would split the tally and the ops dashboard (polls one random worker) would see only a fraction. Add a shared SQLite store selected by THERMOGRAPH_METRICS_DB so every worker tallies into one place; unset keeps the zero-dependency in-memory store for dev/tests/single-worker. Public API and snapshot shape unchanged. - The unit defaults THERMOGRAPH_METRICS_DB and clears it on each (re)start via ExecStartPre, so bumping WORKERS can never silently fragment the dashboard and "since start" tallies keep their old meaning. - Tests: cover the SQLite store — cross-worker aggregation, window roll-off, and env-based selection. Full suite: 178 passed. Note: applying to prod also needs the unit reinstalled + WORKERS=3 in /etc/thermograph.env — deploy.sh only pulls+restarts, it doesn't reinstall the systemd unit. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_016mRqdHWRdUjEvEdfwig3wg --------- Co-authored-by: root Co-authored-by: Claude Opus 4.8 --- metrics.py | 295 ++++++++++++++++++++++++++++++++---------- tests/test_metrics.py | 67 ++++++++++ 2 files changed, 297 insertions(+), 65 deletions(-) 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