云成本治理实战：AI 弹性伸缩与闲置资源回收，月费直降 38%

#!/usr/bin/env python3
"""
闲置资源检测脚本
基于阿里云 OpenAPI + Prometheus 指标
"""
import requests
from datetime import datetime, timedelta
from alibabacloud_ecs20140526.client import Client as ECSClient
from alibabacloud_ecs20140526 import models as ecs_models

PROMETHEUS_URL = "http://localhost:9090"

# 闲置判定阈值
IDLE_THRESHOLD = {
    "cpu_avg_percent": 5,       # CPU 7天均值 < 5%
    "cpu_peak_percent": 15,     # CPU 7天峰值 < 15%
    "network_kb_per_sec": 10,   # 网络流量 < 10KB/s
    "duration_days": 7          # 持续天数
}

def get_ecs_instances() -> list:
    """获取所有运行中的 ECS 实例"""
    client = ECSClient(...)
    request = ecs_models.DescribeInstancesRequest(
        region_id="cn-hangzhou",
        status="Running"
    )
    response = client.describe_instances(request)
    return response.body.instances.instance

def check_instance_utilization(instance_id: str) -> dict:
    """检查实例利用率"""
    end = datetime.now()
    start = end - timedelta(days=7)

    cpu_avg_query = f'avg_over_time(acs_ecs_cpu_total{{instanceId="{instance_id}"}}[7d])'
    cpu_peak_query = f'max_over_time(acs_ecs_cpu_total{{instanceId="{instance_id}"}}[7d])'
    net_query = f'avg_over_time(acs_ecs_network_rate{{instanceId="{instance_id}"}}[7d])'

    cpu_avg = query_prometheus(cpu_avg_query)
    cpu_peak = query_prometheus(cpu_peak_query)
    net_avg = query_prometheus(net_query)

    is_idle = (
        cpu_avg < IDLE_THRESHOLD["cpu_avg_percent"]
        and cpu_peak < IDLE_THRESHOLD["cpu_peak_percent"]
        and net_avg < IDLE_THRESHOLD["network_kb_per_sec"]
    )

    return {
        "instance_id": instance_id,
        "cpu_avg": round(cpu_avg, 2),
        "cpu_peak": round(cpu_peak, 2),
        "network_avg": round(net_avg, 2),
        "is_idle": is_idle
    }

def query_prometheus(query: str) -> float:
    """查询 Prometheus 指标"""
    resp = requests.get(
        f"{PROMETHEUS_URL}/api/v1/query",
        params={"query": query}
    )
    results = resp.json()["data"]["result"]
    ifnot results:
        return0.0
    return float(results[0]["value"][1])

def scan_idle_resources() -> list:
    """扫描所有闲置资源"""
    instances = get_ecs_instances()
    idle_list = []

    for inst in instances:
        util = check_instance_utilization(inst.instance_id)
        if util["is_idle"]:
            idle_list.append({
                **util,
                "instance_name": inst.instance_name,
                "instance_type": inst.instance_type,
                "monthly_cost": estimate_cost(inst.instance_type)
            })

    idle_list.sort(key=lambda x: x["monthly_cost"], reverse=True)
    return idle_list

def estimate_cost(instance_type: str) -> float:
    """估算实例月度费用"""
    PRICE_MAP = {
        "ecs.c6.large": 280,
        "ecs.c6.xlarge": 560,
        "ecs.g6.large": 320,
        "ecs.g6.xlarge": 640,
        "ecs.r6.large": 380,
    }
    return PRICE_MAP.get(instance_type, 200)

=== 闲置资源检测报告 ===
扫描时间: 2026-06-28 10:00:00
扫描范围: cn-hangzhou 运行中实例

闲置实例列表:
1. dev-test-03 (ecs.c6.xlarge)  CPU均值:2.1%  月费:560元
2. staging-api-02 (ecs.g6.large) CPU均值:3.8%  月费:320元
3. legacy-worker (ecs.r6.large)  CPU均值:1.5%  月费:380元

潜在节省: 1,260 元/月

# AI 优化前（默认配置）
apiVersion:autoscaling/v2
kind:HorizontalPodAutoscaler
metadata:
name:order-service-hpa
spec:
minReplicas:3
maxReplicas:10
metrics:
    -type:Resource
      resource:
        name:cpu
        target:
          type:Utilization
          averageUtilization:70# 粗暴固定值

---
# AI 优化后（根据历史流量模式调整）
apiVersion:autoscaling/v2
kind:HorizontalPodAutoscaler
metadata:
name:order-service-hpa
annotations:
    optimized-by:ai-cost-engine
    optimized-at:"2026-06-28"
spec:
minReplicas:2       # 低谷期 2 副本足够
maxReplicas:15      # 高峰期允许更多扩容
behavior:
    scaleDown:
      stabilizationWindowSeconds:300# 缩容稳定窗口 5 分钟
      policies:
        -type:Percent
          value:25        # 每次缩容不超过 25%
          periodSeconds:60
    scaleUp:
      stabilizationWindowSeconds:0    # 扩容立即响应
      policies:
        -type:Percent
          value:100       # 紧急时翻倍扩容
          periodSeconds:30
metrics:
    -type:Resource
      resource:
        name:cpu
        target:
          type:Utilization
          averageUtilization:55# 提前扩容，避免过载
    -type:Pods
      pods:
        metric:
          name:http_requests_per_second
        target:
          type:AverageValue
          averageValue:"500"# 每副本 500 QPS

behavior:
  scaleDown:
    stabilizationWindowSeconds: 300
    policies:
      - type: Percent
        value: 25
        periodSeconds: 60

def recommend_reserved(current_count: int, growth_rate: float = 0.3) -> int:
    """推荐预留实例数量"""
    reserved = int(current_count * 0.7)
    predicted = int(current_count * (1 + growth_rate))
    return min(reserved, predicted)