Kubernetes 成本优化实战：从资源治理到 FinOps 体系

# 典型的过度配置
apiVersion: apps/v1
kind: Deployment
metadata:
  name: api-service
spec:
  template:
    spec:
      containers:
      - name: api
        resources:
          requests:
            cpu: "2000m"    # 实际使用 200m，浪费 90%
            memory: "4Gi"   # 实际使用 512Mi，浪费 87%
          limits:
            cpu: "4000m"
            memory: "8Gi"

# 没有 ResourceQuota 的命名空间 = 无限制的资源消耗
# 这导致一个团队的服务可能挤占其他团队的资源

#!/usr/bin/env python3
"""
Pod QoS 等级判定与资源浪费分析工具
扫描集群中所有 Pod，识别配置问题和成本浪费
"""

import json
from dataclasses import dataclass, asdict
from typing import List, Optional

@dataclass
class PodResourceInfo:
    """Pod 资源配置信息"""
    name: str
    namespace: str
    qos_class: str
    cpu_request_m: float        # millicores
    cpu_limit_m: float
    memory_request_mi: float    # MiB
    memory_limit_mi: float
    cpu_usage_m: float          # 实际使用（来自 metrics-server）
    memory_usage_mi: float

    @property
    def cpu_waste_m(self):
        """CPU 资源浪费量 = Request - 实际使用"""
        return max(0, self.cpu_request_m - self.cpu_usage_m)

    @property
    def memory_waste_mi(self):
        """内存资源浪费量"""
        return max(0, self.memory_request_mi - self.memory_usage_mi)

    @property
    def cpu_utilization(self):
        """Request 利用率"""
        if self.cpu_request_m == 0:
            return 0
        return self.cpu_usage_m / self.cpu_request_m

    @property
    def memory_utilization(self):
        if self.memory_request_mi == 0:
            return 0
        return self.memory_usage_mi / self.memory_request_mi

class ResourceWasteAnalyzer:
    """集群资源浪费分析器"""

    # 优化阈值
    LOW_UTILIZATION_THRESHOLD = 0.30    # 利用率低于 30% 视为浪费
    HIGH_UTILIZATION_THRESHOLD = 0.85    # 利用率高于 85% 视为风险
    OVERREQUEST_MULTIPLIER = 3.0        # Request 超过实际使用 3 倍视为过度申请

    def __init__(self):
        self.pods: List[PodResourceInfo] = []

    def add_pod(self, pod: PodResourceInfo):
        self.pods.append(pod)

    def analyze(self) -> dict:
        """分析集群资源浪费情况"""
        results = {
            'total_pods': len(self.pods),
            'qos_distribution': self._qos_distribution(),
            'waste_summary': self._waste_summary(),
            'recommendations': self._recommendations()
        }
        return results

    def _qos_distribution(self):
        """QoS 分布统计"""
        dist = {'Guaranteed': 0, 'Burstable': 0, 'BestEffort': 0}
        for pod in self.pods:
            if pod.qos_class in dist:
                dist[pod.qos_class] += 1
        return dist

    def _waste_summary(self):
        """资源浪费汇总"""
        total_cpu_request = sum(p.cpu_request_m for p in self.pods)
        total_cpu_usage = sum(p.cpu_usage_m for p in self.pods)
        total_mem_request = sum(p.memory_request_mi for p in self.pods)
        total_mem_usage = sum(p.memory_usage_mi for p in self.pods)

        cpu_waste = total_cpu_request - total_cpu_usage
        mem_waste = total_mem_request - total_mem_usage

        return {
            'cpu': {
                'total_request_m': round(total_cpu_request, 1),
                'total_usage_m': round(total_cpu_usage, 1),
                'waste_m': round(cpu_waste, 1),
                'waste_pct': round(
                    cpu_waste / total_cpu_request * 100, 1
                ) if total_cpu_request > 0 else 0
            },
            'memory': {
                'total_request_mi': round(total_mem_request, 1),
                'total_usage_mi': round(total_mem_usage, 1),
                'waste_mi': round(mem_waste, 1),
                'waste_pct': round(
                    mem_waste / total_mem_request * 100, 1
                ) if total_mem_request > 0 else 0
            }
        }

    def _recommendations(self):
        """生成优化建议"""
        recs = []

        for pod in self.pods:
            # 低利用率检测
            if (pod.cpu_utilization < self.LOW_UTILIZATION_THRESHOLD and
                pod.cpu_request_m > 100):
                suggested_request = max(
                    pod.cpu_usage_m * 1.5,  # 50% buffer
                    50  # 最低 50m
                )
                recs.append({
                    'pod': pod.name,
                    'namespace': pod.namespace,
                    'issue': 'low_cpu_utilization',
                    'current_request_m': pod.cpu_request_m,
                    'actual_usage_m': round(pod.cpu_usage_m, 1),
                    'utilization': f"{pod.cpu_utilization:.0%}",
                    'suggested_request_m': round(suggested_request, 0),
                    'potential_save_m': round(
                        pod.cpu_request_m - suggested_request, 0
                    ),
                    'severity': 'medium'
                })

            # 内存过度申请
            if (pod.memory_utilization < self.LOW_UTILIZATION_THRESHOLD and
                pod.memory_request_mi > 256):
                suggested_memory = max(
                    pod.memory_usage_mi * 1.5,
                    128
                )
                recs.append({
                    'pod': pod.name,
                    'namespace': pod.namespace,
                    'issue': 'low_memory_utilization',
                    'current_request_mi': pod.memory_request_mi,
                    'actual_usage_mi': round(pod.memory_usage_mi, 1),
                    'utilization': f"{pod.memory_utilization:.0%}",
                    'suggested_request_mi': round(suggested_memory, 0),
                    'potential_save_mi': round(
                        pod.memory_request_mi - suggested_memory, 0
                    ),
                    'severity': 'medium'
                })

            # BestEffort Pod 告警
            if pod.qos_class == 'BestEffort':
                recs.append({
                    'pod': pod.name,
                    'namespace': pod.namespace,
                    'issue': 'besteffort_no_resources',
                    'recommendation': 'Set requests and limits to ensure proper scheduling',
                    'severity': 'high'
                })

            # 高利用率风险（可能被 Throttle 或 OOM）
            if pod.cpu_utilization > self.HIGH_UTILIZATION_THRESHOLD:
                recs.append({
                    'pod': pod.name,
                    'namespace': pod.namespace,
                    'issue': 'cpu_near_limit',
                    'utilization': f"{pod.cpu_utilization:.0%}",
                    'recommendation': 'Investigate if CPU throttling is occurring',
                    'severity': 'high'
                })

        recs.sort(key=lambda x: {
            'high': 0, 'medium': 1, 'low': 2
        }.get(x.get('severity', 'low'), 2))

        return recs

