From f36c8906bcb3df5d0f50a571da64015beaf1db87 Mon Sep 17 00:00:00 2001 From: mo Date: Mon, 29 Jun 2026 17:43:37 +0000 Subject: [PATCH] =?UTF-8?q?feat(governance):=20close=20the=206=20data-dise?= =?UTF-8?q?ase=20gaps=20=E2=80=94=20DQ=20monitoring,=20ownership,=20access?= =?UTF-8?q?=20posture,=20lineage=20&=20observability?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- api/Dockerfile | 2 +- api/catalog_governance.py | 314 +++++++++++++++ api/dq_monitor.py | 345 ++++++++++++++++ api/lab_context.py | 36 ++ api/lake_meta.py | 147 +++++++ api/lineage.py | 248 ++++++++++++ api/main.py | 8 + api/observability.py | 370 ++++++++++++++++++ .../components/features/DataExplorerView.tsx | 18 +- .../components/features/DataQualityView.tsx | 9 +- .../components/features/DqMonitoringPanel.tsx | 173 ++++++++ .../features/GovernanceAccessView.tsx | 117 ++++++ .../features/GovernanceOwnershipView.tsx | 246 ++++++++++++ ui/src/components/features/LineageView.tsx | 309 +++++++++++++++ .../components/features/ObservabilityView.tsx | 141 +++++++ 15 files changed, 2479 insertions(+), 4 deletions(-) create mode 100644 api/catalog_governance.py create mode 100644 api/dq_monitor.py create mode 100644 api/lake_meta.py create mode 100644 api/lineage.py create mode 100644 api/observability.py create mode 100644 ui/src/components/features/DqMonitoringPanel.tsx create mode 100644 ui/src/components/features/GovernanceAccessView.tsx create mode 100644 ui/src/components/features/GovernanceOwnershipView.tsx create mode 100644 ui/src/components/features/LineageView.tsx create mode 100644 ui/src/components/features/ObservabilityView.tsx diff --git a/api/Dockerfile b/api/Dockerfile index ac360e8..9a13ba6 100644 --- a/api/Dockerfile +++ b/api/Dockerfile @@ -4,7 +4,7 @@ WORKDIR /app RUN apt-get update && apt-get install -y --no-install-recommends curl && rm -rf /var/lib/apt/lists/* COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt -COPY main.py lab_context.py agent_terminal.py workload.py node_registry.py node_ops.py topology_views.py supervisor.py approval_service.py db.py dockhand_envs.py presentation.py database_inventory.py presentation_upload.py presentation_static.py storage_s3.py elasticsearch_api.py sql_console.py hdfs_api.py ssh_terminal.py pipeline_ops.py hadoop_analytics.py agent_ops.py cdc_consumer.py movements.py dataflow.py streaming_ops.py spark_workbench.py hadoop_sql.py hdfs_kafka.py webhdfs_util.py pii_catalog.py platform_context.py trino_federated.py etl_offload.py agent_activity.py hive_bench_seed.json . +COPY main.py lab_context.py agent_terminal.py workload.py node_registry.py node_ops.py topology_views.py supervisor.py approval_service.py db.py dockhand_envs.py presentation.py database_inventory.py presentation_upload.py presentation_static.py storage_s3.py elasticsearch_api.py sql_console.py hdfs_api.py ssh_terminal.py pipeline_ops.py hadoop_analytics.py agent_ops.py cdc_consumer.py movements.py dataflow.py streaming_ops.py spark_workbench.py hadoop_sql.py hdfs_kafka.py webhdfs_util.py pii_catalog.py platform_context.py trino_federated.py etl_offload.py agent_activity.py lake_meta.py lineage.py dq_monitor.py observability.py catalog_governance.py hive_bench_seed.json . RUN mkdir -p /data ENV DATABASE_URL=sqlite:////data/atc-agents.db EXPOSE 3201 diff --git a/api/catalog_governance.py b/api/catalog_governance.py new file mode 100644 index 0000000..d9a3954 --- /dev/null +++ b/api/catalog_governance.py @@ -0,0 +1,314 @@ +"""Data governance: ownership, stewardship, glossary & posture +(Diseases #3 Ownership and #5 Governance). + +Ownership/steward assignments are kept in a local store (so the feature always +works for the demo regardless of OpenMetadata ingestion state) and are +best-effort mirrored to OpenMetadata. Users/teams for the assignment dropdowns +are read from OpenMetadata when reachable. The governance *posture* view +combines, per dataset: ownership, PII/masking (pii_catalog), live data-quality +score (dq_monitor), observability alerts and a simple data-contract check. + +Endpoints: + GET /api/governance/datasets -> ownership matrix (+ orphan flag) + GET /api/governance/users -> assignable users/teams + POST /api/governance/assign -> set owner/steward/team/tier (native + OM) + GET /api/governance/glossary -> business glossary terms + GET /api/governance/posture -> combined governance posture per dataset +""" + +from __future__ import annotations + +import json +import os +import time +from pathlib import Path +from typing import Any + +import httpx +from fastapi import APIRouter +from fastapi.responses import JSONResponse +from pydantic import BaseModel + +from lake_meta import DATASETS, DATASET_BY_KEY + +router = APIRouter(prefix="/api/governance", tags=["governance"]) + +OPENMETADATA_URL = os.getenv("OPENMETADATA_URL", "").rstrip("/") +OPENMETADATA_TOKEN = os.getenv("OPENMETADATA_TOKEN", "") +OWNERS_PATH = Path(os.getenv("GOVERNANCE_OWNERS_PATH", "/data/governance_owners.json")) +CONTRACTS_PATH = Path(os.getenv("GOVERNANCE_CONTRACTS_PATH", "/data/governance_contracts.json")) + +# Default data contracts (quality SLAs) per dataset — used when none stored. +DEFAULT_CONTRACTS = { + "_default": {"min_score": 80, "min_completeness": 95, "min_freshness_min": 60, + "max_critical_alerts": 0}, +} + +# Seed business glossary so the term list is never empty even before OM ingestion. +SEED_GLOSSARY = [ + {"name": "Customer", "description": "A person or organization that places sales orders.", + "related": ["customer_id", "customer_name", "customer_email"], "domain": "Sales"}, + {"name": "Order", "description": "A sales transaction with an amount, channel and status.", + "related": ["order_id", "amount", "order_status"], "domain": "Sales"}, + {"name": "Revenue", "description": "Sum of order amounts over a period.", + "related": ["amount", "currency"], "domain": "Sales"}, + {"name": "Employee", "description": "A member of the workforce tracked via HR events.", + "related": ["employee_id", "department", "role_name"], "domain": "People"}, + {"name": "PII", "description": "Personally Identifiable Information — masked per policy.", + "related": ["customer_email", "national_id", "billing_iban"], "domain": "Governance"}, + {"name": "Telemetry", "description": "Device metric readings over time.", + "related": ["device_id", "metric_type", "metric_value"], "domain": "IoT"}, +] + + +def _headers() -> dict[str, str]: + h = {"Accept": "application/json"} + if OPENMETADATA_TOKEN: + h["Authorization"] = f"Bearer {OPENMETADATA_TOKEN}" + return h + + +def _load(path: Path) -> dict[str, Any]: + try: + return json.loads(path.read_text()) + except Exception: + return {} + + +def _save(path: Path, data: dict[str, Any]) -> None: + try: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(data, indent=2)) + except Exception: + pass + + +# ── OpenMetadata best-effort ──────────────────────────────────────────────── +def _om_users() -> list[dict[str, Any]]: + if not OPENMETADATA_URL: + return [] + out: list[dict[str, Any]] = [] + try: + with httpx.Client(timeout=8.0, verify=False) as c: + r = c.get(f"{OPENMETADATA_URL}/api/v1/users?limit=50&isBot=false", headers=_headers()) + if r.status_code == 200: + for u in r.json().get("data", []): + out.append({"id": u.get("id"), "name": u.get("name"), + "display": u.get("displayName") or u.get("name"), "type": "user"}) + rt = c.get(f"{OPENMETADATA_URL}/api/v1/teams?limit=50", headers=_headers()) + if rt.status_code == 200: + for t in rt.json().get("data", []): + out.append({"id": t.get("id"), "name": t.get("name"), + "display": t.get("displayName") or t.get("name"), "type": "team"}) + except Exception: + pass + return out + + +def _om_glossary() -> list[dict[str, Any]]: + if not OPENMETADATA_URL: + return [] + try: + with httpx.Client(timeout=8.0, verify=False) as c: + r = c.get(f"{OPENMETADATA_URL}/api/v1/glossaryTerms?limit=100", headers=_headers()) + if r.status_code == 200: + return [{"name": t.get("name"), "description": t.get("description", ""), + "domain": "OpenMetadata", "related": []} + for t in r.json().get("data", [])] + except Exception: + pass + return [] + + +def _om_patch_owner(om_fqn: str, user: dict[str, Any]) -> bool: + """Best-effort: set table owner in OpenMetadata via JSON-Patch.""" + if not OPENMETADATA_URL or not om_fqn or not user.get("id"): + return False + try: + with httpx.Client(timeout=8.0, verify=False) as c: + g = c.get(f"{OPENMETADATA_URL}/api/v1/tables/name/{om_fqn}", headers=_headers()) + if g.status_code != 200: + return False + patch = [{"op": "add", "path": "/owners/0", + "value": {"id": user["id"], "type": user.get("type", "user")}}] + h = {**_headers(), "Content-Type": "application/json-patch+json"} + p = c.patch(f"{OPENMETADATA_URL}/api/v1/tables/name/{om_fqn}", + headers=h, content=json.dumps(patch)) + return p.status_code in (200, 201) + except Exception: + return False + + +# ── helpers pulling from sibling modules (all best-effort) ────────────────── +def _pii_summary(pii_key: str) -> dict[str, Any]: + try: + from pii_catalog import get_pii + for d in get_pii().get("datasets", []): + if d.get("key") == pii_key: + return {"pii_count": d.get("pii_count", 0), + "all_masked": d.get("all_masked", False), + "masked": sum(1 for c in d.get("pii_columns", []) if c.get("masked")), + "unmasked": sum(1 for c in d.get("pii_columns", []) if not c.get("masked"))} + except Exception: + pass + return {"pii_count": 0, "all_masked": False, "masked": 0, "unmasked": 0} + + +def _dq_card(key: str) -> dict[str, Any]: + try: + from dq_monitor import scorecards_view + for c in scorecards_view().get("cards", []): + if c.get("key") == key: + return {"score": c.get("score"), "issues": c.get("issues", []), + "freshness_age_min": c.get("freshness_age_min")} + except Exception: + pass + return {"score": None, "issues": []} + + +def _obs_alerts(key: str) -> list[dict[str, Any]]: + try: + from observability import _state, _lock + with _lock: + return [{"type": a["type"], "severity": a["severity"], "message": a["message"]} + for a in _state["alerts_active"].values() if a["dataset"] == key] + except Exception: + return [] + + +# ── views ─────────────────────────────────────────────────────────────────── +def datasets_view() -> dict[str, Any]: + owners = _load(OWNERS_PATH) + rows = [] + orphans = 0 + stewarded = 0 + for ds in DATASETS: + rec = owners.get(ds["key"], {}) + owner = rec.get("owner") + steward = rec.get("steward") + if not owner: + orphans += 1 + if steward: + stewarded += 1 + rows.append({ + "key": ds["key"], "label": ds["label"], "engine": ds["engine"], "color": ds["color"], + "table": ds["fqtn"], "domain": rec.get("domain") or ds.get("domain"), + "owner": owner, "steward": steward, "team": rec.get("team"), + "tier": rec.get("tier"), "classification": rec.get("classification"), + "updated_at": rec.get("updated_at"), + "orphan": not owner, + "pii": _pii_summary(ds.get("pii_key", ds["key"])), + }) + return {"ok": True, "datasets": rows, + "summary": {"total": len(rows), "orphans": orphans, "stewarded": stewarded, + "owned": len(rows) - orphans}, + "om_connected": bool(OPENMETADATA_URL)} + + +class AssignRequest(BaseModel): + key: str + owner: str | None = None + steward: str | None = None + team: str | None = None + tier: str | None = None + classification: str | None = None + + +def posture_view() -> dict[str, Any]: + owners = _load(OWNERS_PATH) + contracts = _load(CONTRACTS_PATH) + default_c = DEFAULT_CONTRACTS["_default"] + out = [] + compliant = 0 + for ds in DATASETS: + key = ds["key"] + rec = owners.get(key, {}) + contract = {**default_c, **(contracts.get(key, {}))} + pii = _pii_summary(ds.get("pii_key", key)) + dq = _dq_card(key) + alerts = _obs_alerts(key) + crit = sum(1 for a in alerts if a["severity"] == "critical") + checks = [] + score = dq.get("score") + checks.append({"name": "DQ score", "ok": score is not None and score >= contract["min_score"], + "value": score, "target": contract["min_score"]}) + checks.append({"name": "Owner assigned", "ok": bool(rec.get("owner")), + "value": rec.get("owner") or "—", "target": "assigned"}) + checks.append({"name": "PII masked", "ok": pii["unmasked"] == 0, + "value": f'{pii["masked"]}/{pii["pii_count"]}', "target": "all"}) + checks.append({"name": "Critical alerts", "ok": crit <= contract["max_critical_alerts"], + "value": crit, "target": contract["max_critical_alerts"]}) + ok = all(c["ok"] for c in checks) + if ok: + compliant += 1 + out.append({ + "key": key, "label": ds["label"], "engine": ds["engine"], "color": ds["color"], + "table": ds["fqtn"], "owner": rec.get("owner"), "steward": rec.get("steward"), + "tier": rec.get("tier"), "pii": pii, "dq_score": score, "issues": dq.get("issues", []), + "alerts": alerts, "contract": contract, "checks": checks, "compliant": ok, + }) + return {"ok": True, "datasets": out, + "summary": {"total": len(out), "compliant": compliant, + "non_compliant": len(out) - compliant}} + + +def summary_for_llm() -> dict[str, Any]: + owners = _load(OWNERS_PATH) + return { + "owners": {k: {"owner": v.get("owner"), "steward": v.get("steward"), "tier": v.get("tier")} + for k, v in owners.items()}, + "orphan_datasets": [d["key"] for d in DATASETS if not owners.get(d["key"], {}).get("owner")], + } + + +@router.get("/datasets") +async def get_datasets() -> JSONResponse: + return JSONResponse(datasets_view()) + + +@router.get("/users") +async def get_users() -> JSONResponse: + users = _om_users() + if not users: + users = [{"id": None, "name": n, "display": n, "type": "user"} + for n in ("admin", "bart", "mo")] + \ + [{"id": None, "name": "Organization", "display": "Organization", "type": "team"}] + return JSONResponse({"ok": True, "users": users, "om_connected": bool(OPENMETADATA_URL)}) + + +@router.post("/assign") +async def assign(body: AssignRequest) -> JSONResponse: + if body.key not in DATASET_BY_KEY: + return JSONResponse({"ok": False, "error": f"unknown dataset {body.key}"}, status_code=400) + owners = _load(OWNERS_PATH) + rec = dict(owners.get(body.key, {})) + for field in ("owner", "steward", "team", "tier", "classification"): + val = getattr(body, field) + if val is not None: + rec[field] = val or None + rec["updated_at"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) + owners[body.key] = rec + _save(OWNERS_PATH, owners) + + om_synced = False + if body.owner: + ds = DATASET_BY_KEY[body.key] + user = next((u for u in _om_users() if u.get("display") == body.owner or u.get("name") == body.owner), None) + if user: + om_synced = _om_patch_owner(ds.get("om_fqn", ""), user) + return JSONResponse({"ok": True, "key": body.key, "record": rec, "om_synced": om_synced}) + + +@router.get("/glossary") +async def get_glossary() -> JSONResponse: + terms = _om_glossary() + source = "openmetadata" + if not terms: + terms = SEED_GLOSSARY + source = "seed" + return JSONResponse({"ok": True, "source": source, "count": len(terms), "terms": terms}) + + +@router.get("/posture") +async def get_posture() -> JSONResponse: + return JSONResponse(posture_view()) diff --git a/api/dq_monitor.py b/api/dq_monitor.py new file mode 100644 index 0000000..70c03cd --- /dev/null +++ b/api/dq_monitor.py @@ -0,0 +1,345 @@ +"""Continuous data-quality monitoring (Disease #2 — Dirty Data). + +Where dq-api runs Great-Expectations/Soda on *uploaded* files, this agent runs +quality checks continuously against the *live* business tables through Trino and +produces a per-dataset scorecard (0-100) across five dimensions: + + completeness – non-null ratio across columns + uniqueness – distinct/key ratio (duplicate detection) + validity – domain rules (non-negative amounts, present timestamps) + freshness – age of the newest record vs a threshold + volume – row count + delta vs the previous cycle + +A rolling score history powers trend sparklines. The loop streams the exact +SQL it runs into the Data Custodian terminal so operators can see the checks. + +Endpoints: + GET /api/dq/scorecards -> all datasets, dimensions, score, trend + GET /api/dq/scorecard/{key} -> one dataset detail + POST /api/dq/run -> run one cycle now + POST /api/dq/config -> {enabled, interval_s, sample} +""" + +from __future__ import annotations + +import threading +import time +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, DATASET_BY_KEY, trino, trino_scalar, discover_columns + +router = APIRouter(prefix="/api/dq", tags=["data-quality"]) + +FRESHNESS_THRESHOLD_MIN = 60.0 # newest row older than this => freshness degraded +MAX_COLS = 24 + +_lock = threading.Lock() +_state: dict[str, Any] = { + "enabled": True, + "interval_s": 180.0, + "sample": 20000, + "running": False, + "cycles": 0, + "last_cycle_ts": 0.0, + "next_run_ts": 0.0, + "started_at": None, + "cards": {}, # key -> latest scorecard + "history": {d["key"]: deque(maxlen=60) for d in DATASETS}, + "cols_cache": {}, # key -> [{name,type}] + "feed": deque(maxlen=40), +} + + +def _now() -> datetime: + return datetime.now(timezone.utc) + + +def _term(text: str, level: str = "info", phase: str = "dq") -> None: + try: + from agent_terminal import emit_threadsafe + emit_threadsafe("data-custodian", text, level=level, phase=phase) + except Exception: + pass + + +def _feed(text: str, level: str = "info") -> None: + _state["feed"].appendleft({"ts": _now().isoformat(), "text": text, "level": level}) + + +def _score_color(score: float) -> str: + if score >= 90: + return "#34d399" + if score >= 75: + return "#fbbf24" + if score >= 50: + return "#fb923c" + return "#f87171" + + +def _obs_volume_freshness(key: str) -> tuple[int | None, float | None]: + """Read row count + freshness from the observability monitor (in-memory).""" + try: + from observability import _state, _lock + with _lock: + st = _state["datasets"].get(key) or {} + return st.get("rows"), st.get("freshness_age_min") + except Exception: + return None, None + + +def _columns(ds: dict[str, Any]) -> list[dict[str, str]]: + key = ds["key"] + cols = _state["cols_cache"].get(key) + if not cols: + cols = discover_columns(ds) + if cols: + _state["cols_cache"][key] = cols + return cols or [] + + +def _assess(ds: dict[str, Any], sample: int) -> dict[str, Any]: + key = ds["key"] + fq = ds["fqtn"] + cols = _columns(ds) + col_names = [c["name"] for c in cols][:MAX_COLS] + dims: dict[str, Any] = {} + issues: list[str] = [] + col_profiles: list[dict[str, Any]] = [] + + # ── one bounded query: completeness + uniqueness + validity over a sample ── + has_unique_key = ds.get("unique_key", True) + key_col = ds.get("key_col") if ds.get("key_col") in col_names else (col_names[0] if col_names else None) + if not has_unique_key: + key_col = None # no single-column unique key (e.g. composite PK) → skip dedup + ts_col = ds.get("ts_col") if ds.get("ts_col") in col_names else None + amt_col = ds.get("amount_col") if ds.get("amount_col") in col_names else None + + selects = ["count(*) AS n"] + for i, c in enumerate(col_names): + selects.append(f'count("{c}") AS c{i}') + if key_col: + selects.append(f'count(DISTINCT "{key_col}") AS dk') + if amt_col: + selects.append(f'count_if("{amt_col}" >= 0) AS amt_ok') + if ts_col: + selects.append(f'count_if("{ts_col}" IS NOT NULL) AS ts_ok') + + sql = f"SELECT {', '.join(selects)} FROM (SELECT * FROM {fq} LIMIT {sample}) s" + _term(f"$ trino: profile {key} ({len(col_names)} cols · sample {sample:,})", level="cmd", phase="dq") + cols_out, rows = trino(sql, timeout=40.0) + rec = dict(zip(cols_out, rows[0])) if rows else {} + n = int(rec.get("n") or 0) + + if n > 0: + # completeness + non_null_ratios = [] + for i, c in enumerate(col_names): + cnt = int(rec.get(f"c{i}") or 0) + ratio = cnt / n + non_null_ratios.append(ratio) + col_profiles.append({"name": c, "completeness": round(ratio * 100, 1), + "nulls": n - cnt}) + completeness = round(100.0 * sum(non_null_ratios) / max(1, len(non_null_ratios)), 1) + dims["completeness"] = completeness + worst = sorted(col_profiles, key=lambda x: x["completeness"])[:3] + for w in worst: + if w["completeness"] < 95: + issues.append(f'{w["name"]} {100 - w["completeness"]:.0f}% null') + + # uniqueness / dedup + if key_col and rec.get("dk") is not None: + dk = int(rec["dk"]) + uniq = round(100.0 * dk / n, 1) + dims["uniqueness"] = uniq + dups = n - dk + if dups > 0: + issues.append(f"{dups:,} duplicate {key_col} in sample") + + # validity + valid_parts = [] + if amt_col and rec.get("amt_ok") is not None: + valid_parts.append(int(rec["amt_ok"]) / n) + if ts_col and rec.get("ts_ok") is not None: + valid_parts.append(int(rec["ts_ok"]) / n) + if valid_parts: + validity = round(100.0 * sum(valid_parts) / len(valid_parts), 1) + dims["validity"] = validity + if validity < 99 and amt_col: + issues.append(f"negative/invalid {amt_col}") + + # ── volume + freshness: reuse the observability monitor's in-memory probes + # (it already polls count(*) and max(ts)). We deliberately do NOT issue our + # own count/max here — those are expensive full scans on 25-54M-row tables — + # so a DQ cycle stays fast and the score is driven by the cheap sample. ── + volume, fresh_age_min = _obs_volume_freshness(key) + if volume is None: + volume = n # sample size until observability reports the real count + if fresh_age_min is not None: + fresh_score = 100.0 if fresh_age_min <= FRESHNESS_THRESHOLD_MIN else max( + 0.0, 100.0 - (fresh_age_min - FRESHNESS_THRESHOLD_MIN) / 5.0) + dims["freshness"] = round(fresh_score, 1) + if fresh_age_min > FRESHNESS_THRESHOLD_MIN and not ds.get("curated"): + issues.append(f"stale {fresh_age_min:.0f}m") + + score = round(sum(dims.values()) / len(dims), 1) if dims else 0.0 + prev = _state["cards"].get(key, {}) + prev_vol = prev.get("volume") + delta = (volume - prev_vol) if (volume is not None and prev_vol is not None) else None + + card = { + "key": key, "label": ds["label"], "engine": ds["engine"], "color": ds["color"], + "table": fq, "domain": ds.get("domain"), + "score": score, "score_color": _score_color(score), + "dimensions": dims, "volume": volume, "volume_delta": delta, + "freshness_age_min": round(fresh_age_min, 1) if fresh_age_min is not None else None, + "issues": issues[:5], "columns": len(col_names), + "worst_columns": sorted(col_profiles, key=lambda x: x["completeness"])[:5], + "ts": _now().isoformat(), + } + lvl = "ok" if score >= 90 else ("warn" if score >= 60 else "err") + _term(f" ← {key}: score {score} · " + " · ".join(f"{k} {v}" for k, v in dims.items()) + + (f" · {len(issues)} issue(s)" if issues else ""), level=lvl, phase="dq") + return card + + +def run_cycle() -> dict[str, Any]: + if _state["running"]: + return {"ok": True, "skipped": "already running"} + _state["running"] = True + sample = int(_state["sample"]) + n_ok = 0 + try: + _term(f"═══ DQ monitor cycle {_state['cycles'] + 1} — live tables via Trino ═══", + level="info", phase="cycle") + for ds in DATASETS: + if not _state["enabled"]: + break + try: + card = _assess(ds, sample) + except Exception as exc: + card = {"key": ds["key"], "label": ds["label"], "engine": ds["engine"], + "color": ds["color"], "table": ds["fqtn"], "score": None, + "score_color": "#64748b", "dimensions": {}, "error": str(exc)[:160], + "issues": [f"check failed: {str(exc)[:80]}"], "ts": _now().isoformat()} + _term(f" ✗ {ds['key']}: {str(exc)[:120]}", level="err", phase="dq") + with _lock: + _state["cards"][ds["key"]] = card + if card.get("score") is not None: + _state["history"][ds["key"]].append({"t": _now().strftime("%H:%M"), "score": card["score"]}) + n_ok += 1 + _state["cycles"] += 1 + _state["last_cycle_ts"] = time.time() + _state["next_run_ts"] = time.time() + float(_state["interval_s"]) + avg = _overall_score() + _feed(f"cycle {_state['cycles']} — {n_ok}/{len(DATASETS)} datasets scored · platform DQ {avg}") + _term(f"═══ cycle {_state['cycles']} done — platform DQ score {avg} ═══", level="ok", phase="cycle") + finally: + _state["running"] = False + return {"ok": True, "scored": n_ok, "platform_score": _overall_score()} + + +def _overall_score() -> float | None: + vals = [c["score"] for c in _state["cards"].values() if c.get("score") is not None] + return round(sum(vals) / len(vals), 1) if vals else None + + +def _loop() -> None: + _state["started_at"] = _now().isoformat() + time.sleep(20) + 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="dq-monitor").start() + + +def scorecards_view() -> dict[str, Any]: + with _lock: + cards = [] + for ds in DATASETS: + c = _state["cards"].get(ds["key"]) + if c: + cards.append({**c, "trend": list(_state["history"][ds["key"]])}) + else: + cards.append({"key": ds["key"], "label": ds["label"], "engine": ds["engine"], + "color": ds["color"], "table": ds["fqtn"], "score": None, + "score_color": "#64748b", "dimensions": {}, "issues": [], + "trend": [], "pending": True}) + # platform dimension averages + dim_avg: dict[str, list[float]] = {} + for c in cards: + for k, v in (c.get("dimensions") or {}).items(): + dim_avg.setdefault(k, []).append(v) + return { + "ok": True, + "enabled": _state["enabled"], + "interval_s": _state["interval_s"], + "sample": _state["sample"], + "running": _state["running"], + "cycles": _state["cycles"], + "last_cycle_ts": _state["last_cycle_ts"], + "next_run_ts": _state["next_run_ts"], + "platform_score": _overall_score(), + "dimension_averages": {k: round(sum(v) / len(v), 1) for k, v in dim_avg.items()}, + "cards": cards, + "feed": list(_state["feed"])[:20], + } + + +def summary_for_llm() -> dict[str, Any]: + with _lock: + return { + "platform_dq_score": _overall_score(), + "datasets": [{"key": c["key"], "score": c.get("score"), + "issues": c.get("issues", []), "freshness_age_min": c.get("freshness_age_min")} + for c in _state["cards"].values()], + } + + +@router.get("/scorecards") +async def get_scorecards() -> JSONResponse: + return JSONResponse(scorecards_view()) + + +@router.get("/scorecard/{key}") +async def get_scorecard(key: str) -> JSONResponse: + with _lock: + c = _state["cards"].get(key) + if not c: + return JSONResponse({"ok": False, "error": "unknown or not yet scored"}, status_code=404) + return JSONResponse({"ok": True, "card": {**c, "trend": list(_state["history"].get(key, []))}}) + + +@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, "scorecards": scorecards_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(1800.0, float(body["interval_s"]))) + except Exception: + pass + if "sample" in body: + try: + _state["sample"] = max(1000, min(200000, int(body["sample"]))) + except Exception: + pass + return JSONResponse({"ok": True, "enabled": _state["enabled"], + "interval_s": _state["interval_s"], "sample": _state["sample"]}) diff --git a/api/lab_context.py b/api/lab_context.py index a8a3506..ead9f5e 100644 --- a/api/lab_context.py +++ b/api/lab_context.py @@ -671,6 +671,21 @@ def collect_governance(log: TerminalLogFn | None = None) -> dict[str, Any]: "mysql_hr.hr.employee_events -> iceberg.curated_masked.employee_events_masked (PII masked)", "hdfs:/data/historical/sales_orders -> iceberg.hadoop.historical_sales_hdfs", ] + try: + from dq_monitor import summary_for_llm as _dq + out["data_quality"] = _dq() + except Exception as exc: + out["data_quality"] = {"error": str(exc)} + try: + from observability import summary_for_llm as _obs + out["observability"] = _obs() + except Exception as exc: + out["observability"] = {"error": str(exc)} + try: + from catalog_governance import summary_for_llm as _own + out["ownership"] = _own() + except Exception as exc: + out["ownership"] = {"error": str(exc)} return out @@ -700,6 +715,27 @@ def _section_governance(g: dict[str, Any]) -> list[str]: lines.append(" Lineage:") for ln in g.get("lineage") or []: lines.append(f" - {ln}") + dq = g.get("data_quality") or {} + if isinstance(dq, dict) and "error" not in dq: + lines.append(f" Data quality (continuous, live tables): platform score {dq.get('platform_dq_score')}") + for d in dq.get("datasets") or []: + iss = f" issues: {', '.join(d['issues'][:3])}" if d.get("issues") else "" + lines.append(f" - {d['key']}: score {d.get('score')}{iss}") + own = g.get("ownership") or {} + if isinstance(own, dict) and "error" not in own: + orph = own.get("orphan_datasets") or [] + lines.append(f" Ownership: {len(own.get('owners', {}))} assigned" + + (f", orphans (no owner): {', '.join(orph)}" if orph else ", no orphans")) + for k, v in (own.get("owners") or {}).items(): + if v.get("owner"): + lines.append(f" - {k}: owner={v.get('owner')} steward={v.get('steward') or '—'} tier={v.get('tier') or '—'}") + obs = g.get("observability") or {} + if isinstance(obs, dict) and "error" not in obs: + ac = obs.get("active_alerts") or {} + lines.append(f" Observability alerts: {ac.get('total', 0)} active " + f"(critical={ac.get('critical', 0)}, warning={ac.get('warning', 0)})") + for a in (obs.get("alerts") or [])[:5]: + lines.append(f" - [{a['severity']}] {a['dataset']}: {a['message']}") return lines diff --git a/api/lake_meta.py b/api/lake_meta.py new file mode 100644 index 0000000..4e0bc92 --- /dev/null +++ b/api/lake_meta.py @@ -0,0 +1,147 @@ +"""Shared lakehouse metadata + Trino helpers. + +Single source of truth for the business datasets the governance / data-quality / +lineage / observability features operate on, plus a small synchronous Trino +client. Imported by dq_monitor.py, observability.py, lineage.py and +catalog_governance.py so every feature reasons about the exact same tables that +the rest of the Command Center (pii_catalog, etl_offload, trino_federated) +already exposes. +""" + +from __future__ import annotations + +import os +from typing import Any + +import httpx + +TRINO_URL = os.getenv("TRINO_URL", "http://10.0.21.50:8089").rstrip("/") +TRINO_USER = os.getenv("TRINO_USER", "mo") + +# Canonical business datasets, aligned with pii_catalog.DATASETS / etl_offload. +# fqtn = fully-qualified Trino table name (catalog.schema.table) +# om_fqn = OpenMetadata FQN best-effort (governance falls back to native store) +DATASETS: list[dict[str, Any]] = [ + {"key": "orders", "label": "Sales orders", "engine": "PostgreSQL", "color": "#fbbf24", + "catalog": "postgres_sales", "schema": "public", "table": "sales_orders", + "fqtn": "postgres_sales.public.sales_orders", "pii_key": "postgres", + "om_fqn": "atc_trino.postgres_sales.public.sales_orders", + "key_col": "order_id", "ts_col": "order_ts", "amount_col": "amount", + "native_count": ("postgres", "public.sales_orders"), + "domain": "Sales", "topic": "atc.public.sales_orders"}, + {"key": "hr", "label": "HR events", "engine": "MySQL", "color": "#60a5fa", + "catalog": "mysql_hr", "schema": "hr", "table": "employee_events", + "fqtn": "mysql_hr.hr.employee_events", "pii_key": "mysql", + "om_fqn": "atc_trino.mysql_hr.hr.employee_events", + "key_col": "event_id", "ts_col": "event_ts", "amount_col": "salary_change", + "native_count": ("mysql", "hr.employee_events"), + "domain": "People", "topic": "atc.hr.employee_events"}, + {"key": "supply", "label": "Supply chain events", "engine": "MongoDB", "color": "#a78bfa", + "catalog": "mongodb_supplychain", "schema": "supplychain", "table": "events", + "fqtn": "mongodb_supplychain.supplychain.events", "pii_key": "mongodb", + "om_fqn": "atc_trino.mongodb_supplychain.supplychain.events", + "key_col": "event_id", "ts_col": "ts", "amount_col": "amount", + "native_count": ("mongodb", "supplychain.events"), + "domain": "Supply Chain", "topic": "atc.supplychain.events"}, + {"key": "telemetry", "label": "Device telemetry", "engine": "Cassandra", "color": "#22d3ee", + "catalog": "cassandra_telemetry", "schema": "telemetry", "table": "device_metrics", + "fqtn": "cassandra_telemetry.telemetry.device_metrics", "pii_key": "cassandra", + "om_fqn": "atc_trino.cassandra_telemetry.telemetry.device_metrics", + "key_col": "device_id", "ts_col": "metric_ts", "amount_col": "metric_value", + "unique_key": False, + "domain": "IoT", "topic": "atc.telemetry.device_metrics"}, + {"key": "curated", "label": "Curated masked (Iceberg)", "engine": "Iceberg / Trino", "color": "#34d399", + "catalog": "iceberg", "schema": "curated_masked", "table": "sales_orders_masked", + "fqtn": "iceberg.curated_masked.sales_orders_masked", "pii_key": "curated", + "om_fqn": "atc_trino.iceberg.curated_masked.sales_orders_masked", + "key_col": "order_id", "ts_col": "order_ts", "amount_col": "amount", + "domain": "Sales", "curated": True, "topic": None}, + {"key": "hadoop", "label": "Historical sales (HDFS)", "engine": "Iceberg / HDFS", "color": "#f472b6", + "catalog": "iceberg", "schema": "hadoop", "table": "historical_sales_hdfs", + "fqtn": "iceberg.hadoop.historical_sales_hdfs", "pii_key": "hadoop", + "om_fqn": "atc_trino.iceberg.hadoop.historical_sales_hdfs", + "key_col": "order_id", "ts_col": "order_ts", "amount_col": "amount", + "domain": "Sales", "curated": True, "topic": None}, +] + +DATASET_BY_KEY = {d["key"]: d for d in DATASETS} + +# Heuristics for picking the freshness / key column when discovering schema. +_TS_HINTS = ("_ts", "ts", "updated_at", "created_at", "event_time", "modified", "time") +_KEY_HINTS = ("_id", "id", "uuid", "key", "pk") + + +def trino(sql: str, timeout: float = 25.0) -> tuple[list[str], list[list[Any]]]: + """Run a Trino statement, following nextUri pages. Returns (columns, rows). + + `timeout` is both the per-request timeout AND an overall wall-clock deadline, + so a long full-scan (e.g. count(*) on a huge Cassandra table) is aborted and + cancelled instead of looping over nextUri pages for minutes and blocking the + caller's thread. + """ + import time as _t + cols: list[str] = [] + rows: list[list[Any]] = [] + deadline = _t.monotonic() + timeout + with httpx.Client(timeout=min(timeout, 15.0)) as client: + d = client.post(f"{TRINO_URL}/v1/statement", content=sql.encode(), + headers={"X-Trino-User": TRINO_USER}).json() + while True: + if d.get("error"): + raise RuntimeError(d["error"].get("message", "trino error")) + c = d.get("columns") + if c and not cols: + cols = [x["name"] for x in c] + rows += d.get("data") or [] + nxt = d.get("nextUri") + if not nxt: + break + if _t.monotonic() > deadline: + try: + client.delete(nxt) # cancel the running query server-side + except Exception: + pass + raise TimeoutError(f"trino query exceeded {timeout}s deadline") + d = client.get(nxt).json() + return cols, rows + + +def trino_scalar(sql: str, timeout: float = 25.0) -> Any: + _c, rows = trino(sql, timeout=timeout) + if rows and rows[0]: + return rows[0][0] + return None + + +def discover_columns(ds: dict[str, Any], timeout: float = 12.0) -> list[dict[str, str]]: + """[{name, type}] for a dataset's table via information_schema.""" + sql = (f"SELECT column_name, data_type FROM {ds['catalog']}.information_schema.columns " + f"WHERE table_name = '{ds['table']}'") + if ds.get("schema"): + sql += f" AND table_schema = '{ds['schema']}'" + try: + _c, rows = trino(sql, timeout=timeout) + return [{"name": r[0], "type": r[1]} for r in rows] + except Exception: + return [] + + +def pick_ts_col(columns: list[dict[str, str]], default: str | None = None) -> str | None: + names = [c["name"] for c in columns] + for c in columns: + if "timestamp" in (c.get("type") or "").lower() or "date" in (c.get("type") or "").lower(): + return c["name"] + for n in names: + if any(h in n.lower() for h in _TS_HINTS): + return n + return default if default in names else None + + +def pick_key_col(columns: list[dict[str, str]], default: str | None = None) -> str | None: + names = [c["name"] for c in columns] + if default in names: + return default + for n in names: + if any(n.lower() == h or n.lower().endswith(h) for h in _KEY_HINTS): + return n + return names[0] if names else None diff --git a/api/lineage.py b/api/lineage.py new file mode 100644 index 0000000..c8db352 --- /dev/null +++ b/api/lineage.py @@ -0,0 +1,248 @@ +"""Data lineage graph for the Command Center (Disease #6 — Traceability). + +Models the real end-to-end pipeline as an explicit, staged node/edge graph: + + source DB → Debezium/Kafka (CDC) → Spark / Kafka Connect → S3 Parquet lake + → Iceberg lakehouse (curated_masked / hadoop) → Trino federation + → serving (Elasticsearch, RAG/ChromaDB, vLLM, Command Center chat) + +Table-backed nodes are enriched with live Trino row counts. For the sales path +we expose column-level lineage with PII / masking status (reusing pii_catalog), +so an operator can trace any field from source to the masked curated layer. +OpenMetadata table lineage is layered in best-effort where the FQN resolves. + +Endpoints: + GET /api/lineage/graph?dataset= -> nodes + edges (+ column links) + GET /api/lineage/datasets -> selectable datasets +""" + +from __future__ import annotations + +import os +import time +from typing import Any + +import httpx +from fastapi import APIRouter +from fastapi.responses import JSONResponse + +from lake_meta import DATASETS, DATASET_BY_KEY, trino_scalar + +router = APIRouter(prefix="/api/lineage", tags=["lineage"]) + +OPENMETADATA_URL = os.getenv("OPENMETADATA_URL", "").rstrip("/") +OPENMETADATA_TOKEN = os.getenv("OPENMETADATA_TOKEN", "") + +_cache: dict[str, Any] = {"ts": 0.0, "data": None} +_TTL = 30.0 + +# Which source datasets flow through the CDC → lakehouse pipeline. +_SOURCE_KEYS = ["orders", "hr", "supply", "telemetry"] + + +def _is_active() -> tuple[bool, bool]: + gen = arch = False + try: + from trino_federated import generator_active + gen = bool(generator_active()) + except Exception: + pass + try: + from storage_s3 import archive_active + arch = bool(archive_active()) + except Exception: + pass + return gen, arch + + +def _obs_counts() -> dict[str, int | None]: + """Reuse the row counts the observability monitor already polls in-memory, + so the lineage graph never issues its own (potentially slow) count(*).""" + try: + from observability import _state, _lock + with _lock: + return {k: v.get("rows") for k, v in _state["datasets"].items()} + except Exception: + return {} + + +def _row_count(ds_key: str, counts: dict[str, int | None]) -> int | None: + return counts.get(ds_key) + + +def _pii_columns(ds_key: str) -> list[dict[str, Any]]: + """Column-level detail with PII/masking flags from pii_catalog (best-effort). + + ds_key is a lake_meta key; map it to the pii_catalog key first.""" + pii_key = (DATASET_BY_KEY.get(ds_key, {}) or {}).get("pii_key", ds_key) + try: + from pii_catalog import get_pii + data = get_pii() + for d in data.get("datasets", []): + if d.get("key") == pii_key or d.get("node_id") == pii_key: + return [{"name": c["name"], "category": c.get("category"), + "masked": bool(c.get("masked")), "pii": True} + for c in d.get("pii_columns", [])] + except Exception: + pass + return [] + + +def _om_lineage(fqn: str) -> dict[str, int]: + """OpenMetadata table lineage (best-effort, short timeout). Kept out of the + hot path by default; only called when LINEAGE_OM_ENRICH=1 is set.""" + if not OPENMETADATA_URL or not fqn or os.getenv("LINEAGE_OM_ENRICH", "0") != "1": + return {} + url = f"{OPENMETADATA_URL}/api/v1/lineage/table/name/{fqn}?upstreamDepth=1&downstreamDepth=1" + headers = {"Accept": "application/json"} + if OPENMETADATA_TOKEN: + headers["Authorization"] = f"Bearer {OPENMETADATA_TOKEN}" + try: + with httpx.Client(timeout=4.0, verify=False) as c: + r = c.get(url, headers=headers) + if r.status_code != 200: + return {} + j = r.json() + return {"upstream": len(j.get("upstreamEdges") or []), + "downstream": len(j.get("downstreamEdges") or [])} + except Exception: + return {} + + +def _build() -> dict[str, Any]: + gen_active, arch_active = _is_active() + counts = _obs_counts() + nodes: list[dict[str, Any]] = [] + edges: list[dict[str, Any]] = [] + seen: set[str] = set() + + def node(nid: str, label: str, ntype: str, stage: int, **meta: Any) -> None: + if nid in seen: + return + seen.add(nid) + nodes.append({"id": nid, "label": label, "type": ntype, "stage": stage, "meta": meta}) + + def edge(src: str, dst: str, label: str, kind: str, active: bool = False) -> None: + edges.append({"id": f"{src}->{dst}", "source": src, "target": dst, + "label": label, "kind": kind, "active": active}) + + # Stage 5/6 shared serving nodes + node("trino", "Trino", "engine", 5, engine="Trino", note="Federated SQL across every catalog") + node("es", "Elasticsearch", "serving", 6, engine="Elasticsearch", note="Business + infra search indices") + node("rag", "ChromaDB (RAG)", "serving", 6, engine="ChromaDB", note="Vector store / embeddings") + node("vllm", "vLLM", "serving", 6, engine="vLLM", note="LLM inference (GPU)") + node("cc", "Command Center", "serving", 6, engine="React", note="Knowledge Chat & dashboards") + edge("rag", "vllm", "context", "serve", active=True) + edge("vllm", "cc", "answers", "serve", active=True) + edge("trino", "es", "index business data", "serve") + edge("es", "cc", "search", "serve") + edge("trino", "cc", "dashboards", "serve") + + for key in _SOURCE_KEYS: + ds = DATASET_BY_KEY[key] + src_id = f"src_{key}" + topic_id = f"topic_{key}" + lake_id = f"lake_{key}" + rows = _row_count(key, counts) + cols = _pii_columns(key) + node(src_id, f"{ds['engine']}\n{ds['table']}", "source", 0, + engine=ds["engine"], table=ds["fqtn"], rows=rows, domain=ds.get("domain"), + columns=cols, ts_col=ds.get("ts_col"), key_col=ds.get("key_col"), + om=_om_lineage(ds.get("om_fqn", ""))) + + # CDC path (Debezium → Kafka topic) + if ds.get("topic"): + node(topic_id, f"Kafka\n{ds['topic']}", "stream", 1, + engine="Kafka", topic=ds["topic"], note="Debezium CDC topic") + edge(src_id, topic_id, "Debezium CDC", "cdc", active=gen_active) + node("connect", "Kafka Connect", "stream", 2, engine="Kafka Connect", + note="Debezium source + HDFS/S3 sinks") + edge(topic_id, "connect", "consume", "cdc", active=gen_active) + + # ETL offload (direct source → S3 Parquet lake) + node(lake_id, f"S3 lake\nlake/{key}", "storage", 3, engine="ObjectScale S3", + path=f"s3://data/lake/{key}/", note="Parquet parts (ETL offload)") + edge(src_id, lake_id, "ETL offload (Parquet)", "batch", active=arch_active) + + # Spark / Connect → curated + hadoop Iceberg tables + node("spark", "Spark", "compute", 2, engine="Spark", note="Transform & mask → curated") + edge("connect", "spark", "stream", "transform", active=gen_active) + + cur = DATASET_BY_KEY["curated"] + had = DATASET_BY_KEY["hadoop"] + node("iceberg_curated", f"Iceberg\n{cur['table']}", "lakehouse", 4, engine="Iceberg", + table=cur["fqtn"], rows=_row_count("curated", counts), masked_layer=True, + note="PII-masked curated layer", columns=_pii_columns("curated"), + om=_om_lineage(cur.get("om_fqn", ""))) + node("iceberg_hadoop", f"Iceberg/HDFS\n{had['table']}", "lakehouse", 4, engine="Iceberg / HDFS", + table=had["fqtn"], rows=_row_count("hadoop", counts), note="Historical sales on HDFS", + om=_om_lineage(had.get("om_fqn", ""))) + + edge("spark", "iceberg_curated", "mask + write", "transform", active=arch_active) + edge("lake_orders", "iceberg_hadoop", "register external", "batch", active=arch_active) + edge("src_orders", "iceberg_curated", "curate (masked)", "transform") + + # Lakehouse → Trino + for nid in ("iceberg_curated", "iceberg_hadoop"): + edge(nid, "trino", "query", "serve") + for key in _SOURCE_KEYS: + edge(f"src_{key}", "trino", "federate", "serve") + + # Lake → RAG (documents/parquet feeding embeddings is conceptual) + edge("iceberg_curated", "rag", "embed (masked-safe)", "serve") + + # Column-level links: source sales_orders → curated masked + column_links: list[dict[str, Any]] = [] + src_cols = {c["name"] for c in _pii_columns("orders")} + cur_cols = {c["name"] for c in _pii_columns("curated")} + for cn in sorted(src_cols & cur_cols): + masked = any(c["name"] == cn and c["masked"] for c in _pii_columns("curated")) + column_links.append({"source": "src_orders", "target": "iceberg_curated", + "column": cn, "masked": masked}) + + return { + "ok": True, + "generated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), + "active": {"generator": gen_active, "archive": arch_active}, + "stages": ["Sources", "CDC / Stream", "Compute", "Lake storage", + "Lakehouse", "Federation", "Serving"], + "nodes": nodes, + "edges": edges, + "column_links": column_links, + "om_connected": bool(OPENMETADATA_URL), + } + + +def get_graph(use_cache: bool = True) -> dict[str, Any]: + now = time.time() + if use_cache and _cache["data"] and now - _cache["ts"] < _TTL: + return _cache["data"] + data = _build() + _cache["data"] = data + _cache["ts"] = now + return data + + +@router.get("/datasets") +async def datasets() -> JSONResponse: + return JSONResponse({"ok": True, "datasets": [ + {"key": d["key"], "label": d["label"], "engine": d["engine"], + "color": d["color"], "table": d["fqtn"]} + for d in DATASETS]}) + + +@router.get("/graph") +async def graph(dataset: str | None = None, refresh: bool = False) -> JSONResponse: + from starlette.concurrency import run_in_threadpool + data = await run_in_threadpool(get_graph, not refresh) + if dataset and dataset in DATASET_BY_KEY: + # Filter to nodes reachable from / to the selected source, keep serving spine. + keep = {f"src_{dataset}", f"topic_{dataset}", f"lake_{dataset}", + "connect", "spark", "iceberg_curated", "iceberg_hadoop", + "trino", "es", "rag", "vllm", "cc"} + nodes = [n for n in data["nodes"] if n["id"] in keep] + node_ids = {n["id"] for n in nodes} + edges = [e for e in data["edges"] if e["source"] in node_ids and e["target"] in node_ids] + out = {**data, "nodes": nodes, "edges": edges, "focus": dataset} + return JSONResponse(out) + return JSONResponse(data) diff --git a/api/main.py b/api/main.py index 2422c57..bdab09b 100644 --- a/api/main.py +++ b/api/main.py @@ -57,6 +57,10 @@ from spark_workbench import router as spark_workbench_router from pii_catalog import router as pii_router from trino_federated import router as federated_router from etl_offload import router as etl_offload_router +from lineage import router as lineage_router +from dq_monitor import router as dq_router +from observability import router as observability_router +from catalog_governance import router as governance_router from ssh_terminal import ssh_session from node_registry import NODE_IDS, NODE_AGENT, NODE_REGISTRY, is_node_id from node_ops import build_node_detail, probe_node, run_node_probe_task @@ -768,6 +772,10 @@ app.include_router(spark_workbench_router) app.include_router(pii_router) app.include_router(federated_router) app.include_router(etl_offload_router) +app.include_router(lineage_router) +app.include_router(dq_router) +app.include_router(observability_router) +app.include_router(governance_router) app.add_middleware( CORSMiddleware, allow_origins=["*"], diff --git a/api/observability.py b/api/observability.py new file mode 100644 index 0000000..bfe5492 --- /dev/null +++ b/api/observability.py @@ -0,0 +1,370 @@ +"""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 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"]}) diff --git a/ui/src/components/features/DataExplorerView.tsx b/ui/src/components/features/DataExplorerView.tsx index fe735e0..ea644fc 100644 --- a/ui/src/components/features/DataExplorerView.tsx +++ b/ui/src/components/features/DataExplorerView.tsx @@ -16,12 +16,18 @@ import { HardDrive, ShieldCheck, Radio, + GitBranch, + Lock, } from 'lucide-react' import { cn } from '../../lib/utils' import { TrinoFederationView, type SubTab } from './TrinoFederationView' import { LiveDashboard } from './LiveDashboard' +import { LineageView } from './LineageView' +import { GovernanceOwnershipView } from './GovernanceOwnershipView' +import { GovernanceAccessView } from './GovernanceAccessView' +import { ObservabilityView } from './ObservabilityView' -type ExplorerTab = 'business' | 'live' | SubTab +type ExplorerTab = 'business' | 'live' | SubTab | 'lineage' | 'ownership' | 'access' | 'observability' const TABS: { id: ExplorerTab; label: string; icon: typeof Users; hint: string; live?: boolean }[] = [ { id: 'business', label: 'Business Overview', icon: BarChart3, hint: 'Customers, orders, workforce, supply chain & telemetry across every source' }, @@ -29,6 +35,10 @@ const TABS: { id: ExplorerTab; label: string; icon: typeof Users; hint: string; { id: 'federated', label: 'Federated (Trino)', icon: Network, hint: 'One SQL across all 5 databases + region scorecard joined live' }, { id: 'lake', label: 'Hadoop Lake', icon: HardDrive, hint: 'All business data mirrored as external Iceberg tables on HDFS' }, { id: 'dictionary', label: 'Data Dictionary', icon: ShieldCheck, hint: 'Every table & column with PII / masking status — exactly what the assistant sees' }, + { id: 'lineage', label: 'Lineage', icon: GitBranch, hint: 'End-to-end data lineage with column-level PII tracing from source to curated layer' }, + { id: 'ownership', label: 'Ownership', icon: UserCog, hint: 'Data owners, stewards, tiers & business glossary — accountability per dataset' }, + { id: 'access', label: 'Access & Policies', icon: Lock, hint: 'Governance posture: PII masking, ownership, live DQ & alerts vs each data contract' }, + { id: 'observability', label: 'Observability', icon: Activity, hint: 'Volume, freshness & schema-drift monitoring with live alerts across every table' }, ] type Bucket = { key: string; count: number; value?: number } @@ -295,6 +305,12 @@ export function DataExplorerView() { {/* Trino federation / lake / dictionary tabs */} {(view === 'federated' || view === 'lake' || view === 'dictionary') && } + {/* Governance / lineage / observability sub-tabs */} + {view === 'lineage' &&
} + {view === 'ownership' &&
} + {view === 'access' &&
} + {view === 'observability' &&
} + {/* ───────── BUSINESS OVERVIEW ───────── */} {view === 'business' && ( <> diff --git a/ui/src/components/features/DataQualityView.tsx b/ui/src/components/features/DataQualityView.tsx index 298f52f..d31f09c 100644 --- a/ui/src/components/features/DataQualityView.tsx +++ b/ui/src/components/features/DataQualityView.tsx @@ -11,9 +11,11 @@ import { Table2, Upload, XCircle, + Gauge, } from 'lucide-react' import { cn } from '../../lib/utils' import { subTabActive, subTabIdle } from '../../lib/tabActive' +import { DqMonitoringPanel } from './DqMonitoringPanel' type Dimension = { id: string @@ -167,13 +169,13 @@ type ReportSummary = { columns: number } -type Tab = 'assess' | 'docling' | 'reports' +type Tab = 'assess' | 'docling' | 'reports' | 'monitoring' const SCORE_COLOR = (s: number) => (s >= 80 ? 'text-success' : s >= 60 ? 'text-warning' : 'text-danger') const BAR_COLOR = (s: number) => (s >= 80 ? 'bg-success' : s >= 60 ? 'bg-warning' : 'bg-danger') export function DataQualityView() { - const [tab, setTab] = useState('assess') + const [tab, setTab] = useState('monitoring') const [caps, setCaps] = useState(null) const [loading, setLoading] = useState(false) const [assess, setAssess] = useState(null) @@ -252,6 +254,7 @@ export function DataQualityView() { } const tabs: { id: Tab; label: string; icon: typeof FileSearch }[] = [ + { id: 'monitoring', label: 'Live Monitoring', icon: Gauge }, { id: 'assess', label: 'Maturity Assessment', icon: FileSearch }, { id: 'docling', label: 'Docling Parser', icon: FileText }, { id: 'reports', label: 'Reports', icon: CheckCircle2 }, @@ -312,6 +315,8 @@ export function DataQualityView() { )} + {tab === 'monitoring' && } + {tab === 'assess' && (
+ volume: number | null; volume_delta: number | null; freshness_age_min: number | null + issues: string[]; columns?: number; worst_columns?: { name: string; completeness: number; nulls: number }[] + trend: Trend[]; pending?: boolean; error?: string +} +type Resp = { + ok: boolean; enabled: boolean; running: boolean; cycles: number; sample: number + platform_score: number | null; dimension_averages: Record; cards: Card[] + feed: { ts: string; text: string; level: string }[] +} + +const DIMS = ['completeness', 'uniqueness', 'validity', 'freshness'] + +function ScoreRing({ score, color }: { score: number | null; color: string }) { + const r = 26 + const c = 2 * Math.PI * r + const pct = score == null ? 0 : score / 100 + return ( +
+ + + + +
+ {score == null ? '—' : Math.round(score)} +
+
+ ) +} + +function TrendLine({ data, color }: { data: Trend[]; color: string }) { + const pts = data.slice(-30) + if (pts.length < 2) return
+ const w = 200 + const h = 32 + const step = w / (pts.length - 1) + const line = pts.map((p, i) => `${i === 0 ? 'M' : 'L'}${(i * step).toFixed(1)},${(h - (p.score / 100) * (h - 4) - 2).toFixed(1)}`).join(' ') + return ( + + + + ) +} + +export function DqMonitoringPanel() { + const [data, setData] = useState(null) + const [loading, setLoading] = useState(true) + const [running, setRunning] = useState(false) + + const load = useCallback(async () => { + try { + const r = await fetch('/api/dq/scorecards') + if (r.ok) setData(await r.json()) + } catch { /* */ } finally { + setLoading(false) + } + }, []) + + useEffect(() => { load() }, [load]) + useEffect(() => { const t = setInterval(load, 10000); return () => clearInterval(t) }, [load]) + + const runNow = async () => { + setRunning(true) + try { + const r = await fetch('/api/dq/run', { method: 'POST' }) + if (r.ok) { const j = await r.json(); if (j.scorecards) setData(j.scorecards) } + } catch { /* */ } finally { + setRunning(false) + } + } + + const dimColor = (v: number) => (v >= 90 ? '#34d399' : v >= 75 ? '#fbbf24' : v >= 50 ? '#fb923c' : '#f87171') + + return ( +
+ {/* header */} +
+
+ = 80 ? '#34d399' : '#fbbf24'} /> +
+

Platform DQ score

+

{data?.platform_score ?? '—'}

+

{data?.cycles ?? 0} cycles · live via Trino

+
+
+ {DIMS.map((dim) => { + const v = data?.dimension_averages?.[dim] + return ( +
+

{dim}

+

{v ?? '—'}

+
+ ) + })} +
+ +
+ + Continuous quality checks on live tables (completeness · uniqueness/dedup · validity · freshness){data?.running && ' · running…'} + +
+ + +
+
+ + {/* scorecards */} +
+ {(data?.cards || []).map((c) => ( +
+
+ +
+
+ + {c.label} + {c.engine} +
+

{c.table}

+ {c.pending ? ( +

Awaiting first cycle…

+ ) : c.error ? ( +

{c.error}

+ ) : ( +
+ {DIMS.filter((d) => c.dimensions[d] != null).map((d) => ( +
+ {d} +
+
+
+ {c.dimensions[d]} +
+ ))} +
+ )} +
+
+
+
+ {c.volume != null && {c.volume.toLocaleString()} rows} + {c.volume_delta != null && c.volume_delta !== 0 && ( + 0 ? 'text-emerald-400' : 'text-rose-400'}>{c.volume_delta > 0 ? '+' : ''}{c.volume_delta.toLocaleString()} + )} + {c.freshness_age_min != null && {c.freshness_age_min.toFixed(0)}m} +
+
+
+ {c.issues && c.issues.length > 0 && ( +
+ {c.issues.map((iss, i) => ( + {iss} + ))} +
+ )} +
+ ))} +
+
+ ) +} diff --git a/ui/src/components/features/GovernanceAccessView.tsx b/ui/src/components/features/GovernanceAccessView.tsx new file mode 100644 index 0000000..ee2d55d --- /dev/null +++ b/ui/src/components/features/GovernanceAccessView.tsx @@ -0,0 +1,117 @@ +import { useCallback, useEffect, useState } from 'react' +import { ShieldCheck, RefreshCw, Loader2, Lock, Unlock, Check, X, FileCheck2, AlertTriangle } from 'lucide-react' +import { cn } from '../../lib/utils' + +type Check = { name: string; ok: boolean; value: unknown; target: unknown } +type Pii = { pii_count: number; all_masked: boolean; masked: number; unmasked: number } +type Alert = { type: string; severity: string; message: string } +type Posture = { + key: string; label: string; engine: string; color: string; table: string + owner?: string | null; steward?: string | null; tier?: string | null + pii: Pii; dq_score?: number | null; issues: string[]; alerts: Alert[] + contract: Record; checks: Check[]; compliant: boolean +} +type Resp = { ok: boolean; datasets: Posture[]; summary: { total: number; compliant: number; non_compliant: number } } + +export function GovernanceAccessView() { + const [data, setData] = useState(null) + const [loading, setLoading] = useState(true) + + const load = useCallback(async () => { + setLoading(true) + try { + const r = await fetch('/api/governance/posture') + if (r.ok) setData(await r.json()) + } catch { /* */ } finally { + setLoading(false) + } + }, []) + + useEffect(() => { load() }, [load]) + useEffect(() => { const t = setInterval(load, 12000); return () => clearInterval(t) }, [load]) + + const sev = (s: string) => (s === 'critical' ? 'text-rose-400' : s === 'warning' ? 'text-amber-400' : 'text-sky-400') + + return ( +
+
+ + + + +
+ +
+ Governance posture = ownership + PII masking + live DQ + observability alerts vs each data contract + +
+ +
+ {(data?.datasets || []).map((d) => ( +
+
+ + {d.label} + + {d.compliant ? : } {d.compliant ? 'Compliant' : 'Action needed'} + +
+ +
+ {d.checks.map((c) => ( +
+ {c.ok ? : } + {c.name} + {String(c.value)} +
+ ))} +
+ +
+ Owner: {d.owner || '—'} + · Tier: {d.tier || '—'} + · + {d.pii.unmasked === 0 + ? <> {d.pii.masked}/{d.pii.pii_count} PII masked + : <> {d.pii.unmasked} PII visible} + +
+ + {d.alerts.length > 0 && ( +
+ {d.alerts.map((a, i) => ( +

+ {a.message} +

+ ))} +
+ )} + +

+ Contract: DQ ≥ {d.contract.min_score} · completeness ≥ {d.contract.min_completeness}% · freshness ≤ {d.contract.min_freshness_min}m · crit alerts ≤ {d.contract.max_critical_alerts} +

+
+ ))} +
+
+ ) +} + +function Kpi({ icon: Icon, label, value, accent }: { icon: typeof ShieldCheck; label: string; value: string; accent: string }) { + return ( +
+
+ +
+
+

{label}

+

{value}

+
+
+ ) +} diff --git a/ui/src/components/features/GovernanceOwnershipView.tsx b/ui/src/components/features/GovernanceOwnershipView.tsx new file mode 100644 index 0000000..b50b64d --- /dev/null +++ b/ui/src/components/features/GovernanceOwnershipView.tsx @@ -0,0 +1,246 @@ +import { useCallback, useEffect, useState } from 'react' +import { UserCircle, RefreshCw, Loader2, AlertTriangle, BookOpen, ShieldCheck, Check, X } from 'lucide-react' +import { cn } from '../../lib/utils' + +type Pii = { pii_count: number; all_masked: boolean; masked: number; unmasked: number } +type DsRow = { + key: string; label: string; engine: string; color: string; table: string + domain?: string; owner?: string | null; steward?: string | null; team?: string | null + tier?: string | null; classification?: string | null; updated_at?: string | null + orphan: boolean; pii: Pii +} +type Summary = { total: number; orphans: number; stewarded: number; owned: number } +type GUser = { id: string | null; name: string; display: string; type: string } +type Term = { name: string; description: string; domain?: string; related?: string[] } + +const TIERS = ['Tier1', 'Tier2', 'Tier3'] +const CLASSES = ['Public', 'Internal', 'Confidential', 'Restricted'] + +export function GovernanceOwnershipView() { + const [rows, setRows] = useState([]) + const [summary, setSummary] = useState(null) + const [users, setUsers] = useState([]) + const [glossary, setGlossary] = useState([]) + const [glossarySource, setGlossarySource] = useState('') + const [loading, setLoading] = useState(true) + const [omConnected, setOmConnected] = useState(false) + const [editKey, setEditKey] = useState(null) + const [form, setForm] = useState<{ owner: string; steward: string; team: string; tier: string; classification: string }>( + { owner: '', steward: '', team: '', tier: '', classification: '' }, + ) + const [saving, setSaving] = useState(false) + + const load = useCallback(async () => { + setLoading(true) + try { + const [d, u, g] = await Promise.all([ + fetch('/api/governance/datasets').then((r) => r.json()), + fetch('/api/governance/users').then((r) => r.json()), + fetch('/api/governance/glossary').then((r) => r.json()), + ]) + setRows(d.datasets || []) + setSummary(d.summary || null) + setOmConnected(!!d.om_connected) + setUsers(u.users || []) + setGlossary(g.terms || []) + setGlossarySource(g.source || '') + } catch { /* */ } finally { + setLoading(false) + } + }, []) + + useEffect(() => { load() }, [load]) + + const openEdit = (r: DsRow) => { + setEditKey(r.key) + setForm({ owner: r.owner || '', steward: r.steward || '', team: r.team || '', tier: r.tier || '', classification: r.classification || '' }) + } + + const save = async () => { + if (!editKey) return + setSaving(true) + try { + await fetch('/api/governance/assign', { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ key: editKey, ...form }), + }) + setEditKey(null) + await load() + } catch { /* */ } finally { + setSaving(false) + } + } + + const people = users.filter((u) => u.type === 'user') + const teams = users.filter((u) => u.type === 'team') + + return ( +
+ {/* KPIs */} +
+ + + + +
+ +
+ + OpenMetadata {omConnected ? 'connected' : 'offline'} · assignments stored locally{omConnected ? ' + synced to OM' : ''} + + +
+ + {/* ownership matrix */} +
+ + + + + + + + + + + + + {rows.map((r) => ( + + + + + + + + + + ))} + +
DatasetOwnerStewardTeamTierPII +
+
+ + {r.label} +
+ {r.table} + 		+ {r.owner ? {r.owner} + : unassigned} + {r.steward || '—'}{r.team || '—'} + {r.tier ? {r.tier} : '—'} + + {r.pii.pii_count > 0 ? ( + + {r.pii.masked}/{r.pii.pii_count} masked + + ) : none} + + +
+
+ + {/* glossary */} +

+ Business glossary {glossarySource && · {glossarySource}} +

+
+ {glossary.map((t) => ( +
+
+ + {t.name} + {t.domain && {t.domain}} +
+

{t.description}

+ {t.related && t.related.length > 0 && ( +
+ {t.related.map((rl) => {rl})} +
+ )} +
+ ))} +
+ + {/* assign modal */} + {editKey && ( +
setEditKey(null)}> +
e.stopPropagation()}> +
+

Assign ownership — {rows.find((r) => r.key === editKey)?.label}

+ +
+
+ + + + + + + + + +
+ + + + + + +
+
+
+ + +
+
+
+ )} +
+ ) +} + +function Kpi({ icon: Icon, label, value, accent }: { icon: typeof UserCircle; label: string; value: string; accent: string }) { + return ( +
+
+ +
+
+

{label}

+

{value}

+
+
+ ) +} + +function Field({ label, children }: { label: string; children: React.ReactNode }) { + return ( +
+ + {children} +
+ ) +} diff --git a/ui/src/components/features/LineageView.tsx b/ui/src/components/features/LineageView.tsx new file mode 100644 index 0000000..bb2cb30 --- /dev/null +++ b/ui/src/components/features/LineageView.tsx @@ -0,0 +1,309 @@ +import { useCallback, useEffect, useLayoutEffect, useRef, useState } from 'react' +import { GitBranch, RefreshCw, Loader2, Database, Cpu, HardDrive, Layers, Network, Sparkles, Lock, ShieldCheck } from 'lucide-react' +import { cn } from '../../lib/utils' + +type LCol = { name: string; category?: string; masked: boolean; pii?: boolean } +type LNode = { + id: string + label: string + type: string + stage: number + meta: { + engine?: string + table?: string + rows?: number | null + note?: string + columns?: LCol[] + masked_layer?: boolean + domain?: string + topic?: string + path?: string + om?: { upstream?: number; downstream?: number } + } +} +type LEdge = { id: string; source: string; target: string; label: string; kind: string; active: boolean } +type Graph = { + ok: boolean + generated_at: string + active: { generator: boolean; archive: boolean } + stages: string[] + nodes: LNode[] + edges: LEdge[] + column_links: { source: string; target: string; column: string; masked: boolean }[] + om_connected: boolean +} +type DsOpt = { key: string; label: string; engine: string; color: string; table: string } + +const TYPE_ICON: Record = { + source: Database, + stream: Network, + compute: Cpu, + storage: HardDrive, + lakehouse: Layers, + engine: Network, + serving: Sparkles, +} +const KIND_COLOR: Record = { + cdc: '#f472b6', + batch: '#fbbf24', + transform: '#a78bfa', + serve: '#38bdf8', +} + +function fmtRows(n?: number | null) { + if (n == null) return null + if (n >= 1_000_000) return `${(n / 1_000_000).toFixed(1)}M` + if (n >= 1_000) return `${(n / 1_000).toFixed(1)}K` + return String(n) +} + +export function LineageView() { + const [graph, setGraph] = useState(null) + const [datasets, setDatasets] = useState([]) + const [focus, setFocus] = useState('') + const [loading, setLoading] = useState(true) + const [selected, setSelected] = useState(null) + + const wrapRef = useRef(null) + const nodeRefs = useRef>(new Map()) + const [coords, setCoords] = useState>(new Map()) + + const load = useCallback(async () => { + setLoading(true) + try { + const url = focus ? `/api/lineage/graph?dataset=${focus}` : '/api/lineage/graph' + const r = await fetch(url) + if (r.ok) setGraph(await r.json()) + } catch { /* */ } finally { + setLoading(false) + } + }, [focus]) + + useEffect(() => { + fetch('/api/lineage/datasets').then((r) => r.json()).then((d) => setDatasets(d.datasets || [])).catch(() => {}) + }, []) + useEffect(() => { load() }, [load]) + useEffect(() => { + const t = setInterval(load, 15000) + return () => clearInterval(t) + }, [load]) + + // measure node positions for edge drawing + useLayoutEffect(() => { + if (!wrapRef.current || !graph) return + const measure = () => { + const wrap = wrapRef.current + if (!wrap) return + const base = wrap.getBoundingClientRect() + const next = new Map() + nodeRefs.current.forEach((el, id) => { + const r = el.getBoundingClientRect() + next.set(id, { x: r.left - base.left + wrap.scrollLeft, y: r.top - base.top + wrap.scrollTop, w: r.width, h: r.height }) + }) + setCoords(next) + } + measure() + const ro = new ResizeObserver(measure) + if (wrapRef.current) ro.observe(wrapRef.current) + nodeRefs.current.forEach((el) => ro.observe(el)) + return () => ro.disconnect() + }, [graph]) + + const stages = graph?.stages || [] + const byStage: Record = {} + ;(graph?.nodes || []).forEach((n) => { (byStage[n.stage] = byStage[n.stage] || []).push(n) }) + + const selNode = graph?.nodes.find((n) => n.id === selected) || null + const connectedEdges = new Set( + (graph?.edges || []).filter((e) => !selected || e.source === selected || e.target === selected).map((e) => e.id), + ) + + return ( +
+ {/* controls */} +
+ Trace dataset + + {datasets.map((d) => ( + + ))} +
+ {graph?.active && ( + + CDC + ETL + + )} + +
+
+ +
+ {/* graph */} +
+ {/* edges overlay */} + + + + + + + {(graph?.edges || []).map((e) => { + const a = coords.get(e.source) + const b = coords.get(e.target) + if (!a || !b) return null + const x1 = a.x + a.w + const y1 = a.y + a.h / 2 + const x2 = b.x + const y2 = b.y + b.h / 2 + const dx = Math.max(40, Math.abs(x2 - x1) / 2) + const path = `M${x1},${y1} C${x1 + dx},${y1} ${x2 - dx},${y2} ${x2},${y2}` + const color = KIND_COLOR[e.kind] || '#64748b' + const dim = selected && !connectedEdges.has(e.id) + return ( + + + {e.active && ( + + )} + + + ) + })} + + + {/* stage columns */} +
+ {stages.map((label, si) => ( +
+
{label}
+ {(byStage[si] || []).map((n) => { + const Icon = TYPE_ICON[n.type] || Database + const isSel = selected === n.id + const rows = fmtRows(n.meta.rows) + return ( +
{ if (el) nodeRefs.current.set(n.id, el); else nodeRefs.current.delete(n.id) }} + onClick={() => setSelected(isSel ? null : n.id)} + className={cn( + 'relative z-10 cursor-pointer rounded-lg border bg-surface-raised p-2 transition-all', + isSel ? 'border-docker ring-1 ring-docker/40' : 'border-border hover:border-docker/50', + )} + > +
+ + + {n.label.split('\n')[0]} + + {n.meta.masked_layer && } +
+ {n.label.includes('\n') && ( +

+ {n.label.split('\n')[1]} +

+ )} +
+ {rows && {rows} rows} + {n.meta.columns && n.meta.columns.length > 0 && ( + {n.meta.columns.length} PII + )} +
+
+ ) + })} +
+ ))} +
+
+ + {/* detail */} +
+ {!selNode ? ( +
+ +

Click any node to inspect its schema, row count and column-level PII lineage.

+ {graph && ( +

+ {graph.om_connected ? 'OpenMetadata lineage layered in.' : 'OpenMetadata not connected.'} +

+ )} +
+ ) : ( +
+

{selNode.label.split('\n')[0]}

+ {selNode.meta.table &&

{selNode.meta.table}

} +
+ {selNode.meta.engine && } + {selNode.meta.rows != null && } + {selNode.meta.domain && } + {selNode.meta.topic && } + {selNode.meta.om && (selNode.meta.om.upstream != null) && ( + + )} +
+ {selNode.meta.note &&

{selNode.meta.note}

} + {selNode.meta.columns && selNode.meta.columns.length > 0 && ( +
+

+ PII columns +

+
+ {selNode.meta.columns.map((c) => ( +
+ {c.name} + {c.category && {c.category}} + {c.masked + ? masked + : visible} +
+ ))} +
+
+ )} + {graph?.column_links && graph.column_links.length > 0 && (selNode.id === 'src_orders' || selNode.id === 'iceberg_curated') && ( +
+

Column lineage → curated

+
+ {graph.column_links.map((l) => ( +
+ {l.column} + → + {l.masked ? : visible} +
+ ))} +
+
+ )} +
+ )} +
+
+
+ ) +} + +function Info({ label, value }: { label: string; value: string }) { + return ( +
+

{label}

+

{value}

+
+ ) +} diff --git a/ui/src/components/features/ObservabilityView.tsx b/ui/src/components/features/ObservabilityView.tsx new file mode 100644 index 0000000..cacf042 --- /dev/null +++ b/ui/src/components/features/ObservabilityView.tsx @@ -0,0 +1,141 @@ +import { useCallback, useEffect, useState } from 'react' +import { Activity, RefreshCw, Loader2, AlertTriangle, Clock, Database, TrendingUp, TrendingDown } from 'lucide-react' +import { cn } from '../../lib/utils' + +type Point = { t: string; rows: number; delta: number } +type DsMetric = { + key: string; label: string; engine: string; color: string; table: string + rows: number | null; delta: number | null; freshness_age_min: number | null + columns: number | null; stalled_cycles: number; error: string | null; ts: string | null + series: Point[] +} +type Alert = { id: string; dataset: string; type: string; severity: string; message: string; count: number; ts: string; last_ts: string } +type Metrics = { + ok: boolean; enabled: boolean; cycles: number; freshness_min: number + alert_counts: { critical: number; warning: number; info: number; total: number } + datasets: DsMetric[] +} + +function Spark({ data, color }: { data: Point[]; color: string }) { + const pts = data.slice(-40) + if (pts.length < 2) return
+ const w = 240 + const h = 40 + const vals = pts.map((p) => p.rows) + const max = Math.max(...vals) + const min = Math.min(...vals) + const range = max - min || 1 + const step = w / (pts.length - 1) + const line = pts.map((p, i) => `${i === 0 ? 'M' : 'L'}${(i * step).toFixed(1)},${(h - ((p.rows - min) / range) * (h - 6) - 3).toFixed(1)}`).join(' ') + return ( + + + + + ) +} + +export function ObservabilityView() { + const [metrics, setMetrics] = useState(null) + const [alerts, setAlerts] = useState([]) + const [loading, setLoading] = useState(true) + + const load = useCallback(async () => { + setLoading(true) + try { + const [m, a] = await Promise.all([ + fetch('/api/observability/metrics').then((r) => r.json()), + fetch('/api/observability/alerts').then((r) => r.json()), + ]) + setMetrics(m) + setAlerts(a.active || []) + } catch { /* */ } finally { + setLoading(false) + } + }, []) + + useEffect(() => { load() }, [load]) + useEffect(() => { const t = setInterval(load, 8000); return () => clearInterval(t) }, [load]) + + const ac = metrics?.alert_counts + const sevBorder = (s: string) => (s === 'critical' ? 'border-rose-500/40 bg-rose-500/5' : s === 'warning' ? 'border-amber-500/40 bg-amber-500/5' : 'border-sky-500/40 bg-sky-500/5') + const sevText = (s: string) => (s === 'critical' ? 'text-rose-400' : s === 'warning' ? 'text-amber-400' : 'text-sky-400') + + return ( +
+
+ + + + +
+ +
+ Tracking volume, freshness & schema drift across every business table + +
+ + {/* active alerts */} + {alerts.length > 0 && ( +
+ {alerts.map((a) => ( +
+ + {a.dataset} + {a.message} + {a.type} + {a.count > 1 && ×{a.count}} +
+ ))} +
+ )} + + {/* per-dataset volume + freshness */} +
+ {(metrics?.datasets || []).map((d) => { + const stale = d.freshness_age_min != null && d.freshness_age_min > (metrics?.freshness_min ?? 30) + return ( +
+
+ + {d.label} + {d.delta != null && d.delta !== 0 && ( + 0 ? 'text-emerald-400' : 'text-rose-400')}> + {d.delta > 0 ? : } + {d.delta > 0 ? '+' : ''}{d.delta.toLocaleString()} + + )} +
+ +
+ Rows: {d.rows?.toLocaleString() ?? '—'} + + {d.freshness_age_min != null ? `${d.freshness_age_min.toFixed(0)}m old` : 'n/a'} + + {d.columns != null && {d.columns} cols} + {d.stalled_cycles > 0 && stalled ×{d.stalled_cycles}} + {d.error && err} +
+
+ ) + })} +
+
+ ) +} + +function Kpi({ icon: Icon, label, value, accent }: { icon: typeof Activity; label: string; value: string; accent: string }) { + return ( +
+
+ +
+
+

{label}

+

{value}

+
+
+ ) +}