19d - AIOps 实践

2023：基于规则的监控
├── 静态阈值告警（CPU > 80% 就报警）
├── 手动关联日志和指标
└── 人工编写 runbook

2024：ML 增强监控
├── 动态基线（自动学习正常模式）
├── 异常检测（无监督学习）
└── 告警聚合和去重

2025：LLM 驱动诊断
├── 自然语言查询监控数据
├── AI 自动生成根因分析报告
├── Copilot 式运维助手
└── 日志语义分析（不再依赖正则）

2026：Agentic AIOps
├── 自主检测 → 诊断 → 修复闭环
├── AI Agent 执行 runbook
├── 预测性自愈（故障前修复）
└── 多 Agent 协作运维

Solo Founder / 小团队（1-5 人）：
├── 方案 A（零成本）：Uptime Kuma + Grafana + Prometheus
├── 方案 B（低成本）：New Relic 免费层（100GB/月）
└── 方案 C（AI 增强）：New Relic 免费层 + n8n AI 告警 Agent

中型团队（5-50 人）：
├── 推荐：Datadog（$15/主机/月起）
├── 替代：New Relic Pro + PagerDuty
└── 开源方案：Grafana Stack + Robusta

大型企业（50+ 人）：
├── 推荐：Dynatrace（全自动根因分析）
├── 替代：Datadog Enterprise + BigPanda
└── 混合方案：Splunk + PagerDuty AIOps

import numpy as np
from collections import deque
from datetime import datetime
 
class ZScoreDetector:
    """基于滑动窗口 Z-Score 的异常检测器"""
    
    def __init__(self, window_size: int = 100, threshold: float = 3.0):
        self.window = deque(maxlen=window_size)
        self.threshold = threshold
    
    def detect(self, value: float) -> dict:
        self.window.append(value)
        
        if len(self.window) < 10:  # 最少需要 10 个数据点
            return {"is_anomaly": False, "z_score": 0.0}
        
        mean = np.mean(self.window)
        std = np.std(self.window)
        
        if std == 0:
            return {"is_anomaly": False, "z_score": 0.0}
        
        z_score = abs(value - mean) / std
        
        return {
            "is_anomaly": z_score > self.threshold,
            "z_score": round(z_score, 2),
            "value": value,
            "mean": round(mean, 2),
            "std": round(std, 2),
            "timestamp": datetime.now().isoformat()
        }
 
# 使用示例
detector = ZScoreDetector(window_size=60, threshold=3.0)
 
# 模拟正常流量 + 异常峰值
normal_traffic = np.random.normal(100, 10, 50).tolist()
anomaly_traffic = [100, 100, 250, 300, 100]  # 250 和 300 是异常值
 
for value in normal_traffic + anomaly_traffic:
    result = detector.detect(value)
    if result["is_anomaly"]:
        print(f"⚠️ 异常检测: value={result['value']}, "
              f"z_score={result['z_score']}, mean={result['mean']}")

from sklearn.ensemble import IsolationForest
import pandas as pd
import numpy as np
 
class MultiDimAnomalyDetector:
    """基于 Isolation Forest 的多维异常检测"""
    
    def __init__(self, contamination: float = 0.05):
        self.model = IsolationForest(
            contamination=contamination,
            random_state=42,
            n_estimators=200
        )
        self.is_trained = False
    
    def train(self, data: pd.DataFrame):
        """用历史正常数据训练模型"""
        self.model.fit(data)
        self.is_trained = True
        print(f"模型训练完成，使用 {len(data)} 条数据")
    
    def detect(self, data: pd.DataFrame) -> pd.DataFrame:
        """检测异常"""
        if not self.is_trained:
            raise ValueError("模型尚未训练")
        
        predictions = self.model.predict(data)
        scores = self.model.decision_function(data)
        
        data = data.copy()
        data["is_anomaly"] = predictions == -1
        data["anomaly_score"] = scores
        return data
 
# 使用示例：监控服务器指标
np.random.seed(42)
normal_data = pd.DataFrame({
    "cpu_percent": np.random.normal(45, 10, 1000),
    "memory_percent": np.random.normal(60, 8, 1000),
    "request_latency_ms": np.random.normal(200, 30, 1000),
    "error_rate": np.random.normal(0.01, 0.005, 1000)
})
 
