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.

Related Concepts & TermsPipePipeline

Domains & Technologies

Domains
CAPSTONEAIOPSSELF-HEALINGANSIBLEAUTONOMOUS-REMEDIATION
Technologies
KUBERNETESPYTHON

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:

  1. Get paged and wake up (or stop what they are doing)
  2. Open the monitoring dashboard
  3. Figure out what is wrong
  4. Look up the runbook
  5. Run the remediation commands
  6. 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
Bash
## Verify Capstone 1 model exists
ls -lh models/isolation_forest.joblib
## Should show the file
## Verify Capstone 2 runbooks are indexed
python3 -c "
import chromadb
c = chromadb.PersistentClient(path='data/chroma')
col = c.get_collection('runbooks')
print(f'Runbooks indexed: {col.count()} chunks')
"
## Verify Ollama is running
curl -s http://localhost:11434/api/tags | python3 -m json.tool | grep name
## Should show at least one model
## Install Ansible
pip3 install ansible --break-system-packages
ansible --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.

Bash
┌─────────────────────────────────────────────────────────────┐
│ 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

Bash
## This capstone lives in a new directory but imports from capstones 1 and 2
mkdir aiops-self-healing && cd aiops-self-healing
mkdir -p src ansible/playbooks ansible/inventory logs audit k8s
touch src/config.py
touch src/orchestrator.py
touch src/decision_engine.py
touch src/ansible_executor.py
touch src/verifier.py
touch src/audit_logger.py
touch src/slack_approvals.py
touch requirements.txt
echo "✅ Project structure created"
◈ DIAGRAM
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
Bash
# requirements.txt
# Inherits all deps from capstones 1 and 2
pandas==2.1.0
scikit-learn==1.3.0
requests==2.31.0
chromadb==0.4.18
sentence-transformers==2.2.2
joblib==1.3.2
fastapi==0.104.1
uvicorn==0.24.0
ansible-runner==2.3.4
schedule==1.2.0
Bash
pip3 install -r requirements.txt
echo "✅ Dependencies installed"
Configuration — Autonomy Settings

The config file controls what the system is allowed to do

PYTHON
# src/config.py
import 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 remediation
ANOMALY_CONSECUTIVE_THRESHOLD = 2
## ── Slack ─────────────────────────────────────────────────────
SLACK_WEBHOOK_URL = os.getenv("SLACK_WEBHOOK_URL", "")
## For interactive approvals, you need a Slack bot token
SLACK_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 needed
AUTONOMOUS_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 executing
APPROVAL_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 actions
AUTONOMOUS_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 forever
MAX_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 worked
VERIFICATION_WAIT_SECONDS = 90
Remember

The AUTONOMOUS_ACTIONS list is your blast radius control. Start it small — just restart_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."

Bash
## Create the pod restart playbook
cat > 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"
Bash
## Create the scale-up playbook
cat > 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"
Bash
## Create the memory limit increase playbook
cat > 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"
Bash
## 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"
Bash
## Install the Kubernetes Ansible collection
ansible-galaxy collection install kubernetes.core
## Create the inventory file
cat > 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

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_runner
import os
import json
import logging
from 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

PYTHON
# 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 json
import logging
from datetime import datetime, timedelta
from collections import defaultdict
from 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.

PYTHON
# 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 json
import os
import logging
from datetime import datetime
from 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 events
Building the Verifier

Confirming the fix actually worked

PYTHON
# 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 time
import logging
from src.config import VERIFICATION_WAIT_SECONDS, PROMETHEUS_URL
import 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

PYTHON
# 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 requests
import logging
import asyncio
from 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}}'

Bash
*To reject, run:*

curl -X POST http://localhost:8081/approve -d '{{"action": "{action_name}", "service": "{service_name}", "approved": false}}'

◈ DIAGRAM
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

PYTHON
# 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 asyncio
import joblib
import json
import logging
import os
import re
import time
import numpy as np
import pandas as pd
import requests
from datetime import datetime
from src.config import (
PROMETHEUS_URL, MODEL_PATH, DETECTION_INTERVAL_SECONDS,
ANOMALY_CONSECUTIVE_THRESHOLD, OLLAMA_URL, LLM_MODEL,
)
from src.decision_engine import decide
from src.ansible_executor import execute_action
from src.verifier import verify_remediation
from src.audit_logger import log_event
from src.slack_approvals import request_approval, send_action_result
## Import RAG and context from Capstone 2
import sys
sys.path.insert(0, "../aiops-incident-agent")
from src.rag_retriever import search_runbooks, format_runbooks_for_prompt
from 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

Bash
## Terminal 1: Start Ollama
ollama serve
## Terminal 2: Port-forward Prometheus
kubectl port-forward -n monitoring svc/prometheus-operated 9090:9090 &
## Terminal 3: Start the orchestrator
python3 -m src.orchestrator
## Terminal 4: Inject a CPU stress to trigger detection
kubectl 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 automatically

Verify the audit log

Bash
## Check what was written to the audit log
cat 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 remediations
grep '"outcome": "success"' audit/actions.jsonl | wc -l
## Check if any approvals were requested
grep '"decision": "approval_required"' audit/actions.jsonl | wc -l

Test a Tier 3 action manually

Bash
## Simulate a diagnosis that recommends rollback (Tier 3)
python3 -c "
import asyncio
from src.orchestrator import handle_anomaly
## Fake metrics that would suggest a rollback
metrics = {
'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
YAML
## k8s/deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: aiops-self-healing
namespace: monitoring
spec:
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 deployments
apiVersion: v1
kind: ServiceAccount
metadata:
name: aiops-self-healing
namespace: monitoring
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
name: aiops-self-healing
rules:
- 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/v1
kind: ClusterRoleBinding
metadata:
name: aiops-self-healing
subjects:
- kind: ServiceAccount
name: aiops-self-healing
namespace: monitoring
roleRef:
kind: ClusterRole
name: aiops-self-healing
apiGroup: rbac.authorization.k8s.io
Bash
## Apply RBAC and deployment
kubectl apply -f k8s/deployment.yaml
## Verify the system is running
kubectl get pods -n monitoring | grep aiops-self-healing
kubectl logs -n monitoring deployment/aiops-self-healing -f
Production Checklist
Bash
## ─── 1. All three capstone components are working ─────────────
## Capstone 1: model loads
python3 -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 populated
python3 -c "
import chromadb, sys
sys.path.insert(0, '../aiops-incident-agent')
from src.config import CHROMA_DB_PATH, CHROMA_COLLECTION_NAME
c = 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 autonomous
d = 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 1
d = decide('restart_pod', 0.5, 'test-service', {})
assert d['decision'] == 'approval_required', f'Expected approval_required, got {d[\"decision\"]}'
## Tier 3 should always require approval
d = 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_action
from src.config import MAX_ACTIONS_PER_SERVICE_PER_HOUR
## Exhaust the rate limit
for 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 orchestrator
tail -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.