371 lines
14 KiB
Python
371 lines
14 KiB
Python
|
"""Data observability (Disease #6 — Traceability / observability).
|
|||
|
|
|
|||
|
|
A lightweight background monitor that, per business table, tracks over time:
|
|||
|
|
|
|||
|
|
volume – row count and its delta between cycles
|
|||
|
|
freshness – age of the newest record
|
|||
|
|
schema – the column set; a change raises a drift alert
|
|||
|
|
|
|||
|
|
It derives alerts (critical/warning/info) for volume drops, stalled ingestion,
|
|||
|
|
stale data and schema drift, and keeps a rolling time-series per dataset for the
|
|||
|
|
trend charts in the UI.
|
|||
|
|
|
|||
|
|
Endpoints:
|
|||
|
|
GET /api/observability/metrics -> per-dataset series + current state
|
|||
|
|
GET /api/observability/alerts -> active + recent alerts
|
|||
|
|
POST /api/observability/run -> run one probe cycle now
|
|||
|
|
POST /api/observability/config -> {enabled, interval_s, freshness_min}
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import threading
|
|||
|
|
import time
|
|||
|
|
import uuid
|
|||
|
|
from collections import deque
|
|||
|
|
from datetime import datetime, timezone
|
|||
|
|
from typing import Any
|
|||
|
|
|
|||
|
|
from fastapi import APIRouter, Body
|
|||
|
|
from fastapi.responses import JSONResponse
|
|||
|
|
|
|||
|
|
from lake_meta import DATASETS, trino_scalar, discover_columns
|
|||
|
|
|
|||
|
|
router = APIRouter(prefix="/api/observability", tags=["observability"])
|
|||
|
|
|
|||
|
|
_lock = threading.Lock()
|
|||
|
|
_state: dict[str, Any] = {
|
|||
|
|
"enabled": True,
|
|||
|
|
"interval_s": 90.0,
|
|||
|
|
"freshness_min": 30.0,
|
|||
|
|
"running": False,
|
|||
|
|
"cycles": 0,
|
|||
|
|
"last_cycle_ts": 0.0,
|
|||
|
|
"next_run_ts": 0.0,
|
|||
|
|
"started_at": None,
|
|||
|
|
"datasets": {
|
|||
|
|
d["key"]: {"label": d["label"], "engine": d["engine"], "color": d["color"],
|
|||
|
|
"table": d["fqtn"], "rows": None, "prev_rows": None, "delta": None,
|
|||
|
|
"freshness_age_min": None, "columns": None, "schema_hash": None,
|
|||
|
|
"series": deque(maxlen=80), "deltas": deque(maxlen=20),
|
|||
|
|
"stalled_cycles": 0, "ts": None, "error": None}
|
|||
|
|
for d in DATASETS
|
|||
|
|
},
|
|||
|
|
"alerts_active": {}, # dedup_key -> alert
|
|||
|
|
"alerts_history": deque(maxlen=120),
|
|||
|
|
"feed": deque(maxlen=40),
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _now() -> datetime:
|
|||
|
|
return datetime.now(timezone.utc)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _term(text: str, level: str = "info", phase: str = "observe") -> None:
|
|||
|
|
try:
|
|||
|
|
from agent_terminal import emit_threadsafe
|
|||
|
|
emit_threadsafe("infra-sentinel", text, level=level, phase=phase)
|
|||
|
|
except Exception:
|
|||
|
|
pass
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _raise_alert(dataset: str, atype: str, severity: str, message: str) -> None:
|
|||
|
|
dedup = f"{dataset}:{atype}"
|
|||
|
|
existing = _state["alerts_active"].get(dedup)
|
|||
|
|
if existing:
|
|||
|
|
existing["count"] += 1
|
|||
|
|
existing["last_ts"] = _now().isoformat()
|
|||
|
|
existing["message"] = message
|
|||
|
|
return
|
|||
|
|
alert = {"id": uuid.uuid4().hex[:8], "dataset": dataset, "type": atype,
|
|||
|
|
"severity": severity, "message": message, "count": 1,
|
|||
|
|
"ts": _now().isoformat(), "last_ts": _now().isoformat(), "resolved": False}
|
|||
|
|
_state["alerts_active"][dedup] = alert
|
|||
|
|
_state["alerts_history"].appendleft(dict(alert))
|
|||
|
|
lvl = "err" if severity == "critical" else ("warn" if severity == "warning" else "info")
|
|||
|
|
_term(f" ⚠ ALERT [{severity}] {dataset}: {message}", level=lvl, phase="alert")
|
|||
|
|
_state["feed"].appendleft({"ts": _now().isoformat(), "text": f"[{severity}] {dataset}: {message}", "level": lvl})
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _clear_alert(dataset: str, atype: str) -> None:
|
|||
|
|
dedup = f"{dataset}:{atype}"
|
|||
|
|
a = _state["alerts_active"].pop(dedup, None)
|
|||
|
|
if a:
|
|||
|
|
a["resolved"] = True
|
|||
|
|
a["resolved_ts"] = _now().isoformat()
|
|||
|
|
_state["alerts_history"].appendleft({**a, "message": f"resolved: {a['message']}"})
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _native_freshness(ds: dict[str, Any]) -> float | None:
|
|||
|
|
"""Epoch seconds of the newest record, read cheaply via the PK index
|
|||
|
|
(ORDER BY <pk> DESC LIMIT 1) on the source DB. Returns None if unavailable."""
|
|||
|
|
nc = ds.get("native_count")
|
|||
|
|
ts = ds.get("ts_col")
|
|||
|
|
key = ds.get("key_col")
|
|||
|
|
if not nc or not ts:
|
|||
|
|
return None
|
|||
|
|
eng = nc[0]
|
|||
|
|
try:
|
|||
|
|
import sql_console as s
|
|||
|
|
if eng == "postgres" and key:
|
|||
|
|
import psycopg2
|
|||
|
|
conn = psycopg2.connect(host=s.DB_HOST, port=s.PG_PORT, user=s.PG_USER,
|
|||
|
|
password=s.PG_PASS, dbname=s.PG_DB, connect_timeout=6)
|
|||
|
|
try:
|
|||
|
|
cur = conn.cursor()
|
|||
|
|
cur.execute("SET statement_timeout = 6000")
|
|||
|
|
cur.execute(f'SELECT extract(epoch FROM "{ts}") FROM public."{ds["table"]}" '
|
|||
|
|
f'WHERE "{ts}" IS NOT NULL ORDER BY "{key}" DESC LIMIT 1')
|
|||
|
|
row = cur.fetchone()
|
|||
|
|
return float(row[0]) if row and row[0] is not None else None
|
|||
|
|
finally:
|
|||
|
|
conn.close()
|
|||
|
|
if eng == "mysql" and key:
|
|||
|
|
import pymysql
|
|||
|
|
conn = pymysql.connect(host=s.DB_HOST, port=s.MYSQL_PORT, user=s.MYSQL_USER,
|
|||
|
|
password=s.MYSQL_PASS, database=s.MYSQL_DB, connect_timeout=6)
|
|||
|
|
try:
|
|||
|
|
cur = conn.cursor()
|
|||
|
|
cur.execute("SET SESSION MAX_EXECUTION_TIME = 6000")
|
|||
|
|
cur.execute(f"SELECT UNIX_TIMESTAMP(`{ts}`) FROM `{ds['table']}` "
|
|||
|
|
f"WHERE `{ts}` IS NOT NULL ORDER BY `{key}` DESC LIMIT 1")
|
|||
|
|
row = cur.fetchone()
|
|||
|
|
return float(row[0]) if row and row[0] is not None else None
|
|||
|
|
finally:
|
|||
|
|
conn.close()
|
|||
|
|
if eng == "mongodb":
|
|||
|
|
cli = s._mongo_client()
|
|||
|
|
try:
|
|||
|
|
doc = list(cli[ds["schema"]][ds["table"]].find({}, {ts: 1, "_id": 0}).sort("_id", -1).limit(1))
|
|||
|
|
if doc:
|
|||
|
|
import datetime as _dt
|
|||
|
|
v = doc[0].get(ts)
|
|||
|
|
if isinstance(v, _dt.datetime):
|
|||
|
|
return v.timestamp()
|
|||
|
|
finally:
|
|||
|
|
cli.close()
|
|||
|
|
except Exception:
|
|||
|
|
return None
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _probe(ds: dict[str, Any]) -> None:
|
|||
|
|
key = ds["key"]
|
|||
|
|
st = _state["datasets"][key]
|
|||
|
|
fq = ds["fqtn"]
|
|||
|
|
|
|||
|
|
# volume — prefer a fast native planner estimate (postgres/mysql/mongo); use
|
|||
|
|
# Trino metadata count for Iceberg; treat a slow/failed count as a soft miss.
|
|||
|
|
rows = None
|
|||
|
|
estimated = False
|
|||
|
|
nc = ds.get("native_count")
|
|||
|
|
if nc:
|
|||
|
|
try:
|
|||
|
|
import sql_console as _s
|
|||
|
|
rows = _s._table_row_count(nc[0], nc[1])
|
|||
|
|
estimated = True
|
|||
|
|
except Exception:
|
|||
|
|
rows = None
|
|||
|
|
if rows is None:
|
|||
|
|
try:
|
|||
|
|
rows = int(trino_scalar(f"SELECT count(*) FROM {fq}", timeout=30.0) or 0)
|
|||
|
|
except Exception as exc:
|
|||
|
|
st["error"] = str(exc)[:140]
|
|||
|
|
st["count_fails"] = st.get("count_fails", 0) + 1
|
|||
|
|
return
|
|||
|
|
st["error"] = None
|
|||
|
|
st["count_fails"] = 0
|
|||
|
|
st["estimated"] = estimated
|
|||
|
|
|
|||
|
|
prev = st["rows"]
|
|||
|
|
st["prev_rows"] = prev
|
|||
|
|
st["rows"] = rows
|
|||
|
|
delta = (rows - prev) if prev is not None else None
|
|||
|
|
st["delta"] = delta
|
|||
|
|
st["series"].append({"t": _now().strftime("%H:%M:%S"), "rows": rows, "delta": delta or 0})
|
|||
|
|
st["ts"] = _now().isoformat()
|
|||
|
|
|
|||
|
|
if delta is not None:
|
|||
|
|
st["deltas"].append(delta)
|
|||
|
|
# Guard against planner-estimate jitter: only alert on a material drop.
|
|||
|
|
drop_threshold = max(2000, int(0.02 * (prev or 0)))
|
|||
|
|
if delta < -drop_threshold:
|
|||
|
|
_raise_alert(key, "volume_drop", "warning",
|
|||
|
|
f"row count dropped by {abs(delta):,} ({prev:,} → {rows:,})")
|
|||
|
|
else:
|
|||
|
|
_clear_alert(key, "volume_drop")
|
|||
|
|
# stalled ingestion: no growth for several cycles on CDC sources. Skip for
|
|||
|
|
# estimate-based counts (planner stats update lazily → false stalls).
|
|||
|
|
if delta == 0 and not ds.get("curated") and not estimated:
|
|||
|
|
st["stalled_cycles"] += 1
|
|||
|
|
if st["stalled_cycles"] >= 4:
|
|||
|
|
_raise_alert(key, "stalled", "warning",
|
|||
|
|
f"no new rows for {st['stalled_cycles']} cycles")
|
|||
|
|
else:
|
|||
|
|
st["stalled_cycles"] = 0
|
|||
|
|
_clear_alert(key, "stalled")
|
|||
|
|
# spike (informational) — only for exact counts
|
|||
|
|
pos = [d for d in st["deltas"] if d > 0]
|
|||
|
|
if not estimated and pos and delta > (sum(pos) / len(pos)) * 6 and len(pos) >= 4:
|
|||
|
|
_raise_alert(key, "spike", "info", f"volume spike +{delta:,} rows this cycle")
|
|||
|
|
|
|||
|
|
# freshness — cheap PK-indexed newest-row lookup for the CDC sources, with a
|
|||
|
|
# short Trino max(ts) fallback for Iceberg tables.
|
|||
|
|
if ds.get("ts_col"):
|
|||
|
|
epoch = _native_freshness(ds)
|
|||
|
|
if epoch is None:
|
|||
|
|
try:
|
|||
|
|
v = trino_scalar(f'SELECT to_unixtime(max("{ds["ts_col"]}")) FROM {fq}', timeout=10.0)
|
|||
|
|
epoch = float(v) if v is not None else None
|
|||
|
|
except Exception:
|
|||
|
|
epoch = None
|
|||
|
|
if epoch:
|
|||
|
|
age = max(0.0, (time.time() - float(epoch)) / 60.0)
|
|||
|
|
st["freshness_age_min"] = round(age, 1)
|
|||
|
|
if age > float(_state["freshness_min"]) and not ds.get("curated"):
|
|||
|
|
_raise_alert(key, "stale", "warning",
|
|||
|
|
f"newest record is {age:.0f}m old (> {_state['freshness_min']:.0f}m)")
|
|||
|
|
else:
|
|||
|
|
_clear_alert(key, "stale")
|
|||
|
|
|
|||
|
|
# schema drift
|
|||
|
|
cols = discover_columns(ds)
|
|||
|
|
if cols:
|
|||
|
|
names = sorted(c["name"] for c in cols)
|
|||
|
|
h = hash(tuple(names))
|
|||
|
|
if st["schema_hash"] is not None and h != st["schema_hash"]:
|
|||
|
|
old = set(st["columns"] or [])
|
|||
|
|
new = set(names)
|
|||
|
|
added = sorted(new - old)
|
|||
|
|
removed = sorted(old - new)
|
|||
|
|
parts = []
|
|||
|
|
if added:
|
|||
|
|
parts.append(f"+{', '.join(added)}")
|
|||
|
|
if removed:
|
|||
|
|
parts.append(f"-{', '.join(removed)}")
|
|||
|
|
_raise_alert(key, "schema_drift", "critical", "schema changed: " + " ".join(parts))
|
|||
|
|
st["columns"] = names
|
|||
|
|
st["schema_hash"] = h
|
|||
|
|
|
|||
|
|
|
|||
|
|
def run_cycle() -> dict[str, Any]:
|
|||
|
|
if _state["running"]:
|
|||
|
|
return {"ok": True, "skipped": "already running"}
|
|||
|
|
_state["running"] = True
|
|||
|
|
try:
|
|||
|
|
_term(f"═══ observability sweep {_state['cycles'] + 1} — volume · freshness · schema ═══",
|
|||
|
|
level="info", phase="cycle")
|
|||
|
|
for ds in DATASETS:
|
|||
|
|
if not _state["enabled"]:
|
|||
|
|
break
|
|||
|
|
try:
|
|||
|
|
_probe(ds)
|
|||
|
|
except Exception as exc:
|
|||
|
|
_state["datasets"][ds["key"]]["error"] = str(exc)[:140]
|
|||
|
|
_state["cycles"] += 1
|
|||
|
|
_state["last_cycle_ts"] = time.time()
|
|||
|
|
_state["next_run_ts"] = time.time() + float(_state["interval_s"])
|
|||
|
|
active = len(_state["alerts_active"])
|
|||
|
|
_term(f"═══ sweep {_state['cycles']} done — {active} active alert(s) ═══",
|
|||
|
|
level=("warn" if active else "ok"), phase="cycle")
|
|||
|
|
finally:
|
|||
|
|
_state["running"] = False
|
|||
|
|
return {"ok": True, "active_alerts": len(_state["alerts_active"])}
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _loop() -> None:
|
|||
|
|
_state["started_at"] = _now().isoformat()
|
|||
|
|
time.sleep(28)
|
|||
|
|
while True:
|
|||
|
|
try:
|
|||
|
|
if _state["enabled"]:
|
|||
|
|
run_cycle()
|
|||
|
|
except Exception:
|
|||
|
|
pass
|
|||
|
|
time.sleep(max(30.0, float(_state["interval_s"])))
|
|||
|
|
|
|||
|
|
|
|||
|
|
threading.Thread(target=_loop, daemon=True, name="observability").start()
|
|||
|
|
|
|||
|
|
|
|||
|
|
def metrics_view() -> dict[str, Any]:
|
|||
|
|
with _lock:
|
|||
|
|
datasets = []
|
|||
|
|
for ds in DATASETS:
|
|||
|
|
st = _state["datasets"][ds["key"]]
|
|||
|
|
datasets.append({
|
|||
|
|
"key": ds["key"], "label": st["label"], "engine": st["engine"],
|
|||
|
|
"color": st["color"], "table": st["table"], "rows": st["rows"],
|
|||
|
|
"delta": st["delta"], "freshness_age_min": st["freshness_age_min"],
|
|||
|
|
"columns": len(st["columns"]) if st["columns"] else None,
|
|||
|
|
"stalled_cycles": st["stalled_cycles"], "error": st["error"],
|
|||
|
|
"ts": st["ts"], "series": list(st["series"]),
|
|||
|
|
})
|
|||
|
|
return {
|
|||
|
|
"ok": True, "enabled": _state["enabled"], "interval_s": _state["interval_s"],
|
|||
|
|
"freshness_min": _state["freshness_min"], "running": _state["running"],
|
|||
|
|
"cycles": _state["cycles"], "last_cycle_ts": _state["last_cycle_ts"],
|
|||
|
|
"next_run_ts": _state["next_run_ts"],
|
|||
|
|
"alert_counts": _alert_counts(),
|
|||
|
|
"datasets": datasets, "feed": list(_state["feed"])[:20],
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _alert_counts() -> dict[str, int]:
|
|||
|
|
counts = {"critical": 0, "warning": 0, "info": 0}
|
|||
|
|
for a in _state["alerts_active"].values():
|
|||
|
|
counts[a["severity"]] = counts.get(a["severity"], 0) + 1
|
|||
|
|
counts["total"] = len(_state["alerts_active"])
|
|||
|
|
return counts
|
|||
|
|
|
|||
|
|
|
|||
|
|
def summary_for_llm() -> dict[str, Any]:
|
|||
|
|
with _lock:
|
|||
|
|
return {
|
|||
|
|
"active_alerts": _alert_counts(),
|
|||
|
|
"alerts": [{"dataset": a["dataset"], "type": a["type"], "severity": a["severity"],
|
|||
|
|
"message": a["message"]} for a in _state["alerts_active"].values()],
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
|
|||
|
|
@router.get("/metrics")
|
|||
|
|
async def get_metrics() -> JSONResponse:
|
|||
|
|
return JSONResponse(metrics_view())
|
|||
|
|
|
|||
|
|
|
|||
|
|
@router.get("/alerts")
|
|||
|
|
async def get_alerts() -> JSONResponse:
|
|||
|
|
with _lock:
|
|||
|
|
return JSONResponse({
|
|||
|
|
"ok": True,
|
|||
|
|
"counts": _alert_counts(),
|
|||
|
|
"active": sorted(_state["alerts_active"].values(),
|
|||
|
|
key=lambda a: {"critical": 0, "warning": 1, "info": 2}.get(a["severity"], 3)),
|
|||
|
|
"history": list(_state["alerts_history"])[:60],
|
|||
|
|
})
|
|||
|
|
|
|||
|
|
|
|||
|
|
@router.post("/run")
|
|||
|
|
async def post_run() -> JSONResponse:
|
|||
|
|
from starlette.concurrency import run_in_threadpool
|
|||
|
|
res = await run_in_threadpool(run_cycle)
|
|||
|
|
return JSONResponse({**res, "metrics": metrics_view()})
|
|||
|
|
|
|||
|
|
|
|||
|
|
@router.post("/config")
|
|||
|
|
async def post_config(body: dict = Body(default={})) -> JSONResponse:
|
|||
|
|
if "enabled" in body:
|
|||
|
|
_state["enabled"] = bool(body["enabled"])
|
|||
|
|
if "interval_s" in body:
|
|||
|
|
try:
|
|||
|
|
_state["interval_s"] = max(30.0, min(900.0, float(body["interval_s"])))
|
|||
|
|
except Exception:
|
|||
|
|
pass
|
|||
|
|
if "freshness_min" in body:
|
|||
|
|
try:
|
|||
|
|
_state["freshness_min"] = max(1.0, min(1440.0, float(body["freshness_min"])))
|
|||
|
|
except Exception:
|
|||
|
|
pass
|
|||
|
|
return JSONResponse({"ok": True, "enabled": _state["enabled"],
|
|||
|
|
"interval_s": _state["interval_s"], "freshness_min": _state["freshness_min"]})
|