"""Health/alert evaluation shared by /api/v1/health, /metrics and the console.""" from __future__ import annotations from datetime import UTC, datetime from typing import Any from findata_core.config import get_settings from findata_core.db import connection from findata_ingest.lake import Lake from findata_api import metrics as m _cache: dict[str, Any] = {"at": 0.0, "value": None} _refresher_started = False def _refresh_loop(period_s: float) -> None: import time while True: try: v = evaluate() _cache.update(at=time.monotonic(), value=v) except Exception as e: _cache.update(at=time.monotonic(), value={"status": "degraded", "checked_at": datetime.now(UTC).isoformat(), "alerts": [{"severity": "error", "code": "health_eval_failed", "message": str(e)[:200]}], "components": {}}) time.sleep(period_s) def start_refresher(period_s: float = 60.0) -> None: """Health is computed by a background thread; request handlers only read the cache (never block on DB/disk).""" global _refresher_started if _refresher_started: return import threading _refresher_started = True threading.Thread(target=_refresh_loop, args=(period_s,), name="health-refresher", daemon=True).start() def cached_evaluate(max_age_s: float = 60.0) -> dict[str, Any]: start_refresher() v = _cache["value"] if v is None: return {"status": "starting", "checked_at": datetime.now(UTC).isoformat(), "alerts": [], "components": {}, "datasets": []} return v def lake_usage_bytes(conn) -> dict[str, int]: """Lake footprint from the catalog (no filesystem walk): live curated bytes, retired bytes, raw archive bytes.""" r = conn.execute("SELECT coalesce(sum(bytes) FILTER (WHERE version_removed IS NULL),0) AS live, " "coalesce(sum(bytes) FILTER (WHERE version_removed IS NOT NULL),0) AS retired FROM lake_files").fetchone() a = conn.execute("SELECT coalesce(sum(bytes),0) AS raw FROM raw_archives WHERE path IS NOT NULL").fetchone() return {"curated": int(r["live"]), "retired": int(r["retired"]), "raw": int(a["raw"])} def evaluate() -> dict[str, Any]: s = get_settings() lake = Lake() usage = {"curated": 0, "retired": 0, "raw": 0} out: dict[str, Any] = {"status": "ok", "checked_at": datetime.now(UTC).isoformat(), "alerts": [], "components": {}} try: with connection() as conn: conn.execute("SELECT 1") out["components"]["postgres"] = "ok" jobs = {r["status"]: r["n"] for r in conn.execute("SELECT status, count(*) n FROM ingest_jobs GROUP BY status").fetchall()} for st in ("queued", "running", "succeeded", "failed", "dead"): m.JOBS.labels(st).set(jobs.get(st, 0)) ds = conn.execute("SELECT d.id, d.status, d.current_version, (SELECT max(session_date) FROM coverage c WHERE c.dataset_id=d.id) last_session, " "(SELECT value FROM checkpoints WHERE scope='dataset:'||d.id AND key='last_vendor_update:day') ck " "FROM datasets d WHERE d.status='active' AND d.kind IN ('market_bars','derived')").fetchall() out["datasets"] = [] for d in ds: last = d["last_session"] fresh = (datetime.now(UTC).date() - last).days if last else None vendor = (d["ck"] or {}).get("vendor_last_update") if d["ck"] else None lag = None if vendor and last: lag = (datetime.fromisoformat(vendor).date() - last).days out["datasets"].append({"dataset_id": d["id"], "version": d["current_version"], "last_session": str(last) if last else None, "days_since_last_session": fresh, "vendor_last_update": vendor, "vendor_lag_days": lag}) if fresh is not None: m.FRESHNESS.labels(d["id"]).set(fresh) if lag is not None: m.VENDOR_LAG.labels(d["id"]).set(lag) if lag > 0: out["alerts"].append({"severity": "warning", "code": "stale_dataset", "message": f"{d['id']} is {lag} day(s) behind the vendor"}) if fresh is not None and fresh > 5: out["alerts"].append({"severity": "warning", "code": "no_recent_session", "message": f"{d['id']}: last session {last} ({fresh} days ago)"}) dead = jobs.get("dead", 0) if dead: out["alerts"].append({"severity": "error", "code": "dead_jobs", "message": f"{dead} job(s) in the dead-letter queue"}) lb = conn.execute("SELECT value, updated_at FROM settings_kv WHERE key='last_backup'").fetchone() if lb: age_h = (datetime.now(UTC) - lb["updated_at"]).total_seconds() / 3600 m.BACKUP_AGE.set(age_h) out["components"]["backup"] = {"last": lb["updated_at"].isoformat(), "age_hours": round(age_h, 1), "verified": lb["value"].get("verified")} if age_h > 48: out["alerts"].append({"severity": "warning", "code": "backup_stale", "message": f"last backup {age_h:.0f} h ago"}) else: out["components"]["backup"] = {"last": None} out["alerts"].append({"severity": "info", "code": "no_backup_yet", "message": "no backup recorded yet"}) lock = lake.root / ".writer.lock" worker_alive = False if lock.exists(): try: import fcntl fh = open(lock) try: fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB) fcntl.flock(fh, fcntl.LOCK_UN) except BlockingIOError: worker_alive = True finally: fh.close() except OSError: pass out["components"]["ingest_worker"] = "running" if worker_alive else "not running" if not worker_alive: out["alerts"].append({"severity": "error", "code": "worker_down", "message": "ingest worker is not holding the lake lock"}) an = conn.execute("SELECT dataset_id, severity, count(*) n FROM quality_anomalies WHERE created_at > now() - interval '7 days' GROUP BY 1,2").fetchall() for a in an: m.ANOMALIES.labels(a["dataset_id"], a["severity"]).set(a["n"]) usage = lake_usage_bytes(conn) except Exception as e: out["components"]["postgres"] = f"error: {e}" out["alerts"].append({"severity": "error", "code": "postgres_down", "message": str(e)[:200]}) free = lake.free_gb() m.LAKE_FREE.set(free) curated = usage["curated"] m.LAKE_CURATED.set(curated) out["components"]["lake"] = {"root": str(lake.root), "free_gb": round(free, 1), "curated_gb": round(curated / 1e9, 2), "raw_gb": round(usage["raw"] / 1e9, 2), "retired_gb": round(usage["retired"] / 1e9, 2), "soft_limit_gb": s.lake_soft_limit_gb, "hard_min_free_gb": s.lake_hard_min_free_gb, "cold_root": str(lake.cold) if lake.cold else None, "cold_free_gb": round(lake.free_gb(lake.cold), 1) if lake.cold else None} if lake.cold and lake.free_gb(lake.cold) < s.lake_cold_hard_min_free_gb: out["alerts"].append({"severity": "error", "code": "cold_disk_critical", "message": f"external SSD: only {lake.free_gb(lake.cold):.0f} GB free"}) if free < s.lake_hard_min_free_gb: out["alerts"].append({"severity": "error", "code": "disk_critical", "message": f"only {free:.0f} GB free"}) elif free < s.lake_hard_min_free_gb * 2: out["alerts"].append({"severity": "warning", "code": "disk_low", "message": f"{free:.0f} GB free"}) if curated / 1e9 > s.lake_soft_limit_gb: out["alerts"].append({"severity": "warning", "code": "lake_over_budget", "message": f"curated {curated/1e9:.0f} GB > {s.lake_soft_limit_gb} GB"}) try: import subprocess r = subprocess.run(["diskutil", "info", "disk0"], capture_output=True, text=True, timeout=10) smart = next((ln.split(":", 1)[1].strip() for ln in r.stdout.splitlines() if "SMART Status" in ln), "unknown") out["components"]["ssd_smart"] = smart if smart not in ("Verified", "Not Supported", "unknown"): out["alerts"].append({"severity": "error", "code": "ssd_smart", "message": f"SMART status: {smart}"}) except Exception: out["components"]["ssd_smart"] = "unavailable" if any(a["severity"] == "error" for a in out["alerts"]): out["status"] = "degraded" elif out["alerts"]: out["status"] = "warning" return out