Sync main into dev (multi-worker prod + shared metrics store) (#160)
* 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-<date>.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-<date>.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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016mRqdHWRdUjEvEdfwig3wg --------- Co-authored-by: root <root@vmi3417050.contaboserver.net> Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Emi Griffith
GitHub
parent
4dbb060d4d
commit
059f459e8f
2 changed files with 297 additions and 65 deletions
295
metrics.py
295
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
|
Cheap, best-effort counters for **inbound** HTTP requests (bucketed by category) and
|
||||||
category) and **outbound** calls to external data sources (bucketed by source).
|
**outbound** calls to external data sources (bucketed by source). Everything here is
|
||||||
Everything here is since-process-start; durable daily history lives in the
|
since-start; durable daily history lives in the audit/error JSONL logs (see
|
||||||
audit/error JSONL logs (see ``audit.py``). Recording never raises into the request
|
``audit.py``). Recording never raises into the request path.
|
||||||
path.
|
|
||||||
|
|
||||||
Read over the gated ``GET {BASE}/api/v2/metrics`` route (see ``app.py``) and rendered
|
Read over the gated ``GET {BASE}/api/v2/metrics`` route (see ``app.py``) and rendered
|
||||||
by ``scripts/dashboard.py`` over SSH.
|
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 os
|
||||||
|
import sqlite3
|
||||||
import threading
|
import threading
|
||||||
import time
|
import time
|
||||||
|
|
||||||
|
|
@ -32,42 +39,9 @@ SOURCES = sorted(set(_PHASE_SOURCE.values()))
|
||||||
_OUTCOMES = ("ok", "retry", "error", "rate_limited")
|
_OUTCOMES = ("ok", "retry", "error", "rate_limited")
|
||||||
_STATUS_BUCKETS = ("2xx", "3xx", "4xx", "5xx", "other")
|
_STATUS_BUCKETS = ("2xx", "3xx", "4xx", "5xx", "other")
|
||||||
|
|
||||||
_lock = threading.Lock()
|
# Rolling short-window view: per-minute buckets summed over the last WINDOW minutes on
|
||||||
_started_at = time.time()
|
# snapshot(). Cheap and bounded — minutes are pruned as they roll out of the window.
|
||||||
_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.
|
|
||||||
WINDOW_MINUTES = 10
|
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:
|
def source_for_phase(phase: str) -> str:
|
||||||
|
|
@ -130,6 +104,207 @@ def classify_inbound(path: str, base: str = "") -> str:
|
||||||
return "other"
|
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:
|
def record_inbound(category: str, status) -> None:
|
||||||
"""Count one inbound request in ``category``, bucketed by status class."""
|
"""Count one inbound request in ``category``, bucketed by status class."""
|
||||||
# The metrics endpoint is polled by the dashboard itself — counting it would just
|
# 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":
|
if category == "metrics":
|
||||||
return
|
return
|
||||||
try:
|
try:
|
||||||
bucket = _status_bucket(status)
|
_store.record_inbound(category, _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)
|
|
||||||
except Exception: # noqa: BLE001 - instrumentation must never break a request
|
except Exception: # noqa: BLE001 - instrumentation must never break a request
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
|
@ -152,34 +320,31 @@ def record_inbound(category: str, status) -> None:
|
||||||
def record_outbound(phase: str, outcome: str) -> None:
|
def record_outbound(phase: str, outcome: str) -> None:
|
||||||
"""Count one outbound call. ``outcome`` in {ok, retry, error, rate_limited}."""
|
"""Count one outbound call. ``outcome`` in {ok, retry, error, rate_limited}."""
|
||||||
try:
|
try:
|
||||||
source = source_for_phase(phase)
|
_store.record_outbound(source_for_phase(phase), outcome)
|
||||||
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)
|
|
||||||
except Exception: # noqa: BLE001
|
except Exception: # noqa: BLE001
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
|
||||||
def snapshot() -> dict:
|
def snapshot() -> dict:
|
||||||
"""A JSON-serializable copy of all counters plus process facts."""
|
"""A JSON-serializable copy of all counters plus process facts."""
|
||||||
with _lock:
|
try:
|
||||||
inbound = {k: dict(v) for k, v in _inbound.items()}
|
data = _store.read()
|
||||||
outbound = {k: dict(v) for k, v in _outbound.items()}
|
except Exception: # noqa: BLE001 - a locked/broken store must not 500 the dashboard
|
||||||
recent_in = _recent_totals(_recent_in)
|
data = {"started_at": time.time(), "inbound": {}, "outbound": {},
|
||||||
recent_out = _recent_totals(_recent_out)
|
"inbound_recent": {}, "outbound_recent": {}}
|
||||||
|
inbound = data["inbound"]
|
||||||
|
outbound = data["outbound"]
|
||||||
|
started_at = data["started_at"]
|
||||||
return {
|
return {
|
||||||
"pid": os.getpid(),
|
"pid": os.getpid(),
|
||||||
"started_at": _started_at,
|
"started_at": started_at,
|
||||||
"uptime_s": round(time.time() - _started_at, 1),
|
"uptime_s": round(time.time() - started_at, 1),
|
||||||
"inbound": inbound,
|
"inbound": inbound,
|
||||||
"outbound": outbound,
|
"outbound": outbound,
|
||||||
"inbound_total": sum(v.get("total", 0) for v in inbound.values()),
|
"inbound_total": sum(v.get("total", 0) for v in inbound.values()),
|
||||||
"outbound_total": sum(sum(v.values()) for v in outbound.values()),
|
"outbound_total": sum(sum(v.values()) for v in outbound.values()),
|
||||||
# Traffic over the last WINDOW_MINUTES, per category / source.
|
# Traffic over the last WINDOW_MINUTES, per category / source.
|
||||||
"window_minutes": WINDOW_MINUTES,
|
"window_minutes": WINDOW_MINUTES,
|
||||||
"inbound_recent": recent_in,
|
"inbound_recent": data["inbound_recent"],
|
||||||
"outbound_recent": recent_out,
|
"outbound_recent": data["outbound_recent"],
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -132,3 +132,70 @@ def test_recent_excludes_self_polling():
|
||||||
snap = m.snapshot()
|
snap = m.snapshot()
|
||||||
assert "metrics" not in snap["inbound_recent"]
|
assert "metrics" not in snap["inbound_recent"]
|
||||||
assert snap["inbound_recent"]["page"] == 1
|
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
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue