Alert Automation and Remediation: From Alert Storms to Self-Healing Systems

┌─────────────┐     ┌─────────────┐     ┌─────────────┐     ┌─────────────┐     ┌─────────────┐
│  1. Detect   │────▶│  2. Dedup   │────▶│  3. Route   │────▶│  4. Remediate│────▶│  5. Review  │
│  Detection  │     │  Dedup      │     │  Route     │     │  Remediate │     │  Review    │
└─────────────┘     └─────────────┘     └─────────────┘     └─────────────┘     └─────────────┘
     │                    │                   │                   │                   │
  Monitor           Dedup/Aggregate     Severity/Dispatch    AutoRunbook/Human    Archive/Improve

# alertmanager.yml — Grouping configuration
route:
  group_by: ['alertname', 'cluster', 'service']
  group_wait: 30s        # Wait 30s after first alert, collecting same-group alerts
  group_interval: 5m     # Send interval for same-group alerts
  repeat_interval: 4h    # Repeat notification interval for unresolved alerts
  receiver: 'default'

receivers:
  - name: 'default'
    webhook_configs:
      - url: 'http://alert-router:8080/alert'

# alertmanager.yml — Inhibition rules
inhibit_rules:
  # When a node is down, suppress all service alerts for that node
  - source_match:
      alertname: 'NodeDown'
    target_match_re:
      alertname: '.*(Service|Pod|Container).*Down|CrashLoopBackOff'
    equal: ['instance']

  # When cluster is unavailable, suppress all alerts for that cluster
  - source_match:
      severity: 'critical'
      alertname: 'ClusterUnavailable'
    target_match_re:
      severity: 'warning|info'
    equal: ['cluster']

  # During MySQL master switch, suppress replica lag alerts
  - source_match:
      alertname: 'MySQLMasterSwitch'
    target_match:
      alertname: 'MySQLReplicationLag'
    equal: ['cluster', 'service']

#!/usr/bin/env python3
"""Alert fingerprint deduplication engine — generates unique fingerprints for cross-system dedup"""

import hashlib
import time
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class Alert:
    """Alert data structure"""
    alertname: str
    severity: str
    service: str
    instance: str
    cluster: str
    message: str
    labels: dict = field(default_factory=dict)
    fingerprint: str = ""
    first_seen: float = 0
    last_seen: float = 0
    count: int = 0

    def compute_fingerprint(self) -> str:
        """Compute alert fingerprint (based on structural fields, ignoring timestamps and dynamic values)"""
        key_fields = f"{self.alertname}:{self.service}:{self.instance}:{self.cluster}:{self.severity}"
        return hashlib.md5(key_fields.encode()).hexdigest()[:16]

class AlertDeduplicator:
    """Alert deduplicator"""

    def __init__(self, dedup_window: int = 300):
        """
        Args:
            dedup_window: Dedup window (seconds), same fingerprint only kept once within window
        """
        self.dedup_window = dedup_window
        self.alert_store: dict[str, Alert] = {}  # fingerprint -> Alert

    def process(self, alert: Alert) -> Optional[Alert]:
        """
        Process alert, return alert to send (after dedup)
        Returns None if alert is duplicate and suppressed
        """
        alert.fingerprint = alert.compute_fingerprint()
        now = time.time()

        if alert.fingerprint in self.alert_store:
            existing = self.alert_store[alert.fingerprint]
            existing.last_seen = now
            existing.count += 1

            # Within dedup window, suppress duplicate
            if now - existing.first_seen < self.dedup_window:
                return None  # Suppress
            else:
                # Beyond dedup window, treat as new alert
                existing.first_seen = now
                existing.count = 1
                return existing
        else:
            alert.first_seen = now
            alert.last_seen = now
            alert.count = 1
            self.alert_store[alert.fingerprint] = alert
            return alert

    def get_active_alerts(self) -> list[Alert]:
        """Get currently active alert list"""
        now = time.time()
        return [
            a for a in self.alert_store.values()
            if now - a.last_seen < self.dedup_window * 2
        ]

    def cleanup(self, max_age: int = 3600):
        """Clean up expired alert records"""
        now = time.time()
        expired = [
            fp for fp, alert in self.alert_store.items()
            if now - alert.last_seen > max_age
        ]
        for fp in expired:
            del self.alert_store[fp]

if __name__ == "__main__":
    dedup = AlertDeduplicator(dedup_window=300)

    # Simulate alert storm
    alerts = [
        Alert("HighCPU", "warning", "api-gw", "10.0.1.1", "prod", "CPU 92%"),
        Alert("HighCPU", "warning", "api-gw", "10.0.1.1", "prod", "CPU 95%"),  # Duplicate
        Alert("HighCPU", "warning", "api-gw", "10.0.1.1", "prod", "CPU 98%"),  # Duplicate
        Alert("HighMemory", "critical", "api-gw", "10.0.1.1", "prod", "Memory 95%"),
        Alert("DiskFull", "critical", "db", "10.0.2.1", "prod", "Disk 85%"),
    ]

    for alert in alerts:
        result = dedup.process(alert)
        if result:
            print(f"[SEND] {alert.alertname} | {alert.service} | {alert.instance} | fp={alert.fingerprint}")
        else:
            print(f"[SUPPRESS] {alert.alertname} | {alert.service} | {alert.instance} | duplicate")

    print(f"\nActive alerts: {len(dedup.get_active_alerts())}")

# alertmanager.yml — Silence rules (can also be created dynamically via API)
# Create silence via API
# curl -X POST http://alertmanager:9093/api/v2/silences \
#   -H "Content-Type: application/json" \
#   -d '{
#     "matchers": [
#       {"name": "service", "value": "payment-service", "isRegex": false}
#     ],
#     "startsAt": "2026-07-11T02:00:00+08:00",
#     "endsAt": "2026-07-11T04:00:00+08:00",
#     "createdBy": "ops-team",
#     "comment": "Database maintenance window"
#   }'

