From f55c3affcbb65a3ead1fd1b42c0aa0a8220071c0 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 12 Jun 2026 09:25:56 +0000 Subject: [PATCH] refactor(autocon5): restructure Part 3 automation as one flow per diagram block MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Make the Prefect run graph mirror the Part 3 teaching diagram one-to-one: each block in alert → evidence → policy → action is now a flow, and each step inside a block is a task. - evidence_flow: fetch_sot / fetch_metrics / fetch_logs submitted concurrently (independent sources; a retry on one no longer re-fetches the other two), then assemble_evidence decodes enums and bundles. - policy_flow: the two policy stages become visible tasks — evaluate_sot_gate always runs, evaluate_metrics_gate only when stage 1 doesn't short-circuit, then annotate_decision writes the audit record. - action_flow: AI narrative (ai_rca / ai_rca_skipped) plus the deterministic quarantine + annotate_action on proceed only. - quarantine_bgp_flow shrinks to pure orchestration; audit-record labels, decision values, and the flow's return shape are unchanged, so the guide's LogQL queries and try-it checks keep working as-is. The block subflows carry the same device/peer parameters as the parent, so `nobs autocon5 cycle`'s flow-runs panel would have flooded with rows; it now resolves each subflow's parent via the Prefect task-run linkage and renders children indented under their quarantine_bgp run, falling back to the flat list if the lookup fails. Verified offline against an ephemeral Prefect 3 server with the SDK's HTTP clients mocked: proceed, maintenance-skip (incl. ai_rca_skipped), and resolved paths all complete with the expected audit records. https://claude.ai/code/session_01Dn67fTBA32o1Hsyg3HYthf --- workshops/autocon5/automation/flows.py | 304 ++++++++++++------ .../autocon5/src/autocon5_workshop/cycle.py | 75 ++++- 2 files changed, 269 insertions(+), 110 deletions(-) diff --git a/workshops/autocon5/automation/flows.py b/workshops/autocon5/automation/flows.py index c7160e6..654eef6 100644 --- a/workshops/autocon5/automation/flows.py +++ b/workshops/autocon5/automation/flows.py @@ -1,20 +1,33 @@ """ Prefect flows for AutoCon5 — Part 3 of the workshop. -Pipeline (per BgpSessionNotUp alert): - - alert_receiver - └── quarantine_bgp_flow (fires on status=firing) - ├── collect_bgp_evidence_task (metrics + logs + SoT) - ├── evaluate_policy_task (DecisionPolicy: stop/skip/proceed) - ├── annotate_decision_task (audit trail to Loki) - ├── ai_rca_task (opt-in; ENABLE_AI_RCA toggle) - └── if proceed: - ├── quarantine_task (Alertmanager silence) - └── annotate_action_task (audit trail to Loki) - - └── resolved_bgp_flow (fires on status=resolved) - └── annotate_decision_task +The flow structure mirrors the Part 3 teaching diagram one-to-one: each block +in `alert → evidence → policy → action` is a flow, and each step inside a +block is a task. + + alert_receiver (Alert — webhook entrypoint) + └── quarantine_bgp_flow (status=firing; orchestrates the cycle) + ├── evidence_flow (Evidence) + │ ├── fetch_sot (Infrahub GraphQL — intent) + │ ├── fetch_metrics (Prometheus PromQL — reality) + │ ├── fetch_logs (Loki LogQL — recent history) + │ └── assemble_evidence (decode enums, bundle the three) + ├── policy_flow (Policy) + │ ├── evaluate_sot_gate (stage 1 — SoT only) + │ ├── evaluate_metrics_gate (stage 2 — SoT + metrics; only if stage 1 passes) + │ └── annotate_decision (audit trail to Loki) + └── action_flow (Action) + ├── ai_rca / ai_rca_skipped (narrative — always writes a record) + └── if proceed: + ├── quarantine (Alertmanager silence) + └── annotate_action (audit trail to Loki) + + └── resolved_bgp_flow (status=resolved) + └── annotate_decision + +The three fetch tasks inside evidence_flow are submitted concurrently — the +sources are independent, and a retry on one source no longer re-fetches the +other two. The four canonical paths from the AutoCon5 outline map to: actionable / mismatch → quarantine @@ -29,34 +42,57 @@ from prefect import flow, tags, task from prefect.logging import get_run_logger -from workshop_sdk import Decision, DecisionPolicy, EvidenceBundle, WorkshopSDK, is_ai_rca_enabled +from workshop_sdk import ( + Decision, + DecisionPolicy, + EvidenceBundle, + WorkshopSDK, + decode_bgp_states, + is_ai_rca_enabled, +) # --------------------------------------------------------------------------- -# Tasks +# Evidence tasks # --------------------------------------------------------------------------- -@task(retries=2, retry_delay_seconds=3, log_prints=True, task_run_name="collect_evidence[{device}:{peer_address}]") -def collect_bgp_evidence_task( +@task(retries=2, retry_delay_seconds=3, log_prints=True, task_run_name="fetch_sot[{device}:{peer_address}]") +def fetch_sot_task(device: str, peer_address: str, afi_safi: str) -> dict[str, Any]: + print(f"🔎 [evidence] SoT gate for {device}:{peer_address} ({afi_safi})") + return WorkshopSDK().bgp_gate(device=device, peer_address=peer_address, afi_safi=afi_safi) + + +@task(retries=2, retry_delay_seconds=3, log_prints=True, task_run_name="fetch_metrics[{device}:{peer_address}]") +def fetch_metrics_task(device: str, peer_address: str, afi_safi: str, instance_name: str) -> dict[str, float]: + print(f"🔎 [evidence] BGP metrics snapshot for {device}:{peer_address}") + return WorkshopSDK().bgp_metrics_snapshot( + device=device, peer_address=peer_address, afi_safi=afi_safi, instance_name=instance_name + ) + + +@task(retries=2, retry_delay_seconds=3, log_prints=True, task_run_name="fetch_logs[{device}:{peer_address}]") +def fetch_logs_task(device: str, peer_address: str, log_minutes: int, log_limit: int) -> list[str]: + print(f"🔎 [evidence] last {log_minutes}m of logs for {device}:{peer_address}") + return WorkshopSDK().bgp_logs(device=device, peer_address=peer_address, minutes=log_minutes, limit=log_limit) + + +@task(log_prints=True, task_run_name="assemble_evidence[{device}:{peer_address}]") +def assemble_evidence_task( device: str, peer_address: str, afi_safi: str, instance_name: str, - log_minutes: int, - log_limit: int, + sot: dict[str, Any], + metrics: dict[str, float], + logs: list[str], ) -> EvidenceBundle: - print(f"🔎 [collect] device={device} peer={peer_address} afi={afi_safi} instance={instance_name}") - sdk = WorkshopSDK() - ev = sdk.collect_bgp_evidence( - device=device, - peer_address=peer_address, - afi_safi=afi_safi, - instance_name=instance_name, - log_minutes=log_minutes, - log_limit=log_limit, - ) + ev = EvidenceBundle(device=device, peer_address=peer_address, afi_safi=afi_safi, instance_name=instance_name) + ev.sot = sot + ev.metrics = metrics + ev.sot["decoded"] = decode_bgp_states(metrics) + ev.logs = logs print( - "✅ [collect] sot.found={} maintenance={} intended={} expected_state={} reason={!r}".format( + "✅ [evidence] sot.found={} maintenance={} intended={} expected_state={} reason={!r}".format( ev.sot.get("found"), ev.sot.get("maintenance"), ev.sot.get("intended_peer"), @@ -69,19 +105,23 @@ def collect_bgp_evidence_task( return ev -@task(log_prints=True, task_run_name="evaluate_policy[{device}:{peer_address}]") -def evaluate_policy_task(device: str, peer_address: str, ev: EvidenceBundle) -> Decision: - print(f"🧠 [policy] {device}:{peer_address}") - policy = DecisionPolicy() +# --------------------------------------------------------------------------- +# Policy tasks +# --------------------------------------------------------------------------- + - sot_decision = policy.evaluate(ev.sot, metrics=None) - print(f" stage1 SoT-only → {sot_decision.decision} ({sot_decision.reason})") - if sot_decision.decision in {"stop", "skip"}: - return sot_decision +@task(log_prints=True, task_run_name="evaluate_sot_gate[{device}:{peer_address}]") +def evaluate_sot_gate_task(device: str, peer_address: str, ev: EvidenceBundle) -> Decision: + decision = DecisionPolicy().evaluate(ev.sot, metrics=None) + print(f"🧠 [policy] stage1 SoT-only → {decision.decision} ({decision.reason})") + return decision - metrics_decision = policy.evaluate(ev.sot, metrics=ev.metrics) - print(f" stage2 SoT+metrics → {metrics_decision.decision} ({metrics_decision.reason})") - return metrics_decision + +@task(log_prints=True, task_run_name="evaluate_metrics_gate[{device}:{peer_address}]") +def evaluate_metrics_gate_task(device: str, peer_address: str, ev: EvidenceBundle) -> Decision: + decision = DecisionPolicy().evaluate(ev.sot, metrics=ev.metrics) + print(f"🧠 [policy] stage2 SoT+metrics → {decision.decision} ({decision.reason})") + return decision @task(log_prints=True, task_run_name="annotate_decision[{device}:{peer_address}]") @@ -97,6 +137,11 @@ def annotate_decision_task(workflow: str, device: str, peer_address: str, decisi ) +# --------------------------------------------------------------------------- +# Action tasks +# --------------------------------------------------------------------------- + + @task(log_prints=True, task_run_name="ai_rca[{device}:{peer_address}]") def ai_rca_task(workflow: str, device: str, peer_address: str, ev: EvidenceBundle) -> str: """Opt-in LLM RCA. Always returns SOMETHING — disabled-fallback or model output.""" @@ -170,7 +215,110 @@ def annotate_action_task(workflow: str, device: str, peer_address: str, silence_ # --------------------------------------------------------------------------- -# Action flows +# Block flows — one per diagram block (evidence / policy / action) +# --------------------------------------------------------------------------- + + +@flow(log_prints=True, flow_run_name="evidence | {device}:{peer_address}") +def evidence_flow( + device: str, + peer_address: str, + afi_safi: str = "ipv4-unicast", + instance_name: str = "default", + log_minutes: int = 30, + log_limit: int = 50, +) -> EvidenceBundle: + """Evidence block: three concurrent fetches (intent / reality / history), one bundle.""" + sot = fetch_sot_task.submit(device=device, peer_address=peer_address, afi_safi=afi_safi) + metrics = fetch_metrics_task.submit( + device=device, peer_address=peer_address, afi_safi=afi_safi, instance_name=instance_name + ) + logs = fetch_logs_task.submit( + device=device, peer_address=peer_address, log_minutes=log_minutes, log_limit=log_limit + ) + return assemble_evidence_task( + device=device, + peer_address=peer_address, + afi_safi=afi_safi, + instance_name=instance_name, + sot=sot, + metrics=metrics, + logs=logs, + ) + + +@flow(log_prints=True, flow_run_name="policy | {device}:{peer_address}") +def policy_flow( + device: str, + peer_address: str, + ev: EvidenceBundle, + workflow: str = "autocon5_quarantine_bgp", +) -> Decision: + """Policy block: two-stage deterministic evaluation, then the audit record. + + Stage 2 only runs when stage 1 (SoT-only) doesn't short-circuit — so a + maintenance-skip run shows a single evaluate task in the UI, while a + proceed run shows both stages. + """ + print(f"🧠 [policy] {device}:{peer_address}") + decision = evaluate_sot_gate_task(device=device, peer_address=peer_address, ev=ev) + if decision.decision not in {"stop", "skip"}: + decision = evaluate_metrics_gate_task(device=device, peer_address=peer_address, ev=ev) + annotate_decision_task(workflow=workflow, device=device, peer_address=peer_address, decision=decision) + return decision + + +@flow(log_prints=True, flow_run_name="action | {device}:{peer_address}") +def action_flow( + device: str, + peer_address: str, + decision: Decision, + ev: EvidenceBundle, + quarantine_minutes: int = 20, + workflow: str = "autocon5_quarantine_bgp", +) -> dict[str, Any]: + """Action block: AI narrative (always writes) + deterministic action (proceed only).""" + # AI RCA branching: + # 1. ENABLE_AI_RCA=false → ai_rca_task writes the "disabled" annotation + # regardless of decision (the feature is off; that's the only honest + # message to write). + # 2. ENABLE_AI_RCA=true + decision=proceed → real LLM narrative. + # 3. ENABLE_AI_RCA=true + decision != proceed → ai_rca_skipped_task + # writes a brief "not run because policy said skip" annotation. + # Skips the LLM call entirely (saves compute / API cost, keeps the + # audit trail honest — SoT says "don't act", so we don't act + # anywhere, including the LLM step). + if is_ai_rca_enabled() and decision.decision != "proceed": + rca_text = ai_rca_skipped_task( + workflow=workflow, + device=device, + peer_address=peer_address, + decision=decision, + ) + else: + rca_text = ai_rca_task( + workflow=workflow, + device=device, + peer_address=peer_address, + ev=ev, + ) + + if decision.decision != "proceed": + print(f"✅ [action] no deterministic action ({decision.decision} — {decision.reason})") + return {"action": "none", "silence_id": None, "ai_rca": rca_text} + + silence_id = quarantine_task(device=device, peer_address=peer_address, minutes=quarantine_minutes) + annotate_action_task( + workflow=workflow, + device=device, + peer_address=peer_address, + silence_id=silence_id, + ) + return {"action": "quarantine", "silence_id": silence_id, "ai_rca": rca_text} + + +# --------------------------------------------------------------------------- +# Orchestrating flows # --------------------------------------------------------------------------- @@ -194,7 +342,7 @@ def quarantine_bgp_flow( f"instance:{instance_name}", "action:quarantine", ): - ev = collect_bgp_evidence_task( + ev = evidence_flow( device=device, peer_address=peer_address, afi_safi=afi_safi, @@ -203,68 +351,25 @@ def quarantine_bgp_flow( log_limit=log_limit, ) - decision = evaluate_policy_task(device=device, peer_address=peer_address, ev=ev) - annotate_decision_task( - workflow="autocon5_quarantine_bgp", + decision = policy_flow(device=device, peer_address=peer_address, ev=ev) + + outcome = action_flow( device=device, peer_address=peer_address, decision=decision, + ev=ev, + quarantine_minutes=quarantine_minutes, ) - # AI RCA branching: - # 1. ENABLE_AI_RCA=false → ai_rca_task writes the "disabled" annotation - # regardless of decision (the feature is off; that's the only honest - # message to write). - # 2. ENABLE_AI_RCA=true + decision=proceed → real LLM narrative. - # 3. ENABLE_AI_RCA=true + decision != proceed → ai_rca_skipped_task - # writes a brief "not run because policy said skip" annotation. - # Skips the LLM call entirely (saves compute / API cost, keeps the - # audit trail honest — SoT says "don't act", so we don't act - # anywhere, including the LLM step). - if is_ai_rca_enabled() and decision.decision != "proceed": - rca_text = ai_rca_skipped_task( - workflow="autocon5_quarantine_bgp", - device=device, - peer_address=peer_address, - decision=decision, - ) + if outcome["action"] == "quarantine": + logger.info("Quarantine applied: silence_id=%s", outcome["silence_id"]) else: - rca_text = ai_rca_task( - workflow="autocon5_quarantine_bgp", - device=device, - peer_address=peer_address, - ev=ev, - ) - - if decision.decision != "proceed": print(f"✅ [flow] no action ({decision.decision} — {decision.reason})") - return { - "device": device, - "peer_address": peer_address, - "action": "none", - "decision": { - "ok": decision.ok, - "decision": decision.decision, - "reason": decision.reason, - "details": decision.details, - }, - "evidence_summary": ev.summary(), - "ai_rca": rca_text, - } - - silence_id = quarantine_task(device=device, peer_address=peer_address, minutes=quarantine_minutes) - logger.info("Quarantine applied: silence_id=%s", silence_id) - annotate_action_task( - workflow="autocon5_quarantine_bgp", - device=device, - peer_address=peer_address, - silence_id=silence_id, - ) - return { + + result: dict[str, Any] = { "device": device, "peer_address": peer_address, - "action": "quarantine", - "silence_id": silence_id, + "action": outcome["action"], "decision": { "ok": decision.ok, "decision": decision.decision, @@ -272,8 +377,11 @@ def quarantine_bgp_flow( "details": decision.details, }, "evidence_summary": ev.summary(), - "ai_rca": rca_text, + "ai_rca": outcome["ai_rca"], } + if outcome["action"] == "quarantine": + result["silence_id"] = outcome["silence_id"] + return result @flow(log_prints=True, flow_run_name="resolved_bgp | {device}:{peer_address}") diff --git a/workshops/autocon5/src/autocon5_workshop/cycle.py b/workshops/autocon5/src/autocon5_workshop/cycle.py index 1e3e2b0..6c7d6cd 100644 --- a/workshops/autocon5/src/autocon5_workshop/cycle.py +++ b/workshops/autocon5/src/autocon5_workshop/cycle.py @@ -207,25 +207,76 @@ def _render_flow_runs_panel(prefect_url: str, device: str, peer: str, minutes: i ) return + try: + top, children = _group_flow_runs(prefect_url, runs) + except Exception: + # Grouping is cosmetic — fall back to a flat list if the lookup fails. + top, children = runs, {} + table = Table(title=f"Prefect flow runs (last {minutes}m)", show_header=True, header_style="bold") table.add_column("Started") table.add_column("State") table.add_column("Flow") - for r in runs[:5]: - state = r.get("state", {}).get("type", "?") - state_style = {"COMPLETED": "green", "FAILED": "red", "RUNNING": "yellow"}.get(state, "") - run_id = r.get("id", "") - started_cell = r.get("start_time", "")[11:19] - if run_id: - started_cell = f"[link={prefect_url}/runs/flow-run/{run_id}]{started_cell}[/link]" - table.add_row( - started_cell, - f"[{state_style}]{state}[/]" if state_style else state, - r.get("name", "?"), - ) + for r in top[:3]: + _add_run_row(table, prefect_url, r) + kids = children.get(r.get("id", ""), []) + for i, c in enumerate(kids): + branch = "└─" if i == len(kids) - 1 else "├─" + _add_run_row(table, prefect_url, c, prefix=f" [dim]{branch}[/] ") console.print(table) +def _add_run_row(table: Table, prefect_url: str, run: dict, prefix: str = "") -> None: + state = run.get("state", {}).get("type", "?") + state_style = {"COMPLETED": "green", "FAILED": "red", "RUNNING": "yellow"}.get(state, "") + run_id = run.get("id", "") + started_cell = run.get("start_time", "")[11:19] + if run_id: + started_cell = f"[link={prefect_url}/runs/flow-run/{run_id}]{started_cell}[/link]" + name = run.get("name", "?") + if prefix: + # Child rows: the parent row already carries the device:peer suffix. + name = name.split(" | ")[0] + table.add_row( + started_cell, + f"[{state_style}]{state}[/]" if state_style else state, + f"{prefix}{name}", + ) + + +def _group_flow_runs(prefect_url: str, runs: list[dict]) -> tuple[list[dict], dict[str, list[dict]]]: + """Split runs into top-level runs and children grouped by parent flow-run id. + + The evidence/policy/action block flows are subflows of quarantine_bgp and + carry the same device/peer parameters, so they land in the same filtered + set. A run counts as a child only when its parent flow run is *also* in the + set — quarantine_bgp itself is a subflow of alert_receiver, but + alert_receiver carries no device/peer parameters so it never appears here + and quarantine_bgp stays top-level. Subflows are linked to their parent via + a synthetic task run, hence the task-run lookup. + """ + ids = {r.get("id") for r in runs} + task_ids = sorted({r["parent_task_run_id"] for r in runs if r.get("parent_task_run_id")}) + task_to_flow: dict[str, str] = {} + if task_ids: + body = {"task_runs": {"id": {"any_": task_ids}}, "limit": len(task_ids)} + resp = requests.post(f"{prefect_url}/api/task_runs/filter", json=body, timeout=10) + resp.raise_for_status() + task_to_flow = {t.get("id", ""): t.get("flow_run_id", "") for t in resp.json()} + + top: list[dict] = [] + children: dict[str, list[dict]] = {} + for r in runs: + parent_flow = task_to_flow.get(r.get("parent_task_run_id") or "") + if parent_flow in ids: + children.setdefault(parent_flow, []).append(r) + else: + top.append(r) + for kids in children.values(): + kids.sort(key=lambda r: r.get("start_time") or "") + return top, children + + def _render_decision_panel(loki: LokiClient, device: str, peer: str, minutes: int) -> None: # The `decision=~".+"` matcher keeps annotate_action records (which carry no # `decision` label) out of the result, so we land on the actual decision.