# 使用示例
if __name__ == '__main__':
    analyzer = ResourceWasteAnalyzer()

    # 模拟 Pod 数据
    pods_data = [
        PodResourceInfo(
            name='api-service-7f9b-x2k4', namespace='production',
            qos_class='Guaranteed',
            cpu_request_m=2000, cpu_limit_m=2000,
            memory_request_mi=4096, memory_limit_mi=4096,
            cpu_usage_m=180, memory_usage_mi=512
        ),
        PodResourceInfo(
            name='worker-bg-6c8d-m3n1', namespace='production',
            qos_class='Burstable',
            cpu_request_m=500, cpu_limit_m=1000,
            memory_request_mi=512, memory_limit_mi=1024,
            cpu_usage_m=420, memory_usage_mi=480
        ),
        PodResourceInfo(
            name='debug-pod-xyz', namespace='dev',
            qos_class='BestEffort',
            cpu_request_m=0, cpu_limit_m=0,
            memory_request_mi=0, memory_limit_mi=0,
            cpu_usage_m=50, memory_usage_mi=128
        ),
    ]

    for pod in pods_data:
        analyzer.add_pod(pod)

    report = analyzer.analyze()
    print(json.dumps(report, indent=2, ensure_ascii=False))

# Right-Sizing 配置模板
apiVersion: apps/v1
kind: Deployment
metadata:
  name: api-service
  namespace: production
spec:
  template:
    spec:
      containers:
      - name: api
        # 核心服务：Guaranteed QoS，Requests == Limits
        resources:
          requests:
            cpu: "250m"      # 基于 P95 实际使用 * 1.5
            memory: "512Mi"  # 基于 P95 实际使用 * 1.3
          limits:
            cpu: "250m"      # 与 requests 一致，避免 throttle
            memory: "512Mi"  # 与 requests 一常，Guaranteed QoS
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: background-worker
  namespace: production
spec:
  template:
    spec:
      containers:
      - name: worker
        # 非核心服务：Burstable QoS，允许突发
        resources:
          requests:
            cpu: "100m"     # 低保底
            memory: "256Mi"
          limits:
            cpu: "500m"     # 允许突发到 5 倍
            memory: "1Gi"

#!/usr/bin/env python3
"""
基于 Prometheus 历史指标的 Right-Sizing 推荐
分析过去 7 天的资源使用数据，给出 Requests/Limits 建议
"""

import json
from datetime import datetime, timedelta

