告警自动化处理：从告警风暴到自愈系统的工程实践

┌─────────────┐     ┌─────────────┐     ┌─────────────┐     ┌─────────────┐     ┌─────────────┐
│  1. 产生     │────▶│  2. 降噪     │────▶│  3. 路由     │────▶│  4. 处理     │────▶│  5. 复盘     │
│  Detection  │     │  Dedup      │     │  Route     │     │  Remediate │     │  Review    │
└─────────────┘     └─────────────┘     └─────────────┘     └─────────────┘     └─────────────┘
     │                    │                   │                   │                   │
  监控系统          去重/聚合/抑制      分级/分发/升级     自动Runbook/人工     归档/改进/度量

# alertmanager.yml — 分组配置
route:
  group_by: ['alertname', 'cluster', 'service']
  group_wait: 30s        # 首次告警后等待 30s，收集同组告警
  group_interval: 5m     # 同组告警的发送间隔
  repeat_interval: 4h    # 未解决的告警重复通知间隔
  receiver: 'default'

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

# alertmanager.yml — 抑制规则
inhibit_rules:
  # 当节点宕机时，抑制该节点上的所有服务告警
  - source_match:
      alertname: 'NodeDown'
    target_match_re:
      alertname: '.*(Service|Pod|Container).*Down|CrashLoopBackOff'
    equal: ['instance']

  # 当集群不可用时，抑制该集群的所有告警
  - source_match:
      severity: 'critical'
      alertname: 'ClusterUnavailable'
    target_match_re:
      severity: 'warning|info'
    equal: ['cluster']

  # 当数据库主节点切换时，抑制从库同步延迟告警
  - source_match:
      alertname: 'MySQLMasterSwitch'
    target_match:
      alertname: 'MySQLReplicationLag'
    equal: ['cluster', 'service']

#!/usr/bin/env python3
"""告警指纹去重引擎 — 基于告警特征生成唯一指纹，实现跨系统去重"""

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

@dataclass
class Alert:
    """告警数据结构"""
    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:
        """计算告警指纹（基于关键字段，忽略时间戳和动态值）"""
        # 只用结构性字段生成指纹，忽略 message 中的动态数值
        key_fields = f"{self.alertname}:{self.service}:{self.instance}:{self.cluster}:{self.severity}"
        return hashlib.md5(key_fields.encode()).hexdigest()[:16]

class AlertDeduplicator:
    """告警去重器"""

    def __init__(self, dedup_window: int = 300):
        """
        Args:
            dedup_window: 去重窗口（秒），同一指纹在此窗口内只保留一条
        """
        self.dedup_window = dedup_window
        self.alert_store: dict[str, Alert] = {}  # fingerprint -> Alert

    def process(self, alert: Alert) -> Optional[Alert]:
        """
        处理告警，返回需要发送的告警（去重后）
        返回 None 表示该告警是重复的，已被抑制
        """
        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

            # 在去重窗口内，抑制重复告警
            if now - existing.first_seen < self.dedup_window:
                return None  # 抑制
            else:
                # 超过去重窗口，作为新告警处理
                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]:
        """获取当前活跃的告警列表"""
        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):
        """清理过期的告警记录"""
        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)

    # 模拟告警风暴
    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%"),  # 重复
        Alert("HighCPU", "warning", "api-gw", "10.0.1.1", "prod", "CPU 98%"),  # 重复
        Alert("HighMemory", "critical", "api-gw", "10.0.1.1", "prod", "内存 95%"),
        Alert("DiskFull", "critical", "db", "10.0.2.1", "prod", "磁盘 85%"),
    ]

    for alert in alerts:
        result = dedup.process(alert)
        if result:
            print(f"[发送] {alert.alertname} | {alert.service} | {alert.instance} | 指纹={alert.fingerprint}")
        else:
            print(f"[抑制] {alert.alertname} | {alert.service} | {alert.instance} | 重复告警")

    print(f"\n活跃告警数: {len(dedup.get_active_alerts())}")

