From 213350ec75531e01c3057df90326c9ca568799ab Mon Sep 17 00:00:00 2001 From: mo Date: Sun, 28 Jun 2026 22:06:27 +0000 Subject: [PATCH] feat: Generate-data button + generation-script viewer + vector DB explorer Data Flow tab: - Prominent "Generate data" button (500 / 2K / 10K) that inserts a fresh burst of business rows into all source DBs on demand via a new POST /api/federated/generate (fresh connections, safe alongside the background streamer); result toast shows what was inserted, CDC streams it. - "Scripts" button + a "View generation scripts" action on the Data Generator node open a modal listing every generator script with full source, served by GET /api/dataflow/scripts. Sources are the real files: the live streaming generator (sliced live out of trino_federated.py) and the Airflow per-source DAGs + Faker scripts (mounted read-only from infra/airflow into the API). Knowledge Chat: - New "Vector DB" explorer modal: shows the ChromaDB chunking config (RecursiveCharacterTextSplitter 800/120, all-MiniLM-L6-v2, 384-dim, HNSW), collections & documents, and the actual stored chunks with text, metadata and an embedding preview (bars + values) so you can see exactly how files are split and written as vectors. Refactor: generator row-builders shared by the streamer and the on-demand burst. --- api/dataflow.py | 74 ++++++++ api/trino_federated.py | 159 ++++++++++++++--- docker-compose.yml | 1 + ui/src/components/features/DataFlowView.tsx | 151 +++++++++++++++- .../components/features/KnowledgeChatView.tsx | 168 ++++++++++++++++++ 5 files changed, 528 insertions(+), 25 deletions(-) diff --git a/api/dataflow.py b/api/dataflow.py index 0d45116..d4e1391 100644 --- a/api/dataflow.py +++ b/api/dataflow.py @@ -14,6 +14,7 @@ from __future__ import annotations import os import time +from pathlib import Path from typing import Any import httpx @@ -300,6 +301,79 @@ async def get_dataflow(refresh: bool = False) -> JSONResponse: return JSONResponse(data) +GEN_SCRIPTS_DIR = os.getenv("GEN_SCRIPTS_DIR", "/app/gen_scripts") + +# Which generation scripts to surface, in display order. `file` is relative to +# GEN_SCRIPTS_DIR (the mounted infra/airflow dir); `live` slices the running +# streamer source straight out of trino_federated.py so it is always in sync. +_SCRIPT_SPECS: list[dict[str, Any]] = [ + {"id": "live", "title": "Live streaming generator", "engine": "Command Center API", + "desc": "Runs inside this API. While the Live dashboard is open it streams randomly-sized bursts of rows into PostgreSQL, MySQL, MongoDB & Cassandra every few seconds (and powers the manual 'Generate data' button). CDC propagates everything downstream.", + "live": True}, + {"id": "dag", "title": "Airflow per-source DAGs", "engine": "Apache Airflow", + "desc": "One triggerable DAG per database. Each passes a row count via dag_run conf and shells out to the matching generator script below.", + "file": "per_source_gen_dags.py"}, + {"id": "postgres", "title": "PostgreSQL — sales orders", "engine": "Faker → psycopg2", + "desc": "Generates realistic customers, products, regions, channels & amounts (incl. the PII columns that the masking layer later protects).", + "file": "scripts/generate_postgres_sales_data.py"}, + {"id": "mysql", "title": "MySQL — employee events", "engine": "Faker → PyMySQL", + "desc": "HR lifecycle events (hire, promotion, salary change, …) with employee PII.", + "file": "scripts/generate_mysql_employee_data.py"}, + {"id": "mongodb", "title": "MongoDB — supply events", "engine": "Faker → PyMongo", + "desc": "Schemaless supply-chain events with free-form payloads.", + "file": "scripts/generate_mongodb_events_data.py"}, + {"id": "cassandra", "title": "Cassandra — device telemetry", "engine": "Faker → cassandra-driver", + "desc": "High-volume IoT device metrics (temperature, voltage, …) on a time-series schema.", + "file": "scripts/generate_cassandra_telemetry_data.py"}, + {"id": "neo4j", "title": "Neo4j — product & supplier graph", "engine": "Faker → neo4j driver", + "desc": "Product/supplier nodes and relationships for the graph database.", + "file": "scripts/generate_neo4j_graph_data.py"}, +] + + +def _live_generator_source() -> str: + try: + text = Path("/app/trino_federated.py").read_text(encoding="utf-8") + except Exception: + return "# live generator source unavailable" + start = text.find("# Continuous live generator") + end = text.find('@router.get("/live")', start if start >= 0 else 0) + if start >= 0 and end > start: + return text[start:end].rstrip() + return "# live generator source unavailable" + + +_scripts_cache: dict[str, Any] = {"ts": 0.0, "data": None} + + +@router.get("/scripts") +async def get_scripts() -> JSONResponse: + now = time.time() + if _scripts_cache["data"] and now - _scripts_cache["ts"] < 30: + return JSONResponse(_scripts_cache["data"]) + base = Path(GEN_SCRIPTS_DIR) + scripts = [] + for spec in _SCRIPT_SPECS: + src = "" + if spec.get("live"): + src = _live_generator_source() + else: + p = base / spec["file"] + try: + src = p.read_text(encoding="utf-8") + except Exception as exc: + src = f"# source unavailable ({exc})" + scripts.append({ + "id": spec["id"], "title": spec["title"], "engine": spec["engine"], + "desc": spec["desc"], "filename": spec.get("file", "trino_federated.py"), + "language": "python", "lines": src.count("\n") + 1, "source": src, + }) + data = {"ok": True, "scripts": scripts} + _scripts_cache["data"] = data + _scripts_cache["ts"] = now + return JSONResponse(data) + + @router.post("/{movement_id}/run") async def run_dataflow_movement(movement_id: str, body: dict[str, Any] = Body(default={})) -> JSONResponse: try: diff --git a/api/trino_federated.py b/api/trino_federated.py index 9533baf..973375e 100644 --- a/api/trino_federated.py +++ b/api/trino_federated.py @@ -392,7 +392,17 @@ def _gen_reset(key: str): _gen_conns[key] = None -def _gen_orders(n: int): +# Row builders — shared by the background streamer and the on-demand "Generate +# data" button, so both produce identical, realistic business rows. +_PG_INSERT = ("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)") +_MYSQL_INSERT = ("INSERT INTO employee_events " + "(employee_id,department,role_name,region,event_type,salary_change,event_ts) " + "VALUES (%s,%s,%s,%s,%s,%s,%s)") + + +def _order_rows(n: int): import datetime as dt now = dt.datetime.utcnow() by_r: dict[str, int] = {} @@ -408,49 +418,133 @@ def _gen_orders(n: int): 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) + return rows, by_r, by_s, round(val, 2) -def _gen_hr(n: int): +def _hr_rows(n: int): import datetime as dt now = dt.datetime.utcnow() - rows = [(_rnd.randint(1, 100000), _rnd.choice(_DEPTS), _rnd.choice(_ROLES), + return [(_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): +def _supply_docs(n: int): import datetime as dt import uuid now = dt.datetime.utcnow() - docs = [{"event_id": str(uuid.uuid4()), "type": _rnd.choice(_SUPPLY), + return [{"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)] + + +def _tel_rows(n: int): + import datetime as dt + now = dt.datetime.utcnow() + return [(f"device-{_rnd.randint(1, 99999)}", now, _rnd.choice(_METRICS), + round(_rnd.uniform(0, 100), 3), "") for _ in range(n)] + + +_TEL_INSERT_TPL = ("INSERT INTO {ks}.device_metrics " + "(device_id, metric_ts, metric_type, metric_value, payload) VALUES (%s,%s,%s,%s,%s)") + + +def _gen_orders(n: int): + rows, by_r, by_s, val = _order_rows(n) + _gen_pg().cursor().executemany(_PG_INSERT, rows) + return by_r, by_s, val + + +def _gen_hr(n: int): + _gen_mysql().cursor().executemany(_MYSQL_INSERT, _hr_rows(n)) + + +def _gen_supply(n: int): + docs = _supply_docs(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), "")) + sess = _gen_cass() + cql = _TEL_INSERT_TPL.format(ks=s.CASS_KS) + for row in _tel_rows(n): + sess.execute(cql, row) + + +def _generate_once(orders: int, hr: int, supply: int, tel: int) -> dict[str, Any]: + """On-demand burst using FRESH short-lived connections (safe to run from a + request thread alongside the background streamer). Returns inserted counts.""" + import sql_console as s + out = {"orders": 0, "hr_events": 0, "supply_events": 0, "telemetry": 0} + by_r: dict[str, int] = {} + by_s: dict[str, int] = {} + val = 0.0 + if orders > 0: + try: + import psycopg2 + rows, by_r, by_s, val = _order_rows(orders) + 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=8) + try: + c.autocommit = True + c.cursor().executemany(_PG_INSERT, rows) + out["orders"] = orders + finally: + c.close() + except Exception: + pass + if hr > 0: + try: + import pymysql + 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=8, autocommit=True) + try: + c.cursor().executemany(_MYSQL_INSERT, _hr_rows(hr)) + out["hr_events"] = hr + finally: + c.close() + except Exception: + pass + if supply > 0: + try: + cli = s._mongo_client() + try: + cli[s.MONGO_DB]["events"].insert_many(_supply_docs(supply)) + out["supply_events"] = supply + finally: + cli.close() + except Exception: + pass + if tel > 0: + try: + cluster = s._cass_cluster() + sess = cluster.connect() + try: + cql = _TEL_INSERT_TPL.format(ks=s.CASS_KS) + for row in _tel_rows(tel): + sess.execute(cql, row) + out["telemetry"] = tel + finally: + cluster.shutdown() + except Exception: + pass + # fold into the live counters + feed so the dashboard reflects it instantly + with _gen_lock: + c = _GEN["counts"] + for k in out: + c[k] += out[k] + if out["orders"]: + _GEN["by_region"] = by_r + _GEN["by_status"] = by_s + _GEN["tick_value"] = val + top = max(by_r, key=by_r.get) if by_r else "—" + _GEN["feed"].appendleft({ + "ts": datetime.now(timezone.utc).isoformat(), + "text": f"⚡ manual burst: +{out['orders']} orders · €{int(val):,} · top {top} · +{out['telemetry']} telemetry · +{out['hr_events']} HR · +{out['supply_events']} supply", + }) + out["revenue"] = val + return out def _gen_tick(): @@ -528,6 +622,23 @@ async def toggle_generator(body: dict = Body(default={})): return {"ok": True, "enabled": _GEN["enabled"], "interval": _GEN["interval"], "running": _GEN["running"]} +@router.post("/generate") +async def generate_now(body: dict = Body(default={})): + """Manual one-shot burst into the source systems (the Data Flow "Generate + data" button). `rows` controls the order volume; the other sources scale + with it. CDC streams everything downstream automatically.""" + from starlette.concurrency import run_in_threadpool + rows = int(body.get("rows", 500) or 500) + rows = max(1, min(20000, rows)) + orders = rows + hr = max(1, rows // 4) + supply = max(1, rows // 4) + tel = max(1, rows // 2) + out = await run_in_threadpool(_generate_once, orders, hr, supply, tel) + total = out["orders"] + out["hr_events"] + out["supply_events"] + out["telemetry"] + return {"ok": True, "requested": rows, "inserted": out, "total": total} + + @router.get("/live") async def get_live(): import sql_console as s diff --git a/docker-compose.yml b/docker-compose.yml index 514e257..3dc7022 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -55,6 +55,7 @@ services: volumes: - api_data:/data - /root/.ssh:/root/.ssh:ro + - ./infra/airflow:/app/gen_scripts:ro depends_on: redis: condition: service_started diff --git a/ui/src/components/features/DataFlowView.tsx b/ui/src/components/features/DataFlowView.tsx index 5a45fd8..dd2153f 100644 --- a/ui/src/components/features/DataFlowView.tsx +++ b/ui/src/components/features/DataFlowView.tsx @@ -1,5 +1,5 @@ import { useCallback, useEffect, useLayoutEffect, useMemo, useRef, useState } from 'react' -import { GitBranch, Lock, LockOpen, Play, Pause, Square, RefreshCw, ShieldAlert, ShieldCheck, Loader2, Bot } from 'lucide-react' +import { GitBranch, Lock, LockOpen, Play, Pause, Square, RefreshCw, ShieldAlert, ShieldCheck, Loader2, Bot, Zap, Code2, X, FileCode2 } from 'lucide-react' import type { DataflowEdge, DataflowGraph, DataflowNode } from '../../types' import { fetchDataflow, runDataflowMovement, toggleEtlAgent, toggleCustodianOffload, fetchAgentOpsStatus, setPiiMask, setStreamingFlow } from '../../lib/api' import { SparkKafkaPanel } from './SparkKafkaPanel' @@ -85,6 +85,10 @@ export function DataFlowView() { const [triggering, setTriggering] = useState(null) const [etlEnabled, setEtlEnabled] = useState(null) const [custEnabled, setCustEnabled] = useState(null) + const [genRows, setGenRows] = useState(2000) + const [genBusy, setGenBusy] = useState(false) + const [genToast, setGenToast] = useState(null) + const [showScripts, setShowScripts] = useState(false) const canvasRef = useRef(null) const nodeRefs = useRef>({}) @@ -181,6 +185,26 @@ export function DataFlowView() { setTimeout(() => setRefreshing(false), 400) }, [load]) + const onGenerate = useCallback(async () => { + setGenBusy(true) + setGenToast(null) + try { + const r = await fetch('/api/federated/generate', { + method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ rows: genRows }), + }) + const d = await r.json() + const i = d?.inserted || {} + setGenToast(`Generated +${(i.orders ?? 0).toLocaleString()} orders · +${(i.hr_events ?? 0).toLocaleString()} HR · +${(i.supply_events ?? 0).toLocaleString()} supply · +${(i.telemetry ?? 0).toLocaleString()} telemetry → streaming via CDC`) + setTimeout(() => load(true), 700) + setTimeout(() => setGenToast(null), 7000) + } catch { + setGenToast('Generation failed — check the API logs') + setTimeout(() => setGenToast(null), 5000) + } finally { + setGenBusy(false) + } + }, [genRows, load]) + const flowMode = (graph as unknown as { flow?: string })?.flow ?? 'running' const onFlow = useCallback(async (action: 'pause' | 'resume' | 'stop') => { await setStreamingFlow(action) @@ -282,6 +306,34 @@ export function DataFlowView() { > Hadoop offload {custEnabled == null ? '' : custEnabled ? 'on' : 'off'} +
+ + +
+
+ {genToast && ( +
+ {genToast} +
+ )} + + {showScripts && setShowScripts(false)} onGenerate={onGenerate} genBusy={genBusy} />} + {/* canvas */}
{loading && ( @@ -402,6 +462,20 @@ export function DataFlowView() { {(selNode.id === 'spark' || selNode.id === 'kafka') && (

See Spark/Kafka panel below for full UI + job control.

)} + {selNode.id === 'generator' && ( +
+

+ Generates realistic business rows into every source database. Stream it live with “Generate data”, or open the actual scripts. +

+ +
+ )} {selNode.pii?.has_pii ? (
@@ -557,3 +631,78 @@ function NodeCard({ ) } + +type GenScript = { id: string; title: string; engine: string; desc: string; filename: string; language: string; lines: number; source: string } + +function ScriptsModal({ onClose, onGenerate, genBusy }: { onClose: () => void; onGenerate: () => void; genBusy: boolean }) { + const [scripts, setScripts] = useState([]) + const [active, setActive] = useState(null) + const [loading, setLoading] = useState(true) + const [err, setErr] = useState(false) + useEffect(() => { + let alive = true + fetch('/api/dataflow/scripts') + .then((r) => r.json()) + .then((d) => { + if (!alive) return + const list: GenScript[] = d?.scripts || [] + setScripts(list) + setActive(list[0]?.id ?? null) + setLoading(false) + }) + .catch(() => { if (alive) { setErr(true); setLoading(false) } }) + return () => { alive = false } + }, []) + const sel = scripts.find((s) => s.id === active) || null + return ( +
+
e.stopPropagation()}> +
+
+ +
+

Data generation scripts

+

How the Data Generator builds & writes data into every source system

+
+
+
+ + +
+
+ {loading ? ( +
loading scripts…
+ ) : err ? ( +
Could not load scripts.
+ ) : ( +
+
+ {scripts.map((s) => ( + + ))} +
+
+ {sel && ( + <> +
+

{sel.filename}

+

{sel.desc}

+
+ 
{sel.source}
+ + )} +
+
+ )} +
+
+ ) +} diff --git a/ui/src/components/features/KnowledgeChatView.tsx b/ui/src/components/features/KnowledgeChatView.tsx index 83bee39..71d2f98 100644 --- a/ui/src/components/features/KnowledgeChatView.tsx +++ b/ui/src/components/features/KnowledgeChatView.tsx @@ -42,6 +42,7 @@ export function KnowledgeChatView({ onGpuActivity }: Props = {}) { const [activeStage, setActiveStage] = useState(null) const [traceCfg, setTraceCfg] = useState<{ tracing: boolean; project: string; smith_url: string } | null>(null) const [showTraces, setShowTraces] = useState(false) + const [showVdb, setShowVdb] = useState(false) const bottomRef = useRef(null) const loadMeta = useCallback(async () => { @@ -362,6 +363,14 @@ export function KnowledgeChatView({ onGpuActivity }: Props = {}) { )} + +
+ + {/* chunking / embedding config strip */} + {cfg && ( +
+ {cfg.splitter} + chunk {cfg.chunk_size} · overlap {cfg.chunk_overlap} + {cfg.embed_model} + {cfg.embed_dim}-dim + {cfg.index} + {totalVec.toLocaleString()} vectors total +
+ )} + + {loading ? ( +
loading vector store…
+ ) : err ? ( +
{err}
+ ) : ( +
+ {/* collections + documents */} +
+

Collections

+ {cols.map((c) => ( + + ))} + {selCol && selCol.files.length > 0 && ( + <> +

Documents

+ + {selCol.files.map((f) => ( + + ))} + + )} +
+ + {/* chunk viewer */} +
+
+ Showing chunks for {col} + {docId ? <> · doc {docId} : ' · all documents'} + — each card is one vector row in ChromaDB +
+
+ {loadingChunks ? ( +
loading chunks…
+ ) : chunks.length === 0 ? ( +

No chunks in this selection.

+ ) : chunks.map((ch) => ( +
+
+ chunk #{ch.chunk_index ?? '—'} + {ch.chars} chars · ~{ch.tokens_est} tokens + {typeof ch.metadata?.source === 'string' && {ch.metadata.source as string}} + + + {ch.embedding_dim}-dim + +
+

{ch.text}

+

vector id: {ch.id} · embedding [{ch.embedding_preview.slice(0, 6).map((v) => v.toFixed(3)).join(', ')}, …]

+
+ ))} +
+
+
+ )} +
+
+ ) +}