系统规划与管理师视角：企业级云原生故障诊断与自愈系统架构设计与代码实现

┌─────────────────────────────────────────────────────────────┐
│                    接入层（Web Console / API Gateway）       │
└─────────────────────────────────────────────────────────────┘
                               │
┌─────────────────────────────────────────────────────────────┐
│                   决策编排层（决策引擎核心）                  │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────────┐  │
│  │ 规则管理平台  │  │ 工作流引擎   │  │ 权限与熔断控制器 │  │
│  └──────────────┘  └──────────────┘  └──────────────────┘  │
└─────────────────────────────────────────────────────────────┘
                               │
┌─────────────────────────────────────────────────────────────┐
│                   诊断分析层（AI大脑）                       │
│  ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌─────────────┐  │
│  │时序异常  │ │ 日志模式 │ │ 根因定位 │ │ 因果推断AI  │  │
│  │检测引擎  │ │ 匹配引擎 │ │ 算法库   │ │ 辅助模块    │  │
│  └──────────┘ └──────────┘ └──────────┘ └─────────────┘  │
└─────────────────────────────────────────────────────────────┘
                               │
┌─────────────────────────────────────────────────────────────┐
│                  可观测数据层（统一数据中台）                 │
│  ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌─────────────┐  │
│  │Victoria │ │   Loki   │ │  Jaeger  │ │  CMDB配置   │  │
│  │ Metrics  │ │   Logs   │ │  Traces  │ │  关系库     │  │
│  └──────────┘ └──────────┘ └──────────┘ └─────────────┘  │
└─────────────────────────────────────────────────────────────┘
                               │
┌─────────────────────────────────────────────────────────────┐
│                   执行与自动化层（手脚）                     │
│         Kubernetes API / ServiceNow / 自定义脚本           │
└─────────────────────────────────────────────────────────────┘

package anomaly

import (
    "context"
    "fmt"
    "math"
    "sync"
    "time"
    "github.com/prometheus/client_golang/api"
    v1 "github.com/prometheus/client_golang/api/prometheus/v1"
    "github.com/prometheus/common/model"
)

// MultiDimensionAnomalyDetector 多维异常检测器
type MultiDimensionAnomalyDetector struct {
    promClient v1.API
    mu         sync.RWMutex
    // 存储最近 7 天的历史数据用于计算基线 (生产环境使用 TSDB 存储)
    baselineCache map[string][]float64 
}

// MetricSnapshot 某一时刻的多维指标快照
type MetricSnapshot struct {
    CPUUsage    float64
    MemoryUsage float64
    RequestRate float64
    ErrorRate   float64
    Timestamp   time.Time
}

// AnomalyResult 异常检测结果
type AnomalyResult struct {
    IsAnomaly    bool
    Score        float64
    Reason       string
    AnomalyDim   []string // 哪些维度异常
}

