从经验估算到模型预测：基于Prophet+LSTM的云原生资源容量规划工程化落地

一、资源规划的两种失败：要么不够用，要么太浪费

大多数Kubernetes集群的资源配置来自Pod的requests字段——开发者在YAML里填写的值。这个值有两个倾向：要么偏保守（"先设2核4G，不够再加"），要么偏乐观（"我本地跑只要200M内存"）。

保守倾向的结果是资源利用率长期低于15%。一个32核128G的节点只跑了4个总共request了3.5核的服务，剩下的28.5核被浪费。乐观倾向的结果是凌晨三点OOM告警——流量高峰时段内存请求突然翻倍，cgroup限制被击穿。

资源预测的本质是在两个维度上做推断：趋势维度（未来30天Pod数会不会因为业务增长而翻倍）和周期维度（每天20:00的CPU峰值是多少，每周五的流量比周三高多少）。纯靠经验估算只能覆盖一阶变化，二阶的周期性波动和多维度的交叉影响超出人脑计算能力。

flowchart TB
    subgraph 数据采集["数据采集层"]
        P[Prometheus<br/>KSM/Node Exporter] -->|"时序指标"| K[Kafka缓冲]
        CPU[CPU使用率]
        MEM[内存使用量]
        POD[Pod副本数]
        NODE[节点利用率]
    end
    
    K --> F[特征工程<br/>缺失值填充+异常值清洗]
    
    subgraph 预测引擎["双模型融合预测引擎"]
        F --> P1[Prophet<br/>趋势+周期分解]
        F --> P2[LSTM<br/>非线性序列预测]
        P1 --> E[加权融合<br/>Prophet权重0.4<br/>LSTM权重0.6]
        P2 --> E
    end
    
    E --> R[资源建议<br/>Request/Limit/HPA阈值]
    R --> O[输出<br/>容量报告+告警]

二、为什么单模型不够：Prophet负责"看懂"、LSTM负责"算准"

容量预测的输入是Prometheus时序数据——每分钟一个浮点数的CPU使用率、每5分钟一个采样点的内存用量。问题有两个特性：明显的周期规律（每天晚上CPU降、工作日比周末高），叠加的非线性变化（大促前业务方提前扩容，曲线突然跳跃）。

Prophet（Meta开源的时序预测库）把时序分解为三部分：趋势项（长期增长/衰减）、周期项（日周期+周周期）、节假日效应。Prophet对周期性捕捉得很好，但面对业务策略调整（如"下周三多上5个副本"）这类跳跃型变化，Prophet的自动变点检测反应滞后。

LSTM补上了这个短板。LSTM对序列中的非线性模式有很强的拟合能力。将过去168小时（一周）的分钟级数据输入LSTM，预测未来24小时的资源曲线。LSTM的缺点是对数据质量和训练样本量敏感——少于30天的历史数据，LSTM的预测误差会显著上升。

"""
双模型融合的容量预测核心代码
Prophet负责趋势和周期分解，LSTM负责非线性短期预测
"""
import numpy as np
import pandas as pd
from typing import Tuple, Dict, Optional
from dataclasses import dataclass
from datetime import datetime, timedelta


# 注意：需要先安装依赖
# pip install prophet torch scikit-learn

@dataclass
class PredictionResult:
    """容量预测结果"""
    timestamp: pd.DatetimeIndex
    cpu_forecast: np.ndarray      # CPU预测值（核）
    mem_forecast: np.ndarray      # 内存预测值（MB）
    cpu_upper: np.ndarray         # CPU置信区间上界
    cpu_lower: np.ndarray         # CPU置信区间下界
    peak_cpu: float               # 预测窗口内CPU峰值
    peak_mem: float               # 预测窗口内内存峰值
    recommended_cpu_request: float  # 推荐CPU request值
    recommended_mem_request: float  # 推荐内存request值


