预警比告警早 23 分钟：时序异常检测与大模型辅助的故障预警实践

一、 从告警到预警：思维模式的转变

1.1 告警和预警的区别

1.2 预警系统的架构

flowchart TD
    A["Prometheus"] --> B["时序异常检测"]
    B --> C["预警事件"]
    C --> D["LLM 分析引擎"]
    D --> E["严重性判断"]
    E --> F["P0/P1: 实时通知"]
    E --> G["P2/P3: 聚合日报"]

核心差异在于：预警系统不依赖阈值，而是通过机器学习检测数据的趋势变化和模式偏离。

二、 时序预警检测的三层模型

2.1 第一层：短周期趋势检测（5-15 分钟窗口）

# short_trend_detector.py — 短周期趋势检测
import numpy as np
from sklearn.linear_model import LinearRegression

class ShortTrendDetector:
    """短周期趋势检测：捕捉 5-15 分钟内的变化趋势"""
    
    def detect_trend(self, series: np.ndarray) -> dict:
        """检测短期趋势"""
        X = np.arange(len(series)).reshape(-1, 1)
        y = series.values.reshape(-1, 1)
        
        model = LinearRegression()
        model.fit(X, y)
        
        slope = model.coef_[0][0]
        r2 = model.score(X, y)
        
        # 斜率为正说明上升趋势，为负说明下降
        # R²越高说明趋势越显著
        trend_strength = abs(slope) * r2
        
        return {
            'slope': slope,
            'r_squared': r2,
            'trend_strength': trend_strength,
            'prediction': model.predict([[len(series)]])[0][0],  # 预测下一个值
            'direction': 'up' if slope > 0 else 'down',
            'severity': 'high' if trend_strength > 0.5 else (
                        'medium' if trend_strength > 0.2 else 'low')
        }

# 使用示例
detector = ShortTrendDetector()

# 每 1 分钟检测一次 CPU 使用率的趋势
cpu_values = get_metric('node_cpu_seconds_total{mode="idle"}', window='15m')
result = detector.detect_trend(cpu_values)

if result['direction'] == 'down' and result['severity'] == 'high':
    # CPU 空闲率持续下降（使用率持续上升）
    send_warning('CPU 使用率持续上升',
                 f'斜率：{result["slope"]:.2f}, 预测下一个值：{result["prediction"]:.1f}')

2.2 第二层：周期性模式偏离检测（纵向对比）

很多指标具有明显的周期性——比如工作日和周末的 QPS 差异很大。预警系统需要"理解"这种周期性：

# cycle_deviation_detector.py — 周期性偏离检测
class CycleDeviationDetector:
    """周期性偏离检测：对比当前值和历史同期值"""
    
    def __init__(self, prometheus_url: str):
        self.prom = PrometheusConnect(url=prometheus_url, disable_ssl=True)
    
    def detect_deviation(self, metric: str, service: str) -> dict:
        """检测与历史同期相比的偏离度"""
        now = datetime.now()
        
        # 当前值（最近 5 分钟的均值）
        current_query = f'avg_over_time({metric}{{service="{service}"}}[5m])'
        current_value = self._query_one(current_query)
        
        # 上周同一天同一时段的值（历史基线）
        last_week_start = now - timedelta(days=7)
        history_query = f'avg_over_time({metric}{{service="{service}"}}[5m]) offset 1w'
        history_value = self._query_one(history_query)
        
        if history_value == 0:
            return {'deviation_pct': 0, 'severity': 'unknown'}
        
        # 计算偏离度
        deviation = (current_value - history_value) / history_value * 100
        
        severity = 'high' if abs(deviation) > 50 else (
                  'medium' if abs(deviation) > 30 else (
                  'warning' if abs(deviation) > 15 else 'normal'))
        
        return {
            'current': current_value,
            'history_baseline': history_value,
            'deviation_pct': f"{deviation:+.1f}%",
            'direction': 'up' if deviation > 0 else 'down',
            'severity': severity,
            'suggested_action': self._suggest_action(metric, deviation)
        }
    
    def _suggest_action(self, metric: str, deviation: float) -> str:
        """根据指标类型和偏离方向给出建议"""
        if 'error_rate' in metric and deviation > 30:
            return '检查最近一次发布，回滚可能存在问题的版本'
        elif 'latency' in metric and deviation > 30:
            return '检查上游服务状态和数据库连接池'
        elif 'cpu' in metric and deviation > 50:
            return '检查是否有异常进程或流量突增'
        else:
            return '建议关注，持续观察 15 分钟'

