Build a Self-Healing Infrastructure System
Wire anomaly detection, AI diagnosis, and Ansible remediation into one autonomous pipeline that detects, diagnoses, and fixes production incidents without human intervention.
Domains & Technologies
Blueprint Walkthrough
Before You Start — Read This First
What you are building and why it matters
Swiggy processes roughly 1.5 million orders a day. During peak dinner hours, every service matters. When the order-service pods start getting OOMKilled, engineers have about three minutes before users see timeouts, before orders fail silently, before the support queue starts filling up.
In those three minutes, a human on-call engineer needs to:
- Get paged and wake up (or stop what they are doing)
- Open the monitoring dashboard
- Figure out what is wrong
- Look up the runbook
- Run the remediation commands
- Verify the fix worked
Most of that time is not thinking — it is plumbing. Finding the right Prometheus graph. Navigating to the right runbook. Running commands they have run dozens of times before.
Self-healing infrastructure eliminates that plumbing entirely. The system detects the anomaly, diagnoses the cause, selects the remediation, executes it, and verifies the result — all before the on-call engineer's phone has even finished ringing.
This capstone builds that system by wiring together everything from this roadmap:
- Capstone 1's Isolation Forest model detects the anomaly from Prometheus metrics
- Capstone 2's RAG + LLM pipeline diagnoses the cause and selects the remediation
- Ansible playbooks execute the fix safely with full audit logging
- A decision engine decides when to act autonomously vs. when to ask for human approval
The result is a system that handles the 80% of incidents that are predictable and repetitive — restarts, scaling events, cache flushes, pod evictions — automatically. Engineers handle the remaining 20% that genuinely require judgement.
Time to complete: 5-6 hours.
What you need before starting:
- Completed Capstone 1 (trained Isolation Forest model at
models/isolation_forest.joblib) - Completed Capstone 2 (runbooks indexed in ChromaDB, Ollama running)
- Ansible installed locally
## Verify Capstone 1 model existsls -lh models/isolation_forest.joblib## Should show the file ## Verify Capstone 2 runbooks are indexedpython3 -c "import chromadbc = chromadb.PersistentClient(path='data/chroma')col = c.get_collection('runbooks')print(f'Runbooks indexed: {col.count()} chunks')" ## Verify Ollama is runningcurl -s http://localhost:11434/api/tags | python3 -m json.tool | grep name## Should show at least one model ## Install Ansiblepip3 install ansible --break-system-packagesansible --version echo "✅ Ready to build Capstone 3"Understanding the Full System
How all three capstones connect
This is the architecture you are building. Read it carefully before touching any code.
┌─────────────────────────────────────────────────────────────┐│ DETECTION (from Capstone 1) ││ ││ Prometheus metrics → Isolation Forest → Anomaly score ││ Runs every 60 seconds │└──────────────────────────────┬──────────────────────────────┘ │ anomaly detected ▼┌─────────────────────────────────────────────────────────────┐│ DIAGNOSIS (from Capstone 2) ││ ││ Live context (Prometheus + kubectl) → RAG runbook search ││ → LLM reasoning → structured diagnosis with action type │└──────────────────────────────┬──────────────────────────────┘ │ diagnosis + action ▼┌─────────────────────────────────────────────────────────────┐│ DECISION ENGINE (new in Capstone 3) ││ ││ Is this action on the approved list? ││ Is confidence above the threshold? ││ Has this action run too many times recently? ││ ├── YES → execute autonomously ││ └── NO → send to Slack for human approval │└──────────────────────────────┬──────────────────────────────┘ │ approved ▼┌─────────────────────────────────────────────────────────────┐│ REMEDIATION (new in Capstone 3) ││ ││ Ansible playbook executes the fix ││ Full audit log written ││ Verification check runs after completion ││ Result posted to Slack │└─────────────────────────────────────────────────────────────┘The Decision Engine is the most important new piece. Without it, you have a system that might restart pods, scale deployments, or flush caches based on a wrong diagnosis. The decision engine defines what the system is allowed to do on its own — and forces human approval for anything riskier.
The three tiers of autonomy
Not every remediation action carries the same risk. The system organises actions into three tiers:
Tier 1 — Fully autonomous (system acts immediately): Pod restart, cache flush, horizontal scale-up. These are reversible, low-risk, and run hundreds of times a year. Automating them saves the most on-call time with the least risk.
Tier 2 — Autonomous with notification (system acts, then tells humans): Memory limit increase, rolling restart of a deployment, draining a node. Higher impact but still reversible. The system acts immediately and posts the result to Slack so engineers can review.
Tier 3 — Human approval required (system asks before acting): Rollback of a recent deployment, scaling down a cluster, any action involving databases or persistent storage. These actions can cause data loss or downtime if wrong. A human must approve via Slack button before the system proceeds.
Project Structure
Setting up from the capstone 1 and 2 directories
## This capstone lives in a new directory but imports from capstones 1 and 2mkdir aiops-self-healing && cd aiops-self-healing mkdir -p src ansible/playbooks ansible/inventory logs audit k8s touch src/config.pytouch src/orchestrator.pytouch src/decision_engine.pytouch src/ansible_executor.pytouch src/verifier.pytouch src/audit_logger.pytouch src/slack_approvals.pytouch requirements.txt echo "✅ Project structure created"aiops-self-healing/ src/ config.py <- all settings including autonomy thresholds orchestrator.py <- the main loop: detect -> diagnose -> decide -> act decision_engine.py <- decides autonomous vs human-approval actions ansible_executor.py <- runs Ansible playbooks safely verifier.py <- confirms the fix worked after execution audit_logger.py <- writes every action to a tamper-evident log slack_approvals.py <- sends approval requests and handles responses ansible/ playbooks/ <- one playbook per remediation action inventory/ <- Kubernetes inventory for Ansible logs/ <- orchestrator logs audit/ <- audit trail (JSONL, one entry per action) k8s/ <- Kubernetes deployment manifests# requirements.txt# Inherits all deps from capstones 1 and 2pandas==2.1.0scikit-learn==1.3.0requests==2.31.0chromadb==0.4.18sentence-transformers==2.2.2joblib==1.3.2fastapi==0.104.1uvicorn==0.24.0ansible-runner==2.3.4schedule==1.2.0pip3 install -r requirements.txtecho "✅ Dependencies installed"Configuration — Autonomy Settings
The config file controls what the system is allowed to do
# src/config.pyimport os ## ── Inherited from Capstone 1 ────────────────────────────────PROMETHEUS_URL = os.getenv("PROMETHEUS_URL", "http://localhost:9090")MODEL_PATH = "../aiops-anomaly-detection/models/isolation_forest.joblib"DATA_PATH = "../aiops-anomaly-detection/data/metrics.csv" ## ── Inherited from Capstone 2 ────────────────────────────────CHROMA_DB_PATH = "../aiops-incident-agent/data/chroma"CHROMA_COLLECTION_NAME = "runbooks"EMBEDDING_MODEL = "all-MiniLM-L6-v2"OLLAMA_URL = os.getenv("OLLAMA_URL", "http://localhost:11434")LLM_MODEL = os.getenv("LLM_MODEL", "llama3.1") ## ── Detection Settings ───────────────────────────────────────## How often to run the detection loop (seconds)DETECTION_INTERVAL_SECONDS = 60## How many consecutive anomaly readings before triggering remediationANOMALY_CONSECUTIVE_THRESHOLD = 2 ## ── Slack ─────────────────────────────────────────────────────SLACK_WEBHOOK_URL = os.getenv("SLACK_WEBHOOK_URL", "")## For interactive approvals, you need a Slack bot tokenSLACK_BOT_TOKEN = os.getenv("SLACK_BOT_TOKEN", "")SLACK_APPROVAL_CHANNEL = os.getenv("SLACK_APPROVAL_CHANNEL", "#on-call") ## ── Autonomy Configuration ───────────────────────────────────## This is the most important section.## It defines exactly what the system can do autonomously. ## Tier 1: System executes immediately, no human neededAUTONOMOUS_ACTIONS = [ "restart_pod", "flush_redis_cache", "scale_up_deployment", ## only scale UP, never scale down autonomously] ## Tier 2: System executes, then notifies (no prior approval needed)NOTIFY_AFTER_ACTIONS = [ "increase_memory_limit", "rolling_restart_deployment", "evict_overloaded_node",] ## Tier 3: System asks for approval before executingAPPROVAL_REQUIRED_ACTIONS = [ "rollback_deployment", "scale_down_deployment", "delete_pvc", "modify_database_config",] ## Minimum LLM confidence score (0-1) to act autonomously## Below this threshold, always ask for human approval even for Tier 1 actionsAUTONOMOUS_CONFIDENCE_THRESHOLD = 0.75 ## Maximum times the same action can run on the same service per hour## Prevents the system from restart-looping a broken pod foreverMAX_ACTIONS_PER_SERVICE_PER_HOUR = 3 ## ── Ansible ──────────────────────────────────────────────────ANSIBLE_PLAYBOOK_DIR = "ansible/playbooks"ANSIBLE_INVENTORY = "ansible/inventory/hosts.yaml" ## ── Audit ─────────────────────────────────────────────────────AUDIT_LOG_PATH = "audit/actions.jsonl" ## ── Verification ─────────────────────────────────────────────## Seconds to wait after remediation before verifying the fix workedVERIFICATION_WAIT_SECONDS = 90RememberThe
AUTONOMOUS_ACTIONSlist is your blast radius control. Start it small — justrestart_pod— and expand it as you build confidence in the system's accuracy. Never add irreversible actions to this list.
Writing the Ansible Playbooks
Why Ansible instead of running kubectl directly
You could call subprocess.run(["kubectl", "rollout", "restart", ...]) directly from Python. Many systems do. The problem is that direct subprocess calls have no idempotency guarantees, no retry logic, no structured output, and no native audit trail.
Ansible gives you all four. Each playbook is idempotent (safe to run twice), has built-in retry with backoff, produces structured JSON output you can parse, and Ansible Runner writes every execution to a persistent artifact directory.
Think of it this way: kubectl is the screwdriver. Ansible is the procedure that says "use the screwdriver, turn it this many times, verify the screw is seated, mark it done."
## Create the pod restart playbookcat > ansible/playbooks/restart_pod.yaml << 'EOF'---- name: Restart a Kubernetes pod by deleting it (Deployment recreates it) hosts: localhost connection: local gather_facts: false vars: ## These are passed in by the Python executor: ## namespace: the Kubernetes namespace ## pod_name: the specific pod to restart ## service_name: used for verification after restart tasks: - name: Verify the pod exists before trying to delete it kubernetes.core.k8s_info: kind: Pod namespace: "{{ namespace }}" name: "{{ pod_name }}" register: pod_info failed_when: pod_info.resources | length == 0 - name: Delete the pod (the Deployment controller recreates it immediately) kubernetes.core.k8s: state: absent kind: Pod namespace: "{{ namespace }}" name: "{{ pod_name }}" - name: Wait for a new pod to reach Running state kubernetes.core.k8s_info: kind: Pod namespace: "{{ namespace }}" label_selectors: - "app={{ service_name }}" field_selectors: - "status.phase=Running" register: new_pods retries: 10 delay: 10 until: new_pods.resources | length > 0 - name: Report result debug: msg: "Pod {{ pod_name }} restarted. New pods running: {{ new_pods.resources | length }}"EOF echo "✅ restart_pod.yaml created"## Create the scale-up playbookcat > ansible/playbooks/scale_up_deployment.yaml << 'EOF'---- name: Scale up a Kubernetes deployment hosts: localhost connection: local gather_facts: false vars: ## namespace: the Kubernetes namespace ## deployment_name: name of the Deployment to scale ## current_replicas: current replica count (passed in by the executor) ## scale_factor: multiply current replicas by this (default 1.5) tasks: - name: Get current replica count kubernetes.core.k8s_info: kind: Deployment namespace: "{{ namespace }}" name: "{{ deployment_name }}" register: deployment_info failed_when: deployment_info.resources | length == 0 - name: Calculate new replica count (current * scale_factor, minimum 2) set_fact: new_replicas: >- {{ [ (deployment_info.resources[0].spec.replicas | float * (scale_factor | default(1.5))) | round | int, 2 ] | max }} - name: Scale the deployment up kubernetes.core.k8s_scale: kind: Deployment namespace: "{{ namespace }}" name: "{{ deployment_name }}" replicas: "{{ new_replicas }}" - name: Wait for rollout to complete command: > kubectl rollout status deployment/{{ deployment_name }} -n {{ namespace }} --timeout=120s register: rollout_result - name: Report result debug: msg: "Scaled {{ deployment_name }} to {{ new_replicas }} replicas. {{ rollout_result.stdout }}"EOF echo "✅ scale_up_deployment.yaml created"## Create the memory limit increase playbookcat > ansible/playbooks/increase_memory_limit.yaml << 'EOF'---- name: Increase memory limit for a deployment hosts: localhost connection: local gather_facts: false vars: ## namespace: Kubernetes namespace ## deployment_name: name of the Deployment ## container_name: which container to update (usually same as deployment_name) ## new_memory_limit: e.g. "512Mi" or "1Gi" tasks: - name: Verify deployment exists kubernetes.core.k8s_info: kind: Deployment namespace: "{{ namespace }}" name: "{{ deployment_name }}" register: dep_info failed_when: dep_info.resources | length == 0 - name: Patch the memory limit kubernetes.core.k8s_json_patch: kind: Deployment namespace: "{{ namespace }}" name: "{{ deployment_name }}" patch: - op: replace path: "/spec/template/spec/containers/0/resources/limits/memory" value: "{{ new_memory_limit }}" - name: Trigger rolling restart for the new limit to take effect command: > kubectl rollout restart deployment/{{ deployment_name }} -n {{ namespace }} - name: Wait for rollout command: > kubectl rollout status deployment/{{ deployment_name }} -n {{ namespace }} --timeout=180s - name: Report result debug: msg: "Memory limit for {{ deployment_name }} updated to {{ new_memory_limit }}"EOF echo "✅ increase_memory_limit.yaml created"## Create the rollback playbook (Tier 3 — requires approval)cat > ansible/playbooks/rollback_deployment.yaml << 'EOF'---- name: Roll back a Kubernetes deployment to the previous revision hosts: localhost connection: local gather_facts: false vars: ## namespace: Kubernetes namespace ## deployment_name: which deployment to roll back tasks: - name: Show rollout history before rolling back command: > kubectl rollout history deployment/{{ deployment_name }} -n {{ namespace }} register: history - name: Log the history for audit purposes debug: msg: "{{ history.stdout }}" - name: Execute rollback to previous revision command: > kubectl rollout undo deployment/{{ deployment_name }} -n {{ namespace }} register: rollback_result - name: Wait for rollback to complete command: > kubectl rollout status deployment/{{ deployment_name }} -n {{ namespace }} --timeout=180s - name: Report result debug: msg: "Rollback complete. {{ rollback_result.stdout }}"EOF echo "✅ rollback_deployment.yaml created"## Install the Kubernetes Ansible collectionansible-galaxy collection install kubernetes.core ## Create the inventory filecat > ansible/inventory/hosts.yaml << 'EOF'all: hosts: localhost: ansible_connection: local ansible_python_interpreter: "{{ ansible_playbook_python }}"EOF echo "✅ Ansible setup complete"Building the Ansible Executor
Running playbooks safely from Python
# src/ansible_executor.py# Executes Ansible playbooks and returns structured results.# Uses ansible-runner which gives us async execution, structured# output, and persistent artifacts for auditing. import ansible_runnerimport osimport jsonimport loggingfrom src.config import ANSIBLE_PLAYBOOK_DIR, ANSIBLE_INVENTORY logger = logging.getLogger(__name__) def run_playbook(playbook_name, extra_vars, private_data_dir="ansible/artifacts"): """ Execute an Ansible playbook with the given variables. Parameters: playbook_name: filename of the playbook (e.g. "restart_pod.yaml") extra_vars: dict of variables to pass to the playbook private_data_dir: where ansible-runner stores artifacts (logs, facts, etc.) Returns: dict with keys: success (bool), stdout, stderr, status, rc """ playbook_path = os.path.join(ANSIBLE_PLAYBOOK_DIR, playbook_name) if not os.path.exists(playbook_path): return { "success": False, "error": f"Playbook not found: {playbook_path}", "stdout": "", "stderr": "", } logger.info(f" Running playbook: {playbook_name}") logger.info(f" Variables: {json.dumps(extra_vars, indent=2)}") os.makedirs(private_data_dir, exist_ok=True) ## ansible_runner.run() is synchronous -- it blocks until the playbook finishes ## For production, use ansible_runner.run_async() and poll for status result = ansible_runner.run( private_data_dir=private_data_dir, playbook=playbook_path, inventory=ANSIBLE_INVENTORY, extravars=extra_vars, quiet=False, ## set True in production to reduce log noise ) ## ansible-runner returns status: "successful", "failed", "canceled", "timeout" success = result.status == "successful" ## Collect stdout from all events stdout_lines = [] for event in result.events: if event.get("event") in ("runner_on_ok", "runner_on_failed", "debug"): stdout = event.get("event_data", {}).get("res", {}).get("msg", "") if stdout: stdout_lines.append(stdout) if success: logger.info(f" ✅ Playbook succeeded: {playbook_name}") else: logger.error(f" ❌ Playbook failed: {playbook_name} (status: {result.status})") return { "success": success, "status": result.status, "rc": result.rc, "stdout": "\n".join(stdout_lines), "stderr": result.stderr.read() if result.stderr else "", "artifacts_dir": private_data_dir, } ## ── Action-to-playbook mapping ────────────────────────────────## Maps action names (from the LLM decision) to playbook filenames## and required variables. ACTION_PLAYBOOK_MAP = { "restart_pod": { "playbook": "restart_pod.yaml", "required_vars": ["namespace", "pod_name", "service_name"], }, "scale_up_deployment": { "playbook": "scale_up_deployment.yaml", "required_vars": ["namespace", "deployment_name"], }, "increase_memory_limit": { "playbook": "increase_memory_limit.yaml", "required_vars": ["namespace", "deployment_name", "container_name", "new_memory_limit"], }, "rollback_deployment": { "playbook": "rollback_deployment.yaml", "required_vars": ["namespace", "deployment_name"], },} def execute_action(action_name, action_vars): """ Execute a named remediation action with the given variables. Parameters: action_name: one of the keys in ACTION_PLAYBOOK_MAP action_vars: dict of variables the playbook needs Returns: dict with execution result """ if action_name not in ACTION_PLAYBOOK_MAP: return { "success": False, "error": f"Unknown action: {action_name}. Known actions: {list(ACTION_PLAYBOOK_MAP.keys())}", } mapping = ACTION_PLAYBOOK_MAP[action_name] ## Verify all required variables are present missing = [v for v in mapping["required_vars"] if v not in action_vars] if missing: return { "success": False, "error": f"Missing required variables for {action_name}: {missing}", } return run_playbook(mapping["playbook"], action_vars)Building the Decision Engine
The gate between diagnosis and action
# src/decision_engine.py# Decides whether to execute a remediation action autonomously,# execute and notify, or request human approval.# Also enforces rate limits to prevent action loops. import jsonimport loggingfrom datetime import datetime, timedeltafrom collections import defaultdictfrom src.config import ( AUTONOMOUS_ACTIONS, NOTIFY_AFTER_ACTIONS, APPROVAL_REQUIRED_ACTIONS, AUTONOMOUS_CONFIDENCE_THRESHOLD, MAX_ACTIONS_PER_SERVICE_PER_HOUR,) logger = logging.getLogger(__name__) ## In-memory rate limiter: tracks recent actions per service## {service_name: [datetime, datetime, ...]}_action_history = defaultdict(list) def _count_recent_actions(service_name): """Count how many actions ran for this service in the last hour.""" cutoff = datetime.now() - timedelta(hours=1) ## Remove entries older than one hour _action_history[service_name] = [ t for t in _action_history[service_name] if t > cutoff ] return len(_action_history[service_name]) def _record_action(service_name): """Record that an action was taken for this service.""" _action_history[service_name].append(datetime.now()) def decide(action_name, confidence, service_name, context): """ Make an autonomy decision for a proposed remediation action. Parameters: action_name: the remediation action (e.g. "restart_pod") confidence: float 0-1, how confident the LLM is in this diagnosis service_name: which service to act on context: dict with action variables (namespace, pod_name, etc.) Returns: dict with: - decision: "autonomous", "notify_after", "approval_required", "blocked" - reason: human-readable explanation - can_execute: bool (True means go ahead, False means wait for approval) """ ## ── Rule 1: Is this action known? ──────────────────────── all_known = AUTONOMOUS_ACTIONS + NOTIFY_AFTER_ACTIONS + APPROVAL_REQUIRED_ACTIONS if action_name not in all_known: return { "decision": "blocked", "reason": f"Action '{action_name}' is not in the approved action list. Add it to config.py to enable it.", "can_execute": False, } ## ── Rule 2: Rate limit check ───────────────────────────── recent_count = _count_recent_actions(service_name) if recent_count >= MAX_ACTIONS_PER_SERVICE_PER_HOUR: return { "decision": "blocked", "reason": ( f"Rate limit reached: {recent_count} actions taken on {service_name} " f"in the last hour (max {MAX_ACTIONS_PER_SERVICE_PER_HOUR}). " f"Automatic action paused to prevent an action loop. " f"A human should investigate why this service keeps triggering." ), "can_execute": False, } ## ── Rule 3: Confidence threshold ───────────────────────── if confidence < AUTONOMOUS_CONFIDENCE_THRESHOLD: return { "decision": "approval_required", "reason": ( f"Confidence {confidence:.0%} is below the autonomous threshold " f"({AUTONOMOUS_CONFIDENCE_THRESHOLD:.0%}). " f"Low confidence means the diagnosis may be wrong. " f"Requesting human approval before acting." ), "can_execute": False, } ## ── Rule 4: Tier classification ────────────────────────── if action_name in AUTONOMOUS_ACTIONS: _record_action(service_name) return { "decision": "autonomous", "reason": ( f"'{action_name}' is a Tier 1 action. " f"Confidence {confidence:.0%} is above threshold. " f"Rate limit OK ({recent_count}/{MAX_ACTIONS_PER_SERVICE_PER_HOUR} actions this hour). " f"Executing immediately." ), "can_execute": True, } if action_name in NOTIFY_AFTER_ACTIONS: _record_action(service_name) return { "decision": "notify_after", "reason": ( f"'{action_name}' is a Tier 2 action. " f"Executing now, will notify Slack after completion." ), "can_execute": True, } if action_name in APPROVAL_REQUIRED_ACTIONS: return { "decision": "approval_required", "reason": ( f"'{action_name}' is a Tier 3 action. " f"This action requires human approval due to its potential impact." ), "can_execute": False, } ## Should never reach here but be safe return { "decision": "blocked", "reason": "Action did not match any tier. Blocked by default.", "can_execute": False, }Building the Audit Logger
Every action needs a permanent record
In any automated system that can modify production infrastructure, every action must be auditable. Compliance frameworks like SOC 2 require it. Post-incident investigations depend on it. And if the system ever makes a wrong call, you need to know exactly what it did and why.
# src/audit_logger.py# Writes every action — attempted, approved, executed, failed —# to a JSONL file. JSONL (JSON Lines) means one JSON object per line,# making it easy to grep, parse, and stream into log management systems. import jsonimport osimport loggingfrom datetime import datetimefrom src.config import AUDIT_LOG_PATH logger = logging.getLogger(__name__) os.makedirs(os.path.dirname(AUDIT_LOG_PATH), exist_ok=True) def log_event(event_type, service_name, action_name, details, outcome=None): """ Write a structured audit event to the JSONL log. Parameters: event_type: "anomaly_detected", "diagnosis_complete", "action_decided", "action_executed", "action_failed", "approval_requested", "approval_received", "verification_passed", "verification_failed" service_name: which service was affected action_name: which remediation action (None for detection events) details: dict of additional context (metrics, LLM response, playbook output) outcome: "success", "failure", "pending", or None """ entry = { "timestamp": datetime.utcnow().isoformat() + "Z", "event_type": event_type, "service_name": service_name, "action_name": action_name, "outcome": outcome, "details": details, } try: with open(AUDIT_LOG_PATH, "a") as f: f.write(json.dumps(entry) + "\n") except Exception as e: ## Audit logging failure should never stop the remediation logger.error(f"Audit log write failed: {e}") logger.info(f"📋 Audit: {event_type} | {service_name} | {action_name or '-'} | {outcome or '-'}") def get_recent_events(service_name=None, hours=24): """ Read recent audit events. Optionally filter by service name. Returns a list of event dicts. """ if not os.path.exists(AUDIT_LOG_PATH): return [] cutoff = datetime.utcnow().timestamp() - (hours * 3600) events = [] with open(AUDIT_LOG_PATH, "r") as f: for line in f: line = line.strip() if not line: continue try: event = json.loads(line) ## Parse ISO timestamp and compare ts = datetime.fromisoformat(event["timestamp"].replace("Z", "")) if ts.timestamp() < cutoff: continue if service_name and event.get("service_name") != service_name: continue events.append(event) except Exception: continue return eventsBuilding the Verifier
Confirming the fix actually worked
# src/verifier.py# After a remediation action completes, the verifier checks whether# the incident condition has actually resolved.# If it has not, it escalates to human review. import timeimport loggingfrom src.config import VERIFICATION_WAIT_SECONDS, PROMETHEUS_URLimport requests logger = logging.getLogger(__name__) def query_prometheus(promql): """Run a PromQL query and return the float result, or None.""" try: r = requests.get( f"{PROMETHEUS_URL}/api/v1/query", params={"query": promql}, timeout=10, ) data = r.json() if data["status"] == "success" and data["data"]["result"]: return float(data["data"]["result"][0]["value"][1]) except Exception: pass return None def verify_remediation(action_name, service_name, namespace, original_metrics): """ Wait for the system to stabilise then check if the problem resolved. Parameters: action_name: which action was taken (used to know what to verify) service_name: the service that was fixed namespace: Kubernetes namespace original_metrics: dict of metric values at the time of the anomaly Returns: dict with: resolved (bool), current_metrics, message """ logger.info(f" Waiting {VERIFICATION_WAIT_SECONDS}s for system to stabilise...") time.sleep(VERIFICATION_WAIT_SECONDS) logger.info(f" Running verification checks...") ## ── Check pod health ────────────────────────────────────── ## For any action, the basic check is: are pods running and not crashing? import subprocess result = subprocess.run( ["kubectl", "get", "pods", "-n", namespace, "-l", f"app={service_name}", "--no-headers", "-o", "custom-columns=STATUS:.status.phase,RESTARTS:.status.containerStatuses[0].restartCount"], capture_output=True, text=True, timeout=10, ) pod_status_lines = result.stdout.strip().split("\n") if result.returncode == 0 else [] all_running = all("Running" in line for line in pod_status_lines if line) ## ── Check memory for OOM-related actions ───────────────── current_metrics = {} resolved = all_running ## start with pod health as the baseline if action_name in ("restart_pod", "increase_memory_limit"): mem_usage = query_prometheus( f'avg(container_memory_working_set_bytes{{container="{service_name}",namespace="{namespace}"}})' ) mem_limit = query_prometheus( f'avg(kube_pod_container_resource_limits{{container="{service_name}",namespace="{namespace}",resource="memory"}})' ) if mem_usage and mem_limit: mem_pct = (mem_usage / mem_limit) * 100 current_metrics["memory_utilization_pct"] = round(mem_pct, 1) ## If memory is still above 90% after fix, the issue is likely not resolved if mem_pct > 90: resolved = False logger.warning(f" ⚠️ Memory still at {mem_pct:.1f}% after remediation") ## ── Check error rate for error-related actions ──────────── if action_name in ("rollback_deployment", "scale_up_deployment", "rolling_restart_deployment"): error_rate = query_prometheus( f'sum(rate(http_requests_total{{service="{service_name}",status=~"5.."}}[5m])) / ' f'sum(rate(http_requests_total{{service="{service_name}"}}[5m]))' ) if error_rate is not None: error_pct = error_rate * 100 current_metrics["http_error_rate_pct"] = round(error_pct, 2) if error_pct > 5: resolved = False logger.warning(f" ⚠️ Error rate still at {error_pct:.2f}% after remediation") current_metrics["pods_all_running"] = all_running if resolved: message = f"✅ Verification passed — {service_name} appears healthy after {action_name}" else: message = ( f"⚠️ Verification failed — {service_name} may still have issues after {action_name}. " f"Current state: {current_metrics}. Manual investigation recommended." ) logger.info(f" {message}") return { "resolved": resolved, "current_metrics": current_metrics, "message": message, }Building the Slack Approval System
Getting human approval for Tier 3 actions
# src/slack_approvals.py# Sends approval requests to Slack for Tier 3 actions# and provides a simple approval endpoint that Slack calls back.# In production you would implement full Slack interactive messages.# This version sends the request and the human approves by# calling a local endpoint (simplest implementation). import requestsimport loggingimport asynciofrom src.config import SLACK_WEBHOOK_URL, SLACK_APPROVAL_CHANNEL logger = logging.getLogger(__name__) async def request_approval(action_name, service_name, namespace, action_vars, diagnosis, reason): """ Send an approval request to Slack and wait for a response. For production: implement Slack interactive buttons with a callback URL. For this capstone: the message includes the curl command to approve. Parameters: action_name: what the system wants to do service_name: which service namespace: which namespace action_vars: variables for the playbook diagnosis: the LLM's diagnosis text reason: why this action was chosen Returns: bool: True if approved, False if rejected or timed out """ vars_text = "\n".join([f" {k}: `{v}`" for k, v in action_vars.items()]) message = f"""🔔 *Approval Required — Self-Healing System* *Proposed Action:* `{action_name}`*Service:* `{service_name}` in namespace `{namespace}` *Diagnosis:*{diagnosis[:300]}... *Playbook Variables:*{vars_text} *Why this action requires approval:*{reason} *To approve, run:*curl -X POST http://localhost:8081/approve -d '{{"action": "{action_name}", "service": "{service_name}", "approved": true}}'
*To reject, run:*curl -X POST http://localhost:8081/approve -d '{{"action": "{action_name}", "service": "{service_name}", "approved": false}}'
This request expires in 10 minutes.""" await _send_slack_message(message) logger.info(f" Approval request sent to {SLACK_APPROVAL_CHANNEL}") logger.info(f" Waiting up to 10 minutes for human response...") ## Poll for approval response ## In production, use a proper callback mechanism ## Here we check a simple in-memory approval store return await _wait_for_approval(action_name, service_name, timeout_seconds=600) ## Simple in-memory store for approvals## In production: use Redis or a database_pending_approvals = {} def record_approval_response(action_name, service_name, approved): """Called by the approval endpoint when a human responds.""" key = f"{action_name}:{service_name}" _pending_approvals[key] = approved logger.info(f" Approval response recorded: {key} -> {'approved' if approved else 'rejected'}") async def _wait_for_approval(action_name, service_name, timeout_seconds=600): """Poll the approval store every 5 seconds until response or timeout.""" key = f"{action_name}:{service_name}" elapsed = 0 while elapsed < timeout_seconds: if key in _pending_approvals: result = _pending_approvals.pop(key) return result await asyncio.sleep(5) elapsed += 5 logger.warning(f" Approval request timed out after {timeout_seconds}s") return False async def _send_slack_message(message): """Send a plain text message to Slack.""" if not SLACK_WEBHOOK_URL: logger.info(f"📢 [SLACK NOT CONFIGURED]\n{message}") return try: requests.post( SLACK_WEBHOOK_URL, json={"text": message, "channel": SLACK_APPROVAL_CHANNEL}, timeout=10, ) except Exception as e: logger.error(f"Slack send failed: {e}") async def send_action_result(action_name, service_name, success, details): """Send the result of an executed action to Slack.""" icon = "✅" if success else "❌" status = "succeeded" if success else "FAILED" message = f"""{icon} *Self-Healing Action {status}*Action: `{action_name}` on `{service_name}` {details}""" await _send_slack_message(message)Building the Main Orchestrator
The loop that connects everything
# src/orchestrator.py# The main orchestration loop.# Runs detection every 60 seconds. When an anomaly is found,# runs the full pipeline: diagnose -> decide -> act -> verify. import asyncioimport joblibimport jsonimport loggingimport osimport reimport timeimport numpy as npimport pandas as pdimport requestsfrom datetime import datetimefrom src.config import ( PROMETHEUS_URL, MODEL_PATH, DETECTION_INTERVAL_SECONDS, ANOMALY_CONSECUTIVE_THRESHOLD, OLLAMA_URL, LLM_MODEL,)from src.decision_engine import decidefrom src.ansible_executor import execute_actionfrom src.verifier import verify_remediationfrom src.audit_logger import log_eventfrom src.slack_approvals import request_approval, send_action_result ## Import RAG and context from Capstone 2import syssys.path.insert(0, "../aiops-incident-agent")from src.rag_retriever import search_runbooks, format_runbooks_for_promptfrom src.context_collector import collect_service_context, format_context_for_prompt os.makedirs("logs", exist_ok=True)logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", handlers=[ logging.FileHandler("logs/orchestrator.log"), logging.StreamHandler(), ],)logger = logging.getLogger(__name__) def load_detection_model(): """Load the Isolation Forest model from Capstone 1.""" if not os.path.exists(MODEL_PATH): logger.error(f"Model not found at {MODEL_PATH}") logger.error("Run Capstone 1's train_model.py first") exit(1) package = joblib.load(MODEL_PATH) logger.info(f"Detection model loaded from {MODEL_PATH}") logger.info(f"Features: {package['feature_columns']}") return package["model"], package["scaler"], package["feature_columns"] def collect_current_metrics(feature_columns): """Collect one data point from Prometheus for all required features.""" metrics = {} QUERIES = { "cpu_usage": 'avg(rate(container_cpu_usage_seconds_total{container!=""}[5m]))', "memory_usage": 'avg(container_memory_working_set_bytes{container!=""})', "http_request_rate": 'sum(rate(http_requests_total[5m]))', "http_error_rate": 'sum(rate(http_requests_total{status=~"5.."}[5m])) / sum(rate(http_requests_total[5m]))', "response_time_p95": 'histogram_quantile(0.95, sum(rate(http_request_duration_seconds_bucket[5m])) by (le))', } for col in feature_columns: if col in QUERIES: try: r = requests.get( f"{PROMETHEUS_URL}/api/v1/query", params={"query": QUERIES[col]}, timeout=10, ) data = r.json() if data["status"] == "success" and data["data"]["result"]: metrics[col] = float(data["data"]["result"][0]["value"][1]) else: metrics[col] = 0.0 except Exception: metrics[col] = 0.0 return metrics def run_detection(model, scaler, feature_columns): """ Collect metrics and run the Isolation Forest model. Returns (is_anomaly, score, metrics_dict). """ metrics = collect_current_metrics(feature_columns) X = pd.DataFrame([{col: metrics.get(col, 0.0) for col in feature_columns}])[feature_columns] X_scaled = scaler.transform(X) score = model.score_samples(X_scaled)[0] is_anomaly = model.predict(X_scaled)[0] == -1 return is_anomaly, float(score), metrics def call_llm_for_action(context_text, runbook_text, metrics): """ Ask the LLM to diagnose the anomaly and recommend a specific action. Returns a structured response we can parse for action type and confidence. """ prompt = f"""You are an SRE at a high-traffic Indian fintech company.An anomaly has been detected in the production cluster. === LIVE INFRASTRUCTURE STATE ==={context_text} === RELEVANT RUNBOOKS ==={runbook_text} === YOUR TASK ===Diagnose the most likely cause and recommend the single best remediation action. Respond ONLY in this exact JSON format:{{ "diagnosis": "2 sentence description of what is likely happening", "action": "one of: restart_pod, scale_up_deployment, increase_memory_limit, rollback_deployment, flush_redis_cache, rolling_restart_deployment", "action_vars": {{ "namespace": "the namespace name", "service_name": "the service name", "deployment_name": "the deployment name", "pod_name": "specific pod if known, else null", "container_name": "container name if needed", "new_memory_limit": "e.g. 512Mi if increasing memory" }}, "confidence": 0.85, "reasoning": "why this action was chosen over alternatives"}} Base namespace and service_name on the context above.Only include action_vars that are needed for the chosen action.Confidence is 0-1 based on how certain you are.""" try: response = requests.post( f"{OLLAMA_URL}/api/generate", json={ "model": LLM_MODEL, "prompt": prompt, "stream": False, "options": {"num_predict": 500, "temperature": 0.05}, }, timeout=120, ) raw = response.json().get("response", "").strip() ## Extract JSON from the response ## The LLM sometimes adds preamble text before the JSON json_match = re.search(r'\{.*\}', raw, re.DOTALL) if json_match: return json.loads(json_match.group()) except Exception as e: logger.error(f"LLM call failed: {e}") return None async def handle_anomaly(metrics, score): """ Run the full pipeline when an anomaly is detected: 1. Collect context 2. Search runbooks 3. Ask LLM for action recommendation 4. Decision engine evaluates the action 5. Execute or request approval 6. Verify the result """ logger.info("=" * 55) logger.info("🚨 ANOMALY DETECTED — starting remediation pipeline") logger.info(f" Score: {score:.4f}") logger.info("=" * 55) ## Use a default service for context collection ## In production, detect the specific service from which metric spiked namespace = "production" service_name = "payment-service" ## Step 1: Collect live context logger.info("Step 1: Collecting infrastructure context...") context = collect_service_context(namespace, service_name, "AnomalyDetected") context_text = format_context_for_prompt(context) log_event("anomaly_detected", service_name, None, { "anomaly_score": score, "metrics": metrics, }) ## Step 2: RAG search logger.info("Step 2: Searching runbooks...") search_query = f"anomaly detected metrics: {' '.join([f'{k}={v:.3f}' for k, v in metrics.items()])}" runbook_results = search_runbooks(search_query) runbook_text = format_runbooks_for_prompt(runbook_results) ## Step 3: LLM diagnosis and action recommendation logger.info("Step 3: Getting LLM diagnosis...") llm_result = call_llm_for_action(context_text, runbook_text, metrics) if not llm_result: logger.error("LLM failed to return a valid action recommendation") log_event("diagnosis_complete", service_name, None, {"error": "LLM failed"}, outcome="failure") return action_name = llm_result.get("action") action_vars = llm_result.get("action_vars", {}) confidence = float(llm_result.get("confidence", 0.0)) diagnosis = llm_result.get("diagnosis", "") logger.info(f" Diagnosis: {diagnosis}") logger.info(f" Recommended action: {action_name} (confidence: {confidence:.0%})") log_event("diagnosis_complete", service_name, action_name, { "diagnosis": diagnosis, "confidence": confidence, "runbooks_used": [r["source"] for r in runbook_results], "action_vars": action_vars, }) ## Step 4: Decision engine logger.info("Step 4: Running decision engine...") decision = decide(action_name, confidence, service_name, action_vars) logger.info(f" Decision: {decision['decision']}") logger.info(f" Reason: {decision['reason']}") log_event("action_decided", service_name, action_name, { "decision": decision["decision"], "reason": decision["reason"], "confidence": confidence, }) ## Step 5: Execute or request approval if decision["can_execute"]: logger.info(f"Step 5: Executing {action_name}...") exec_result = execute_action(action_name, action_vars) outcome = "success" if exec_result["success"] else "failure" log_event("action_executed", service_name, action_name, exec_result, outcome=outcome) if decision["decision"] == "notify_after": await send_action_result(action_name, service_name, exec_result["success"], f"Executed automatically.\n{exec_result.get('stdout', '')}") elif decision["decision"] == "approval_required": logger.info(f"Step 5: Requesting approval for {action_name}...") approved = await request_approval( action_name, service_name, action_vars.get("namespace", namespace), action_vars, diagnosis, decision["reason"], ) if approved: logger.info(f" ✅ Approved — executing {action_name}...") exec_result = execute_action(action_name, action_vars) log_event("action_executed", service_name, action_name, exec_result, outcome="success" if exec_result["success"] else "failure") await send_action_result(action_name, service_name, exec_result["success"], f"Executed after human approval.\n{exec_result.get('stdout', '')}") else: logger.info(" ❌ Rejected or timed out — no action taken") log_event("action_decided", service_name, action_name, {"outcome": "rejected_or_timed_out"}, outcome="failure") return else: ## Blocked by rate limit or unknown action logger.warning(f" Action blocked: {decision['reason']}") await send_action_result(action_name, service_name, False, f"Action blocked: {decision['reason']}") return ## Step 6: Verify logger.info("Step 6: Verifying remediation...") verification = verify_remediation( action_name, service_name, action_vars.get("namespace", namespace), metrics, ) event_type = "verification_passed" if verification["resolved"] else "verification_failed" log_event(event_type, service_name, action_name, verification) if not verification["resolved"]: logger.warning("⚠️ Verification failed — escalating to human") await send_action_result( action_name, service_name, False, f"⚠️ Verification failed after {action_name}.\n" f"{verification['message']}\n" f"Human investigation required." ) else: logger.info("✅ Incident resolved automatically") await send_action_result( action_name, service_name, True, f"Incident resolved by {action_name}.\n{verification['message']}" ) async def run_orchestrator(): """The main async loop.""" model, scaler, feature_columns = load_detection_model() logger.info("=" * 55) logger.info("🚀 AIOps Self-Healing System Started") logger.info(f" Detection interval: {DETECTION_INTERVAL_SECONDS}s") logger.info(f" Consecutive anomalies before action: {ANOMALY_CONSECUTIVE_THRESHOLD}") logger.info("=" * 55) consecutive_anomalies = 0 currently_remediating = False while True: try: cycle_time = datetime.now().strftime("%H:%M:%S") is_anomaly, score, metrics = run_detection(model, scaler, feature_columns) status = "⚠️ ANOMALY" if is_anomaly else "✅ normal" logger.info(f"[{cycle_time}] Score: {score:.4f} ({status})") if is_anomaly: consecutive_anomalies += 1 logger.warning(f" Consecutive anomalies: {consecutive_anomalies}/{ANOMALY_CONSECUTIVE_THRESHOLD}") if consecutive_anomalies >= ANOMALY_CONSECUTIVE_THRESHOLD and not currently_remediating: currently_remediating = True try: await handle_anomaly(metrics, score) finally: currently_remediating = False consecutive_anomalies = 0 else: consecutive_anomalies = 0 except KeyboardInterrupt: logger.info("\n⛔ Orchestrator stopped by user") break except Exception as e: logger.error(f"Orchestrator error: {e}", exc_info=True) await asyncio.sleep(DETECTION_INTERVAL_SECONDS) if __name__ == "__main__": asyncio.run(run_orchestrator())Testing the Full Pipeline End to End
Run the complete self-healing demo
## Terminal 1: Start Ollamaollama serve ## Terminal 2: Port-forward Prometheuskubectl port-forward -n monitoring svc/prometheus-operated 9090:9090 & ## Terminal 3: Start the orchestratorpython3 -m src.orchestrator ## Terminal 4: Inject a CPU stress to trigger detectionkubectl run cpu-stress \ --image=containerstack/cpustress \ --restart=Never \ -- --cpu 4 --timeout 180s ## Watch Terminal 3 for the full pipeline:#### [14:23:01] Score: -0.0823 (✅ normal)## [14:24:01] Score: -0.0891 (✅ normal)## [14:25:01] Score: -0.3122 (⚠️ ANOMALY)## Consecutive anomalies: 1/2## [14:26:01] Score: -0.2987 (⚠️ ANOMALY)## Consecutive anomalies: 2/2## ========================================================## 🚨 ANOMALY DETECTED — starting remediation pipeline## Step 1: Collecting infrastructure context...## Step 2: Searching runbooks...## Step 3: Getting LLM diagnosis...## Diagnosis: Elevated CPU usage detected across cluster...## Recommended action: scale_up_deployment (confidence: 82%)## Step 4: Running decision engine...## Decision: autonomous## Reason: 'scale_up_deployment' is a Tier 1 action...## Step 5: Executing scale_up_deployment...## ✅ Playbook succeeded: scale_up_deployment.yaml## Step 6: Verifying remediation...## ✅ Incident resolved automaticallyVerify the audit log
## Check what was written to the audit logcat audit/actions.jsonl | python3 -m json.tool | grep event_type ## Expected:## "event_type": "anomaly_detected"## "event_type": "diagnosis_complete"## "event_type": "action_decided"## "event_type": "action_executed"## "event_type": "verification_passed" ## Count successful autonomous remediationsgrep '"outcome": "success"' audit/actions.jsonl | wc -l ## Check if any approvals were requestedgrep '"decision": "approval_required"' audit/actions.jsonl | wc -lTest a Tier 3 action manually
## Simulate a diagnosis that recommends rollback (Tier 3)python3 -c "import asynciofrom src.orchestrator import handle_anomaly ## Fake metrics that would suggest a rollbackmetrics = { 'cpu_usage': 0.15, 'memory_usage': 200000000, 'http_request_rate': 500, 'http_error_rate': 0.45, ## 45% error rate strongly suggests bad deployment 'response_time_p95': 8.5,} asyncio.run(handle_anomaly(metrics, -0.41))" ## The system should detect the high error rate,## diagnose a bad deployment, and REQUEST APPROVAL for rollback## Check the Slack output (or terminal if Slack not configured)Deploy to Kubernetes
## k8s/deployment.yamlapiVersion: apps/v1kind: Deploymentmetadata: name: aiops-self-healing namespace: monitoringspec: replicas: 1 selector: matchLabels: app: aiops-self-healing template: metadata: labels: app: aiops-self-healing spec: ## The system needs to run kubectl and Ansible against the cluster ## This ServiceAccount grants it the minimum required permissions serviceAccountName: aiops-self-healing containers: - name: orchestrator image: your-registry/aiops-self-healing:v1.0.0 env: - name: PROMETHEUS_URL value: "http://prometheus-operated.monitoring.svc:9090" - name: OLLAMA_URL value: "http://ollama.monitoring.svc:11434" - name: SLACK_WEBHOOK_URL valueFrom: secretKeyRef: name: aiops-secrets key: slack-webhook-url resources: requests: cpu: "200m" memory: "768Mi" limits: cpu: "1000m" memory: "1.5Gi" volumeMounts: - name: audit-storage mountPath: /app/audit - name: model-storage mountPath: /app/models volumes: - name: audit-storage persistentVolumeClaim: claimName: aiops-audit-pvc - name: model-storage persistentVolumeClaim: claimName: aiops-model-pvc## RBAC: the system needs permission to manage pods and deploymentsapiVersion: v1kind: ServiceAccountmetadata: name: aiops-self-healing namespace: monitoringapiVersion: rbac.authorization.k8s.io/v1kind: ClusterRolemetadata: name: aiops-self-healingrules: - apiGroups: ["apps"] resources: ["deployments", "deployments/scale"] verbs: ["get", "list", "patch", "update"] - apiGroups: [""] resources: ["pods", "pods/log", "events", "nodes"] verbs: ["get", "list", "delete"]apiVersion: rbac.authorization.k8s.io/v1kind: ClusterRoleBindingmetadata: name: aiops-self-healingsubjects: - kind: ServiceAccount name: aiops-self-healing namespace: monitoringroleRef: kind: ClusterRole name: aiops-self-healing apiGroup: rbac.authorization.k8s.io## Apply RBAC and deploymentkubectl apply -f k8s/deployment.yaml ## Verify the system is runningkubectl get pods -n monitoring | grep aiops-self-healingkubectl logs -n monitoring deployment/aiops-self-healing -fProduction Checklist
## ─── 1. All three capstone components are working ─────────────## Capstone 1: model loadspython3 -c "import joblib; pkg = joblib.load('$(python3 -c \"from src.config import MODEL_PATH; print(MODEL_PATH)\")'); print('Model OK:', pkg['feature_columns'])" ## Capstone 2: runbook index is populatedpython3 -c "import chromadb, syssys.path.insert(0, '../aiops-incident-agent')from src.config import CHROMA_DB_PATH, CHROMA_COLLECTION_NAMEc = chromadb.PersistentClient(path=CHROMA_DB_PATH)col = c.get_collection(CHROMA_COLLECTION_NAME)print(f'Runbooks indexed: {col.count()} chunks')" ## ─── 2. Decision engine respects autonomy tiers ───────────────python3 -c "from src.decision_engine import decide ## Tier 1 should be autonomousd = decide('restart_pod', 0.9, 'test-service', {})assert d['decision'] == 'autonomous', f'Expected autonomous, got {d[\"decision\"]}' ## Low confidence should require approval even for Tier 1d = decide('restart_pod', 0.5, 'test-service', {})assert d['decision'] == 'approval_required', f'Expected approval_required, got {d[\"decision\"]}' ## Tier 3 should always require approvald = decide('rollback_deployment', 0.99, 'test-service', {})assert d['decision'] == 'approval_required', f'Expected approval_required, got {d[\"decision\"]}' print('✅ Decision engine autonomy tiers are correct')" ## ─── 3. Rate limiting is working ─────────────────────────────python3 -c "from src.decision_engine import decide, _record_actionfrom src.config import MAX_ACTIONS_PER_SERVICE_PER_HOUR ## Exhaust the rate limitfor i in range(MAX_ACTIONS_PER_SERVICE_PER_HOUR): _record_action('rate-test-service') d = decide('restart_pod', 0.95, 'rate-test-service', {})assert d['decision'] == 'blocked', f'Expected blocked, got {d[\"decision\"]}'print('✅ Rate limiting is working')" ## ─── 4. Audit log is being written ───────────────────────────ls -lh audit/actions.jsonl## Should exist after running the orchestratortail -3 audit/actions.jsonl | python3 -m json.tool ## ─── 5. Ansible playbooks are valid ─────────────────────────for playbook in ansible/playbooks/*.yaml; do ansible-playbook --syntax-check "$playbook" -i ansible/inventory/hosts.yaml echo "✅ $playbook syntax OK"done echo "✅ All production checks passed"Common Production Mistakes
Adding irreversible actions to the autonomous tier. The most dangerous mistake you can make with this system is putting rollback_deployment or scale_down_deployment in AUTONOMOUS_ACTIONS. A wrong diagnosis combined with autonomous rollback can take a working deployment down at 2 AM and cause more damage than the original anomaly. The rule is: if the action cannot be undone in 60 seconds, it goes in APPROVAL_REQUIRED_ACTIONS. Always.
Not setting a rate limit and creating an action loop. A broken pod that keeps crashing will keep triggering the detection model. Without MAX_ACTIONS_PER_SERVICE_PER_HOUR, the system will restart it again and again in an infinite loop. Three restarts in an hour is usually the right limit. If the same service needs more than three restarts in an hour, the underlying cause is not something a restart fixes — a human needs to investigate.
Running two replicas of the orchestrator. With replicas: 2, both pods detect the same anomaly and both start the remediation pipeline simultaneously. You get two Ansible playbook executions, two scale-up operations, and two Slack messages. Keep this at exactly one replica and use a liveness probe to handle pod failures.
Skipping the verification step during testing. The verification step waits 90 seconds and then checks whether the problem resolved. This is slow during development. Developers often set VERIFICATION_WAIT_SECONDS = 0 to speed up testing and then forget to restore it for production. If verification is skipped, the system never learns when its remediations fail — it will report success for actions that did nothing. Keep the wait at 90 seconds in production minimum.
Training the detection model during a past incident. The Isolation Forest model from Capstone 1 learns what "normal" looks like from the training data. If you train it on a 24-hour window that included an incident, it learns that incident conditions are normal. The self-healing system will never trigger during the same type of incident because the model thinks it is expected behaviour. Before training, check your monitoring dashboards to confirm the training window was genuinely incident-free.
Not testing the approval workflow before it matters. The Tier 3 approval flow goes untested until a real production incident requires a rollback. Then it fails because the Slack bot token expired, or the approval endpoint is unreachable, or nobody knows the curl command format. Run a monthly drill: force a Tier 3 action and verify the approval flow works end to end.
Quick Reference
| Component | File | What it Does |
|---|---|---|
| Anomaly detection | Capstone 1 model | Runs every 60s, flags unusual metrics |
| RAG search | Capstone 2 ChromaDB | Finds relevant runbooks |
| LLM reasoning | orchestrator.py |
Diagnoses cause, picks action |
| Decision engine | decision_engine.py |
Decides autonomous vs approval |
| Ansible executor | ansible_executor.py |
Runs the remediation playbook |
| Verifier | verifier.py |
Confirms the fix worked |
| Audit logger | audit_logger.py |
Records everything permanently |
| Action | Tier | Approval Needed |
|---|---|---|
restart_pod |
1 | No |
scale_up_deployment |
1 | No |
flush_redis_cache |
1 | No |
increase_memory_limit |
2 | No (notifies after) |
rolling_restart_deployment |
2 | No (notifies after) |
rollback_deployment |
3 | Yes |
scale_down_deployment |
3 | Yes |
| Command | What it Does |
|---|---|
python3 -m src.orchestrator |
Start the full self-healing loop |
cat audit/actions.jsonl | python3 -m json.tool |
Read the audit log |
grep '"outcome": "failure"' audit/actions.jsonl |
Find failed remediations |
ansible-playbook --syntax-check playbook.yaml -i hosts.yaml |
Validate a playbook |
kubectl logs deployment/aiops-self-healing -f -n monitoring |
Watch live logs in production |
Videos & Guides
No external walkthroughs, video tutorials, or reference files attached to this blueprint yet.