From 88ca338f5a8564d9ffdab274d6af09c1b14a76b8 Mon Sep 17 00:00:00 2001 From: mo Date: Fri, 26 Jun 2026 08:51:39 +0000 Subject: [PATCH] feat: neo4j in topology + agent-driven datagen with activity log; fix all-sources generate --- api/pipeline_ops.py | 105 +++++++++++++++++- ui/src/components/features/DataGenView.tsx | 68 ++++++++++-- .../components/features/PlatformTopology.tsx | 8 +- ui/src/lib/infraCatalog.ts | 2 +- 4 files changed, 172 insertions(+), 11 deletions(-) diff --git a/api/pipeline_ops.py b/api/pipeline_ops.py index 37812da..555d92f 100644 --- a/api/pipeline_ops.py +++ b/api/pipeline_ops.py @@ -35,6 +35,23 @@ SOURCE_DAG = { "hadoop": "gen_hadoop_history", } +# UI source key -> the agent responsible for that part of the platform +SOURCE_AGENT = { + "postgres": "data-custodian", + "mysql": "data-custodian", + "mongodb": "data-custodian", + "cassandra": "data-custodian", + "neo4j": "data-custodian", + "all": "data-custodian", + "hadoop": "hadoop-ranger", +} +AGENT_NAME = { + "data-custodian": "Data Custodian", + "hadoop-ranger": "Hadoop Ranger", + "etl-guardian": "ETL Guardian", + "lakehouse-ops": "Lakehouse Ops", +} + # UI source key -> Trino fully-qualified table for live row counts SOURCE_COUNT_SQL = { "postgres": "SELECT count(*) FROM postgres_sales.public.sales_orders", @@ -85,6 +102,43 @@ async def _airflow_token(client: httpx.AsyncClient) -> str: return tok +def _feed(agent_id: str, message: str, level: str = "info") -> None: + """Write an entry to the shared agent activity feed (Comms log).""" + try: + from main import add_feed # lazy: main is fully loaded by request time + add_feed(agent_id, message, level) + except Exception: + pass + + +async def _watch_run(source: str, dag_id: str, run_id: str, agent_id: str, rows: int | None) -> None: + """Poll an Airflow run to completion and log the outcome to the feed.""" + name = AGENT_NAME.get(agent_id, agent_id) + label = f"{rows} rijen" if rows else "data" + try: + async with httpx.AsyncClient() as client: + tok = await _airflow_token(client) + headers = {"Authorization": f"Bearer {tok}"} + for _ in range(180): # up to ~15 min + await asyncio.sleep(5) + try: + r = await client.get( + f"{AIRFLOW_URL}/api/v2/dags/{dag_id}/dagRuns/{run_id}", + headers=headers, timeout=10, + ) + state = r.json().get("state") + except Exception: + continue + if state == "success": + _feed(agent_id, f"[datagen] {name} genereerde {label} in {source} — klaar, data stroomt via CDC naar Kafka/S3", "info") + return + if state == "failed": + _feed(agent_id, f"[datagen] {name}: generatie voor {source} is mislukt (zie Airflow logs)", "err") + return + except Exception: + pass + + async def _trino_scalar(sql: str, deadline_s: float = 8.0) -> int | None: """Run a scalar Trino query with a hard wall-clock deadline. @@ -138,6 +192,9 @@ async def generate(source: str, body: dict[str, Any] = Body(default={})) -> JSON conf["rows"] = max(1, min(int(rows), 2_000_000)) except (TypeError, ValueError): return JSONResponse({"ok": False, "error": "rows must be an integer"}, status_code=400) + agent_id = body.get("agent_id") or SOURCE_AGENT.get(source, "data-custodian") + autonomous = bool(body.get("autonomous")) + name = AGENT_NAME.get(agent_id, agent_id) try: async with httpx.AsyncClient() as client: tok = await _airflow_token(client) @@ -148,20 +205,66 @@ async def generate(source: str, body: dict[str, Any] = Body(default={})) -> JSON timeout=15, ) if r.status_code >= 400: + _feed(agent_id, f"[datagen] {name}: kon generatie voor {source} niet starten (Airflow {r.status_code})", "err") return JSONResponse({"ok": False, "error": f"Airflow {r.status_code}: {r.text[:300]}"}, status_code=200) j = r.json() + run_id = j.get("dag_run_id") + verb = "genereert zelf" if autonomous else "startte generatie:" + rows_txt = f"{conf['rows']} rijen" if conf.get("rows") else "data" + _feed(agent_id, f"[datagen] {name} {verb} {rows_txt} in {source}", "info") + if run_id: + asyncio.create_task(_watch_run(source, dag_id, run_id, agent_id, conf.get("rows"))) return JSONResponse({ "ok": True, "source": source, "dag_id": dag_id, - "run_id": j.get("dag_run_id"), + "run_id": run_id, "state": j.get("state"), "rows": conf.get("rows"), + "agent_id": agent_id, + "agent_name": name, }) except Exception as exc: return JSONResponse({"ok": False, "error": str(exc)}, status_code=200) +@router.get("/agents") +async def agents() -> JSONResponse: + """Which agent is responsible for generating each source.""" + out = {src: {"agent_id": aid, "agent_name": AGENT_NAME.get(aid, aid)} + for src, aid in SOURCE_AGENT.items()} + return JSONResponse({"ok": True, "agents": out}) + + +@router.get("/activity") +async def activity(limit: int = Query(25)) -> JSONResponse: + """Recent data-generation activity performed by agents (from the feed).""" + try: + from main import FeedEntry + from db import SessionLocal + from sqlalchemy import select + with SessionLocal() as db: + rows = db.execute( + select(FeedEntry).order_by(FeedEntry.ts.desc()).limit(400) + ).scalars().all() + items = [] + for r in rows: + if r.message and "[datagen]" in r.message: + items.append({ + "id": r.id, + "ts": r.ts.isoformat() if r.ts else None, + "agent_id": r.agent_id, + "agent_name": AGENT_NAME.get(r.agent_id, r.agent_id), + "message": r.message.replace("[datagen] ", ""), + "level": r.level, + }) + if len(items) >= limit: + break + return JSONResponse({"ok": True, "activity": items}) + except Exception as exc: + return JSONResponse({"ok": False, "error": str(exc), "activity": []}, status_code=200) + + @router.get("/runs/{source}") async def runs(source: str, limit: int = Query(5)) -> JSONResponse: dag_id = SOURCE_DAG.get(source) diff --git a/ui/src/components/features/DataGenView.tsx b/ui/src/components/features/DataGenView.tsx index a49ef85..b35cf9c 100644 --- a/ui/src/components/features/DataGenView.tsx +++ b/ui/src/components/features/DataGenView.tsx @@ -1,5 +1,5 @@ import { useCallback, useEffect, useRef, useState } from 'react' -import { Database, Boxes, Activity, Cpu, Network, Layers, Play, Loader2, CheckCircle2, XCircle, RefreshCw } from 'lucide-react' +import { Database, Boxes, Activity, Cpu, Network, Layers, Play, Loader2, CheckCircle2, XCircle, RefreshCw, Bot, ScrollText } from 'lucide-react' import { cn } from '../../lib/utils' import { subTabActive, subTabIdle } from '../../lib/tabActive' @@ -26,6 +26,8 @@ const SOURCES: SourceMeta[] = [ ] type RunInfo = { run_id?: string; state?: string; start?: string; end?: string; conf?: { rows?: number } } +type AgentInfo = { agent_id: string; agent_name: string } +type ActivityItem = { id: string; ts?: string; agent_id: string; agent_name: string; message: string; level: string } type Props = { onPulse: () => void @@ -43,6 +45,8 @@ export function DataGenView({ onPulse, onOpenPlatform }: Props) { const [runs, setRuns] = useState>({}) const [counts, setCounts] = useState>({}) const [msg, setMsg] = useState(null) + const [agentMap, setAgentMap] = useState>({}) + const [activity, setActivity] = useState([]) const pollRef = useRef>>({}) const meta = SOURCES.find((s) => s.key === active)! @@ -55,6 +59,14 @@ export function DataGenView({ onPulse, onOpenPlatform }: Props) { } catch { /* */ } }, []) + const loadActivity = useCallback(async () => { + try { + const r = await fetch('/api/pipeline/activity?limit=25') + const j = await r.json() + if (j.ok) setActivity(j.activity || []) + } catch { /* */ } + }, []) + const loadRuns = useCallback(async (source: SourceKey) => { try { const r = await fetch(`/api/pipeline/runs/${source}?limit=5`) @@ -68,9 +80,13 @@ export function DataGenView({ onPulse, onOpenPlatform }: Props) { useEffect(() => { loadCounts() + loadActivity() SOURCES.forEach((s) => loadRuns(s.key)) - return () => { Object.values(pollRef.current).forEach(clearInterval) } - }, [loadCounts, loadRuns]) + fetch('/api/pipeline/agents').then((r) => r.json()).then((j) => { if (j.ok) setAgentMap(j.agents || {}) }).catch(() => {}) + const act = setInterval(loadActivity, 8000) + const poll = pollRef.current + return () => { Object.values(poll).forEach(clearInterval); clearInterval(act) } + }, [loadCounts, loadRuns, loadActivity]) const startPolling = useCallback((source: SourceKey) => { if (pollRef.current[source]) clearInterval(pollRef.current[source]) @@ -87,7 +103,7 @@ export function DataGenView({ onPulse, onOpenPlatform }: Props) { }, 3000) }, [loadRuns, loadCounts]) - const generate = useCallback(async (source: SourceKey) => { + const generate = useCallback(async (source: SourceKey, autonomous = false) => { setMsg(null) setBusy((b) => ({ ...b, [source]: true })) onPulse() @@ -95,7 +111,7 @@ export function DataGenView({ onPulse, onOpenPlatform }: Props) { const r = await fetch(`/api/pipeline/generate/${source}`, { method: 'POST', headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ rows: rows[source] }), + body: JSON.stringify({ rows: rows[source], autonomous }), }) const j = await r.json() if (!j.ok) { @@ -103,13 +119,15 @@ export function DataGenView({ onPulse, onOpenPlatform }: Props) { setBusy((b) => ({ ...b, [source]: false })) return } - setMsg(`${source}: gestart (run ${String(j.run_id).slice(-8)})`) + const who = autonomous ? `${j.agent_name || 'Agent'} genereert zelf` : 'gestart' + setMsg(`${source}: ${who} (run ${String(j.run_id).slice(-8)})`) startPolling(source) + setTimeout(loadActivity, 800) } catch { setMsg('API niet bereikbaar') setBusy((b) => ({ ...b, [source]: false })) } - }, [rows, onPulse, startPolling]) + }, [rows, onPulse, startPolling, loadActivity]) const stateBadge = (state?: string) => { if (state === 'success') return success @@ -170,6 +188,11 @@ export function DataGenView({ onPulse, onOpenPlatform }: Props) { ) : ( geen CDC-stream )} + {agentMap[active] && ( + + {agentMap[active].agent_name} + + )}

