Merge origin/main into running Command Center

Keep the deployed tree on conflict; integrate the remote Dockhand/CDC tip.

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
+    agg = _aggregate_window(minutes)
-    by_source: dict[str, int] = {}
+    return {
-    total = 0
+        "connected": _state["connected"],
-    for c in _ring:
+        "consumed": _state["consumed"],
-        try:
+        "buffered": len(_ring),
-            if datetime.fromisoformat(c["ts"]).timestamp() < cutoff:
+        "window_total": agg["total"],
-                continue
+        "by_source": agg["by_source"],
-        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}
 
 
 # ── 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:
+async def change_stats(minutes: int = Query(15, le=1440)) -> JSONResponse:
-    now = datetime.now(timezone.utc)
+    agg = _aggregate_window(minutes)
-    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
     return JSONResponse({
         "ok": True,
         "window_minutes": minutes,
-        "total": total,
+        **agg,
-        "by_source": by_source,
-        "by_op": by_op,
-        "by_table": by_table,
-        "buckets": [{"t": k, "n": v} for k, v in sorted(buckets.items())],
         "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})

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)

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"))),

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=

ui/src/components/features/ChangesView.tsx

@@ -21,6 +21,12 @@ const SOURCE_COLOR: Record<string, string> = {
 
 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[] }) {
           </p>
         </div>
         <div className="flex items-center gap-3">
+          <div className="flex items-center gap-1 rounded border border-border/60 p-0.5">
+            {TIME_WINDOWS.map((w) => (
+              <button
+                key={w.minutes}
+                type="button"
+                onClick={() => setWindowMin(w.minutes)}
+                className={cn(
+                  'rounded px-2 py-0.5 text-[10px]',
+                  windowMin === w.minutes
+                    ? 'bg-docker/20 text-docker'
+                    : 'text-foreground-muted hover:text-foreground',
+                )}
+              >
+                {w.label}
+              </button>
+            ))}
+          </div>
           <span className={cn('flex items-center gap-1.5 rounded-full border px-2.5 py-1 text-[10px] font-medium',
             connected ? 'border-emerald-500/40 bg-emerald-500/10 text-emerald-300' : 'border-rose-500/40 bg-rose-500/10 text-rose-300')}>
             <Radio className={cn('h-3 w-3', connected && 'animate-pulse')} /> {connected ? 'STREAMING' : 'OFFLINE'}

ui/src/lib/api.ts

@@ -150,15 +150,16 @@ export async function fetchPresentation(): Promise<PresentationData | null> {
   return fetchJson<PresentationData>('/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))
+  p.set('limit', String(opts.limit ?? 200))
-  const j = await fetchJson<{ changes?: CdcChange[]; connected?: boolean; consumed?: number }>(
+  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<CdcStats | null> {

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<string, number>
   by_op: Record<string, number>
   by_table: Record<string, number>
   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 }