#!/bin/bash
# schedule-silence.sh — Auto-create alert silence (for maintenance windows)

ALERTMANAGER="http://alertmanager:9093"
SERVICE="${1:-payment-service}"
DURATION="${2:-120}"  # minutes

START_TIME=$(date -u -d "+1 minute" '+%Y-%m-%dT%H:%M:%S.000Z')
END_TIME=$(date -u -d "+${DURATION} minutes" '+%Y-%m-%dT%H:%M:%S.000Z')

curl -s -X POST "${ALERTMANAGER}/api/v2/silences" \
  -H "Content-Type: application/json" \
  -d "{
    \"matchers\": [
      {\"name\": \"service\", \"value\": \"${SERVICE}\", \"isRegex\": false}
    ],
    \"startsAt\": \"${START_TIME}\",
    \"endsAt\": \"${END_TIME}\",
    \"createdBy\": \"automation\",
    \"comment\": \"Auto-silence: ${SERVICE} maintenance ${DURATION}min\"
  }"

echo "Created ${DURATION}min silence window for ${SERVICE}"

#!/usr/bin/env python3
"""Multi-dimensional alert correlation engine — correlates alerts from different sources into events"""

import time
from dataclasses import dataclass, field
from typing import Optional
from collections import defaultdict

@dataclass
class AlertEvent:
    """Correlated alert event"""
    event_id: str
    service: str
    cluster: str
    severity: str  # Takes highest severity
    alerts: list = field(default_factory=list)
    first_seen: float = 0
    last_seen: float = 0
    status: str = "firing"  # firing / resolved

    def add_alert(self, alert: dict):
        self.alerts.append(alert)
        self.last_seen = time.time()
        severity_order = {"info": 0, "warning": 1, "critical": 2, "fatal": 3}
        if severity_order.get(alert.get("severity", "info"), 0) > \
           severity_order.get(self.severity, 0):
            self.severity = alert["severity"]

class AlertCorrelator:
    """Alert correlator — correlates alerts based on service, cluster, time window"""

    CORRELATION_WINDOW = 300  # 5-minute window

    def __init__(self):
        self.events: dict[str, AlertEvent] = {}  # correlation_key -> AlertEvent

    def _correlation_key(self, alert: dict) -> str:
        """Generate correlation key"""
        service = alert.get("labels", {}).get("service", "unknown")
        cluster = alert.get("labels", {}).get("cluster", "unknown")
        return f"{service}:{cluster}"

    def process(self, alert: dict) -> Optional[AlertEvent]:
        """Process alert, return correlated event"""
        key = self._correlation_key(alert)
        now = time.time()

        if key in self.events:
            event = self.events[key]
            # Check if within correlation window
            if now - event.last_seen <= self.CORRELATION_WINDOW:
                event.add_alert(alert)
                return event
            else:
                # Window expired, create new event
                event.status = "resolved"

        # Create new event
        event_id = f"EVT-{int(now)}-{key.replace(':', '-')}"
        event = AlertEvent(
            event_id=event_id,
            service=alert.get("labels", {}).get("service", "unknown"),
            cluster=alert.get("labels", {}).get("cluster", "unknown"),
            severity=alert.get("severity", "info"),
            first_seen=now,
            last_seen=now,
        )
        event.add_alert(alert)
        self.events[key] = event
        return event

if __name__ == "__main__":
    correlator = AlertCorrelator()

    # Simulate multi-source alerts
    raw_alerts = [
        {"alertname": "HighCPU", "severity": "warning",
         "labels": {"service": "api-gw", "cluster": "prod"}},
        {"alertname": "HighLatency", "severity": "critical",
         "labels": {"service": "api-gw", "cluster": "prod"}},  # Correlated
        {"alertname": "ErrorRateHigh", "severity": "critical",
         "labels": {"service": "api-gw", "cluster": "prod"}},  # Correlated
        {"alertname": "DiskFull", "severity": "critical",
         "labels": {"service": "db", "cluster": "prod"}},       # Different service
    ]

    for alert in raw_alerts:
        event = correlator.process(alert)
        print(f"Alert: {alert['alertname']:20s} → Event: {event.event_id} | "
              f"Severity: {event.severity:8s} | Correlated: {len(event.alerts)}")

#!/usr/bin/env python3
"""Alert routing engine — rule-based alert classification and dispatch"""

import re
import json
from dataclasses import dataclass, field
from typing import Optional
from enum import Enum

class Severity(Enum):
    P0 = "critical"
    P1 = "high"
    P2 = "medium"
    P3 = "low"

@dataclass
class RouteRule:
    """Routing rule"""
    name: str
    match_labels: dict           # Labels to match
    match_severity: list[str]    # Severities to match
    receivers: list[str]         # Receiver list
    escalate_after: int = 0     # Escalation time if no response (seconds)
    escalate_to: list[str] = field(default_factory=list)  # Escalation receivers
    auto_remediation: str = ""   # Associated auto-remediation Runbook

