From b6d7d3dc74d2f6792c089df632ef57273cd6052d Mon Sep 17 00:00:00 2001 From: mo Date: Sun, 28 Jun 2026 23:33:21 +0000 Subject: [PATCH] feat: realtime ETL offload to S3 + live business dashboards - etl_offload.py: autonomous agent backfills/tails source DBs (PG/MySQL/ Mongo/Cassandra) to S3 as Parquet in small chunks, accumulates a live federated business matrix (/api/etl/status, /api/etl/business, /run, /config). - storage_s3.py: buffer generated CDC + masked curated rows to S3, overlay live last-write into analytics; put_object_bytes for Parquet parts. - trino_federated.py: capture generated rows + archive to S3; generator_active. - dataflow.py: pulse generate + kafka/spark->S3 archive edges when active. - StorageView: realtime ETL ingest panel; TrinoFederationView: realtime business KPIs/charts from /api/etl/business. - ChangesView: top KPIs/charts now overlay the live WS stream on server stats so they update in lock-step with the bottom feed; faster 2.5s refresh. - useCommandCenter: retain 800 live CDC changes. --- api/Dockerfile | 2 +- api/dataflow.py | 10 +- api/etl_offload.py | 556 ++++++++++++++++++ api/main.py | 2 + api/requirements.txt | 1 + api/storage_s3.py | 160 ++++- api/trino_federated.py | 55 +- ui/src/components/features/ChangesView.tsx | 100 +++- ui/src/components/features/StorageView.tsx | 117 +++- .../features/TrinoFederationView.tsx | 102 ++++ ui/src/hooks/useCommandCenter.ts | 2 +- 11 files changed, 1062 insertions(+), 45 deletions(-) create mode 100644 api/etl_offload.py diff --git a/api/Dockerfile b/api/Dockerfile index 89a259d..6014407 100644 --- a/api/Dockerfile +++ b/api/Dockerfile @@ -4,7 +4,7 @@ WORKDIR /app RUN apt-get update && apt-get install -y --no-install-recommends curl && rm -rf /var/lib/apt/lists/* COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt -COPY main.py lab_context.py agent_terminal.py workload.py node_registry.py node_ops.py topology_views.py supervisor.py approval_service.py db.py dockhand_envs.py presentation.py database_inventory.py presentation_upload.py presentation_static.py storage_s3.py elasticsearch_api.py sql_console.py hdfs_api.py ssh_terminal.py pipeline_ops.py hadoop_analytics.py agent_ops.py cdc_consumer.py movements.py dataflow.py streaming_ops.py spark_workbench.py hadoop_sql.py hdfs_kafka.py webhdfs_util.py pii_catalog.py platform_context.py trino_federated.py hive_bench_seed.json . +COPY main.py lab_context.py agent_terminal.py workload.py node_registry.py node_ops.py topology_views.py supervisor.py approval_service.py db.py dockhand_envs.py presentation.py database_inventory.py presentation_upload.py presentation_static.py storage_s3.py elasticsearch_api.py sql_console.py hdfs_api.py ssh_terminal.py pipeline_ops.py hadoop_analytics.py agent_ops.py cdc_consumer.py movements.py dataflow.py streaming_ops.py spark_workbench.py hadoop_sql.py hdfs_kafka.py webhdfs_util.py pii_catalog.py platform_context.py trino_federated.py etl_offload.py hive_bench_seed.json . RUN mkdir -p /data ENV DATABASE_URL=sqlite:////data/atc-agents.db EXPOSE 3201 diff --git a/api/dataflow.py b/api/dataflow.py index 4ce62fd..3092011 100644 --- a/api/dataflow.py +++ b/api/dataflow.py @@ -178,6 +178,11 @@ async def _build() -> dict[str, Any]: gen_active = generator_active() except Exception: gen_active = False + try: + from storage_s3 import archive_active + arch_active = archive_active() + except Exception: + arch_active = False try: from pii_catalog import get_pii pii = get_pii() @@ -274,7 +279,10 @@ async def _build() -> dict[str, Any]: elif e.get("from") == "spark" and e.get("to") == "iceberg_curated": 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") + edge["active"] = bool(edge_live.get("spark→s3")) or edge.get("active") or arch_active + elif e["kind"] == "archive" and e.get("from") == "kafka" and e.get("to") == "s3_cdc": + # Kafka → S3 CDC archive pulses while the pipeline lands objects in S3 + edge["active"] = arch_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()) diff --git a/api/etl_offload.py b/api/etl_offload.py new file mode 100644 index 0000000..1aeca0d --- /dev/null +++ b/api/etl_offload.py @@ -0,0 +1,556 @@ +"""Autonomous ETL offload agent — source databases → S3 Parquet lake. + +A background agent ("Lakehouse Loader") that, on a fixed cadence, pulls the next +small chunk of rows from every source database and writes it to the S3 object +store as a partitioned Parquet part: + + lake//dt=YYYY-MM-DD/part-.parquet + +It progressively *backfills* the entire history in small chunks (so a 50M-row +table lands as thousands of small files) and then *tails* newly generated rows, +so the object store keeps filling and the analytics stay realtime. While +offloading it also accumulates a live federated business matrix (revenue by +region/status/channel, HR by dept, supply by type, telemetry by metric, …) +straight from the rows it actually moved — which therefore always reflects the +newest generated data, and powers the realtime Trino / Object-store dashboards. + +Endpoints: + GET /api/etl/status -> per-dataset progress, ingest rate, recent parts, feed + GET /api/etl/business -> accumulated realtime federated business matrix + POST /api/etl/run -> trigger one offload cycle now + POST /api/etl/config -> {enabled, interval_s, chunk} +""" + +from __future__ import annotations + +import io +import json +import random +import threading +import time +from collections import deque +from datetime import datetime, timezone +from typing import Any + +from fastapi import APIRouter, Body +from fastapi.responses import JSONResponse + +router = APIRouter(prefix="/api/etl", tags=["etl"]) + +DATASETS = [ + {"key": "orders", "label": "Sales orders", "engine": "PostgreSQL", "color": "#fbbf24"}, + {"key": "hr_events", "label": "HR events", "engine": "MySQL", "color": "#60a5fa"}, + {"key": "supply_events", "label": "Supply events", "engine": "MongoDB", "color": "#a78bfa"}, + {"key": "telemetry", "label": "Device telemetry", "engine": "Cassandra", "color": "#22d3ee"}, +] + +_lock = threading.Lock() +_state: dict[str, Any] = { + "enabled": True, + "interval_s": 60.0, + "chunk": 5000, + "running_cycle": False, + "started_at": None, + "last_cycle_ts": 0.0, + "last_cycle_rows": 0, + "next_run_ts": 0.0, + "cycles": 0, + "datasets": { + d["key"]: {"label": d["label"], "engine": d["engine"], "color": d["color"], + "parts": 0, "rows": 0, "bytes": 0, "backfilled": False, "cursor": None, + "keycol": None, "mode": None, "total_source": None, "last_ts": None, + "last_rows": 0, "last_key": None, "error": None} + for d in DATASETS + }, + "feed": deque(maxlen=60), + "minute": deque(maxlen=60), # (minute_epoch, rows, bytes) + "series": deque(maxlen=48), # per-cycle points for the realtime chart +} + +# Accumulated federated business matrix, built from the rows we actually offload. +_business: dict[str, Any] = { + "orders": {"count": 0, "revenue": 0.0, "by_region": {}, "by_status": {}, "by_channel": {}, "by_currency": {}}, + "hr": {"count": 0, "by_department": {}, "by_event": {}, "by_region": {}}, + "supply": {"count": 0, "amount": 0.0, "by_type": {}, "by_region": {}, "by_source": {}}, + "telemetry": {"count": 0, "by_metric": {}}, + "ts": deque(maxlen=48), # {t, orders, revenue, rows} +} + + +# ── helpers ─────────────────────────────────────────────────────────────────── +def _now() -> datetime: + return datetime.now(timezone.utc) + + +def _bump(d: dict, k: Any, n: int = 1, val: float | None = None) -> None: + if k is None or k == "": + return + k = str(k) + if val is None: + d[k] = d.get(k, 0) + n + else: + cur = d.get(k) or {"count": 0, "value": 0.0} + cur["count"] += n + cur["value"] += val + d[k] = cur + + +def _feed(text: str, level: str = "info") -> None: + _state["feed"].appendleft({"ts": _now().isoformat(), "text": text, "level": level}) + try: + from main import add_feed + add_feed("etl-guardian", f"[lakehouse] {text}", level) + except Exception: + pass + + +def _scalar(v: Any) -> Any: + import datetime as _dt + import decimal + if v is None or isinstance(v, (str, int, float, bool)): + return v + if isinstance(v, decimal.Decimal): + return float(v) + if isinstance(v, (_dt.datetime, _dt.date)): + return v.isoformat() + if isinstance(v, (dict, list)): + return json.dumps(v, default=str) + return str(v) + + +def _normalize(rows: list[dict]) -> list[dict]: + keys: list[str] = [] + seen: set[str] = set() + for r in rows: + for k in r.keys(): + if k not in seen: + seen.add(k) + keys.append(k) + return [{k: _scalar(r.get(k)) for k in keys} for r in rows] + + +def _write_parquet(dataset_key: str, rows: list[dict]) -> tuple[str, int]: + import pyarrow as pa + import pyarrow.parquet as pq + table = pa.Table.from_pylist(_normalize(rows)) + buf = io.BytesIO() + pq.write_table(table, buf, compression="snappy") + body = buf.getvalue() + now = _now() + key = (f"lake/{dataset_key}/dt={now.strftime('%Y-%m-%d')}/" + f"part-{int(now.timestamp() * 1000)}-{random.randint(1000, 9999)}.parquet") + from storage_s3 import put_object_bytes + put_object_bytes(key, body, "application/vnd.apache.parquet") + return key, len(body) + + +# ── source readers: return (rows, new_cursor, done) ───────────────────────────── +def _read_orders(cursor, n): + import psycopg2 + import psycopg2.extras + import sql_console as s + conn = 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: + conn.autocommit = True + cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) + wm = int(cursor or 0) + cur.execute("SELECT * FROM public.sales_orders WHERE order_id > %s ORDER BY order_id LIMIT %s", (wm, n)) + rows = [dict(r) for r in cur.fetchall()] + new_cursor = rows[-1].get("order_id", wm) if rows else wm + return rows, new_cursor, len(rows) < n + finally: + conn.close() + + +def _mysql_keycol(cur) -> tuple[str, str]: + import sql_console as s + try: + cur.execute( + "SELECT column_name FROM information_schema.columns WHERE table_schema=%s AND table_name='employee_events' " + "AND extra LIKE '%%auto_increment%%' LIMIT 1", (s.MYSQL_DB,)) + r = cur.fetchone() + if r: + return (list(r.values())[0] if isinstance(r, dict) else r[0]), "key" + cur.execute( + "SELECT column_name FROM information_schema.key_column_usage WHERE table_schema=%s " + "AND table_name='employee_events' AND constraint_name='PRIMARY' ORDER BY ordinal_position LIMIT 1", + (s.MYSQL_DB,)) + r = cur.fetchone() + if r: + return (list(r.values())[0] if isinstance(r, dict) else r[0]), "key" + except Exception: + pass + return "", "offset" + + +def _read_hr(cursor, n, st): + import pymysql + import sql_console as s + conn = 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, cursorclass=pymysql.cursors.DictCursor) + try: + cur = conn.cursor() + if not st.get("keycol") and st.get("mode") is None: + kc, mode = _mysql_keycol(cur) + st["keycol"] = kc + st["mode"] = mode + if st.get("mode") == "key" and st.get("keycol"): + kc = st["keycol"] + wm = int(cursor or 0) + cur.execute(f"SELECT * FROM employee_events WHERE `{kc}` > %s ORDER BY `{kc}` LIMIT %s", (wm, n)) + rows = list(cur.fetchall()) + new_cursor = rows[-1].get(kc, wm) if rows else wm + else: # offset fallback + off = int(cursor or 0) + cur.execute("SELECT * FROM employee_events LIMIT %s OFFSET %s", (n, off)) + rows = list(cur.fetchall()) + new_cursor = off + len(rows) + return rows, new_cursor, len(rows) < n + finally: + conn.close() + + +def _read_supply(cursor, n): + from bson import ObjectId + import sql_console as s + cli = s._mongo_client() + try: + col = cli[s.MONGO_DB]["events"] + q = {"_id": {"$gt": ObjectId(cursor)}} if cursor else {} + docs = list(col.find(q).sort("_id", 1).limit(n)) + new_cursor = str(docs[-1]["_id"]) if docs else cursor + for d in docs: + d["_id"] = str(d["_id"]) + return docs, new_cursor, len(docs) < n + finally: + cli.close() + + +def _read_telemetry(cursor, n): + from cassandra.query import SimpleStatement + import sql_console as s + cluster = s._cass_cluster() + try: + sess = cluster.connect() + stmt = SimpleStatement(f"SELECT * FROM {s.CASS_KS}.device_metrics", fetch_size=n) + kwargs = {} + if cursor: + try: + kwargs["paging_state"] = bytes.fromhex(cursor) + except Exception: + kwargs = {} + rs = sess.execute(stmt, **kwargs) + rows = [dict(r._asdict()) for r in rs.current_rows] + ps = rs.paging_state + new_cursor = ps.hex() if ps else None + return rows, new_cursor, new_cursor is None + finally: + cluster.shutdown() + + +# ── business matrix accumulation ──────────────────────────────────────────────── +def _num(v) -> float: + try: + return float(v) + except Exception: + return 0.0 + + +def _agg(dataset_key: str, rows: list[dict]) -> tuple[int, float]: + """Fold a freshly-offloaded chunk into the live business matrix. Returns + (order_rows, revenue) for the realtime time-series.""" + o_rows = 0 + o_rev = 0.0 + with _lock: + if dataset_key == "orders": + b = _business["orders"] + for r in rows: + amt = _num(r.get("amount") if r.get("amount") is not None else r.get("total_amount")) + b["count"] += 1 + b["revenue"] += amt + _bump(b["by_region"], r.get("region"), val=amt) + _bump(b["by_status"], r.get("order_status") or r.get("status")) + _bump(b["by_channel"], r.get("sales_channel") or r.get("channel"), val=amt) + _bump(b["by_currency"], r.get("currency")) + o_rows += 1 + o_rev += amt + elif dataset_key == "hr_events": + b = _business["hr"] + for r in rows: + b["count"] += 1 + _bump(b["by_department"], r.get("department")) + _bump(b["by_event"], r.get("event_type") or r.get("event")) + _bump(b["by_region"], r.get("region")) + elif dataset_key == "supply_events": + b = _business["supply"] + for r in rows: + amt = _num(r.get("amount")) + b["count"] += 1 + b["amount"] += amt + _bump(b["by_type"], r.get("type"), val=amt) + _bump(b["by_region"], r.get("region")) + _bump(b["by_source"], r.get("source")) + elif dataset_key == "telemetry": + b = _business["telemetry"] + for r in rows: + b["count"] += 1 + mt = r.get("metric_type") or r.get("metric_name") + if mt: + cur = b["by_metric"].get(str(mt)) or {"count": 0, "sum": 0.0} + cur["count"] += 1 + cur["sum"] += _num(r.get("metric_value")) + b["by_metric"][str(mt)] = cur + return o_rows, o_rev + + +_READERS = {"orders": _read_orders, "hr_events": _read_hr, + "supply_events": _read_supply, "telemetry": _read_telemetry} + + +def _offload_dataset(key: str) -> dict[str, Any]: + st = _state["datasets"][key] + chunk = int(_state["chunk"]) + res = {"rows": 0, "bytes": 0, "order_rows": 0, "revenue": 0.0} + try: + if key == "hr_events": + rows, new_cursor, done = _read_hr(st["cursor"], chunk, st) + else: + rows, new_cursor, done = _READERS[key](st["cursor"], chunk) + except Exception as exc: + st["error"] = str(exc)[:160] + return res + st["error"] = None + if not rows: + if done: + st["backfilled"] = True + if key == "telemetry": # no global order — loop back to keep tailing + st["cursor"] = None + return res + try: + obj_key, nbytes = _write_parquet(key, rows) + except Exception as exc: + st["error"] = f"parquet/s3: {str(exc)[:140]}" + return res + o_rows, o_rev = _agg(key, rows) + with _lock: + st["parts"] += 1 + st["rows"] += len(rows) + st["bytes"] += nbytes + st["cursor"] = new_cursor + st["backfilled"] = bool(done) + st["last_ts"] = _now().isoformat() + st["last_rows"] = len(rows) + st["last_key"] = obj_key + if key == "telemetry" and done: + st["cursor"] = None + res.update({"rows": len(rows), "bytes": nbytes, "order_rows": o_rows, "revenue": o_rev}) + return res + + +def _source_totals() -> None: + """Best-effort backfill targets so the dashboard can show progress %.""" + import sql_console as s + d = _state["datasets"] + try: + if d["orders"]["total_source"] is None: + d["orders"]["total_source"] = s._table_row_count("postgres", "public.sales_orders") + except Exception: + pass + try: + if d["hr_events"]["total_source"] is None: + d["hr_events"]["total_source"] = s._table_row_count("mysql", "hr.employee_events") + except Exception: + pass + try: + if d["supply_events"]["total_source"] is None: + d["supply_events"]["total_source"] = s._table_row_count("mongodb", "supplychain.events") + except Exception: + pass + + +def run_cycle() -> dict[str, Any]: + if _state["running_cycle"]: + return {"ok": True, "skipped": "cycle already running"} + _state["running_cycle"] = True + total = 0 + cbytes = 0 + order_rows = 0 + revenue = 0.0 + try: + _source_totals() + for d in DATASETS: + if not _state["enabled"]: + break + r = _offload_dataset(d["key"]) + total += r["rows"] + cbytes += r["bytes"] + order_rows += r["order_rows"] + revenue += r["revenue"] + now = time.time() + _state["cycles"] += 1 + _state["last_cycle_ts"] = now + _state["last_cycle_rows"] = total + _state["next_run_ts"] = now + float(_state["interval_s"]) + if total: + minute = int(now // 60) * 60 + if _state["minute"] and _state["minute"][-1][0] == minute: + m, r0, b0 = _state["minute"][-1] + _state["minute"][-1] = (minute, r0 + total, b0 + cbytes) + else: + _state["minute"].append((minute, total, cbytes)) + point = {"t": _now().strftime("%H:%M:%S"), "rows": total, "bytes": cbytes, + "orders": order_rows, "revenue": round(revenue, 2)} + _state["series"].append(point) + with _lock: + _business["ts"].append(point) + _feed(f"offloaded {total:,} rows → S3 Parquet · €{int(revenue):,} (cycle {_state['cycles']})") + finally: + _state["running_cycle"] = False + return {"ok": True, "rows": total, "bytes": cbytes, "revenue": round(revenue, 2)} + + +def _loop() -> None: + _state["started_at"] = _now().isoformat() + time.sleep(12) # let the API + sources settle + while True: + try: + if _state["enabled"]: + run_cycle() + except Exception: + pass + time.sleep(max(10.0, float(_state["interval_s"]))) + + +threading.Thread(target=_loop, daemon=True, name="etl-offload").start() + + +# ── views ─────────────────────────────────────────────────────────────────────── +def _top(d: dict, n: int = 12, value: bool = False) -> list[dict]: + if value: + items = sorted(d.items(), key=lambda kv: -(kv[1].get("value", 0) if isinstance(kv[1], dict) else kv[1])) + return [{"key": k, "count": v.get("count", 0), "value": round(v.get("value", 0.0), 2)} for k, v in items[:n]] + items = sorted(d.items(), key=lambda kv: -kv[1]) + return [{"key": k, "count": v} for k, v in items[:n]] + + +def status_view() -> dict[str, Any]: + with _lock: + datasets = [] + tot_parts = tot_rows = tot_bytes = 0 + for d in DATASETS: + st = _state["datasets"][d["key"]] + total_src = st.get("total_source") + pct = None + if total_src and total_src > 0: + pct = min(100.0, round(100.0 * st["rows"] / total_src, 1)) + tot_parts += st["parts"] + tot_rows += st["rows"] + tot_bytes += st["bytes"] + datasets.append({"key": d["key"], **{k: st[k] for k in ( + "label", "engine", "color", "parts", "rows", "bytes", "backfilled", + "total_source", "last_ts", "last_rows", "last_key", "error")}, + "progress_pct": pct}) + now_min = int(time.time() // 60) * 60 + rate = 0 + for m, r, _b in _state["minute"]: + if m >= now_min - 60: + rate += r + return { + "ok": True, + "enabled": _state["enabled"], + "interval_s": _state["interval_s"], + "chunk": _state["chunk"], + "running_cycle": _state["running_cycle"], + "cycles": _state["cycles"], + "started_at": _state["started_at"], + "last_cycle_ts": _state["last_cycle_ts"], + "last_cycle_rows": _state["last_cycle_rows"], + "next_run_ts": _state["next_run_ts"], + "totals": {"parts": tot_parts, "rows": tot_rows, "bytes": tot_bytes}, + "rate_rows_per_min": rate, + "datasets": datasets, + "series": list(_state["series"]), + "feed": list(_state["feed"])[:24], + } + + +def business_view() -> dict[str, Any]: + with _lock: + o = _business["orders"] + hr = _business["hr"] + sup = _business["supply"] + tel = _business["telemetry"] + tel_metrics = sorted( + ({"key": k, "count": v["count"], "avg": round(v["sum"] / v["count"], 2) if v["count"] else 0} + for k, v in tel["by_metric"].items()), key=lambda x: -x["count"])[:12] + # combine into a region matrix across datasets + regions: dict[str, dict] = {} + for k, v in o["by_region"].items(): + regions.setdefault(k, {})["orders"] = v.get("count", 0) + regions[k]["revenue"] = round(v.get("value", 0.0), 2) + for k, v in hr["by_region"].items(): + regions.setdefault(k, {})["hr_events"] = v + for k, v in sup["by_region"].items(): + regions.setdefault(k, {})["supply_events"] = v + region_matrix = sorted( + ({"region": k, "orders": v.get("orders", 0), "revenue": v.get("revenue", 0), + "hr_events": v.get("hr_events", 0), "supply_events": v.get("supply_events", 0)} + for k, v in regions.items() if k), key=lambda r: -(r["revenue"] or 0)) + return { + "ok": True, + "generated_at": _now().isoformat(), + "kpis": { + "orders": o["count"], "revenue": round(o["revenue"], 2), + "avg_order": round(o["revenue"] / o["count"], 2) if o["count"] else 0, + "hr_events": hr["count"], "supply_events": sup["count"], + "supply_amount": round(sup["amount"], 2), "telemetry": tel["count"], + "rows_total": o["count"] + hr["count"] + sup["count"] + tel["count"], + }, + "orders_by_region": _top(o["by_region"], 12, value=True), + "orders_by_status": _top(o["by_status"], 8), + "orders_by_channel": _top(o["by_channel"], 8, value=True), + "orders_by_currency": _top(o["by_currency"], 6), + "hr_by_department": _top(hr["by_department"], 10), + "hr_by_event": _top(hr["by_event"], 8), + "supply_by_type": _top(sup["by_type"], 10, value=True), + "telemetry_by_metric": tel_metrics, + "region_matrix": region_matrix, + "ts": list(_business["ts"]), + } + + +@router.get("/status") +async def get_status() -> JSONResponse: + return JSONResponse(status_view()) + + +@router.get("/business") +async def get_business() -> JSONResponse: + return JSONResponse(business_view()) + + +@router.post("/run") +async def post_run() -> JSONResponse: + from starlette.concurrency import run_in_threadpool + res = await run_in_threadpool(run_cycle) + return JSONResponse({**res, "status": status_view()}) + + +@router.post("/config") +async def post_config(body: dict = Body(default={})) -> JSONResponse: + if "enabled" in body: + _state["enabled"] = bool(body["enabled"]) + if "interval_s" in body: + try: + _state["interval_s"] = max(10.0, min(900.0, float(body["interval_s"]))) + except Exception: + pass + if "chunk" in body: + try: + _state["chunk"] = max(200, min(50000, int(body["chunk"]))) + except Exception: + pass + _feed(f"config updated · interval={_state['interval_s']}s · chunk={_state['chunk']} · enabled={_state['enabled']}") + return JSONResponse({"ok": True, "enabled": _state["enabled"], + "interval_s": _state["interval_s"], "chunk": _state["chunk"]}) diff --git a/api/main.py b/api/main.py index 642a62a..3d24539 100644 --- a/api/main.py +++ b/api/main.py @@ -54,6 +54,7 @@ from streaming_ops import router as streaming_router from spark_workbench import router as spark_workbench_router from pii_catalog import router as pii_router from trino_federated import router as federated_router +from etl_offload import router as etl_offload_router from ssh_terminal import ssh_session from node_registry import NODE_IDS, NODE_AGENT, NODE_REGISTRY, is_node_id from node_ops import build_node_detail, probe_node, run_node_probe_task @@ -759,6 +760,7 @@ app.include_router(streaming_router) app.include_router(spark_workbench_router) app.include_router(pii_router) app.include_router(federated_router) +app.include_router(etl_offload_router) app.add_middleware( CORSMiddleware, allow_origins=["*"], diff --git a/api/requirements.txt b/api/requirements.txt index 9987891..545379e 100644 --- a/api/requirements.txt +++ b/api/requirements.txt @@ -16,3 +16,4 @@ python-pptx==1.0.2 boto3==1.35.99 paramiko==3.5.0 aiokafka==0.12.0 +pyarrow==18.1.0 diff --git a/api/storage_s3.py b/api/storage_s3.py index edc2ac9..c03ee52 100644 --- a/api/storage_s3.py +++ b/api/storage_s3.py @@ -2,7 +2,10 @@ from __future__ import annotations +import json as _json import os +import random as _rnd +import threading as _threading import time from collections import deque from datetime import datetime, timezone @@ -34,6 +37,148 @@ _activity: deque[tuple[float, str]] = deque(maxlen=20000) _analytics_cache: dict[str, Any] = {"ts": 0.0, "data": None} _ANALYTICS_TTL = 45.0 +# ── Pipeline → S3 archiver ──────────────────────────────────────────────────── +# When data is generated, the streaming pipeline must actually land objects in +# S3 so the dashboard reflects it (Last write / growth / activity). We mirror two +# real stages straight into the object store: +# • Kafka → S3 CDC archive -> cdc-archive/dt=YYYY-MM-DD/events-*.json (raw NDJSON) +# • Spark → S3 curated layer -> curated/sales_orders_masked/dt=…/part-*.json (PII masked) +# Writes are buffered like a Kafka-Connect S3 sink (flush on size or interval) so +# we don't create a flood of tiny objects, and force-flushed on a manual burst. +S3_ARCHIVE_BUCKET = os.getenv("S3_ARCHIVE_BUCKET", "data") +S3_FLUSH_INTERVAL = float(os.getenv("S3_ARCHIVE_FLUSH_S", "12")) +S3_FLUSH_SIZE = int(os.getenv("S3_ARCHIVE_FLUSH_SIZE", "400")) + +_arch_lock = _threading.Lock() +_arch_buf: dict[str, list] = {"cdc": [], "curated": []} +_arch_since: dict[str, float] = {"t": 0.0} +_last_write: dict[str, Any] = {} +_arch_stats: dict[str, int] = {"objects": 0, "bytes": 0, "rows": 0} + + +def _iso(v: Any) -> str: + return v.isoformat() if hasattr(v, "isoformat") else str(v) + + +def _mask_cust(cid: Any) -> str: + h = abs(hash(("cust", cid))) % 0xFFFFFF + return f"cust_{h:06x}***" + + +def _order_to_cdc(t: tuple, ts: str) -> dict[str, Any]: + cid, pid, region, channel, ots, amt, curr, status = t + return {"op": "c", "source": "postgres", "db": "sales", "table": "sales_orders", "ts": ts, + "after": {"customer_id": cid, "product_id": pid, "region": region, "sales_channel": channel, + "amount": amt, "currency": curr, "order_status": status, "order_ts": _iso(ots)}} + + +def _order_to_curated(t: tuple, ts: str) -> dict[str, Any]: + cid, pid, region, channel, ots, amt, curr, status = t + return {"customer_ref": _mask_cust(cid), "product_id": pid, "region": region, "sales_channel": channel, + "amount": amt, "currency": curr, "order_status": status, "order_ts": _iso(ots), + "ingested_ts": ts, "pii_masked": True} + + +def _hr_to_cdc(t: tuple, ts: str) -> dict[str, Any]: + eid, dept, role, region, evt, sal, ets = t + return {"op": "c", "source": "mysql", "db": "hr", "table": "employee_events", "ts": ts, + "after": {"employee_id": eid, "department": dept, "role_name": role, "region": region, + "event_type": evt, "salary_change": sal, "event_ts": _iso(ets)}} + + +def _supply_to_cdc(d: dict, ts: str) -> dict[str, Any]: + return {"op": "c", "source": "mongodb", "db": "supplychain", "table": "events", "ts": ts, "after": dict(d)} + + +def _tel_to_cdc(t: tuple, ts: str) -> dict[str, Any]: + dev, mts, mtype, mval, _payload = t + return {"op": "c", "source": "cassandra", "db": "telemetry", "table": "device_metrics", "ts": ts, + "after": {"device_id": dev, "metric_ts": _iso(mts), "metric_type": mtype, "metric_value": mval}} + + +def _put(s3, bucket: str, key: str, body: bytes, content_type: str) -> None: + s3.put_object(Bucket=bucket, Key=key, Body=body, ContentType=content_type) + _track("write") + _last_write.update({"ts": datetime.now(timezone.utc).isoformat(), "mono": time.time(), + "bucket": bucket, "key": key, "bytes": len(body)}) + _arch_stats["objects"] += 1 + _arch_stats["bytes"] += len(body) + + +def _flush_locked() -> list[dict[str, Any]] | None: + cdc = _arch_buf["cdc"] + cur = _arch_buf["curated"] + if not cdc and not cur: + return None + s3 = _client() + now = datetime.now(timezone.utc) + day = now.strftime("%Y-%m-%d") + ms = int(now.timestamp() * 1000) + rid = _rnd.randint(1000, 9999) + written: list[dict[str, Any]] = [] + if cdc: + body = ("\n".join(_json.dumps(e, default=str) for e in cdc) + "\n").encode() + key = f"cdc-archive/dt={day}/events-{ms}-{rid}.json" + _put(s3, S3_ARCHIVE_BUCKET, key, body, "application/x-ndjson") + written.append({"stage": "kafka→s3", "key": key, "rows": len(cdc), "bytes": len(body)}) + if cur: + body = ("\n".join(_json.dumps(e, default=str) for e in cur) + "\n").encode() + key = f"curated/sales_orders_masked/dt={day}/part-{ms}-{rid}.json" + _put(s3, S3_ARCHIVE_BUCKET, key, body, "application/x-ndjson") + written.append({"stage": "spark→s3", "key": key, "rows": len(cur), "bytes": len(body)}) + _arch_buf["cdc"] = [] + _arch_buf["curated"] = [] + _arch_since["t"] = time.time() + return written + + +def archive_generated_batch(orders_rows=None, hr_rows=None, supply_docs=None, tel_rows=None, + *, force: bool = False) -> list[dict[str, Any]] | dict[str, Any] | None: + """Stage a freshly generated batch into S3 (Kafka→S3 CDC archive + Spark→S3 + curated masked). Buffered; flushes on size/interval or when force=True.""" + try: + with _arch_lock: + ts = datetime.now(timezone.utc).isoformat() + for t in (orders_rows or []): + _arch_buf["cdc"].append(_order_to_cdc(t, ts)) + _arch_buf["curated"].append(_order_to_curated(t, ts)) + _arch_stats["rows"] += 1 + for t in (hr_rows or []): + _arch_buf["cdc"].append(_hr_to_cdc(t, ts)); _arch_stats["rows"] += 1 + for d in (supply_docs or []): + _arch_buf["cdc"].append(_supply_to_cdc(d, ts)); _arch_stats["rows"] += 1 + for t in (tel_rows or []): + _arch_buf["cdc"].append(_tel_to_cdc(t, ts)); _arch_stats["rows"] += 1 + if _arch_since["t"] == 0.0: + _arch_since["t"] = time.time() + buffered = len(_arch_buf["cdc"]) + len(_arch_buf["curated"]) + age = time.time() - _arch_since["t"] + if force or buffered >= S3_FLUSH_SIZE or age >= S3_FLUSH_INTERVAL: + return _flush_locked() + except Exception as exc: # never break the generator on an S3 hiccup + return {"error": str(exc)} + return None + + +def put_object_bytes(key: str, body: bytes, content_type: str = "application/octet-stream", + bucket: str | None = None) -> dict[str, Any]: + """Write raw bytes to S3 (used by the ETL offload agent for Parquet parts). + Tracks last-write + activity so the storage dashboard reflects it live.""" + b = bucket or S3_ARCHIVE_BUCKET + s3 = _client() + _put(s3, b, key, body, content_type) + return {"ok": True, "bucket": b, "key": key, "bytes": len(body)} + + +def archive_active(window_s: float = 25.0) -> bool: + """True if the pipeline wrote to S3 recently — drives the kafka→S3 edge pulse.""" + return (time.time() - float(_last_write.get("mono") or 0.0)) < window_s + + +def archive_info() -> dict[str, Any]: + return {"last_write": dict(_last_write) or None, "objects": _arch_stats["objects"], + "bytes": _arch_stats["bytes"], "rows": _arch_stats["rows"]} + def _client(): return boto3.client( @@ -494,5 +639,18 @@ async def analytics(refresh: bool = Query(False)): _analytics_cache["ts"] = now except Exception as exc: return JSONResponse({"ok": False, "error": str(exc), "endpoint": S3_ENDPOINT}, status_code=502) + + # Overlay the live last-write so the dashboard reflects pipeline writes + # immediately, without waiting for the (bounded, cached) full rescan. + summary = dict(data.get("summary") or {}) + recent = list(data.get("recent") or []) + lw = dict(_last_write) + if lw.get("ts"): + if not summary.get("newest") or lw["ts"] > summary["newest"]: + summary["newest"] = lw["ts"] + recent = ([{"modified": lw["ts"], "bucket": lw.get("bucket"), "key": lw.get("key"), + "bytes": lw.get("bytes", 0), "size_human": _human_size(lw.get("bytes", 0))}] + + [r for r in recent if r.get("key") != lw.get("key")])[:15] return {"ok": True, "endpoint": S3_ENDPOINT, "generated_at": datetime.now(timezone.utc).isoformat(), - "activity": _activity_view(), **data} + "activity": _activity_view(), **data, "summary": summary, "recent": recent, + "archive": archive_info()} diff --git a/api/trino_federated.py b/api/trino_federated.py index 8ffe1aa..7f9e333 100644 --- a/api/trino_federated.py +++ b/api/trino_federated.py @@ -453,25 +453,30 @@ _TEL_INSERT_TPL = ("INSERT INTO {ks}.device_metrics " 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 + return rows, by_r, by_s, val def _gen_hr(n: int): - _gen_mysql().cursor().executemany(_MYSQL_INSERT, _hr_rows(n)) + rows = _hr_rows(n) + _gen_mysql().cursor().executemany(_MYSQL_INSERT, rows) + return rows def _gen_supply(n: int): docs = _supply_docs(n) if docs: - _gen_mongo()["events"].insert_many(docs) + _gen_mongo()["events"].insert_many([dict(d) for d in docs]) + return docs def _gen_tel(n: int): import sql_console as s sess = _gen_cass() cql = _TEL_INSERT_TPL.format(ks=s.CASS_KS) - for row in _tel_rows(n): + rows = _tel_rows(n) + for row in rows: sess.execute(cql, row) + return rows def _generate_once(orders: int, hr: int, supply: int, tel: int) -> dict[str, Any]: @@ -482,10 +487,15 @@ def _generate_once(orders: int, hr: int, supply: int, tel: int) -> dict[str, Any by_r: dict[str, int] = {} by_s: dict[str, int] = {} val = 0.0 + order_built: list = [] + hr_built: list = [] + supply_built: list = [] + tel_built: list = [] if orders > 0: try: import psycopg2 rows, by_r, by_s, val = _order_rows(orders) + order_built = rows 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 @@ -498,9 +508,10 @@ def _generate_once(orders: int, hr: int, supply: int, tel: int) -> dict[str, Any if hr > 0: try: import pymysql + hr_built = _hr_rows(hr) 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)) + c.cursor().executemany(_MYSQL_INSERT, hr_built) out["hr_events"] = hr finally: c.close() @@ -508,9 +519,10 @@ def _generate_once(orders: int, hr: int, supply: int, tel: int) -> dict[str, Any pass if supply > 0: try: + supply_built = _supply_docs(supply) cli = s._mongo_client() try: - cli[s.MONGO_DB]["events"].insert_many(_supply_docs(supply)) + cli[s.MONGO_DB]["events"].insert_many([dict(d) for d in supply_built]) out["supply_events"] = supply finally: cli.close() @@ -518,17 +530,28 @@ def _generate_once(orders: int, hr: int, supply: int, tel: int) -> dict[str, Any pass if tel > 0: try: + tel_built = _tel_rows(tel) cluster = s._cass_cluster() sess = cluster.connect() try: cql = _TEL_INSERT_TPL.format(ks=s.CASS_KS) - for row in _tel_rows(tel): + for row in tel_built: sess.execute(cql, row) out["telemetry"] = tel finally: cluster.shutdown() except Exception: pass + # Land the generated batch in S3 through the pipeline stages (Kafka→S3 CDC + # archive + Spark→S3 curated masked) so Object Storage reflects it at once. + try: + from storage_s3 import archive_generated_batch + archive_generated_batch(order_built if out["orders"] else [], + hr_built if out["hr_events"] else [], + supply_built if out["supply_events"] else [], + tel_built if out["telemetry"] else [], force=True) + except Exception: + pass # fold into the live counters + feed so the dashboard reflects it instantly with _gen_lock: c = _GEN["counts"] @@ -561,22 +584,32 @@ def _gen_tick(): by_r: dict[str, int] = {} by_s: dict[str, int] = {} val = 0.0 + order_built: list = [] + hr_built: list = [] + supply_built: list = [] + tel_built: list = [] try: - by_r, by_s, val = _gen_orders(no) + order_built, by_r, by_s, val = _gen_orders(no) except Exception: _gen_reset("pg"); no = 0 try: - _gen_hr(nh) + hr_built = _gen_hr(nh) except Exception: _gen_reset("mysql"); nh = 0 try: - _gen_supply(ns) + supply_built = _gen_supply(ns) except Exception: _gen_reset("mongo"); ns = 0 try: - _gen_tel(nt) + tel_built = _gen_tel(nt) except Exception: _gen_reset("cass"); nt = 0 + # stream the batch into S3 (buffered like a Kafka-Connect S3 sink) + try: + from storage_s3 import archive_generated_batch + archive_generated_batch(order_built, hr_built, supply_built, tel_built, force=False) + except Exception: + pass with _gen_lock: c = _GEN["counts"] c["orders"] += no diff --git a/ui/src/components/features/ChangesView.tsx b/ui/src/components/features/ChangesView.tsx index 36d0cd1..cbd6c84 100644 --- a/ui/src/components/features/ChangesView.tsx +++ b/ui/src/components/features/ChangesView.tsx @@ -224,32 +224,59 @@ export function ChangesView({ liveChanges }: { liveChanges: CdcChange[] }) { const [expanded, setExpanded] = useState(null) const [connected, setConnected] = useState(false) const [flash, setFlash] = useState(false) - const prevTotal = useRef(0) + // 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: 200 }), fetchChangeStats(15)]) setSeed(c.changes) setConnected(c.connected) - if (s) setStats(s) - }, []) + applyStats(s) + }, [applyStats]) useEffect(() => { load() - const iv = setInterval(() => fetchChangeStats(15).then((s) => s && setStats(s)), 4000) + const iv = setInterval(() => fetchChangeStats(15).then((s) => applyStats(s)), 2500) return () => clearInterval(iv) - }, [load]) + }, [load, applyStats]) - // Pulse the header when fresh changes arrive. + // Fold freshly-arrived WS changes into the overlay → instant top-of-page update. useEffect(() => { - const t = stats?.total ?? 0 - if (t > prevTotal.current) { - setFlash(true) - const id = setTimeout(() => setFlash(false), 900) - prevTotal.current = t - return () => clearTimeout(id) + if (!liveChanges.length) return + if (!primed.current) { + primed.current = true + lastSeenId.current = liveChanges[0].id + return } - prevTotal.current = t - }, [stats?.total]) + 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(() => { @@ -264,29 +291,44 @@ export function ChangesView({ liveChanges }: { liveChanges: CdcChange[] }) { [merged, source, op], ) - 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 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 = 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 })) + .map((k) => ({ label: k, value: (stats?.by_op?.[k] || 0) + (overlay.by_op[k] || 0), color: opOf(k).color })) .filter((s) => s.value > 0) - ), [byOp]) + ), [stats?.by_op, overlay]) - const sourceRows = useMemo(() => ( - Object.entries(stats?.by_source || {}).sort((a, b) => b[1] - a[1]) - ), [stats?.by_source]) + 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(() => ( - Object.entries(stats?.by_table || {}).sort((a, b) => b[1] - a[1]).slice(0, 7) - ), [stats?.by_table]) + 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 */} @@ -324,7 +366,7 @@ export function ChangesView({ liveChanges }: { liveChanges: CdcChange[] }) {
Change volume — last 15 minutes
- +
diff --git a/ui/src/components/features/StorageView.tsx b/ui/src/components/features/StorageView.tsx index 04dfe6c..8db4833 100644 --- a/ui/src/components/features/StorageView.tsx +++ b/ui/src/components/features/StorageView.tsx @@ -184,6 +184,118 @@ function Panel({ title, icon: Icon, children, className, right }: { title: strin /* ── Main ────────────────────────────────────────────────────────────────── */ +type EtlDataset = { + key: string; label: string; engine: string; color: string; parts: number; rows: number; bytes: number + backfilled: boolean; total_source: number | null; last_ts: string | null; last_rows: number + last_key: string | null; error: string | null; progress_pct: number | null +} +type EtlStatus = { + ok: boolean; enabled: boolean; interval_s: number; chunk: number; running_cycle: boolean; cycles: number + last_cycle_rows: number; totals: { parts: number; rows: number; bytes: number }; rate_rows_per_min: number + datasets: EtlDataset[]; series: { t: string; rows: number; bytes: number; orders: number; revenue: number }[] + feed: { ts: string; text: string; level: string }[] +} + +function EtlIngestPanel() { + const [etl, setEtl] = useState(null) + const [busy, setBusy] = useState(false) + const load = useCallback(async () => { + try { const r = await fetch('/api/etl/status'); if (r.ok) setEtl(await r.json()) } catch { /* */ } + }, []) + useEffect(() => { load(); const t = setInterval(load, 5000); return () => clearInterval(t) }, [load]) + const runNow = async () => { + setBusy(true) + try { await fetch('/api/etl/run', { method: 'POST' }) } catch { /* */ } + setTimeout(() => { load(); setBusy(false) }, 900) + } + const cfg = async (body: Record) => { + await fetch('/api/etl/config', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify(body) }) + load() + } + const ds = etl?.datasets || [] + return ( + + + + {etl?.running_cycle ? 'OFFLOADING' : etl?.enabled ? 'STREAMING' : 'PAUSED'} + + {etl?.cycles ?? 0} cycles · {fmtNum(etl?.rate_rows_per_min || 0)} rows/min + + } + > +
+ A background ETL agent pulls small chunks from every source and writes partitioned Parquet to s3://data/lake/ every + + + + + {fmtNum(etl?.totals.parts || 0)} parts · {fmtNum(etl?.totals.rows || 0)} rows · {fmtBytes(etl?.totals.bytes || 0)} + +
+ +
+ {ds.map((d) => ( +
+
+ + {d.label} + + {d.backfilled + ? TAILING + : BACKFILL} +
+

{d.engine}

+

{fmtNum(d.rows)}

+

rows · {fmtNum(d.parts)} parts · {fmtBytes(d.bytes)}

+
+
+
+

+ {d.progress_pct != null ? `${d.progress_pct}% of ${fmtNum(d.total_source || 0)}` : 'streaming'} + {d.last_rows ? +{fmtNum(d.last_rows)} : null} +

+ {d.error &&

{d.error}

} +
+ ))} +
+ +
+
+

Rows offloaded per cycle (realtime)

+ p.rows)} color="#34d399" /> +
+
+

ETL agent activity

+
+ {(etl?.feed || []).slice(0, 6).map((f, i) => ( +

+ {f.ts.slice(11, 19)} {f.text} +

+ ))} + {!etl?.feed?.length && } +
+
+
+ + ) +} + export function StorageView() { const [tab, setTab] = useState<'overview' | 'browser'>('overview') const [an, setAn] = useState(null) @@ -207,7 +319,7 @@ export function StorageView() { useEffect(() => { loadAnalytics() - const t = setInterval(() => loadAnalytics(), 30000) + const t = setInterval(() => loadAnalytics(), 12000) return () => clearInterval(t) }, [loadAnalytics]) @@ -264,6 +376,9 @@ export function StorageView() {
+ {/* Realtime ETL offload (source → S3 Parquet) */} + + {/* Growth + bucket distribution */}
1 ? (w - pad * 2) / (d.length - 1) : 0 + const pts = d.map((v, i) => [pad + i * step, h - pad - (v / max) * (h - pad * 2)] 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` + return ( +
+ {label &&

{label}

} + + + + + + + {values.length > 0 && } + {values.length > 0 && } + {values.length > 0 && } + +
+ ) +} + export function TrinoFederationView({ embedded = false, activeTab }: { embedded?: boolean; activeTab?: SubTab } = {}) { const [tabState, setTab] = useState('federated') const tab = embedded ? activeTab ?? 'federated' : tabState @@ -124,9 +149,26 @@ export function TrinoFederationView({ embedded = false, activeTab }: { embedded? const [marquee, setMarquee] = useState(null) const [lake, setLake] = useState(null) const [dict, setDict] = useState(null) + const [biz, setBiz] = useState(null) const [loading, setLoading] = useState(false) const [matRunning, setMatRunning] = useState(false) + const loadBiz = useCallback(async () => { + try { + const r = await fetch('/api/etl/business') + if (r.ok) setBiz(await r.json()) + } catch { /* */ } + }, []) + + // Poll the live federated business model (built from the ETL lakehouse offload) + // while the federated tab is open, so the graphs move with newly generated data. + useEffect(() => { + if (tab !== 'federated') return + loadBiz() + const t = setInterval(loadBiz, 5000) + return () => clearInterval(t) + }, [tab, loadBiz]) + const loadFederated = useCallback(async () => { setLoading(true) try { @@ -198,6 +240,66 @@ export function TrinoFederationView({ embedded = false, activeTab }: { embedded? {/* ───────── FEDERATED ───────── */} {tab === 'federated' && ( <> + {/* ───────── REALTIME FEDERATED BUSINESS MODEL ───────── */} + +

+ + Business matrices built continuously from the lakehouse offload across all five data points — orders, HR, supply & telemetry — and they move as new data is generated & streamed to S3. +

+
+ + + + + + +
+
+ p.orders)} color="#fbbf24" /> + p.revenue)} color="#34d399" /> +
+ + +
+ + + {biz?.region_matrix?.length ? ( +
+ + + + + + + {biz.region_matrix.map((r: any, i: number) => ( + + + + + + + + ))} + +
RegionOrdersRevenueHRSupply
{r.region}{fmtNum(r.orders)}{fmtMoney(r.revenue)}{fmtNum(r.hr_events)}{fmtNum(r.supply_events)}
+
+ ) :

Building the model from the lakehouse offload…

} + +
+ + +
+ + + + ({ key: x.key, count: x.count, value: x.avg }))} valueKind="num" /> + +
+
diff --git a/ui/src/hooks/useCommandCenter.ts b/ui/src/hooks/useCommandCenter.ts index 46f656d..fd0dd37 100644 --- a/ui/src/hooks/useCommandCenter.ts +++ b/ui/src/hooks/useCommandCenter.ts @@ -129,7 +129,7 @@ export function useCommandCenter() { if (msg.type === 'terminal') appendTerminal(msg.line) if (msg.type === 'terminal_history' && msg.terminals) setTerminals(msg.terminals) if (msg.type === 'feed') setFeed((prev) => [msg.entry, ...prev].slice(0, 100)) - if (msg.type === 'cdc_change' && msg.entry) setChanges((prev) => [msg.entry, ...prev].slice(0, 400)) + if (msg.type === 'cdc_change' && msg.entry) setChanges((prev) => [msg.entry, ...prev].slice(0, 800)) if (msg.type === 'agent_dispatch') { setSelectedAgentId(msg.agent_id) setAnims((p) => ({ ...p, [msg.agent_id]: { agentId: msg.agent_id, state: 'walk', zone: msg.zone } }))