class ProphetForecaster:
    """Prophet趋势+周期预测器"""
    
    def __init__(self, growth: str = "linear"):
        """
        Args:
            growth: 趋势模式，"linear"线性或"logistic"逻辑回归
        """
        self.growth = growth
        self.model = None
    
    def fit(self, df: pd.DataFrame) -> None:
        """
        训练Prophet模型
        
        Args:
            df: 包含 ds(时间戳) 和 y(指标值) 两列的DataFrame
        """
        try:
            from prophet import Prophet
        except ImportError:
            raise ImportError(
                "请先安装prophet: pip install prophet"
            )
        
        self.model = Prophet(
            growth=self.growth,
            yearly_seasonality=False,
            weekly_seasonality=True,     # 启用周周期
            daily_seasonality=True,      # 启用日周期
            changepoint_prior_scale=0.05,  # 变点检测灵敏度
            seasonality_prior_scale=10.0,
            interval_width=0.95          # 95%置信区间
        )
        self.model.fit(df)
    
    def predict(self, periods: int, freq: str = "5min") -> pd.DataFrame:
        """预测未来periods个时间点"""
        if self.model is None:
            raise RuntimeError("模型未训练，请先调用fit()")
        
        future = self.model.make_future_dataframe(
            periods=periods, freq=freq
        )
        forecast = self.model.predict(future)
        return forecast


class LSTMForecaster:
    """基于PyTorch的LSTM短期预测器"""
    
    def __init__(
        self,
        input_window: int = 168,   # 输入窗口（小时数，默认一周）
        output_window: int = 24,   # 预测窗口（小时数，默认一天）
        hidden_size: int = 64,
        num_layers: int = 2
    ):
        self.input_window = input_window
        self.output_window = output_window
        self.hidden_size = hidden_size
        self.num_layers = num_layers
        self.model = None
        self._scaler_params: Dict[str, Tuple[float, float]] = {}
    
    def _minmax_scale(
        self, data: np.ndarray
    ) -> Tuple[np.ndarray, float, float]:
        """Min-Max归一化到[0, 1]"""
        d_min = data.min()
        d_max = data.max()
        if d_max == d_min:
            return np.zeros_like(data), d_min, d_min + 1e-8
        scaled = (data - d_min) / (d_max - d_min)
        return scaled, d_min, d_max
    
    def fit(self, series: np.ndarray, epochs: int = 50) -> None:
        """
        训练LSTM模型
        
        Args:
            series: 形状为(n,)的时序数据
            epochs: 训练轮数
        """
        import torch
        import torch.nn as nn
        
        scaled, vmin, vmax = self._minmax_scale(series)
        self._scaler_params["default"] = (vmin, vmax)
        
        # 构造训练样本：滑动窗口
        X, y = [], []
        for i in range(len(scaled) - self.input_window - self.output_window):
            X.append(scaled[i:i + self.input_window])
            y.append(scaled[
                i + self.input_window:
                i + self.input_window + self.output_window
            ])
        
        if not X:
            raise ValueError(
                f"数据长度({len(series)})不足以构造训练样本，"
                f"需要至少{self.input_window + self.output_window}个点"
            )
        
        X = np.array(X).reshape(-1, self.input_window, 1)
        y = np.array(y).reshape(-1, self.output_window)
        
        X_tensor = torch.tensor(X, dtype=torch.float32)
        y_tensor = torch.tensor(y, dtype=torch.float32)
        
        # 定义LSTM模型
        class SimpleLSTM(nn.Module):
            def __init__(self, input_size, hidden_size, num_layers, output_size):
                super().__init__()
                self.lstm = nn.LSTM(
                    input_size, hidden_size, num_layers,
                    batch_first=True
                )
                self.fc = nn.Linear(hidden_size, output_size)
            
            def forward(self, x):
                out, _ = self.lstm(x)
                # 取最后一个时间步的输出
                out = self.fc(out[:, -1, :])
                return out
        
        self.model = SimpleLSTM(
            input_size=1,
            hidden_size=self.hidden_size,
            num_layers=self.num_layers,
            output_size=self.output_window
        )
        
        criterion = nn.MSELoss()
        optimizer = torch.optim.Adam(self.model.parameters(), lr=0.001)
        
        # 训练循环
        for epoch in range(epochs):
            self.model.train()
            optimizer.zero_grad()
            outputs = self.model(X_tensor)
            loss = criterion(outputs, y_tensor)
            loss.backward()
            optimizer.step()
            
            if (epoch + 1) % 10 == 0:
                print(f"Epoch [{epoch+1}/{epochs}], Loss: {loss.item():.6f}")
    
    def predict(self, recent_data: np.ndarray) -> np.ndarray:
        """基于最近的input_window个数据点预测"""
        import torch
        
        if self.model is None:
            raise RuntimeError("模型未训练，请先调用fit()")
        
        scaled, _, _ = self._minmax_scale(recent_data)
        X = torch.tensor(
            scaled[-self.input_window:].reshape(1, -1, 1),
            dtype=torch.float32
        )
        
        self.model.eval()
        with torch.no_grad():
            pred = self.model(X).numpy().flatten()
        
        # 反归一化
        vmin, vmax = self._scaler_params["default"]
        pred = pred * (vmax - vmin) + vmin
        return np.maximum(pred, 0)  # 资源值不能为负