class RightSizingRecommender:
    """基于历史指标给出资源配置建议"""

    # 建议系数
    CPU_REQUEST_MULTIPLIER = 1.5    # P95 使用 * 1.5
    CPU_LIMIT_MULTIPLIER = 2.0     # P95 使用 * 2.0
    MEM_REQUEST_MULTIPLIER = 1.3   # P95 使用 * 1.3
    MEM_LIMIT_MULTIPLIER = 1.5     # P95 使用 * 1.5

    # 最低值
    MIN_CPU_REQUEST_M = 50         # 50m
    MIN_MEM_REQUEST_MI = 128       # 128Mi

    def __init__(self):
        self.metrics = []

    def add_metric(self, timestamp, cpu_m, memory_mi):
        """添加一条指标数据"""
        self.metrics.append({
            'timestamp': timestamp,
            'cpu_m': cpu_m,
            'memory_mi': memory_mi
        })

    def recommend(self, service_name, qos='burstable'):
        """
        生成资源配置建议

        Args:
            service_name: 服务名
            qos: 目标 QoS 等级 ('guaranteed' 或 'burstable')
        """
        if not self.metrics:
            return {'error': 'No metrics data'}

        cpu_values = sorted([m['cpu_m'] for m in self.metrics])
        mem_values = sorted([m['memory_mi'] for m in self.metrics])

        # 计算 P50, P95, P99
        stats = {
            'p50_cpu': self._percentile(cpu_values, 50),
            'p95_cpu': self._percentile(cpu_values, 95),
            'p99_cpu': self._percentile(cpu_values, 99),
            'p50_mem': self._percentile(mem_values, 50),
            'p95_mem': self._percentile(mem_values, 95),
            'p99_mem': self._percentile(mem_values, 99),
            'max_cpu': max(cpu_values),
            'max_mem': max(mem_values),
        }

        # 生成建议
        p95_cpu = stats['p95_cpu']
        p95_mem = stats['p95_mem']

        cpu_request = max(
            p95_cpu * self.CPU_REQUEST_MULTIPLIER,
            self.MIN_CPU_REQUEST_M
        )
        mem_request = max(
            p95_mem * self.MEM_REQUEST_MULTIPLIER,
            self.MIN_MEM_REQUEST_MI
        )

        if qos == 'guaranteed':
            # Guaranteed: requests == limits
            cpu_limit = cpu_request
            mem_limit = mem_request
        else:
            # Burstable: limits > requests
            cpu_limit = max(
                p95_cpu * self.CPU_LIMIT_MULTIPLIER,
                cpu_request
            )
            mem_limit = max(
                p95_mem * self.MEM_LIMIT_MULTIPLIER,
                mem_request
            )

        return {
            'service': service_name,
            'qos_target': qos,
            'statistics': {
                'cpu_p50_m': round(stats['p50_cpu'], 1),
                'cpu_p95_m': round(stats['p95_cpu'], 1),
                'cpu_p99_m': round(stats['p99_cpu'], 1),
                'cpu_max_m': round(stats['max_cpu'], 1),
                'mem_p50_mi': round(stats['p50_mem'], 1),
                'mem_p95_mi': round(stats['p95_mem'], 1),
                'mem_p99_mi': round(stats['p99_mem'], 1),
                'mem_max_mi': round(stats['max_mem'], 1),
            },
            'recommendation': {
                'cpu_request': f"{round(cpu_request)}m",
                'cpu_limit': f"{round(cpu_limit)}m",
                'memory_request': f"{round(mem_request)}Mi",
                'memory_limit': f"{round(mem_limit)}Mi",
            },
            'yaml_snippet': self._generate_yaml(
                cpu_request, cpu_limit, mem_request, mem_limit
            )
        }

    def _percentile(self, sorted_list, p):
        """计算百分位数"""
        if not sorted_list:
            return 0
        index = int(len(sorted_list) * p / 100)
        index = min(index, len(sorted_list) - 1)
        return sorted_list[index]

    def _generate_yaml(self, cpu_req, cpu_lim, mem_req, mem_lim):
        """生成 YAML 配置片段"""
        return f"""resources:
  requests:
    cpu: "{round(cpu_req)}m"
    memory: "{round(mem_req)}Mi"
  limits:
    cpu: "{round(cpu_lim)}m"
    memory: "{round(mem_lim)}Mi\""""

# 使用示例
if __name__ == '__main__':
    recommender = RightSizingRecommender()

    # 模拟 7 天的历史数据（每小时一个数据点）
    import random
    random.seed(42)
    base_time = datetime(2026, 7, 4)

    for i in range(168):  # 7天 * 24小时
        ts = base_time + timedelta(hours=i)
        # 模拟日间高峰、夜间低谷的 CPU 模式
        hour = ts.hour
        if 9 <= hour <= 18:  # 工作时间
            cpu = random.uniform(150, 250)
            mem = random.uniform(400, 600)
        else:
            cpu = random.uniform(30, 80)
            mem = random.uniform(200, 350)

        recommender.add_metric(ts, cpu, mem)

    result = recommender.recommend('api-service', qos='burstable')
    print(json.dumps(result, indent=2, ensure_ascii=False))