detector = MultiDimAnomalyDetector(contamination=0.05)
detector.train(normal_data)
 
# 模拟异常数据（内存泄漏 + 延迟飙升）
test_data = pd.DataFrame({
    "cpu_percent": [50, 55, 48, 90, 95],
    "memory_percent": [62, 65, 85, 92, 95],
    "request_latency_ms": [210, 220, 500, 800, 1200],
    "error_rate": [0.01, 0.02, 0.05, 0.15, 0.25]
})
 
results = detector.detect(test_data)
anomalies = results[results["is_anomaly"]]
print(f"检测到 {len(anomalies)} 个异常点")
print(anomalies)

模式 1：Zero-Shot 日志分析
├── 直接将日志片段发送给 LLM
├── LLM 基于预训练知识判断是否异常
├── 优势：无需训练数据
└── 局限：成本高，延迟大

模式 2：RAG 增强诊断
├── 将历史事件和 runbook 存入向量数据库
├── 异常发生时检索相似历史事件
├── LLM 结合历史经验生成诊断
└── 优势：结合组织特定知识

模式 3：Fine-tuned 专用模型
├── 用组织的日志数据微调小模型（如 Llama 3）
├── 部署为本地推理服务
├── 优势：低延迟、低成本、数据不出域
└── 局限：需要标注数据和训练资源

你是一位资深 SRE 工程师。请分析以下系统日志，识别异常模式并给出诊断。

## 日志片段
[粘贴最近 50-100 行日志]

## 系统上下文
- 服务名称：[服务名]
- 部署环境：[生产/预发布]
- 最近变更：[最近的部署或配置变更]
- 当前告警：[活跃告警列表]

## 请回答
1. 是否存在异常模式？如果是，描述异常的具体表现
2. 可能的根因是什么？（列出 Top 3 可能性，按概率排序）
3. 建议的排查步骤（按优先级排序）
4. 是否需要立即人工介入？（紧急/重要/可观察）
5. 建议的修复方案

告警触发
    │
    ▼
┌─────────────────────┐
│  1. 上下文收集       │
│  ├── 收集相关指标    │
│  ├── 拉取最近日志    │
│  ├── 查询最近变更    │
│  └── 获取拓扑关系    │
└─────────┬───────────┘
          │
          ▼
┌─────────────────────┐
│  2. 关联分析         │
│  ├── 时间关联        │
│  ├── 拓扑关联        │
│  ├── 变更关联        │
│  └── 历史相似事件    │
└─────────┬───────────┘
          │
          ▼
┌─────────────────────┐
│  3. 根因推理         │
│  ├── ML 模型推理     │
│  ├── LLM 语义分析    │
│  └── 规则引擎匹配    │
└─────────┬───────────┘
          │
          ▼
┌─────────────────────┐
│  4. 输出诊断报告     │
│  ├── 根因描述        │
│  ├── 影响范围        │
│  ├── 修复建议        │
│  └── 置信度评分      │
└─────────────────────┘

工作流：AI 自动诊断 Agent

触发器：Webhook（接收告警）或定时（每 5 分钟）
    │
    ▼
节点 1：HTTP Request — 健康检查
    - GET https://your-api.com/health
    - 超时设置：5 秒
    - 记录响应时间和状态码
    │
    ▼
节点 2：条件判断（IF）
    - 条件：响应时间 > 2000ms OR 状态码 != 200
    │
    ├── 否 → 结束（一切正常）
    │
    └── 是 ↓
          │
节点 3：并行收集上下文
    ├── HTTP Request → 获取最近 100 行日志（/api/logs?limit=100）
    ├── HTTP Request → 获取系统指标（/api/metrics）
    └── HTTP Request → 获取最近部署记录（GitHub API）
    │
    ▼
节点 4：AI Agent 节点（Claude/GPT）
    - System Prompt：你是 SRE 专家，请分析以下监控数据...
    - 输入：合并的日志 + 指标 + 部署记录
    - 输出：JSON 格式的诊断报告
    │
    ▼
