Benchmark: Cilium 1.17 vs Calico 3.29 vs Flannel 0.25: Kubernetes CNI Latency for 500 Node Clusters

package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "os"
    "os/exec"
    "strconv"
    "time"

    "gopkg.in/yaml.v3"
)

// BenchmarkConfig holds test parameters for CNI latency benchmarks
type BenchmarkConfig struct {
    ClusterNodes    int    `yaml:"clusterNodes"`
    TestDuration    string `yaml:"testDuration"`
    CNIVersions     []string `yaml:"cniVersions"`
    LoadPercent     int    `yaml:"loadPercent"`
    MetricsEndpoint string `yaml:"metricsEndpoint"`
}

// CNIBenchmarker orchestrates CNI benchmark runs
type CNIBenchmarker struct {
    config BenchmarkConfig
    logger *log.Logger
}

// NewCNIBenchmarker initializes a new benchmarker with config path
func NewCNIBenchmarker(configPath string, logger *log.Logger) (*CNIBenchmarker, error) {
    data, err := os.ReadFile(configPath)
    if err != nil {
        return nil, fmt.Errorf("failed to read config: %w", err)
    }

    var cfg BenchmarkConfig
    if err := yaml.Unmarshal(data, &cfg); err != nil {
        return nil, fmt.Errorf("failed to parse config: %w", err)
    }

    // Validate config
    if cfg.ClusterNodes <= 0 {
        return nil, fmt.Errorf("invalid cluster nodes: %d", cfg.ClusterNodes)
    }
    if cfg.LoadPercent < 0 || cfg.LoadPercent > 100 {
        return nil, fmt.Errorf("invalid load percent: %d", cfg.LoadPercent)
    }

    return &CNIBenchmarker{config: cfg, logger: logger}, nil
}

// Run executes the full benchmark suite for all CNI versions
func (b *CNIBenchmarker) Run(ctx context.Context) error {
    b.logger.Printf("Starting benchmark for %d-node cluster", b.config.ClusterNodes)

    for _, cniVer := range b.config.CNIVersions {
        b.logger.Printf("Testing CNI version: %s", cniVer)

        // Deploy CNI
        if err := b.deployCNI(ctx, cniVer); err != nil {
            b.logger.Printf("Failed to deploy %s: %v", cniVer, err)
            continue
        }

        // Run load test
        latency, err := b.runLoadTest(ctx, cniVer)
        if err != nil {
            b.logger.Printf("Load test failed for %s: %v", cniVer, err)
            continue
        }

        // Collect metrics
        if err := b.writeResults(cniVer, latency); err != nil {
            b.logger.Printf("Failed to write results for %s: %v", cniVer, err)
        }

        // Cleanup
        if err := b.cleanupCNI(ctx, cniVer); err != nil {
            b.logger.Printf("Cleanup failed for %s: %v", cniVer, err)
        }
    }

    return nil
}

// deployCNI handles CNI installation via Helm
func (b *CNIBenchmarker) deployCNI(ctx context.Context, cniVer string) error {
    // Split CNI name and version
    var cniName, version string
    switch {
    case contains(cniVer, "cilium"):
        cniName = "cilium"
        version = cniVer
    case contains(cniVer, "calico"):
        cniName = "calico"
        version = cniVer
    case contains(cniVer, "flannel"):
        cniName = "flannel"
        version = cniVer
    default:
        return fmt.Errorf("unsupported CNI: %s", cniVer)
    }

    cmd := exec.CommandContext(ctx, "helm", "install", cniName, fmt.Sprintf("%s/%s", cniName, cniName),
        "--version", version, "--namespace", "kube-system", "--create-namespace")
    output, err := cmd.CombinedOutput()
    if err != nil {
        return fmt.Errorf("helm install failed: %s: %w", output, err)
    }

    // Wait for CNI pods to be ready
    waitCmd := exec.CommandContext(ctx, "kubectl", "wait", "--for=condition=ready", "pod",
        "-l", fmt.Sprintf("k8s-app=%s", cniName), "-n", "kube-system", "--timeout=5m")
    if err := waitCmd.Run(); err != nil {
        return fmt.Errorf("CNI pod wait failed: %w", err)
    }

    return nil
}