# 1. LimitRange：约束单个 Pod 的资源配置
apiVersion: v1
kind: LimitRange
metadata:
  name: production-limits
  namespace: production
spec:
  limits:
  # 默认值（未显式设置时的默认值）
  - type: Container
    default:          # default = limits
      cpu: "500m"
      memory: "512Mi"
    defaultRequest:   # defaultRequest = requests
      cpu: "100m"
      memory: "128Mi"
    # 上下限约束
    max:
      cpu: "4"
      memory: "8Gi"
    min:
      cpu: "50m"
      memory: "64Mi"
    # Limit/Limit Request 比值约束
    maxLimitRequestRatio:
      cpu: 4           # limit 最多是 request 的 4 倍
      memory: 2        # 内存不建议大比值

---
# 2. ResourceQuota：约束命名空间总资源
apiVersion: v1
kind: ResourceQuota
metadata:
  name: production-quota
  namespace: production
spec:
  hard:
    requests.cpu: "100"       # 命名空间总 CPU 上限
    requests.memory: 200Gi
    limits.cpu: "200"
    limits.memory: 400Gi
    pods: "200"                # Pod 数量上限
    services: "50"
    configmaps: "100"
    persistentvolumeclaims: "20"
    requests.storage: "500Gi"

---
# 3. 多级 Quota（按团队分配）
apiVersion: v1
kind: ResourceQuota
metadata:
  name: team-a-quota
  namespace: team-a
spec:
  hard:
    requests.cpu: "30"
    requests.memory: 60Gi
    limits.cpu: "60"
    limits.memory: 120Gi
    pods: "50"

# VPA Recommender 模式：只给出建议，不自动修改
apiVersion: autoscaling.k8s.io/v1
kind: VerticalPodAutoscaler
metadata:
  name: api-service-vpa
  namespace: production
spec:
  targetRef:
    apiVersion: "apps/v1"
    kind: Deployment
    name: api-service
  updatePolicy:
    updateMode: "Off"    # Off = 只建议不修改
                          # Initial = 只在 Pod 创建时应用
                          # Auto = 自动调整（会重启 Pod）
  resourcePolicy:
    containerPolicies:
    - containerName: api
      minAllowed:
        cpu: 50m
        memory: 128Mi
      maxAllowed:
        cpu: 2000m
        memory: 4Gi
      controlledResources: ["cpu", "memory"]

# 查看 VPA 推荐
kubectl describe vpa api-service-vpa -n production

# 输出示例：
# Recommendation:
#   Container Recommendations:
#     Target:
#       Cpu: 250m
#       Memory: 512Mi
#     Lower Bound:
#       Cpu: 100m
#       Memory: 256Mi
#     Upper Bound:
#       Cpu: 500m
#       Memory: 1Gi
#     Uncapped Target:
#       Cpu: 180m
#       Memory: 380Mi

# HPA 基于 CPU 和内存使用率
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: api-service-hpa
  namespace: production
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: api-service
  minReplicas: 3              # 最少 3 副本（保证可用性）
  maxReplicas: 30             # 最多 30 副本
  metrics:
  # CPU 利用率（相对 Requests）
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70  # 目标 CPU 利用率 70%
  # 内存利用率
  - type: Resource
    resource:
      name: memory
      target:
        type: Utilization
        averageUtilization: 80
  behavior:
    # 扩容行为：快速扩容
    scaleUp:
      stabilizationWindowSeconds: 0  # 无需稳定窗口，立即扩容
      policies:
      - type: Percent
        value: 100              # 每次最多扩容 100%
        periodSeconds: 30
      - type: Pods
        value: 6                # 或每次最多加 6 个 Pod
        periodSeconds: 30
      selectPolicy: Max         # 取两个策略中更大的
    # 缩容行为：缓慢缩容
    scaleDown:
      stabilizationWindowSeconds: 300  # 5 分钟稳定窗口
      policies:
      - type: Percent
        value: 10                # 每次最多缩容 10%
        periodSeconds: 60

# Cluster Autoscaler 配置（以 AWS EKS 为例）
# 注意：这是 AWS Auto Scaling Group 的配置策略
# Cluster Autoscaler 根据不可调度的 Pod 自动扩容节点