2.3 第三层：跨指标关联预警

单个指标异常可能是噪声，但多个指标同时异常一定是信号：

# cross_metric_warning.py — 跨指标关联预警
class CrossMetricWarning:
    """跨指标关联预警"""
    
    def check_correlation(self, metrics: dict) -> list:
        """检查多个指标之间的关联模式"""
        warnings = []
        
        # 规则 1: 请求量正常但错误率上升 → 服务本身有问题
        if (metrics['qps']['deviation'] < 10 and 
            metrics['error_rate']['deviation'] > 50):
            warnings.append({
                'type': 'service_degradation',
                'confidence': 0.85,
                'message': '请求量正常但错误率飙升，服务本身存在缺陷',
                'suggestions': ['检查最近代码变更', '查看错误日志分布']
            })
        
        # 规则 2: 请求量上升 + 延迟上升 + CPU 上升 → 正常扩展
        if (metrics['qps']['deviation'] > 30 and 
            metrics['latency']['deviation'] > 20 and 
            metrics['cpu']['deviation'] > 20):
            warnings.append({
                'type': 'normal_scaling',
                'confidence': 0.90,
                'message': '流量增长导致延迟和 CPU 同步上升，建议提前扩容',
                'suggestions': ['提前扩容 Pod', '检查 HPA 配置是否合理']
            })
        
        # 规则 3: 请求量不变 + 延迟上升 + CPU 不变 → 线程阻塞/code issue
        if (abs(metrics['qps']['deviation']) < 10 and 
            metrics['latency']['deviation'] > 30 and 
            abs(metrics['cpu']['deviation']) < 10):
            warnings.append({
                'type': 'thread_contention',
                'confidence': 0.75,
                'message': '请求量不变但延迟上升且 CPU 未同步上升，可能有线程阻塞',
                'suggestions': ['jstack 查看线程状态', '检查数据库连接池']
            })
        
        return warnings

三、 LLM 辅助预警判断

预警系统的最大挑战是误报率。如果预警太多，团队会进入"预警疲劳"，最终无视所有预警。

我们引入 LLM 做"二次判断"——在预警发送前，让 LLM 检查这个预警是否合理：

# llm_warning_filter.py — LLM 预警过滤器
class LLMWarningFilter:
    """用 LLM 过滤和分级预警"""
    
    async def should_alert(self, warning: dict, context: dict) -> dict:
        """判断预警是否需要发送"""
        
        prompt = f"""你是一个 SRE 预警过滤专家。请判断以下预警是否需要立即通知工程师。

## 预警信息
- 类型：{warning['type']}
- 置信度：{warning['confidence']}
- 消息：{warning['message']}

## 当前系统上下文
- 当前时间：{context['current_time']}
- 是否工作时间：{context['is_working_hours']}
- 最近是否有发布：{context['recent_deploy']}
- 类似预警的历史命中率：{context['historical_accuracy']}

## 判断标准
1. 如果是深夜（22:00-08:00）且置信度 < 0.8 → 延迟到早上聚合发送
2. 如果预警类型在历史上误报率 > 50% → 标记为"低优先级"
3. 如果是已知的周期性波动 → 不发送
4. 如果影响核心交易链路 → 即使置信度低也要发送

请返回：
- decision: send / delay / suppress
- reason: 简短理由
- priority: P1/P2/P3
"""
        # 调用 LLM...
        result = await self._call_llm(prompt)
        return result

四、 实践效果

最让我们惊喜的数据是：72% 的预警在 24 小时内被验证为有效。也就是说，每发出 10 条预警，有 7 条以上是真正有价值、需要关注的。

总结

从"被动响应告警"到"主动接收预警"，看起来只是换个词，但背后是整个运维思维模式的转变。

告警告诉你"已经出事了"；预警告诉你"即将出事了"。前者需要止损，后者只需要预防——两者的响应成本和影响范围天差地别。

部署这套预警系统半年后，团队形成了一种新习惯：每天早上先看"预警日报"，了解今天可能出现的风险，而不是等告警响了再去救火。

预防的价值，永远大于止损。