Kubernetes Cost Optimization in Practice: From Resource Governance to FinOps

# Typical over-provisioning
apiVersion: apps/v1
kind: Deployment
metadata:
  name: api-service
spec:
  template:
    spec:
      containers:
      - name: api
        resources:
          requests:
            cpu: "2000m"    # Actual usage 200m, 90% waste
            memory: "4Gi"   # Actual usage 512Mi, 87% waste
          limits:
            cpu: "4000m"
            memory: "8Gi"

# A namespace without ResourceQuota = unlimited resource consumption
# This means one team's services can crowd out other teams' resources

#!/usr/bin/env python3
"""
Pod QoS class determination and resource waste analysis tool
Scans all Pods in a cluster, identifies configuration issues and cost waste
"""

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

@dataclass
class PodResourceInfo:
    """Pod resource configuration info"""
    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          # Actual usage (from metrics-server)
    memory_usage_mi: float

    @property
    def cpu_waste_m(self):
        """CPU resource waste = Request - actual usage"""
        return max(0, self.cpu_request_m - self.cpu_usage_m)

    @property
    def memory_waste_mi(self):
        """Memory resource waste"""
        return max(0, self.memory_request_mi - self.memory_usage_mi)

    @property
    def cpu_utilization(self):
        """Request utilization rate"""
        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:
    """Cluster resource waste analyzer"""

    # Optimization thresholds
    LOW_UTILIZATION_THRESHOLD = 0.30    # Below 30% utilization = waste
    HIGH_UTILIZATION_THRESHOLD = 0.85   # Above 85% utilization = risk
    OVERREQUEST_MULTIPLIER = 3.0       # Request > 3x actual usage = over-provisioned

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

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

    def analyze(self) -> dict:
        """Analyze cluster resource waste"""
        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 distribution statistics"""
        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):
        """Resource waste summary"""
        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):
        """Generate optimization recommendations"""
        recs = []

        for pod in self.pods:
            # Low utilization detection
            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  # minimum 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'
                })

            # Memory over-provisioning
            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 alert
            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'
                })

            # High utilization risk (may be Throttled or 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

# Usage example
if __name__ == '__main__':
    analyzer = ResourceWasteAnalyzer()

    # Simulate Pod data
    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 configuration template
apiVersion: apps/v1
kind: Deployment
metadata:
  name: api-service
  namespace: production
spec:
  template:
    spec:
      containers:
      - name: api
        # Core service: Guaranteed QoS, Requests == Limits
        resources:
          requests:
            cpu: "250m"      # Based on P95 actual usage * 1.5
            memory: "512Mi"  # Based on P95 actual usage * 1.3
          limits:
            cpu: "250m"      # Same as requests, avoid throttle
            memory: "512Mi"  # Same as requests, Guaranteed QoS
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: background-worker
  namespace: production
spec:
  template:
    spec:
      containers:
      - name: worker
        # Non-core service: Burstable QoS, allows bursting
        resources:
          requests:
            cpu: "100m"     # Low baseline
            memory: "256Mi"
          limits:
            cpu: "500m"     # Allow bursting to 5x
            memory: "1Gi"

#!/usr/bin/env python3
"""
Right-Sizing recommendations based on Prometheus historical metrics
Analyzes past 7 days of resource usage data, provides Requests/Limits suggestions
"""

import json
from datetime import datetime, timedelta

class RightSizingRecommender:
    """Resource configuration recommender based on historical metrics"""

    # Recommendation multipliers
    CPU_REQUEST_MULTIPLIER = 1.5    # P95 usage * 1.5
    CPU_LIMIT_MULTIPLIER = 2.0     # P95 usage * 2.0
    MEM_REQUEST_MULTIPLIER = 1.3   # P95 usage * 1.3
    MEM_LIMIT_MULTIPLIER = 1.5     # P95 usage * 1.5

    # Minimum values
    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):
        """Add a metric data point"""
        self.metrics.append({
            'timestamp': timestamp,
            'cpu_m': cpu_m,
            'memory_mi': memory_mi
        })

    def recommend(self, service_name, qos='burstable'):
        """
        Generate resource configuration recommendations

        Args:
            service_name: Service name
            qos: Target QoS class ('guaranteed' or '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])

        # Calculate 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),
        }

        # Generate recommendations
        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):
        """Calculate percentile"""
        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):
        """Generate YAML configuration snippet"""
        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\""""

# Usage example
if __name__ == '__main__':
    recommender = RightSizingRecommender()

    # Simulate 7 days of historical data (one data point per hour)
    import random
    random.seed(42)
    base_time = datetime(2026, 7, 4)

    for i in range(168):  # 7 days * 24 hours
        ts = base_time + timedelta(hours=i)
        # Simulate daytime peak, nighttime trough CPU pattern
        hour = ts.hour
        if 9 <= hour <= 18:  # Working hours
            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: constrain individual Pod resource configuration
apiVersion: v1
kind: LimitRange
metadata:
  name: production-limits
  namespace: production
spec:
  limits:
  # Default values (when not explicitly set)
  - type: Container
    default:          # default = limits
      cpu: "500m"
      memory: "512Mi"
    defaultRequest:   # defaultRequest = requests
      cpu: "100m"
      memory: "128Mi"
    # Upper and lower bound constraints
    max:
      cpu: "4"
      memory: "8Gi"
    min:
      cpu: "50m"
      memory: "64Mi"
    # Limit-to-Request ratio constraint
    maxLimitRequestRatio:
      cpu: 4           # Limit can be at most 4x request
      memory: 2        # Memory does not recommend large ratios

---
# 2. ResourceQuota: constrain total namespace resources
apiVersion: v1
kind: ResourceQuota
metadata:
  name: production-quota
  namespace: production
spec:
  hard:
    requests.cpu: "100"       # Namespace total CPU cap
    requests.memory: 200Gi
    limits.cpu: "200"
    limits.memory: 400Gi
    pods: "200"                # Pod count limit
    services: "50"
    configmaps: "100"
    persistentvolumeclaims: "20"
    requests.storage: "500Gi"

---
# 3. Multi-level Quota (allocated by team)
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 mode: only gives recommendations, does not modify
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 = recommend only
                          # Initial = apply only at Pod creation
                          # Auto = auto-adjust (will restart Pod)
  resourcePolicy:
    containerPolicies:
    - containerName: api
      minAllowed:
        cpu: 50m
        memory: 128Mi
      maxAllowed:
        cpu: 2000m
        memory: 4Gi
      controlledResources: ["cpu", "memory"]

# View VPA recommendations
kubectl describe vpa api-service-vpa -n production

# Example output:
# 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 based on CPU and memory utilization
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: api-service-hpa
  namespace: production
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: api-service
  minReplicas: 3              # Minimum 3 replicas (ensure availability)
  maxReplicas: 30             # Maximum 30 replicas
  metrics:
  # CPU utilization (relative to Requests)
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70  # Target CPU utilization 70%
  # Memory utilization
  - type: Resource
    resource:
      name: memory
      target:
        type: Utilization
        averageUtilization: 80
  behavior:
    # Scale-up behavior: fast scaling
    scaleUp:
      stabilizationWindowSeconds: 0  # No stabilization window, scale immediately
      policies:
      - type: Percent
        value: 100              # Scale up at most 100% each time
        periodSeconds: 30
      - type: Pods
        value: 6                # Or add at most 6 Pods each time
        periodSeconds: 30
      selectPolicy: Max         # Take the larger of the two policies
    # Scale-down behavior: slow scaling
    scaleDown:
      stabilizationWindowSeconds: 300  # 5-minute stabilization window
      policies:
      - type: Percent
        value: 10                # Scale down at most 10% each time
        periodSeconds: 60

# Cluster Autoscaler configuration (AWS EKS example)
# Note: This is the AWS Auto Scaling Group configuration policy
# Cluster Autoscaler automatically scales nodes based on unschedulable Pods

# Node group configuration recommendations
nodeGroups:
  # On-demand node group: guarantees baseline capacity
  - name: on-demand-base
    instanceType: m6i.large
    minSize: 3           # Minimum 3 nodes for high availability
    maxSize: 10
    spot: false

  # Spot instance node group: absorbs elastic load
  - name: spot-elastic
    instanceType:
    - m6i.large
    - m5.large
    - m5a.large
    minSize: 0           # Can scale to 0
    maxSize: 20
    spot: true

# KEDA ScaledObject: scale based on Kafka consumption lag
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       # Scale to 1 when idle
  maxReplicaCount: 20      # Scale to 20 at peak
  pollingInterval: 30       # Check every 30 seconds
  cooldownPeriod: 300       # 5-minute cooldown for scale-down
  triggers:
  - type: kafka
    metadata:
      bootstrapServers: kafka-broker.data.svc.cluster.local:9092
      consumerGroup: order-consumer-group
      topic: orders
      lagThreshold: "100"    # Trigger scaling when backlog exceeds 100 messages
      offsetResetPolicy: latest

#!/usr/bin/env python3
"""
Autoscaling strategy evaluator
Simulates scaling behavior under different scenarios to assess strategy effectiveness
"""

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

@dataclass
class TrafficPoint:
    """Traffic data at a point in time"""
    timestamp: int        # unix timestamp
    rps: float           # requests per second
    cpu_per_pod_m: float  # CPU usage per Pod (millicores)

@dataclass
class ScalingDecision:
    """Scaling decision record"""
    timestamp: int
    current_replicas: int
    target_replicas: int
    action: str          # 'scale_up', 'scale_down', 'no_change'
    reason: str

class HPASimulator:
    """HPA strategy simulator"""

    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]:
        """Simulate HPA behavior"""
        decisions = []
        current_replicas = self.min_replicas
        last_scale_up = 0
        last_scale_down = 0

        for point in traffic:
            # Calculate current total CPU utilization
            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 = ''

            # Scale-up logic
            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

            # Scale-down logic
            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):
        """Evaluate strategy effectiveness"""
        total_pod_hours = sum(d.target_replicas for d in decisions) / 60  # assuming one point per minute
        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)
        }

# Usage example
if __name__ == '__main__':
    import random
    random.seed(42)

    # Generate 24 hours of traffic data (one point per minute)
    traffic = []
    for minute in range(1440):
        hour = minute / 60
        if 9 <= hour < 12 or 14 <= hour < 18:
            rps = random.uniform(80, 120)   # Peak
        elif 0 <= hour < 6:
            rps = random.uniform(5, 15)     # Trough
        else:
            rps = random.uniform(30, 60)     # Normal

        traffic.append(TrafficPoint(
            timestamp=minute * 60,
            rps=rps,
            cpu_per_pod_m=2.5  # Each RPS consumes 2.5m CPU
        ))

    # Simulate conservative vs aggressive strategy
    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("=== Conservative Strategy (target 50%, cooldown 10 min) ===")
    print(json.dumps(conservative.evaluate(traffic, cons_decisions),
                     indent=2, ensure_ascii=False))

    print("\n=== Aggressive Strategy (target 75%, cooldown 2 min) ===")
    print(json.dumps(aggressive.evaluate(traffic, aggr_decisions),
                     indent=2, ensure_ascii=False))

# Spot node pool configuration (AWS EKS)
apiVersion: apps/v1
kind: Deployment
metadata:
  name: batch-processor
  namespace: production
spec:
  replicas: 10
  template:
    metadata:
      annotations:
        # Mark as schedulable on Spot nodes
        sqs.amazonaws.com/queue-name: "batch-queue"
    spec:
      nodeSelector:
        kubernetes.io/arch: amd64
      tolerations:
      # Tolerate Spot node taints
      - key: "spot-instance"
        operator: "Equal"
        value: "true"
        effect: "NoPrefer"
      # Graceful termination: give tasks time to finish processing
      terminationGracePeriodSeconds: 300
      containers:
      - name: processor
        image: registry.example.com/processor:v2.1
        resources:
          requests:
            cpu: "500m"
            memory: "1Gi"
          limits:
            cpu: "1000m"
            memory: "2Gi"
        # Graceful termination hook
        lifecycle:
          preStop:
            exec:
              command:
              - /bin/sh
              - -c
              - |
                # Notify task manager that current task needs requeuing
                curl -X POST http://task-manager:8080/requeue \
                  -d '{"pod": "$HOSTNAME", "action": "graceful_shutdown"}'
                sleep 30  # Wait for in-flight tasks to complete                

#!/usr/bin/env python3
"""
Spot instance interruption handler
Listens for AWS Spot interruption notices and gracefully drains nodes
"""

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

logger = logging.getLogger(__name__)

class SpotInterruptionHandler(BaseHTTPRequestHandler):
    """Handle Spot instance interruption notices"""

    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):
        """Handle node termination"""
        logger.info(f"Starting graceful drain for {instance_id}")

        # 1. Mark node as unschedulable
        subprocess.run([
            'kubectl', 'cordon', instance_id
        ], check=False)

        # 2. Drain node, give Pods graceful termination time
        subprocess.run([
            'kubectl', 'drain', instance_id,
            '--ignore-daemonsets',
            '--delete-emptydir-data',
            '--grace-period=120',    # 2-minute graceful termination
            '--timeout=300s'         # Max wait 5 minutes
        ], 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):
    """Start interruption listener service"""
    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()

# Ensure at least 2 replicas are always available
apiVersion: policy/v1
kind: PodDisruptionBudget
metadata:
  name: api-service-pdb
  namespace: production
spec:
  minAvailable: 2            # Or use maxUnavailable: 1
  selector:
    matchLabels:
      app: api-service

---
# PDB for batch processing tasks
apiVersion: policy/v1
kind: PodDisruptionBudget
metadata:
  name: batch-processor-pdb
  namespace: production
spec:
  maxUnavailable: 30%        # At most 30% unavailable simultaneously
  selector:
    matchLabels:
      app: batch-processor

#!/usr/bin/env python3
"""
Kubernetes cost allocation calculator
Allocates cluster costs by namespace/label to teams
"""

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

class CostAllocator:
    """Cluster cost allocation calculator"""

    def __init__(self):
        self.nodes = []
        self.pods = []
        # Cloud provider pricing (example, USD/hour)
        self.instance_pricing = {
            'm6i.large': 0.096,      # On-demand
            '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: Node name
            instance_type: Instance type
            is_spot: Whether Spot instance
            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):
        """Calculate cost allocation"""
        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)

            # Calculate monthly cost
            monthly_cost = hourly_cost * 24 * 30

            # Aggregate resource requests by namespace on this node
            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)

            # Allocate node cost by resource ratio
            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']

        # Summarize per namespace
        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
        }

# Usage example
if __name__ == '__main__':
    allocator = CostAllocator()

    # Simulate cluster data
    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 deployment (via Helm)
# helm install opencost opencost/opencost \
#   --namespace opencost \
#   --create-namespace \
#   --set opencost.exporter.cloudProvider=aws \
#   --set opencost.exporter.clusterName=production-cluster

# OpenCost provides Prometheus metrics, query costs with PromQL
# Example PromQL queries:
# Monthly cost by namespace
# container_cost_per_namespace_usd

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

# Quick view of resource consumption by namespace using 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 node pool (Graviton processors, better price-performance)
# AWS Graviton instances are typically 20% cheaper and perform better than x86
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"

# Declare extra overhead for Pods using sandbox runtimes like Kata Containers
apiVersion: node.k8s.io/v1
kind: RuntimeClass
metadata:
  name: kata-containers
handler: kata-qemu
overhead:
  podFixed:
    cpu: "150m"        # VMM extra overhead
    memory: "160Mi"    # VM extra memory

---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: secure-workload
spec:
  template:
    spec:
      runtimeClassName: kata-containers  # Use sandbox runtime
      containers:
      - name: app
        image: app:v1
        resources:
          requests:
            cpu: "500m"
            memory: "1Gi"
          # Actual scheduling: 500m + 150m = 650m CPU, 1Gi + 160Mi = 1184Mi

# Kubernetes Descheduler configuration
apiVersion: "descheduler/v1alpha1"
kind: "DeschedulerPolicy"
strategies:
  # Remove Pods on low-utilization nodes (trigger rescheduling to denser nodes)
  - name: "LowNodeUtilization"
    enabled: true
    params:
      nodeResourceUtilizationThresholds:
        thresholds:
          cpu: 20         # Nodes with CPU usage below 20% are low-utilization
          memory: 20
          pods: 30
        targetThresholds:
          cpu: 50          # Target utilization 50%
          memory: 50
          pods: 50

  # Remove Pods violating topology spread constraints
  - name: "RemovePodsViolatingTopologySpreadConstraint"
    enabled: true

  # Remove duplicate Pods (multiple replicas of the same Deployment on one node)
  - name: "RemoveDuplicates"
    enabled: true
    params:
      nodeFit: true        # Ensure evicted Pods can be rescheduled