# 节点组配置建议
nodeGroups:
  # 按需实例节点组：保证基线容量
  - name: on-demand-base
    instanceType: m6i.large
    minSize: 3           # 最少 3 节点保证高可用
    maxSize: 10
    spot: false

  # Spot 实例节点组：承接弹性负载
  - name: spot-elastic
    instanceType:
    - m6i.large
    - m5.large
    - m5a.large
    minSize: 0           # 可以缩到 0
    maxSize: 20
    spot: true

# KEDA ScaledObject：基于 Kafka 消费延迟扩缩容
apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: kafka-consumer-scaler
  namespace: production
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: kafka-consumer
  minReplicaCount: 1       # 空闲时缩到 1
  maxReplicaCount: 20      # 高峰扩到 20
  pollingInterval: 30       # 30 秒检查一次
  cooldownPeriod: 300       # 缩容冷却 5 分钟
  triggers:
  - type: kafka
    metadata:
      bootstrapServers: kafka-broker.data.svc.cluster.local:9092
      consumerGroup: order-consumer-group
      topic: orders
      lagThreshold: "100"    # 积压超过 100 条触发扩容
      offsetResetPolicy: latest

#!/usr/bin/env python3
"""
自动扩缩容策略评估器
模拟不同场景下的扩缩容行为，评估策略效果
"""

import json
from dataclasses import dataclass, field
from typing import List

@dataclass
class TrafficPoint:
    """一个时间点的流量数据"""
    timestamp: int        # unix timestamp
    rps: float           # 每秒请求数
    cpu_per_pod_m: float  # 单 Pod CPU 使用 (millicores)

@dataclass
class ScalingDecision:
    """扩缩容决策记录"""
    timestamp: int
    current_replicas: int
    target_replicas: int
    action: str          # 'scale_up', 'scale_down', 'no_change'
    reason: str

class HPASimulator:
    """HPA 策略模拟器"""

    def __init__(self, min_replicas=3, max_replicas=30,
                 target_cpu=70, scale_up_delay=0,
                 scale_down_delay=300, cpu_request_m=250):
        self.min_replicas = min_replicas
        self.max_replicas = max_replicas
        self.target_cpu = target_cpu
        self.scale_up_delay = scale_up_delay
        self.scale_down_delay = scale_down_delay
        self.cpu_request_m = cpu_request_m

    def simulate(self, traffic: List[TrafficPoint]) -> List[ScalingDecision]:
        """模拟 HPA 行为"""
        decisions = []
        current_replicas = self.min_replicas
        last_scale_up = 0
        last_scale_down = 0

        for point in traffic:
            # 计算当前总 CPU 使用率
            total_cpu_needed = point.rps * point.cpu_per_pod_m
            current_cpu = current_replicas * self.cpu_request_m
            utilization = (total_cpu_needed / current_cpu * 100
                          if current_cpu > 0 else 100)

            action = 'no_change'
            reason = ''

            # 扩容逻辑
            if utilization > self.target_cpu:
                if point.timestamp - last_scale_up >= self.scale_up_delay:
                    needed_replicas = int(
                        total_cpu_needed / (self.cpu_request_m *
                                           self.target_cpu / 100)
                    ) + 1
                    target = min(needed_replicas, self.max_replicas)
                    if target > current_replicas:
                        current_replicas = target
                        action = 'scale_up'
                        reason = f'CPU {utilization:.0f}% > target {self.target_cpu}%'
                        last_scale_up = point.timestamp

            # 缩容逻辑
            elif utilization < self.target_cpu * 0.5:
                if point.timestamp - last_scale_down >= self.scale_down_delay:
                    needed_replicas = int(
                        total_cpu_needed / (self.cpu_request_m *
                                           self.target_cpu / 100)
                    ) + 1
                    target = max(needed_replicas, self.min_replicas)
                    if target < current_replicas:
                        current_replicas = target
                        action = 'scale_down'
                        reason = f'CPU {utilization:.0f}% < {self.target_cpu * 0.5:.0f}%'
                        last_scale_down = point.timestamp

            decisions.append(ScalingDecision(
                timestamp=point.timestamp,
                current_replicas=current_replicas,
                target_replicas=current_replicas,
                action=action,
                reason=reason
            ))

        return decisions

    def evaluate(self, traffic, decisions):
        """评估策略效果"""
        total_pod_hours = sum(d.target_replicas for d in decisions) / 60  # 假设每分钟一个点
        total_needed_pod_hours = sum(
            max(1, int(p.rps * p.cpu_per_pod_m / self.cpu_request_m))
            for p in traffic
        ) / 60

        waste_pct = ((total_pod_hours - total_needed_pod_hours) /
                     total_pod_hours * 100) if total_pod_hours > 0 else 0

        scale_events = sum(1 for d in decisions if d.action != 'no_change')

        return {
            'total_pod_hours': round(total_pod_hours, 1),
            'needed_pod_hours': round(total_needed_pod_hours, 1),
            'waste_pct': round(waste_pct, 1),
            'scale_events': scale_events,
            'avg_replicas': round(
                sum(d.target_replicas for d in decisions) / len(decisions), 1
            ),
            'max_replicas_used': max(d.target_replicas for d in decisions)
        }