// Detect 执行多维异常检测
func (d *MultiDimensionAnomalyDetector) Detect(ctx context.Context, podName string, namespace string) (*AnomalyResult, error) {
    // 1. 获取当前实时数据 (PromQL 聚合查询)
    snapshot, err := d.fetchCurrentMetrics(ctx, podName, namespace)
    if err != nil {
        return nil, err
    }

    // 2. 获取历史基线 (从内存缓存或 TSDB 中读取)
    baselineKey := fmt.Sprintf("%s_%s", namespace, podName)
    d.mu.RLock()
    history, ok := d.baselineCache[baselineKey]
    d.mu.RUnlock()
    
    if !ok || len(history) < 100 { // 数据不足时降级为静态阈值
        return d.fallbackStaticCheck(snapshot), nil
    }

    // 3. 计算多维马氏距离 (Mahalanobis Distance) 或 简单的 Z-Score 组合
    // 此处简化：分别计算 Z-Score，若超过 3 个维度超过 2.5 标准差则报警
    dimResults := make(map[string]bool)
    abnormalDims := []string{}
    totalScore := 0.0

    // 模拟计算 CPU Z-Score (实际需从 history 中计算均值和方差)
    cpuZ := d.calculateZScore(snapshot.CPUUsage, history, "cpu")
    memZ := d.calculateZScore(snapshot.MemoryUsage, history, "mem")
    reqZ := d.calculateZScore(snapshot.RequestRate, history, "req")
    errZ := d.calculateZScore(snapshot.ErrorRate, history, "err")

    if math.Abs(cpuZ) > 2.5 {
        abnormalDims = append(abnormalDims, "CPU")
        dimResults["CPU"] = true
    }
    if math.Abs(memZ) > 2.5 {
        abnormalDims = append(abnormalDims, "Memory")
        dimResults["Memory"] = true
    }
    if math.Abs(reqZ) > 2.5 {
        abnormalDims = append(abnormalDims, "RequestRate")
        dimResults["RequestRate"] = true
    }
    if math.Abs(errZ) > 2.5 {
        abnormalDims = append(abnormalDims, "ErrorRate")
        dimResults["ErrorRate"] = true
    }

    // 综合评分：异常维度数量 * 平均偏离度
    totalScore = (math.Abs(cpuZ) + math.Abs(memZ) + math.Abs(reqZ) + math.Abs(errZ)) / 4.0

    result := &AnomalyResult{
        IsAnomaly:  len(abnormalDims) >= 3,
        Score:      totalScore,
        AnomalyDim: abnormalDims,
    }

    if result.IsAnomaly {
        result.Reason = fmt.Sprintf("多维度异常，异常维度: %v, 综合偏离度: %.2f", abnormalDims, totalScore)
    }
    return result, nil
}

// fetchCurrentMetrics 通过 PromQL 获取当前指标 (实际代码省略查询语法)
func (d *MultiDimensionAnomalyDetector) fetchCurrentMetrics(ctx context.Context, pod, ns string) (*MetricSnapshot, error) {
    // 伪代码实现: 调用 Prometheus API
    // queryCPU := fmt.Sprintf(`rate(container_cpu_usage_seconds_total{pod="%s", namespace="%s"}[1m])`, pod, ns)
    return &MetricSnapshot{
        CPUUsage:    75.6,  // 示例数据
        MemoryUsage: 82.3,
        RequestRate: 1250.0,
        ErrorRate:   8.2,
        Timestamp:   time.Now(),
    }, nil
}

// calculateZScore 计算当前值在历史基线中的标准分数
func (d *MultiDimensionAnomalyDetector) calculateZScore(value float64, history []float64, dim string) float64 {
    // 实际需要分维度存储历史，此处简化为计算均值和标准差
    // 生产环境建议使用 EWMA (指数加权移动平均) 来避免毛刺
    mean, std := d.computeStats(history)
    if std == 0 {
        return 0
    }
    return (value - mean) / std
}

func (d *MultiDimensionAnomalyDetector) computeStats(data []float64) (mean, std float64) {
    sum := 0.0
    for _, v := range data {
        sum += v
    }
    mean = sum / float64(len(data))
    var variance float64
    for _, v := range data {
        variance += math.Pow(v-mean, 2)
    }
    std = math.Sqrt(variance / float64(len(data)))
    return
}

// fallbackStaticCheck 静态阈值降级
func (d *MultiDimensionAnomalyDetector) fallbackStaticCheck(snapshot *MetricSnapshot) *AnomalyResult {
    if snapshot.CPUUsage > 95 || snapshot.MemoryUsage > 95 || snapshot.ErrorRate > 10 {
        return &AnomalyResult{IsAnomaly: true, Reason: "静态阈值触发", Score: 1.0}
    }
    return &AnomalyResult{IsAnomaly: false}
}

# 根因定位模块 (Python 实现，利用 K8s client 与 Loki)
import re
from collections import Counter
from kubernetes import client, watch
from loki import LokiClient  # 假设存在