class AlertRouter:
    """Alert router"""

    def __init__(self):
        self.rules: list[RouteRule] = []
        self.default_receivers = ["on-call-team"]

    def add_rule(self, rule: RouteRule):
        self.rules.append(rule)

    def route(self, alert: dict) -> dict:
        """Route alert to correct receivers and handling flow"""
        severity = alert.get("severity", "info")
        labels = alert.get("labels", {})

        for rule in self.rules:
            if self._match(alert, rule):
                return {
                    "alert": alert,
                    "severity": severity,
                    "receivers": rule.receivers,
                    "escalate_after": rule.escalate_after,
                    "escalate_to": rule.escalate_to,
                    "auto_remediation": rule.auto_remediation,
                    "routed_at": alert.get("startsAt", ""),
                }

        return {
            "alert": alert,
            "severity": severity,
            "receivers": self.default_receivers,
            "escalate_after": 0,
            "escalate_to": [],
            "auto_remediation": "",
            "routed_at": alert.get("startsAt", ""),
        }

    def _match(self, alert: dict, rule: RouteRule) -> bool:
        """Check if alert matches routing rule"""
        severity = alert.get("severity", "info")
        labels = alert.get("labels", {})

        if rule.match_severity and severity not in rule.match_severity:
            return False

        for key, value in rule.match_labels.items():
            if key not in labels:
                return False
            if isinstance(value, str) and value.startswith("~"):
                pattern = value[1:]
                if not re.match(pattern, str(labels.get(key, ""))):
                    return False
            elif str(labels.get(key, "")) != str(value):
                return False

        return True

if __name__ == "__main__":
    router = AlertRouter()

    # Define routing rules
    router.add_rule(RouteRule(
        name="Database P0",
        match_labels={"service": "mysql", "cluster": "prod"},
        match_severity=[Severity.P0.value],
        receivers=["on-call-dba", "on-call-sre", "tech-lead"],
        escalate_after=300,  # Escalate after 5 min no response
        escalate_to=["cto", "vp-engineering"],
        auto_remediation="runbook:mysql-failover",
    ))

    router.add_rule(RouteRule(
        name="API Service P1",
        match_labels={"service": "~api-.*", "cluster": "prod"},
        match_severity=[Severity.P1.value, Severity.P0.value],
        receivers=["on-call-sre"],
        escalate_after=600,
        escalate_to=["tech-lead"],
        auto_remediation="runbook:api-restart",
    ))

    router.add_rule(RouteRule(
        name="Disk Space Warning",
        match_labels={"alertname": "DiskSpaceWarning"},
        match_severity=[Severity.P2.value, Severity.P3.value],
        receivers=["ops-team-im"],
        escalate_after=0,
        auto_remediation="runbook:disk-cleanup",
    ))

    # Test routing
    test_alerts = [
        {"alertname": "MySQLDown", "severity": "critical",
         "labels": {"service": "mysql", "cluster": "prod"},
         "startsAt": "2026-07-11T02:25:59+08:00"},
        {"alertname": "HighErrorRate", "severity": "high",
         "labels": {"service": "api-gateway", "cluster": "prod"},
         "startsAt": "2026-07-11T02:25:59+08:00"},
        {"alertname": "DiskSpaceWarning", "severity": "medium",
         "labels": {"service": "web", "cluster": "prod", "instance": "10.0.1.5"},
         "startsAt": "2026-07-11T02:25:59+08:00"},
    ]

    for alert in test_alerts:
        result = router.route(alert)
        print(f"\nAlert: {alert['alertname']} ({alert['severity']})")
        print(f"  Receivers: {result['receivers']}")
        print(f"  Auto-remediation: {result['auto_remediation']}")
        print(f"  Escalation: {result['escalate_after']}s → {result['escalate_to']}")

#!/usr/bin/env python3
"""Alert escalation manager — tracks alert response status, auto-escalates on timeout"""

import time
import threading
from dataclasses import dataclass, field

@dataclass
class EscalationPolicy:
    """Escalation policy"""
    initial_receivers: list[str]
    escalate_after: int           # seconds
    escalate_to: list[str]        # escalation receivers
    max_escalations: int = 3      # max escalation levels
    auto_resolve_on_action: bool = True  # stop escalation on acknowledgment

@dataclass
class AlertEscalation:
    """Alert escalation tracking"""
    alert_id: str
    policy: EscalationPolicy
    created_at: float
    acknowledged: bool = False
    current_level: int = 0
    escalation_history: list = field(default_factory=list)

    def acknowledge(self, user: str):
        self.acknowledged = True
        self.escalation_history.append({
            "action": "acknowledged",
            "user": user,
            "timestamp": time.time(),
        })

    def escalate(self) -> list[str]:
        """Execute escalation, return new receiver list"""
        if self.acknowledged:
            return []

        self.current_level += 1
        if self.current_level > self.policy.max_escalations:
            return ["final-escalation", "cto"]

        receivers = self.policy.escalate_to[:self.current_level]
        self.escalation_history.append({
            "action": "escalated",
            "level": self.current_level,
            "receivers": receivers,
            "timestamp": time.time(),
        })
        return receivers

class EscalationManager:
    """Alert escalation manager"""

    def __init__(self):
        self.tracked: dict[str, AlertEscalation] = {}
        self._timer: threading.Timer = None

    def track(self, alert_id: str, policy: EscalationPolicy):
        """Start tracking alert escalation"""
        escalation = AlertEscalation(
            alert_id=alert_id,
            policy=policy,
            created_at=time.time(),
        )
        self.tracked[alert_id] = escalation

        # Set escalation timer
        timer = threading.Timer(
            policy.escalate_after,
            self._check_escalation,
            args=[alert_id]
        )
        timer.daemon = True
        timer.start()

    def _check_escalation(self, alert_id: str):
        """Check if escalation is needed"""
        if alert_id not in self.tracked:
            return

        escalation = self.tracked[alert_id]
        if escalation.acknowledged:
            return

        new_receivers = escalation.escalate()
        if new_receivers:
            print(f"[ESCALATE] Alert {alert_id} → level {escalation.current_level}")
            print(f"  New receivers: {new_receivers}")
            # Trigger actual notification sending here

    def acknowledge(self, alert_id: str, user: str):
        """Acknowledge alert"""
        if alert_id in self.tracked:
            self.tracked[alert_id].acknowledge(user)
            print(f"[ACK] Alert {alert_id} acknowledged by {user}")

