diff --git a/api/dataflow.py b/api/dataflow.py index e29a19f..0d45116 100644 --- a/api/dataflow.py +++ b/api/dataflow.py @@ -52,6 +52,11 @@ NODES: list[dict[str, Any]] = [ # governance — bottom centre {"id": "openmetadata", "label": "OpenMetadata", "sub": "catalog · lineage · PII", "kind": "governance", "x": 47, "y": 93, "url": "http://10.0.21.47:8585"}, + # AI serving lane — how governed data reaches the LLM & the Command Center chat + {"id": "chromadb", "label": "ChromaDB", "sub": "vectors · embeddings", "kind": "vector", "x": 60, "y": 65}, + {"id": "rag", "label": "RAG · LangChain", "sub": "retrieve · augment · agent", "kind": "rag", "x": 74, "y": 65}, + {"id": "vllm", "label": "vLLM Gateway", "sub": "Llama3-70B · GPT-4o", "kind": "llm", "x": 88, "y": 68}, + {"id": "chat", "label": "Knowledge Chat", "sub": "Command Center", "kind": "chat", "x": 92, "y": 90}, ] # Edges. movement_id (optional) links to movements.py so the edge is triggerable. @@ -86,6 +91,14 @@ EDGES: list[dict[str, Any]] = [ {"from": "cassandra", "to": "openmetadata", "kind": "catalog"}, {"from": "neo4j", "to": "openmetadata", "kind": "catalog"}, {"from": "trino", "to": "openmetadata", "kind": "catalog"}, + # AI serving lane: governed business data + catalog + vectors → RAG → vLLM → chat + {"from": "trino", "to": "rag", "kind": "context"}, + {"from": "openmetadata", "to": "rag", "kind": "context"}, + {"from": "iceberg_curated", "to": "rag", "kind": "context"}, + {"from": "chromadb", "to": "rag", "kind": "retrieve"}, + {"from": "rag", "to": "vllm", "kind": "prompt"}, + {"from": "vllm", "to": "chat", "kind": "answer"}, + {"from": "rag", "to": "chat", "kind": "answer"}, ] _cache: dict[str, Any] = {"ts": 0.0, "data": None} @@ -117,6 +130,25 @@ def _trino_scalar(sql: str, deadline_s: float = 8.0) -> int | None: return None +RAG_URL = os.getenv("RAG_URL", "http://rag-api:5020").rstrip("/") +_rag_cache: dict[str, Any] = {"ts": 0.0, "val": None} + + +def _rag_info() -> dict[str, Any]: + now = time.time() + if _rag_cache["val"] is not None and now - _rag_cache["ts"] < 60: + return _rag_cache["val"] + info: dict[str, Any] = {} + try: + with httpx.Client(timeout=2.0) as client: + info = client.get(f"{RAG_URL}/config").json() or {} + except Exception: + info = {} + _rag_cache["val"] = info + _rag_cache["ts"] = now + return info + + def _iceberg_hadoop_count() -> int | None: now = time.time() if _count_cache["val"] is not None and now - _count_cache["ts"] < 60: @@ -180,6 +212,19 @@ async def _build() -> dict[str, Any]: metric = f"{c:,} rows" if c is not None else "iceberg table" elif n["id"] == "generator": metric = "Airflow gen DAGs" + elif n["id"] in ("vllm", "rag", "chromadb"): + info = _rag_info() + model = info.get("llm_model") or "gpt-4o" + embed = (info.get("embed_model") or "all-MiniLM-L6-v2").split("/")[-1] + if n["id"] == "vllm": + metric = f"{model} · OpenAI-compat" + elif n["id"] == "rag": + metric = f"LangChain · {embed}" + else: + metric = f"embeddings · {embed}" + node["level"] = "ok" if info else "warn" + elif n["id"] == "chat": + metric = "RAG chat · agent mode" # PII overlay p = pii_by_node.get(n["id"]) if p: @@ -219,6 +264,9 @@ async def _build() -> dict[str, Any]: edge["active"] = bool(edge_live.get("spark→iceberg")) or edge.get("active") elif e.get("from") == "spark" and e.get("to") == "s3_cdc": edge["active"] = bool(edge_live.get("spark→s3")) or edge.get("active") + elif e["kind"] in ("context", "retrieve", "prompt", "answer"): + # AI serving lane pulses while governed data is being served to the LLM + edge["active"] = bool(_rag_info()) if e.get("offload"): try: from agent_ops import custodian_recent diff --git a/api/trino_federated.py b/api/trino_federated.py index 46100fe..9533baf 100644 --- a/api/trino_federated.py +++ b/api/trino_federated.py @@ -20,7 +20,7 @@ from datetime import datetime, timezone from pathlib import Path from typing import Any -from fastapi import APIRouter +from fastapi import APIRouter, Body from fastapi.responses import JSONResponse router = APIRouter(prefix="/api/federated", tags=["federated"]) @@ -297,45 +297,299 @@ def _live_business_aggs() -> dict[str, Any]: return data +# ────────────────────────────────────────────────────────────────────────────── +# Continuous live generator — keeps the platform "alive": while the Live +# dashboard is open it streams small, randomly-sized batches of business rows +# into the real source databases (PostgreSQL / MySQL / MongoDB / Cassandra), +# which CDC then propagates downstream. Batch sizes fluctuate every tick so the +# throughput visibly goes up and down. It only runs while someone is watching +# (the /live poll refreshes a heartbeat) so the tables don't grow unbounded. +# ────────────────────────────────────────────────────────────────────────────── +import random as _rnd +from collections import deque as _deque + +_GEN: dict[str, Any] = { + "enabled": True, + "running": False, + "interval": 4.0, + "last_seen": 0.0, + "last_tick": 0.0, + "counts": {"orders": 0, "hr_events": 0, "supply_events": 0, "telemetry": 0}, + "last_batch": {"orders": 0, "hr_events": 0, "supply_events": 0, "telemetry": 0}, + "by_region": {}, + "by_status": {}, + "tick_value": 0.0, + "feed": _deque(maxlen=14), + "base": None, +} +_gen_lock = threading.Lock() +_gen_conns: dict[str, Any] = {"pg": None, "mysql": None, "mongo": None, "cass": None} + +_REGIONS = ["NA", "EU", "APAC", "LATAM", "EMEA", "MEA"] +_CHANNELS = ["B2B", "B2C", "ONLINE", "PARTNER", "RETAIL"] +_STATUSES = ["NEW", "PAID", "SHIPPED", "DELIVERED", "RETURNED", "CANCELLED"] +_CURR = ["EUR", "USD", "GBP", "JPY"] +_DEPTS = ["Engineering", "Sales", "Support", "Operations", "Finance", "HR", "Marketing"] +_ROLES = ["Analyst", "Engineer", "Manager", "Lead", "Specialist", "Director"] +_EVT = ["HIRE", "PROMOTION", "SALARY_CHANGE", "TRANSFER", "REVIEW", "EXIT"] +_SUPPLY = ["INSERT", "UPDATE", "REPLENISH", "SHIPMENT", "RETURN"] +_SRC = ["CRM", "ERP", "WMS", "API"] +_METRICS = ["temperature", "humidity", "pressure", "voltage", "current"] + + +def _gen_pg(): + import psycopg2 + import sql_console as s + c = _gen_conns["pg"] + if c is None or getattr(c, "closed", 1): + c = psycopg2.connect(host=s.DB_HOST, port=s.PG_PORT, user=s.PG_USER, + password=s.PG_PASS, dbname=s.PG_DB, connect_timeout=6) + c.autocommit = True + _gen_conns["pg"] = c + return c + + +def _gen_mysql(): + import pymysql + import sql_console as s + c = _gen_conns["mysql"] + if c is None: + c = pymysql.connect(host=s.DB_HOST, port=s.MYSQL_PORT, user=s.MYSQL_USER, + password=s.MYSQL_PASS, database=s.MYSQL_DB, connect_timeout=6, + autocommit=True) + _gen_conns["mysql"] = c + else: + c.ping(reconnect=True) + return c + + +def _gen_mongo(): + import sql_console as s + c = _gen_conns["mongo"] + if c is None: + c = s._mongo_client() + _gen_conns["mongo"] = c + return c[s.MONGO_DB] + + +def _gen_cass(): + import sql_console as s + sess = _gen_conns["cass"] + if sess is None: + cluster = s._cass_cluster() + sess = cluster.connect() + _gen_conns["cass"] = sess + return sess + + +def _gen_reset(key: str): + try: + c = _gen_conns.get(key) + if c is not None: + c.close() if key != "cass" else c.cluster.shutdown() + except Exception: + pass + _gen_conns[key] = None + + +def _gen_orders(n: int): + import datetime as dt + now = dt.datetime.utcnow() + by_r: dict[str, int] = {} + by_s: dict[str, int] = {} + val = 0.0 + rows = [] + for _ in range(n): + r = _rnd.choice(_REGIONS) + st = _rnd.choices(_STATUSES, weights=[5, 6, 5, 8, 2, 2])[0] + ch = _rnd.choice(_CHANNELS) + amt = round(_rnd.uniform(15, 9500), 2) + rows.append((_rnd.randint(1, 20000), _rnd.randint(1, 5000), r, ch, now, amt, _rnd.choice(_CURR), st)) + by_r[r] = by_r.get(r, 0) + 1 + by_s[st] = by_s.get(st, 0) + 1 + val += amt + cur = _gen_pg().cursor() + cur.executemany( + "INSERT INTO public.sales_orders " + "(customer_id,product_id,region,sales_channel,order_ts,amount,currency,order_status) " + "VALUES (%s,%s,%s,%s,%s,%s,%s,%s)", rows) + return by_r, by_s, round(val, 2) + + +def _gen_hr(n: int): + import datetime as dt + now = dt.datetime.utcnow() + rows = [(_rnd.randint(1, 100000), _rnd.choice(_DEPTS), _rnd.choice(_ROLES), + _rnd.choice(_REGIONS), _rnd.choice(_EVT), round(_rnd.uniform(-2000, 6000), 2), now) + for _ in range(n)] + cur = _gen_mysql().cursor() + cur.executemany( + "INSERT INTO employee_events " + "(employee_id,department,role_name,region,event_type,salary_change,event_ts) " + "VALUES (%s,%s,%s,%s,%s,%s,%s)", rows) + + +def _gen_supply(n: int): + import datetime as dt + import uuid + now = dt.datetime.utcnow() + docs = [{"event_id": str(uuid.uuid4()), "type": _rnd.choice(_SUPPLY), + "region": _rnd.choice(_REGIONS), "source": _rnd.choice(_SRC), + "amount": round(_rnd.uniform(10, 40000), 2), "ts": now.isoformat()} + for _ in range(n)] + if docs: + _gen_mongo()["events"].insert_many(docs) + + +def _gen_tel(n: int): + import datetime as dt + now = dt.datetime.utcnow() + sess = _gen_cass() + import sql_console as s + cql = (f"INSERT INTO {s.CASS_KS}.device_metrics " + "(device_id, metric_ts, metric_type, metric_value, payload) VALUES (%s,%s,%s,%s,%s)") + for _ in range(n): + sess.execute(cql, (f"device-{_rnd.randint(1, 99999)}", now, + _rnd.choice(_METRICS), round(_rnd.uniform(0, 100), 3), "")) + + +def _gen_tick(): + # fluctuating batch sizes, with the occasional spike, so throughput moves up & down + no = _rnd.randint(2, 40) + if _rnd.random() < 0.18: + no += _rnd.randint(25, 70) + nh = _rnd.randint(0, 18) + ns = _rnd.randint(0, 16) + nt = _rnd.randint(8, 55) + by_r: dict[str, int] = {} + by_s: dict[str, int] = {} + val = 0.0 + try: + by_r, by_s, val = _gen_orders(no) + except Exception: + _gen_reset("pg"); no = 0 + try: + _gen_hr(nh) + except Exception: + _gen_reset("mysql"); nh = 0 + try: + _gen_supply(ns) + except Exception: + _gen_reset("mongo"); ns = 0 + try: + _gen_tel(nt) + except Exception: + _gen_reset("cass"); nt = 0 + with _gen_lock: + c = _GEN["counts"] + c["orders"] += no + c["hr_events"] += nh + c["supply_events"] += ns + c["telemetry"] += nt + _GEN["last_batch"] = {"orders": no, "hr_events": nh, "supply_events": ns, "telemetry": nt} + _GEN["by_region"] = by_r + _GEN["by_status"] = by_s + _GEN["tick_value"] = val + _GEN["last_tick"] = time.time() + if no: + top = max(by_r, key=by_r.get) if by_r else "—" + _GEN["feed"].appendleft({ + "ts": datetime.now(timezone.utc).isoformat(), + "text": f"+{no} orders · €{int(val):,} · top {top} ({by_r.get(top, 0)}) · +{nt} telemetry · +{nh} HR", + }) + + +def _gen_loop(): + while True: + try: + if _GEN["enabled"] and (time.time() - _GEN["last_seen"] < 25): + _GEN["running"] = True + _gen_tick() + else: + _GEN["running"] = False + except Exception: + _GEN["running"] = False + time.sleep(max(2.0, float(_GEN["interval"]))) + + +threading.Thread(target=_gen_loop, daemon=True, name="live-generator").start() + + +@router.post("/live/generator") +async def toggle_generator(body: dict = Body(default={})): + if "enabled" in body: + _GEN["enabled"] = bool(body["enabled"]) + if "interval" in body: + try: + _GEN["interval"] = max(2.0, min(30.0, float(body["interval"]))) + except Exception: + pass + _GEN["last_seen"] = time.time() + return {"ok": True, "enabled": _GEN["enabled"], "interval": _GEN["interval"], "running": _GEN["running"]} + + @router.get("/live") async def get_live(): import sql_console as s - orders = s._table_row_count("postgres", "public.sales_orders") or 0 - # MySQL event_id is monotonic, so max(event_id) tracks inserts in real time - # (the planner estimate only refreshes after ANALYZE). - hr_res = _trino("SELECT max(event_id) FROM mysql_hr.hr.employee_events", 1) - hr = 0 - if hr_res.get("ok"): - try: - hr = int((hr_res.get("rows") or [[0]])[0][0] or 0) - except Exception: - hr = 0 - if not hr: - hr = s._table_row_count("mysql", "hr.employee_events") or 0 - supply = s._table_row_count("mongodb", "supplychain.events") or 0 + _GEN["last_seen"] = time.time() # heartbeat: keeps the generator running while watched + # Cassandra has no cheap estimate — reuse the exact count from the cached # federated matrix query when available. - telemetry = 0 if _marquee.get("data") is None: _load_marquee() mq = _marquee.get("data") or {} + cass_base = 0 for r in ((mq.get("matrix") or {}).get("rows") or []): if r.get("catalog") == "cassandra_telemetry": try: - telemetry = int(r.get("records") or 0) + cass_base = int(r.get("records") or 0) except Exception: - telemetry = 0 + cass_base = 0 + + # One-time base snapshot of source sizes; every subsequent reading is + # base + rows the generator has streamed in, so the counters move smoothly + # and in lock-step with the live activity feed. + with _gen_lock: + if _GEN["base"] is None: + _GEN["base"] = { + "orders": s._table_row_count("postgres", "public.sales_orders") or 0, + "hr_events": s._table_row_count("mysql", "hr.employee_events") or 0, + "supply_events": s._table_row_count("mongodb", "supplychain.events") or 0, + "telemetry": cass_base, + } + elif cass_base and not _GEN["base"].get("telemetry"): + _GEN["base"]["telemetry"] = cass_base + base = dict(_GEN["base"]) + gc = dict(_GEN["counts"]) + gen_view = { + "enabled": _GEN["enabled"], + "running": _GEN["running"], + "interval": _GEN["interval"], + "counts": dict(_GEN["counts"]), + "last_batch": dict(_GEN["last_batch"]), + "tick_value": _GEN["tick_value"], + "by_region": [{"key": k, "count": v} for k, v in sorted(_GEN["by_region"].items(), key=lambda kv: -kv[1])], + "by_status": [{"key": k, "count": v} for k, v in sorted(_GEN["by_status"].items(), key=lambda kv: -kv[1])], + "feed": list(_GEN["feed"]), + } + + orders = base["orders"] + gc["orders"] + hr = base["hr_events"] + gc["hr_events"] + supply = base["supply_events"] + gc["supply_events"] + telemetry = base["telemetry"] + gc["telemetry"] + avg_order = _avg_order_value() sources = [ - {"key": "orders", "label": "Orders", "engine": "PostgreSQL", "catalog": "postgres_sales", "rows": orders, "color": "#fbbf24"}, - {"key": "hr_events", "label": "HR events", "engine": "MySQL", "catalog": "mysql_hr", "rows": hr, "color": "#60a5fa"}, - {"key": "supply_events", "label": "Supply events", "engine": "MongoDB", "catalog": "mongodb_supplychain", "rows": supply, "color": "#a78bfa"}, - {"key": "telemetry", "label": "Telemetry", "engine": "Cassandra", "catalog": "cassandra_telemetry", "rows": telemetry, "color": "#22d3ee"}, + {"key": "orders", "label": "Orders", "engine": "PostgreSQL", "catalog": "postgres_sales", "rows": orders, "added": gc["orders"], "color": "#fbbf24"}, + {"key": "hr_events", "label": "HR events", "engine": "MySQL", "catalog": "mysql_hr", "rows": hr, "added": gc["hr_events"], "color": "#60a5fa"}, + {"key": "supply_events", "label": "Supply events", "engine": "MongoDB", "catalog": "mongodb_supplychain", "rows": supply, "added": gc["supply_events"], "color": "#a78bfa"}, + {"key": "telemetry", "label": "Telemetry", "engine": "Cassandra", "catalog": "cassandra_telemetry", "rows": telemetry, "added": gc["telemetry"], "color": "#22d3ee"}, ] return { "ok": True, "ts": datetime.now(timezone.utc).isoformat(), "sources": sources, + "generator": gen_view, "totals": { "records": orders + hr + supply + telemetry, "revenue_est": round(orders * avg_order, 2), diff --git a/ui/src/components/features/DataFlowView.tsx b/ui/src/components/features/DataFlowView.tsx index 8da37a3..5a45fd8 100644 --- a/ui/src/components/features/DataFlowView.tsx +++ b/ui/src/components/features/DataFlowView.tsx @@ -18,6 +18,10 @@ const NODE_KIND: Record = { compute: { ring: 'border-violet-400/60', chip: 'bg-violet-500/15 text-violet-300 border-violet-400/40', dot: '#a78bfa' }, engine: { ring: 'border-violet-400/60', chip: 'bg-violet-500/15 text-violet-300 border-violet-400/40', dot: '#a78bfa' }, governance: { ring: 'border-fuchsia-400/60', chip: 'bg-fuchsia-500/15 text-fuchsia-300 border-fuchsia-400/40', dot: '#d946ef' }, + vector: { ring: 'border-teal-400/60', chip: 'bg-teal-500/15 text-teal-300 border-teal-400/40', dot: '#2dd4bf' }, + rag: { ring: 'border-pink-400/60', chip: 'bg-pink-500/15 text-pink-300 border-pink-400/40', dot: '#f472b6' }, + llm: { ring: 'border-rose-400/70', chip: 'bg-rose-500/15 text-rose-200 border-rose-400/50', dot: '#fb7185' }, + chat: { ring: 'border-indigo-400/60', chip: 'bg-indigo-500/15 text-indigo-300 border-indigo-400/40', dot: '#818cf8' }, } const EDGE_COLOR: Record = { @@ -28,6 +32,10 @@ const EDGE_COLOR: Record = { mask: '#fb7185', query: '#818cf8', catalog: '#d946ef', + context: '#2dd4bf', + retrieve: '#f472b6', + prompt: '#fb7185', + answer: '#818cf8', } const EDGE_LEGEND: { kind: string; label: string }[] = [ @@ -38,6 +46,10 @@ const EDGE_LEGEND: { kind: string; label: string }[] = [ { kind: 'mask', label: 'PII masking' }, { kind: 'query', label: 'Query' }, { kind: 'catalog', label: 'Catalog (OpenMetadata)' }, + { kind: 'context', label: 'LLM context' }, + { kind: 'retrieve', label: 'Vector retrieve' }, + { kind: 'prompt', label: 'Prompt' }, + { kind: 'answer', label: 'Answer → chat' }, ] type Anchor = { x: number; y: number; w: number; h: number } diff --git a/ui/src/components/features/LiveDashboard.tsx b/ui/src/components/features/LiveDashboard.tsx index d1c5f02..50a349f 100644 --- a/ui/src/components/features/LiveDashboard.tsx +++ b/ui/src/components/features/LiveDashboard.tsx @@ -1,14 +1,26 @@ import { useCallback, useEffect, useRef, useState } from 'react' -import { Activity, Pause, Play, ShoppingCart, Users, Boxes, Cpu, DollarSign, Database, Gauge } from 'lucide-react' +import { Activity, Pause, Play, ShoppingCart, Users, Boxes, Cpu, DollarSign, Database, Gauge, Zap, Sparkles, Radio } from 'lucide-react' import { cn } from '../../lib/utils' type Bucket = { key: string; count: number; value?: number } -type Source = { key: string; label: string; engine: string; catalog: string; rows: number; color: string } +type Source = { key: string; label: string; engine: string; catalog: string; rows: number; added?: number; color: string } type RegionRow = { region: string; orders: number; revenue: number; hr_events: number; supply_events: number } +type Gen = { + enabled: boolean + running: boolean + interval: number + counts: { orders: number; hr_events: number; supply_events: number; telemetry: number } + last_batch: { orders: number; hr_events: number; supply_events: number; telemetry: number } + tick_value: number + by_region: Bucket[] + by_status: Bucket[] + feed: { ts: string; text: string }[] +} type Live = { ok: boolean ts: string sources: Source[] + generator?: Gen totals: { records: number; revenue_est: number; avg_order: number } business: { orders_by_region: Bucket[] @@ -174,9 +186,11 @@ export function LiveDashboard() { const [paused, setPaused] = useState(false) const [err, setErr] = useState(false) const [totalHist, setTotalHist] = useState([]) + const [ordHist, setOrdHist] = useState([]) const [rateBySrc, setRateBySrc] = useState>({}) const [added, setAdded] = useState(0) - const prev = useRef<{ ts: number; rows: Record; total: number } | null>(null) + const [genBusy, setGenBusy] = useState(false) + const prev = useRef<{ ts: number; rows: Record; total: number; genOrders: number } | null>(null) const startTotal = useRef(null) const poll = useCallback(async () => { @@ -187,10 +201,13 @@ export function LiveDashboard() { setErr(false) const now = Date.parse(d.ts) || Date.now() const total = d.totals.records + const genOrders = d.generator?.counts.orders ?? 0 if (prev.current) { const dt = Math.max(0.5, (now - prev.current.ts) / 1000) const totRate = Math.max(0, (total - prev.current.total) / dt) setTotalHist((h) => [...h, totRate].slice(-90)) + // orders added between polls — fluctuates up and down with each batch + setOrdHist((h) => [...h, Math.max(0, genOrders - prev.current!.genOrders)].slice(-60)) const rmap: Record = {} d.sources.forEach((s) => { const p = prev.current!.rows[s.key] ?? s.rows @@ -200,13 +217,23 @@ export function LiveDashboard() { } if (startTotal.current == null) startTotal.current = total setAdded(Math.max(0, total - (startTotal.current || total))) - prev.current = { ts: now, rows: Object.fromEntries(d.sources.map((s) => [s.key, s.rows])), total } + prev.current = { ts: now, rows: Object.fromEntries(d.sources.map((s) => [s.key, s.rows])), total, genOrders } setData(d) } catch { setErr(true) } }, []) + const toggleGen = useCallback(async (enabled: boolean) => { + setGenBusy(true) + try { + await fetch('/api/federated/live/generator', { + method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ enabled }), + }) + await poll() + } catch { /* ignore */ } finally { setGenBusy(false) } + }, [poll]) + useEffect(() => { poll() if (paused) return @@ -215,7 +242,10 @@ export function LiveDashboard() { }, [poll, paused]) const b = data?.business + const gen = data?.generator + const lb = gen?.last_batch const totalRate = totalHist.length ? totalHist[totalHist.length - 1] : 0 + const ordPeak = Math.max(1, ...ordHist) const matrix = b?.region_matrix || [] const maxRev = Math.max(1, ...matrix.map((m) => Number(m.revenue) || 0)) @@ -236,7 +266,20 @@ export function LiveDashboard() { · +{fmtNum(added)} since opened {err && reconnecting…} + {gen && ( + + stream {gen.running ? `every ${gen.interval}s` : 'idle'} + + )} {data ? `updated ${new Date(data.ts).toLocaleTimeString()}` : 'connecting…'} + {gen && ( + + )} @@ -251,6 +294,52 @@ export function LiveDashboard() { ))} + {/* live activity — the per-tick pulse: orders & events streaming into the sources right now */} + {gen && ( +
+ +
+ {([ + { k: 'orders', label: 'orders', color: '#fbbf24', icon: ShoppingCart }, + { k: 'telemetry', label: 'telemetry', color: '#22d3ee', icon: Cpu }, + { k: 'hr_events', label: 'HR', color: '#60a5fa', icon: Users }, + { k: 'supply_events', label: 'supply', color: '#a78bfa', icon: Boxes }, + ] as const).map(({ k, label, color, icon: Icon }) => { + const v = lb ? (lb as Record)[k] : 0 + return ( +
+ {label} + +{fmtNum(v)} +
+ ) + })} +
+ +
+ orders per {(POLL_MS / 1000).toFixed(1)}s tick — watch it rise & fall + peak {fmtNum(ordPeak)} · €{fmtNum(gen.tick_value)} last burst +
+ + +
+ {(gen.feed || []).length ? gen.feed.map((f, i) => ( +
+ {new Date(f.ts).toLocaleTimeString()} · {f.text} +
+ )) :

waiting for the next burst…

} +
+ +
+ )} + + {/* orders by region this burst (changes every tick) */} + {gen && (gen.by_region?.length || gen.by_status?.length) ? ( +
+ + +
+ ) : null} + {/* throughput + per-source rates */}
@@ -326,7 +415,7 @@ export function LiveDashboard() {

- Counters & throughput are live source estimates (Trino over PostgreSQL, MySQL, MongoDB & Cassandra); breakdown charts aggregate the materialized Hadoop lake. Polling every {POLL_MS / 1000}s. + While this tab is open a live generator streams randomly-sized bursts of real rows into PostgreSQL, MySQL, MongoDB & Cassandra (picked up by CDC) — counters & the activity feed move every {gen?.interval ?? 4}s; the region scorecard & breakdown charts aggregate the materialized Hadoop lake. Polling every {POLL_MS / 1000}s.

)