节点 5：条件判断 — 严重程度
    ├── Critical → Slack 通知 + PagerDuty 告警 + 自动执行修复
    ├── Warning → Slack 通知 + 创建 Jira 工单
    └── Info → 记录到日志数据库

import httpx
import json
from datetime import datetime, timedelta
 
class AutoDiagnostics:
    """自动诊断引擎"""
    
    def __init__(self, llm_client, metrics_client, log_client):
        self.llm = llm_client
        self.metrics = metrics_client
        self.logs = log_client
    
    async def diagnose(self, alert: dict) -> dict:
        """接收告警，执行自动诊断"""
        
        # 1. 收集上下文
        context = await self._collect_context(alert)
        
        # 2. LLM 分析
        diagnosis = await self._llm_analyze(alert, context)
        
        # 3. 生成报告
        report = self._generate_report(alert, context, diagnosis)
        
        return report
    
    async def _collect_context(self, alert: dict) -> dict:
        """并行收集诊断所需的上下文信息"""
        service = alert.get("service", "unknown")
        time_range = timedelta(minutes=30)
        
        # 并行获取指标、日志、最近变更
        metrics = await self.metrics.query(
            service=service,
            metrics=["cpu", "memory", "latency", "error_rate"],
            duration=time_range
        )
        
        logs = await self.logs.search(
            service=service,
            level=["ERROR", "WARN"],
            limit=100,
            duration=time_range
        )
        
        return {
            "metrics": metrics,
            "logs": logs,
            "alert": alert,
            "timestamp": datetime.now().isoformat()
        }
    
    async def _llm_analyze(self, alert: dict, context: dict) -> dict:
        """使用 LLM 进行根因分析"""
        prompt = f"""
你是一位资深 SRE 工程师。请根据以下信息进行根因分析。
 
## 告警信息
- 告警名称：{alert.get('name')}
- 严重程度：{alert.get('severity')}
- 触发时间：{alert.get('triggered_at')}
- 受影响服务：{alert.get('service')}
 
## 系统指标（最近 30 分钟）
{json.dumps(context['metrics'], indent=2, ensure_ascii=False)}
 
## 错误日志（最近 100 条）
{chr(10).join(context['logs'][:20])}
 
请以 JSON 格式返回诊断结果：
{{
  "root_cause": "根因描述",
  "confidence": 0.0-1.0,
  "impact": "影响范围描述",
  "severity": "critical/warning/info",
  "suggested_fixes": ["修复建议1", "修复建议2"],
  "auto_fixable": true/false,
  "fix_command": "如果可自动修复，给出命令"
}}
"""
        response = await self.llm.chat(prompt)
        return json.loads(response)
    
    def _generate_report(self, alert, context, diagnosis) -> dict:
        """生成诊断报告"""
        return {
            "alert_id": alert.get("id"),
            "diagnosed_at": datetime.now().isoformat(),
            "root_cause": diagnosis.get("root_cause"),
            "confidence": diagnosis.get("confidence"),
            "severity": diagnosis.get("severity"),
            "impact": diagnosis.get("impact"),
            "suggested_fixes": diagnosis.get("suggested_fixes"),
            "auto_fixable": diagnosis.get("auto_fixable"),
            "fix_command": diagnosis.get("fix_command"),
            "context_summary": {
                "metrics_analyzed": len(context.get("metrics", {})),
                "logs_analyzed": len(context.get("logs", []))
            }
        }

Level 0：手动修复
├── 人工收到告警 → 人工排查 → 人工修复
└── MTTR：30 分钟 - 数小时

Level 1：自动化脚本修复
├── 告警触发 → 执行预定义脚本 → 通知结果
├── 示例：自动重启崩溃的服务
└── MTTR：1-5 分钟

Level 2：智能决策修复
├── 告警触发 → AI 分析根因 → 选择修复策略 → 执行 → 验证
├── 示例：根据错误类型选择回滚/重启/扩容
└── MTTR：30 秒 - 2 分钟

