智能容量规划与资源优化推荐：从"拍脑袋"到数据驱动的资源决策

一、容量规划的"猜谜游戏"：为什么总是"要么不够，要么浪费"？

容量规划是运维团队最头疼的决策之一。申请少了，业务高峰时服务崩溃；申请多了，资源闲置率超过 60%，成本账单触目惊心。传统容量规划依赖经验判断——"去年双十一峰值是日常的 5 倍，今年按 6 倍准备"。但业务增长不是线性的，新功能上线可能改变流量模式，突发事件的流量特征也无法从历史数据中预测。

AI 驱动的容量规划通过时序预测和资源画像，将"拍脑袋"决策转变为数据驱动决策。核心思路是：基于历史指标预测未来负载，结合资源利用率画像识别浪费，生成精确的资源申请和优化建议。

二、容量规划的数据模型与预测流程

容量规划分为三个阶段：负载预测（未来需要多少算力）、资源映射（需要多少实例/配额）、优化推荐（哪些资源可以释放）。负载预测是基础，资源映射是核心，优化推荐是价值落地。

flowchart TD
    A[历史指标数据<br/>CPU / Memory / QPS] --> B[时序预测模型<br/>Prophet / LSTM]
    B --> C[未来负载预测<br/>7天 / 30天 / 90天]
    C --> D[资源映射引擎<br/>负载 → 实例数 / 配额]

    E[当前资源利用率<br/>实例画像] --> F[浪费识别<br/>低利用率实例]
    F --> G[优化推荐<br/>缩容 / 降配 / 释放]

    D --> H[容量报告<br/>需要多少资源]
    G --> I[优化报告<br/>可以省多少资源]
    H --> J[决策建议<br/>扩容 + 缩容 + 成本预估]
    I --> J

    subgraph "预测维度"
        K[常规负载<br/>工作日 / 周末模式]
        L[周期性峰值<br/>促销 / 节假日]
        M[趋势性增长<br/>用户量增长]
    end

    B --> K
    B --> L
    B --> M

关键预测维度：

• 常规负载：工作日与周末的周期性模式，日内的流量波峰波谷
• 周期性峰值：促销活动、节假日的流量突增，需要提前预留容量
• 趋势性增长：用户量持续增长带来的长期负载上升趋势

三、智能容量规划系统的实现

# capacity_planner.py — AI 驱动的智能容量规划系统
# 设计意图：基于历史指标预测未来负载，结合资源利用率画像
# 生成精确的容量申请和优化建议

import numpy as np
from dataclasses import dataclass
from typing import List, Dict, Tuple, Optional
from datetime import datetime, timedelta
from collections import defaultdict


@dataclass
class MetricPoint:
    """指标数据点"""
    timestamp: datetime
    cpu_usage: float       # 0-1
    memory_usage: float    # 0-1
    qps: float             # 请求速率
    latency_p99: float     # P99 延迟 ms


@dataclass
class ResourceProfile:
    """资源画像"""
    service_name: str
    instance_type: str
    instance_count: int
    avg_cpu: float         # 平均 CPU 利用率
    avg_memory: float      # 平均内存利用率
    peak_cpu: float        # 峰值 CPU 利用率
    peak_memory: float     # 峰值内存利用率
    cost_per_month: float  # 月成本


@dataclass
class CapacityRecommendation:
    """容量建议"""
    service_name: str
    current_instances: int
    recommended_instances: int
    action: str            # scale_up / scale_down / maintain / downgrade
    reason: str
    estimated_cost_change: float  # 成本变化（正=增加，负=减少）