if __name__ == "__main__":
    manager = EscalationManager()

    policy = EscalationPolicy(
        initial_receivers=["on-call-sre"],
        escalate_after=5,  # 5 seconds (demo, should be 300-600 in production)
        escalate_to=["tech-lead", "vp-engineering", "cto"],
        max_escalations=3,
    )

    manager.track("EVT-001", policy)
    print("Alert tracked, awaiting response...")
    print("(Not acknowledged, auto-escalating in 5 seconds)")

    time.sleep(8)  # Wait for escalation trigger

    # Simulate acknowledgment
    # manager.acknowledge("EVT-001", "John")

#!/usr/bin/env python3
"""Runbook registry — manages and executes automated remediation scripts"""

import subprocess
import logging
import time
import json
import os
from dataclasses import dataclass, field
from typing import Optional, Callable
from enum import Enum

logger = logging.getLogger(__name__)

class RunbookStatus(Enum):
    SUCCESS = "success"
    FAILED = "failed"
    PARTIAL = "partial"
    SKIPPED = "skipped"

class RunbookRisk(Enum):
    SAFE = "safe"           # Safe operation, auto-execute
    CAUTION = "caution"     # Use with caution, recommend human confirmation
    DANGEROUS = "dangerous" # High risk, must confirm manually

@dataclass
class RunbookResult:
    """Runbook execution result"""
    runbook_name: str
    status: RunbookStatus
    message: str
    execution_time: float
    output: str = ""
    actions_taken: list = field(default_factory=list)

@dataclass
class Runbook:
    """Runbook definition"""
    name: str
    description: str
    risk_level: RunbookRisk
    match_conditions: dict       # Alert conditions to match
    handler: Callable            # Handler function
    max_retries: int = 1        # Max retry count
    timeout: int = 60           # Timeout (seconds)
    cooldown: int = 300         # Cooldown between executions (seconds)
    last_executed: float = 0    # Last execution timestamp

    def matches(self, alert: dict) -> bool:
        """Check if alert matches this Runbook"""
        labels = alert.get("labels", {})
        for key, value in self.match_conditions.items():
            if key not in labels:
                return False
            if str(labels[key]) != str(value):
                return False
        return True

    def can_execute(self) -> bool:
        """Check if can execute (cooldown check)"""
        if self.last_executed == 0:
            return True
        return time.time() - self.last_executed >= self.cooldown

    def execute(self, alert: dict) -> RunbookResult:
        """Execute the Runbook"""
        if not self.can_execute():
            return RunbookResult(
                runbook_name=self.name,
                status=RunbookStatus.SKIPPED,
                message=f"Runbook in cooldown ({self.cooldown}s)",
                execution_time=0,
            )

        self.last_executed = time.time()
        start = time.time()

        for attempt in range(self.max_retries + 1):
            try:
                result = self.handler(alert)
                elapsed = time.time() - start

                if result.get("success", False):
                    return RunbookResult(
                        runbook_name=self.name,
                        status=RunbookStatus.SUCCESS,
                        message=result.get("message", "Success"),
                        execution_time=elapsed,
                        output=result.get("output", ""),
                        actions_taken=result.get("actions", []),
                    )
                else:
                    if attempt < self.max_retries:
                        logger.warning(f"Runbook {self.name} attempt {attempt+1} failed, retrying...")
                        time.sleep(2 ** attempt)  # Exponential backoff
                    else:
                        return RunbookResult(
                            runbook_name=self.name,
                            status=RunbookStatus.FAILED,
                            message=result.get("message", "Failed"),
                            execution_time=elapsed,
                            output=result.get("output", ""),
                            actions_taken=result.get("actions", []),
                        )
            except Exception as e:
                elapsed = time.time() - start
                logger.error(f"Runbook {self.name} exception: {e}")
                if attempt >= self.max_retries:
                    return RunbookResult(
                        runbook_name=self.name,
                        status=RunbookStatus.FAILED,
                        message=f"Exception: {str(e)}",
                        execution_time=elapsed,
                    )
                time.sleep(2 ** attempt)

        return RunbookResult(
            runbook_name=self.name,
            status=RunbookStatus.FAILED,
            message="Retries exhausted",
            execution_time=time.time() - start,
        )

class RunbookRegistry:
    """Runbook registry"""

    def __init__(self):
        self.runbooks: list[Runbook] = []
        self.execution_history: list[dict] = []

    def register(self, runbook: Runbook):
        """Register a Runbook"""
        self.runbooks.append(runbook)
        logger.info(f"Registered Runbook: {runbook.name} (risk: {runbook.risk_level.value})")

    def find_and_execute(self, alert: dict, auto_execute: bool = True) -> Optional[RunbookResult]:
        """Find matching Runbook and execute"""
        for runbook in self.runbooks:
            if runbook.matches(alert):
                if not auto_execute and runbook.risk_level != RunbookRisk.SAFE:
                    logger.info(f"Runbook {runbook.name} risk level {runbook.risk_level.value}, needs confirmation")
                    return RunbookResult(
                        runbook_name=runbook.name,
                        status=RunbookStatus.SKIPPED,
                        message=f"Risk level {runbook.risk_level.value}, awaiting confirmation",
                        execution_time=0,
                    )

                logger.info(f"Executing Runbook: {runbook.name}")
                result = runbook.execute(alert)

                # Record execution history
                self.execution_history.append({
                    "timestamp": time.time(),
                    "runbook": runbook.name,
                    "alert": alert.get("alertname", ""),
                    "status": result.status.value,
                    "message": result.message,
                    "execution_time": result.execution_time,
                    "actions": result.actions_taken,
                })

                return result

        return None

# ====== Six Auto-Remediation Scenarios ======