Level 3：预测性自愈
├── AI 预测即将发生的故障 → 提前执行预防措施
├── 示例：检测到内存泄漏趋势，在 OOM 前自动重启
└── MTTR：0（故障未发生）

# deployment-with-self-healing.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: api-server
spec:
  replicas: 3
  strategy:
    type: RollingUpdate
    rollingUpdate:
      maxUnavailable: 1
      maxSurge: 1
  template:
    spec:
      containers:
      - name: api
        image: myapp:v1.2.3
        
        # 存活探针：检测容器是否需要重启
        livenessProbe:
          httpGet:
            path: /healthz
            port: 8080
          initialDelaySeconds: 15
          periodSeconds: 10
          failureThreshold: 3  # 连续 3 次失败则重启
        
        # 就绪探针：检测容器是否可以接收流量
        readinessProbe:
          httpGet:
            path: /ready
            port: 8080
          initialDelaySeconds: 5
          periodSeconds: 5
          failureThreshold: 2
        
        # 启动探针：慢启动应用的保护
        startupProbe:
          httpGet:
            path: /healthz
            port: 8080
          failureThreshold: 30
          periodSeconds: 10
        
        resources:
          requests:
            cpu: "250m"
            memory: "256Mi"
          limits:
            cpu: "1000m"
            memory: "512Mi"
      
      # Pod 中断预算：确保滚动更新时的可用性
      terminationGracePeriodSeconds: 30
 
---
# HPA：基于指标的自动扩缩容
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: api-server-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: api-server
  minReplicas: 2
  maxReplicas: 10
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70
  - type: Resource
    resource:
      name: memory
      target:
        type: Utilization
        averageUtilization: 80
  behavior:
    scaleUp:
      stabilizationWindowSeconds: 60
      policies:
      - type: Pods
        value: 2
        periodSeconds: 60
    scaleDown:
      stabilizationWindowSeconds: 300
      policies:
      - type: Pods
        value: 1
        periodSeconds: 120

# robusta-playbook.yaml
# Robusta 是一个开源的 K8s 故障自动修复平台
customPlaybooks:
  # 场景 1：OOMKilled 自动增加内存限制
  - triggers:
    - on_pod_oom_killed: {}
    actions:
    - resource_babysitter:
        increase_pct: 25  # 自动增加 25% 内存限制
        max_memory: "2Gi"
    - create_finding:
        title: "Pod OOMKilled - 已自动增加内存限制"
        severity: MEDIUM
    - slack_sender:
        channel: "#ops-alerts"
 
  # 场景 2：CrashLoopBackOff 自动收集诊断信息
  - triggers:
    - on_pod_crash_loop:
        restart_reason: "CrashLoopBackOff"
    actions:
    - pod_bash_enricher:
        bash_command: "cat /var/log/app.log | tail -50"
    - logs_enricher: {}
    - slack_sender:
        channel: "#ops-alerts"
 
  # 场景 3：节点资源不足时自动驱逐低优先级 Pod
  - triggers:
    - on_node_pressure:
        pressure_type: "MemoryPressure"
    actions:
    - node_running_pods_enricher: {}
    - create_finding:
        title: "节点内存压力 - 检查低优先级 Pod"
        severity: HIGH

#!/usr/bin/env python3
"""
自动回滚脚本：监控部署后的健康状态，异常时自动回滚
配合 CI/CD 使用，在部署后执行
"""
import subprocess
import time
import httpx
import sys
 
