"""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.""" # The metrics endpoint is polled by the dashboard itself — counting it would just # show the monitor watching itself, so it's excluded from traffic entirely. 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 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()), }