# 使用示例
if __name__ == '__main__':
    import random
    random.seed(42)

    # 生成 24 小时流量数据（每分钟一个点）
    traffic = []
    for minute in range(1440):
        hour = minute / 60
        if 9 <= hour < 12 or 14 <= hour < 18:
            rps = random.uniform(80, 120)   # 高峰
        elif 0 <= hour < 6:
            rps = random.uniform(5, 15)     # 低谷
        else:
            rps = random.uniform(30, 60)     # 平峰

        traffic.append(TrafficPoint(
            timestamp=minute * 60,
            rps=rps,
            cpu_per_pod_m=2.5  # 每个 RPS 消耗 2.5m CPU
        ))

    # 模拟保守策略 vs 激进策略
    conservative = HPASimulator(
        min_replicas=3, max_replicas=30,
        target_cpu=50, scale_down_delay=600
    )
    aggressive = HPASimulator(
        min_replicas=1, max_replicas=30,
        target_cpu=75, scale_down_delay=120
    )

    cons_decisions = conservative.simulate(traffic)
    aggr_decisions = aggressive.simulate(traffic)

    print("=== 保守策略（目标50%，冷却10分钟）===")
    print(json.dumps(conservative.evaluate(traffic, cons_decisions),
                     indent=2, ensure_ascii=False))

    print("\n=== 激进策略（目标75%，冷却2分钟）===")
    print(json.dumps(aggressive.evaluate(traffic, aggr_decisions),
                     indent=2, ensure_ascii=False))

# Spot 节点池配置（AWS EKS）
apiVersion: apps/v1
kind: Deployment
metadata:
  name: batch-processor
  namespace: production
spec:
  replicas: 10
  template:
    metadata:
      annotations:
        # 标记为可调度到 Spot 节点
        sqs.amazonaws.com/queue-name: "batch-queue"
    spec:
      nodeSelector:
        kubernetes.io/arch: amd64
      tolerations:
      # 容忍 Spot 节点的 taint
      - key: "spot-instance"
        operator: "Equal"
        value: "true"
        effect: "NoPrefer"
      # 优雅终止：给任务时间完成处理
      terminationGracePeriodSeconds: 300
      containers:
      - name: processor
        image: registry.example.com/processor:v2.1
        resources:
          requests:
            cpu: "500m"
            memory: "1Gi"
          limits:
            cpu: "1000m"
            memory: "2Gi"
        # 优雅终止钩子
        lifecycle:
          preStop:
            exec:
              command:
              - /bin/sh
              - -c
              - |
                # 通知任务管理器当前任务需要重新排队
                curl -X POST http://task-manager:8080/requeue \
                  -d '{"pod": "$HOSTNAME", "action": "graceful_shutdown"}'
                sleep 30  # 等待正在处理的任务完成                

#!/usr/bin/env python3
"""
Spot 实例中断处理器
监听 AWS Spot 中断通知，优雅地排空节点
"""

import json
import logging
import subprocess
import time
from http.server import HTTPServer, BaseHTTPRequestHandler

logger = logging.getLogger(__name__)

class SpotInterruptionHandler(BaseHTTPRequestHandler):
    """处理 Spot 实例中断通知"""

    def do_PUT(self):
        if self.path == '/spot/interrupt':
            content_length = int(self.headers['Content-Length'])
            body = self.rfile.read(content_length)
            notice = json.loads(body)

            logger.warning(f"Spot interruption notice received: {notice}")

            instance_id = notice.get('instance-id')
            instance_action = notice.get('instance-action')

            if instance_action == 'terminate':
                self._handle_termination(instance_id)

            self.send_response(200)
            self.end_headers()
            self.wfile.write(b'OK')

    def _handle_termination(self, instance_id):
        """处理节点终止"""
        logger.info(f"Starting graceful drain for {instance_id}")

        # 1. 标记节点为不可调度
        subprocess.run([
            'kubectl', 'cordon', instance_id
        ], check=False)

        # 2. 排空节点，给 Pod 优雅终止时间
        subprocess.run([
            'kubectl', 'drain', instance_id,
            '--ignore-daemonsets',
            '--delete-emptydir-data',
            '--grace-period=120',    # 2 分钟优雅终止
            '--timeout=300s'         # 最长等 5 分钟
        ], check=False)

        logger.info(f"Node {instance_id} drained successfully")

    def log_message(self, format, *args):
        logger.info(format % args)

