diff --git a/api/lab_context.py b/api/lab_context.py index 186d008..2823617 100644 --- a/api/lab_context.py +++ b/api/lab_context.py @@ -632,6 +632,77 @@ def _section_command_center(c: dict[str, Any]) -> list[str]: return lines +def collect_governance(log: TerminalLogFn | None = None) -> dict[str, Any]: + """In-process governance snapshot: CDC, movements, PII catalog, lineage. + + Reads directly from the live modules (no HTTP) so the LLM always has the + current data-platform governance picture. + """ + out: dict[str, Any] = {"openmetadata_url": os.getenv("OPENMETADATA_URL", "http://10.0.21.47:8585")} + try: + from cdc_consumer import snapshot as _cdc_snapshot + out["cdc"] = _cdc_snapshot(15) + except Exception as exc: + out["cdc"] = {"error": str(exc)} + try: + from movements import MOVEMENTS, last_runs + runs = last_runs() + out["movements"] = [ + {"id": m["id"], "label": m["label"], "kind": m["kind"], "from": m["from"], "to": m["to"], + "last": runs.get(m["id"])} + for m in MOVEMENTS + ] + except Exception as exc: + out["movements"] = {"error": str(exc)} + try: + from pii_catalog import get_pii + pii = get_pii() + out["pii_summary"] = pii.get("summary", {}) + out["pii_source"] = pii.get("source") + out["pii_datasets"] = [ + {"label": d["label"], "pii_count": d["pii_count"], "all_masked": d["all_masked"], + "columns": [f"{c['name']}:{c['category']}{'(masked)' if c['masked'] else ''}" for c in d["pii_columns"]]} + for d in pii.get("datasets", []) + ] + except Exception as exc: + out["pii_summary"] = {"error": str(exc)} + out["lineage"] = [ + "postgres_sales.public.sales_orders -> iceberg.curated_masked.sales_orders_masked (PII masked)", + "mysql_hr.hr.employee_events -> iceberg.curated_masked.employee_events_masked (PII masked)", + "hdfs:/data/historical/sales_orders -> iceberg.hadoop.historical_sales_hdfs", + ] + return out + + +def _section_governance(g: dict[str, Any]) -> list[str]: + lines = ["Data governance (CDC · movements · PII · lineage):", + f" OpenMetadata catalog: {g.get('openmetadata_url')} (catalog, lineage, PII auto-classification)"] + cdc = g.get("cdc") or {} + if "error" not in cdc: + by_src = ", ".join(f"{k}={v}" for k, v in (cdc.get("by_source") or {}).items()) or "none" + lines.append(f" CDC stream: connected={cdc.get('connected')} consumed={cdc.get('consumed')} " + f"changes/15m={cdc.get('window_total')} ({by_src})") + movements = g.get("movements") + if isinstance(movements, list): + lines.append(" Data movements (trigger via Command Center / ETL agent):") + for m in movements: + last = m.get("last") or {} + st = last.get("state", "never run") + extra = f" rows={last.get('rows')}" if last.get("rows") is not None else "" + lines.append(f" - {m['id']}: {m['from']}→{m['to']} [{m['kind']}] last={st}{extra}") + ps = g.get("pii_summary") or {} + if "error" not in ps: + lines.append(f" PII catalog ({g.get('pii_source')}): {ps.get('pii_columns', 0)} PII cols, " + f"{ps.get('masked_columns', 0)} masked / {ps.get('unmasked_columns', 0)} unmasked") + for d in g.get("pii_datasets") or []: + tag = "all masked" if d["all_masked"] else "UNMASKED" + lines.append(f" - {d['label']}: {d['pii_count']} PII [{tag}] {', '.join(d['columns'][:8])}") + lines.append(" Lineage:") + for ln in g.get("lineage") or []: + lines.append(f" - {ln}") + return lines + + def _section_cluster_registry(_: dict[str, Any]) -> list[str]: """Static cluster map — always available even when probes fail.""" lines = ["Cluster infrastructure map (Proxmox VMs & roles):"] @@ -661,10 +732,11 @@ SECTION_BUILDERS = { "gpu": _section_gpu, "objectscale": _section_objectscale, "command_center": _section_command_center, + "governance": _section_governance, "cluster_registry": _section_cluster_registry, } -DOMAIN_ORDER = ["docker", "databases", "lakehouse", "etl", "hadoop", "gpu", "objectscale", "command_center", "cluster_registry"] +DOMAIN_ORDER = ["docker", "databases", "lakehouse", "etl", "hadoop", "gpu", "objectscale", "command_center", "governance", "cluster_registry"] async def collect_full_lab_context( @@ -700,6 +772,13 @@ async def collect_full_lab_context( lakehouse = await collect_lakehouse(client, lake_raw, log) command_center = await collect_command_center(client, cc_raw, log) + try: + governance = collect_governance(log) + await _log(log, "ok", "fetch", "← Governance: CDC/movements/PII/lineage") + except Exception as exc: + await _log(log, "warn", "fetch", f"✗ Governance: {exc}") + governance = {"error": str(exc)} + await _log(log, "ok", "fetch", "═══ Lab snapshot complete ═══") return { "ts": datetime.now(timezone.utc).isoformat(), @@ -711,6 +790,7 @@ async def collect_full_lab_context( "gpu": gpu_data, "objectscale": objectscale, "command_center": command_center, + "governance": governance, }