class DeploymentGuard:
    """部署守卫：监控新部署并在异常时自动回滚"""
    
    def __init__(
        self,
        health_url: str,
        deployment_name: str,
        namespace: str = "default",
        check_interval: int = 10,
        max_checks: int = 30,
        error_threshold: float = 0.1
    ):
        self.health_url = health_url
        self.deployment = deployment_name
        self.namespace = namespace
        self.check_interval = check_interval
        self.max_checks = max_checks
        self.error_threshold = error_threshold
    
    def watch(self) -> bool:
        """监控部署健康状态，返回 True 表示健康，False 触发回滚"""
        print(f"🔍 开始监控部署 {self.deployment}...")
        errors = 0
        
        for i in range(self.max_checks):
            try:
                resp = httpx.get(self.health_url, timeout=5)
                if resp.status_code == 200:
                    print(f"  ✅ 检查 {i+1}/{self.max_checks}: 健康")
                else:
                    errors += 1
                    print(f"  ⚠️ 检查 {i+1}: 状态码 {resp.status_code}")
            except Exception as e:
                errors += 1
                print(f"  ❌ 检查 {i+1}: 连接失败 - {e}")
            
            error_rate = errors / (i + 1)
            if error_rate > self.error_threshold and i >= 5:
                print(f"🚨 错误率 {error_rate:.0%} 超过阈值，触发回滚！")
                self._rollback()
                return False
            
            time.sleep(self.check_interval)
        
        print(f"✅ 部署监控完成，错误率: {errors/self.max_checks:.0%}")
        return True
    
    def _rollback(self):
        """执行 Kubernetes 回滚"""
        cmd = [
            "kubectl", "rollout", "undo",
            f"deployment/{self.deployment}",
            f"-n", self.namespace
        ]
        print(f"⏪ 执行回滚: {' '.join(cmd)}")
        result = subprocess.run(cmd, capture_output=True, text=True)
        
        if result.returncode == 0:
            print(f"✅ 回滚成功: {result.stdout}")
        else:
            print(f"❌ 回滚失败: {result.stderr}")
            sys.exit(1)
 
# 使用示例
if __name__ == "__main__":
    guard = DeploymentGuard(
        health_url="https://api.example.com/health",
        deployment_name="api-server",
        namespace="production",
        check_interval=10,   # 每 10 秒检查一次
        max_checks=30,       # 最多检查 30 次（5 分钟）
        error_threshold=0.1  # 错误率超过 10% 触发回滚
    )
    
    success = guard.watch()
    sys.exit(0 if success else 1)

你是一位资深 SRE 工程师。请分析以下告警并给出根因分析。

## 告警详情
- 告警名称：[告警名称]
- 严重程度：[Critical/Warning/Info]
- 触发时间：[时间]
- 受影响服务：[服务名]
- 告警描述：[描述]

## 相关指标
[粘贴最近 30 分钟的关键指标数据]

## 最近变更
[最近 24 小时的部署或配置变更]

## 请回答
1. 最可能的根因（Top 3，按概率排序）
2. 影响范围评估
3. 建议的排查步骤（按优先级）
4. 推荐的修复方案
5. 预防措施建议

请为以下服务生成 Prometheus 告警规则（PrometheusRule YAML 格式）。

## 服务信息
- 服务名称：[服务名]
- 服务类型：[Web API / 后台任务 / 数据库 / 消息队列]
- SLO 目标：[可用性 99.9% / 延迟 P99 < 500ms]

## 需要监控的指标
1. [指标1，如：请求延迟]
2. [指标2，如：错误率]
3. [指标3，如：CPU/内存使用率]
4. [指标4，如：队列积压]

## 要求
- 使用动态阈值（基于历史数据的百分比偏差）而非静态阈值
- 包含 for 持续时间（避免瞬时抖动误报）
- 告警分为 warning 和 critical 两级
- 每条规则包含有意义的 summary 和 description 注解
- 包含 runbook_url 注解（指向修复文档）

请为以下告警场景生成标准化的 Runbook（运维手册）。

## 告警场景
- 告警名称：[告警名称]
- 触发条件：[触发条件描述]
- 影响：[对用户/业务的影响]

## Runbook 要求
请包含以下部分：
1. **概述**：告警含义和严重程度
2. **排查步骤**：从最可能的原因开始，逐步排查
3. **修复操作**：每种根因对应的修复命令
4. **验证步骤**：如何确认问题已修复
5. **升级路径**：什么情况下需要升级处理
6. **预防措施**：如何避免再次发生

每个步骤请包含具体的命令（kubectl、curl、SQL 等）。

请为以下项目设计一套 AIOps 监控方案。

## 项目信息
- 技术栈：[前端框架 / 后端语言 / 数据库 / 消息队列]
- 部署方式：[K8s / Docker / Serverless / VPS]
- 团队规模：[人数]
- 月预算：[监控预算]
- 当前痛点：[描述当前运维痛点]

