From dfd5d4da8a74bfa3d8a14be6e44390db12f61496 Mon Sep 17 00:00:00 2001 From: mo Date: Sun, 28 Jun 2026 22:41:48 +0000 Subject: [PATCH] feat: pulse generator edges on data gen + New & Changed Data dashboard Generated data is now visible flowing into the source systems on the Data Flow graph: a manual "Generate data" burst (and the continuous live loop) stamps a last_tick, and a new generator_active() helper lights up the generator -> postgres/mysql/mongodb/cassandra edges for ~12s. Neo4j stays dark since the live generator does not write to it. Redesigned the Changes tab into a "New & Changed Data" dashboard driven by the live CDC stream: animated KPI cards (new/updated/deleted + throughput), a change-volume area chart, an operation-mix donut, per-system and top-table breakdown bars, plus the existing filterable change feed. --- api/dataflow.py | 12 +- api/trino_federated.py | 10 + ui/src/components/features/ChangesView.tsx | 283 +++++++++++++++++---- 3 files changed, 254 insertions(+), 51 deletions(-) diff --git a/api/dataflow.py b/api/dataflow.py index d4e1391..4ce62fd 100644 --- a/api/dataflow.py +++ b/api/dataflow.py @@ -173,6 +173,11 @@ async def _build() -> dict[str, Any]: cdc = cdc_snapshot(15) except Exception: cdc = {"connected": False, "consumed": 0, "by_source": {}, "window_total": 0} + try: + from trino_federated import generator_active + gen_active = generator_active() + except Exception: + gen_active = False try: from pii_catalog import get_pii pii = get_pii() @@ -255,7 +260,12 @@ async def _build() -> dict[str, Any]: edge["last_rows"] = lr.get("rows") edge["last_duration_s"] = lr.get("duration_s") edge["active"] = lr.get("state") == "running" - if e["kind"] == "cdc": + if e["kind"] == "generate": + # The live generator (continuous loop or the 'Generate data' burst) + # writes into these four sources; pulse the edge while it is active. + if e["to"] in ("postgres", "mysql", "mongodb", "cassandra"): + edge["active"] = gen_active or bool(edge.get("active")) + elif e["kind"] == "cdc": edge["active"] = cdc.get("by_source", {}).get(e["from"], 0) > 0 elif e.get("from") == "hdfs" and e.get("to") == "kafka": edge["active"] = bool(edge_live.get("hdfs→kafka")) diff --git a/api/trino_federated.py b/api/trino_federated.py index d817266..8ffe1aa 100644 --- a/api/trino_federated.py +++ b/api/trino_federated.py @@ -534,6 +534,9 @@ def _generate_once(orders: int, hr: int, supply: int, tel: int) -> dict[str, Any c = _GEN["counts"] for k in out: c[k] += out[k] + _GEN["last_batch"] = {"orders": out["orders"], "hr_events": out["hr_events"], + "supply_events": out["supply_events"], "telemetry": out["telemetry"]} + _GEN["last_tick"] = time.time() # marks recent activity → Data Flow CDC edges pulse if out["orders"]: _GEN["by_region"] = by_r _GEN["by_status"] = by_s @@ -609,6 +612,13 @@ def _gen_loop(): threading.Thread(target=_gen_loop, daemon=True, name="live-generator").start() +def generator_active(window_s: float = 12.0) -> bool: + """True when the live generator (continuous loop OR a manual 'Generate data' + burst) wrote rows very recently. The Data Flow graph uses this to pulse the + generator→source edges so generated data is visible flowing into the sources.""" + return (time.time() - float(_GEN.get("last_tick") or 0.0)) < window_s + + @router.post("/live/generator") async def toggle_generator(body: dict = Body(default={})): if "enabled" in body: diff --git a/ui/src/components/features/ChangesView.tsx b/ui/src/components/features/ChangesView.tsx index 5af3563..36d0cd1 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 { useCallback, useEffect, useMemo, useRef, useState } from 'react' +import { Activity, Radio, RefreshCw, PlusCircle, Pencil, Trash2, Database, Layers, TrendingUp } from 'lucide-react' import { fetchChanges, fetchChangeStats } 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 = { @@ -23,7 +23,7 @@ const SOURCES = ['all', 'postgres', 'mysql', 'mongodb', 'cassandra', 'neo4j'] const OPS = ['all', 'insert', 'update', 'delete'] 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 +34,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,6 +223,8 @@ 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 prevTotal = useRef(0) const load = useCallback(async () => { const [c, s] = await Promise.all([fetchChanges({ limit: 150 }), fetchChangeStats(15)]) @@ -92,10 +235,22 @@ export function ChangesView({ liveChanges }: { liveChanges: CdcChange[] }) { useEffect(() => { load() - const iv = setInterval(() => fetchChangeStats(15).then((s) => s && setStats(s)), 5000) + const iv = setInterval(() => fetchChangeStats(15).then((s) => s && setStats(s)), 4000) return () => clearInterval(iv) }, [load]) + // Pulse the header when fresh changes arrive. + useEffect(() => { + const t = stats?.total ?? 0 + if (t > prevTotal.current) { + setFlash(true) + const id = setTimeout(() => setFlash(false), 900) + prevTotal.current = t + return () => clearTimeout(id) + } + prevTotal.current = t + }, [stats?.total]) + // Merge live (WS) with seeded backlog, dedupe by id, newest first. const merged = useMemo(() => { const byId = new Map() @@ -109,18 +264,39 @@ export function ChangesView({ liveChanges }: { liveChanges: CdcChange[] }) { [merged, source, op], ) - const maxBucket = Math.max(1, ...(stats?.buckets || []).map((b) => b.n)) + const byOp = stats?.by_op || {} + const inserts = byOp.insert || 0 + const updates = byOp.update || 0 + const deletes = byOp.delete || 0 + const total = stats?.total ?? 0 + const perMin = total / Math.max(1, stats?.window_minutes ?? 15) + + const opSegments = useMemo(() => ( + ['insert', 'update', 'delete', 'snapshot'] + .map((k) => ({ label: k, value: byOp[k] || 0, color: opOf(k).color })) + .filter((s) => s.value > 0) + ), [byOp]) + + const sourceRows = useMemo(() => ( + Object.entries(stats?.by_source || {}).sort((a, b) => b[1] - a[1]) + ), [stats?.by_source]) + const maxSource = Math.max(1, ...sourceRows.map(([, v]) => v)) + + const tableRows = useMemo(() => ( + Object.entries(stats?.by_table || {}).sort((a, b) => b[1] - a[1]).slice(0, 7) + ), [stats?.by_table]) + const maxTable = Math.max(1, ...tableRows.map(([, v]) => v)) 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

@@ -134,53 +310,60 @@ export function ChangesView({ liveChanges }: { liveChanges: CdcChange[] }) {
- {/* 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 15 minutes
+
-
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) => (