class HybridCapacityPredictor:
    """Prophet + LSTM混合容量预测器"""
    
    def __init__(self, prophet_weight: float = 0.4):
        """
        Args:
            prophet_weight: Prophet预测值在融合中的权重（0~1）
        """
        self.prophet_weight = prophet_weight
        self.lstm_weight = 1.0 - prophet_weight
        self.prophet = ProphetForecaster()
        self.lstm = LSTMForecaster()
    
    def fit(
        self,
        timestamps: pd.DatetimeIndex,
        cpu_values: np.ndarray,
        mem_values: np.ndarray
    ) -> None:
        """训练双模型"""
        # Prophet训练
        df_cpu = pd.DataFrame({
            "ds": timestamps,
            "y": cpu_values
        })
        df_mem = pd.DataFrame({
            "ds": timestamps,
            "y": mem_values
        })
        
        print("训练Prophet模型...")
        self.prophet.fit(df_cpu)
        
        print("训练LSTM模型...")
        self.lstm.fit(cpu_values, epochs=30)
    
    def predict(
        self,
        recent_data: np.ndarray,
        future_periods: int = 288  # 未来24小时（每5分钟一个点）
    ) -> PredictionResult:
        """生成融合预测结果"""
        # Prophet预测
        prophet_result = self.prophet.predict(
            periods=future_periods, freq="5min"
        )
        prophet_cpu = prophet_result["yhat"].values[-future_periods:]
        prophet_cpu_upper = (
            prophet_result["yhat_upper"].values[-future_periods:]
        )
        prophet_cpu_lower = (
            prophet_result["yhat_lower"].values[-future_periods:]
        )
        
        # LSTM预测
        lstm_cpu = self.lstm.predict(recent_data)
        
        # 加权融合（对齐长度）
        min_len = min(len(prophet_cpu), len(lstm_cpu))
        prophet_cpu = prophet_cpu[:min_len]
        lstm_cpu = lstm_cpu[:min_len]
        
        fused_cpu = (
            self.prophet_weight * prophet_cpu
            + self.lstm_weight * lstm_cpu
        )
        
        # 计算推荐值：95分位预测值 × 1.2倍安全系数
        peak_cpu = float(np.percentile(fused_cpu, 95))
        recommended_cpu = peak_cpu * 1.2
        
        # 内存预测（简化：基于CPU内存比估算）
        mem_ratio = 4.0  # 默认 1核 : 4GB
        peak_mem = float(np.percentile(fused_cpu, 95) * mem_ratio)
        recommended_mem = peak_mem * 1.2
        
        # 构建未来时间戳
        last_ts = datetime.now()
        future_ts = pd.date_range(
            start=last_ts + timedelta(minutes=5),
            periods=min_len,
            freq="5min"
        )
        
        return PredictionResult(
            timestamp=future_ts,
            cpu_forecast=np.maximum(fused_cpu, 0),
            mem_forecast=np.maximum(fused_cpu * mem_ratio, 0),
            cpu_upper=prophet_cpu_upper[:min_len],
            cpu_lower=prophet_cpu_lower[:min_len],
            peak_cpu=peak_cpu,
            peak_mem=peak_mem,
            recommended_cpu_request=recommended_cpu,
            recommended_mem_request=recommended_mem,
        )