# alertmanager.yml — 静默规则（也可通过 API 动态创建）
# 通过 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": "数据库维护窗口"
#   }'

#!/bin/bash
# schedule-silence.sh — 自动创建告警静默（用于维护窗口）

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

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\": \"自动静默: ${SERVICE} 维护 ${DURATION}分钟\"
  }"

echo "已为 ${SERVICE} 创建 ${DURATION} 分钟静默窗口"

#!/usr/bin/env python3
"""多维度告警关联引擎 — 将不同来源的告警关联为同一事件"""

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

@dataclass
class AlertEvent:
    """关联后的告警事件"""
    event_id: str
    service: str
    cluster: str
    severity: str  # 取最高级别
    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:
    """告警关联器 — 基于服务、集群、时间窗口关联告警"""

    CORRELATION_WINDOW = 300  # 5分钟窗口

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

    def _correlation_key(self, alert: dict) -> str:
        """生成关联键"""
        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]:
        """处理告警，返回关联后的事件"""
        key = self._correlation_key(alert)
        now = time.time()

        if key in self.events:
            event = self.events[key]
            # 检查是否在关联窗口内
            if now - event.last_seen <= self.CORRELATION_WINDOW:
                event.add_alert(alert)
                return event
            else:
                # 窗口过期，创建新事件
                event.status = "resolved"

        # 创建新事件
        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()

    # 模拟多源告警
    raw_alerts = [
        {"alertname": "HighCPU", "severity": "warning",
         "labels": {"service": "api-gw", "cluster": "prod"}},
        {"alertname": "HighLatency", "severity": "critical",
         "labels": {"service": "api-gw", "cluster": "prod"}},  # 关联
        {"alertname": "ErrorRateHigh", "severity": "critical",
         "labels": {"service": "api-gw", "cluster": "prod"}},  # 关联
        {"alertname": "DiskFull", "severity": "critical",
         "labels": {"service": "db", "cluster": "prod"}},       # 不同服务
    ]

    for alert in raw_alerts:
        event = correlator.process(alert)
        print(f"告警: {alert['alertname']:20s} → 事件: {event.event_id} | "
              f"严重度: {event.severity:8s} | 关联告警数: {len(event.alerts)}")

#!/usr/bin/env python3
"""告警路由引擎 — 基于规则的告警分级与分发"""

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:
    """路由规则"""
    name: str
    match_labels: dict           # 匹配的标签
    match_severity: list[str]    # 匹配的严重度
    receivers: list[str]         # 接收者列表
    escalate_after: int = 0     # 未响应后升级时间（秒）
    escalate_to: list[str] = field(default_factory=list)  # 升级接收者
    auto_remediation: str = ""   # 关联的自动修复 Runbook