def start_interruption_listener(port=5000):
    """启动中断监听服务"""
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s [%(levelname)s] %(message)s'
    )

    server = HTTPServer(('0.0.0.0', port), SpotInterruptionHandler)
    logger.info(f"Spot interruption listener started on port {port}")
    server.serve_forever()

if __name__ == '__main__':
    start_interruption_listener()

# 确保至少 2 个副本始终可用
apiVersion: policy/v1
kind: PodDisruptionBudget
metadata:
  name: api-service-pdb
  namespace: production
spec:
  minAvailable: 2            # 或使用 maxUnavailable: 1
  selector:
    matchLabels:
      app: api-service

---
# 批处理任务的 PDB
apiVersion: policy/v1
kind: PodDisruptionBudget
metadata:
  name: batch-processor-pdb
  namespace: production
spec:
  maxUnavailable: 30%        # 同时最多 30% 不可用
  selector:
    matchLabels:
      app: batch-processor

#!/usr/bin/env python3
"""
Kubernetes 成本分摊计算器
将集群成本按命名空间/标签分摊到各团队
"""

import json
from collections import defaultdict
from datetime import datetime, timedelta

class CostAllocator:
    """集群成本分摊计算器"""

    def __init__(self):
        # 节点信息
        self.nodes = []
        # Pod 资源使用
        self.pods = []
        # 云厂商定价（示例，美元/小时）
        self.instance_pricing = {
            'm6i.large': 0.096,      # 按需
            'm6i.large_spot': 0.029, # Spot
            'm5.large': 0.096,
            'm5.large_spot': 0.029,
            'r6i.large': 0.126,
            'c6i.large': 0.085,
        }

    def add_node(self, name, instance_type, is_spot, namespace_pods):
        """
        Args:
            name: 节点名
            instance_type: 实例类型
            is_spot: 是否 Spot 实例
            namespace_pods: {namespace: [{cpu_request_m, memory_request_mi}]}
        """
        self.nodes.append({
            'name': name,
            'instance_type': instance_type,
            'is_spot': is_spot,
            'namespace_pods': namespace_pods
        })

    def calculate_allocation(self):
        """计算成本分摊"""
        allocation = defaultdict(lambda: {
            'cpu_request_m': 0,
            'memory_request_mi': 0,
            'node_cost': 0,
            'pod_count': 0
        })

        for node in self.nodes:
            pricing_key = (
                f"{node['instance_type']}_spot"
                if node['is_spot']
                else node['instance_type']
            )
            hourly_cost = self.instance_pricing.get(pricing_key, 0.10)

            # 按小时计算
            monthly_cost = hourly_cost * 24 * 30

            # 统计该节点上各命名空间的资源请求
            ns_resources = defaultdict(lambda: {
                'cpu_request_m': 0,
                'memory_request_mi': 0,
                'pod_count': 0
            })

            total_cpu = 0
            total_mem = 0
            for ns, pods in node['namespace_pods'].items():
                for pod in pods:
                    ns_resources[ns]['cpu_request_m'] += pod.get('cpu_request_m', 0)
                    ns_resources[ns]['memory_request_mi'] += pod.get('memory_request_mi', 0)
                    ns_resources[ns]['pod_count'] += 1
                    total_cpu += pod.get('cpu_request_m', 0)
                    total_mem += pod.get('memory_request_mi', 0)

            # 按资源占比分摊节点成本
            if total_cpu > 0:
                for ns, res in ns_resources.items():
                    cpu_ratio = res['cpu_request_m'] / total_cpu
                    allocated_cost = monthly_cost * cpu_ratio

                    allocation[ns]['cpu_request_m'] += res['cpu_request_m']
                    allocation[ns]['memory_request_mi'] += res['memory_request_mi']
                    allocation[ns]['node_cost'] += allocated_cost
                    allocation[ns]['pod_count'] += res['pod_count']

        # 计算每命名空间的汇总
        result = []
        for ns, data in sorted(allocation.items(),
                                key=lambda x: x[1]['node_cost'],
                                reverse=True):
            result.append({
                'namespace': ns,
                'monthly_cost_usd': round(data['node_cost'], 2),
                'pod_count': data['pod_count'],
                'cpu_request_cores': round(data['cpu_request_m'] / 1000, 2),
                'memory_request_gib': round(data['memory_request_mi'] / 1024, 2),
                'cost_per_pod': round(
                    data['node_cost'] / data['pod_count'], 2
                ) if data['pod_count'] > 0 else 0
            })

        total_cost = sum(r['monthly_cost_usd'] for r in result)
        return {
            'period': 'monthly',
            'total_cluster_cost': round(total_cost, 2),
            'namespace_breakdown': result
        }