def handle_high_cpu(alert: dict) -> dict:
    """Scenario 1: High CPU — restart process or scale"""
    instance = alert.get("labels", {}).get("instance", "")
    service = alert.get("labels", {}).get("service", "")

    actions = []

    # Step 1: Collect diagnostic info
    actions.append(f"Collecting top processes on {instance}")
    top_output = subprocess.run(
        ["ssh", instance, "top", "-b", "-n", "1"],
        capture_output=True, text=True, timeout=10
    ).stdout

    # Step 2: Identify high CPU process
    actions.append("Analyzing high CPU processes")
    if "runaway_worker" in top_output:
        # Step 3: Restart service
        actions.append(f"Restarting service {service} on {instance}")
        subprocess.run(
            ["ssh", instance, "systemctl", "restart", service],
            timeout=30
        )
        return {"success": True, "message": f"Restarted {service} on {instance}", "actions": actions}

    return {"success": False, "message": "No anomalous process identified, needs manual investigation", "actions": actions}

def handle_disk_full(alert: dict) -> dict:
    """Scenario 2: Disk full — clean expired logs and temp files"""
    instance = alert.get("labels", {}).get("instance", "")
    partition = alert.get("labels", {}).get("partition", "/")

    actions = []

    # Find large files
    actions.append(f"Scanning large files on {instance}:{partition}")
    du_output = subprocess.run(
        ["ssh", instance, "du", "-sh", f"{partition}/*"],
        capture_output=True, text=True, timeout=30
    ).stdout

    # Cleanup strategy
    cleanup_dirs = ["/var/log", "/tmp", "/var/cache/apt/archives"]
    total_cleaned = 0

    for dir_path in cleanup_dirs:
        if dir_path.startswith(partition):
            actions.append(f"Cleaning files older than 7 days in {dir_path}")
            result = subprocess.run(
                ["ssh", instance, "find", dir_path, "-type", "f", "-mtime", "+7", "-delete"],
                capture_output=True, text=True, timeout=60
            )
            actions.append(f"Cleaned {dir_path}")

    # Clean Docker unused images
    actions.append("Cleaning Docker unused images and containers")
    subprocess.run(
        ["ssh", instance, "docker", "system", "prune", "-f"],
        capture_output=True, text=True, timeout=60
    )

    # Verify disk space
    df_output = subprocess.run(
        ["ssh", instance, "df", "-h", partition],
        capture_output=True, text=True, timeout=10
    ).stdout

    return {
        "success": True,
        "message": f"Disk cleanup complete\n{df_output}",
        "actions": actions,
    }

def handle_memory_oom(alert: dict) -> dict:
    """Scenario 3: Memory OOM — restart OOM process and adjust limits"""
    instance = alert.get("labels", {}).get("instance", "")
    service = alert.get("labels", {}).get("service", "")

    actions = []

    # Check OOM records
    actions.append("Checking dmesg for OOM records")
    oom_log = subprocess.run(
        ["ssh", instance, "dmesg", "-T", "|", "grep", "-i", "oom"],
        capture_output=True, text=True, timeout=10, shell=True
    ).stdout

    if "Out of memory" in oom_log or "Killed process" in oom_log:
        actions.append(f"OOM Kill detected, restarting service {service}")
        subprocess.run(["ssh", instance, "systemctl", "restart", service], timeout=30)
        actions.append("Increasing cgroup memory limit")
        return {"success": True, "message": f"Handled OOM and restarted {service}", "actions": actions}

    return {"success": False, "message": "No OOM Kill record found", "actions": actions}

def handle_cert_expiry(alert: dict) -> dict:
    """Scenario 4: Certificate expiring — auto-renew"""
    instance = alert.get("labels", {}).get("instance", "")
    domain = alert.get("labels", {}).get("domain", "")

    actions = [f"Renewing certificate for {domain}"]

    # Renew with certbot
    result = subprocess.run(
        ["ssh", instance, "certbot", "renew", "--quiet", "--deploy-hook",
         "systemctl reload nginx"],
        capture_output=True, text=True, timeout=120
    )

    if result.returncode == 0:
        actions.append("Certificate renewed successfully, nginx reloaded")
        return {"success": True, "message": f"Certificate {domain} auto-renewed", "actions": actions}
    else:
        return {"success": False, "message": f"Certificate renewal failed: {result.stderr}", "actions": actions}

def handle_service_down(alert: dict) -> dict:
    """Scenario 5: Service down — attempt restart and health check"""
    instance = alert.get("labels", {}).get("instance", "")
    service = alert.get("labels", {}).get("service", "")

    actions = []

    # Attempt restart
    actions.append(f"Restarting service {service} on {instance}")
    subprocess.run(["ssh", instance, "systemctl", "restart", service], timeout=30)

    # Wait and check health
    time.sleep(10)
    health_check = subprocess.run(
        ["ssh", instance, "systemctl", "is-active", service],
        capture_output=True, text=True, timeout=10
    )

    if health_check.stdout.strip() == "active":
        actions.append("Service recovered")
        return {"success": True, "message": f"{service} restarted and recovered", "actions": actions}
    else:
        actions.append("Service still down after restart")
        return {"success": False, "message": f"{service} still down after restart, needs manual intervention",
                "actions": actions}

def handle_disk_io_high(alert: dict) -> dict:
    """Scenario 6: High disk IO — identify and throttle"""
    instance = alert.get("labels", {}).get("instance", "")

    actions = []

    # Identify high IO processes
    actions.append(f"Identifying high IO processes on {instance}")
    iotop_output = subprocess.run(
        ["ssh", instance, "iotop", "-b", "-n", "1", "-o"],
        capture_output=True, text=True, timeout=10
    ).stdout

    # Apply cgroup v2 IO limits
    actions.append("Applying cgroup limits to high IO processes")

    return {"success": True, "message": "Identified and throttled high IO processes", "actions": actions}