class AlertRouter:
    """告警路由器"""

    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:
        """路由告警到正确的接收者和处理流程"""
        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:
        """检查告警是否匹配路由规则"""
        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()

    # 定义路由规则
    router.add_rule(RouteRule(
        name="数据库 P0",
        match_labels={"service": "mysql", "cluster": "prod"},
        match_severity=[Severity.P0.value],
        receivers=["on-call-dba", "on-call-sre", "tech-lead"],
        escalate_after=300,  # 5 分钟未响应升级
        escalate_to=["cto", "vp-engineering"],
        auto_remediation="runbook:mysql-failover",
    ))

    router.add_rule(RouteRule(
        name="API 服务 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="磁盘空间预警",
        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_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"\n告警: {alert['alertname']} ({alert['severity']})")
        print(f"  接收者: {result['receivers']}")
        print(f"  自动修复: {result['auto_remediation']}")
        print(f"  升级策略: {result['escalate_after']}s → {result['escalate_to']}")

#!/usr/bin/env python3
"""告警升级管理器 — 跟踪告警响应状态，超时自动升级"""

import time
import threading
from dataclasses import dataclass, field

@dataclass
class EscalationPolicy:
    """升级策略"""
    initial_receivers: list[str]
    escalate_after: int           # 秒
    escalate_to: list[str]        # 升级后的接收者
    max_escalations: int = 3      # 最大升级次数
    auto_resolve_on_action: bool = True  # 有人响应后停止升级

@dataclass
class AlertEscalation:
    """告警升级跟踪"""
    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]:
        """执行升级，返回新的接收者列表"""
        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:
    """告警升级管理器"""

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

    def track(self, alert_id: str, policy: EscalationPolicy):
        """开始跟踪告警的升级"""
        escalation = AlertEscalation(
            alert_id=alert_id,
            policy=policy,
            created_at=time.time(),
        )
        self.tracked[alert_id] = escalation

        # 设置升级定时器
        timer = threading.Timer(
            policy.escalate_after,
            self._check_escalation,
            args=[alert_id]
        )
        timer.daemon = True
        timer.start()

    def _check_escalation(self, alert_id: str):
        """检查是否需要升级"""
        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"[升级] 告警 {alert_id} → 级别 {escalation.current_level}")
            print(f"  新接收者: {new_receivers}")
            # 这里触发实际的通知发送

    def acknowledge(self, alert_id: str, user: str):
        """确认告警"""
        if alert_id in self.tracked:
            self.tracked[alert_id].acknowledge(user)
            print(f"[确认] 告警 {alert_id} 已被 {user} 确认")

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

    policy = EscalationPolicy(
        initial_receivers=["on-call-sre"],
        escalate_after=5,  # 5秒后升级（演示用，实际应为 300-600 秒）
        escalate_to=["tech-lead", "vp-engineering", "cto"],
        max_escalations=3,
    )

    manager.track("EVT-001", policy)
    print("告警已跟踪，等待响应...")
    print("(未确认，5秒后自动升级)")

    time.sleep(8)  # 等待升级触发

    # 模拟确认
    # manager.acknowledge("EVT-001", "张三")

#!/usr/bin/env python3
"""Runbook 注册表 — 管理和执行自动化修复脚本"""

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"           # 安全操作，可自动执行
    CAUTION = "caution"     # 需谨慎，建议人工确认后执行
    DANGEROUS = "dangerous" # 高风险，必须人工确认

@dataclass
class RunbookResult:
    """Runbook 执行结果"""
    runbook_name: str
    status: RunbookStatus
    message: str
    execution_time: float
    output: str = ""
    actions_taken: list = field(default_factory=list)

