War Story: Debugging a CPU Throttling Issue in Kubernetes 1.33 Pods Using cAdvisor 0.49 and Prometheus

// cadvisor_query.go
// Queries cAdvisor 0.49 API on a Kubernetes node to fetch CPU throttling metrics
// Usage: go run cadvisor_query.go --node-ip=192.168.1.10 --scrape-interval=10s
package main

import (
    "encoding/json"
    "flag"
    "fmt"
    "io"
    "net/http"
    "os"
    "time"
)

// ThrottlingData represents CFS throttling stats from cAdvisor
type ThrottlingData struct {
    Periods          uint64  `json:"periods"`
    ThrottledPeriods uint64  `json:"throttled_periods"`
    ThrottledTime    float64 `json:"throttled_time"` // Seconds
}

// CpuStats represents CPU stats from cAdvisor
type CpuStats struct {
    UsageCoreNanoSeconds uint64         `json:"usage_core_nanoseconds"`
    Throttling           ThrottlingData `json:"throttling"`
}

// ContainerStats represents per-container stats from cAdvisor
type ContainerStats struct {
    Timestamp time.Time `json:"timestamp"`
    Cpu       CpuStats  `json:"cpu"`
    Name      string    `json:"name"`
}

// MachineStats represents top-level cAdvisor response
type MachineStats struct {
    Containers []ContainerStats `json:"containers"`
    MachineID  string           `json:"machine_id"`
}

func main() {
    // Parse command line flags
    nodeIP := flag.String("node-ip", "", "IP address of the Kubernetes node running cAdvisor")
    scrapeInterval := flag.Duration("scrape-interval", 10*time.Second, "Scrape interval for metrics (unused here, for reference)")
    flag.Parse()

    if *nodeIP == "" {
        fmt.Fprintf(os.Stderr, "Error: --node-ip is required\n")
        flag.Usage()
        os.Exit(1)
    }

    // cAdvisor default port is 4194
    cadvisorURL := fmt.Sprintf("http://%s:4194/api/v2.0/stats?type=container&recursive=true", *nodeIP)
    client := &http.Client{
        Timeout: 5 * time.Second,
    }

    // Query cAdvisor API
    resp, err := client.Get(cadvisorURL)
    if err != nil {
        fmt.Fprintf(os.Stderr, "Error querying cAdvisor: %v\n", err)
        os.Exit(1)
    }
    defer resp.Body.Close()

    // Check HTTP status code
    if resp.StatusCode != http.StatusOK {
        fmt.Fprintf(os.Stderr, "cAdvisor returned non-200 status: %d\n", resp.StatusCode)
        os.Exit(1)
    }

    // Read response body
    body, err := io.ReadAll(resp.Body)
    if err != nil {
        fmt.Fprintf(os.Stderr, "Error reading response body: %v\n", err)
        os.Exit(1)
    }

    // Parse JSON response
    var stats MachineStats
    if err := json.Unmarshal(body, &stats); err != nil {
        fmt.Fprintf(os.Stderr, "Error parsing JSON: %v\n", err)
        os.Exit(1)
    }

    // Print throttling metrics for each container
    fmt.Printf("Machine ID: %s\n", stats.MachineID)
    fmt.Printf("Scrape Interval (reference): %v\n", *scrapeInterval)
    fmt.Println("--- Container CPU Throttling Stats ---")
    for _, container := range stats.Containers {
        // Skip non-pod containers (e.g., system containers)
        if len(container.Name) < 10 { // Pod names are longer than 10 characters
            continue
        }
        throttlingRatio := float64(0)
        if container.Cpu.Throttling.Periods > 0 {
            throttlingRatio = float64(container.Cpu.Throttling.ThrottledPeriods) / float64(container.Cpu.Throttling.Periods) * 100
        }
        fmt.Printf("Container: %s\n", container.Name)
        fmt.Printf("  Total Periods: %d\n", container.Cpu.Throttling.Periods)
        fmt.Printf("  Throttled Periods: %d (%.2f%%)\n", container.Cpu.Throttling.ThrottledPeriods, throttlingRatio)
        fmt.Printf("  Throttled Time: %.2f seconds\n", container.Cpu.Throttling.ThrottledTime)
        fmt.Printf("  Timestamp: %v\n\n", container.Timestamp)
    }
}

"""
prometheus_throttling_alert.py
Fetches CPU throttling metrics from Prometheus, calculates throttling ratios,
and generates alerts if thresholds are exceeded.
Dependencies: requests, pyyaml
Usage: python prometheus_throttling_alert.py --config=config.yaml
"""

import argparse
import json
import os
import sys
import time
from datetime import datetime

import requests
import yaml

# Default configuration
DEFAULT_CONFIG = {
    "prometheus_url": "http://prometheus:9090",
    "throttling_threshold": 5.0,  # Throttling ratio % threshold
    "scrape_interval": 10,  # Seconds between checks
    "alert_webhook": "https://alert-webhook:8080/alerts",
    "metrics": {
        "throttled_periods": "container_cpu_cfs_throttled_periods_total",
        "total_periods": "container_cpu_cfs_periods_total",
        "pod_info": "kube_pod_info"
    }
}

