1"""Health/alert evaluation shared by /api/v1/health, /metrics and the console."""2from __future__ import annotations34from datetime import UTC, datetime5from typing import Any67from findata_core.config import get_settings8from findata_core.db import connection9from findata_ingest.lake import Lake1011from findata_api import metrics as m1213_cache: dict[str, Any] = {"at": 0.0, "value": None}14_refresher_started = False151617def _refresh_loop(period_s: float) -> None:18 import time19 while True:20 try:21 v = evaluate()22 _cache.update(at=time.monotonic(), value=v)23 except Exception as e:24 _cache.update(at=time.monotonic(), value={"status": "degraded", "checked_at": datetime.now(UTC).isoformat(),25 "alerts": [{"severity": "error", "code": "health_eval_failed", "message": str(e)[:200]}],26 "components": {}})27 time.sleep(period_s)282930def start_refresher(period_s: float = 60.0) -> None:31 """Health is computed by a background thread; request handlers only read the cache (never block on DB/disk)."""32 global _refresher_started33 if _refresher_started:34 return35 import threading36 _refresher_started = True37 threading.Thread(target=_refresh_loop, args=(period_s,), name="health-refresher", daemon=True).start()383940def cached_evaluate(max_age_s: float = 60.0) -> dict[str, Any]:41 start_refresher()42 v = _cache["value"]43 if v is None:44 return {"status": "starting", "checked_at": datetime.now(UTC).isoformat(), "alerts": [], "components": {}, "datasets": []}45 return v464748def lake_usage_bytes(conn) -> dict[str, int]:49 """Lake footprint from the catalog (no filesystem walk): live curated bytes, retired bytes, raw archive bytes."""50 r = conn.execute("SELECT coalesce(sum(bytes) FILTER (WHERE version_removed IS NULL),0) AS live, "51 "coalesce(sum(bytes) FILTER (WHERE version_removed IS NOT NULL),0) AS retired FROM lake_files").fetchone()52 a = conn.execute("SELECT coalesce(sum(bytes),0) AS raw FROM raw_archives WHERE path IS NOT NULL").fetchone()53 return {"curated": int(r["live"]), "retired": int(r["retired"]), "raw": int(a["raw"])}545556def evaluate() -> dict[str, Any]:57 s = get_settings()58 lake = Lake()59 usage = {"curated": 0, "retired": 0, "raw": 0}60 out: dict[str, Any] = {"status": "ok", "checked_at": datetime.now(UTC).isoformat(), "alerts": [], "components": {}}61 try:62 with connection() as conn:63 conn.execute("SELECT 1")64 out["components"]["postgres"] = "ok"65 jobs = {r["status"]: r["n"] for r in conn.execute("SELECT status, count(*) n FROM ingest_jobs GROUP BY status").fetchall()}66 for st in ("queued", "running", "succeeded", "failed", "dead"):67 m.JOBS.labels(st).set(jobs.get(st, 0))68 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, "69 "(SELECT value FROM checkpoints WHERE scope='dataset:'||d.id AND key='last_vendor_update:day') ck "70 "FROM datasets d WHERE d.status='active' AND d.kind IN ('market_bars','derived')").fetchall()71 out["datasets"] = []72 for d in ds:73 last = d["last_session"]74 fresh = (datetime.now(UTC).date() - last).days if last else None75 vendor = (d["ck"] or {}).get("vendor_last_update") if d["ck"] else None76 lag = None77 if vendor and last:78 lag = (datetime.fromisoformat(vendor).date() - last).days79 out["datasets"].append({"dataset_id": d["id"], "version": d["current_version"], "last_session": str(last) if last else None,80 "days_since_last_session": fresh, "vendor_last_update": vendor, "vendor_lag_days": lag})81 if fresh is not None:82 m.FRESHNESS.labels(d["id"]).set(fresh)83 if lag is not None:84 m.VENDOR_LAG.labels(d["id"]).set(lag)85 if lag > 0:86 out["alerts"].append({"severity": "warning", "code": "stale_dataset", "message": f"{d['id']} is {lag} day(s) behind the vendor"})87 if fresh is not None and fresh > 5:88 out["alerts"].append({"severity": "warning", "code": "no_recent_session", "message": f"{d['id']}: last session {last} ({fresh} days ago)"})89 dead = jobs.get("dead", 0)90 if dead:91 out["alerts"].append({"severity": "error", "code": "dead_jobs", "message": f"{dead} job(s) in the dead-letter queue"})92 lb = conn.execute("SELECT value, updated_at FROM settings_kv WHERE key='last_backup'").fetchone()93 if lb:94 age_h = (datetime.now(UTC) - lb["updated_at"]).total_seconds() / 360095 m.BACKUP_AGE.set(age_h)96 out["components"]["backup"] = {"last": lb["updated_at"].isoformat(), "age_hours": round(age_h, 1), "verified": lb["value"].get("verified")}97 if age_h > 48:98 out["alerts"].append({"severity": "warning", "code": "backup_stale", "message": f"last backup {age_h:.0f} h ago"})99 else:100 out["components"]["backup"] = {"last": None}101 out["alerts"].append({"severity": "info", "code": "no_backup_yet", "message": "no backup recorded yet"})102 lock = lake.root / ".writer.lock"103 worker_alive = False104 if lock.exists():105 try:106 import fcntl107 fh = open(lock)108 try:109 fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB)110 fcntl.flock(fh, fcntl.LOCK_UN)111 except BlockingIOError:112 worker_alive = True113 finally:114 fh.close()115 except OSError:116 pass117 out["components"]["ingest_worker"] = "running" if worker_alive else "not running"118 if not worker_alive:119 out["alerts"].append({"severity": "error", "code": "worker_down", "message": "ingest worker is not holding the lake lock"})120 an = conn.execute("SELECT dataset_id, severity, count(*) n FROM quality_anomalies WHERE created_at > now() - interval '7 days' GROUP BY 1,2").fetchall()121 for a in an:122 m.ANOMALIES.labels(a["dataset_id"], a["severity"]).set(a["n"])123 usage = lake_usage_bytes(conn)124 except Exception as e:125 out["components"]["postgres"] = f"error: {e}"126 out["alerts"].append({"severity": "error", "code": "postgres_down", "message": str(e)[:200]})127 free = lake.free_gb()128 m.LAKE_FREE.set(free)129 curated = usage["curated"]130 m.LAKE_CURATED.set(curated)131 out["components"]["lake"] = {"root": str(lake.root), "free_gb": round(free, 1), "curated_gb": round(curated / 1e9, 2),132 "raw_gb": round(usage["raw"] / 1e9, 2), "retired_gb": round(usage["retired"] / 1e9, 2),133 "soft_limit_gb": s.lake_soft_limit_gb, "hard_min_free_gb": s.lake_hard_min_free_gb,134 "cold_root": str(lake.cold) if lake.cold else None,135 "cold_free_gb": round(lake.free_gb(lake.cold), 1) if lake.cold else None}136 if lake.cold and lake.free_gb(lake.cold) < s.lake_cold_hard_min_free_gb:137 out["alerts"].append({"severity": "error", "code": "cold_disk_critical", "message": f"external SSD: only {lake.free_gb(lake.cold):.0f} GB free"})138 if free < s.lake_hard_min_free_gb:139 out["alerts"].append({"severity": "error", "code": "disk_critical", "message": f"only {free:.0f} GB free"})140 elif free < s.lake_hard_min_free_gb * 2:141 out["alerts"].append({"severity": "warning", "code": "disk_low", "message": f"{free:.0f} GB free"})142 if curated / 1e9 > s.lake_soft_limit_gb:143 out["alerts"].append({"severity": "warning", "code": "lake_over_budget", "message": f"curated {curated/1e9:.0f} GB > {s.lake_soft_limit_gb} GB"})144 try:145 import subprocess146 r = subprocess.run(["diskutil", "info", "disk0"], capture_output=True, text=True, timeout=10)147 smart = next((ln.split(":", 1)[1].strip() for ln in r.stdout.splitlines() if "SMART Status" in ln), "unknown")148 out["components"]["ssd_smart"] = smart149 if smart not in ("Verified", "Not Supported", "unknown"):150 out["alerts"].append({"severity": "error", "code": "ssd_smart", "message": f"SMART status: {smart}"})151 except Exception:152 out["components"]["ssd_smart"] = "unavailable"153 if any(a["severity"] == "error" for a in out["alerts"]):154 out["status"] = "degraded"155 elif out["alerts"]:156 out["status"] = "warning"157 return out158