class TimeSeriesForecaster:
    """时序预测器：基于指数平滑和周期分解"""

    def __init__(self):
        self.seasonal_period = 7 * 24  # 一周的小时数

    def forecast(
        self,
        history: List[MetricPoint],
        horizon_days: int = 30,
    ) -> List[MetricPoint]:
        """预测未来 N 天的负载"""
        if len(history) < self.seasonal_period * 2:
            # 数据不足时使用简单外推
            return self._simple_extrapolate(history, horizon_days)

        # 分解趋势和周期分量
        qps_values = [p.qps for p in history]
        trend, seasonal = self._decompose(qps_values)

        # 预测
        last_ts = history[-1].timestamp
        predictions = []

        for i in range(horizon_days * 24):
            future_ts = last_ts + timedelta(hours=i + 1)
            trend_val = trend[-1] + (trend[-1] - trend[-2]) * (i / 24)
            seasonal_idx = (len(history) + i) % self.seasonal_period
            seasonal_val = seasonal[seasonal_idx]

            predicted_qps = max(0, trend_val + seasonal_val)

            predictions.append(MetricPoint(
                timestamp=future_ts,
                cpu_usage=0,  # 由资源映射引擎计算
                memory_usage=0,
                qps=predicted_qps,
                latency_p99=0,
            ))

        return predictions

    def _decompose(
        self, values: List[float]
    ) -> Tuple[List[float], List[float]]:
        """简单时序分解：提取趋势和周期分量"""
        n = len(values)
        period = self.seasonal_period

        # 移动平均提取趋势
        trend = []
        window = period
        for i in range(n):
            start = max(0, i - window // 2)
            end = min(n, i + window // 2 + 1)
            trend.append(np.mean(values[start:end]))

        # 提取周期分量
        detrended = [v - t for v, t in zip(values, trend)]
        seasonal = [0.0] * period
        counts = [0] * period

        for i, val in enumerate(detrended):
            idx = i % period
            seasonal[idx] += val
            counts[idx] += 1

        seasonal = [s / max(c, 1) for s, c in zip(seasonal, counts)]

        return trend, seasonal

    def _simple_extrapolate(
        self, history: List[MetricPoint], horizon_days: int
    ) -> List[MetricPoint]:
        """简单线性外推（数据不足时的降级方案）"""
        if len(history) < 2:
            return []

        qps_values = [p.qps for p in history]
        avg_qps = np.mean(qps_values)
        growth_rate = (qps_values[-1] - qps_values[0]) / max(len(qps_values), 1)

        last_ts = history[-1].timestamp
        predictions = []

        for i in range(horizon_days * 24):
            future_ts = last_ts + timedelta(hours=i + 1)
            predicted_qps = max(0, avg_qps + growth_rate * (len(qps_values) + i))

            predictions.append(MetricPoint(
                timestamp=future_ts,
                cpu_usage=0,
                memory_usage=0,
                qps=predicted_qps,
                latency_p99=0,
            ))

        return predictions


class ResourceMapper:
    """资源映射引擎：将预测负载映射为资源需求"""

    def __init__(self, target_cpu: float = 0.7, target_memory: float = 0.8):
        self.target_cpu = target_cpu          # 目标 CPU 利用率
        self.target_memory = target_memory    # 目标内存利用率

    def map_to_instances(
        self,
        service_name: str,
        predictions: List[MetricPoint],
        current_profile: ResourceProfile,
        qps_per_instance: float,
    ) -> CapacityRecommendation:
        """将预测 QPS 映射为实例数"""
        # 计算预测期内的峰值 QPS
        peak_qps = max(p.qps for p in predictions) if predictions else 0

        # 计算所需实例数（考虑安全裕度）
        safety_margin = 1.2  # 20% 安全裕度
        required_instances = int(
            np.ceil(peak_qps / qps_per_instance * safety_margin)
        )
        required_instances = max(required_instances, 2)  # 最少 2 个实例

        # 生成建议
        current = current_profile.instance_count
        if required_instances > current:
            action = "scale_up"
            reason = (
                f"预测峰值 QPS={peak_qps:.0f}，"
                f"当前 {current} 实例无法承载，"
                f"建议扩容至 {required_instances} 实例"
            )
        elif required_instances < current * 0.6:
            action = "scale_down"
            reason = (
                f"预测峰值 QPS={peak_qps:.0f}，"
                f"当前 {current} 实例利用率不足，"
                f"建议缩容至 {required_instances} 实例"
            )
        else:
            action = "maintain"
            reason = "当前实例数满足预测负载，无需调整"

        # 估算成本变化
        unit_cost = current_profile.cost_per_month / max(current, 1)
        cost_change = (required_instances - current) * unit_cost

        return CapacityRecommendation(
            service_name=service_name,
            current_instances=current,
            recommended_instances=required_instances,
            action=action,
            reason=reason,
            estimated_cost_change=cost_change,
        )


class WasteDetector:
    """浪费检测器：识别低利用率资源"""

    def __init__(
        self,
        low_cpu_threshold: float = 0.15,
        low_memory_threshold: float = 0.25,
        sustained_hours: int = 72,
    ):
        self.low_cpu_threshold = low_cpu_threshold
        self.low_memory_threshold = low_memory_threshold
        self.sustained_hours = sustained_hours

    def detect_waste(
        self, profiles: List[ResourceProfile]
    ) -> List[CapacityRecommendation]:
        """检测低利用率资源，生成优化建议"""
        recommendations = []

        for profile in profiles:
            # 判断是否持续低利用率
            is_low_cpu = profile.avg_cpu < self.low_cpu_threshold
            is_low_memory = profile.avg_memory < self.low_memory_threshold

            if is_low_cpu and is_low_memory:
                # CPU 和内存都低：建议缩容或降配
                if profile.instance_count > 2:
                    target_count = max(2, profile.instance_count // 2)
                    action = "scale_down"
                    reason = (
                        f"CPU 均值 {profile.avg_cpu:.0%}，"
                        f"内存均值 {profile.avg_memory:.0%}，"
                        f"建议缩容至 {target_count} 实例"
                    )
                else:
                    action = "downgrade"
                    reason = (
                        f"CPU 均值 {profile.avg_cpu:.0%}，"
                        f"建议降配实例类型"
                    )

                unit_cost = profile.cost_per_month / max(profile.instance_count, 1)
                if action == "scale_down":
                    cost_change = -(profile.instance_count - target_count) * unit_cost
                else:
                    cost_change = -unit_cost * 0.4  # 降配约省 40%

                recommendations.append(CapacityRecommendation(
                    service_name=profile.service_name,
                    current_instances=profile.instance_count,
                    recommended_instances=(
                        target_count if action == "scale_down"
                        else profile.instance_count
                    ),
                    action=action,
                    reason=reason,
                    estimated_cost_change=cost_change,
                ))

        return recommendations


class CapacityPlanner:
    """智能容量规划系统"""

    def __init__(self):
        self.forecaster = TimeSeriesForecaster()
        self.mapper = ResourceMapper()
        self.waste_detector = WasteDetector()

    def generate_plan(
        self,
        history: Dict[str, List[MetricPoint]],
        profiles: List[ResourceProfile],
        qps_per_instance: Dict[str, float],
        horizon_days: int = 30,
    ) -> Dict:
        """生成完整的容量规划报告"""
        scale_recommendations = []
        waste_recommendations = []

        # Step 1: 负载预测与资源映射
        for service_name, metrics in history.items():
            if not metrics:
                continue

            predictions = self.forecaster.forecast(metrics, horizon_days)

            profile = next(
                (p for p in profiles if p.service_name == service_name),
                None,
            )
            if not profile:
                continue

            qps_cap = qps_per_instance.get(service_name, 1000)
            rec = self.mapper.map_to_instances(
                service_name, predictions, profile, qps_cap
            )
            scale_recommendations.append(rec)

        # Step 2: 浪费检测
        waste_recs = self.waste_detector.detect_waste(profiles)
        waste_recommendations.extend(waste_recs)

        # 汇总
        total_cost_change = sum(
            r.estimated_cost_change
            for r in scale_recommendations + waste_recommendations
        )

        return {
            "period": f"Next {horizon_days} days",
            "scale_recommendations": [
                {
                    "service": r.service_name,
                    "action": r.action,
                    "current": r.current_instances,
                    "recommended": r.recommended_instances,
                    "reason": r.reason,
                    "cost_change": r.estimated_cost_change,
                }
                for r in scale_recommendations
            ],
            "waste_recommendations": [
                {
                    "service": r.service_name,
                    "action": r.action,
                    "reason": r.reason,
                    "cost_change": r.estimated_cost_change,
                }
                for r in waste_recommendations
            ],
            "total_estimated_cost_change": total_cost_change,
        }

四、智能容量规划的 Trade-offs

预测精度与数据量的矛盾：时序预测的精度高度依赖历史数据的长度和质量。少于 2 个完整周期的数据，预测结果不可靠。但新服务上线时往往没有足够的历史数据。解决方案是使用同类服务的指标作为先验，或使用保守的线性外推作为降级方案。

安全裕度的选择：安全裕度过大导致资源浪费，过小导致高峰期容量不足。20% 的安全裕度是常见选择，但不同业务的风险承受能力不同。核心交易服务可能需要 50% 的裕度，而内部工具 10% 即可。建议按服务等级（SLA）分级设置安全裕度。

预测与现实的偏差：突发事件（如社交媒体爆文导致的流量突增）无法从历史数据中预测。容量规划需要与弹性伸缩（HPA）配合——规划提供基础容量，HPA 应对突发流量。两者互补而非替代。

成本优化的业务风险：缩容和降配可以节省成本，但也降低了系统的冗余能力。一个被缩容的服务在突发流量时可能无法及时扩容。建议在非核心服务上优先执行优化，核心服务保持较高的冗余度。

五、总结

AI 驱动的智能容量规划将资源决策从"拍脑袋"推向数据驱动。时序预测模型基于历史指标预测未来负载，资源映射引擎将负载转化为实例需求，浪费检测器识别低利用率资源。三者结合生成"扩容 + 缩容 + 降配"的综合建议。但预测精度受数据量限制、安全裕度需要按业务分级、突发事件需要弹性伸缩兜底、成本优化存在业务风险。在实际落地中，建议将容量规划作为月度例行流程，结合 HPA 的弹性伸缩应对短期波动，按服务等级差异化设置安全裕度。容量规划的目标不是"精确预测未来"，而是"在不确定性中做出最优的资源分配决策"。