if __name__ == "__main__":
    registry = RunbookRegistry()

    # Register Runbooks
    registry.register(Runbook(
        name="auto-cpu-restart",
        description="Auto-restart service on high CPU",
        risk_level=RunbookRisk.CAUTION,
        match_conditions={"alertname": "HighCPU"},
        handler=handle_high_cpu,
        max_retries=1,
        timeout=60,
        cooldown=300,
    ))

    registry.register(Runbook(
        name="auto-disk-cleanup",
        description="Auto-cleanup on disk space warning",
        risk_level=RunbookRisk.SAFE,
        match_conditions={"alertname": "DiskSpaceWarning"},
        handler=handle_disk_full,
        max_retries=1,
        timeout=120,
        cooldown=600,
    ))

    registry.register(Runbook(
        name="auto-oom-restart",
        description="Auto-restart and adjust memory on OOM",
        risk_level=RunbookRisk.CAUTION,
        match_conditions={"alertname": "OOMKilled"},
        handler=handle_memory_oom,
        max_retries=1,
        timeout=30,
        cooldown=300,
    ))

    registry.register(Runbook(
        name="auto-cert-renew",
        description="Auto-renew expiring certificates",
        risk_level=RunbookRisk.SAFE,
        match_conditions={"alertname": "CertExpiringSoon"},
        handler=handle_cert_expiry,
        max_retries=1,
        timeout=120,
        cooldown=3600,
    ))

    registry.register(Runbook(
        name="auto-service-restart",
        description="Auto-restart on service down",
        risk_level=RunbookRisk.CAUTION,
        match_conditions={"alertname": "ServiceDown"},
        handler=handle_service_down,
        max_retries=2,
        timeout=30,
        cooldown=300,
    ))

    registry.register(Runbook(
        name="auto-io-throttle",
        description="Auto-throttle on high IO",
        risk_level=RunbookRisk.SAFE,
        match_conditions={"alertname": "HighDiskIO"},
        handler=handle_disk_io_high,
        max_retries=1,
        timeout=30,
        cooldown=300,
    ))

    # Simulate alert triggering Runbook
    test_alert = {
        "alertname": "DiskSpaceWarning",
        "severity": "medium",
        "labels": {
            "instance": "10.0.1.5",
            "service": "web-server",
            "partition": "/",
        },
    }

    print(f"Received alert: {test_alert['alertname']}")
    result = registry.find_and_execute(test_alert, auto_execute=True)
    if result:
        print(f"Runbook: {result.runbook_name}")
        print(f"Status: {result.status.value}")
        print(f"Message: {result.message}")
        print(f"Execution time: {result.execution_time:.2f}s")
        print(f"Actions: {result.actions_taken}")

#!/usr/bin/env python3
"""Automated safety layer — whitelists, rate limiting, circuit breakers"""

import time
from collections import defaultdict
from dataclasses import dataclass, field
from threading import Lock

@dataclass
class CircuitBreaker:
    """Circuit breaker — trips on consecutive failures"""
    failure_threshold: int = 3      # Consecutive failure threshold
    recovery_timeout: int = 600     # Recovery time (seconds)
    failure_count: int = 0
    last_failure_time: float = 0
    state: str = "closed"           # closed / open / half-open
    lock: Lock = field(default_factory=Lock)

    def can_execute(self) -> bool:
        with self.lock:
            if self.state == "open":
                # Check if can enter half-open
                if time.time() - self.last_failure_time > self.recovery_timeout:
                    self.state = "half-open"
                    return True
                return False
            return True

    def record_success(self):
        with self.lock:
            self.failure_count = 0
            self.state = "closed"

    def record_failure(self):
        with self.lock:
            self.failure_count += 1
            self.last_failure_time = time.time()
            if self.failure_count >= self.failure_threshold:
                self.state = "open"
                print(f"[CIRCUIT BREAKER] {self.failure_count} consecutive failures, tripping for {self.recovery_timeout}s")

class RateLimiter:
    """Rate limiter — limits Runbook execution frequency"""

    def __init__(self):
        self.executions: dict[str, list[float]] = defaultdict(list)
        self.lock = Lock()

    def can_execute(self, runbook_name: str, max_per_hour: int = 5) -> bool:
        """Check if execution is allowed (max per hour)"""
        with self.lock:
            now = time.time()
            # Clean records older than 1 hour
            self.executions[runbook_name] = [
                t for t in self.executions[runbook_name]
                if now - t < 3600
            ]
            if len(self.executions[runbook_name]) >= max_per_hour:
                return False
            self.executions[runbook_name].append(now)
            return True

class SafeExecutor:
    """Safe executor — integrates whitelist, rate limiting, circuit breaker"""

    def __init__(self):
        self.circuit_breakers: dict[str, CircuitBreaker] = defaultdict(CircuitBreaker)
        self.rate_limiter = RateLimiter()
        # Whitelist: only auto-remediate specific services
        self.service_whitelist = {
            "nginx", "redis", "web-server", "api-gateway",
            "log-collector", "metric-exporter",
        }
        # Blacklist: never auto-remediate these services
        self.service_blacklist = {
            "mysql-master", "postgresql-primary", "etcd",
            "consul", "zookeeper",
        }

    def can_execute(self, runbook_name: str, alert: dict) -> tuple[bool, str]:
        """Check if safe to execute"""
        labels = alert.get("labels", {})
        service = labels.get("service", "")

        # Check blacklist
        if service in self.service_blacklist:
            return False, f"Service {service} is blacklisted, auto-remediation forbidden"

        # Check whitelist
        if service and service not in self.service_whitelist:
            return False, f"Service {service} not in whitelist, needs confirmation"

        # Check circuit breaker
        cb = self.circuit_breakers[runbook_name]
        if not cb.can_execute():
            return False, f"Runbook {runbook_name} circuit breaker tripped, awaiting recovery"

        # Check rate limit
        if not self.rate_limiter.can_execute(runbook_name, max_per_hour=5):
            return False, f"Runbook {runbook_name} reached hourly execution limit"

        return True, "Allowed"

    def record_result(self, runbook_name: str, success: bool):
        """Record execution result"""
        cb = self.circuit_breakers[runbook_name]
        if success:
            cb.record_success()
        else:
            cb.record_failure()

