health: catalog-based lake usage and 60 s cached evaluation (probes were timing out → mld heal restart loop)
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
5 changed files +46 −7
modified
services/api/src/findata_api/health.py
+35 −2
@@ -10,10 +10,37 @@ from findata_ingest.lake import Lake | ||
| 10 | 10 | |
| 11 | 11 | from findata_api import metrics as m |
| 12 | 12 | |
| 13 | +_cache: dict[str, Any] = {"at": 0.0, "value": None} | |
| 14 | +_lock = __import__("threading").Lock() | |
| 15 | + | |
| 16 | + | |
| 17 | +def 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 cheap | |
| 19 | + even while hundreds of thousands of files are being published.""" | |
| 20 | + import time | |
| 21 | + 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 v | |
| 30 | + | |
| 31 | + | |
| 32 | +def 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"])} | |
| 38 | + | |
| 13 | 39 | |
| 14 | 40 | def evaluate() -> dict[str, Any]: |
| 15 | 41 | s = get_settings() |
| 16 | 42 | lake = Lake() |
| 43 | + usage = {"curated": 0, "retired": 0, "raw": 0} | |
| 17 | 44 | out: dict[str, Any] = {"status": "ok", "checked_at": datetime.now(UTC).isoformat(), "alerts": [], "components": {}} |
| 18 | 45 | try: |
| 19 | 46 | with connection() as conn: |
@@ -77,15 +104,21 @@ def evaluate() -> dict[str, Any]: | ||
| 77 | 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() |
| 78 | 105 | for a in an: |
| 79 | 106 | m.ANOMALIES.labels(a["dataset_id"], a["severity"]).set(a["n"]) |
| 107 | + usage = lake_usage_bytes(conn) | |
| 80 | 108 | except Exception as e: |
| 81 | 109 | out["components"]["postgres"] = f"error: {e}" |
| 82 | 110 | out["alerts"].append({"severity": "error", "code": "postgres_down", "message": str(e)[:200]}) |
| 83 | 111 | free = lake.free_gb() |
| 84 | 112 | m.LAKE_FREE.set(free) |
| 85 | − curated = lake.used_bytes("curated") | |
| 113 | + curated = usage["curated"] | |
| 86 | 114 | m.LAKE_CURATED.set(curated) |
| 87 | 115 | 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} | |
| 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"}) | |
| 89 | 122 | if free < s.lake_hard_min_free_gb: |
| 90 | 123 | out["alerts"].append({"severity": "error", "code": "disk_critical", "message": f"only {free:.0f} GB free"}) |
| 91 | 124 | elif free < s.lake_hard_min_free_gb * 2: |
modified
services/api/src/findata_api/main.py
+1 −1
@@ -141,7 +141,7 @@ def create_app() -> FastAPI: | ||
| 141 | 141 | private = host in ("127.0.0.1", "::1") or host.startswith(("192.168.", "10.67.", "10.", "172.16.")) |
| 142 | 142 | if not private and not p.has("admin"): |
| 143 | 143 | return JSONResponse({"error": {"code": "forbidden", "message": "metrics require admin or a private network"}}, status_code=403) |
| 144 | − from findata_api.health import evaluate | |
| 144 | + from findata_api.health import cached_evaluate as evaluate | |
| 145 | 145 | evaluate() |
| 146 | 146 | body, ctype = M.render() |
| 147 | 147 | return Response(body, media_type=ctype) |
modified
services/api/src/findata_api/routers/admin.py
+5 −2
@@ -123,12 +123,15 @@ def storage(p: A.Principal = Depends(A.require("admin"))): | ||
| 123 | 123 | retired = conn.execute("SELECT count(*) n, coalesce(sum(bytes),0) bytes FROM lake_files WHERE version_removed IS NOT NULL").fetchone() |
| 124 | 124 | db = conn.execute("SELECT pg_database_size(current_database()) bytes").fetchone() |
| 125 | 125 | backup = conn.execute("SELECT value, updated_at FROM settings_kv WHERE key='last_backup'").fetchone() |
| 126 | + from findata_api.health import lake_usage_bytes | |
| 127 | + usage = lake_usage_bytes(conn) | |
| 126 | 128 | import os |
| 127 | 129 | st = os.statvfs(lake.root) |
| 128 | 130 | return {"lake_root": str(lake.root), "volume": {"total_gb": round(st.f_blocks * st.f_frsize / 1e9, 1), "free_gb": round(st.f_bavail * st.f_frsize / 1e9, 1)}, |
| 129 | 131 | "budget": {"soft_limit_gb": s.lake_soft_limit_gb, "hard_min_free_gb": s.lake_hard_min_free_gb}, |
| 130 | − "curated_bytes": lake.used_bytes("curated"), "raw_bytes": lake.used_bytes("raw"), "retired_bytes": lake.used_bytes("_retired"), | |
| 131 | − "exports_bytes": lake.used_bytes("exports"), "per_dataset": per_ds, "raw_archives": raw, "exports": exports, "retired_files": retired, | |
| 132 | + "curated_bytes": usage["curated"], "raw_bytes": usage["raw"], "retired_bytes": usage["retired"], | |
| 133 | + "exports_bytes": sum(int(e["bytes"] or 0) for e in exports), "per_dataset": per_ds, | |
| 134 | + "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, "raw_archives": raw, "exports": exports, "retired_files": retired, | |
| 132 | 135 | "postgres_bytes": db["bytes"], "last_backup": backup, |
| 133 | 136 | "capacity_model": {"bytes_per_1m_bar_curated": 13, "note": "see docs/DISCOVERY.md §5 — measured on the prior lake; recomputed from per_dataset above"}} |
| 134 | 137 | |
modified
services/api/src/findata_api/routers/health.py
+1 −1
@@ -4,7 +4,7 @@ from fastapi import APIRouter, Depends | ||
| 4 | 4 | from findata_core.db import connection |
| 5 | 5 | |
| 6 | 6 | from findata_api import auth as A |
| 7 | −from findata_api.health import evaluate | |
| 7 | +from findata_api.health import cached_evaluate as evaluate | |
| 8 | 8 | |
| 9 | 9 | router = APIRouter(tags=["health"]) |
| 10 | 10 | |
modified
services/ingest/src/findata_ingest/cli.py
+4 −1
@@ -201,7 +201,10 @@ def status() -> None: | ||
| 201 | 201 | |
| 202 | 202 | from findata_ingest.lake import Lake |
| 203 | 203 | lake = Lake() |
| 204 | − typer.echo(f"lake {lake.root} free {lake.free_gb():.1f} GB curated {lake.used_bytes('curated')/1e9:.2f} GB raw {lake.used_bytes('raw')/1e9:.2f} GB") | |
| 204 | + u = fetch_one("SELECT coalesce(sum(bytes) FILTER (WHERE version_removed IS NULL),0) AS live FROM lake_files") | |
| 205 | + r = fetch_one("SELECT coalesce(sum(bytes),0) AS raw FROM raw_archives WHERE path IS NOT NULL") | |
| 206 | + cold = f" cold {lake.cold} free {lake.free_gb(lake.cold):.0f} GB" if lake.cold else "" | |
| 207 | + typer.echo(f"lake {lake.root} free {lake.free_gb():.1f} GB curated {u['live']/1e9:.2f} GB raw {r['raw']/1e9:.2f} GB{cold}") | |
| 205 | 208 | for r in fetch_all("SELECT d.id, d.status, d.current_version, v.file_count, v.row_count, v.bytes FROM datasets d LEFT JOIN dataset_versions v ON v.dataset_id=d.id AND v.version=d.current_version ORDER BY d.id"): |
| 206 | 209 | typer.echo(f" {r['id']:<36} {r['status']:<22} v{r['current_version']:<4} files={r['file_count'] or 0:<6} rows={r['row_count'] or 0:<12} {(r['bytes'] or 0)/1e6:.1f} MB") |
| 207 | 210 | q = fetch_one("SELECT count(*) FILTER (WHERE status='queued') q, count(*) FILTER (WHERE status='running') r, count(*) FILTER (WHERE status='dead') d FROM ingest_jobs") |
| 208 | 211 | |