class RootCauseLocator:
    def __init__(self):
        self.k8s_core = client.CoreV1Api()
        self.loki = LokiClient(url="http://loki:3100")
    
    def locate(self, pod_name, namespace, anomaly_time, window_minutes=5):
        """定位根因：返回 top 3 可能原因"""
        causes = []
        
        # 1. 查询 K8s 事件 (重点关注 Failed, Error, OOMKilled)
        events = self.k8s_core.list_namespaced_event(
            namespace=namespace,
            field_selector=f"involvedObject.name={pod_name}"
        )
        for event in events.items:
            if event.type in ["Warning", "Error"]:
                causes.append(f"[K8s Event] {event.reason}: {event.message}")
                # 若检测到 OOMKilled，直接返回
                if "OOMKilled" in event.message:
                    return ["OOMKilled (内存溢出)", event.message]
        
        # 2. 查询 Loki 日志进行词频突增分析
        query = f'{{pod="{pod_name}", namespace="{namespace}"}} |= "error" or "panic" or "timeout"'
        # 获取异常时间点前后 window 的日志
        logs = self.loki.query_range(
            query=query,
            start=anomaly_time - timedelta(minutes=window_minutes),
            end=anomaly_time + timedelta(minutes=window_minutes)
        )
        
        # 提取错误码、SQL 异常、NullPointer 等模式
        pattern_counter = Counter()
        for log_line in logs:
            # 正则提取核心错误信息
            match = re.search(r'(ERROR|Exception|panic|timeout|failed to)\s*[:]?\s*(.+)', log_line, re.IGNORECASE)
            if match:
                pattern_counter[match.group(2).strip()[:50]] += 1
        
        # 取高频模式
        for pattern, count in pattern_counter.most_common(3):
            causes.append(f"[Log Pattern] {pattern} (出现 {count} 次)")
        
        return causes if causes else ["未找到明确根因，建议人工介入"]

package executor

import (
    "context"
    "fmt"
    "strconv"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/client-go/kubernetes"
)

// CircuitBreaker 熔断器
type CircuitBreaker struct {
    client    *kubernetes.Clientset
    maxRatio  float64 // 最大允许操作比例，默认 0.3
}

// CheckDryRun 执行前检查: 计算操作目标数量，若超过阈值则熔断
func (cb *CircuitBreaker) CheckDryRun(ctx context.Context, namespace, deploymentName string, targetReplicas int32) error {
    // 1. 获取当前 Deployment 信息
    deploy, err := cb.client.AppsV1().Deployments(namespace).Get(ctx, deploymentName, metav1.GetOptions{})
    if err != nil {
        return err
    }
    
    currentReplicas := *deploy.Spec.Replicas
    if currentReplicas == 0 {
        return fmt.Errorf("当前副本数为0，拒绝执行")
    }
    
    // 2. 计算操作比例 (重启/回滚 影响比例)
    // 若重启，通常等价于全部替换；若仅扩容，则影响新增部分
    ratio := float64(targetReplicas) / float64(currentReplicas)
    
    // 3. 动态阈值: 若应用等级为 P0 (核心) 则阈值更严格
    appLevel := cb.getAppLevel(namespace, deploymentName)
    threshold := cb.maxRatio
    if appLevel == "P0" {
        threshold = 0.1 // 核心应用每次最多操作 10%
    }
    
    if ratio > threshold {
        return fmt.Errorf("熔断触发: 操作比例 %.2f%% 超过阈值 %.2f%%，需人工审批", ratio*100, threshold*100)
    }
    
    return nil
}

// ExecuteWithRollback 带有回滚机制的执行
func (cb *CircuitBreaker) ExecuteWithRollback(ctx context.Context, action func() error, rollback func() error, checkInterval time.Duration) error {
    // 1. 执行动作
    if err := action(); err != nil {
        return err
    }
    
    // 2. 观察业务指标 (假设通过 Prometheus 查询)
    ticker := time.NewTicker(checkInterval)
    defer ticker.Stop()
    
    for i := 0; i < 5; i++ { // 等待 5 个周期
        <-ticker.C
        // 查询 error_rate
        errorRate, err := cb.queryErrorRate(ctx)
        if err != nil {
            continue
        }
        // 若错误率继续升高 (超过动作前的 1.2 倍)，触发回滚
        if errorRate > 1.2 * cb.baselineErrorRate {
            fmt.Printf("检测到错误率上升至 %.2f%%，触发自动回滚\n", errorRate)
            return rollback()
        }
    }
    return nil
}