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_lock = __import__("threading").Lock()151617def cached_evaluate(max_age_s: float = 60.0) -> dict[str, Any]:18 """Health evaluation is served from a 60 s cache so that probes (mld heal, agentd, Prometheus) stay cheap19 even while hundreds of thousands of files are being published."""20 import time21 now = time.monotonic()22 if _cache["value"] is not None and now - _cache["at"] < max_age_s:23 return _cache["value"]24 with _lock:25 if _cache["value"] is not None and time.monotonic() - _cache["at"] < max_age_s:26 return _cache["value"]27 v = evaluate()28 _cache.update(at=time.monotonic(), value=v)29 return v303132def lake_usage_bytes(conn) -> dict[str, int]:33 """Lake footprint from the catalog (no filesystem walk): live curated bytes, retired bytes, raw archive bytes."""34 r = conn.execute("SELECT coalesce(sum(bytes) FILTER (WHERE version_removed IS NULL),0) AS live, "35 "coalesce(sum(bytes) FILTER (WHERE version_removed IS NOT NULL),0) AS retired FROM lake_files").fetchone()36 a = conn.execute("SELECT coalesce(sum(bytes),0) AS raw FROM raw_archives WHERE path IS NOT NULL").fetchone()37 return {"curated": int(r["live"]), "retired": int(r["retired"]), "raw": int(a["raw"])}383940def evaluate() -> dict[str, Any]:41 s = get_settings()42 lake = Lake()43 usage = {"curated": 0, "retired": 0, "raw": 0}44 out: dict[str, Any] = {"status": "ok", "checked_at": datetime.now(UTC).isoformat(), "alerts": [], "components": {}}45 try:46 with connection() as conn:47 conn.execute("SELECT 1")48 out["components"]["postgres"] = "ok"49 jobs = {r["status"]: r["n"] for r in conn.execute("SELECT status, count(*) n FROM ingest_jobs GROUP BY status").fetchall()}50 for st in ("queued", "running", "succeeded", "failed", "dead"):51 m.JOBS.labels(st).set(jobs.get(st, 0))52 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, "53 "(SELECT value FROM checkpoints WHERE scope='dataset:'||d.id AND key='last_vendor_update:day') ck "54 "FROM datasets d WHERE d.status='active' AND d.kind IN ('market_bars','derived')").fetchall()55 out["datasets"] = []56 for d in ds:57 last = d["last_session"]58 fresh = (datetime.now(UTC).date() - last).days if last else None59 vendor = (d["ck"] or {}).get("vendor_last_update") if d["ck"] else None60 lag = None61 if vendor and last:62 lag = (datetime.fromisoformat(vendor).date() - last).days63 out["datasets"].append({"dataset_id": d["id"], "version": d["current_version"], "last_session": str(last) if last else None,64 "days_since_last_session": fresh, "vendor_last_update": vendor, "vendor_lag_days": lag})65 if fresh is not None:66 m.FRESHNESS.labels(d["id"]).set(fresh)67 if lag is not None:68 m.VENDOR_LAG.labels(d["id"]).set(lag)69 if lag > 0:70 out["alerts"].append({"severity": "warning", "code": "stale_dataset", "message": f"{d['id']} is {lag} day(s) behind the vendor"})71 if fresh is not None and fresh > 5:72 out["alerts"].append({"severity": "warning", "code": "no_recent_session", "message": f"{d['id']}: last session {last} ({fresh} days ago)"})73 dead = jobs.get("dead", 0)74 if dead:75 out["alerts"].append({"severity": "error", "code": "dead_jobs", "message": f"{dead} job(s) in the dead-letter queue"})76 lb = conn.execute("SELECT value, updated_at FROM settings_kv WHERE key='last_backup'").fetchone()77 if lb:78 age_h = (datetime.now(UTC) - lb["updated_at"]).total_seconds() / 360079 m.BACKUP_AGE.set(age_h)80 out["components"]["backup"] = {"last": lb["updated_at"].isoformat(), "age_hours": round(age_h, 1), "verified": lb["value"].get("verified")}81 if age_h > 48:82 out["alerts"].append({"severity": "warning", "code": "backup_stale", "message": f"last backup {age_h:.0f} h ago"})83 else:84 out["components"]["backup"] = {"last": None}85 out["alerts"].append({"severity": "info", "code": "no_backup_yet", "message": "no backup recorded yet"})86 lock = lake.root / ".writer.lock"87 worker_alive = False88 if lock.exists():89 try:90 import fcntl91 fh = open(lock)92 try:93 fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB)94 fcntl.flock(fh, fcntl.LOCK_UN)95 except BlockingIOError:96 worker_alive = True97 finally:98 fh.close()99 except OSError:100 pass101 out["components"]["ingest_worker"] = "running" if worker_alive else "not running"102 if not worker_alive:103 out["alerts"].append({"severity": "error", "code": "worker_down", "message": "ingest worker is not holding the lake lock"})104 an = conn.execute("SELECT dataset_id, severity, count(*) n FROM quality_anomalies WHERE created_at > now() - interval '7 days' GROUP BY 1,2").fetchall()105 for a in an:106 m.ANOMALIES.labels(a["dataset_id"], a["severity"]).set(a["n"])107 usage = lake_usage_bytes(conn)108 except Exception as e:109 out["components"]["postgres"] = f"error: {e}"110 out["alerts"].append({"severity": "error", "code": "postgres_down", "message": str(e)[:200]})111 free = lake.free_gb()112 m.LAKE_FREE.set(free)113 curated = usage["curated"]114 m.LAKE_CURATED.set(curated)115 out["components"]["lake"] = {"root": str(lake.root), "free_gb": round(free, 1), "curated_gb": round(curated / 1e9, 2),116 "raw_gb": round(usage["raw"] / 1e9, 2), "retired_gb": round(usage["retired"] / 1e9, 2),117 "soft_limit_gb": s.lake_soft_limit_gb, "hard_min_free_gb": s.lake_hard_min_free_gb,118 "cold_root": str(lake.cold) if lake.cold else None,119 "cold_free_gb": round(lake.free_gb(lake.cold), 1) if lake.cold else None}120 if lake.cold and lake.free_gb(lake.cold) < s.lake_cold_hard_min_free_gb:121 out["alerts"].append({"severity": "error", "code": "cold_disk_critical", "message": f"external SSD: only {lake.free_gb(lake.cold):.0f} GB free"})122 if free < s.lake_hard_min_free_gb:123 out["alerts"].append({"severity": "error", "code": "disk_critical", "message": f"only {free:.0f} GB free"})124 elif free < s.lake_hard_min_free_gb * 2:125 out["alerts"].append({"severity": "warning", "code": "disk_low", "message": f"{free:.0f} GB free"})126 if curated / 1e9 > s.lake_soft_limit_gb:127 out["alerts"].append({"severity": "warning", "code": "lake_over_budget", "message": f"curated {curated/1e9:.0f} GB > {s.lake_soft_limit_gb} GB"})128 try:129 import subprocess130 r = subprocess.run(["diskutil", "info", "disk0"], capture_output=True, text=True, timeout=10)131 smart = next((ln.split(":", 1)[1].strip() for ln in r.stdout.splitlines() if "SMART Status" in ln), "unknown")132 out["components"]["ssd_smart"] = smart133 if smart not in ("Verified", "Not Supported", "unknown"):134 out["alerts"].append({"severity": "error", "code": "ssd_smart", "message": f"SMART status: {smart}"})135 except Exception:136 out["components"]["ssd_smart"] = "unavailable"137 if any(a["severity"] == "error" for a in out["alerts"]):138 out["status"] = "degraded"139 elif out["alerts"]:140 out["status"] = "warning"141 return out142