if __name__ == "__main__":
    executor = SafeExecutor()

    # Test whitelist checking
    alerts = [
        {"alertname": "HighCPU", "labels": {"service": "nginx"}},
        {"alertname": "HighCPU", "labels": {"service": "mysql-master"}},
        {"alertname": "HighCPU", "labels": {"service": "unknown-service"}},
    ]

    for alert in alerts:
        can_run, reason = executor.can_execute("auto-cpu-restart", alert)
        status = "ALLOW" if can_run else "DENY"
        print(f"[{status}] {alert['labels']['service']:20s} | {reason}")

┌──────────────────────────────────────────────────────────────────────────┐
│                    Self-Healing Platform Architecture                      │
├──────────────────────────────────────────────────────────────────────────┤
│                                                                          │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐    │
│  │ Prometheus   │  │ ELK Stack   │  │ APM (Jaeger)│  │ Zabbix      │    │
│  │ (Metrics)    │  │ (Logs)      │  │ (Traces)    │  │ (Infra)     │    │
│  └──────┬───────┘  └──────┬───────┘  └──────┬───────┘  └──────┬───────┘    │
│         │                 │                 │                 │            │
│         ▼                 ▼                 ▼                 ▼            │
│  ┌──────────────────────────────────────────────────────────────────┐    │
│  │                   Alert Ingestion Layer                          │    │
│  │         Webhook Receiver | API Gateway | Message Queue           │    │
│  └──────────────────────────────┬───────────────────────────────────┘    │
│                                 │                                        │
│                                 ▼                                        │
│  ┌──────────────────────────────────────────────────────────────────┐    │
│  │                   Alert Processing Layer                         │    │
│  │  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────────┐    │    │
│  │  │ Dedup    │  │ Correlate│  │ Inhibit  │  │ Silence Mgmt │    │    │
│  │  └──────────┘  └──────────┘  └──────────┘  └──────────────┘    │    │
│  └──────────────────────────────┬───────────────────────────────────┘    │
│                                 │                                        │
│                                 ▼                                        │
│  ┌──────────────────────────────────────────────────────────────────┐    │
│  │               Routing & Dispatch Layer                          │    │
│  │  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────────┐    │    │
│  │  │ Severity │  │ Route    │  │ Escalate │  │ Notify       │    │    │
│  │  └──────────┘  └──────────┘  └──────────┘  └──────────────┘    │    │
│  └──────────────────────────────┬───────────────────────────────────┘    │
│                                 │                                        │
│                  ┌──────────────┴──────────────┐                        │
│                  ▼                              ▼                        │
│  ┌──────────────────────┐     ┌──────────────────────────────────┐     │
│  │  Auto-Remediation    │     │  Human Notification              │     │
│  │  ┌──────────────────┐│     │  ┌──────────────────────────┐    │     │
│  │  │ Runbook Registry ││     │  │ PagerDuty / Slack / SMS  │    │     │
│  │  │ (match+execute)   ││     │  └──────────────────────────┘    │     │
│  │  └────────┬─────────┘│     └──────────────────────────────────┘     │
│  │           │           │                                              │
│  │  ┌────────▼─────────┐ │                                              │
│  │  │ Safe Executor    │ │  (whitelist+rate limit+circuit breaker)    │
│  │  └────────┬─────────┘ │                                              │
│  │           │           │                                              │
│  │  ┌────────▼─────────┐ │                                              │
│  │  │ Action Executor  │ │  (SSH/API/K8s/Cloud)                        │
│  │  └────────┬─────────┘ │                                              │
│  └───────────┼──────────┘                                              │
│              │                                                           │
│              ▼                                                           │
│  ┌──────────────────────────────────────────────────────────────────┐    │
│  │               Verify & Feedback Layer                            │    │
│  │  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────────┐    │    │
│  │  │ Health   │  │ Verify   │  │ Rollback │  │ Event Archive│    │    │
│  │  └──────────┘  └──────────┘  └──────────┘  └──────────────┘    │    │
│  └──────────────────────────────────────────────────────────────────┘    │
│                                                                          │
└──────────────────────────────────────────────────────────────────────────┘

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"sync"
	"time"
)

// Alert structure
type Alert struct {
	ID        string            `json:"id"`
	AlertName string            `json:"alertname"`
	Severity  string            `json:"severity"`
	Labels    map[string]string `json:"labels"`
	Message   string            `json:"message"`
	StartsAt  time.Time         `json:"startsAt"`
}

// RemediationAction represents a remediation action
type RemediationAction struct {
	Name       string                 `json:"name"`
	Type       string                 `json:"type"` // ssh, api, k8s, cloud
	Params     map[string]interface{} `json:"params"`
	Result     string                 `json:"result"`
	ExecutedAt time.Time              `json:"executedAt"`
}

// RemediationResult represents the result of remediation
type RemediationResult struct {
	AlertID    string              `json:"alertId"`
	Success    bool                `json:"success"`
	Message    string              `json:"message"`
	Actions    []RemediationAction `json:"actions"`
	Duration   float64             `json:"duration"`
	Timestamp  time.Time           `json:"timestamp"`
}

// AutoRemediationPlatform is the core self-healing platform
type AutoRemediationPlatform struct {
	mu             sync.RWMutex
	deduplicator   *Deduplicator
	correlator     *Correlator
	router         *Router
	runbookReg     *RunbookRegistry
	safeExecutor   *SafeExecutor
	verifier       *HealthVerifier
	eventStore     []RemediationResult
}