# 使用示例
if __name__ == '__main__':
    allocator = CostAllocator()

    # 模拟集群数据
    allocator.add_node('node-1', 'm6i.large', is_spot=False,
        namespace_pods={
            'production': [
                {'cpu_request_m': 250, 'memory_request_mi': 512},
                {'cpu_request_m': 500, 'memory_request_mi': 1024},
            ],
            'staging': [
                {'cpu_request_m': 100, 'memory_request_mi': 256},
            ]
        })

    allocator.add_node('node-2', 'm6i.large', is_spot=True,
        namespace_pods={
            'production': [
                {'cpu_request_m': 500, 'memory_request_mi': 1024},
                {'cpu_request_m': 500, 'memory_request_mi': 1024},
            ],
            'dev': [
                {'cpu_request_m': 50, 'memory_request_mi': 128},
            ]
        })

    result = allocator.calculate_allocation()
    print(json.dumps(result, indent=2, ensure_ascii=False))

# OpenCost 部署（通过 Helm）
# helm install opencost opencost/opencost \
#   --namespace opencost \
#   --create-namespace \
#   --set opencost.exporter.cloudProvider=aws \
#   --set opencost.exporter.clusterName=production-cluster

# OpenCost 提供 Prometheus 指标，可用 PromQL 查询成本
# 示例 PromQL 查询：
# 按命名空间查询月度成本
# container_cost_per_namespace_usd

# 按 Pod 查询 CPU 浪费
# sum by (pod) (
#   kube_pod_container_resource_requests{resource="cpu"}
#   - on(pod) group_left()
#   rate(container_cpu_usage_seconds_total[5m])
# )

# 使用 kubectl + jq 快速查看各命名空间资源消耗
kubectl get pods --all-namespaces -o json | \
  jq '.items[] |
  {
    namespace: .metadata.namespace,
    cpu_request: (
      .spec.containers[].resources.requests.cpu // "0"
      | sub("m$"; "") | tonumber
    ),
    memory_request: (
      .spec.containers[].resources.requests.memory // "0"
      | sub("Gi$"; "*1024") | sub("Mi$"; "") | eval
    )
  } |
  .cpu_request as $cpu |
  .memory_request as $mem |
  {namespace, cpu_m: $cpu, memory_mi: $mem}
  ' | \
  jq -s 'group_by(.namespace) |
    map({
      namespace: .[0].namespace,
      total_cpu_m: (map(.cpu_m) | add),
      total_memory_mi: (map(.memory_mi) | add),
      pod_count: length
    }) | sort_by(-.total_cpu_m)'

# ARM 节点池（Graviton 处理器，性价比高）
# AWS Graviton 实例通常比 x86 便宜 20% 且性能更好
apiVersion: apps/v1
kind: Deployment
metadata:
  name: api-service-arm
  namespace: production
spec:
  template:
    spec:
      nodeSelector:
        kubernetes.io/arch: arm64
      containers:
      - name: api
        image: registry.example.com/api-service:arm64-v2.1
        resources:
          requests:
            cpu: "200m"
            memory: "384Mi"
          limits:
            cpu: "200m"
            memory: "384Mi"

# 为使用 Kata Containers 等沙箱运行时的 Pod 声明额外开销
apiVersion: node.k8s.io/v1
kind: RuntimeClass
metadata:
  name: kata-containers
handler: kata-qemu
overhead:
  podFixed:
    cpu: "150m"        # VMM 额外开销
    memory: "160Mi"    # VM 额外内存

---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: secure-workload
spec:
  template:
    spec:
      runtimeClassName: kata-containers  # 使用沙箱运行时
      containers:
      - name: app
        image: app:v1
        resources:
          requests:
            cpu: "500m"
            memory: "1Gi"
          # 实际调度时算: 500m + 150m = 650m CPU, 1Gi + 160Mi = 1184Mi

# Kubernetes Descheduler 配置
apiVersion: "descheduler/v1alpha1"
kind: "DeschedulerPolicy"
strategies:
  # 移除低利用率的 Pod（触发重新调度到更紧凑的节点）
  - name: "LowNodeUtilization"
    enabled: true
    params:
      nodeResourceUtilizationThresholds:
        thresholds:
          cpu: 20         # CPU 使用低于 20% 的节点视为低利用
          memory: 20
          pods: 30
        targetThresholds:
          cpu: 50          # 目标利用率 50%
          memory: 50
          pods: 50

  # 移除违反拓扑分布约束的 Pod
  - name: "RemovePodsViolatingTopologySpreadConstraint"
    enabled: true

  # 移除重复的 Pod（同一节点上同一 Deployment 的多个副本）
  - name: "RemoveDuplicates"
    enabled: true
    params:
      nodeFit: true        # 确保被驱逐的 Pod 能重新调度