智能运维落地复盘：我踩过的 10 个坑和解决方案，落地成功率从 30% 提升到 85%

❌ 错误目标：全自动运维（AI 替代人工）
✅ 正确目标：P3/P2 告警自动处理率 80%，P0/P1 人工介入率确保全覆盖

"""智能运维数据预处理流水线"""
import pandas as pd
import numpy as np
from sklearn.preprocessing import StandardScaler


class DataPreprocessor:
    """监控数据预处理"""

    def __init__(self, config: dict):
        self.fill_method = config.get("fill_method", "ffill")
        self.outlier_threshold = config.get("outlier_threshold", 3.0)
        self.scaler = StandardScaler()

    def run(self, df: pd.DataFrame) -> pd.DataFrame:
        """执行完整预处理流水线"""
        df = self._fill_missing(df)
        df = self._remove_outliers(df)
        df = self._align_timestamps(df)
        df = self._normalize(df)
        return df

    def _fill_missing(self, df: pd.DataFrame) -> pd.DataFrame:
        """缺失值填充"""
        missing_pct = df.isnull().sum() / len(df)
        # 缺失超过 40% 的列直接丢弃
        drop_cols = missing_pct[missing_pct > 0.4].index.tolist()
        df = df.drop(columns=drop_cols)
        # 其余用前向填充
        df = df.fillna(method=self.fill_method)
        return df

    def _remove_outliers(self, df: pd.DataFrame) -> pd.DataFrame:
        """基于 Z-score 的异常点平滑"""
        numeric_cols = df.select_dtypes(include=[np.number]).columns
        for col in numeric_cols:
            z = np.abs((df[col] - df[col].mean()) / df[col].std())
            mask = z > self.outlier_threshold
            df.loc[mask, col] = df[col].median()
        return df

    def _align_timestamps(self, df: pd.DataFrame) -> pd.DataFrame:
        """时间对齐：统一采样间隔"""
        if "timestamp" in df.columns:
            df = df.set_index("timestamp")
            df = df.resample("1min").mean().interpolate()
        return df

    def _normalize(self, df: pd.DataFrame) -> pd.DataFrame:
        """标准化：消除量纲差异"""
        numeric_cols = df.select_dtypes(include=[np.number]).columns
        df[numeric_cols] = self.scaler.fit_transform(df[numeric_cols])
        return df

L1 层：告警聚合 — 同一故障源的告警合并（5 分钟窗口）
L2 层：智能去重 — 重复告警只保留告警状态变化
L3 层：优先级排序 — 基于业务影响评分，只推送 Top N

#!/bin/bash
# =====================================================
# 脚本功能：同源告警 5 分钟窗口聚合去重
# 作者：运维团队
# 日期：2026-03-10
# 依赖：jq、curl
# =====================================================

# 配置变量
ALERT_API="http://alertmanager:9093/api/v2/alerts"
WINDOW=300  # 聚合窗口（秒）
DEDUP_KEY="alertname,instance,job"

# 拉取当前活跃告警
curl -s "$ALERT_API" | jq -c '.[]' > /tmp/raw_alerts.json

# 按 DEDUP_KEY 聚合，5 分钟窗口内只保留最近一条
jq -s --arg key "$DEDUP_KEY" --arg w "$WINDOW" '
  group_by(.labels[$key]) |
  map(select(.startsAt >= (now - ($w | tonumber)))) |
  max_by(.startsAt)
' /tmp/raw_alerts.json > /tmp/dedup_alerts.json

# 输出聚合后告警数量对比
RAW=$(jq -s 'length' /tmp/raw_alerts.json)
DEDUP=$(jq -s 'length' /tmp/dedup_alerts.json)
echo "[AGG] 原始告警 $RAW 条 → 聚合后 $DEDUP 条（降噪率 $(echo "scale=1; ($RAW-$DEDUP)*100/$RAW" | bc)%）"

# 每 5 分钟执行告警聚合
*/5 * * * * /opt/scripts/alert_aggregator.sh >> /var/log/aiops/aggregator.log 2>&1

# 1. 赋予执行权限
chmod +x /opt/scripts/alert_aggregator.sh

# 2. 安装依赖
yum install -y jq curl bc

