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

@@ -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")

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,

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()),
+    }

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