面向容量规划的微服务架构分层与关键抽象设计

capacity-planner/
├── app.py                      # FastAPI应用主入口
├── core/
│   ├── __init__.py
│   ├── models.py               # 核心数据模型定义（Service, Instance, Metric, Plan）
│   ├── algorithm.py            # 容量规划核心算法
│   ├── storage.py              # 数据存储抽象与内存实现
│   └── config.py               # 应用配置
├── api/
│   ├── __init__.py
│   ├── endpoints.py            # RESTful API路由定义
│   └── dependencies.py         # FastAPI依赖注入（如获取存储实例）
├── requirements.txt            # Python依赖列表
└── .env.example                # 环境变量示例文件

"""
容量规划核心模型定义。
"""
from datetime import datetime
from enum import Enum
from typing import List, Optional, Dict, Any
from pydantic import BaseModel, Field, validator

class ServiceStatus(str, Enum):
    """服务状态枚举"""
    HEALTHY = "healthy"
    WARNING = "warning"
    CRITICAL = "critical"

class PlanAction(str, Enum):
    """规划动作枚举"""
    SCALE_OUT = "scale_out"
    SCALE_IN = "scale_in"
    DO_NOTHING = "do_nothing"

class Service(BaseModel):
    """服务抽象"""
    id: str = Field(..., description="服务唯一标识")
    name: str = Field(..., description="服务名称")
    min_instances: int = Field(1, ge=1, description="最小实例数")
    max_instances: int = Field(10, ge=1, description="最大实例数")
    target_cpu_utilization: float = Field(0.65, ge=0.1, le=1.0, description="目标CPU利用率")
    current_instances: int = Field(1, description="当前运行实例数")
    status: ServiceStatus = Field(ServiceStatus.HEALTHY, description="服务状态")
    created_at: datetime = Field(default_factory=datetime.utcnow)
    updated_at: datetime = Field(default_factory=datetime.utcnow)

    class Config:
        use_enum_values = True

    @validator('max_instances')
    def max_instances_greater_than_min(cls, v, values):
        if 'min_instances' in values and v < values['min_instances']:
            raise ValueError('max_instances must be >= min_instances')
        return v

class Instance(BaseModel):
    """服务实例抽象"""
    id: str = Field(..., description="实例唯一标识")
    service_id: str = Field(..., description="所属服务ID")
    host: str = Field(..., description="运行主机")
    port: int = Field(..., description="服务端口")
    metadata: Dict[str, Any] = Field(default_factory=dict, description="实例元数据")
    created_at: datetime = Field(default_factory=datetime.utcnow)
    is_active: bool = Field(True, description="实例是否活跃")

class Metric(BaseModel):
    """性能指标抽象"""
    id: Optional[str] = Field(None, description="指标记录ID（由存储生成）")
    instance_id: str = Field(..., description="对应的实例ID")
    service_id: str = Field(..., description="对应的服务ID")
    cpu_utilization: float = Field(..., ge=0.0, le=1.0, description="CPU使用率")
    memory_utilization: float = Field(..., ge=0.0, le=1.0, description="内存使用率")
    requests_per_second: float = Field(0.0, ge=0.0, description="每秒请求数")
    timestamp: datetime = Field(default_factory=datetime.utcnow, description="指标采集时间")

class Plan(BaseModel):
    """容量规划建议"""
    id: str = Field(..., description="规划唯一标识")
    service_id: str = Field(..., description="目标服务ID")
    action: PlanAction = Field(..., description="规划动作")
    reason: str = Field(..., description="规划原因")
    details: Dict[str, Any] = Field(default_factory=dict, description="规划详情，如目标实例数")
    generated_at: datetime = Field(default_factory=datetime.utcnow, description="规划生成时间")
    is_executed: bool = Field(False, description="是否已执行")

"""
数据存储抽象层与内存实现。
"""
from abc import ABC, abstractmethod
from typing import List, Optional, Dict, Any
from uuid import uuid4
from datetime import datetime, timedelta
from .models import Service, Instance, Metric, Plan