// HandleAlert processes an alert through the complete pipeline
func (p *AutoRemediationPlatform) HandleAlert(alert Alert) (*RemediationResult, error) {
	start := time.Now()

	// 1. Dedup check
	if p.deduplicator.IsDuplicate(alert) {
		return &RemediationResult{
			AlertID:   alert.ID,
			Success:   true,
			Message:   "Alert deduplicated, no action needed",
			Timestamp: time.Now(),
		}, nil
	}

	// 2. Correlation check
	event := p.correlator.Correlate(alert)

	// 3. Routing decision
	route := p.router.Route(alert)

	// 4. Attempt auto-remediation
	var result *RemediationResult
	if route.AutoRemediation != "" {
		canExecute, reason := p.safeExecutor.Check(route.AutoRemediation, alert)
		if canExecute {
			result = p.runbookReg.Execute(route.AutoRemediation, alert)
			p.safeExecutor.RecordResult(route.AutoRemediation, result.Success)

			// 5. Verify remediation effectiveness
			if result.Success {
				verified := p.verifier.Verify(alert)
				if !verified {
					result.Success = false
					result.Message = "Remediation executed but verification failed, needs manual check"
				}
			}
		} else {
			log.Printf("[SAFETY] Runbook %s blocked: %s", route.AutoRemediation, reason)
		}
	}

	// 6. If auto-remediation failed or not configured, notify humans
	if result == nil || !result.Success {
		p.notifyHumans(route, alert)
	}

	// 7. Archive
	result.Duration = time.Since(start).Seconds()
	result.Timestamp = time.Now()
	p.storeEvent(*result)

	return result, nil
}

func (p *AutoRemediationPlatform) notifyHumans(route *RouteResult, alert Alert) {
	log.Printf("[HUMAN NOTIFY] Alert %s → %v (escalate: %ds → %v)",
		alert.AlertName, route.Receivers, route.EscalateAfter, route.EscalateTo)
}

func (p *AutoRemediationPlatform) storeEvent(result RemediationResult) {
	p.mu.Lock()
	defer p.mu.Unlock()
	p.eventStore = append(p.eventStore, result)
}

// WebhookHandler receives Alertmanager webhooks
func (p *AutoRemediationPlatform) WebhookHandler(w http.ResponseWriter, r *http.Request) {
	var payload struct {
		Alerts []Alert `json:"alerts"`
	}

	if err := json.NewDecoder(r.Body).Decode(&payload); err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}

	for _, alert := range payload.Alerts {
		go func(a Alert) {
			result, err := p.HandleAlert(a)
			if err != nil {
				log.Printf("[ERROR] Failed to handle alert %s: %v", a.AlertName, err)
			} else {
				log.Printf("[DONE] Alert %s → %s (%.2fs)",
					a.AlertName, result.Message, result.Duration)
			}
		}(alert)
	}

	w.WriteHeader(http.StatusOK)
	json.NewEncoder(w).Encode(map[string]string{"status": "received"})
}

func main() {
	platform := &AutoRemediationPlatform{
		deduplicator: NewDeduplicator(300),
		correlator:   NewCorrelator(),
		router:       NewRouter(),
		runbookReg:   NewRunbookRegistry(),
		safeExecutor: NewSafeExecutor(),
		verifier:     NewHealthVerifier(),
	}

	http.HandleFunc("/api/v1/alerts", platform.WebhookHandler)

	// Health check
	http.HandleFunc("/health", func(w http.ResponseWriter, r *http.Request) {
		w.WriteHeader(http.StatusOK)
		fmt.Fprintln(w, "OK")
	})

	// Stats endpoint
	http.HandleFunc("/api/v1/stats", func(w http.ResponseWriter, r *http.Request) {
		platform.mu.RLock()
		defer platform.mu.RUnlock()

		total := len(platform.eventStore)
		success := 0
		for _, e := range platform.eventStore {
			if e.Success {
				success++
			}
		}

		json.NewEncoder(w).Encode(map[string]interface{}{
			"total_events":  total,
			"success_count": success,
			"success_rate": func() float64 {
				if total == 0 {
					return 0
				}
				return float64(success) / float64(total) * 100
			}(),
		})
	})

	log.Println("Self-healing platform starting on :8080")
	log.Fatal(http.ListenAndServe(":8080", nil))
}

// Simplified component implementations (full implementations in production)

type Deduplicator struct{ window int }
type Correlator struct{}
type Router struct{}
type RouteResult struct {
	Receivers        []string
	EscalateAfter    int
	EscalateTo       []string
	AutoRemediation  string
}
type RunbookRegistry struct{}
type SafeExecutor struct{}
type HealthVerifier struct{}

func NewDeduplicator(w int) *Deduplicator { return &Deduplicator{window: w} }
func (d *Deduplicator) IsDuplicate(a Alert) bool { return false }
func NewCorrelator() *Correlator { return &Correlator{} }
func (c *Correlator) Correlate(a Alert) interface{} { return nil }
func NewRouter() *Router { return &Router{} }
func (r *Router) Route(a Alert) *RouteResult {
	return &RouteResult{Receivers: []string{"on-call"}, AutoRemediation: ""}
}
func NewRunbookRegistry() *RunbookRegistry { return &RunbookRegistry{} }
func (r *RunbookRegistry) Execute(name string, a Alert) *RemediationResult {
	return &RemediationResult{Success: false, Message: "no runbook matched"}
}
func NewSafeExecutor() *SafeExecutor { return &SafeExecutor{} }
func (s *SafeExecutor) Check(name string, a Alert) (bool, string) { return false, "safe check" }
func (s *SafeExecutor) RecordResult(name string, success bool) {}
func NewHealthVerifier() *HealthVerifier { return &HealthVerifier{} }
func (h *HealthVerifier) Verify(a Alert) bool { return true }