From d6a62cc2de59b1181ab92dd6a3a387186a8e0175 Mon Sep 17 00:00:00 2001 From: Emi Griffith Date: Thu, 16 Jul 2026 13:46:59 -0700 Subject: [PATCH] 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. --- app.py | 35 +++++++++++ climate.py | 4 ++ metrics.py | 138 ++++++++++++++++++++++++++++++++++++++++++ tests/test_metrics.py | 81 +++++++++++++++++++++++++ 4 files changed, 258 insertions(+) create mode 100644 metrics.py create mode 100644 tests/test_metrics.py diff --git a/app.py b/app.py index d29c2df..d985314 100644 --- a/app.py +++ b/app.py @@ -3,6 +3,7 @@ import contextlib import datetime import functools import hashlib +import hmac import json import mimetypes import os @@ -21,6 +22,7 @@ import climate import content import db import grid +import metrics import notify import places import store @@ -149,6 +151,10 @@ async def revalidate_static(request, call_next): deploy immediately (no "my change isn't showing" bugs).""" response = await call_next(request) path = request.url.path + try: + metrics.record_inbound(metrics.classify_inbound(path, BASE), response.status_code) + except Exception: # noqa: BLE001 - never let instrumentation break a response + pass pages = (BASE, f"{BASE}/", f"{BASE}/calendar", f"{BASE}/day", f"{BASE}/compare", f"{BASE}/legend", f"{BASE}/alerts") if path.endswith((".js", ".css", ".html")) or path in pages: response.headers["Cache-Control"] = "no-cache" @@ -537,6 +543,34 @@ def api_cell( return _json_response(_encode(payload), etag) +# --- ops metrics (gated) --------------------------------------------------- +def _metrics_allowed(request: Request) -> bool: + """Guard the ops metrics: a valid token allows from anywhere (for SSH tunnels); + with no token it's direct-loopback-only, and any forwarded header (i.e. proxied + via Caddy) is refused — so prod ops data never leaks through the public domain.""" + token = os.environ.get("THERMOGRAPH_METRICS_TOKEN", "").strip() + if token: + supplied = request.headers.get("x-metrics-token") or request.query_params.get("token") or "" + if hmac.compare_digest(supplied, token): + return True + client = request.client.host if request.client else None + forwarded = any(h in request.headers for h in + ("x-forwarded-for", "x-forwarded-host", "x-forwarded-proto")) + return client in ("127.0.0.1", "::1") and not forwarded + + +def api_metrics(request: Request): + """Live in-process traffic counters + worker facts for the ops dashboard.""" + if not _metrics_allowed(request): + raise HTTPException(status_code=404) # 404, not 403 — don't reveal the route + snap = metrics.snapshot() + snap["warm_queue_depth"] = _warm_queue.qsize() + snap["warm_seen"] = len(_WARM_SEEN) + snap["threads"] = threading.active_count() + snap["thread_names"] = sorted(t.name for t in threading.enumerate()) + return snap + + # --- API versioning -------------------------------------------------------- # Non-backward-compatible additions land in a new version; every version stays # mounted and served simultaneously. v1 is the original contract (grade/geocode); @@ -555,6 +589,7 @@ v2.add_api_route("/calendar", api_calendar, methods=["GET"]) v2.add_api_route("/day", api_day, methods=["GET"]) v2.add_api_route("/forecast", api_forecast, methods=["GET"]) v2.add_api_route("/cell", api_cell, methods=["GET"]) +v2.add_api_route("/metrics", api_metrics, methods=["GET"], include_in_schema=False) app.include_router(v1, prefix=f"{BASE}/api") # legacy unversioned == v1 (backward compatible) app.include_router(v1, prefix=f"{BASE}/api/v1") diff --git a/climate.py b/climate.py index 8be4633..d2415a1 100644 --- a/climate.py +++ b/climate.py @@ -15,6 +15,7 @@ import numpy as np import polars as pl import audit +import metrics import store CACHE_DIR = os.path.join(os.path.dirname(__file__), "..", "data", "cache") @@ -154,12 +155,15 @@ def _request(url, params, timeout, *, phase, headers=None, attempts=MAX_ATTEMPTS try: r = httpx.get(url, params=params, timeout=timeout, headers=headers) r.raise_for_status() + metrics.record_outbound(phase, "ok") return r except Exception as e: # noqa: BLE001 - upstream/network failures are expected last = e status = getattr(getattr(e, "response", None), "status_code", None) rate_limited = status == 429 final = rate_limited or attempt == attempts + metrics.record_outbound( + phase, "rate_limited" if rate_limited else ("error" if final else "retry")) audit.log_event( "error" if final else "retry", {"phase": phase, "attempt": attempt, "max_attempts": attempts, diff --git a/metrics.py b/metrics.py new file mode 100644 index 0000000..82d241f --- /dev/null +++ b/metrics.py @@ -0,0 +1,138 @@ +"""In-process 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. + +Read over the gated ``GET {BASE}/api/v2/metrics`` route (see ``app.py``) and rendered +by ``scripts/dashboard.py`` over SSH. +""" +import os +import threading +import time + +# The ``phase=`` tag passed to ``climate._request`` maps 1:1 onto an external source. +# Keep this in sync with the phase names used at each call site in ``climate.py``. +_PHASE_SOURCE = { + "history_fetch": "open-meteo-archive", + "history_topup": "open-meteo-archive", + "recent_forecast_fetch": "open-meteo-forecast", + "geocode": "open-meteo-geocoding", + "history_nasa": "nasa-power", + "forecast_metno": "met-norway", + "reverse_geocode": "nominatim", +} + +# Every source we know about, so the dashboard shows a stable set of rows even before +# any traffic has hit a given source. +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 +} + + +def source_for_phase(phase: str) -> str: + """Map a ``climate._request`` phase tag to an external-source label.""" + return _PHASE_SOURCE.get(phase, "other") + + +def _status_bucket(status) -> str: + try: + code = int(status) + except (TypeError, ValueError): + return "other" + if 200 <= code < 300: + return "2xx" + if 300 <= code < 400: + return "3xx" + if 400 <= code < 500: + return "4xx" + if 500 <= code < 600: + return "5xx" + return "other" + + +def classify_inbound(path: str, base: str = "") -> str: + """Bucket a request path into a dashboard category. + + ``path`` is ``request.url.path`` (still carrying the ``BASE`` prefix); ``base`` is + the app's mount prefix (e.g. ``/thermograph`` or ``""``). + """ + p = path or "/" + if base and p.startswith(base): + p = p[len(base):] or "/" + if p.startswith("/api/"): + rest = p[len("/api/"):] + if rest.startswith(("v1/", "v2/")): + rest = rest[3:] + seg = rest.split("/", 1)[0].split("?", 1)[0] + if seg in ("auth", "users"): + return "auth" + if seg in ("subscriptions", "notifications", "push"): + return "accounts" + if seg == "metrics": + return "metrics" + return f"api:{seg}" if seg else "api:other" + if p.endswith((".js", ".css", ".html", ".webmanifest", ".png", ".svg", ".ico", + ".json", ".woff", ".woff2", ".map", ".txt", ".xml")): + # robots.txt / sitemap.xml are crawler SEO endpoints, not page views. + if p in ("/robots.txt", "/sitemap.xml"): + return "seo" + return "static" + if p in ("", "/", "/calendar", "/day", "/compare", "/legend", "/alerts"): + return "page" + if p == "/about" or p.startswith("/climate") or p.startswith("/glossary"): + return "page" + return "other" + + +def record_inbound(category: str, status) -> None: + """Count one inbound request in ``category``, bucketed by status class.""" + 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 + except Exception: # noqa: BLE001 - instrumentation must never break a request + pass + + +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 + 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()} + return { + "pid": os.getpid(), + "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()), + } diff --git a/tests/test_metrics.py b/tests/test_metrics.py new file mode 100644 index 0000000..c7fd244 --- /dev/null +++ b/tests/test_metrics.py @@ -0,0 +1,81 @@ +"""Unit tests for the in-process traffic counters (metrics.py).""" +import importlib + +import metrics + + +def _fresh(): + # Counters are module-level; reload for an isolated starting state per test. + return importlib.reload(metrics) + + +def test_phase_source_map_covers_every_source(): + m = _fresh() + assert m.source_for_phase("history_fetch") == "open-meteo-archive" + assert m.source_for_phase("history_topup") == "open-meteo-archive" + assert m.source_for_phase("recent_forecast_fetch") == "open-meteo-forecast" + assert m.source_for_phase("geocode") == "open-meteo-geocoding" + assert m.source_for_phase("history_nasa") == "nasa-power" + assert m.source_for_phase("forecast_metno") == "met-norway" + assert m.source_for_phase("reverse_geocode") == "nominatim" + assert m.source_for_phase("unknown-phase") == "other" + + +def test_classify_inbound_categories(): + m = _fresh() + c = m.classify_inbound + # BASE stripped, API namespaces + assert c("/thermograph/api/v2/place", "/thermograph") == "api:place" + assert c("/thermograph/api/v2/grade", "/thermograph") == "api:grade" + assert c("/thermograph/api/v1/geocode", "/thermograph") == "api:geocode" + assert c("/thermograph/api/v2/auth/login", "/thermograph") == "auth" + assert c("/thermograph/api/v2/users/me", "/thermograph") == "auth" + assert c("/thermograph/api/v2/subscriptions", "/thermograph") == "accounts" + assert c("/thermograph/api/v2/notifications", "/thermograph") == "accounts" + assert c("/thermograph/api/v2/push/subscribe", "/thermograph") == "accounts" + assert c("/thermograph/api/v2/metrics", "/thermograph") == "metrics" + # pages, static, seo + assert c("/thermograph/", "/thermograph") == "page" + assert c("/thermograph/calendar", "/thermograph") == "page" + assert c("/thermograph/climate/tokyo", "/thermograph") == "page" + assert c("/thermograph/app.js", "/thermograph") == "static" + assert c("/thermograph/style.css", "/thermograph") == "static" + assert c("/thermograph/robots.txt", "/thermograph") == "seo" + # root-mounted (BASE == "") + assert c("/api/v2/place", "") == "api:place" + assert c("/", "") == "page" + + +def test_record_inbound_buckets_by_status(): + m = _fresh() + m.record_inbound("api:place", 200) + m.record_inbound("api:place", 200) + m.record_inbound("api:place", 503) + m.record_inbound("page", 304) + snap = m.snapshot() + assert snap["inbound"]["api:place"]["total"] == 3 + assert snap["inbound"]["api:place"]["2xx"] == 2 + assert snap["inbound"]["api:place"]["5xx"] == 1 + assert snap["inbound"]["page"]["3xx"] == 1 + assert snap["inbound_total"] == 4 + + +def test_record_outbound_by_source_and_outcome(): + m = _fresh() + m.record_outbound("history_fetch", "ok") + m.record_outbound("history_topup", "ok") # same source (archive) + m.record_outbound("reverse_geocode", "error") + m.record_outbound("recent_forecast_fetch", "rate_limited") + snap = m.snapshot() + assert snap["outbound"]["open-meteo-archive"]["ok"] == 2 + assert snap["outbound"]["nominatim"]["error"] == 1 + assert snap["outbound"]["open-meteo-forecast"]["rate_limited"] == 1 + assert snap["outbound_total"] == 4 + + +def test_snapshot_has_process_facts(): + m = _fresh() + snap = m.snapshot() + assert isinstance(snap["pid"], int) + assert snap["uptime_s"] >= 0 + assert "open-meteo-archive" in snap["outbound"] # stable source rows exist