@dataclass
class Runbook:
    """Runbook 定义"""
    name: str
    description: str
    risk_level: RunbookRisk
    match_conditions: dict       # 匹配的告警条件
    handler: Callable            # 处理函数
    max_retries: int = 1        # 最大重试次数
    timeout: int = 60           # 超时时间（秒）
    cooldown: int = 300         # 冷却时间（秒），同一 Runbook 的执行间隔
    last_executed: float = 0    # 上次执行时间

    def matches(self, alert: dict) -> bool:
        """检查告警是否匹配此 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:
        """检查是否可以执行（冷却时间检查）"""
        if self.last_executed == 0:
            return True
        return time.time() - self.last_executed >= self.cooldown

    def execute(self, alert: dict) -> RunbookResult:
        """执行 Runbook"""
        if not self.can_execute():
            return RunbookResult(
                runbook_name=self.name,
                status=RunbookStatus.SKIPPED,
                message=f"Runbook 处于冷却期（{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", "执行成功"),
                        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+1} 次执行失败，重试中...")
                        time.sleep(2 ** attempt)  # 指数退避
                    else:
                        return RunbookResult(
                            runbook_name=self.name,
                            status=RunbookStatus.FAILED,
                            message=result.get("message", "执行失败"),
                            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} 执行异常: {e}")
                if attempt >= self.max_retries:
                    return RunbookResult(
                        runbook_name=self.name,
                        status=RunbookStatus.FAILED,
                        message=f"执行异常: {str(e)}",
                        execution_time=elapsed,
                    )
                time.sleep(2 ** attempt)

        return RunbookResult(
            runbook_name=self.name,
            status=RunbookStatus.FAILED,
            message="重试次数耗尽",
            execution_time=time.time() - start,
        )

class RunbookRegistry:
    """Runbook 注册表"""

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

    def register(self, runbook: Runbook):
        """注册 Runbook"""
        self.runbooks.append(runbook)
        logger.info(f"已注册 Runbook: {runbook.name} (风险: {runbook.risk_level.value})")

    def find_and_execute(self, alert: dict, auto_execute: bool = True) -> Optional[RunbookResult]:
        """查找匹配的 Runbook 并执行"""
        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} 风险等级 {runbook.risk_level.value}，需人工确认")
                    return RunbookResult(
                        runbook_name=runbook.name,
                        status=RunbookStatus.SKIPPED,
                        message=f"风险等级 {runbook.risk_level.value}，等待人工确认",
                        execution_time=0,
                    )

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

                # 记录执行历史
                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

# ====== 六大自动修复场景 ======

def handle_high_cpu(alert: dict) -> dict:
    """场景1：CPU 使用率过高 — 重启进程或扩容"""
    instance = alert.get("labels", {}).get("instance", "")
    service = alert.get("labels", {}).get("service", "")

    actions = []

    # 步骤1：采集诊断信息
    actions.append(f"采集 {instance} 的 top 进程信息")
    top_output = subprocess.run(
        ["ssh", instance, "top", "-b", "-n", "1"],
        capture_output=True, text=True, timeout=10
    ).stdout

    # 步骤2：识别高 CPU 进程
    actions.append("分析高 CPU 进程")
    if "runaway_worker" in top_output:
        # 步骤3：重启服务
        actions.append(f"重启服务 {service} on {instance}")
        subprocess.run(
            ["ssh", instance, "systemctl", "restart", service],
            timeout=30
        )
        return {"success": True, "message": f"已重启 {service} on {instance}", "actions": actions}

    return {"success": False, "message": "未识别到异常进程，需人工排查", "actions": actions}

def handle_disk_full(alert: dict) -> dict:
    """场景2：磁盘空间不足 — 清理过期日志和临时文件"""
    instance = alert.get("labels", {}).get("instance", "")
    partition = alert.get("labels", {}).get("partition", "/")

    actions = []

    # 查找大文件
    actions.append(f"扫描 {instance}:{partition} 的大文件")
    du_output = subprocess.run(
        ["ssh", instance, "du", "-sh", f"{partition}/*"],
        capture_output=True, text=True, timeout=30
    ).stdout

    # 清理策略
    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"清理 {dir_path} 中 7 天前的文件")
            result = subprocess.run(
                ["ssh", instance, "find", dir_path, "-type", "f", "-mtime", "+7", "-delete"],
                capture_output=True, text=True, timeout=60
            )
            actions.append(f"已清理 {dir_path}")

    # 清理 Docker 无用镜像
    actions.append("清理 Docker 无用镜像和容器")
    subprocess.run(
        ["ssh", instance, "docker", "system", "prune", "-f"],
        capture_output=True, text=True, timeout=60
    )

    # 验证磁盘空间
    df_output = subprocess.run(
        ["ssh", instance, "df", "-h", partition],
        capture_output=True, text=True, timeout=10
    ).stdout

    return {
        "success": True,
        "message": f"磁盘清理完成\n{df_output}",
        "actions": actions,
    }

def handle_memory_oom(alert: dict) -> dict:
    """场景3：内存 OOM — 重启 OOM 进程并调整限制"""
    instance = alert.get("labels", {}).get("instance", "")
    service = alert.get("labels", {}).get("service", "")

    actions = []

    # 检查 OOM 记录
    actions.append("检查 dmesg 中的 OOM 记录")
    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，重启服务 {service}")
        subprocess.run(["ssh", instance, "systemctl", "restart", service], timeout=30)
        actions.append("增加 cgroup 内存限制")
        return {"success": True, "message": f"已处理 OOM 并重启 {service}", "actions": actions}

    return {"success": False, "message": "未检测到 OOM Kill 记录", "actions": actions}

def handle_cert_expiry(alert: dict) -> dict:
    """场景4：证书即将过期 — 自动续期"""
    instance = alert.get("labels", {}).get("instance", "")
    domain = alert.get("labels", {}).get("domain", "")

    actions = [f"为 {domain} 执行证书续期"]

    # 使用 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("证书续期成功，已重新加载 nginx")
        return {"success": True, "message": f"证书 {domain} 已自动续期", "actions": actions}
    else:
        return {"success": False, "message": f"证书续期失败: {result.stderr}", "actions": actions}

def handle_service_down(alert: dict) -> dict:
    """场景5：服务不可用 — 尝试重启并检查健康"""
    instance = alert.get("labels", {}).get("instance", "")
    service = alert.get("labels", {}).get("service", "")

    actions = []

    # 尝试重启
    actions.append(f"重启服务 {service} on {instance}")
    subprocess.run(["ssh", instance, "systemctl", "restart", service], timeout=30)

    # 等待并检查健康
    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("服务已恢复")
        return {"success": True, "message": f"{service} 已重启并恢复", "actions": actions}
    else:
        actions.append("服务重启后仍未恢复")
        return {"success": False, "message": f"{service} 重启后仍未恢复，需人工介入",
                "actions": actions}

def handle_disk_io_high(alert: dict) -> dict:
    """场景6：磁盘 IO 过高 — 识别并限制 IO"""
    instance = alert.get("labels", {}).get("instance", "")

    actions = []

    # 识别高 IO 进程
    actions.append(f"识别 {instance} 上的高 IO 进程")
    iotop_output = subprocess.run(
        ["ssh", instance, "iotop", "-b", "-n", "1", "-o"],
        capture_output=True, text=True, timeout=10
    ).stdout

    # 使用 cgroup v2 限制 IO
    actions.append("对高 IO 进程应用 cgroup 限制")
    # 这里简化处理，实际场景需要解析 iotop 输出获取 PID

    return {"success": True, "message": "已识别并限制高 IO 进程", "actions": actions}

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

    # 注册 Runbook
    registry.register(Runbook(
        name="auto-cpu-restart",
        description="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="磁盘空间不足时自动清理",
        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="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="证书即将过期时自动续期",
        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="服务不可用时自动重启",
        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="IO 过高时自动限制",
        risk_level=RunbookRisk.SAFE,
        match_conditions={"alertname": "HighDiskIO"},
        handler=handle_disk_io_high,
        max_retries=1,
        timeout=30,
        cooldown=300,
    ))

    # 模拟告警触发 Runbook
    test_alert = {
        "alertname": "DiskSpaceWarning",
        "severity": "medium",
        "labels": {
            "instance": "10.0.1.5",
            "service": "web-server",
            "partition": "/",
        },
    }

    print(f"收到告警: {test_alert['alertname']}")
    result = registry.find_and_execute(test_alert, auto_execute=True)
    if result:
        print(f"Runbook: {result.runbook_name}")
        print(f"状态: {result.status.value}")
        print(f"消息: {result.message}")
        print(f"执行时间: {result.execution_time:.2f}s")
        print(f"执行动作: {result.actions_taken}")

#!/usr/bin/env python3
"""自动化安全防护层 — 白名单、限流、熔断"""

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

@dataclass
class CircuitBreaker:
    """熔断器 — 连续失败时熔断自动修复"""
    failure_threshold: int = 3      # 连续失败阈值
    recovery_timeout: int = 600     # 熔断恢复时间（秒）
    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":
                # 检查是否可以进入 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"[熔断] 连续失败 {self.failure_count} 次，熔断 {self.recovery_timeout}s")

class RateLimiter:
    """限流器 — 限制同一 Runbook 的执行频率"""

    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:
        """检查是否允许执行（每小时最多 max_per_hour 次）"""
        with self.lock:
            now = time.time()
            # 清理1小时前的记录
            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:
    """安全执行器 — 集成白名单、限流、熔断"""

    def __init__(self):
        self.circuit_breakers: dict[str, CircuitBreaker] = defaultdict(CircuitBreaker)
        self.rate_limiter = RateLimiter()
        # 白名单：只允许对特定服务执行自动修复
        self.service_whitelist = {
            "nginx", "redis", "web-server", "api-gateway",
            "log-collector", "metric-exporter",
        }
        # 黑名单：永远不自动修复的服务
        self.service_blacklist = {
            "mysql-master", "postgresql-primary", "etcd",
            "consul", "zookeeper",
        }

    def can_execute(self, runbook_name: str, alert: dict) -> tuple[bool, str]:
        """检查是否可以安全执行"""
        labels = alert.get("labels", {})
        service = labels.get("service", "")

        # 检查黑名单
        if service in self.service_blacklist:
            return False, f"服务 {service} 在黑名单中，禁止自动修复"

        # 检查白名单
        if service and service not in self.service_whitelist:
            return False, f"服务 {service} 不在白名单中，需人工确认"

        # 检查熔断器
        cb = self.circuit_breakers[runbook_name]
        if not cb.can_execute():
            return False, f"Runbook {runbook_name} 已熔断，等待恢复"

        # 检查限流
        if not self.rate_limiter.can_execute(runbook_name, max_per_hour=5):
            return False, f"Runbook {runbook_name} 已达每小时执行上限"

        return True, "允许执行"

    def record_result(self, runbook_name: str, success: bool):
        """记录执行结果"""
        cb = self.circuit_breakers[runbook_name]
        if success:
            cb.record_success()
        else:
            cb.record_failure()

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

    # 测试白名单检查
    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 = "允许" if can_run else "拒绝"
        print(f"[{status}] {alert['labels']['service']:20s} | {reason}")

┌──────────────────────────────────────────────────────────────────────────┐
│                          自愈平台架构                                      │
├──────────────────────────────────────────────────────────────────────────┤
│                                                                          │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐    │
│  │ Prometheus   │  │ ELK Stack   │  │ APM (Jaeger)│  │ Zabbix      │    │
│  │ (Metrics)    │  │ (Logs)      │  │ (Traces)    │  │ (Infra)     │    │
│  └──────┬───────┘  └──────┬───────┘  └──────┬───────┘  └──────┬───────┘    │
│         │                 │                 │                 │            │
│         ▼                 ▼                 ▼                 ▼            │
│  ┌──────────────────────────────────────────────────────────────────┐    │
│  │                    告警接入层 (Alert Ingestion)                    │    │
│  │         Webhook Receiver | API Gateway | Message Queue            │    │
│  └──────────────────────────────┬───────────────────────────────────┘    │
│                                 │                                        │
│                                 ▼                                        │
│  ┌──────────────────────────────────────────────────────────────────┐    │
│  │                    告警处理层 (Alert Processing)                   │    │
│  │  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────────┐    │    │
│  │  │ 去重引擎  │  │ 关联引擎  │  │ 抑制引擎  │  │ 静默管理     │    │    │
│  │  └──────────┘  └──────────┘  └──────────┘  └──────────────┘    │    │
│  └──────────────────────────────┬───────────────────────────────────┘    │
│                                 │                                        │
│                                 ▼                                        │
│  ┌──────────────────────────────────────────────────────────────────┐    │
│  │                    路由决策层 (Routing & Dispatch)                  │    │
│  │  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────────┐    │    │
│  │  │ 分级引擎  │  │ 路由规则  │  │ 升级管理  │  │ 通知分发     │    │    │
│  │  └──────────┘  └──────────┘  └──────────┘  └──────────────┘    │    │
│  └──────────────────────────────┬───────────────────────────────────┘    │
│                                 │                                        │
│                  ┌──────────────┴──────────────┐                        │
│                  ▼                              ▼                        │
│  ┌──────────────────────┐     ┌──────────────────────────────────┐     │
│  │  自动修复层            │     │  人工通知层                       │     │
│  │  ┌──────────────────┐ │     │  ┌──────────────────────────┐   │     │
│  │  │ Runbook Registry │ │     │  │ PagerDuty / 飞书 / 短信   │   │     │
│  │  │ (匹配+执行)       │ │     │  └──────────────────────────┘   │     │
│  │  └────────┬─────────┘ │     └──────────────────────────────────┘     │
│  │           │           │                                              │
│  │  ┌────────▼─────────┐ │                                              │
│  │  │ Safe Executor    │ │  (白名单+限流+熔断)                           │
│  │  └────────┬─────────┘ │                                              │
│  │           │           │                                              │
│  │  ┌────────▼─────────┐ │                                              │
│  │  │ Action Executor  │ │  (SSH/API/K8s/Cloud)                        │
│  │  └────────┬─────────┘ │                                              │
│  └───────────┼──────────┘                                              │
│              │                                                           │
│              ▼                                                           │
│  ┌──────────────────────────────────────────────────────────────────┐    │
│  │                    验证与反馈层 (Verify & Feedback)                 │    │
│  │  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────────┐    │    │
│  │  │ 健康检查  │  │ 效果验证  │  │ 失败回滚  │  │ 事件归档     │    │    │
│  │  └──────────┘  └──────────┘  └──────────┘  └──────────────┘    │    │
│  └──────────────────────────────────────────────────────────────────┘    │
│                                                                          │
└──────────────────────────────────────────────────────────────────────────┘

package main

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

// Alert 告警结构
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 修复动作
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 修复结果
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 自愈平台核心
type AutoRemediationPlatform struct {
	mu             sync.RWMutex
	deduplicator   *Deduplicator
	correlator     *Correlator
	router         *Router
	runbookReg     *RunbookRegistry
	safeExecutor   *SafeExecutor
	verifier       *HealthVerifier
	eventStore     []RemediationResult
}

// HandleAlert 处理告警的完整流程
func (p *AutoRemediationPlatform) HandleAlert(alert Alert) (*RemediationResult, error) {
	start := time.Now()

	// 1. 去重检查
	if p.deduplicator.IsDuplicate(alert) {
		return &RemediationResult{
			AlertID:   alert.ID,
			Success:   true,
			Message:   "告警已去重，无需处理",
			Timestamp: time.Now(),
		}, nil
	}

	// 2. 关联检查
	event := p.correlator.Correlate(alert)

	// 3. 路由决策
	route := p.router.Route(alert)

	// 4. 尝试自动修复
	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. 验证修复效果
			if result.Success {
				verified := p.verifier.Verify(alert)
				if !verified {
					result.Success = false
					result.Message = "修复执行成功但验证未通过，需人工检查"
				}
			}
		} else {
			log.Printf("[安全拦截] Runbook %s 被拦截: %s", route.AutoRemediation, reason)
		}
	}

	// 6. 如果自动修复失败或未配置，发送人工通知
	if result == nil || !result.Success {
		p.notifyHumans(route, alert)
	}

	// 7. 归档
	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("[人工通知] 告警 %s → %v (升级: %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 Alertmanager webhook 接收器
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("[错误] 处理告警 %s 失败: %v", a.AlertName, err)
			} else {
				log.Printf("[完成] 告警 %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)

	// 健康检查
	http.HandleFunc("/health", func(w http.ResponseWriter, r *http.Request) {
		w.WriteHeader(http.StatusOK)
		fmt.Fprintln(w, "OK")
	})

	// 统计接口
	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("自愈平台启动，监听 :8080")
	log.Fatal(http.ListenAndServe(":8080", nil))
}

// 以下是各组件的简化实现（实际项目中应有完整实现）

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 }