// runLoadTest executes k6 load test and returns p99 latency
func (b *CNIBenchmarker) runLoadTest(ctx context.Context, cniVer string) (float64, error) {
    // Start k6 test
    cmd := exec.CommandContext(ctx, "k6", "run", "--vus", "10000", "--duration", b.config.TestDuration,
        "--out", fmt.Sprintf("json=results-%s.json", cniVer), "load-test.js")
    if err := cmd.Run(); err != nil {
        return 0, fmt.Errorf("k6 test failed: %w", err)
    }

    // Parse results
    data, err := os.ReadFile(fmt.Sprintf("results-%s.json", cniVer))
    if err != nil {
        return 0, fmt.Errorf("failed to read results: %w", err)
    }

    // Extract p99 latency (simplified for example)
    var results map[string]interface{}
    if err := json.Unmarshal(data, &results); err != nil {
        return 0, fmt.Errorf("failed to parse results: %w", err)
    }

    p99, ok := results["p99_latency_us"].(float64)
    if !ok {
        return 0, fmt.Errorf("p99 latency not found in results")
    }

    return p99, nil
}

// writeResults persists benchmark results to disk
func (b *CNIBenchmarker) writeResults(cniVer string, latency float64) error {
    // Implementation omitted for brevity
    return nil
}

// cleanupCNI removes CNI installation
func (b *CNIBenchmarker) cleanupCNI(ctx context.Context, cniVer string) error {
    // Implementation omitted for brevity
    return nil
}

// Helper to check string contains
func contains(s, substr string) bool {
    return len(s) >= len(substr) && (s == substr || len(s) > 0 && (s[0:len(substr)] == substr || contains(s[1:], substr)))
}

func main() {
    logger := log.New(os.Stdout, "[BENCH] ", log.LstdFlags)

    bench, err := NewCNIBenchmarker("benchmark-config.yaml", logger)
    if err != nil {
        logger.Fatalf("Failed to initialize benchmarker: %v", err)
    }

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

    if err := bench.Run(ctx); err != nil {
        logger.Fatalf("Benchmark failed: %v", err)
    }
}

#include 
#include 
#include 
#include 
#include 

// Define latency histogram map: key is latency bucket (μs), value is count
struct {
    __uint(type, BPF_MAP_TYPE_HASH);
    __type(key, u32);
    __type(value, u64);
    __uint(max_entries, 1024);
} latency_map SEC(".maps");

// XDP program to measure ingress latency
SEC("xdp")
int measure_ingress_latency(struct xdp_md *ctx) {
    void *data = (void *)(long)ctx->data;
    void *data_end = (void *)(long)ctx->data_end;

    // Parse Ethernet header
    struct ethhdr *eth = data;
    if ((void *)(eth + 1) > data_end) {
        return XDP_PASS;
    }

    // Only process IPv4 packets
    if (bpf_ntohs(eth->h_proto) != ETH_P_IP) {
        return XDP_PASS;
    }

    // Parse IP header
    struct iphdr *ip = (void *)(eth + 1);
    if ((void *)(ip + 1) > data_end) {
        return XDP_PASS;
    }

    // Check if packet is destined to a Kubernetes service CIDR (10.96.0.0/12)
    if ((ip->daddr & 0xFFF00000) != 0x0A600000) {
        return XDP_PASS;
    }

    // Get current timestamp (nanoseconds since boot)
    u64 now = bpf_ktime_get_ns();

    // Read timestamp from packet (set by egress probe, omitted for brevity)
    u64 *pkt_ts = (u64 *)(data_end - sizeof(u64));
    if ((void *)(pkt_ts + 1) > data_end) {
        return XDP_PASS;
    }

    // Calculate latency in microseconds
    u64 latency_ns = now - *pkt_ts;
    u32 latency_us = (u32)(latency_ns / 1000);

    // Increment latency bucket
    u64 *count = bpf_map_lookup_elem(&latency_map, &latency_us);
    if (count) {
        __sync_fetch_and_add(count, 1);
    } else {
        u64 init = 1;
        bpf_map_update_elem(&latency_map, &latency_us, &init, BPF_ANY);
    }

    return XDP_PASS;
}

// User space program to read latency histogram
#include 
#include 
#include 
#include 
#include 

int main(int argc, char **argv) {
    struct bpf_object *obj;
    struct bpf_map *map;
    int map_fd;

    // Load BPF object
    obj = bpf_object__open_file("latency.o", NULL);
    if (!obj) {
        fprintf(stderr, "Failed to open BPF object\n");
        return 1;
    }

    if (bpf_object__load(obj)) {
        fprintf(stderr, "Failed to load BPF object\n");
        return 1;
    }

    // Get latency map
    map = bpf_object__find_map_by_name(obj, "latency_map");
    if (!map) {
        fprintf(stderr, "Failed to find latency map\n");
        return 1;
    }

    map_fd = bpf_map__fd(map);

    // Read and print histogram
    u32 key, next_key;
    u64 value;

    printf("Latency (μs) | Count\n");
    printf("------------------\n");

    while (bpf_map_get_next_key(map_fd, &key, &next_key) == 0) {
        if (bpf_map_lookup_elem(map_fd, &next_key, &value) == 0) {
            printf("%10u | %llu\n", next_key, value);
        }
        key = next_key;
    }

    // Calculate p99 latency
    // Implementation omitted for brevity

    bpf_object__close(obj);
    return 0;
}

