从发现到定位的最后一公里：机器学习时序异常检测与根因分析结合实践

一、为什么要结合？

1.1 孤立的痛点

先看两个常见的失败场景：

场景一：异常洪水

03:15 Prometheus告警：CPU > 90%
03:16 Prometheus告警：内存 > 85%
03:17 Prometheus告警：磁盘IO > 95%
03:18 Prometheus告警：延迟P99 > 2s
... 接下来10分钟，又触发了20条告警

值班工程师面对一堆告警，无从下手。这些指标到底哪个是因、哪个是果？

场景二：有异常无告警

服务响应变慢，用户投诉
但所有指标都在阈值范围内
——因为没有达到告警阈值

这就是传统阈值告警的盲区。而机器学习时序检测可以捕捉到"指标虽然没超阈值，但趋势已经明显异常"的情况。

二、系统设计：从检测到关联再到定位

2.1 总体架构

flowchart TD
    A["[时序指标]"] --> B["异常检测(多算法)"] --> C["异常事件"]
    D["[错误日志]"] --> E["错误聚合(聚类)"] --> F["错误模式"]
    C --> G["异常关联引擎"]
    F --> G
    G --> H["LLM根因分析器"] --> I["根因分析报告"]

2.2 多算法时序异常检测

单一算法容易误判，我们采用多算法投票机制：

# multi_detector.py — 多算法异常检测
import numpy as np
from scipy import stats
from sklearn.ensemble import IsolationForest
import pandas as pd

class MultiAlgorithmDetector:
    """多算法异常检测引擎"""
    
    def __init__(self):
        self.detectors = {
            '3sigma': self._detect_3sigma,
            'mad': self._detect_mad,
            'iqr': self._detect_iqr,
            'isolation_forest': self._detect_iforest
        }
    
    def _detect_3sigma(self, series: pd.Series):
        """3-sigma检测"""
        mean = series.mean()
        std = series.std()
        anomalies = (series > mean + 3*std) | (series < mean - 3*std)
        return anomalies, mean, std
    
    def _detect_mad(self, series: pd.Series):
        """MAD（中位数绝对偏差）检测，对异常值更鲁棒"""
        median = series.median()
        mad = np.median(np.abs(series - median))
        # 阈值：median ± 3 * 1.4826 * MAD
        threshold = 3 * 1.4826 * mad
        anomalies = (series > median + threshold) | (series < median - threshold)
        return anomalies, median, threshold
    
    def _detect_iqr(self, series: pd.Series):
        """IQR四分位距检测"""
        q1 = series.quantile(0.25)
        q3 = series.quantile(0.75)
        iqr = q3 - q1
        lower = q1 - 1.5 * iqr
        upper = q3 + 1.5 * iqr
        anomalies = (series < lower) | (series > upper)
        return anomalies, lower, upper
    
    def _detect_iforest(self, series: pd.Series):
        """Isolation Forest检测"""
        values = series.values.reshape(-1, 1)
        model = IsolationForest(contamination=0.05, random_state=42)
        preds = model.fit_predict(values)
        anomalies = pd.Series(preds == -1, index=series.index)
        return anomalies, None, None
    
    def ensemble_detect(self, series: pd.Series, 
                        min_votes: int = 2):
        """集成投票：至少2个算法判定为异常才算异常"""
        votes = np.zeros(len(series))
        
        for name, detector in self.detectors.items():
            anomalies, _, _ = detector(series)
            votes += anomalies.astype(int)
        
        # 投票决定
        final_anomaly = votes >= min_votes
        anomaly_score = votes / len(self.detectors)
        
        return {
            'is_anomaly': final_anomaly,
            'confidence': anomaly_score,
            'votes': votes,
            'details': {name: detector(series)[0].tolist() 
                       for name, detector in self.detectors.items()}
        }

2.3 日志错误模式聚类

# log_cluster.py — 日志错误模式聚类
import re
from collections import Counter
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.cluster import DBSCAN

class ErrorLogCluster:
    """错误日志模式聚类"""
    
    def __init__(self):
        self.patterns = {}
        self.vectorizer = TfidfVectorizer(
            analyzer='char_wb',
            ngram_range=(3, 5),
            max_features=1000
        )
    
    def normalize_log(self, message: str) -> str:
        """将变量部分替换为占位符"""
        # 替换IP地址
        message = re.sub(r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}', '{ip}', message)
        # 替换数字
        message = re.sub(r'\d+', '{n}', message)
        # 替换UUID
        message = re.sub(r'[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}', 
                        '{uuid}', message)
        return message
    
    def cluster(self, log_entries: list) -> dict:
        """聚类错误日志"""
        messages = [entry['message'] for entry in log_entries]
        normalized = [self.normalize_log(m) for m in messages]
        
        # TF-IDF向量化 + DBSCAN聚类
        X = self.vectorizer.fit_transform(normalized)
        clustering = DBSCAN(eps=0.3, min_samples=2, metric='cosine')
        labels = clustering.fit_predict(X)
        
        # 统计每个聚类的数量和样本
        clusters = {}
        for i, label in enumerate(labels):
            if label == -1:
                continue  # 噪点忽略
            if label not in clusters:
                clusters[label] = {
                    'count': 0,
                    'pattern': normalized[i],
                    'sample_message': messages[i]
                }
            clusters[label]['count'] += 1
        
        # 按数量排序
        sorted_clusters = sorted(
            clusters.values(),
            key=lambda x: x['count'],
            reverse=True
        )
        
        return {
            'total_errors': len(log_entries),
            'clusters': sorted_clusters[:10],
            'unique_patterns': len(clusters)
        }

2.4 异常关联引擎

这是最关键的一步——把指标异常和日志错误关联起来：