{meta.desc} Doel: {meta.target}

@@ -194,6 +217,16 @@ export function DataGenView({ onPulse, onOpenPlatform }: Props) { {busy[active] ? : } Genereer data + {active !== 'all' && COUNT_KEYS.includes(active) && (
Huidige rijen: {counts[active]?.toLocaleString() ?? '…'} @@ -242,6 +275,27 @@ export function DataGenView({ onPulse, onOpenPlatform }: Props) { ))}
+ +
+

+ Agent-activiteit (wat de agents deden) +

+ {activity.length === 0 ? ( +

Nog geen agent-acties gelogd.

+ ) : ( +
    + {activity.map((a) => ( +
  • + {a.ts?.slice(11, 19) || ''} + + {a.agent_name} + + {a.message} +
    • + ))} +
+ )} +
diff --git a/ui/src/components/features/PlatformTopology.tsx b/ui/src/components/features/PlatformTopology.tsx index 4e87bfc..6a3c3bb 100644 --- a/ui/src/components/features/PlatformTopology.tsx +++ b/ui/src/components/features/PlatformTopology.tsx @@ -34,6 +34,7 @@ const STAGES: TopoStage[] = [ { id: 'mysql', label: 'MySQL', sub: 'Replica set', metricKey: 'mysql' }, { id: 'mongodb', label: 'MongoDB', sub: 'Document store', metricKey: 'mongodb' }, { id: 'cassandra', label: 'Cassandra', sub: 'Wide-column', metricKey: 'cassandra' }, + { id: 'neo4j', label: 'Neo4j', sub: 'Graph store', metricKey: 'neo4j' }, ], }, { @@ -78,11 +79,13 @@ const FLOW_EDGES: FlowEdge[] = [ { from: 'airflow', to: 'mysql', kind: 'orchestration', label: 'Daily Python gen' }, { from: 'airflow', to: 'mongodb', kind: 'orchestration', label: 'Daily Python gen' }, { from: 'airflow', to: 'cassandra', kind: 'orchestration', label: 'Daily Python gen' }, + { from: 'airflow', to: 'neo4j', kind: 'orchestration', label: 'Daily Python gen' }, // CDC capture from sources { from: 'postgresql', to: 'debezium', kind: 'cdc', label: 'CDC' }, { from: 'mysql', to: 'debezium', kind: 'cdc', label: 'CDC' }, { from: 'mongodb', to: 'debezium', kind: 'cdc', label: 'CDC' }, { from: 'cassandra', to: 'debezium', kind: 'cdc', label: 'CDC' }, + { from: 'neo4j', to: 'debezium', kind: 'cdc', label: 'CDC' }, // Streaming bus { from: 'debezium', to: 'kafka', kind: 'stream', label: 'Events' }, { from: 'airflow', to: 'kafka', kind: 'orchestration', label: 'DAG trigger' }, @@ -144,7 +147,7 @@ const PARTICLE_FILL: Record = { } const NODE_CLICK_MAP: Record = { - postgresql: 'src-postgres', mysql: 'src-mysql', mongodb: 'src-mongo', cassandra: 'src-cassandra', + postgresql: 'src-postgres', mysql: 'src-mysql', mongodb: 'src-mongo', cassandra: 'src-cassandra', neo4j: 'src-neo4j', debezium: 'cdc-postgres', kafka: 'stream-kafka', airflow: 'src-airflow', spark: 'stream-spark', trino: 'query-trino', iceberg: 'lake-iceberg', s3: 'lake-s3', hadoop: 'hadoop', bi: 'cons-bi', jupyter: 'cons-notebooks', elasticsearch: 'cons-elastic', kibana: 'cons-kibana', llm: 'cons-ml', @@ -187,7 +190,7 @@ type MetricState = Record function seedMetrics(): MetricState { return { - postgresql: '12.4k rows/s', mysql: '8.1k rows/s', mongodb: '2.3k docs/s', cassandra: '5.6k ops/s', + postgresql: '12.4k rows/s', mysql: '8.1k rows/s', mongodb: '2.3k docs/s', cassandra: '5.6k ops/s', neo4j: '1.4k nodes/s', debezium: '4 connectors active', kafka: '142 MB/s', airflow: '18 DAGs · daily 02:00 UTC', spark: '6 executors live', trino: '3 queries active', iceberg: '847 tables · 2.1 TB', s3: '14.2 TB stored', hadoop: 'Historical archive', @@ -235,6 +238,7 @@ function jitterMetric(key: string, current: string, workload: WorkloadData | nul mysql: () => `${(8.1 + n() * 0.6).toFixed(1)}k rows/s`, mongodb: () => `${(2.3 + n() * 0.3).toFixed(1)}k docs/s`, cassandra: () => `${(5.6 + n() * 0.5).toFixed(1)}k ops/s`, + neo4j: () => `${(1.4 + n() * 0.3).toFixed(1)}k nodes/s`, debezium: () => `${Math.max(3, Math.round(4 + n()))} connectors active`, kafka: () => `${Math.max(80, Math.round(142 + n() * 18))} MB/s`, airflow: () => `${Math.max(12, Math.round(18 + n() * 2))} DAGs · daily 02:00 UTC`, diff --git a/ui/src/lib/infraCatalog.ts b/ui/src/lib/infraCatalog.ts index 1b09d02..62839a3 100644 --- a/ui/src/lib/infraCatalog.ts +++ b/ui/src/lib/infraCatalog.ts @@ -47,7 +47,7 @@ export const INFRA_CATALOG: InfraNode[] = [ { label: 'PostgreSQL', url: 'postgresql://10.0.21.51:5432/postgres', port: '5432' }, { label: 'MongoDB', url: 'mongodb://10.0.21.51:27017/', port: '27017' }, ], - topoIds: ['postgresql', 'mysql', 'mongodb', 'cassandra', 'src-postgres', 'src-mysql', 'src-mongo', 'src-cassandra'], + topoIds: ['postgresql', 'mysql', 'mongodb', 'cassandra', 'neo4j', 'src-postgres', 'src-mysql', 'src-mongo', 'src-cassandra', 'src-neo4j'], }, { id: 'airflow',