char _license[] SEC("license") = "GPL";

#!/usr/bin/env python3
import json
import sys
from typing import Dict, List, Optional

# Constants
NODE_COST_PER_MONTH = 400  # AWS m6i.4xlarge on-demand cost
SLA_PENALTY_PER_MS = 100   # $100 per ms over 200μs p99 latency
OPERATIONAL_OVERHEAD = 0.1  # 10% of node cost for CNI operations

class CNIBenchmarkResult:
    """Holds benchmark results for a single CNI version."""
    def __init__(self, name: str, version: str, p99_latency_us: float, throughput_gbps: float,
                 cpu_percent: float, memory_mb: float, node_count: int):
        self.name = name
        self.version = version
        self.p99_latency_us = p99_latency_us
        self.throughput_gbps = throughput_gbps
        self.cpu_percent = cpu_percent
        self.memory_mb = memory_mb
        self.node_count = node_count
        self._validate()

    def _validate(self) -> None:
        """Validate benchmark result fields."""
        if self.p99_latency_us <= 0:
            raise ValueError(f"Invalid p99 latency: {self.p99_latency_us}")
        if self.node_count <= 0:
            raise ValueError(f"Invalid node count: {self.node_count}")

    def calculate_monthly_cost(self) -> float:
        """Calculate total monthly cost for this CNI."""
        # Node cost
        node_cost = self.node_count * NODE_COST_PER_MONTH

        # CPU overhead cost (assume 1 vCPU = $30/month)
        cpu_cost = self.node_count * (self.cpu_percent / 100) * 16 * 30

        # Memory overhead cost (assume 1GB = $10/month)
        memory_cost = self.node_count * (self.memory_mb / 1024) * 10

        # SLA penalty cost
        latency_ms = self.p99_latency_us / 1000
        sla_penalty = max(0, (latency_ms - 0.2)) * SLA_PENALTY_PER_MS

        # Operational overhead
        ops_cost = node_cost * OPERATIONAL_OVERHEAD

        return node_cost + cpu_cost + memory_cost + sla_penalty + ops_cost

    def calculate_savings_vs(self, other: "CNIBenchmarkResult") -> float:
        """Calculate monthly savings compared to another CNI."""
        return other.calculate_monthly_cost() - self.calculate_monthly_cost()

def load_benchmark_results(path: str) -> List[CNIBenchmarkResult]:
    """Load benchmark results from JSON file."""
    try:
        with open(path, "r") as f:
            data = json.load(f)
    except (IOError, json.JSONDecodeError) as e:
        raise RuntimeError(f"Failed to load benchmark results: {e}")

    results = []
    for entry in data:
        try:
            result = CNIBenchmarkResult(
                name=entry["name"],
                version=entry["version"],
                p99_latency_us=entry["p99_latency_us"],
                throughput_gbps=entry["throughput_gbps"],
                cpu_percent=entry["cpu_percent"],
                memory_mb=entry["memory_mb"],
                node_count=entry["node_count"]
            )
            results.append(result)
        except KeyError as e:
            raise RuntimeError(f"Missing field in benchmark entry: {e}")

    return results

def print_cost_comparison(results: List[CNIBenchmarkResult]) -> None:
    """Print cost comparison table."""
    print(f"{'CNI Version':<25} {'Monthly Cost':<15} {'Savings vs Flannel':<20}")
    print("-" * 60)

    flannel = next(r for r in results if r.name == "flannel")

    for result in sorted(results, key=lambda x: x.calculate_monthly_cost()):
        cost = result.calculate_monthly_cost()
        savings = result.calculate_savings_vs(flannel)
        print(f"{result.name} {result.version:<10} ${cost:<14.2f} ${savings:<19.2f}")

def main() -> None:
    """Main entry point."""
    if len(sys.argv) != 2:
        print(f"Usage: {sys.argv[0]} ", file=sys.stderr)
        sys.exit(1)

    try:
        results = load_benchmark_results(sys.argv[1])
    except RuntimeError as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)

    print_cost_comparison(results)

if __name__ == "__main__":
    main()

import http from 'k6/http';
import { sleep } from 'k6';

export const options = {
  vus: 10000,
  duration: '14d',
  thresholds: {
    http_req_duration: ['p(99)<200'],
  },
};

export default function () {
  const res = http.get('http://kubernetes.default.svc:443');
  sleep(0.1);
}

bpf:
  mapSizeConntrack: 512000
  mapSizeNAT: 256000
  enableXDP: true
  enableL7Policy: true

calicoNetwork:
  linuxDataplane: eBPF
  bgp: Disabled
  ipam:
    type: HostLocal
  encapsulation: None