From b54f5c7a06b3f3cf9585c25455f0c580521c49db Mon Sep 17 00:00:00 2001 From: mo Date: Wed, 22 Jul 2026 00:31:35 +0000 Subject: [PATCH] feat: Nessie catalog UI + Iceberg structure explorer + live Data Flow online/offline Add Project Nessie on lake01 with Command Center Iceberg tab (snapshots, manifests, data files, time-travel SQL). Data Flow pulses only when endpoints are reachable; offline nodes/edges render red. --- api/Dockerfile | 2 +- api/dataflow.py | 102 ++++- api/main.py | 2 + api/nessie_iceberg.py | 378 ++++++++++++++++++ config/command-center/atc.env.example | 3 + ui/src/components/features/ChangesView.tsx | 17 +- .../components/features/DataExplorerView.tsx | 6 +- ui/src/components/features/DataFlowView.tsx | 26 +- .../components/features/IcebergNessieView.tsx | 302 ++++++++++++++ 9 files changed, 815 insertions(+), 23 deletions(-) create mode 100644 api/nessie_iceberg.py create mode 100644 ui/src/components/features/IcebergNessieView.tsx diff --git a/api/Dockerfile b/api/Dockerfile index 8d831a8..035528b 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 auth.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 gpu_config.py hive_bench_seed.json . +COPY main.py auth.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 gpu_config.py nessie_iceberg.py hive_bench_seed.json . RUN mkdir -p /data ENV DATABASE_URL=sqlite:////data/atc-agents.db EXPOSE 3201 diff --git a/api/dataflow.py b/api/dataflow.py index 3092011..36ce8e9 100644 --- a/api/dataflow.py +++ b/api/dataflow.py @@ -26,6 +26,76 @@ 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") +NESSIE_URL = os.getenv("NESSIE_URL", "http://10.0.21.50:19120").rstrip("/") +SRC_DB_HOST = os.getenv("SRC_DB_HOST", "10.0.21.51") +SPARK_UI_URL = os.getenv("SPARK_UI_URL", "http://10.0.21.50:8081").rstrip("/") +KAFKA_UI_URL = os.getenv("KAFKA_UI_URL", "http://10.0.21.36:9000").rstrip("/") +HDFS_NN_URL = os.getenv("HDFS_NN_URL", "http://10.0.21.61:9870").rstrip("/") +S3_ENDPOINT = os.getenv("S3_ENDPOINT", "http://10.0.20.111:9020").rstrip("/") + +# TCP ports for source DB reachability on the shared db host +_SOURCE_PORTS = { + "postgres": 5432, + "mysql": 3306, + "mongodb": 27017, + "cassandra": 9042, + "neo4j": 7687, +} + + +async def _tcp_open(host: str, port: int, timeout: float = 1.2) -> bool: + import asyncio + try: + conn = asyncio.open_connection(host, port) + reader, writer = await asyncio.wait_for(conn, timeout=timeout) + writer.close() + try: + await writer.wait_closed() + except Exception: + pass + return True + except Exception: + return False + + +async def _http_ok(url: str, timeout: float = 2.0) -> bool: + try: + async with httpx.AsyncClient(timeout=timeout) as client: + r = await client.get(url) + return r.status_code < 500 + except Exception: + return False + + +async def _probe_online() -> dict[str, bool]: + """Best-effort live reachability for dataflow nodes.""" + online: dict[str, bool] = {} + for sid, port in _SOURCE_PORTS.items(): + online[sid] = await _tcp_open(SRC_DB_HOST, port) + online["trino"] = await _http_ok(f"{TRINO_URL}/v1/info") + online["nessie"] = await _http_ok(f"{NESSIE_URL}/api/v2/config") + online["spark"] = await _http_ok(f"{SPARK_UI_URL}/json/") + online["kafka"] = await _http_ok(KAFKA_UI_URL) + online["hdfs"] = await _http_ok(HDFS_NN_URL) + # S3: TCP to endpoint host:port + try: + from urllib.parse import urlparse + u = urlparse(S3_ENDPOINT) + online["s3_cdc"] = await _tcp_open(u.hostname or "10.0.20.111", u.port or 9020) + except Exception: + online["s3_cdc"] = False + # lakehouse iceberg nodes follow trino catalog availability + online["iceberg_hadoop"] = online.get("trino", False) + online["iceberg_curated"] = online.get("trino", False) + online["generator"] = True # logical + online["openmetadata"] = await _http_ok("http://10.0.21.47:8585") + online["chromadb"] = True + online["rag"] = True + online["vllm"] = await _http_ok(os.getenv("LLM_URL", "http://10.0.10.106:8001/v1").rstrip("/") + "/models") + online["chat"] = True + return online + + # Static landscape (x,y in 0..100). kind drives UI styling. # Laid out as clean left→right pipeline stages so lineage reads in order: # producers (col 0) → sources (col 1) → CDC (col 2) → storage/lakehouse (col 3) @@ -48,6 +118,8 @@ NODES: list[dict[str, Any]] = [ {"id": "s3_cdc", "label": "S3 CDC Archive", "sub": "object store", "kind": "sink", "x": 67, "y": 13}, {"id": "iceberg_curated", "label": "Iceberg · curated_masked", "sub": "masked PII", "kind": "lakehouse", "x": 67, "y": 45}, {"id": "iceberg_hadoop", "label": "Iceberg · hadoop", "sub": "historical_sales_hdfs", "kind": "lakehouse", "x": 67, "y": 80}, + {"id": "nessie", "label": "Nessie", "sub": "Iceberg catalog · Git-like", "kind": "catalog", "x": 78, "y": 62, + "url": "http://10.0.21.50:19120"}, # col 4 — query engine {"id": "trino", "label": "Trino", "sub": "query engine", "kind": "engine", "x": 88, "y": 45}, # governance — bottom centre @@ -83,6 +155,10 @@ EDGES: list[dict[str, Any]] = [ {"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": "nessie", "kind": "catalog"}, + {"from": "iceberg_curated", "to": "nessie", "kind": "catalog"}, + {"from": "nessie", "to": "s3_cdc", "kind": "archive"}, + {"from": "nessie", "to": "trino", "kind": "query"}, {"from": "iceberg_hadoop", "to": "trino", "kind": "query"}, {"from": "iceberg_curated", "to": "trino", "kind": "query"}, {"from": "kafka", "to": "trino", "kind": "query"}, @@ -198,10 +274,14 @@ async def _build() -> dict[str, Any]: edge_live = streaming.get("edges") or {} pii_by_node = {d["node_id"]: d for d in pii.get("datasets", [])} + online = await _probe_online() + nodes = [] for n in NODES: node = dict(n) - node["level"] = "ok" + is_up = online.get(n["id"], True) + node["online"] = is_up + node["level"] = "ok" if is_up else "err" metric = None if n["id"] == "openmetadata": metric = f"{pii.get('summary', {}).get('pii_columns', 0)} PII cols cataloged" @@ -217,10 +297,15 @@ async def _build() -> dict[str, Any]: metric = f"{spark.get('alive_workers', 0)} workers · {used}/{cores} cores · {apps} apps" node["level"] = "ok" if spark.get("ui_ok") and (spark.get("status") or "").upper() == "ALIVE" else "warn" elif n["id"] in ("postgres", "mysql", "mongodb", "cassandra", "neo4j"): - metric = f"{cdc.get('by_source', {}).get(n['id'], 0)} CDC/15m" + if not is_up: + metric = "offline" + else: + 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"] == "nessie": + metric = "catalog commits" if online.get("nessie") else "offline" elif n["id"] == "generator": metric = "Airflow gen DAGs" elif n["id"] in ("vllm", "rag", "chromadb"): @@ -283,6 +368,10 @@ async def _build() -> dict[str, Any]: elif e["kind"] == "archive" and e.get("from") == "kafka" and e.get("to") == "s3_cdc": # Kafka → S3 CDC archive pulses while the pipeline lands objects in S3 edge["active"] = arch_active + elif e["kind"] == "catalog" and e.get("to") == "nessie": + edge["active"] = bool(online.get("nessie")) and bool(online.get(e.get("from"), True)) + elif e["kind"] == "catalog": + edge["active"] = bool(online.get(e.get("from"), True)) and bool(online.get(e.get("to"), True)) elif e["kind"] in ("context", "retrieve", "prompt", "answer"): # AI serving lane pulses while governed data is being served to the LLM edge["active"] = bool(_rag_info()) @@ -294,6 +383,15 @@ async def _build() -> dict[str, Any]: pass if _flow != "running": edge["active"] = False + + # Realtime online/offline: never pulse if an endpoint is down + frm_up = online.get(e.get("from"), True) + to_up = online.get(e.get("to"), True) + if not (frm_up and to_up): + edge["active"] = False + edge["offline"] = True + else: + edge["offline"] = False edges.append(edge) return { diff --git a/api/main.py b/api/main.py index 0c91a15..90cea59 100644 --- a/api/main.py +++ b/api/main.py @@ -62,6 +62,7 @@ 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 nessie_iceberg import router as nessie_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 @@ -1044,6 +1045,7 @@ app.include_router(lineage_router) app.include_router(dq_router) app.include_router(observability_router) app.include_router(governance_router) +app.include_router(nessie_router) app.add_middleware( CORSMiddleware, allow_origins=["*"], diff --git a/api/nessie_iceberg.py b/api/nessie_iceberg.py new file mode 100644 index 0000000..e2e3dd5 --- /dev/null +++ b/api/nessie_iceberg.py @@ -0,0 +1,378 @@ +"""Nessie catalog + Iceberg table structure for Command Center. + +Nessie (http://lake01:19120) holds Git-like catalog commits. Trino's existing +`iceberg` catalog exposes $snapshots / $manifests / $files for structure. +Trino 405 cannot use iceberg.catalog.type=nessie, so we combine both. +""" + +from __future__ import annotations + +import os +import time +from typing import Annotated, Any +from urllib.parse import quote + +import httpx +from fastapi import APIRouter, Body, Query +from fastapi.responses import JSONResponse + +router = APIRouter(prefix="/api/nessie", tags=["nessie"]) + +NESSIE_URL = os.getenv("NESSIE_URL", "http://10.0.21.50:19120").rstrip("/") +NESSIE_UI_URL = os.getenv("NESSIE_UI_URL", NESSIE_URL) +TRINO_URL = os.getenv("TRINO_URL", "http://10.0.21.50:8089").rstrip("/") +TRINO_USER = os.getenv("TRINO_USER", "mo") +ICEBERG_CATALOG = os.getenv("ICEBERG_CATALOG", "iceberg") +S3_ENDPOINT = os.getenv("S3_ENDPOINT", "http://10.0.20.111:9020") +S3_ACCESS_KEY = os.getenv("S3_ACCESS_KEY", "") +S3_SECRET_KEY = os.getenv("S3_SECRET_KEY", "") +S3_BUCKET = os.getenv("S3_ARCHIVE_BUCKET", "data") + +_cache: dict[str, Any] = {"ts": 0.0, "health": None} +_TTL = 8.0 + + +async def _nessie(client: httpx.AsyncClient, path: str) -> Any: + r = await client.get(f"{NESSIE_URL}{path}", timeout=12.0) + r.raise_for_status() + return r.json() + + +async def _trino_rows(sql: str, deadline_s: float = 45.0) -> list[list[Any]]: + end = time.time() + deadline_s + rows: list[list[Any]] = [] + async with httpx.AsyncClient(timeout=20.0) as client: + r = await client.post( + f"{TRINO_URL}/v1/statement", + content=sql.encode(), + headers={"X-Trino-User": TRINO_USER}, + ) + data = r.json() + rows.extend(data.get("data") or []) + nxt = data.get("nextUri") + while nxt: + if time.time() > end: + try: + await client.delete(nxt, timeout=3.0) + except Exception: + pass + break + rr = await client.get(nxt) + data = rr.json() + rows.extend(data.get("data") or []) + nxt = data.get("nextUri") + if data.get("stats", {}).get("state") in ("FAILED", "FINISHED"): + if data.get("error"): + raise RuntimeError(str(data["error"].get("message") or data["error"])) + if not nxt: + break + return rows + + +def _fqn(schema: str, table: str) -> str: + return f'{ICEBERG_CATALOG}.{schema}."{table}"' + + +def _meta_table(schema: str, table: str, kind: str) -> str: + # Trino Iceberg metadata tables: schema."table$snapshots" + return f'{ICEBERG_CATALOG}.{schema}."{table}${kind}"' + + +@router.get("/health") +async def health() -> JSONResponse: + now = time.time() + if _cache["health"] and now - _cache["ts"] < _TTL: + return JSONResponse(_cache["health"]) + out: dict[str, Any] = { + "ok": False, + "nessie_url": NESSIE_URL, + "nessie_ui": NESSIE_UI_URL, + "trino_url": TRINO_URL, + "iceberg_catalog": ICEBERG_CATALOG, + "nessie_ok": False, + "trino_ok": False, + "default_branch": None, + "note": "Trino 405 has no native Nessie catalog; structure comes from iceberg.* metadata tables.", + } + try: + async with httpx.AsyncClient(timeout=8.0) as client: + cfg = await _nessie(client, "/api/v2/config") + out["nessie_ok"] = True + out["default_branch"] = cfg.get("defaultBranch", "main") + info = await client.get(f"{TRINO_URL}/v1/info") + out["trino_ok"] = info.status_code == 200 + except Exception as exc: + out["error"] = str(exc)[:200] + out["ok"] = bool(out["nessie_ok"]) + _cache["health"] = out + _cache["ts"] = now + return JSONResponse(out) + + +@router.get("/trees") +async def trees() -> JSONResponse: + try: + async with httpx.AsyncClient(timeout=12.0) as client: + data = await _nessie(client, "/api/v2/trees") + refs = [] + for item in data.get("references") or data.get("tokens") or []: + # v2 may return {references: [...]} or paginated + ref = item.get("reference") if isinstance(item, dict) and "reference" in item else item + if not isinstance(ref, dict): + continue + refs.append({ + "type": ref.get("type"), + "name": ref.get("name"), + "hash": ref.get("hash"), + }) + if not refs: + # fallback: main only + async with httpx.AsyncClient(timeout=12.0) as client: + main = await _nessie(client, "/api/v2/trees/main") + ref = main.get("reference") or main + refs = [{"type": ref.get("type"), "name": ref.get("name"), "hash": ref.get("hash")}] + return JSONResponse({"ok": True, "references": refs, "ui": NESSIE_UI_URL}) + except Exception as exc: + return JSONResponse({"ok": False, "error": str(exc)[:240], "references": []}) + + +@router.get("/history") +async def history(ref: Annotated[str, Query()] = "main", max_records: Annotated[int, Query(ge=1, le=200)] = 50) -> JSONResponse: + try: + async with httpx.AsyncClient(timeout=15.0) as client: + data = await _nessie(client, f"/api/v2/trees/{quote(str(ref), safe='')}/history?maxRecords={max_records}") + entries = data.get("logEntries") or data.get("commits") or [] + out = [] + for e in entries: + meta = e.get("commitMeta") or e.get("commit") or {} + out.append({ + "hash": e.get("hash") or e.get("commitHash") or meta.get("hash"), + "message": meta.get("message"), + "author": (meta.get("authors") or [meta.get("author")])[0] if (meta.get("authors") or meta.get("author")) else None, + "commitTime": meta.get("commitTime") or meta.get("authorTime"), + }) + return JSONResponse({"ok": True, "ref": ref, "commits": out}) + except Exception as exc: + return JSONResponse({"ok": False, "error": str(exc)[:240], "commits": []}) + + +@router.get("/contents") +async def contents(ref: Annotated[str, Query()] = "main") -> JSONResponse: + try: + async with httpx.AsyncClient(timeout=15.0) as client: + data = await _nessie(client, f"/api/v2/trees/{quote(str(ref), safe='')}/entries") + tables = [] + namespaces = [] + for e in data.get("entries") or []: + name = e.get("name") or {} + elements = name.get("elements") or [] + typ = e.get("type") + if typ == "NAMESPACE": + namespaces.append(".".join(elements)) + elif typ == "ICEBERG_TABLE" and len(elements) >= 2: + tables.append({ + "schema": elements[0], + "table": elements[1], + "key": ".".join(elements), + "contentId": e.get("contentId"), + "trino_fqn": _fqn(elements[0], elements[1]), + }) + return JSONResponse({ + "ok": True, + "ref": ref, + "namespaces": namespaces, + "tables": tables, + "ui": NESSIE_UI_URL, + }) + except Exception as exc: + return JSONResponse({"ok": False, "error": str(exc)[:240], "tables": []}) + + +@router.get("/tables/{schema}/{table}/structure") +async def table_structure( + schema: str, + table: str, + snapshot_id: Annotated[str | None, Query()] = None, +) -> JSONResponse: + """Iceberg structure via Trino metadata tables (works with file metastore).""" + result: dict[str, Any] = { + "ok": False, + "schema": schema, + "table": table, + "fqn": _fqn(schema, table), + "snapshots": [], + "manifests": [], + "files": [], + "selected_snapshot": None, + "as_of_sql": None, + } + try: + snap_sql = ( + f'SELECT snapshot_id, committed_at, operation, summary ' + f'FROM {_meta_table(schema, table, "snapshots")} ' + f'ORDER BY committed_at DESC LIMIT 40' + ) + snaps = await _trino_rows(snap_sql) + snapshots = [] + for row in snaps: + snapshots.append({ + "snapshot_id": str(row[0]), + "committed_at": str(row[1]) if row[1] is not None else None, + "operation": row[2], + "summary": (str(row[3]) if row[3] is not None else None) if len(row) > 3 else None, + }) + result["snapshots"] = snapshots + current = snapshot_id or (snapshots[0]["snapshot_id"] if snapshots else None) + result["selected_snapshot"] = current + if current: + result["as_of_sql"] = ( + f'SELECT * FROM {_fqn(schema, table)} FOR VERSION AS OF {current} LIMIT 100' + ) + + # manifests + try: + man_sql = ( + f'SELECT path, length, partition_spec_id, added_snapshot_id, ' + f'added_data_files_count, existing_data_files_count, deleted_data_files_count ' + f'FROM {_meta_table(schema, table, "manifests")} LIMIT 200' + ) + mans = await _trino_rows(man_sql) + result["manifests"] = [ + { + "path": r[0], + "length": r[1], + "partition_spec_id": r[2], + "added_snapshot_id": str(r[3]) if r[3] is not None else None, + "added_data_files_count": r[4], + "existing_data_files_count": r[5], + "deleted_data_files_count": r[6], + } + for r in mans + ] + except Exception as exc: + result["manifests_error"] = str(exc)[:200] + + # data files (current snapshot view) + try: + files_sql = ( + f'SELECT file_path, file_format, record_count, file_size_in_bytes ' + f'FROM {_meta_table(schema, table, "files")} LIMIT 500' + ) + files = await _trino_rows(files_sql) + result["files"] = [ + { + "path": r[0], + "format": r[1], + "record_count": r[2], + "size_bytes": r[3], + "is_s3": str(r[0]).startswith("s3") if r[0] else False, + } + for r in files + ] + except Exception as exc: + result["files_error"] = str(exc)[:200] + + result["ok"] = True + result["restore_warning"] = ( + "Restore re-points table metadata to an older snapshot. " + "It cannot undelete files already removed from S3/disk." + ) + return JSONResponse(result) + except Exception as exc: + result["error"] = str(exc)[:300] + return JSONResponse(result) + + +@router.post("/tables/{schema}/{table}/restore") +async def restore_snapshot( + schema: str, + table: str, + body: dict[str, Any] = Body(default={}), +) -> JSONResponse: + """Rollback Iceberg table to a previous snapshot via Trino CALL (if supported) or documented SQL. + + Trino 405 may not expose system.rollback; we attempt CALL and fall back to guidance. + """ + snapshot_id = str((body or {}).get("snapshot_id") or "").strip() + if not snapshot_id: + return JSONResponse({"ok": False, "error": "snapshot_id required"}, status_code=400) + fqn = _fqn(schema, table) + # Prefer Iceberg procedure when available + attempts = [ + f"CALL {ICEBERG_CATALOG}.system.rollback_to_snapshot('{schema}', '{table}', {snapshot_id})", + f"ALTER TABLE {fqn} EXECUTE rollback_to_snapshot({snapshot_id})", + ] + errors: list[str] = [] + for sql in attempts: + try: + await _trino_rows(sql, deadline_s=60.0) + return JSONResponse({ + "ok": True, + "schema": schema, + "table": table, + "snapshot_id": snapshot_id, + "method": sql, + "warning": "Old data files must still exist on storage for this to succeed.", + }) + except Exception as exc: + errors.append(f"{sql}: {exc}") + return JSONResponse({ + "ok": False, + "error": "Rollback procedure not available on this Trino version", + "hint": f'Query historically with: SELECT * FROM {fqn} FOR VERSION AS OF {snapshot_id}', + "as_of_sql": f'SELECT * FROM {fqn} FOR VERSION AS OF {snapshot_id} LIMIT 100', + "details": errors[:3], + }) + + +@router.get("/s3/parquet") +async def list_s3_parquet(prefix: Annotated[str, Query()] = "lake/,iceberg/", limit: Annotated[int, Query(ge=1, le=1000)] = 200) -> JSONResponse: + """List parquet (and orc) objects under lake/ and iceberg/ prefixes.""" + if not S3_ACCESS_KEY or not S3_SECRET_KEY: + return JSONResponse({"ok": False, "error": "S3 credentials not configured", "objects": []}) + try: + import boto3 + from botocore.client import Config + s3 = boto3.client( + "s3", + endpoint_url=S3_ENDPOINT, + aws_access_key_id=S3_ACCESS_KEY, + aws_secret_access_key=S3_SECRET_KEY, + region_name=os.getenv("S3_REGION", "us-east-1"), + config=Config(s3={"addressing_style": "path"}, signature_version="s3v4"), + ) + prefixes = [p.strip() for p in prefix.split(",") if p.strip()] + objects: list[dict[str, Any]] = [] + for pref in prefixes: + token = None + while len(objects) < limit: + kw: dict[str, Any] = {"Bucket": S3_BUCKET, "Prefix": pref, "MaxKeys": min(500, limit - len(objects))} + if token: + kw["ContinuationToken"] = token + resp = s3.list_objects_v2(**kw) + for o in resp.get("Contents") or []: + key = o["Key"] + low = key.lower() + if not (low.endswith(".parquet") or low.endswith(".orc") or "/data/" in low): + continue + objects.append({ + "bucket": S3_BUCKET, + "key": key, + "size": o.get("Size"), + "last_modified": o["LastModified"].isoformat() if o.get("LastModified") else None, + "format": "parquet" if low.endswith(".parquet") else ("orc" if low.endswith(".orc") else "data"), + }) + if len(objects) >= limit: + break + if not resp.get("IsTruncated"): + break + token = resp.get("NextContinuationToken") + return JSONResponse({ + "ok": True, + "bucket": S3_BUCKET, + "prefixes": prefixes, + "objects": objects, + "note": "Raw S3 objects — not all are Iceberg-managed; lake/ ETL parquet has no snapshot history until registered as Iceberg.", + }) + except Exception as exc: + return JSONResponse({"ok": False, "error": str(exc)[:240], "objects": []}) diff --git a/config/command-center/atc.env.example b/config/command-center/atc.env.example index a4a54e5..5077572 100644 --- a/config/command-center/atc.env.example +++ b/config/command-center/atc.env.example @@ -19,3 +19,6 @@ KAFKA_UI_URL=http://10.0.21.36:9000 # Dockhand API token (Profile → API tokens in Dockhand UI; prefix dh_) DOCKHAND_API_TOKEN= + +NESSIE_URL=http://10.0.21.50:19120 +NESSIE_UI_URL=http://10.0.21.50:19120 diff --git a/ui/src/components/features/ChangesView.tsx b/ui/src/components/features/ChangesView.tsx index 9554a26..c90f694 100644 --- a/ui/src/components/features/ChangesView.tsx +++ b/ui/src/components/features/ChangesView.tsx @@ -232,6 +232,7 @@ export function ChangesView({ liveChanges }: { liveChanges: CdcChange[] }) { const [flash, setFlash] = useState(false) const [resyncing, setResyncing] = useState(false) const [resyncMsg, setResyncMsg] = useState(null) + const [windowMin, setWindowMin] = useState(15) // Live overlay: CDC events counted straight off the WebSocket stream since the // last server stats snapshot. The top KPIs/charts = authoritative server stats // (refreshed every 2.5s) + this overlay, so they move in lock-step with the @@ -248,11 +249,11 @@ export function ChangesView({ liveChanges }: { liveChanges: CdcChange[] }) { }, []) const load = useCallback(async () => { - const [c, s] = await Promise.all([fetchChanges({ limit: 200 }), fetchChangeStats(15)]) + const [c, s] = await Promise.all([fetchChanges({ limit: 200, minutes: windowMin }), fetchChangeStats(windowMin)]) setSeed(c.changes) setConnected(c.connected) applyStats(s) - }, [applyStats]) + }, [applyStats, windowMin]) const doResync = useCallback(async () => { setResyncing(true) @@ -275,9 +276,9 @@ export function ChangesView({ liveChanges }: { liveChanges: CdcChange[] }) { useEffect(() => { load() - const iv = setInterval(() => fetchChangeStats(15).then((s) => applyStats(s)), 2500) + const iv = setInterval(() => fetchChangeStats(windowMin).then((s) => applyStats(s)), 2500) return () => clearInterval(iv) - }, [load, applyStats]) + }, [load, applyStats, windowMin]) // Fold freshly-arrived WS changes into the overlay → instant top-of-page update. useEffect(() => { @@ -410,9 +411,9 @@ export function ChangesView({ liveChanges }: { liveChanges: CdcChange[] }) { {/* KPI row */}
- - - + + +
@@ -420,7 +421,7 @@ export function ChangesView({ liveChanges }: { liveChanges: CdcChange[] }) {
- Change volume — last 15 minutes + Change volume — selected window
diff --git a/ui/src/components/features/DataExplorerView.tsx b/ui/src/components/features/DataExplorerView.tsx index ea644fc..d15cde2 100644 --- a/ui/src/components/features/DataExplorerView.tsx +++ b/ui/src/components/features/DataExplorerView.tsx @@ -18,6 +18,7 @@ import { Radio, GitBranch, Lock, + Layers, } from 'lucide-react' import { cn } from '../../lib/utils' import { TrinoFederationView, type SubTab } from './TrinoFederationView' @@ -26,14 +27,16 @@ import { LineageView } from './LineageView' import { GovernanceOwnershipView } from './GovernanceOwnershipView' import { GovernanceAccessView } from './GovernanceAccessView' import { ObservabilityView } from './ObservabilityView' +import { IcebergNessieView } from './IcebergNessieView' -type ExplorerTab = 'business' | 'live' | SubTab | 'lineage' | 'ownership' | 'access' | 'observability' +type ExplorerTab = 'business' | 'live' | SubTab | 'lineage' | 'ownership' | 'access' | 'observability' | 'iceberg' 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' }, { id: 'live', label: 'Live', icon: Radio, hint: 'Realtime business activity — live counters, ingestion throughput & region matrix', live: true }, { 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: 'iceberg', label: 'Iceberg', icon: Layers, hint: 'Nessie catalog · snapshots · manifests · data files · time travel' }, { 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' }, @@ -310,6 +313,7 @@ export function DataExplorerView() { {view === 'ownership' &&
} {view === 'access' &&
} {view === 'observability' &&
} + {view === 'iceberg' &&
} {/* ───────── BUSINESS OVERVIEW ───────── */} {view === 'business' && ( diff --git a/ui/src/components/features/DataFlowView.tsx b/ui/src/components/features/DataFlowView.tsx index 420395d..121ebdd 100644 --- a/ui/src/components/features/DataFlowView.tsx +++ b/ui/src/components/features/DataFlowView.tsx @@ -22,6 +22,7 @@ const NODE_KIND: Record = { rag: { ring: 'border-pink-400/60', chip: 'bg-pink-500/15 text-pink-300 border-pink-400/40', dot: '#f472b6' }, llm: { ring: 'border-rose-400/70', chip: 'bg-rose-500/15 text-rose-200 border-rose-400/50', dot: '#fb7185' }, chat: { ring: 'border-indigo-400/60', chip: 'bg-indigo-500/15 text-indigo-300 border-indigo-400/40', dot: '#818cf8' }, + catalog: { ring: 'border-lime-400/60', chip: 'bg-lime-500/15 text-lime-300 border-lime-400/40', dot: '#a3e635' }, } const EDGE_COLOR: Record = { @@ -425,9 +426,10 @@ export function DataFlowView() { const b = anchors[edge.to] if (!a || !b) return null const d = edgePath(a, b) - const color = EDGE_COLOR[edge.kind] || '#64748b' - const running = edge.state === 'running' - const active = edge.active || running + const offline = Boolean((edge as any).offline) + const color = offline ? '#f43f5e' : (EDGE_COLOR[edge.kind] || '#64748b') + const running = edge.state === 'running' && !offline + const active = Boolean(edge.active) && !offline const dur = 1.6 + (i % 5) * 0.3 return ( @@ -435,9 +437,9 @@ export function DataFlowView() { d={d} fill="none" stroke={color} - strokeWidth={running ? 2.6 : 1.4} - strokeOpacity={active ? 0.85 : 0.28} - strokeDasharray={edge.movement_id && !active ? '4 4' : undefined} + strokeWidth={running ? 2.6 : offline ? 1.8 : 1.4} + strokeOpacity={active ? 0.85 : offline ? 0.75 : 0.28} + strokeDasharray={offline || (edge.movement_id && !active) ? '4 4' : undefined} /> {active && ( <> @@ -629,6 +631,7 @@ function NodeCard({ const kind = NODE_KIND[node.kind] || NODE_KIND.source const pii = node.pii const showPii = piiOverlay && pii?.has_pii + const offline = (node as any).online === false || node.level === 'err' return (
{node.sub} {node.metric && ( - - {node.metric} + + {offline ? `offline · ${node.metric}` : node.metric} )} {showPii && ( diff --git a/ui/src/components/features/IcebergNessieView.tsx b/ui/src/components/features/IcebergNessieView.tsx new file mode 100644 index 0000000..6fb05b1 --- /dev/null +++ b/ui/src/components/features/IcebergNessieView.tsx @@ -0,0 +1,302 @@ +import { useCallback, useEffect, useMemo, useState } from 'react' +import { + ExternalLink, + GitBranch, + Layers, + Loader2, + RefreshCw, + RotateCcw, + FileBox, + HardDrive, + AlertTriangle, +} from 'lucide-react' +import { cn } from '../../lib/utils' + +type NessieTable = { schema: string; table: string; key: string; trino_fqn: string } +type Snapshot = { snapshot_id: string; committed_at?: string | null; operation?: string | null } +type Manifest = { + path: string + length?: number + added_snapshot_id?: string | null + added_data_files_count?: number + existing_data_files_count?: number +} +type DataFile = { path: string; format?: string; record_count?: number; size_bytes?: number; is_s3?: boolean } +type Commit = { hash?: string; message?: string; author?: string; commitTime?: string } +type S3Obj = { key: string; size?: number; last_modified?: string | null; format?: string } + +function fmtBytes(n?: number | null) { + if (n == null) return '—' + if (n >= 1e9) return `${(n / 1e9).toFixed(2)} GB` + if (n >= 1e6) return `${(n / 1e6).toFixed(1)} MB` + if (n >= 1e3) return `${(n / 1e3).toFixed(1)} KB` + return `${n} B` +} + +export function IcebergNessieView() { + const [loading, setLoading] = useState(true) + const [health, setHealth] = useState(null) + const [tables, setTables] = useState([]) + const [commits, setCommits] = useState([]) + const [sel, setSel] = useState(null) + const [structure, setStructure] = useState(null) + const [snapId, setSnapId] = useState(null) + const [s3objs, setS3objs] = useState([]) + const [err, setErr] = useState(null) + const [busy, setBusy] = useState(false) + const [toast, setToast] = useState(null) + + const load = useCallback(async () => { + setLoading(true) + setErr(null) + try { + const [h, c, hist, s3] = await Promise.all([ + fetch('/api/nessie/health').then((r) => r.json()), + fetch('/api/nessie/contents').then((r) => r.json()), + fetch('/api/nessie/history?max_records=30').then((r) => r.json()), + fetch('/api/nessie/s3/parquet?limit=80').then((r) => r.json()), + ]) + setHealth(h) + setTables(c.tables || []) + setCommits(hist.commits || []) + setS3objs(s3.objects || []) + if (!sel && (c.tables || []).length) setSel(c.tables[0]) + } catch (e: any) { + setErr(String(e?.message || e)) + } finally { + setLoading(false) + } + }, [sel]) + + const loadStructure = useCallback(async (t: NessieTable, snapshot?: string | null) => { + setBusy(true) + try { + const q = snapshot ? `?snapshot_id=${encodeURIComponent(snapshot)}` : '' + const data = await fetch(`/api/nessie/tables/${t.schema}/${t.table}/structure${q}`).then((r) => r.json()) + setStructure(data) + setSnapId(data.selected_snapshot || null) + } catch (e: any) { + setStructure({ ok: false, error: String(e?.message || e) }) + } finally { + setBusy(false) + } + }, []) + + useEffect(() => { + void load() + }, []) + + useEffect(() => { + if (sel) void loadStructure(sel, null) + }, [sel?.key]) + + const restore = async () => { + if (!sel || !snapId) return + if (!confirm(`Restore ${sel.schema}.${sel.table} to snapshot ${snapId}?\n\nOnly works if old data files still exist. Nessie cannot undelete wiped S3 objects.`)) return + setBusy(true) + try { + const res = await fetch(`/api/nessie/tables/${sel.schema}/${sel.table}/restore`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ snapshot_id: snapId }), + }).then((r) => r.json()) + setToast(res.ok ? `Restored to ${snapId}` : (res.hint || res.error || 'Restore failed')) + if (res.as_of_sql) setToast((t) => `${t || ''}\n${res.as_of_sql}`) + await loadStructure(sel, snapId) + } finally { + setBusy(false) + } + } + + const uiUrl = health?.nessie_ui || 'http://10.0.21.50:19120' + const snapshots: Snapshot[] = structure?.snapshots || [] + const manifests: Manifest[] = structure?.manifests || [] + const files: DataFile[] = structure?.files || [] + + return ( +
+
+ +
+
Iceberg · Nessie catalog
+
+ Catalog versioning (Nessie) + manifest → data files (Iceberg via Trino) + {health?.nessie_ok ? ' · Nessie up' : ' · Nessie down'} + {health?.trino_ok ? ' · Trino up' : ' · Trino down'} +
+
+ + Nessie UI + + +
+ + {err && ( +
{err}
+ )} + {toast && ( +
{toast}
+ )} + + {loading ? ( +
+ loading Nessie / Iceberg… +
+ ) : ( +
+ {/* Catalog */} +
+
+ Tables on main +
+
+ {tables.length === 0 && ( +
No ICEBERG_TABLE entries in Nessie yet.
+ )} + {tables.map((t) => ( + + ))} +
+
+
+ Nessie commits +
+
+ {commits.map((c, i) => ( +
+
{c.message || '(no message)'}
+
{(c.hash || '').slice(0, 10)} · {c.commitTime || '—'}
+
+ ))} +
+
+
+ + {/* Timeline + structure */} +
+
+
+ Timeline · {sel ? `${sel.schema}.${sel.table}` : '—'} +
+ {busy && } + +
+
+ {snapshots.length === 0 && ( +
No snapshots (table missing in Trino iceberg catalog?)
+ )} + {snapshots.map((s) => ( + + ))} +
+
+
+ + {structure?.restore_warning || + 'Restore only re-points metadata. Hard-deleted S3/disk files cannot be recovered via Nessie.'} +
+
Manifests ({manifests.length})
+ {manifests.slice(0, 40).map((m, i) => ( +
+
{m.path}
+
+ +{m.added_data_files_count ?? 0} / exist {m.existing_data_files_count ?? 0} · {fmtBytes(m.length)} +
+
+ ))} + {structure?.as_of_sql && ( + 
+                  {structure.as_of_sql}
+
+ )} + {structure?.error &&
{structure.error}
} +
+
+ + {/* Files */} +
+
+
+ Data files ({files.length}) +
+
+ {files.map((f, i) => ( +
+
{f.path}
+
+ {f.format || '?'} · {f.record_count?.toLocaleString?.() ?? f.record_count} rows · {fmtBytes(f.size_bytes)} + {f.is_s3 ? ' · S3' : ' · local'} +
+
+ ))} + {!files.length &&
No data files for this table view.
} +
+
+
+
+ Raw S3 parquet/orc +
+
+ {s3objs.map((o) => ( +
+ {o.key} · {fmtBytes(o.size)} +
+ ))} + {!s3objs.length && ( +
+ No objects under lake/,iceberg/ (or S3 creds missing). Use Object Storage for full bucket browse. +
+ )} +
+
+
+
+ )} +
+ ) +}