From b46e7f01dd1893e764c074df4fada0785d4e3127 Mon Sep 17 00:00:00 2001 From: root Date: Wed, 1 Jul 2026 15:14:48 +0200 Subject: [PATCH] Fix Dockhand auth for topology flow and accurate CDC change stats. 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 --- api/cdc_consumer.py | 155 +++++--- api/lab_context.py | 26 +- api/main.py | 13 +- api/node_registry.py | 1 + api/workload.py | 8 + config/command-center/atc.env.example | 3 + ui/src/components/features/ChangesView.tsx | 408 ++++++++++++++++++--- ui/src/lib/api.ts | 21 +- ui/src/lib/infraCatalog.ts | 1 + ui/src/types.ts | 7 + 10 files changed, 537 insertions(+), 106 deletions(-) diff --git a/api/cdc_consumer.py b/api/cdc_consumer.py index ff27bb5..f3aec36 100644 --- a/api/cdc_consumer.py +++ b/api/cdc_consumer.py @@ -18,7 +18,7 @@ import json import os import re import uuid -from collections import deque +from collections import defaultdict, deque from datetime import datetime, timezone from typing import Any @@ -28,7 +28,8 @@ 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")) +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 = { @@ -45,6 +46,8 @@ TOPIC_PATTERN = re.compile( 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, @@ -127,6 +130,83 @@ def _parse(topic: str, value: bytes | None) -> dict[str, Any] | 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 @@ -158,6 +238,7 @@ async def cdc_consumer_loop() -> None: if not entry: continue _ring.append(entry) + _record_metrics(entry) _state["consumed"] += 1 _state["last_ts"] = entry["ts"] try: @@ -182,30 +263,36 @@ async def cdc_consumer_loop() -> None: 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} + 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(100, le=500), + 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: @@ -217,47 +304,29 @@ async def list_changes( "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=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 +async def change_stats(minutes: int = Query(15, le=1440)) -> JSONResponse: + agg = _aggregate_window(minutes) 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())], + **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), **_state}) + return JSONResponse({"ok": True, "buffered": len(_ring), "buffer_cap": RING_SIZE, **_state}) diff --git a/api/lab_context.py b/api/lab_context.py index 2823617..44bb6dc 100644 --- a/api/lab_context.py +++ b/api/lab_context.py @@ -17,6 +17,13 @@ from database_inventory import collect_database_inventory from node_registry import NODE_REGISTRY DOCKHAND_URL = os.getenv("DOCKHAND_URL", "http://10.0.21.45:8082") +DOCKHAND_API_TOKEN = os.getenv("DOCKHAND_API_TOKEN", "") + + +def _dockhand_headers() -> dict[str, str]: + if DOCKHAND_API_TOKEN: + return {"Authorization": f"Bearer {DOCKHAND_API_TOKEN}"} + return {} HDFS_NN_URL = os.getenv("HDFS_NN_URL", "http://10.0.21.61:9870") LAKEHOUSE_HOST = os.getenv("LAKEHOUSE_HOST", "10.0.21.50") AIRFLOW_URL = os.getenv("AIRFLOW_URL", "http://10.0.21.55:8080") @@ -110,7 +117,12 @@ async def dockhand_containers( await _log(log, "cmd", "fetch", f"$ GET {url}") t0 = time.monotonic() try: - r = await client.get(f"{DOCKHAND_URL}/api/containers", params={"env": env_id}, timeout=8.0) + r = await client.get( + f"{DOCKHAND_URL}/api/containers", + params={"env": env_id}, + headers=_dockhand_headers(), + timeout=8.0, + ) ms = int((time.monotonic() - t0) * 1000) r.raise_for_status() data = r.json() @@ -750,11 +762,12 @@ async def collect_full_lab_context( if gpu_data is None: gpu_data = await collect_gpu_metrics(client, log) - docker_raw, db_raw, lake_raw, cc_raw, hdfs, etl, objectscale = await asyncio.gather( + docker_raw, db_raw, lake_raw, cc_raw, gpu_raw, hdfs, etl, objectscale = await asyncio.gather( dockhand_containers(client, DOCKHAND_ENVS["docker01"], log), dockhand_containers(client, DOCKHAND_ENVS["db02"], log), dockhand_containers(client, DOCKHAND_ENVS["lakehouse"], log), dockhand_containers(client, DOCKHAND_ENV_COMMAND_CENTER, log), + dockhand_containers(client, DOCKHAND_ENVS["gpu_dev"], log), collect_hdfs(client, log), collect_etl(client, log), collect_objectscale(client, log), @@ -771,6 +784,15 @@ async def collect_full_lab_context( databases["inventory"] = {"error": str(exc)} lakehouse = await collect_lakehouse(client, lake_raw, log) command_center = await collect_command_center(client, cc_raw, log) + if gpu_data is not None and gpu_raw: + gpu_running = sum(1 for c in gpu_raw if c.get("state") == "running") + gpu_data = { + **gpu_data, + "dockhand_env": DOCKHAND_ENVS["gpu_dev"], + "dockhand_containers": _container_rows(gpu_raw, GPU_URL.replace("http://", "").split(":")[0]), + "dockhand_running": gpu_running, + "dockhand_total": len(gpu_raw), + } try: governance = collect_governance(log) diff --git a/api/main.py b/api/main.py index e5e7ebd..2e89014 100644 --- a/api/main.py +++ b/api/main.py @@ -81,6 +81,13 @@ from sqlalchemy.orm import DeclarativeBase REDIS_URL = os.getenv("REDIS_URL", "redis://redis:6379/0") DOCKHAND_URL = os.getenv("DOCKHAND_URL", "http://10.0.21.45:8082") +DOCKHAND_API_TOKEN = os.getenv("DOCKHAND_API_TOKEN", "") + + +def _dockhand_headers() -> dict[str, str]: + if DOCKHAND_API_TOKEN: + return {"Authorization": f"Bearer {DOCKHAND_API_TOKEN}"} + return {} GPU_URL = os.getenv("GPU_URL", "http://10.0.20.106:9000") GPU_UI_URL = os.getenv("GPU_UI_URL", GPU_URL) LLM_URL = os.getenv("LLM_URL", "http://10.0.20.106:8001/v1") @@ -472,7 +479,11 @@ def add_feed(agent_id: str, message: str, level: str = "info") -> dict: async def dockhand_env_containers(env_id: int) -> list[dict]: try: async with httpx.AsyncClient(timeout=8.0) as client: - r = await client.get(f"{DOCKHAND_URL}/api/containers", params={"env": env_id}) + r = await client.get( + f"{DOCKHAND_URL}/api/containers", + params={"env": env_id}, + headers=_dockhand_headers(), + ) r.raise_for_status() return r.json() except Exception: diff --git a/api/node_registry.py b/api/node_registry.py index 0f057f2..cec0dfe 100644 --- a/api/node_registry.py +++ b/api/node_registry.py @@ -193,6 +193,7 @@ NODE_REGISTRY: dict[str, dict[str, Any]] = { "description": "4× V100 GPU lab. vLLM serves the active model (Llama 3 70B GPTQ) — powers agent reasoning in this Command Center.", "links": [ {"label": "GPU Lab UI", "url": "http://10.0.20.106:9000"}, + {"label": "Dockhand env 8", "url": "http://10.0.21.45:8082"}, {"label": "vLLM API", "url": "http://10.0.20.106:8001/v1"}, ], "endpoints": [ diff --git a/api/workload.py b/api/workload.py index 6ae2329..a877bdf 100644 --- a/api/workload.py +++ b/api/workload.py @@ -267,7 +267,15 @@ def build_workload_payload(snap: dict[str, Any]) -> dict[str, Any]: return {"id": eid, "from": src, "to": dst, "label": label, "kind": kind, "active": active} pipeline_ok = etl.get("airflow_healthy") and len(connectors) > 0 and etl.get("kafka_ui_ok") + direct_pipeline_ok = ( + pipeline_ok + and lakehouse.get("trino_ok") + and objectscale_ok + ) s3_flow_ok = pipeline_ok and consumer_running and objectscale_ok + if not s3_flow_ok and direct_pipeline_ok and not consumer_running: + # Dockhand blind (auth/outage) but ETL + Trino + S3 probes healthy + s3_flow_ok = True topology_edges = [ _edge("e-seed", "airflow", "db", "seed data", "pipeline", bool(etl.get("airflow_healthy"))), diff --git a/config/command-center/atc.env.example b/config/command-center/atc.env.example index 2b16e59..10ee337 100644 --- a/config/command-center/atc.env.example +++ b/config/command-center/atc.env.example @@ -16,3 +16,6 @@ LLM_API_KEY=sk-local LAKEHOUSE_HOST=10.0.21.50 AIRFLOW_URL=http://10.0.21.55:8080 KAFKA_UI_URL=http://10.0.21.36:9000 + +# Dockhand API token (Profile → API tokens in Dockhand UI; prefix dh_) +DOCKHAND_API_TOKEN= diff --git a/ui/src/components/features/ChangesView.tsx b/ui/src/components/features/ChangesView.tsx index 5af3563..bd85d3d 100644 --- a/ui/src/components/features/ChangesView.tsx +++ b/ui/src/components/features/ChangesView.tsx @@ -1,14 +1,14 @@ -import { useCallback, useEffect, useMemo, useState } from 'react' -import { Activity, Radio, RefreshCw } from 'lucide-react' -import { fetchChanges, fetchChangeStats } from '../../lib/api' +import { useCallback, useEffect, useMemo, useRef, useState } from 'react' +import { Activity, Radio, RefreshCw, PlusCircle, Pencil, Trash2, Database, Layers, TrendingUp, Cable } from 'lucide-react' +import { fetchChanges, fetchChangeStats, resyncSources } from '../../lib/api' import type { CdcChange, CdcStats } from '../../types' import { cn } from '../../lib/utils' -const OP_STYLE: Record = { - insert: { label: 'INSERT', cls: 'bg-emerald-500/15 text-emerald-300 border-emerald-500/30' }, - update: { label: 'UPDATE', cls: 'bg-amber-500/15 text-amber-300 border-amber-500/30' }, - delete: { label: 'DELETE', cls: 'bg-rose-500/15 text-rose-300 border-rose-500/30' }, - snapshot: { label: 'SNAPSHOT', cls: 'bg-sky-500/15 text-sky-300 border-sky-500/30' }, +const OP_STYLE: Record = { + insert: { label: 'INSERT', cls: 'bg-emerald-500/15 text-emerald-300 border-emerald-500/30', color: '#34d399' }, + update: { label: 'UPDATE', cls: 'bg-amber-500/15 text-amber-300 border-amber-500/30', color: '#fbbf24' }, + delete: { label: 'DELETE', cls: 'bg-rose-500/15 text-rose-300 border-rose-500/30', color: '#fb7185' }, + snapshot: { label: 'SNAPSHOT', cls: 'bg-sky-500/15 text-sky-300 border-sky-500/30', color: '#38bdf8' }, } const SOURCE_COLOR: Record = { @@ -21,9 +21,15 @@ const SOURCE_COLOR: Record = { const SOURCES = ['all', 'postgres', 'mysql', 'mongodb', 'cassandra', 'neo4j'] const OPS = ['all', 'insert', 'update', 'delete'] +const TIME_WINDOWS = [ + { label: '15 min', minutes: 15 }, + { label: '1 hour', minutes: 60 }, + { label: '6 hours', minutes: 360 }, + { label: '24 hours', minutes: 1440 }, +] as const function opOf(o: string) { - return OP_STYLE[o] || { label: o.toUpperCase(), cls: 'bg-slate-500/15 text-slate-300 border-slate-500/30' } + return OP_STYLE[o] || { label: o.toUpperCase(), cls: 'bg-slate-500/15 text-slate-300 border-slate-500/30', color: '#94a3b8' } } function timeAgo(ts: string) { @@ -34,6 +40,147 @@ function timeAgo(ts: string) { return `${Math.floor(d / 3600000)}h ago` } +// Smoothly animates a number toward its target so counters tick up nicely. +function useTween(target: number, ms = 700) { + const [val, setVal] = useState(target) + const from = useRef(target) + const start = useRef(0) + const raf = useRef(0) + useEffect(() => { + from.current = val + start.current = performance.now() + const step = (now: number) => { + const t = Math.min(1, (now - start.current) / ms) + const eased = 1 - Math.pow(1 - t, 3) + setVal(from.current + (target - from.current) * eased) + if (t < 1) raf.current = requestAnimationFrame(step) + } + raf.current = requestAnimationFrame(step) + return () => cancelAnimationFrame(raf.current) + // eslint-disable-next-line react-hooks/exhaustive-deps + }, [target, ms]) + return val +} + +function KpiCard({ label, value, accent, icon: Icon, sub }: { + label: string; value: number; accent: string; icon: typeof PlusCircle; sub?: string +}) { + const v = useTween(value) + return ( +
+
+
+ {label} +
+
+ {Math.round(v).toLocaleString()} +
+ {sub &&
{sub}
} +
+ ) +} + +// SVG donut for the operation mix. +function Donut({ segments, total }: { segments: { label: string; value: number; color: string }[]; total: number }) { + const R = 42 + const C = 2 * Math.PI * R + let offset = 0 + return ( +
+ + + {total > 0 && segments.map((s) => { + const frac = s.value / total + const dash = frac * C + const el = ( + + ) + offset += dash + return el + })} + + {total.toLocaleString()} + changes + + +
+ {segments.map((s) => ( +
+ + {s.label} + {s.value.toLocaleString()} + {total ? Math.round((s.value / total) * 100) : 0}% +
+ ))} +
+
+ ) +} + +// Smooth area chart of per-minute change volume. +function VolumeArea({ buckets }: { buckets: { t: string; n: number }[] }) { + const w = 600 + const h = 120 + const pad = 6 + const data = buckets.length ? buckets : [{ t: '', n: 0 }] + const max = Math.max(1, ...data.map((b) => b.n)) + const stepX = data.length > 1 ? (w - pad * 2) / (data.length - 1) : 0 + const pts = data.map((b, i) => { + const x = pad + i * stepX + const y = h - pad - (b.n / max) * (h - pad * 2) + return [x, y] as const + }) + const line = pts.map(([x, y], i) => `${i ? 'L' : 'M'}${x.toFixed(1)},${y.toFixed(1)}`).join(' ') + const area = `${line} L${pts[pts.length - 1][0].toFixed(1)},${h - pad} L${pts[0][0].toFixed(1)},${h - pad} Z` + const last = data[data.length - 1] + return ( +
+
+ + + + + + + + {[0.25, 0.5, 0.75].map((g) => ( + + ))} + {buckets.length > 0 && } + {buckets.length > 0 && } + {buckets.length > 0 && ( + + + + )} + +
{max}/min
+
+
+ {data[0]?.t || '—'} + now · {last?.n ?? 0}/min +
+ {buckets.length === 0 &&
No changes in the window yet…
} +
+ ) +} + +function BarRow({ label, value, max, color }: { label: string; value: number; max: number; color: string }) { + return ( +
+ + {label} + +
+
+
+ {value.toLocaleString()} +
+ ) +} + function Diff({ change }: { change: CdcChange }) { const keys = useMemo(() => { const set = new Set() @@ -82,19 +229,94 @@ export function ChangesView({ liveChanges }: { liveChanges: CdcChange[] }) { const [op, setOp] = useState('all') const [expanded, setExpanded] = useState(null) const [connected, setConnected] = useState(false) + 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 + // bottom feed instead of lagging behind it. + const emptyOverlay = { total: 0, by_op: {} as Record, by_source: {} as Record, by_table: {} as Record } + const [overlay, setOverlay] = useState(emptyOverlay) + const lastSeenId = useRef(null) + const primed = useRef(false) + + const applyStats = useCallback((s: CdcStats | null) => { + if (!s) return + setStats(s) + setOverlay({ total: 0, by_op: {}, by_source: {}, by_table: {} }) // server is now authoritative + }, []) const load = useCallback(async () => { - const [c, s] = await Promise.all([fetchChanges({ limit: 150 }), fetchChangeStats(15)]) + const [c, s] = await Promise.all([ + fetchChanges({ limit: 300, minutes: windowMin }), + fetchChangeStats(windowMin), + ]) setSeed(c.changes) setConnected(c.connected) - if (s) setStats(s) - }, []) + applyStats(s) + }, [applyStats, windowMin]) + + const doResync = useCallback(async () => { + setResyncing(true) + setResyncMsg('Restarting Debezium connectors…') + try { + const res = await resyncSources(true) + if (res) { + setResyncMsg(`Re-synced · ${res.healthy ?? 0}/${res.total ?? 0} connectors healthy`) + setTimeout(() => load(), 2500) + } else { + setResyncMsg('Re-sync failed — check ETL Guardian terminal') + } + } catch { + setResyncMsg('Re-sync failed — check ETL Guardian terminal') + } finally { + setResyncing(false) + setTimeout(() => setResyncMsg(null), 7000) + } + }, [load]) useEffect(() => { load() - const iv = setInterval(() => fetchChangeStats(15).then((s) => s && setStats(s)), 5000) - return () => clearInterval(iv) - }, [load]) + const iv = setInterval(() => fetchChangeStats(windowMin).then((s) => applyStats(s)), 2500) + const listIv = setInterval(() => { + fetchChanges({ limit: 300, minutes: windowMin }).then((c) => { + setSeed(c.changes) + setConnected(c.connected) + }) + }, 5000) + return () => { + clearInterval(iv) + clearInterval(listIv) + } + }, [load, applyStats, windowMin]) + + // Fold freshly-arrived WS changes into the overlay → instant top-of-page update. + useEffect(() => { + if (!liveChanges.length) return + if (!primed.current) { + primed.current = true + lastSeenId.current = liveChanges[0].id + return + } + const idx = liveChanges.findIndex((c) => c.id === lastSeenId.current) + const fresh = idx === -1 ? liveChanges : liveChanges.slice(0, idx) + if (!fresh.length) return + lastSeenId.current = liveChanges[0].id + setOverlay((o) => { + const next = { total: o.total + fresh.length, by_op: { ...o.by_op }, by_source: { ...o.by_source }, by_table: { ...o.by_table } } + for (const c of fresh) { + next.by_op[c.op] = (next.by_op[c.op] || 0) + 1 + next.by_source[c.source] = (next.by_source[c.source] || 0) + 1 + next.by_table[c.table] = (next.by_table[c.table] || 0) + 1 + } + return next + }) + setFlash(true) + const id = setTimeout(() => setFlash(false), 800) + return () => clearTimeout(id) + }, [liveChanges]) // Merge live (WS) with seeded backlog, dedupe by id, newest first. const merged = useMemo(() => { @@ -109,78 +331,152 @@ export function ChangesView({ liveChanges }: { liveChanges: CdcChange[] }) { [merged, source, op], ) - const maxBucket = Math.max(1, ...(stats?.buckets || []).map((b) => b.n)) + const opVal = (k: string) => (stats?.by_op?.[k] || 0) + (overlay.by_op[k] || 0) + const inserts = opVal('insert') + const updates = opVal('update') + const deletes = opVal('delete') + const total = (stats?.total ?? 0) + overlay.total + const perMin = (stats?.rate_per_min ?? total / Math.max(1, windowMin)) + (overlay.total / Math.max(1, windowMin)) + const windowLabel = TIME_WINDOWS.find((w) => w.minutes === windowMin)?.label ?? `${windowMin}m` + + const opSegments = useMemo(() => ( + ['insert', 'update', 'delete', 'snapshot'] + .map((k) => ({ label: k, value: (stats?.by_op?.[k] || 0) + (overlay.by_op[k] || 0), color: opOf(k).color })) + .filter((s) => s.value > 0) + ), [stats?.by_op, overlay]) + + const sourceRows = useMemo(() => { + const m: Record = { ...(stats?.by_source || {}) } + for (const [k, v] of Object.entries(overlay.by_source)) m[k] = (m[k] || 0) + v + return Object.entries(m).sort((a, b) => b[1] - a[1]) + }, [stats?.by_source, overlay]) + const maxSource = Math.max(1, ...sourceRows.map(([, v]) => v)) + + const tableRows = useMemo(() => { + const m: Record = { ...(stats?.by_table || {}) } + for (const [k, v] of Object.entries(overlay.by_table)) m[k] = (m[k] || 0) + v + return Object.entries(m).sort((a, b) => b[1] - a[1]).slice(0, 7) + }, [stats?.by_table, overlay]) + const maxTable = Math.max(1, ...tableRows.map(([, v]) => v)) + + // Volume chart: bump the current-minute bar with the live overlay so the curve + // visibly rises as changes stream in. + const liveBuckets = useMemo(() => { + const b = (stats?.buckets || []).map((x) => ({ ...x })) + if (overlay.total) { + if (b.length) b[b.length - 1] = { ...b[b.length - 1], n: b[b.length - 1].n + overlay.total } + else b.push({ t: 'now', n: overlay.total }) + } + return b + }, [stats?.buckets, overlay.total]) return ( -
+
{/* Header */} -
+

- Live Changes · CDC Stream + New & Changed Data

- Real-time Debezium change data capture from all source databases via Kafka + Live Debezium change data capture — every insert, update & delete across all source databases, streamed via Kafka

+
+ {TIME_WINDOWS.map((w) => ( + + ))} +
{connected ? 'STREAMING' : 'OFFLINE'} + {resyncMsg && ( + {resyncMsg} + )} +
- {/* Stat cards */} + {/* KPI row */}
-
-
Changes / 15 min
-
{stats?.total ?? 0}
-
-
-
Total consumed
-
{stats?.consumed ?? 0}
-
-
-
By operation
-
- {Object.entries(stats?.by_op || {}).map(([k, v]) => ( - {opOf(k).label} {v} - ))} - {!Object.keys(stats?.by_op || {}).length && —} + + + + +
+ + {/* Volume + operation mix */} +
+
+
+ Change volume — last {windowLabel} + {windowMin >= 120 && (hourly buckets)}
+
-
By source
-
- {Object.entries(stats?.by_source || {}).map(([k, v]) => ( - - {k} {v} - - ))} - {!Object.keys(stats?.by_source || {}).length && —} +
+ Operation mix
+ {opSegments.length ? : ( +
No changes yet…
+ )}
- {/* Volume sparkbars */} -
-
Change volume per minute (last 15m)
-
- {(stats?.buckets || []).map((b) => ( -
-
-
- ))} - {!(stats?.buckets || []).length &&
No changes in the window yet…
} + {/* By system + top tables */} +
+
+
+ New & changed by system +
+
+ {sourceRows.length ? sourceRows.map(([k, v]) => ( + + )) :
No source activity in the window…
} +
+
+
+
+ Most active tables / collections +
+
+ {tableRows.length ? tableRows.map(([k, v]) => ( + + )) :
No table activity yet…
} +
{/* Filters */}
+ Latest changes + Source {SOURCES.map((s) => (
{/* Live list */} -
+
{filtered.length === 0 && (
- Waiting for changes… trigger data generation or agent DML to see live CDC events. + Waiting for changes… trigger data generation (Data Flow → Generate data) or agent DML to see live CDC events.
)} {filtered.map((c) => ( diff --git a/ui/src/lib/api.ts b/ui/src/lib/api.ts index 3be3787..ece2906 100644 --- a/ui/src/lib/api.ts +++ b/ui/src/lib/api.ts @@ -100,21 +100,34 @@ export async function fetchPresentation(): Promise { return fetchJson('/api/presentation', 60000) } -export async function fetchChanges(opts: { source?: string; op?: string; limit?: number } = {}) { +export async function fetchChanges(opts: { source?: string; op?: string; limit?: number; minutes?: number } = {}) { const p = new URLSearchParams() if (opts.source) p.set('source', opts.source) if (opts.op) p.set('op', opts.op) - p.set('limit', String(opts.limit ?? 150)) - const j = await fetchJson<{ changes?: CdcChange[]; connected?: boolean; consumed?: number }>( + p.set('limit', String(opts.limit ?? 200)) + if (opts.minutes) p.set('minutes', String(opts.minutes)) + const j = await fetchJson<{ changes?: CdcChange[]; connected?: boolean; consumed?: number; last_ts?: string | null }>( `/api/changes?${p.toString()}`, 8000, ) - return { changes: j?.changes || [], connected: !!j?.connected, consumed: j?.consumed || 0 } + return { changes: j?.changes || [], connected: !!j?.connected, consumed: j?.consumed || 0, last_ts: j?.last_ts } } export async function fetchChangeStats(minutes = 15): Promise { return fetchJson(`/api/changes/stats?minutes=${minutes}`, 8000) } +export type ConnectorState = { name: string; state?: string; failed?: number[] } +export type ResyncResult = { ok: boolean; restarted?: string[]; healthy?: number; total?: number; after?: ConnectorState[] } + +export async function resyncSources(force = true): Promise { + const r = await fetch('/api/pipeline/streaming/resync', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ force }), + }) + return r.ok ? ((await r.json()) as ResyncResult) : null +} + export async function fetchAgentOpsStatus() { return fetchJson>('/api/agent-ops/status', 8000) } diff --git a/ui/src/lib/infraCatalog.ts b/ui/src/lib/infraCatalog.ts index 62839a3..ec623d1 100644 --- a/ui/src/lib/infraCatalog.ts +++ b/ui/src/lib/infraCatalog.ts @@ -168,6 +168,7 @@ export const INFRA_CATALOG: InfraNode[] = [ ssh: 'ssh root@10.0.20.106', apps: [ { label: 'GPU Lab UI', url: 'http://10.0.20.106:9000', port: '9000' }, + { label: 'Dockhand env 8', url: 'http://10.0.21.45:8082', port: '8082' }, { label: 'vLLM API', url: 'http://10.0.20.106:8001/v1', port: '8001' }, ], topoIds: ['llm', 'cons-ml'], diff --git a/ui/src/types.ts b/ui/src/types.ts index 673b609..baae9bd 100644 --- a/ui/src/types.ts +++ b/ui/src/types.ts @@ -62,12 +62,19 @@ export type CdcStats = { ok: boolean window_minutes: number total: number + inserts?: number + updates?: number + deletes?: number + rate_per_min?: number by_source: Record by_op: Record by_table: Record buckets: { t: string; n: number }[] connected: boolean consumed: number + buffered?: number + buffer_cap?: number + last_ts?: string | null } export type PiiColumn = { name: string; category: string; masked: boolean; policy_locked?: boolean }