diff --git a/api/cdc_consumer.py b/api/cdc_consumer.py index ff27bb5..f3aec36 100644 --- a/api/cdc_consumer.py +++ b/api/cdc_consumer.py @@ -18,7 +18,7 @@ import json import os import re import uuid -from collections import deque +from collections import defaultdict, deque from datetime import datetime, timezone from typing import Any @@ -28,7 +28,8 @@ from fastapi.responses import JSONResponse router = APIRouter(prefix="/api/changes", tags=["changes"]) KAFKA_BOOTSTRAP = os.getenv("KAFKA_BOOTSTRAP", "10.0.21.36:9092") -RING_SIZE = int(os.getenv("CDC_RING_SIZE", "1000")) +RING_SIZE = int(os.getenv("CDC_RING_SIZE", "5000")) +METRICS_RETENTION_MIN = int(os.getenv("CDC_METRICS_RETENTION_MIN", str(24 * 60))) # Active CDC topic prefixes -> logical source. Matches Debezium topic.prefix. PREFIX_SOURCE = { @@ -45,6 +46,8 @@ TOPIC_PATTERN = re.compile( OP_MAP = {"c": "insert", "u": "update", "d": "delete", "r": "snapshot"} _ring: deque[dict[str, Any]] = deque(maxlen=RING_SIZE) +# Minute-level rollups — accurate counts beyond the display ring buffer cap. +_minute_rollups: dict[int, dict[str, Any]] = {} _state: dict[str, Any] = { "started": False, "connected": False, @@ -127,6 +130,83 @@ def _parse(topic: str, value: bytes | None) -> dict[str, Any] | None: } +def _record_metrics(entry: dict[str, Any]) -> None: + """Track per-minute aggregates for accurate window stats (not ring-limited).""" + try: + ts = datetime.fromisoformat(entry["ts"].replace("Z", "+00:00")) + except Exception: + ts = datetime.now(timezone.utc) + minute_key = int(ts.timestamp()) // 60 + bucket = _minute_rollups.setdefault( + minute_key, + {"total": 0, "by_source": defaultdict(int), "by_op": defaultdict(int), "by_table": defaultdict(int)}, + ) + bucket["total"] += 1 + bucket["by_source"][entry["source"]] += 1 + bucket["by_op"][entry["op"]] += 1 + bucket["by_table"][entry["table"]] += 1 + + cutoff = minute_key - METRICS_RETENTION_MIN + for old_key in list(_minute_rollups.keys()): + if old_key < cutoff: + del _minute_rollups[old_key] + + +def _aggregate_window(minutes: int) -> dict[str, Any]: + """Aggregate rollup buckets for the requested time window.""" + now = datetime.now(timezone.utc) + cutoff_minute = int((now.timestamp() - minutes * 60) // 60) + use_hour_buckets = minutes >= 120 + + by_source: dict[str, int] = defaultdict(int) + by_op: dict[str, int] = defaultdict(int) + by_table: dict[str, int] = defaultdict(int) + buckets: dict[str, int] = defaultdict(int) + total = 0 + + for minute_key, bucket in _minute_rollups.items(): + if minute_key < cutoff_minute: + continue + total += bucket["total"] + for k, v in bucket["by_source"].items(): + by_source[k] += v + for k, v in bucket["by_op"].items(): + by_op[k] += v + for k, v in bucket["by_table"].items(): + by_table[k] += v + ts = datetime.fromtimestamp(minute_key * 60, timezone.utc) + label = ts.strftime("%H:00") if use_hour_buckets else ts.strftime("%H:%M") + buckets[label] += bucket["total"] + + # Fill chart timeline with zero buckets so the window span is visible. + if use_hour_buckets: + span = max(1, (minutes + 59) // 60) + filled: dict[str, int] = {} + for i in range(span): + t = datetime.fromtimestamp(now.timestamp() - (span - 1 - i) * 3600, timezone.utc) + filled[t.strftime("%H:00")] = buckets.get(t.strftime("%H:00"), 0) + buckets = filled + else: + span = max(1, minutes) + filled = {} + for i in range(span): + t = datetime.fromtimestamp(now.timestamp() - (span - 1 - i) * 60, timezone.utc) + filled[t.strftime("%H:%M")] = buckets.get(t.strftime("%H:%M"), 0) + buckets = filled + + return { + "total": total, + "by_source": dict(by_source), + "by_op": dict(by_op), + "by_table": dict(by_table), + "buckets": [{"t": k, "n": v} for k, v in sorted(buckets.items())], + "inserts": by_op.get("insert", 0), + "updates": by_op.get("update", 0), + "deletes": by_op.get("delete", 0), + "rate_per_min": round(total / max(1, minutes), 1), + } + + async def cdc_consumer_loop() -> None: """Background loop: consume CDC topics and publish each change live.""" _state["started"] = True @@ -158,6 +238,7 @@ async def cdc_consumer_loop() -> None: if not entry: continue _ring.append(entry) + _record_metrics(entry) _state["consumed"] += 1 _state["last_ts"] = entry["ts"] try: @@ -182,30 +263,36 @@ async def cdc_consumer_loop() -> None: def snapshot(minutes: int = 15) -> dict[str, Any]: """Lightweight CDC snapshot for other modules (Data Flow graph).""" - cutoff = datetime.now(timezone.utc).timestamp() - minutes * 60 - by_source: dict[str, int] = {} - total = 0 - for c in _ring: - try: - if datetime.fromisoformat(c["ts"]).timestamp() < cutoff: - continue - except Exception: - continue - total += 1 - by_source[c["source"]] = by_source.get(c["source"], 0) + 1 - return {"connected": _state["connected"], "consumed": _state["consumed"], - "buffered": len(_ring), "window_total": total, "by_source": by_source} + agg = _aggregate_window(minutes) + return { + "connected": _state["connected"], + "consumed": _state["consumed"], + "buffered": len(_ring), + "window_total": agg["total"], + "by_source": agg["by_source"], + } # ── Endpoints ──────────────────────────────────────────────────────────────── @router.get("") async def list_changes( - limit: int = Query(100, le=500), + limit: int = Query(200, le=1000), source: str | None = None, op: str | None = None, table: str | None = None, + minutes: int | None = Query(None, le=1440), ) -> JSONResponse: items = list(_ring) + if minutes is not None: + cutoff = datetime.now(timezone.utc).timestamp() - minutes * 60 + filtered_by_time: list[dict[str, Any]] = [] + for c in items: + try: + if datetime.fromisoformat(c["ts"].replace("Z", "+00:00")).timestamp() >= cutoff: + filtered_by_time.append(c) + except Exception: + continue + items = filtered_by_time if source: items = [c for c in items if c["source"] == source] if op: @@ -217,47 +304,29 @@ async def list_changes( "ok": True, "changes": items, "buffered": len(_ring), + "buffer_cap": RING_SIZE, "connected": _state["connected"], "consumed": _state["consumed"], "last_error": _state["last_error"], + "last_ts": _state["last_ts"], }) @router.get("/stats") -async def change_stats(minutes: int = Query(15, le=240)) -> JSONResponse: - now = datetime.now(timezone.utc) - cutoff = now.timestamp() - minutes * 60 - by_source: dict[str, int] = {} - by_op: dict[str, int] = {} - by_table: dict[str, int] = {} - buckets: dict[str, int] = {} - total = 0 - for c in _ring: - try: - t = datetime.fromisoformat(c["ts"]).timestamp() - except Exception: - continue - if t < cutoff: - continue - total += 1 - by_source[c["source"]] = by_source.get(c["source"], 0) + 1 - by_op[c["op"]] = by_op.get(c["op"], 0) + 1 - by_table[c["table"]] = by_table.get(c["table"], 0) + 1 - bucket = datetime.fromtimestamp(t, timezone.utc).strftime("%H:%M") - buckets[bucket] = buckets.get(bucket, 0) + 1 +async def change_stats(minutes: int = Query(15, le=1440)) -> JSONResponse: + agg = _aggregate_window(minutes) return JSONResponse({ "ok": True, "window_minutes": minutes, - "total": total, - "by_source": by_source, - "by_op": by_op, - "by_table": by_table, - "buckets": [{"t": k, "n": v} for k, v in sorted(buckets.items())], + **agg, "connected": _state["connected"], "consumed": _state["consumed"], + "buffered": len(_ring), + "buffer_cap": RING_SIZE, + "last_ts": _state["last_ts"], }) @router.get("/status") async def changes_status() -> JSONResponse: - return JSONResponse({"ok": True, "buffered": len(_ring), **_state}) + return JSONResponse({"ok": True, "buffered": len(_ring), "buffer_cap": RING_SIZE, **_state}) diff --git a/api/lab_context.py b/api/lab_context.py index 0cb4980..a3e56d5 100644 --- a/api/lab_context.py +++ b/api/lab_context.py @@ -808,11 +808,12 @@ async def collect_full_lab_context( if gpu_data is None: gpu_data = await collect_gpu_metrics(client, log) - docker_raw, db_raw, lake_raw, cc_raw, hdfs, etl, objectscale = await asyncio.gather( + docker_raw, db_raw, lake_raw, cc_raw, gpu_raw, hdfs, etl, objectscale = await asyncio.gather( dockhand_containers(client, DOCKHAND_ENVS["docker01"], log), dockhand_containers(client, DOCKHAND_ENVS["db02"], log), dockhand_containers(client, DOCKHAND_ENVS["lakehouse"], log), dockhand_containers(client, DOCKHAND_ENV_COMMAND_CENTER, log), + dockhand_containers(client, DOCKHAND_ENVS["gpu_dev"], log), collect_hdfs(client, log), collect_etl(client, log), collect_objectscale(client, log), @@ -829,6 +830,15 @@ async def collect_full_lab_context( databases["inventory"] = {"error": str(exc)} lakehouse = await collect_lakehouse(client, lake_raw, log) command_center = await collect_command_center(client, cc_raw, log) + if gpu_data is not None and gpu_raw: + gpu_running = sum(1 for c in gpu_raw if c.get("state") == "running") + gpu_data = { + **gpu_data, + "dockhand_env": DOCKHAND_ENVS["gpu_dev"], + "dockhand_containers": _container_rows(gpu_raw, GPU_URL.replace("http://", "").split(":")[0]), + "dockhand_running": gpu_running, + "dockhand_total": len(gpu_raw), + } try: governance = collect_governance(log) diff --git a/api/workload.py b/api/workload.py index 56a4d5d..3a9d1e6 100644 --- a/api/workload.py +++ b/api/workload.py @@ -292,7 +292,15 @@ def build_workload_payload(snap: dict[str, Any]) -> dict[str, Any]: return {"id": eid, "from": src, "to": dst, "label": label, "kind": kind, "active": active} pipeline_ok = etl.get("airflow_healthy") and len(connectors) > 0 and etl.get("kafka_ui_ok") + direct_pipeline_ok = ( + pipeline_ok + and lakehouse.get("trino_ok") + and objectscale_ok + ) s3_flow_ok = pipeline_ok and consumer_running and objectscale_ok + if not s3_flow_ok and direct_pipeline_ok and not consumer_running: + # Dockhand blind (auth/outage) but ETL + Trino + S3 probes healthy + s3_flow_ok = True topology_edges = [ _edge("e-seed", "airflow", "db", "seed data", "pipeline", bool(etl.get("airflow_healthy"))), diff --git a/config/command-center/atc.env.example b/config/command-center/atc.env.example index 7396bd7..a4a54e5 100644 --- a/config/command-center/atc.env.example +++ b/config/command-center/atc.env.example @@ -16,3 +16,6 @@ LLM_API_KEY=sk-local LAKEHOUSE_HOST=10.0.21.50 AIRFLOW_URL=http://10.0.21.55:8080 KAFKA_UI_URL=http://10.0.21.36:9000 + +# Dockhand API token (Profile → API tokens in Dockhand UI; prefix dh_) +DOCKHAND_API_TOKEN= diff --git a/ui/src/components/features/ChangesView.tsx b/ui/src/components/features/ChangesView.tsx index 6ad24cb..9554a26 100644 --- a/ui/src/components/features/ChangesView.tsx +++ b/ui/src/components/features/ChangesView.tsx @@ -21,6 +21,12 @@ const SOURCE_COLOR: Record = { const SOURCES = ['all', 'postgres', 'mysql', 'mongodb', 'cassandra', 'neo4j'] const OPS = ['all', 'insert', 'update', 'delete'] +const TIME_WINDOWS = [ + { label: '15 min', minutes: 15 }, + { label: '1 hour', minutes: 60 }, + { label: '6 hours', minutes: 360 }, + { label: '24 hours', minutes: 1440 }, +] as const function opOf(o: string) { return OP_STYLE[o] || { label: o.toUpperCase(), cls: 'bg-slate-500/15 text-slate-300 border-slate-500/30', color: '#94a3b8' } @@ -363,6 +369,23 @@ export function ChangesView({ liveChanges }: { liveChanges: CdcChange[] }) {

