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