## 请设计
1. **监控架构**：推荐的工具组合和数据流
2. **关键指标**：需要监控的核心指标列表（按 RED/USE 方法论）
3. **告警策略**：告警规则设计和分级策略
4. **仪表板**：Grafana 仪表板布局建议
5. **自愈方案**：可自动化修复的场景和实现方式
6. **成本估算**：各工具的预估月费用

预算：$0-50/月
目标：实现基础 AIOps 能力

架构：
┌─────────────────────────────────────────────┐
│                  告警通知层                    │
│         Slack / Telegram / 邮件               │
└──────────────────┬──────────────────────────┘
                   │
┌──────────────────┴──────────────────────────┐
│              AI 诊断层（n8n）                  │
│  ┌─────────┐  ┌──────────┐  ┌────────────┐  │
│  │健康检查  │  │日志分析   │  │成本监控     │  │
│  │Agent    │  │Agent     │  │Agent       │  │
│  └─────────┘  └──────────┘  └────────────┘  │
└──────────────────┬──────────────────────────┘
                   │
┌──────────────────┴──────────────────────────┐
│              数据采集层                        │
│  ┌──────────┐  ┌──────────┐  ┌───────────┐  │
│  │Uptime    │  │New Relic │  │Railway    │  │
│  │Kuma      │  │免费层     │  │Metrics    │  │
│  └──────────┘  └──────────┘  └───────────┘  │
└─────────────────────────────────────────────┘

# docker-compose.yml — 部署在同一 VPS 或 Railway
version: "3"
services:
  uptime-kuma:
    image: louislam/uptime-kuma:1
    volumes:
      - ./data:/app/data
    ports:
      - "3001:3001"
    restart: unless-stopped

// 在 Node.js API 入口文件最顶部添加
require('newrelic');
 
// newrelic.js 配置文件
'use strict';
exports.config = {
  app_name: ['my-saas-api'],
  license_key: process.env.NEW_RELIC_LICENSE_KEY,
  logging: { level: 'info' },
  
  // AI 异常检测（免费层包含）
  ai_monitoring: { enabled: true },
  
  // 慢查询追踪
  slow_sql: { enabled: true, max_samples: 10 },
  
  // 分布式追踪
  distributed_tracing: { enabled: true },
  
  // 错误收集
  error_collector: {
    enabled: true,
    ignore_status_codes: [404]
  }
};

n8n 工作流配置：

工作流 1：健康检查 + AI 诊断
├── 触发器：Cron（每 5 分钟）
├── HTTP Request：GET /api/health
├── IF：status != 200 OR latency > 3000ms
├── AI Agent（Claude Haiku / GPT-4o-mini）：
│   "分析以下健康检查失败信息，给出可能原因和建议：
│    状态码: {{$json.statusCode}}
│    响应时间: {{$json.responseTime}}ms
│    错误信息: {{$json.body}}"
└── Slack/Telegram：发送诊断结果

工作流 2：每日成本报告
├── 触发器：Cron（每天 9:00）
├── HTTP Request：Railway API 获取用量数据
├── AI Agent：分析用量趋势，预测月底费用
└── Slack：发送日报

工作流 3：部署后自动验证
├── 触发器：Webhook（GitHub Actions 部署完成后调用）
├── Wait：等待 60 秒（让新版本稳定）
├── HTTP Request：连续 5 次健康检查
├── IF：任何一次失败
├── AI Agent：分析失败原因
└── Slack：通知 + 建议是否回滚

告警分级策略：

🔴 Critical（立即通知 + 电话）：
├── 服务完全不可用（连续 3 次健康检查失败）
├── 数据库连接失败
└── 错误率 > 10%

🟡 Warning（Slack 通知）：
├── 响应时间 > 2 秒
├── 错误率 > 2%
├── 内存使用 > 80%
└── 数据库慢查询 > 5 秒

🔵 Info（日报汇总）：
├── 部署完成通知
├── 每日成本报告
└── 每周性能趋势