+
+ {TIME_WINDOWS.map((w) => ( + + ))} +
{connected ? 'STREAMING' : 'OFFLINE'} diff --git a/ui/src/lib/api.ts b/ui/src/lib/api.ts index b92a2a7..7495432 100644 --- a/ui/src/lib/api.ts +++ b/ui/src/lib/api.ts @@ -150,15 +150,16 @@ export async function fetchPresentation(): Promise { return fetchJson('/api/presentation', 60000) } -export async function fetchChanges(opts: { source?: string; op?: string; limit?: number } = {}) { +export async function fetchChanges(opts: { source?: string; op?: string; limit?: number; minutes?: number } = {}) { const p = new URLSearchParams() if (opts.source) p.set('source', opts.source) if (opts.op) p.set('op', opts.op) - p.set('limit', String(opts.limit ?? 150)) - const j = await fetchJson<{ changes?: CdcChange[]; connected?: boolean; consumed?: number }>( + p.set('limit', String(opts.limit ?? 200)) + if (opts.minutes) p.set('minutes', String(opts.minutes)) + const j = await fetchJson<{ changes?: CdcChange[]; connected?: boolean; consumed?: number; last_ts?: string | null }>( `/api/changes?${p.toString()}`, 8000, ) - return { changes: j?.changes || [], connected: !!j?.connected, consumed: j?.consumed || 0 } + return { changes: j?.changes || [], connected: !!j?.connected, consumed: j?.consumed || 0, last_ts: j?.last_ts } } export async function fetchChangeStats(minutes = 15): Promise { diff --git a/ui/src/types.ts b/ui/src/types.ts index be18250..d337096 100644 --- a/ui/src/types.ts +++ b/ui/src/types.ts @@ -62,12 +62,19 @@ export type CdcStats = { ok: boolean window_minutes: number total: number + inserts?: number + updates?: number + deletes?: number + rate_per_min?: number by_source: Record by_op: Record by_table: Record buckets: { t: string; n: number }[] connected: boolean consumed: number + buffered?: number + buffer_cap?: number + last_ts?: string | null } export type PiiColumn = { name: string; category: string; masked: boolean; policy_locked?: boolean }