root
b46e7f01dd
Pass DOCKHAND_API_TOKEN to container inventory calls so pipeline_active and topology animation work after Authentik. Replace ring-buffer-only CDC stats with minute rollups (no 1000 cap), add 15m/1h/6h/24h window selector on the Live Changes tab, and poll recent events on an interval. Co-authored-by: Cursor <cursoragent@cursor.com>
333 lines
11 KiB
Python
333 lines
11 KiB
Python
"""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 defaultdict, 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", "5000"))
|
|
METRICS_RETENTION_MIN = int(os.getenv("CDC_METRICS_RETENTION_MIN", str(24 * 60)))
|
|
|
|
# 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)
|
|
# Minute-level rollups — accurate counts beyond the display ring buffer cap.
|
|
_minute_rollups: dict[int, dict[str, Any]] = {}
|
|
_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,
|
|
}
|
|
|
|
|
|
def _record_metrics(entry: dict[str, Any]) -> None:
|
|
"""Track per-minute aggregates for accurate window stats (not ring-limited)."""
|
|
try:
|
|
ts = datetime.fromisoformat(entry["ts"].replace("Z", "+00:00"))
|
|
except Exception:
|
|
ts = datetime.now(timezone.utc)
|
|
minute_key = int(ts.timestamp()) // 60
|
|
bucket = _minute_rollups.setdefault(
|
|
minute_key,
|
|
{"total": 0, "by_source": defaultdict(int), "by_op": defaultdict(int), "by_table": defaultdict(int)},
|
|
)
|
|
bucket["total"] += 1
|
|
bucket["by_source"][entry["source"]] += 1
|
|
bucket["by_op"][entry["op"]] += 1
|
|
bucket["by_table"][entry["table"]] += 1
|
|
|
|
cutoff = minute_key - METRICS_RETENTION_MIN
|
|
for old_key in list(_minute_rollups.keys()):
|
|
if old_key < cutoff:
|
|
del _minute_rollups[old_key]
|
|
|
|
|
|
def _aggregate_window(minutes: int) -> dict[str, Any]:
|
|
"""Aggregate rollup buckets for the requested time window."""
|
|
now = datetime.now(timezone.utc)
|
|
cutoff_minute = int((now.timestamp() - minutes * 60) // 60)
|
|
use_hour_buckets = minutes >= 120
|
|
|
|
by_source: dict[str, int] = defaultdict(int)
|
|
by_op: dict[str, int] = defaultdict(int)
|
|
by_table: dict[str, int] = defaultdict(int)
|
|
buckets: dict[str, int] = defaultdict(int)
|
|
total = 0
|
|
|
|
for minute_key, bucket in _minute_rollups.items():
|
|
if minute_key < cutoff_minute:
|
|
continue
|
|
total += bucket["total"]
|
|
for k, v in bucket["by_source"].items():
|
|
by_source[k] += v
|
|
for k, v in bucket["by_op"].items():
|
|
by_op[k] += v
|
|
for k, v in bucket["by_table"].items():
|
|
by_table[k] += v
|
|
ts = datetime.fromtimestamp(minute_key * 60, timezone.utc)
|
|
label = ts.strftime("%H:00") if use_hour_buckets else ts.strftime("%H:%M")
|
|
buckets[label] += bucket["total"]
|
|
|
|
# Fill chart timeline with zero buckets so the window span is visible.
|
|
if use_hour_buckets:
|
|
span = max(1, (minutes + 59) // 60)
|
|
filled: dict[str, int] = {}
|
|
for i in range(span):
|
|
t = datetime.fromtimestamp(now.timestamp() - (span - 1 - i) * 3600, timezone.utc)
|
|
filled[t.strftime("%H:00")] = buckets.get(t.strftime("%H:00"), 0)
|
|
buckets = filled
|
|
else:
|
|
span = max(1, minutes)
|
|
filled = {}
|
|
for i in range(span):
|
|
t = datetime.fromtimestamp(now.timestamp() - (span - 1 - i) * 60, timezone.utc)
|
|
filled[t.strftime("%H:%M")] = buckets.get(t.strftime("%H:%M"), 0)
|
|
buckets = filled
|
|
|
|
return {
|
|
"total": total,
|
|
"by_source": dict(by_source),
|
|
"by_op": dict(by_op),
|
|
"by_table": dict(by_table),
|
|
"buckets": [{"t": k, "n": v} for k, v in sorted(buckets.items())],
|
|
"inserts": by_op.get("insert", 0),
|
|
"updates": by_op.get("update", 0),
|
|
"deletes": by_op.get("delete", 0),
|
|
"rate_per_min": round(total / max(1, minutes), 1),
|
|
}
|
|
|
|
|
|
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)
|
|
_record_metrics(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
|
|
|
|
|
|
def snapshot(minutes: int = 15) -> dict[str, Any]:
|
|
"""Lightweight CDC snapshot for other modules (Data Flow graph)."""
|
|
agg = _aggregate_window(minutes)
|
|
return {
|
|
"connected": _state["connected"],
|
|
"consumed": _state["consumed"],
|
|
"buffered": len(_ring),
|
|
"window_total": agg["total"],
|
|
"by_source": agg["by_source"],
|
|
}
|
|
|
|
|
|
# ── Endpoints ────────────────────────────────────────────────────────────────
|
|
@router.get("")
|
|
async def list_changes(
|
|
limit: int = Query(200, le=1000),
|
|
source: str | None = None,
|
|
op: str | None = None,
|
|
table: str | None = None,
|
|
minutes: int | None = Query(None, le=1440),
|
|
) -> JSONResponse:
|
|
items = list(_ring)
|
|
if minutes is not None:
|
|
cutoff = datetime.now(timezone.utc).timestamp() - minutes * 60
|
|
filtered_by_time: list[dict[str, Any]] = []
|
|
for c in items:
|
|
try:
|
|
if datetime.fromisoformat(c["ts"].replace("Z", "+00:00")).timestamp() >= cutoff:
|
|
filtered_by_time.append(c)
|
|
except Exception:
|
|
continue
|
|
items = filtered_by_time
|
|
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),
|
|
"buffer_cap": RING_SIZE,
|
|
"connected": _state["connected"],
|
|
"consumed": _state["consumed"],
|
|
"last_error": _state["last_error"],
|
|
"last_ts": _state["last_ts"],
|
|
})
|
|
|
|
|
|
@router.get("/stats")
|
|
async def change_stats(minutes: int = Query(15, le=1440)) -> JSONResponse:
|
|
agg = _aggregate_window(minutes)
|
|
return JSONResponse({
|
|
"ok": True,
|
|
"window_minutes": minutes,
|
|
**agg,
|
|
"connected": _state["connected"],
|
|
"consumed": _state["consumed"],
|
|
"buffered": len(_ring),
|
|
"buffer_cap": RING_SIZE,
|
|
"last_ts": _state["last_ts"],
|
|
})
|
|
|
|
|
|
@router.get("/status")
|
|
async def changes_status() -> JSONResponse:
|
|
return JSONResponse({"ok": True, "buffered": len(_ring), "buffer_cap": RING_SIZE, **_state})
|