# 使用示例
if __name__ == "__main__":
    # 模拟14天的CPU使用率数据（每5分钟一个点）
    np.random.seed(42)
    n_points = 14 * 24 * 12  # 14天 × 24小时 × 12个5分钟
    timestamps = pd.date_range(
        start="2026-06-22", periods=n_points, freq="5min"
    )
    
    # 模拟数据：基值2核 + 日周期波动 + 趋势增长 + 噪声
    base = 2.0
    daily_pattern = (
        1.0 * np.sin(2 * np.pi * np.arange(n_points) / (24 * 12))
    )
    trend = 0.00002 * np.arange(n_points)  # 缓慢增长趋势
    noise = 0.1 * np.random.randn(n_points)
    cpu_data = base + daily_pattern + trend + noise
    cpu_data = np.maximum(cpu_data, 0.1)
    
    predictor = HybridCapacityPredictor()
    predictor.fit(timestamps, cpu_data, cpu_data * 4)
    
    result = predictor.predict(cpu_data)
    
    print(f"未来24小时CPU峰值预测: {result.peak_cpu:.2f}核")
    print(f"推荐CPU request: {result.recommended_cpu_request:.2f}核")
    print(f"推荐内存request: {result.recommended_mem_request:.2f}MB")

三、从预测到行动：与HPA和VPA的联动闭环

预测模型的价值不在于给出一个数字，而在于驱动自动化的资源调整。具体落地路径分三步：

第一步：预测值驱动HPA阈值。传统HPA基于当前CPU使用率触发扩缩容。预测模型将"未来1小时CPU将达80%"作为信号提前注入HPA，实现预扩容。在Kubernetes中，这可以通过KEDA（Kubernetes Event-Driven Autoscaling）的Prometheus Scaler实现。

apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: capacity-predictive-scaler
spec:
  scaleTargetRef:
    name: payment-api
  triggers:
  - type: prometheus
    metadata:
      serverAddress: http://thanos-query:9090
      metricName: predicted_cpu_usage_1h
      query: |
        capacity_prediction{service="payment-api", window="1h"}
      threshold: "70"  # 预测1小时后CPU超70%则提前扩容

第二步：预测报告自动推送。定时（如每周一）生成各服务的容量预测报告，包含"未来7天CPU/内存峰值预测"和"当前request是否充足"两项核心指标，推送到企业微信或飞书群。

第三步：VPA建议与人工审批结合。Vertical Pod Autoscaler基于历史数据生成资源建议。预测模型可以作为VPA的"先验知识"，在VPA的推荐值上叠加趋势修正。

四、预测模型的边界：什么时候应该信任模型、什么时候应该信任人

容量预测模型的能力边界需要被清晰定义。以下四种场景下，预测模型的结论应该被降级处理：

新上线服务（冷启动）。没有历史数据的服务，预测模型只能退化为固定值初始化。此时应依赖LOAD TEST的结果设定初始值，并在前两周用实际数据快速迭代模型。

业务策略突变。如果运营团队决定下周做一次大促活动，Prophet的节假日效应需要手动标记事件日期。更稳妥的做法是在预测接口中暴露一个override_factor参数，允许业务方临时乘以1.5或2.0的系数。

周期性模式被打破。如果某天凌晨的服务出现异常流量（如定时任务配置错误导致循环调用），预测模型会将异常峰值当作正常周期的一部分学习进去。特征工程中的异常值清洗（基于IQR或Z-score）是必选项。

预测置信度不足时回退。对预测结果的置信区间做评估——如果95%置信区间的宽度超过均值的30%，说明预测质量有问题，此时不应采纳模型的建议值。

五、总结

容量预测的本质不是"猜未来"，而是"把周期性规律从噪声中分离出来"。Prophet+LSTM的双模型融合策略在工程上提供了一个实用的折中：Prophet保证可解释性（能告诉你"为什么预测CPU会在晚上8点上升"），LSTM保证短期精度（能捕捉到Prophet漏掉的非线性突变）。

落地时的核心产出不是预测曲线本身，而是两个数字：推荐的CPU request值和推荐的内存request值。这两个数字如果能在每一次Deployment变更时被自动注入到YAML中，才是真正的闭环。