class Storage(ABC):
    """存储抽象基类"""

    @abstractmethod
    def add_service(self, service: Service) -> Service:
        pass

    @abstractmethod
    def get_service(self, service_id: str) -> Optional[Service]:
        pass

    @abstractmethod
    def update_service(self, service: Service) -> Service:
        pass

    @abstractmethod
    def list_services(self) -> List[Service]:
        pass

    @abstractmethod
    def add_instance(self, instance: Instance) -> Instance:
        pass

    @abstractmethod
    def get_instances_by_service(self, service_id: str, active_only: bool = True) -> List[Instance]:
        pass

    @abstractmethod
    def add_metric(self, metric: Metric) -> Metric:
        pass

    @abstractmethod
    def get_metrics(self, service_id: str, start_time: datetime, end_time: datetime) -> List[Metric]:
        pass

    @abstractmethod
    def add_plan(self, plan: Plan) -> Plan:
        pass

    @abstractmethod
    def get_pending_plans(self, service_id: Optional[str] = None) -> List[Plan]:
        pass

    @abstractmethod
    def mark_plan_executed(self, plan_id: str):
        pass


class InMemoryStorage(Storage):
    """内存存储实现（用于演示）"""

    def __init__(self):
        self._services: Dict[str, Service] = {}
        self._instances: Dict[str, Instance] = {}
        self._metrics: List[Metric] = []
        self._plans: Dict[str, Plan] = {}

    def add_service(self, service: Service) -> Service:
        service.id = service.id or f"svc-{uuid4().hex[:8]}"
        service.updated_at = datetime.utcnow()
        self._services[service.id] = service
        return service

    def get_service(self, service_id: str) -> Optional[Service]:
        return self._services.get(service_id)

    def update_service(self, service: Service) -> Service:
        if service.id not in self._services:
            raise KeyError(f"Service {service.id} not found")
        service.updated_at = datetime.utcnow()
        self._services[service.id] = service
        return service

    def list_services(self) -> List[Service]:
        return list(self._services.values())

    def add_instance(self, instance: Instance) -> Instance:
        instance.id = instance.id or f"ins-{uuid4().hex[:8]}"
        self._instances[instance.id] = instance
        return instance

    def get_instances_by_service(self, service_id: str, active_only: bool = True) -> List[Instance]:
        instances = [i for i in self._instances.values() if i.service_id == service_id]
        if active_only:
            instances = [i for i in instances if i.is_active]
        return instances

    def add_metric(self, metric: Metric) -> Metric:
        metric.id = f"met-{uuid4().hex[:8]}"
        self._metrics.append(metric)
        return metric

    def get_metrics(self, service_id: str, start_time: datetime, end_time: datetime) -> List[Metric]:
        # 简单的时间过滤，实际项目需考虑索引
        filtered = [
            m for m in self._metrics
            if m.service_id == service_id and start_time <= m.timestamp <= end_time
        ]
        return sorted(filtered, key=lambda x: x.timestamp)

    def add_plan(self, plan: Plan) -> Plan:
        plan.id = plan.id or f"plan-{uuid4().hex[:8]}"
        self._plans[plan.id] = plan
        return plan

    def get_pending_plans(self, service_id: Optional[str] = None) -> List[Plan]:
        plans = [p for p in self._plans.values() if not p.is_executed]
        if service_id:
            plans = [p for p in plans if p.service_id == service_id]
        return sorted(plans, key=lambda x: x.generated_at)

    def mark_plan_executed(self, plan_id: str):
        if plan_id in self._plans:
            self._plans[plan_id].is_executed = True

"""
容量规划核心算法。
"""
from datetime import datetime, timedelta
from typing import List, Tuple, Optional
from .models import Service, Metric, Plan, PlanAction
from .storage import Storage

