mo
f36c8906bc
Adds native Command Center features (no new containers) integrated as sub-tabs in the existing Data Explorer and Data Quality views: - Continuous Data Quality (dq_monitor.py): live completeness/uniqueness/validity/ freshness scorecards via Trino with rolling trends → DataQuality "Live Monitoring". - Ownership & stewardship (catalog_governance.py): owner/steward/tier matrix, orphan detection, business glossary; local store best-effort synced to OpenMetadata (owner PATCH) → Data Explorer "Ownership". - Access & policy posture: per-dataset compliance combining PII masking, ownership, live DQ and observability alerts vs data contracts → Data Explorer "Access & Policies". - Lineage (lineage.py): staged source→CDC→Spark→S3→Iceberg→Trino→serving graph with live row counts and column-level PII/masking tracing → Data Explorer "Lineage". - Observability (observability.py): volume/freshness/schema-drift monitoring with alerts → Data Explorer "Observability". - Shared lake_meta.py dataset registry + bounded Trino client; fast native row-count and PK-indexed freshness so monitors stay cheap on 25-54M-row tables. - LLM context (lab_context.py) enriched with DQ scores, ownership and active alerts.
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"]})
|