From 1a454f76cfb992fb8a43d37f8bcde510111bb5db Mon Sep 17 00:00:00 2001 From: mo Date: Sat, 27 Jun 2026 01:31:18 +0200 Subject: [PATCH] feat(cdc): live Debezium CDC consumer + /api/changes for the Changes dashboard Add api/cdc_consumer.py: aiokafka background consumer subscribes to the CDC topics (postgres_sales/mysql_hr/mongodb_supplychain + cassandra/neo4j), parses Debezium before/after envelopes, keeps a ring buffer and publishes each change live as type=cdc_change. Endpoints /api/changes, /api/changes/stats, /status. --- api/Dockerfile | 2 +- api/cdc_consumer.py | 246 +++++++++++++++++++++++++++++++++++++++++++ api/main.py | 4 + api/requirements.txt | 1 + 4 files changed, 252 insertions(+), 1 deletion(-) create mode 100644 api/cdc_consumer.py diff --git a/api/Dockerfile b/api/Dockerfile index fba1167..07299c7 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 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 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 new file mode 100644 index 0000000..0d11536 --- /dev/null +++ b/api/cdc_consumer.py @@ -0,0 +1,246 @@ +"""Live CDC consumer for the Command Center "Changes" dashboard. + +A background aiokafka consumer subscribes to the Debezium CDC topics on the +Kafka broker, parses each change event (insert/update/delete with before/after +images), keeps a bounded in-memory ring buffer, and fans every change out over +the WebSocket bus as a ``cdc_change`` event so the UI can render new & changed +data in real time. + +Endpoints: + GET /api/changes -> recent changes from the ring buffer (filterable) + GET /api/changes/stats -> volume per source/table/op + per-minute buckets +""" + +from __future__ import annotations + +import asyncio +import json +import os +import re +import uuid +from collections import deque +from datetime import datetime, timezone +from typing import Any + +from fastapi import APIRouter, Query +from fastapi.responses import JSONResponse + +router = APIRouter(prefix="/api/changes", tags=["changes"]) + +KAFKA_BOOTSTRAP = os.getenv("KAFKA_BOOTSTRAP", "10.0.21.36:9092") +RING_SIZE = int(os.getenv("CDC_RING_SIZE", "1000")) + +# Active CDC topic prefixes -> logical source. Matches Debezium topic.prefix. +PREFIX_SOURCE = { + "postgres_sales": "postgres", + "mysql_hr": "mysql", + "mongodb_supplychain": "mongodb", + "cassandra_telemetry": "cassandra", + "neo4j_graph": "neo4j", +} +TOPIC_PATTERN = re.compile( + r"^(postgres_sales|mysql_hr|mongodb_supplychain|cassandra_telemetry|neo4j_graph)\..+" +) + +OP_MAP = {"c": "insert", "u": "update", "d": "delete", "r": "snapshot"} + +_ring: deque[dict[str, Any]] = deque(maxlen=RING_SIZE) +_state: dict[str, Any] = { + "started": False, + "connected": False, + "consumed": 0, + "last_error": None, + "last_ts": None, + "topics": [], +} + + +def _source_of(topic: str) -> tuple[str, str]: + prefix = topic.split(".", 1)[0] + source = PREFIX_SOURCE.get(prefix, prefix) + table = topic.split(".")[-1] + return source, table + + +def _coerce(v: Any) -> Any: + """Debezium/Mongo nests the document as a JSON string sometimes.""" + if isinstance(v, str): + try: + return json.loads(v) + except Exception: + return v + return v + + +def _key_fields(source: str, table: str, after: Any, before: Any) -> str: + row = after or before or {} + if not isinstance(row, dict): + return "" + prefer = ["order_id", "event_id", "_id", "id", "region", "order_status", + "event_type", "type", "amount", "salary_change"] + parts = [] + for k in prefer: + if k in row and row[k] is not None: + val = row[k] + if isinstance(val, (dict, list)): + continue + sval = str(val) + if len(sval) > 40: + sval = sval[:40] + "…" + parts.append(f"{k}={sval}") + if len(parts) >= 4: + break + return " ".join(parts) + + +def _parse(topic: str, value: bytes | None) -> dict[str, Any] | None: + if value is None: # tombstone + return None + try: + payload = json.loads(value.decode("utf-8")) + except Exception: + return None + # Debezium envelope: {schema, payload:{before,after,op,source,ts_ms}} + if isinstance(payload, dict) and "payload" in payload and isinstance(payload["payload"], dict): + payload = payload["payload"] + if not isinstance(payload, dict): + return None + op = OP_MAP.get(payload.get("op"), payload.get("op") or "change") + before = _coerce(payload.get("before")) + after = _coerce(payload.get("after")) + src = payload.get("source") or {} + source, table = _source_of(topic) + if isinstance(src, dict) and src.get("table"): + table = src.get("table") + ts_ms = payload.get("ts_ms") + ts = datetime.fromtimestamp(ts_ms / 1000, timezone.utc).isoformat() if ts_ms else datetime.now(timezone.utc).isoformat() + return { + "id": uuid.uuid4().hex[:10], + "ts": ts, + "source": source, + "table": table, + "topic": topic, + "op": op, + "summary": _key_fields(source, table, after, before), + "before": before if isinstance(before, dict) else None, + "after": after if isinstance(after, dict) else None, + } + + +async def cdc_consumer_loop() -> None: + """Background loop: consume CDC topics and publish each change live.""" + _state["started"] = True + await asyncio.sleep(6) + try: + from aiokafka import AIOKafkaConsumer + except Exception as exc: + _state["last_error"] = f"aiokafka import failed: {exc}" + return + + while True: + consumer = None + try: + consumer = AIOKafkaConsumer( + bootstrap_servers=KAFKA_BOOTSTRAP, + group_id=f"atc-cc-cdc-{uuid.uuid4().hex[:8]}", + auto_offset_reset="latest", + enable_auto_commit=False, + client_id="atc-command-center-cdc", + ) + consumer.subscribe(pattern=TOPIC_PATTERN) + await consumer.start() + _state["connected"] = True + _state["last_error"] = None + from main import publish_event + + async for msg in consumer: + entry = _parse(msg.topic, msg.value) + if not entry: + continue + _ring.append(entry) + _state["consumed"] += 1 + _state["last_ts"] = entry["ts"] + try: + _state["topics"] = sorted(consumer.subscription() or []) + except Exception: + pass + try: + await publish_event({"type": "cdc_change", "entry": entry}) + except Exception: + pass + except Exception as exc: + _state["connected"] = False + _state["last_error"] = str(exc) + await asyncio.sleep(10) # backoff then reconnect + finally: + if consumer is not None: + try: + await consumer.stop() + except Exception: + pass + + +# ── Endpoints ──────────────────────────────────────────────────────────────── +@router.get("") +async def list_changes( + limit: int = Query(100, le=500), + source: str | None = None, + op: str | None = None, + table: str | None = None, +) -> JSONResponse: + items = list(_ring) + if source: + items = [c for c in items if c["source"] == source] + if op: + items = [c for c in items if c["op"] == op] + if table: + items = [c for c in items if c["table"] == table] + items = list(reversed(items))[:limit] + return JSONResponse({ + "ok": True, + "changes": items, + "buffered": len(_ring), + "connected": _state["connected"], + "consumed": _state["consumed"], + "last_error": _state["last_error"], + }) + + +@router.get("/stats") +async def change_stats(minutes: int = Query(15, le=240)) -> JSONResponse: + now = datetime.now(timezone.utc) + cutoff = now.timestamp() - minutes * 60 + by_source: dict[str, int] = {} + by_op: dict[str, int] = {} + by_table: dict[str, int] = {} + buckets: dict[str, int] = {} + total = 0 + for c in _ring: + try: + t = datetime.fromisoformat(c["ts"]).timestamp() + except Exception: + continue + if t < cutoff: + continue + total += 1 + by_source[c["source"]] = by_source.get(c["source"], 0) + 1 + by_op[c["op"]] = by_op.get(c["op"], 0) + 1 + by_table[c["table"]] = by_table.get(c["table"], 0) + 1 + bucket = datetime.fromtimestamp(t, timezone.utc).strftime("%H:%M") + buckets[bucket] = buckets.get(bucket, 0) + 1 + return JSONResponse({ + "ok": True, + "window_minutes": minutes, + "total": total, + "by_source": by_source, + "by_op": by_op, + "by_table": by_table, + "buckets": [{"t": k, "n": v} for k, v in sorted(buckets.items())], + "connected": _state["connected"], + "consumed": _state["consumed"], + }) + + +@router.get("/status") +async def changes_status() -> JSONResponse: + return JSONResponse({"ok": True, "buffered": len(_ring), **_state}) diff --git a/api/main.py b/api/main.py index 5bb2dec..20003de 100644 --- a/api/main.py +++ b/api/main.py @@ -43,6 +43,7 @@ from hadoop_analytics import router as hadoop_router from elasticsearch_api import router as elasticsearch_router from sql_console import router as sql_router from agent_ops import router as agent_ops_router, agent_dml_loop +from cdc_consumer import router as cdc_router, cdc_consumer_loop 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 @@ -713,10 +714,12 @@ async def lifespan(app: FastAPI): await terminal_log(nid, f"{meta['label']} shell ready — click node to connect", level="info", phase="boot") task = asyncio.create_task(heartbeat_loop()) dml_task = asyncio.create_task(agent_dml_loop()) + cdc_task = asyncio.create_task(cdc_consumer_loop()) add_feed("infra-sentinel", "ATC Command Center API online", "info") yield task.cancel() dml_task.cancel() + cdc_task.cancel() if redis_client: await redis_client.close() @@ -729,6 +732,7 @@ app.include_router(hadoop_router) app.include_router(elasticsearch_router) app.include_router(sql_router) app.include_router(agent_ops_router) +app.include_router(cdc_router) app.add_middleware( CORSMiddleware, allow_origins=["*"], diff --git a/api/requirements.txt b/api/requirements.txt index c94c875..9987891 100644 --- a/api/requirements.txt +++ b/api/requirements.txt @@ -15,3 +15,4 @@ neo4j==5.26.0 python-pptx==1.0.2 boto3==1.35.99 paramiko==3.5.0 +aiokafka==0.12.0