class CapacityPlanner:
    """容量规划器"""

    def __init__(self, storage: Storage):
        self.storage = storage

    def analyze_and_plan(self, service_id: str) -> Optional[Plan]:
        """
        分析指定服务的指标并生成规划建议。
        核心算法步骤：

        1. 获取服务配置与当前实例信息。
        2. 获取最近一段时间的历史指标。
        3. 预测下一时间段的负载（如CPU使用率）。
        4. 将预测负载与目标阈值、实例数进行比较。
        5. 生成扩容、缩容或无操作建议。
        """
        service = self.storage.get_service(service_id)
        if not service:
            return None

        # 1. 获取当前活跃实例
        current_instances = self.storage.get_instances_by_service(service_id, active_only=True)
        current_instance_count = len(current_instances)
        if current_instance_count == 0:
            # 没有活跃实例，可能是新服务或全部宕机，建议扩容到最小实例数
            if service.min_instances > 0:
                return self._create_plan(
                    service_id,
                    PlanAction.SCALE_OUT,
                    reason="No active instances found.",
                    target_instances=service.min_instances
                )
            return None

        # 2. 获取历史指标（例如过去30分钟）
        end_time = datetime.utcnow()
        start_time = end_time - timedelta(minutes=30)
        historical_metrics = self.storage.get_metrics(service_id, start_time, end_time)

        if not historical_metrics:
            # 无历史数据，无法分析，建议保持现状
            return None

        # 3. 简单预测：使用最近N个指标的平均值作为预测值 (N=5)
        recent_metrics = historical_metrics[-5:] if len(historical_metrics) >= 5 else historical_metrics
        avg_cpu = sum(m.cpu_utilization for m in recent_metrics) / len(recent_metrics)
        avg_rps = sum(m.requests_per_second for m in recent_metrics) / len(recent_metrics)

        # 4. 基于预测的CPU利用率进行决策
        projected_cpu_per_instance = avg_cpu  # 假设负载均匀分布

        # 计算满足目标利用率所需的理想实例数
        # 所需实例数 = ceil(预测总负载 / (单个实例容量 * 目标利用率))
        # 简化：假设单个实例容量为1 (100% CPU)，总负载为 `avg_cpu * current_instance_count`
        total_projected_load = projected_cpu_per_instance * current_instance_count
        desired_instance_count = self._calculate_desired_instances(
            total_projected_load, service.target_cpu_utilization
        )

        # 约束在最小/最大实例数之间
        desired_instance_count = max(service.min_instances, min(service.max_instances, desired_instance_count))

        # 5. 生成规划
        if desired_instance_count > current_instance_count:
            return self._create_plan(
                service_id,
                PlanAction.SCALE_OUT,
                reason=f"Projected CPU load ({projected_cpu_per_instance:.2%}) per instance exceeds "
                       f"target ({service.target_cpu_utilization:.0%}). "
                       f"Avg RPS: {avg_rps:.2f}",
                target_instances=desired_instance_count
            )
        elif desired_instance_count < current_instance_count and current_instance_count > service.min_instances:
            # 只有当当前实例数大于最小值时才考虑缩容
            return self._create_plan(
                service_id,
                PlanAction.SCALE_IN,
                reason=f"Projected CPU load ({projected_cpu_per_instance:.2%}) per instance is below "
                       f"target ({service.target_cpu_utilization:.0%}). Opportunity to save resources. "
                       f"Avg RPS: {avg_rps:.2f}",
                target_instances=desired_instance_count
            )
        else:
            # 无需操作
            return self._create_plan(
                service_id,
                PlanAction.DO_NOTHING,
                reason=f"Current configuration is optimal. "
                       f"Projected CPU: {projected_cpu_per_instance:.2%}, "
                       f"Target: {service.target_cpu_utilization:.0%}, "
                       f"Instances: {current_instance_count}",
                target_instances=current_instance_count
            )

    def _calculate_desired_instances(self, total_load: float, target_utilization: float) -> int:
        """计算达到目标利用率所需的实例数（向上取整）"""
        if target_utilization <= 0:
            return 1
        import math
        return math.ceil(total_load / target_utilization)

    def _create_plan(self,
                     service_id: str,
                     action: PlanAction,
                     reason: str,
                     target_instances: int) -> Plan:
        """创建规划对象"""
        from uuid import uuid4
        return Plan(
            id=f"plan-{uuid4().hex[:8]}",
            service_id=service_id,
            action=action,
            reason=reason,
            details={
                "target_instance_count": target_instances,
                "generated_by": "simple_capacity_planner_v1"
            }
        )

"""
应用配置。
"""
import os
from pydantic import BaseSettings

class Settings(BaseSettings):
    """应用设置"""
    app_name: str = "Capacity Planning Service"
    app_version: str = "1.0.0"
    debug: bool = False

    # API配置
    api_prefix: str = "/api/v1"
    host: str = "0.0.0.0"
    port: int = 8000

    # 规划器配置
    planner_interval_seconds: int = 60  # 自动规划任务执行间隔

    class Config:
        env_file = ".env"

settings = Settings()

"""
FastAPI 依赖项。
"""
from core.storage import InMemoryStorage
from core.algorithm import CapacityPlanner

# 创建全局单例（简单起见，实际生产环境可能需要更复杂的管理）
_storage = InMemoryStorage()
_planner = CapacityPlanner(_storage)

def get_storage():
    """获取存储实例依赖"""
    return _storage

def get_planner():
    """获取容量规划器实例依赖"""
    return _planner

