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 Lake10from findata_api import metrics as m111213def evaluate() -> dict[str, Any]:14 s = get_settings()15 lake = Lake()16 out: dict[str, Any] = {"status": "ok", "checked_at": datetime.now(UTC).isoformat(), "alerts": [], "components": {}}17 try:18 with connection() as conn:19 conn.execute("SELECT 1")20 out["components"]["postgres"] = "ok"21 jobs = {r["status"]: r["n"] for r in conn.execute("SELECT status, count(*) n FROM ingest_jobs GROUP BY status").fetchall()}22 for st in ("queued", "running", "succeeded", "failed", "dead"):23 m.JOBS.labels(st).set(jobs.get(st, 0))24 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, "25 "(SELECT value FROM checkpoints WHERE scope='dataset:'||d.id AND key='last_vendor_update:day') ck "26 "FROM datasets d WHERE d.status='active' AND d.kind IN ('market_bars','derived')").fetchall()27 out["datasets"] = []28 for d in ds:29 last = d["last_session"]30 fresh = (datetime.now(UTC).date() - last).days if last else None31 vendor = (d["ck"] or {}).get("vendor_last_update") if d["ck"] else None32 lag = None33 if vendor and last:34 lag = (datetime.fromisoformat(vendor).date() - last).days35 out["datasets"].append({"dataset_id": d["id"], "version": d["current_version"], "last_session": str(last) if last else None,36 "days_since_last_session": fresh, "vendor_last_update": vendor, "vendor_lag_days": lag})37 if fresh is not None:38 m.FRESHNESS.labels(d["id"]).set(fresh)39 if lag is not None:40 m.VENDOR_LAG.labels(d["id"]).set(lag)41 if lag > 0:42 out["alerts"].append({"severity": "warning", "code": "stale_dataset", "message": f"{d['id']} is {lag} day(s) behind the vendor"})43 if fresh is not None and fresh > 5:44 out["alerts"].append({"severity": "warning", "code": "no_recent_session", "message": f"{d['id']}: last session {last} ({fresh} days ago)"})45 dead = jobs.get("dead", 0)46 if dead:47 out["alerts"].append({"severity": "error", "code": "dead_jobs", "message": f"{dead} job(s) in the dead-letter queue"})48 lb = conn.execute("SELECT value, updated_at FROM settings_kv WHERE key='last_backup'").fetchone()49 if lb:50 age_h = (datetime.now(UTC) - lb["updated_at"]).total_seconds() / 360051 m.BACKUP_AGE.set(age_h)52 out["components"]["backup"] = {"last": lb["updated_at"].isoformat(), "age_hours": round(age_h, 1), "verified": lb["value"].get("verified")}53 if age_h > 48:54 out["alerts"].append({"severity": "warning", "code": "backup_stale", "message": f"last backup {age_h:.0f} h ago"})55 else:56 out["components"]["backup"] = {"last": None}57 out["alerts"].append({"severity": "info", "code": "no_backup_yet", "message": "no backup recorded yet"})58 lock = lake.root / ".writer.lock"59 worker_alive = False60 if lock.exists():61 try:62 import fcntl63 fh = open(lock)64 try:65 fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB)66 fcntl.flock(fh, fcntl.LOCK_UN)67 except BlockingIOError:68 worker_alive = True69 finally:70 fh.close()71 except OSError:72 pass73 out["components"]["ingest_worker"] = "running" if worker_alive else "not running"74 if not worker_alive:75 out["alerts"].append({"severity": "error", "code": "worker_down", "message": "ingest worker is not holding the lake lock"})76 an = conn.execute("SELECT dataset_id, severity, count(*) n FROM quality_anomalies WHERE created_at > now() - interval '7 days' GROUP BY 1,2").fetchall()77 for a in an:78 m.ANOMALIES.labels(a["dataset_id"], a["severity"]).set(a["n"])79 except Exception as e: # noqa: BLE00180 out["components"]["postgres"] = f"error: {e}"81 out["alerts"].append({"severity": "error", "code": "postgres_down", "message": str(e)[:200]})82 free = lake.free_gb()83 m.LAKE_FREE.set(free)84 curated = lake.used_bytes("curated")85 m.LAKE_CURATED.set(curated)86 out["components"]["lake"] = {"root": str(lake.root), "free_gb": round(free, 1), "curated_gb": round(curated / 1e9, 2),87 "raw_gb": round(lake.used_bytes("raw") / 1e9, 2), "soft_limit_gb": s.lake_soft_limit_gb, "hard_min_free_gb": s.lake_hard_min_free_gb}88 if free < s.lake_hard_min_free_gb:89 out["alerts"].append({"severity": "error", "code": "disk_critical", "message": f"only {free:.0f} GB free"})90 elif free < s.lake_hard_min_free_gb * 2:91 out["alerts"].append({"severity": "warning", "code": "disk_low", "message": f"{free:.0f} GB free"})92 if curated / 1e9 > s.lake_soft_limit_gb:93 out["alerts"].append({"severity": "warning", "code": "lake_over_budget", "message": f"curated {curated/1e9:.0f} GB > {s.lake_soft_limit_gb} GB"})94 try:95 import subprocess96 r = subprocess.run(["diskutil", "info", "disk0"], capture_output=True, text=True, timeout=10)97 smart = next((ln.split(":", 1)[1].strip() for ln in r.stdout.splitlines() if "SMART Status" in ln), "unknown")98 out["components"]["ssd_smart"] = smart99 if smart not in ("Verified", "Not Supported", "unknown"):100 out["alerts"].append({"severity": "error", "code": "ssd_smart", "message": f"SMART status: {smart}"})101 except Exception: # noqa: BLE001102 out["components"]["ssd_smart"] = "unavailable"103 if any(a["severity"] == "error" for a in out["alerts"]):104 out["status"] = "degraded"105 elif out["alerts"]:106 out["status"] = "warning"107 return out108