class PrometheusThrottlingChecker:
    def __init__(self, config):
        self.prometheus_url = config["prometheus_url"]
        self.threshold = config["throttling_threshold"]
        self.scrape_interval = config["scrape_interval"]
        self.alert_webhook = config["alert_webhook"]
        self.metrics = config["metrics"]
        self.session = requests.Session()
        self.session.headers.update({"Accept": "application/json"})

    def query_prometheus(self, query):
        """Execute a PromQL query against Prometheus API."""
        url = f"{self.prometheus_url}/api/v1/query"
        try:
            response = self.session.get(url, params={"query": query}, timeout=5)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            print(f"Error querying Prometheus: {e}", file=sys.stderr)
            return None

    def get_throttling_ratio(self):
        """Calculate per-pod CPU throttling ratio (throttled periods / total periods * 100)."""
        query = f"""
            sum(rate({self.metrics["throttled_periods"]}[1m])) by (pod, namespace) 
            / 
            sum(rate({self.metrics["total_periods"]}[1m])) by (pod, namespace) 
            * 100
        """
        result = self.query_prometheus(query)
        if not result or result.get("status") != "success":
            return []

        ratios = []
        for item in result.get("data", {}).get("result", []):
            metric = item.get("metric", {})
            pod = metric.get("pod", "unknown")
            namespace = metric.get("namespace", "unknown")
            value = float(item.get("value", [0, 0])[1])
            ratios.append({
                "pod": pod,
                "namespace": namespace,
                "ratio": value,
                "timestamp": datetime.now().isoformat()
            })
        return ratios

    def send_alert(self, pod, namespace, ratio):
        """Send alert to webhook if throttling ratio exceeds threshold."""
        alert = {
            "alert": "HighCPUThrottling",
            "pod": pod,
            "namespace": namespace,
            "throttling_ratio": f"{ratio:.2f}%".
            "threshold": f"{self.threshold:.2f}%".
            "timestamp": datetime.now().isoformat(),
            "description": f"Pod {pod} in namespace {namespace} has CPU throttling ratio of {ratio:.2f}%, exceeding threshold of {self.threshold:.2f}%"
        }
        try:
            response = self.session.post(self.alert_webhook, json=alert, timeout=5)
            response.raise_for_status()
            print(f"Sent alert for {pod}/{namespace}: {ratio:.2f}%")
        except requests.exceptions.RequestException as e:
            print(f"Error sending alert: {e}", file=sys.stderr)

    def run(self):
        """Main run loop: check throttling ratios and send alerts."""
        print(f"Starting throttling checker. Threshold: {self.threshold}%, Interval: {self.scrape_interval}s")
        while True:
            try:
                ratios = self.get_throttling_ratio()
                for item in ratios:
                    pod = item["pod"]
                    namespace = item["namespace"]
                    ratio = item["ratio"]
                    if ratio >= self.threshold:
                        print(f"High throttling detected: {pod}/{namespace} {ratio:.2f}%")
                        self.send_alert(pod, namespace, ratio)
                    else:
                        print(f"Normal throttling: {pod}/{namespace} {ratio:.2f}%")
            except Exception as e:
                print(f"Unexpected error: {e}", file=sys.stderr)
            time.sleep(self.scrape_interval)

def load_config(config_path):
    """Load configuration from YAML file, merge with defaults."""
    config = DEFAULT_CONFIG.copy()
    if config_path and os.path.exists(config_path):
        with open(config_path, "r") as f:
            user_config = yaml.safe_load(f)
            config.update(user_config)
    return config

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Prometheus CPU Throttling Checker")
    parser.add_argument("--config", help="Path to configuration YAML file")
    args = parser.parse_args()

    config = load_config(args.config)
    checker = PrometheusThrottlingChecker(config)
    try:
        checker.run()
    except KeyboardInterrupt:
        print("Shutting down...")
        sys.exit(0)

// pod_cpu_reconciler.go
// Reconciles pod CPU limits with actual throttling metrics from cAdvisor
// Requires cluster-admin permissions to list pods and patch resources
package main

import (
    "context"
    "encoding/json"
    "flag"
    "fmt"
    "os"
    "time"

    corev1 "k8s.io/api/core/v1"
    "k8s.io/apimachinery/pkg/api/resource"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/client-go/kubernetes"
    "k8s.io/client-go/tools/clientcmd"
    "k8s.io/client-go/util/homedir"
)

// ThrottlingRecommendation represents a CPU limit recommendation
type ThrottlingRecommendation struct {
    Pod       string    `json:"pod"`
    Namespace string    `json:"namespace"`
    Current   string    `json:"current_limit"`
    Recommended string `json:"recommended_limit"`
    ThrottlingRatio float64 `json:"throttling_ratio"`
}

