diff --git a/api/main.py b/api/main.py index 84caa05..2422c57 100644 --- a/api/main.py +++ b/api/main.py @@ -47,6 +47,7 @@ from elasticsearch_api import router as elasticsearch_router from sql_console import router as sql_router from agent_ops import router as agent_ops_router, agent_dml_loop, etl_agent_loop, custodian_offload_loop from agent_activity import agent_activity_loop +from streaming_ops import connector_autoheal_loop from cdc_consumer import router as cdc_router, cdc_consumer_loop from movements import router as movements_router from movements import MOVEMENT_BY_ID, trigger_and_watch @@ -737,6 +738,7 @@ async def lifespan(app: FastAPI): etl_task = asyncio.create_task(etl_agent_loop()) cust_task = asyncio.create_task(custodian_offload_loop()) act_task = asyncio.create_task(agent_activity_loop()) + heal_task = asyncio.create_task(connector_autoheal_loop()) add_feed("infra-sentinel", "ATC Command Center API online", "info") yield task.cancel() @@ -745,6 +747,7 @@ async def lifespan(app: FastAPI): etl_task.cancel() cust_task.cancel() act_task.cancel() + heal_task.cancel() if redis_client: await redis_client.close() diff --git a/api/streaming_ops.py b/api/streaming_ops.py index 4215c03..aa2a4b6 100644 --- a/api/streaming_ops.py +++ b/api/streaming_ops.py @@ -433,3 +433,126 @@ async def resume_kafka_connector(name: str) -> JSONResponse: return JSONResponse({"ok": True, "connector": name, "action": "resume"}) except Exception as exc: return JSONResponse({"ok": False, "error": str(exc)}, status_code=500) + + +# ── Source CDC re-sync + autonomous self-heal ─────────────────────────────── +# When a source database briefly drops, the Debezium tasks land in FAILED (or +# stall while still reporting RUNNING) and never recover on their own. The +# re-sync restarts the connectors so CDC catches up; the auto-heal loop does the +# same automatically for FAILED tasks. +SOURCE_CONNECTORS = [c for c in os.getenv( + "CDC_SOURCE_CONNECTORS", + "postgres-sales-connector,mysql-hr-connector,mongodb-supplychain-connector", +).split(",") if c.strip()] + + +async def _term(agent_id: str, text: str, level: str = "info", phase: str = "resync") -> None: + try: + from agent_terminal import terminal_log + await terminal_log(agent_id, text, level=level, phase=phase) + except Exception: + pass + + +async def _connector_status(client: httpx.AsyncClient, name: str) -> dict[str, Any]: + try: + r = await client.get(f"{KAFKA_CONNECT_URL}/connectors/{name}/status") + if r.status_code >= 400: + return {"name": name, "state": "MISSING", "tasks": [], "failed": []} + st = r.json() + tasks = st.get("tasks") or [] + return { + "name": name, + "state": (st.get("connector") or {}).get("state"), + "tasks": [{"id": t.get("id"), "state": t.get("state")} for t in tasks], + "failed": [t.get("id") for t in tasks if t.get("state") == "FAILED"], + } + except Exception as exc: + return {"name": name, "state": "ERROR", "error": str(exc)[:120], "tasks": [], "failed": []} + + +async def _discover_connectors(client: httpx.AsyncClient) -> list[str]: + try: + r = await client.get(f"{KAFKA_CONNECT_URL}/connectors") + if r.status_code < 400 and isinstance(r.json(), list) and r.json(): + return r.json() + except Exception: + pass + return list(SOURCE_CONNECTORS) + + +async def resync_connectors(force: bool = True, names: list[str] | None = None) -> dict[str, Any]: + """force=True → restart every connector incl. all tasks (full re-sync, also + recovers stalled-but-RUNNING tasks). force=False → only restart connectors + that have a FAILED connector or task (self-heal).""" + result: dict[str, Any] = {"ok": True, "restarted": [], "skipped": [], "before": []} + async with httpx.AsyncClient(timeout=20.0) as client: + targets = names or await _discover_connectors(client) + for name in targets: + before = await _connector_status(client, name) + result["before"].append(before) + unhealthy = before.get("state") in ("FAILED", "ERROR", "MISSING") or before.get("failed") + if not force and not unhealthy: + result["skipped"].append(name) + continue + try: + qs = "includeTasks=true" if force else "includeTasks=true&onlyFailed=true" + rr = await client.post(f"{KAFKA_CONNECT_URL}/connectors/{name}/restart?{qs}") + if rr.status_code < 400: + result["restarted"].append(name) + await _term("etl-guardian", f"$ kafka-connect restart {name} ({'full re-sync' if force else 'failed tasks'})", level="cmd") + else: + result["ok"] = False + await _term("etl-guardian", f" ✗ {name}: HTTP {rr.status_code}", level="err") + except Exception as exc: + result["ok"] = False + await _term("etl-guardian", f" ✗ {name}: {str(exc)[:100]}", level="err") + _cache["ts"] = 0 + return result + + +@router.get("/connectors") +async def connectors_status() -> JSONResponse: + async with httpx.AsyncClient(timeout=12.0) as client: + names = await _discover_connectors(client) + items = [await _connector_status(client, n) for n in names] + healthy = sum(1 for c in items if c.get("state") == "RUNNING" and not c.get("failed")) + return JSONResponse({"ok": True, "connect_url": KAFKA_CONNECT_URL, + "healthy": healthy, "total": len(items), "connectors": items}) + + +@router.post("/resync") +async def resync_sources(body: dict[str, Any] = Body(default={})) -> JSONResponse: + force = bool(body.get("force", True)) + _feed("etl-guardian", f"[kafka] Source CDC re-sync requested ({'full' if force else 'failed only'})", "warn") + await _term("etl-guardian", "═══ Source CDC re-sync — restarting Debezium connectors ═══", level="info") + res = await resync_connectors(force=force) + await asyncio.sleep(3) # let tasks transition + async with httpx.AsyncClient(timeout=15.0) as client: + after = [await _connector_status(client, c["name"]) for c in res["before"]] + res["after"] = after + healthy = sum(1 for c in after if c.get("state") == "RUNNING" and not c.get("failed")) + res["healthy"] = healthy + res["total"] = len(after) + await _term("etl-guardian", f" ← re-sync requested for {len(res['restarted'])} connectors · {healthy}/{len(after)} healthy", level="ok") + return JSONResponse(res) + + +async def connector_autoheal_loop() -> None: + """Autonomously restart FAILED Debezium tasks so CDC recovers after a source + DB outage without operator action.""" + await asyncio.sleep(40) + interval = max(30.0, float(os.getenv("CONNECTOR_AUTOHEAL_SECONDS", "60"))) + while True: + try: + async with httpx.AsyncClient(timeout=15.0) as client: + for name in await _discover_connectors(client): + st = await _connector_status(client, name) + if st.get("state") in ("FAILED", "ERROR") or st.get("failed"): + await client.post(f"{KAFKA_CONNECT_URL}/connectors/{name}/restart?includeTasks=true&onlyFailed=true") + _feed("etl-guardian", f"[kafka] auto-heal restarted {name}", "warn") + await _term("etl-guardian", f" ⟳ auto-heal: restarted FAILED task(s) on {name}", level="warn", phase="autoheal") + _cache["ts"] = 0 + except Exception: + pass + await asyncio.sleep(interval) diff --git a/ui/src/components/features/ChangesView.tsx b/ui/src/components/features/ChangesView.tsx index cbd6c84..6ad24cb 100644 --- a/ui/src/components/features/ChangesView.tsx +++ b/ui/src/components/features/ChangesView.tsx @@ -1,6 +1,6 @@ import { useCallback, useEffect, useMemo, useRef, useState } from 'react' -import { Activity, Radio, RefreshCw, PlusCircle, Pencil, Trash2, Database, Layers, TrendingUp } from 'lucide-react' -import { fetchChanges, fetchChangeStats } from '../../lib/api' +import { Activity, Radio, RefreshCw, PlusCircle, Pencil, Trash2, Database, Layers, TrendingUp, Cable } from 'lucide-react' +import { fetchChanges, fetchChangeStats, resyncSources } from '../../lib/api' import type { CdcChange, CdcStats } from '../../types' import { cn } from '../../lib/utils' @@ -224,6 +224,8 @@ export function ChangesView({ liveChanges }: { liveChanges: CdcChange[] }) { const [expanded, setExpanded] = useState(null) const [connected, setConnected] = useState(false) const [flash, setFlash] = useState(false) + const [resyncing, setResyncing] = useState(false) + const [resyncMsg, setResyncMsg] = useState(null) // 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 @@ -246,6 +248,25 @@ export function ChangesView({ liveChanges }: { liveChanges: CdcChange[] }) { applyStats(s) }, [applyStats]) + const doResync = useCallback(async () => { + setResyncing(true) + setResyncMsg('Restarting Debezium connectors…') + try { + const res = await resyncSources(true) + if (res) { + setResyncMsg(`Re-synced · ${res.healthy ?? 0}/${res.total ?? 0} connectors healthy`) + setTimeout(() => load(), 2500) + } else { + setResyncMsg('Re-sync failed — check ETL Guardian terminal') + } + } catch { + setResyncMsg('Re-sync failed — check ETL Guardian terminal') + } finally { + setResyncing(false) + setTimeout(() => setResyncMsg(null), 7000) + } + }, [load]) + useEffect(() => { load() const iv = setInterval(() => fetchChangeStats(15).then((s) => applyStats(s)), 2500) @@ -346,6 +367,18 @@ export function ChangesView({ liveChanges }: { liveChanges: CdcChange[] }) { connected ? 'border-emerald-500/40 bg-emerald-500/10 text-emerald-300' : 'border-rose-500/40 bg-rose-500/10 text-rose-300')}> {connected ? 'STREAMING' : 'OFFLINE'} + {resyncMsg && ( + {resyncMsg} + )} + diff --git a/ui/src/lib/api.ts b/ui/src/lib/api.ts index 3be3787..4d684c6 100644 --- a/ui/src/lib/api.ts +++ b/ui/src/lib/api.ts @@ -115,6 +115,18 @@ export async function fetchChangeStats(minutes = 15): Promise { return fetchJson(`/api/changes/stats?minutes=${minutes}`, 8000) } +export type ConnectorState = { name: string; state?: string; failed?: number[] } +export type ResyncResult = { ok: boolean; restarted?: string[]; healthy?: number; total?: number; after?: ConnectorState[] } + +export async function resyncSources(force = true): Promise { + const r = await fetch('/api/pipeline/streaming/resync', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ force }), + }) + return r.ok ? ((await r.json()) as ResyncResult) : null +} + export async function fetchAgentOpsStatus() { return fetchJson>('/api/agent-ops/status', 8000) }