"""
RESTful API 端点定义。
"""
from datetime import datetime, timedelta
from typing import List
from fastapi import APIRouter, Depends, HTTPException, BackgroundTasks
from core.models import Service, Instance, Metric, Plan
from core.algorithm import CapacityPlanner
from .dependencies import get_storage, get_planner
from core.storage import Storage

router = APIRouter()

# ---------- Service 相关接口 ----------
@router.post("/services", response_model=Service, tags=["services"])
async def create_service(service: Service, storage: Storage = Depends(get_storage)):
    """注册一个新的微服务"""
    existing = storage.get_service(service.id)
    if existing:
        raise HTTPException(status_code=409, detail=f"Service with ID {service.id} already exists.")
    return storage.add_service(service)

@router.get("/services", response_model=List[Service], tags=["services"])
async def list_services(storage: Storage = Depends(get_storage)):
    """获取所有已注册服务"""
    return storage.list_services()

@router.get("/services/{service_id}", response_model=Service, tags=["services"])
async def get_service(service_id: str, storage: Storage = Depends(get_storage)):
    """获取指定服务详情"""
    service = storage.get_service(service_id)
    if not service:
        raise HTTPException(status_code=404, detail=f"Service {service_id} not found.")
    return service

# ---------- Instance 相关接口 ----------
@router.post("/instances", response_model=Instance, tags=["instances"])
async def register_instance(instance: Instance, storage: Storage = Depends(get_storage)):
    """注册一个服务实例（通常由服务启动时调用）"""
    # 检查对应服务是否存在
    service = storage.get_service(instance.service_id)
    if not service:
        raise HTTPException(status_code=404, detail=f"Service {instance.service_id} not found.")
    return storage.add_instance(instance)

@router.get("/services/{service_id}/instances", response_model=List[Instance], tags=["instances"])
async def list_service_instances(service_id: str,
                                 active_only: bool = True,
                                 storage: Storage = Depends(get_storage)):
    """获取指定服务的所有实例"""
    return storage.get_instances_by_service(service_id, active_only=active_only)

# ---------- Metric 相关接口 ----------
@router.post("/metrics", response_model=Metric, tags=["metrics"])
async def submit_metric(metric: Metric, storage: Storage = Depends(get_storage)):
    """提交性能指标数据（通常由监控Agent周期性推送）"""
    # 可选：验证对应的实例是否存在
    return storage.add_metric(metric)

@router.get("/services/{service_id}/metrics", response_model=List[Metric], tags=["metrics"])
async def get_service_metrics(service_id: str,
                              minutes: int = 30,
                              storage: Storage = Depends(get_storage)):
    """获取指定服务的历史指标"""
    end_time = datetime.utcnow()
    start_time = end_time - timedelta(minutes=minutes)
    return storage.get_metrics(service_id, start_time, end_time)

# ---------- Plan 相关接口 ----------
@router.post("/services/{service_id}/plan", response_model=Plan, tags=["plan"])
async def trigger_planning(service_id: str,
                           background_tasks: BackgroundTasks,
                           planner: CapacityPlanner = Depends(get_planner)):
    """手动触发对指定服务的容量规划分析"""
    plan = planner.analyze_and_plan(service_id)
    if not plan:
        raise HTTPException(status_code=404, detail=f"No actionable plan generated for service {service_id}.")
    # 注意：这里直接返回规划，实际应由后台任务保存到存储
    return plan

@router.get("/plans/pending", response_model=List[Plan], tags=["plan"])
async def get_pending_plans(service_id: str = None, storage: Storage = Depends(get_storage)):
    """获取所有（或指定服务的）待执行规划"""
    return storage.get_pending_plans(service_id)

@router.post("/plans/{plan_id}/execute", response_model=Plan, tags=["plan"])
async def execute_plan(plan_id: str, storage: Storage = Depends(get_storage)):
    """标记一个规划为已执行（通常由外部编排系统调用）"""
    # 简化：只标记为已执行。实际场景会触发真正的扩缩容操作并更新服务实例数。
    storage.mark_plan_executed(plan_id)
    # 返回更新后的规划（需要从存储重新获取，这里简化）
    # 实际应实现一个 get_plan_by_id 方法
    all_plans = storage.get_pending_plans() + [p for p in storage._plans.values() if p.is_executed]
    for p in all_plans:
        if p.id == plan_id:
            return p
    raise HTTPException(status_code=404, detail=f"Plan {plan_id} not found.")

"""
容量规划系统主应用入口。
"""
import logging
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from core.config import settings
from api.endpoints import router as api_router
from core.storage import InMemoryStorage
from core.algorithm import CapacityPlanner