# 3. 测试运行
/opt/scripts/alert_aggregator.sh

# 1. 查看自愈引擎执行日志
grep "auto-healing" /var/log/aiops/executor.log | tail -20
# 关键输出：[2024-08-15 03:12:07] ACTION=pg_restart target=master reason="sync_lag"

# 2. 查数据库主库状态切换记录
kubectl get pods -n db -l role=master --show-labels
# 关键输出：postgres-master-0 (Terminating) → postgres-master-1 (Running)

# 3. 查 binlog 夺点
psql -U postgres -c "SELECT pg_current_wal_lsn(), pg_last_wal_replay_lsn();"
# 关键输出：0/F1A3C000 vs 0/F1A2E000 → 缺失 14MB 未复制

"""分级审批策略"""
from enum import Enum


class RiskLevel(Enum):
    LOW = "low"       # 全自动：单 Pod 重启
    MEDIUM = "medium"  # 通知 + 自动执行：资源扩容
    HIGH = "high"     # 人工确认后执行：配置变更
    BLOCKED = "blocked"  # 禁止 AI 执行：数据操作


APPROVAL_POLICY = {
    RiskLevel.LOW: {"auto_execute": True, "notify": True},
    RiskLevel.MEDIUM: {"auto_execute": True, "notify": True, "approval_timeout": 300},
    RiskLevel.HIGH: {"auto_execute": False, "require_approval": True},
    RiskLevel.BLOCKED: {"auto_execute": False, "allow_manual_only": True},
}

L1 层：数据脱敏 — 日志传入 AI 前自动脱敏（手机号→138****5678）
L2 层：访问控制 — AI 服务用只读账号，限制可访问的日志源
L3 层：审计追踪 — 所有 AI 访问记录写入审计日志

1. 新模型先以 shadow 模式运行（仅记录，不触发动作）
2. 对比 shadow 模式和线上模型的差异
3. 逐步放量：5% → 20% → 50% → 全量
4. 每个阶段观察误报率和漏报率
5. 任何阶段指标恶化，一键回滚到上一版本

"""模型健康度监控"""
import numpy as np
from scipy import stats


class ModelDriftDetector:
    """模型漂移检测器"""

    def __init__(self, reference_data: np.ndarray, threshold: float = 0.05):
        self.reference = reference_data
        self.threshold = threshold  # p-value 阈值

    def detect(self, current_data: np.ndarray) -> dict:
        """使用 KS 检验检测数据分布漂移"""
        stat, p_value = stats.ks_2samp(self.reference, current_data)

        result = {
            "statistic": stat,
            "p_value": p_value,
            "drift_detected": p_value < self.threshold,
            "severity": self._severity(stat)
        }

        if result["drift_detected"]:
            self._alert(result)

        return result

    def _severity(self, stat: float) -> str:
        """漂移严重度"""
        if stat > 0.3:
            return "critical"
        elif stat > 0.15:
            return "warning"
        return "info"

    def _alert(self, result: dict):
        """发送漂移告警"""
        print(f"[DRIFT ALERT] 检测到模型漂移! "
              f"严重度={result['severity']}, "
              f"统计量={result['statistic']:.4f}, "
              f"p值={result['p_value']:.6f}")

# prometheus_rules.yaml — 模型漂移监控告警规则
groups:
  - name: aiops_model_drift
    rules:
      - alert: ModelDriftWarning
        expr: aiops_model_drift_score > 0.15
        for: 10m
        labels:
          severity: warning
          team: aiops
        annotations:
          summary: "模型分布出现 warning 级漂移（{{ $labels.model }}）"
          description: "KS 统计量={{ $value }}，建议本周内安排重训"

      - alert: ModelDriftCritical
        expr: aiops_model_drift_score > 0.3
        for: 2m
        labels:
          severity: critical
          team: aiops
        annotations:
          summary: "模型分布出现 critical 级漂移（{{ $labels.model }}）"
          description: "KS 统计量={{ $value }}，已触发自动切回上一版本"

1. 让一线运维参与需求评审 — 他们知道哪些痛点值得做
2. AI 推荐附带可解释性 — "为什么判断是数据库慢查询？"附上证据链
3. 先做减法再做加法 — 先减少告警噪音，再增加自动修复能力