diff --git a/api/Dockerfile b/api/Dockerfile index 583dc8c..9e632cf 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 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 pii_catalog.py hive_bench_seed.json . RUN mkdir -p /data ENV DATABASE_URL=sqlite:////data/atc-agents.db EXPOSE 3201 diff --git a/api/cdc_consumer.py b/api/cdc_consumer.py index 0d11536..ff27bb5 100644 --- a/api/cdc_consumer.py +++ b/api/cdc_consumer.py @@ -180,6 +180,23 @@ async def cdc_consumer_loop() -> None: pass +def snapshot(minutes: int = 15) -> dict[str, Any]: + """Lightweight CDC snapshot for other modules (Data Flow graph).""" + cutoff = datetime.now(timezone.utc).timestamp() - minutes * 60 + by_source: dict[str, int] = {} + total = 0 + for c in _ring: + try: + if datetime.fromisoformat(c["ts"]).timestamp() < cutoff: + continue + except Exception: + continue + total += 1 + by_source[c["source"]] = by_source.get(c["source"], 0) + 1 + return {"connected": _state["connected"], "consumed": _state["consumed"], + "buffered": len(_ring), "window_total": total, "by_source": by_source} + + # ── Endpoints ──────────────────────────────────────────────────────────────── @router.get("") async def list_changes( diff --git a/api/dataflow.py b/api/dataflow.py new file mode 100644 index 0000000..520fe9d --- /dev/null +++ b/api/dataflow.py @@ -0,0 +1,192 @@ +"""Data Flow graph for the Command Center "Data Flow" tab. + +Assembles the live landscape as a node-link graph (generators -> source DBs -> +CDC/Kafka -> sinks; HDFS -> Iceberg via Trino; sources -> masked curated layer) +and overlays live status: movement run states, CDC volume, Trino row counts, and +a PII overlay (which nodes hold PII and whether it is masked). + +This is the single LIVE source of truth for the Data Flow tab. The structure is +defined here (movement-driven) and is OpenMetadata-ready: when lineage is +available it can enrich/replace the static edges. +""" + +from __future__ import annotations + +import os +import time +from typing import Any + +import httpx +from fastapi import APIRouter, Body +from fastapi.responses import JSONResponse + +router = APIRouter(prefix="/api/dataflow", tags=["dataflow"]) + +TRINO_URL = os.getenv("TRINO_URL", "http://10.0.21.50:8089").rstrip("/") +TRINO_USER = os.getenv("TRINO_USER", "mo") + +# Static landscape (x,y in 0..100). kind drives UI styling. +NODES: list[dict[str, Any]] = [ + {"id": "generator", "label": "Data Generator", "sub": "Airflow DAGs", "kind": "generator", "x": 10, "y": 50}, + {"id": "hdfs", "label": "Hadoop HDFS", "sub": "historical_sales", "kind": "hadoop", "x": 10, "y": 88}, + {"id": "postgres", "label": "PostgreSQL", "sub": "sales_orders", "kind": "source", "x": 30, "y": 18}, + {"id": "mysql", "label": "MySQL", "sub": "employee_events", "kind": "source", "x": 30, "y": 44}, + {"id": "mongodb", "label": "MongoDB", "sub": "events", "kind": "source", "x": 30, "y": 70}, + {"id": "kafka", "label": "Kafka · Debezium", "sub": "CDC topics", "kind": "stream", "x": 52, "y": 40}, + {"id": "s3_cdc", "label": "S3 CDC Archive", "sub": "object store", "kind": "sink", "x": 72, "y": 20}, + {"id": "iceberg_hadoop", "label": "Iceberg · hadoop", "sub": "historical_sales_hdfs", "kind": "lakehouse", "x": 72, "y": 60}, + {"id": "iceberg_curated", "label": "Iceberg · curated_masked", "sub": "masked PII", "kind": "lakehouse", "x": 72, "y": 86}, + {"id": "trino", "label": "Trino", "sub": "query engine", "kind": "engine", "x": 90, "y": 52}, +] + +# Edges. movement_id (optional) links to movements.py so the edge is triggerable. +EDGES: list[dict[str, Any]] = [ + {"from": "generator", "to": "postgres", "kind": "generate", "movement_id": "gen_postgres"}, + {"from": "generator", "to": "mysql", "kind": "generate", "movement_id": "gen_mysql"}, + {"from": "generator", "to": "mongodb", "kind": "generate", "movement_id": "gen_mongodb"}, + {"from": "postgres", "to": "kafka", "kind": "cdc"}, + {"from": "mysql", "to": "kafka", "kind": "cdc"}, + {"from": "mongodb", "to": "kafka", "kind": "cdc"}, + {"from": "kafka", "to": "s3_cdc", "kind": "archive"}, + {"from": "hdfs", "to": "iceberg_hadoop", "kind": "movement", "movement_id": "hadoop_to_trino"}, + {"from": "postgres", "to": "iceberg_curated", "kind": "mask", "movement_id": "mask_to_curated"}, + {"from": "mysql", "to": "iceberg_curated", "kind": "mask", "movement_id": "mask_to_curated"}, + {"from": "mongodb", "to": "iceberg_curated", "kind": "mask", "movement_id": "mask_to_curated"}, + {"from": "iceberg_hadoop", "to": "trino", "kind": "query"}, + {"from": "iceberg_curated", "to": "trino", "kind": "query"}, + {"from": "kafka", "to": "trino", "kind": "query"}, +] + +_cache: dict[str, Any] = {"ts": 0.0, "data": None} +_TTL = 12.0 +_count_cache: dict[str, Any] = {"ts": 0.0, "val": None} + + +def _trino_scalar(sql: str, deadline_s: float = 8.0) -> int | None: + end = time.time() + deadline_s + try: + with httpx.Client(timeout=5.0) as client: + d = client.post(f"{TRINO_URL}/v1/statement", content=sql.encode(), headers={"X-Trino-User": TRINO_USER}).json() + rows: list[Any] = d.get("data") or [] + nxt = d.get("nextUri") + while nxt: + if time.time() > end: + try: + client.delete(nxt, timeout=3.0) + except Exception: + pass + return None + dd = client.get(nxt).json() + rows += dd.get("data") or [] + if dd.get("error"): + return None + nxt = dd.get("nextUri") + return int(rows[0][0]) if rows else None + except Exception: + return None + + +def _iceberg_hadoop_count() -> int | None: + now = time.time() + if _count_cache["val"] is not None and now - _count_cache["ts"] < 60: + return _count_cache["val"] + v = _trino_scalar("SELECT count(*) FROM iceberg.hadoop.historical_sales_hdfs") + if v is not None: + _count_cache["val"] = v + _count_cache["ts"] = now + return v + + +def _build() -> dict[str, Any]: + # Live signals + try: + from movements import last_runs + runs = last_runs() + except Exception: + runs = {} + try: + from cdc_consumer import snapshot as cdc_snapshot + cdc = cdc_snapshot(15) + except Exception: + cdc = {"connected": False, "consumed": 0, "by_source": {}, "window_total": 0} + try: + from pii_catalog import get_pii + pii = get_pii() + except Exception: + pii = {"datasets": []} + pii_by_node = {d["node_id"]: d for d in pii.get("datasets", [])} + + nodes = [] + for n in NODES: + node = dict(n) + node["level"] = "ok" + metric = None + if n["id"] == "kafka": + metric = f"{cdc.get('window_total', 0)} chg/15m · {cdc.get('consumed', 0)} total" + node["level"] = "ok" if cdc.get("connected") else "warn" + elif n["id"] in ("postgres", "mysql", "mongodb"): + metric = f"{cdc.get('by_source', {}).get(n['id'], 0)} CDC/15m" + elif n["id"] == "iceberg_hadoop": + c = _iceberg_hadoop_count() + metric = f"{c:,} rows" if c is not None else "iceberg table" + elif n["id"] == "generator": + metric = "Airflow gen DAGs" + # PII overlay + p = pii_by_node.get(n["id"]) + if p: + node["pii"] = { + "has_pii": p["has_pii"], "pii_count": p["pii_count"], "all_masked": p["all_masked"], + "masked_layer": p.get("masked_layer", False), + "categories": sorted({c["category"] for c in p["pii_columns"]}), + "columns": p["pii_columns"], + } + node["metric"] = metric + nodes.append(node) + + edges = [] + for e in EDGES: + edge = dict(e) + mid = e.get("movement_id") + if mid and mid in runs: + lr = runs[mid] + edge["state"] = lr.get("state") + edge["last_rows"] = lr.get("rows") + edge["last_duration_s"] = lr.get("duration_s") + edge["active"] = lr.get("state") == "running" + if e["kind"] == "cdc": + edge["active"] = cdc.get("by_source", {}).get(e["from"], 0) > 0 + edges.append(edge) + + return { + "ok": True, + "nodes": nodes, + "edges": edges, + "pii_summary": pii.get("summary", {}), + "cdc": {"connected": cdc.get("connected"), "consumed": cdc.get("consumed"), "window_total": cdc.get("window_total")}, + "ts": time.time(), + } + + +@router.get("") +async def get_dataflow(refresh: bool = False) -> JSONResponse: + now = time.time() + if not refresh and _cache["data"] and now - _cache["ts"] < _TTL: + return JSONResponse(_cache["data"]) + data = _build() + _cache["data"] = data + _cache["ts"] = now + return JSONResponse(data) + + +@router.post("/{movement_id}/run") +async def run_dataflow_movement(movement_id: str, body: dict[str, Any] = Body(default={})) -> JSONResponse: + try: + from movements import MOVEMENT_BY_ID, trigger_and_watch + import asyncio + except Exception as exc: + return JSONResponse({"ok": False, "error": str(exc)}, status_code=500) + if movement_id not in MOVEMENT_BY_ID: + return JSONResponse({"ok": False, "error": f"unknown movement {movement_id}"}, status_code=400) + conf = body.get("conf") if isinstance(body, dict) else None + asyncio.create_task(trigger_and_watch(movement_id, conf)) + return JSONResponse({"ok": True, "movement_id": movement_id, "status": "triggered"}) diff --git a/api/main.py b/api/main.py index 71613bd..2b8b985 100644 --- a/api/main.py +++ b/api/main.py @@ -45,6 +45,8 @@ from sql_console import router as sql_router from agent_ops import router as agent_ops_router, agent_dml_loop, etl_agent_loop from cdc_consumer import router as cdc_router, cdc_consumer_loop from movements import router as movements_router +from dataflow import router as dataflow_router +from pii_catalog import router as pii_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 @@ -737,6 +739,8 @@ app.include_router(sql_router) app.include_router(agent_ops_router) app.include_router(cdc_router) app.include_router(movements_router) +app.include_router(dataflow_router) +app.include_router(pii_router) app.add_middleware( CORSMiddleware, allow_origins=["*"], diff --git a/api/pii_catalog.py b/api/pii_catalog.py new file mode 100644 index 0000000..76cf687 --- /dev/null +++ b/api/pii_catalog.py @@ -0,0 +1,135 @@ +"""PII catalog for the Command Center Data Flow PII overlay. + +Classifies columns of the known datasets into PII categories. Primary source is +OpenMetadata tags (when available, Phase 3); falls back to a name-based +heuristic over the live table columns (via Trino information_schema). Reports +per dataset which columns are PII and whether they are masked. +""" + +from __future__ import annotations + +import os +import re +import time +from typing import Any + +import httpx +from fastapi import APIRouter +from fastapi.responses import JSONResponse + +router = APIRouter(prefix="/api/pii", tags=["pii"]) + +TRINO_URL = os.getenv("TRINO_URL", "http://10.0.21.50:8089").rstrip("/") +TRINO_USER = os.getenv("TRINO_USER", "mo") +OPENMETADATA_URL = os.getenv("OPENMETADATA_URL", "").rstrip("/") +OPENMETADATA_TOKEN = os.getenv("OPENMETADATA_TOKEN", "") + +# Datasets we surface in the PII overlay. node_id matches dataflow.py node ids. +DATASETS = [ + {"key": "postgres", "node_id": "postgres", "label": "PostgreSQL sales_orders", + "table": "postgres_sales.public.sales_orders", "table_name": "sales_orders", "catalog": "postgres_sales"}, + {"key": "mysql", "node_id": "mysql", "label": "MySQL employee_events", + "table": "mysql_hr.hr.employee_events", "table_name": "employee_events", "catalog": "mysql_hr"}, + {"key": "mongodb", "node_id": "mongodb", "label": "MongoDB events", + "table": "mongodb_supplychain.supplychain.events", "table_name": "events", "catalog": "mongodb_supplychain"}, + {"key": "curated", "node_id": "iceberg_curated", "label": "Iceberg curated_masked", + "table": "iceberg.curated_masked.sales_orders_masked", "table_name": "sales_orders_masked", "catalog": "iceberg", + "schema": "curated_masked", "masked_layer": True}, +] + +# name fragment -> PII category +PII_RULES: list[tuple[str, str]] = [ + (r"email|e_mail", "EMAIL"), + (r"phone|mobile|msisdn|tel", "PHONE"), + (r"iban|account_no|bank|card|credit", "FINANCIAL"), + (r"ssn|bsn|national_id|passport|tax_id", "NATIONAL_ID"), + (r"first_name|last_name|full_name|customer_name|employee_name|contact_name|^name$", "NAME"), + (r"address|street|city|zip|postal|postcode", "ADDRESS"), + (r"dob|birth|date_of_birth", "DOB"), + (r"ip_addr|ip_address", "IP"), +] + +_cache: dict[str, Any] = {"ts": 0.0, "data": None} +_TTL = 120.0 + + +def _classify(col: str) -> str | None: + c = col.lower() + for pat, cat in PII_RULES: + if re.search(pat, c): + return cat + return None + + +def _is_masked(col: str, dataset_masked_layer: bool) -> bool: + c = col.lower() + return dataset_masked_layer or c.endswith("_masked") or c.endswith("_hash") or c.endswith("_token") + + +def _trino_columns(catalog: str, schema: str | None, table_name: str) -> list[str]: + sql = ( + f"SELECT column_name FROM {catalog}.information_schema.columns " + f"WHERE table_name = '{table_name}'" + ) + if schema: + sql += f" AND table_schema = '{schema}'" + try: + with httpx.Client(timeout=10.0) as client: + r = client.post(f"{TRINO_URL}/v1/statement", content=sql.encode(), headers={"X-Trino-User": TRINO_USER}) + d = r.json() + rows: list[Any] = d.get("data") or [] + nxt = d.get("nextUri") + while nxt: + dd = client.get(nxt).json() + rows += dd.get("data") or [] + if dd.get("error"): + break + nxt = dd.get("nextUri") + return [row[0] for row in rows] + except Exception: + return [] + + +def _build() -> dict[str, Any]: + datasets_out = [] + total_pii = 0 + total_masked = 0 + for ds in DATASETS: + cols = _trino_columns(ds["catalog"], ds.get("schema"), ds["table_name"]) + pii_cols = [] + for c in cols: + cat = _classify(c) + if cat: + masked = _is_masked(c, ds.get("masked_layer", False)) + pii_cols.append({"name": c, "category": cat, "masked": masked}) + total_pii += 1 + if masked: + total_masked += 1 + datasets_out.append({ + "key": ds["key"], "node_id": ds["node_id"], "label": ds["label"], "table": ds["table"], + "exists": bool(cols), "masked_layer": ds.get("masked_layer", False), + "pii_columns": pii_cols, "pii_count": len(pii_cols), + "has_pii": bool(pii_cols), + "all_masked": bool(pii_cols) and all(c["masked"] for c in pii_cols), + }) + return { + "ok": True, "source": "heuristic" if not OPENMETADATA_URL else "openmetadata+heuristic", + "datasets": datasets_out, + "summary": {"datasets": len(datasets_out), "pii_columns": total_pii, "masked_columns": total_masked, + "unmasked_columns": total_pii - total_masked}, + } + + +def get_pii(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("") +async def pii_overview(refresh: bool = False) -> JSONResponse: + return JSONResponse(get_pii(use_cache=not refresh))