# 配置日志
logging.basicConfig(level=logging.INFO if not settings.debug else logging.DEBUG)
logger = logging.getLogger(__name__)

# 全局存储与规划器实例
storage = None
planner = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    """
    应用生命周期管理。
    启动时：初始化全局组件。
    关闭时：清理资源（本示例无）。
    """
    global storage, planner
    # 启动
    logger.info(f"Starting {settings.app_name} v{settings.app_version}")
    storage = InMemoryStorage()
    planner = CapacityPlanner(storage)
    logger.info("Global storage and planner initialized.")

    # 创建一些示例数据用于演示
    from core.models import Service, Instance, Metric
    from datetime import datetime, timedelta
    import random
    try:
        demo_svc = Service(id="demo-frontend", name="Frontend Web Service", min_instances=2, max_instances=10)
        storage.add_service(demo_svc)
        for i in range(3):
            ins = Instance(id=f"demo-ins-{i}", service_id="demo-frontend", host=f"host-{i}.example.com", port=8080)
            storage.add_instance(ins)
        # 生成一些过去30分钟的模拟指标
        now = datetime.utcnow()
        for i in range(30):
            for ins in storage.get_instances_by_service("demo-frontend"):
                m_time = now - timedelta(minutes=(30 - i))
                metric = Metric(
                    instance_id=ins.id,
                    service_id="demo-frontend",
                    cpu_utilization=random.uniform(0.3, 0.8),  # 模拟30%-80%的CPU
                    memory_utilization=random.uniform(0.4, 0.6),
                    requests_per_second=random.uniform(50, 200),
                    timestamp=m_time
                )
                storage.add_metric(metric)
        logger.info("Demo data created for service 'demo-frontend'.")
    except Exception as e:
        logger.warning(f"Could not create demo data: {e}")

    yield
    # 关闭
    logger.info("Shutting down capacity planning service.")

# 创建FastAPI应用实例
app = FastAPI(
    title=settings.app_name,
    version=settings.app_version,
    debug=settings.debug,
    lifespan=lifespan,
    openapi_url=f"{settings.api_prefix}/openapi.json",
    docs_url="/docs",
    redoc_url="/redoc"
)

# 添加CORS中间件（便于前端调试）
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # 生产环境应限制
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 挂载API路由
app.include_router(api_router, prefix=settings.api_prefix)

@app.get("/", tags=["root"])
async def root():
    """根端点，返回服务信息"""
    return {
        "service": settings.app_name,
        "version": settings.app_version,
        "docs": "/docs",
        "api_base": settings.api_prefix
    }

@app.get("/health", tags=["health"])
async def health_check():
    """健康检查端点"""
    return {"status": "healthy"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(
        "app:app",
        host=settings.host,
        port=settings.port,
        reload=settings.debug,
        log_level="info"
    )

fastapi==0.104.1
uvicorn[standard]==0.24.0
pydantic==2.5.0
pydantic-settings==2.1.0

# 应用配置
DEBUG=False
APP_NAME="Capacity Planner Demo"
APP_VERSION="1.0.0"

# 服务器配置
HOST="0.0.0.0"
PORT=8000
API_PREFIX="/api/v1"

# 规划器配置
PLANNER_INTERVAL_SECONDS=60

# 克隆或创建项目目录
mkdir capacity-planner-demo && cd capacity-planner-demo

# 创建并激活虚拟环境（可选但推荐）
python -m venv venv
# Linux/Mac:
source venv/bin/activate
# Windows:
# venv\Scripts\activate

# 安装依赖包
pip install -r requirements.txt

python app.py

INFO:     Started server process [xxxx]
INFO:     Waiting for application startup.
INFO:     Global storage and planner initialized.
INFO:     Demo data created for service 'demo-frontend'.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)

curl -X GET "http://localhost:8000/api/v1/services" -H "accept: application/json"

curl -X GET "http://localhost:8000/api/v1/services/demo-frontend/instances" -H "accept: application/json"

curl -X POST "http://localhost:8000/api/v1/services/demo-frontend/plan" -H "accept: application/json"

curl -X GET "http://localhost:8000/api/v1/plans/pending" -H "accept: application/json"

curl -X POST "http://localhost:8000/api/v1/metrics" \
  -H "Content-Type: application/json" \
  -d '{
    "instance_id": "demo-ins-0",
    "service_id": "demo-frontend",
    "cpu_utilization": 0.95,
    "memory_utilization": 0.5,
    "requests_per_second": 500
  }'