func main() {
    // Parse flags
    var kubeconfig *string
    if home := homedir.HomeDir(); home != "" {
        kubeconfig = flag.String("kubeconfig", home+"/.kube/config", "Absolute path to kubeconfig file")
    } else {
        kubeconfig = flag.String("kubeconfig", "", "Absolute path to kubeconfig file")
    }
    dryRun := flag.Bool("dry-run", true, "If true, only print recommendations without patching")
    flag.Parse()

    // Build kubernetes config
    config, err := clientcmd.BuildConfigFromFlags("", *kubeconfig)
    if err != nil {
        fmt.Fprintf(os.Stderr, "Error building kubeconfig: %v\n", err)
        os.Exit(1)
    }

    // Create clientset
    clientset, err := kubernetes.NewForConfig(config)
    if err != nil {
        fmt.Fprintf(os.Stderr, "Error creating clientset: %v\n", err)
        os.Exit(1)
    }

    ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
    defer cancel()

    // List all pods in all namespaces
    pods, err := clientset.CoreV1().Pods("").List(ctx, metav1.ListOptions{})
    if err != nil {
        fmt.Fprintf(os.Stderr, "Error listing pods: %v\n", err)
        os.Exit(1)
    }

    fmt.Printf("Found %d pods. Generating CPU limit recommendations...\n", len(pods.Items))

    recommendations := []ThrottlingRecommendation{}

    for _, pod := range pods.Items {
        // Skip system pods
        if pod.Namespace == "kube-system" {
            continue
        }

        // Get current CPU limit for the first container (simplified)
        if len(pod.Spec.Containers) == 0 {
            continue
        }
        container := pod.Spec.Containers[0]
        currentLimit := container.Resources.Limits.Cpu()
        if currentLimit == nil || currentLimit.IsZero() {
            currentLimit = resource.NewMilliQuantity(1000, resource.DecimalSI) // Default 1 core if not set
        }

        // In production, this would query cAdvisor for actual throttling ratio
        // For this example, we simulate a throttling ratio based on pod labels
        throttlingRatio := 0.0
        if workload, ok := pod.Labels["workload"]; ok && workload == "batch" {
            throttlingRatio = 12.5 // Simulated 12.5% throttling for batch workloads
        }

        // Calculate recommended limit: increase by 20% if throttling > 5%
        recommendedLimit := currentLimit.Copy()
        if throttlingRatio > 5.0 {
            recommendedLimit.Add(*resource.NewMilliQuantity(currentLimit.MilliValue()/5, resource.DecimalSI)) // Add 20%
        }

        recommendations = append(recommendations, ThrottlingRecommendation{
            Pod:       pod.Name,
            Namespace: pod.Namespace,
            Current:   currentLimit.String(),
            Recommended: recommendedLimit.String(),
            ThrottlingRatio: throttlingRatio,
        })
    }

    // Print recommendations
    jsonOutput, err := json.MarshalIndent(recommendations, "", "  ")
    if err != nil {
        fmt.Fprintf(os.Stderr, "Error marshaling recommendations: %v\n", err)
        os.Exit(1)
    }
    fmt.Println(string(jsonOutput))

    // Patch pods if not dry run
    if !*dryRun {
        fmt.Println("Patching pods with recommended limits...")
        for _, rec := range recommendations {
            if rec.ThrottlingRatio <= 5.0 {
                continue
            }
            // Patch the pod's container resources
            pod, err := clientset.CoreV1().Pods(rec.Namespace).Get(ctx, rec.Pod, metav1.GetOptions{})
            if err != nil {
                fmt.Fprintf(os.Stderr, "Error getting pod %s/%s: %v\n", rec.Namespace, rec.Pod, err)
                continue
            }
            // Update first container's CPU limit
            pod.Spec.Containers[0].Resources.Limits[corev1.ResourceCPU] = resource.MustParse(rec.Recommended)
            _, err = clientset.CoreV1().Pods(rec.Namespace).Update(ctx, pod, metav1.UpdateOptions{})
            if err != nil {
                fmt.Fprintf(os.Stderr, "Error updating pod %s/%s: %v\n", rec.Namespace, rec.Pod, err)
                continue
            }
            fmt.Printf("Patched pod %s/%s: new limit %s\n", rec.Namespace, rec.Pod, rec.Recommended)
        }
    }
}

scrape_configs:
  - job_name: 'cadvisor'
    scrape_interval: 10s
    static_configs:
      - targets: ['cadvisor:4194']
    metric_relabel_configs:
      - source_labels: [__name__]
        regex: 'container_cpu_cfs_throttled.*'
        action: keep

kubectl get --raw /api/v1/nodes/${NODE_NAME}/proxy/metrics | grep container_cpu_cfs_throttled_seconds_total

sum(rate(container_cpu_cfs_throttled_periods_total[5m])) by (pod, namespace) 
/ 
sum(rate(container_cpu_cfs_periods_total[5m])) by (pod, namespace) 
* 100