# anomaly_correlator.py — 异常关联引擎
from datetime import datetime, timedelta
import numpy as np

class AnomalyCorrelator:
    """异常关联引擎：找到指标异常和日志错误的因果关系"""
    
    def __init__(self, time_window: int = 5):
        self.time_window = time_window  # 关联时间窗口（分钟）
    
    def correlate(self, metric_anomalies: dict, 
                  log_clusters: dict,
                  time_range: tuple) -> dict:
        """核心关联逻辑"""
        correlations = []
        
        # 对每个异常指标，检查同一时间窗口内的日志模式
        for metric_name, anomaly_info in metric_anomalies.items():
            anomaly_times = anomaly_info['anomaly_times']
            
            for atime in anomaly_times:
                window_start = atime - timedelta(minutes=self.time_window)
                window_end = atime + timedelta(minutes=self.time_window)
                
                # 找这个时间窗口内的错误日志
                matching_errors = [
                    c for c in log_clusters['clusters']
                    if any(window_start <= e['timestamp'] <= window_end 
                          for e in c.get('entries', []))
                ]
                
                if matching_errors:
                    correlations.append({
                        'metric': metric_name,
                        'anomaly_time': atime.isoformat(),
                        'anomaly_value': anomaly_info['value'],
                        'baseline': anomaly_info['baseline'],
                        'deviation': f"{((anomaly_info['value'] / anomaly_info['baseline']) - 1) * 100:.1f}%",
                        'related_errors': [
                            {
                                'pattern': c['pattern'],
                                'count': c['count'],
                                'sample': c['sample_message']
                            }
                            for c in matching_errors
                        ],
                        'correlation_strength': len(matching_errors) / len(log_clusters['clusters'])
                    })
        
        # 按关联强度排序
        correlations.sort(key=lambda x: x['correlation_strength'], reverse=True)
        
        return {
            'total_correlations': len(correlations),
            'top_correlations': correlations[:5],
            'uncorrelated_anomalies': self._find_uncorrelated(
                metric_anomalies, correlations
            )
        }
    
    def _find_uncorrelated(self, anomalies, correlations):
        """找出来关联不上的异常（可能是未知问题）"""
        correlated_metrics = {c['metric'] for c in correlations}
        return [
            name for name in anomalies 
            if name not in correlated_metrics
        ]

2.5 LLM根因分析

# llm_analyzer.py — LLM根因分析器
class LLMRootCauseAnalyzer:
    """基于关联结果，用LLM做根因分析"""
    
    def analyze(self, correlation_result: dict, 
                service_name: str) -> str:
        """生成根因分析"""
        correlations = correlation_result['top_correlations']
        
        # 按时间线排序
        timeline = sorted(correlations, 
                         key=lambda x: x['anomaly_time'])
        
        # 找到最早出现的异常（极有可能是根因）
        earliest = timeline[0] if timeline else None
        
        prompt = f"""根据以下关联分析结果，判断{service_name}的故障根因：

## 执行时间线（按时间排序）
{self._format_timeline(timeline)}

## 最早出现的异常
{json.dumps(earliest, indent=2, ensure_ascii=False) if earliest else '无'}

## 未关联的异常指标
{json.dumps(correlation_result.get('uncorrelated_anomalies', []), ensure_ascii=False)}

请分析：
1. 最早异常的指标是什么？它是否可能是根因？
2. 其他异常是指标级联效应还是独立事件？
3. 给出最可能的根因（最多3个，按概率排序）
4. 下一步排查建议
"""
        # 调用LLM...
        return self._call_llm(prompt)

三、生产案例

在一次生产故障中，我们的系统检测到了以下关联：

flowchart TD
    Start["检测到5个指标异常，2个错误日志模式"] --> Analysis["关联分析结果"]

    subgraph AnalysisDetails [关联分析详情]
        direction TB
        A1["1. [最早] order-db 活跃连接数 50→200 (16:30:05)"] --> A1_Err["关联错误：conn pool exhausted (count: 847)"]
        A1 --> A1_Str["关联强度：0.83"]

        A2["2. order-service CPU 40%→85% (16:30:30)"] --> A2_Err["关联错误：connection timeout (count: 423)"]
        A2 --> A2_Str["关联强度：0.67"]

        A3["3. api-gateway 延迟P99 80ms→3.2s (16:31:00)"] --> A3_Err["关联错误：upstream timeout (count: 215)"]
        A3 --> A3_Str["关联强度：0.54"]
    end

    Analysis --> RootCause["根因概率"]
    
    subgraph RootCauseDetails [根因分析]
        direction TB
        RC1["1. 数据库连接池耗尽 — 80%<br/>（最早异常，关联最强）"]
        RC2["2. 数据库慢查询 — 15%"]
        RC3["3. 网络问题 — 5%"]
    end

    RootCause --> RC1
    RootCause --> RC2
    RootCause --> RC3

    RC1 --> Action["排查建议"]
    
    subgraph ActionDetails [排查步骤]
        direction TB
        S1["1. 检查MySQL max_connections和连接池配置"]
        S2["2. 查看慢查询日志"]
        S3["3. 检查是否有未关闭的事务"]
    end

    Action --> S1
    Action --> S2
    Action --> S3

实际根因：数据库连接池max_connections配置为200，但应用实例从3个扩容到10个时，每个实例的连接池扩大，总和超过了200。

四、总结

从"多个维度独立检测"到"跨维度关联分析"，这是运维可观测性从L2向L3演进的关键一步。机器学习做时序检测发现异常，日志聚类聚合错误模式，关联引擎连接指标和日志，LLM给出最终分析——这四层组合起来，让"发现到定位"的平均时间从40分钟降到了8分钟。