"""Build live presentation deck from cluster snapshot + registry.""" from __future__ import annotations import json from typing import Any from node_registry import NODE_REGISTRY from topology_views import build_all_topologies from workload import build_workload_payload def _status_badge(level: str) -> str: return {"ok": "● Online", "warn": "◐ Degraded", "down": "○ Offline", "unknown": "? Unknown"}.get(level, level) def _slide(slide_id: str, title: str, subtitle: str, bullets: list[str], **extra: Any) -> dict[str, Any]: return {"id": slide_id, "title": title, "subtitle": subtitle, "bullets": bullets, **extra} def _node_slide(nid: str, extra_bullets: list[str] | None = None) -> dict[str, Any]: reg = NODE_REGISTRY.get(nid, {}) bullets: list[str] = [] if reg.get("description"): bullets.append(reg["description"]) bullets.append(f"VM: {reg.get('vm', '?')} (VMID {reg.get('vmid', '?')}) @ {reg.get('ip', '?')}") for link in reg.get("links", [])[:5]: bullets.append(f"{link.get('label', 'Link')}: {link.get('url', '')}") for ep in reg.get("endpoints", [])[:6]: bullets.append(f"{ep.get('name', '?')}: {ep.get('host', '?')}:{ep.get('port', '?')}") if extra_bullets: bullets.extend(extra_bullets) kind = "gpu" if nid == "gpu" else "command" if nid == "command" else "zone" if nid in ("openmetadata", "elastic"): kind = "architecture" return _slide( f"node-{nid}", reg.get("label", nid), f"{reg.get('role', 'service').upper()} · {reg.get('vm', '')}", bullets[:14], kind=kind, ) def build_presentation_payload(snap: dict[str, Any]) -> dict[str, Any]: workload = build_workload_payload(snap) topologies = workload.get("topologies") or build_all_topologies(snap) totals = workload.get("totals", {}) zones = workload.get("zones", []) gpu = workload.get("gpu", {}) etl = snap.get("etl", {}) hadoop = snap.get("hadoop", {}) objectscale = snap.get("objectscale", {}) command = snap.get("command_center", {}) databases = snap.get("databases", {}) docker = snap.get("docker", {}) lakehouse = snap.get("lakehouse", {}) governance = snap.get("governance") or {} slides: list[dict[str, Any]] = [] slides.append(_slide( "title", "Dell ATC Data Lab", "Live infrastructure presentation — Command Center", [ f"Snapshot: {snap.get('ts', 'now')}", f"Pipeline: {'ACTIVE' if totals.get('pipeline_active') else 'INACTIVE'}", f"Apps running: {totals.get('apps_running', 0)}/{totals.get('apps_total', 0)}", f"CDC connectors: {totals.get('connectors', 0)} · Source DBs: 5 engines on db02", f"LLM: {gpu.get('model') or 'offline'} ({gpu.get('gpu_count', 0)}× V100)", "Command Center → http://10.0.21.33/ · Data Platform tab → Presentation", ], kind="hero", )) slides.append(_slide( "mission", "Mission", "End-to-end modern data platform on Dell infrastructure", [ "Ingest CDC from PostgreSQL, MySQL, MongoDB (+ Cassandra & Neo4j for analytics)", "Stream via Kafka & Debezium into Spark, Trino & Iceberg on the lakehouse", "Land curated data on ObjectScale S3 — federated SQL with Trino", "Govern with OpenMetadata — catalog, lineage, PII classification", "Search & observe via Elasticsearch/Kibana · BI via Superset", "Parallel 9-node Hadoop HDFS cluster for batch workloads", "GPU lab (4× V100) powers autonomous agents with local vLLM inference", "This Command Center orchestrates agents, approvals & live visibility", ], kind="narrative", )) slides.append(_slide( "command-center-ui", "Command Center UI", "Everything you operate from this dashboard", [ "Data Platform — interactive topology + live presentation deck", "Data Sources UI — browse, filter & edit all 5 source databases", "Live Changes — real-time Debezium CDC event stream", "Data Flow — OpenMetadata catalog, lineage & pipeline map", "Data Quality — Docling document QA + RAG ingest", "Knowledge Chat — GPU-backed RAG over lab documentation", "Object Storage · HDFS · Elasticsearch · SSH terminal", "Agent fleet · Approval inbox · Live GPU matrix", ], kind="command", )) arch = topologies.get("architecture") or workload.get("topology") or {} arch_nodes = arch.get("nodes", []) slides.append(_slide( "architecture", "Data Platform Architecture", arch.get("subtitle", "Sources → Ingestion → Compute → Storage → Consumers"), [f"{n.get('label', n.get('id'))}: {n.get('subtitle', n.get('role', ''))}" for n in arch_nodes[:16]], kind="topology", topology=arch, )) db_lines = [ f"Host atc-db02 (10.0.21.51) — {databases.get('running', 0)}/{databases.get('total', 0)} containers up", "postgres_sales — PostgreSQL sales_orders (CDC → Debezium)", "mysql_hr — MySQL employee_events (CDC → Debezium)", "mongodb_supplychain — MongoDB supplychain.events (CDC → Debezium)", "cassandra_telemetry — Cassandra device_metrics (Trino federated)", "neo4j_graph — Product/Supplier graph · 4.5M nodes", ] for c in (databases.get("containers") or [])[:8]: db_lines.append(f" · {c.get('name', '?')}: {c.get('state', '?')}") slides.append(_slide( "source-databases", "Source Databases", "DB Vault · atc-db02 · 10.0.21.51", db_lines, kind="topology", )) pipeline = topologies.get("pipeline", {}) connector_lines = [ f" · {cs['name']}: {cs.get('state', '?')}" for cs in (etl.get("connector_status") or [])[:8] ] slides.append(_slide( "pipeline", "CDC Pipeline", pipeline.get("subtitle", "Airflow → DB → Debezium → Kafka → Lakehouse → S3"), [ f"Airflow: {'healthy' if etl.get('airflow_healthy') else 'degraded'} — {etl.get('airflow_url', 'http://10.0.21.55:8080')}", f"Kafka UI: {'UP' if etl.get('kafka_ui_ok') else 'DOWN'} — http://10.0.21.36:9000", f"Debezium Connect: http://10.0.21.50:8083", f"Connectors: {', '.join(etl.get('connectors') or []) or 'none'}", *connector_lines, f"Spark UI: {'UP' if etl.get('spark_ui_ok') else 'DOWN'} — http://10.0.21.50:8080", f"ObjectScale S3: {'reachable' if objectscale.get('reachable') else 'down'} — bucket={objectscale.get('bucket', 'data')}", ], kind="topology", topology=pipeline, )) lake_lines = [ f"atc-lake01 @ {lakehouse.get('host', '10.0.21.50')} — {lakehouse.get('running', 0)}/{lakehouse.get('total', 0)} containers", f"Trino: {'UP' if lakehouse.get('trino_ok') else 'DOWN'} — http://10.0.21.50:8089", "Spark — batch & streaming compute", "Kafka Connect + Debezium — CDC ingestion", "s3-kafka-consumer — events → ObjectScale S3", "Iceberg catalog — bronze → silver → gold tables", ] for c in (lakehouse.get("containers") or [])[:8]: lake_lines.append(f" · {c.get('name', '?')}: {c.get('state', '?')}") slides.append(_slide( "lakehouse", "Lakehouse Hub", "Spark · Trino · Iceberg · Kafka Connect", lake_lines, kind="topology", )) docker_lines = [ f"atc-docker01 @ 10.0.21.45 — {docker.get('running', 0)}/{docker.get('total', 0)} containers", "Homepage — http://10.0.21.45", "Dockhand — container management http://10.0.21.45:8082", "Apache Superset — BI dashboards http://10.0.21.45:8088", "Forgejo / Gitea · monitoring · nginx · redis", ] for c in (docker.get("containers") or [])[:10]: docker_lines.append(f" · {c.get('name', '?')}: {c.get('state', '?')}") slides.append(_slide( "docker-rack", "Docker Rack & Analytics", "Platform services on atc-docker01", docker_lines, kind="zone", )) cdc = governance.get("cdc") or {} gov_lines = [ f"OpenMetadata UI: {governance.get('openmetadata_url', 'http://10.0.21.47:8585')}", "Ingestion Airflow: http://10.0.21.47:8080", "Catalog · lineage · data quality · PII auto-classification (Presidio NER)", ] if "error" not in cdc: by_src = ", ".join(f"{k}={v}" for k, v in (cdc.get("by_source") or {}).items()) or "none" gov_lines.append(f"CDC stream: connected={cdc.get('connected')} · {cdc.get('window_total', 0)} changes/15m ({by_src})") ps = governance.get("pii_summary") or {} if "error" not in ps: gov_lines.append(f"PII: {ps.get('pii_columns', 0)} columns · {ps.get('masked_columns', 0)} masked") for ln in (governance.get("lineage") or [])[:4]: gov_lines.append(f"Lineage: {ln}") slides.append(_slide( "governance", "Governance & Metadata", "OpenMetadata · CDC · PII · Lineage", gov_lines, kind="architecture", )) slides.append(_node_slide("openmetadata")) slides.append(_node_slide("elastic")) for zone in zones: apps = zone.get("apps") or [] app_lines = [ f"{a['name']}: {a['state']}" + (f" ({a.get('host', '')})" if a.get("host") else "") for a in apps[:12] ] slides.append(_slide( f"zone-{zone['id']}", zone["label"], f"{zone.get('vm', '')} · {zone.get('ip', '')} · {_status_badge(zone.get('level', 'unknown'))}", [ f"Containers: {zone.get('running', 0)}/{zone.get('total', 0)} running", *app_lines, ], kind="zone", zone=zone, )) PRESENTATION_NODES = [ "airflow", "db", "debezium", "kafka", "lakehouse", "s3", "docker", "hadoop", "gpu", "command", ] slides.append(_slide( "infrastructure-map", "Infrastructure Map", "Proxmox VMs & services across VLAN 20/21", [ f"{NODE_REGISTRY[nid]['label']} — {NODE_REGISTRY[nid].get('vm')} " f"(VMID {NODE_REGISTRY[nid].get('vmid', '?')}) @ {NODE_REGISTRY[nid].get('ip')}" for nid in PRESENTATION_NODES if nid in NODE_REGISTRY ], kind="registry", )) for nid in PRESENTATION_NODES: if nid in NODE_REGISTRY and nid not in ("docker", "db", "lakehouse"): slides.append(_node_slide(nid)) dn_lines = [ f" · {dn['host']}: {dn.get('used_gb', 0)} GB — {dn.get('state', '')}" for dn in (hadoop.get("datanodes") or [])[:6] ] slides.append(_slide( "hadoop", "Hadoop HDFS", "9-node parallel storage cluster", [ f"NameNode: {'UP' if hadoop.get('reachable') else 'DOWN'} — {hadoop.get('namenode', 'http://10.0.21.61:9870')}", f"Capacity: {hadoop.get('capacity_used_gb', '?')} / {hadoop.get('capacity_total_gb', '?')} GB", f"DataNodes: {hadoop.get('live_datanodes', 0)} live, {hadoop.get('dead_datanodes', 0)} dead", f"Files: {hadoop.get('files_total', 0)}, Blocks: {hadoop.get('blocks_total', 0)}", *dn_lines, ], kind="data", )) gpus = gpu.get("gpus") or snap.get("gpu", {}).get("gpus") or [] try: from gpu_config import resolve_gpu_identity gpu_id = resolve_gpu_identity(snap.get("gpu") or gpu) except Exception: host = (snap.get("gpu") or gpu).get("host") or "10.0.10.106" gpu_id = { "vm": "atc-gpu-prod", "vmid": 306, "ui_url": f"http://{host}:9000", "llm_url": f"http://{host}:8001/v1", } gpu_lines = [ f"GPU{g['index']}: {g.get('util_gpu', 0):.0f}% util, " f"{g.get('memory_used_mib', 0):.0f}/{g.get('memory_total_mib', 0):.0f} MiB" for g in gpus[:4] ] slides.append(_slide( "gpu", "GPU Lab & GenAI", f"{gpu.get('model') or 'vLLM'} on {gpu_id['vm']} (VM {gpu_id['vmid']})", [ f"Inference: {'ON' if gpu.get('inference_active') else 'OFF'}", f"API: {gpu_id['llm_url']}", f"GPU Lab UI: {gpu_id['ui_url']}", "Kibana/Elastic: http://10.0.21.46:5601", *gpu_lines, ], kind="gpu", )) slides.append(_slide( "agents", "Autonomous Agents", "Mo & Bart supervise 5 domain operators + MCP hub", [ "ETL Guardian — Airflow, Kafka, Debezium, connectors", "Data Custodian — PostgreSQL, MySQL, MongoDB, Cassandra, Neo4j", "Lakehouse Ops — Spark, Trino, Iceberg, ObjectScale S3, OpenMetadata", "Hadoop Ranger — HDFS NameNode, DataNodes, block health", "Infra Sentinel — Docker rack, GPU lab, Command Center", "All agents receive LIVE cluster snapshot in every LLM prompt", ], kind="agents", )) cc_apps = [f"{c['name']}: {c['state']}" for c in (command.get("containers") or [])] slides.append(_slide( "command", "Command Center Stack", "VM 304 — this dashboard runs here", [ f"Host: {command.get('host', '10.0.21.33')} (VMID {command.get('vmid', 304)})", f"Stack: {command.get('running', 0)}/{command.get('total', 0)} containers (api, ui, caddy, redis, postgres)", *cc_apps, "WebSocket ops feed · Approval inbox · Agent terminals · Data Sources UI", ], kind="command", )) slides.append(_slide( "demo", "Live Demo Tips", "Use this deck during customer presentations", [ "Press ← → or click dots to navigate slides · F = fullscreen", "Data Platform tab → Presentation sub-tab (this deck)", "Data Platform → Topology for interactive pipeline map", "Export HTML opens a standalone deck for projectors / offline", "Ask agents in the Command Bar — they see full cluster context", "Trigger data generation in Data Sources UI → Generate tab", ], kind="cta", )) return { "ts": snap.get("ts"), "title": "Dell ATC Data Lab", "subtitle": "Live Infrastructure Presentation", "totals": totals, "pipeline_active": totals.get("pipeline_active"), "slides": slides, "slide_count": len(slides), "workload": workload, "topologies": topologies, } def render_presentation_html(payload: dict[str, Any]) -> str: slides_json = json.dumps(payload.get("slides", []), default=str) title = payload.get("title", "ATC Lab") ts = payload.get("ts", "") return f""" {title} — Presentation

{title}

Dell ATC · Live snapshot {ts}
"""