Deep Dive: How Karpenter 1.0's Just-in-Time Node Provisioning Works with AWS EC2

package main

import (
    "context"
    "fmt"
    "log"
    "strings"

    corev1 "k8s.io/api/core/v1"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/client-go/kubernetes/fake"
)

// NodePoolConstraint mirrors Karpenter's v1beta1.NodePool spec
// Source reference: https://github.com/kubernetes-sigs/karpenter/blob/main/pkg/apis/v1beta1/nodepool.go
type NodePoolConstraint struct {
    Name                string
    AllowedInstanceTypes []string
    Taints              []corev1.Taint
    Labels              map[string]string
    MinVCPU             int64
    MinMemoryMiB        int64
}

// PendingPodFilter replicates Karpenter's pkg/controllers/pod/pod.go filtering logic
// Source reference: https://github.com/kubernetes-sigs/karpenter/blob/main/pkg/controllers/pod/pod.go
func PendingPodFilter(ctx context.Context, pod *corev1.Pod, constraints []NodePoolConstraint) (*NodePoolConstraint, error) {
    if pod.Status.Phase != corev1.PodPending {
        return nil, fmt.Errorf("pod %s/%s is not pending (phase: %s)", pod.Namespace, pod.Name, pod.Status.Phase)
    }
    if pod.Spec.NodeName != "" {
        return nil, fmt.Errorf("pod %s/%s already assigned to node %s", pod.Namespace, pod.Name, pod.Spec.NodeName)
    }

    // Check for Karpenter-specific pod annotations
    _, isKarpenterManaged := pod.Annotations["karpenter.sh/provider-name"]
    if !isKarpenterManaged {
        return nil, fmt.Errorf("pod %s/%s is not managed by Karpenter", pod.Namespace, pod.Name)
    }

    // Match pod requests against NodePool constraints
    podVCPU := int64(0)
    podMemoryMiB := int64(0)
    for _, container := range pod.Spec.Containers {
        if req, ok := container.Resources.Requests[corev1.ResourceCPU]; ok {
            podVCPU += req.MilliValue() / 1000 // Convert milliCPU to whole vCPU
        }
        if req, ok := container.Resources.Requests[corev1.ResourceMemory]; ok {
            podMemoryMiB += req.Value() / (1024 * 1024) // Convert bytes to MiB
        }
    }

    for _, c := range constraints {
        // Check label selectors
        matchesLabels := true
        for k, v := range c.Labels {
            if pod.Labels[k] != v {
                matchesLabels = false
                break
            }
        }
        if !matchesLabels {
            continue
        }

        // Check taint tolerations
        toleratesTaints := true
        for _, taint := range c.Taints {
            tolerated := false
            for _, tol := range pod.Spec.Tolerations {
                if tol.Key == taint.Key && (tol.Value == taint.Value || tol.Operator == corev1.TolerationOpExists) {
                    tolerated = true
                    break
                }
            }
            if !tolerated {
                toleratesTaints = false
                break
            }
        }
        if !toleratesTaints {
            continue
        }

        // Check resource requirements
        if podVCPU < c.MinVCPU || podMemoryMiB < c.MinMemoryMiB {
            continue
        }

        return &c, nil
    }

    return nil, fmt.Errorf("no matching NodePool found for pod %s/%s", pod.Namespace, pod.Name)
}

func main() {
    // Initialize fake K8s client for demo
    clientset := fake.NewSimpleClientset()

    // Create test pod
    testPod := &corev1.Pod{
        ObjectMeta: metav1.ObjectMeta{
            Name:      "test-batch-pod",
            Namespace: "default",
            Labels:    map[string]string{"workload-type": "batch"},
            Annotations: map[string]string{"karpenter.sh/provider-name": "aws"},
        },
        Spec: corev1.PodSpec{
            Containers: []corev1.Container{
                {
                    Name:  "batch-container",
                    Image: "ubuntu:latest",
                    Resources: corev1.ResourceRequirements{
                        Requests: corev1.ResourceList{
                            corev1.ResourceCPU:    corev1.MustParse("2"),
                            corev1.ResourceMemory: corev1.MustParse("4Gi"),
                        },
                    },
                },
            },
            Tolerations: []corev1.Toleration{
                {Key: "batch-workload", Operator: corev1.TolerationOpExists},
            },
        },
        Status: corev1.PodStatus{Phase: corev1.PodPending},
    }

    // Define NodePool constraints
    constraints := []NodePoolConstraint{
        {
            Name:                "batch-nodepool",
            AllowedInstanceTypes: []string{"m5.large", "m5.xlarge", "c6i.2xlarge"},
            Taints:              []corev1.Taint{{Key: "batch-workload", Value: "true", Effect: corev1.TaintEffectNoSchedule}},
            Labels:              map[string]string{"workload-type": "batch"},
            MinVCPU:             2,
            MinMemoryMiB:        4096,
        },
    }

    // Run filter
    matchingPool, err := PendingPodFilter(context.Background(), testPod, constraints)
    if err != nil {
        log.Fatalf("Filter failed: %v", err)
    }

    fmt.Printf("Pod %s/%s matched NodePool: %s\n", testPod.Namespace, testPod.Name, matchingPool.Name)
}

package main

import (
    "encoding/json"
    "fmt"
    "log"
    "math"

    "github.com/aws/aws-sdk-go/aws"
    "github.com/aws/aws-sdk-go/aws/session"
    "github.com/aws/aws-sdk-go/service/ec2"
)

// EC2Instance mirrors Karpenter's pkg/cloudprovider/aws/instance_type.go type
// Source reference: https://github.com/kubernetes-sigs/karpenter/blob/main/pkg/cloudprovider/aws/instance_type.go
type EC2Instance struct {
    InstanceType   string
    VCPU           int64
    MemoryMiB      int64
    StorageGB      int64
    OnDemandPrice  float64
    SpotPrice      float64
    Accelerators   []string
}

// BinpackResult holds the optimal instance for a set of pending pods
type BinpackResult struct {
    Instance      EC2Instance
    PodCount      int
    UtilizedVCPU  int64
    UtilizedMemMiB int64
    CostPerPod    float64
}

// KarpenterBinpacker replicates the binpacking logic from pkg/scheduling/binpack.go
// Implements first-fit decreasing algorithm for pod-to-instance matching
func KarpenterBinpacker(pendingPods []PodRequest, availableInstances []EC2Instance, useSpot bool) (*BinpackResult, error) {
    if len(pendingPods) == 0 {
        return nil, fmt.Errorf("no pending pods to binpack")
    }
    if len(availableInstances) == 0 {
        return nil, fmt.Errorf("no available EC2 instances to provision")
    }

    // Sort pods by vCPU request descending (first-fit decreasing)
    sortedPods := make([]PodRequest, len(pendingPods))
    copy(sortedPods, pendingPods)
    for i := 0; i < len(sortedPods)-1; i++ {
        for j := i+1; j < len(sortedPods); j++ {
            if sortedPods[i].VCPU < sortedPods[j].VCPU {
                sortedPods[i], sortedPods[j] = sortedPods[j], sortedPods[i]
            }
        }
    }

    // Sort instances by price ascending (prefer cheaper first)
    sortedInstances := make([]EC2Instance, len(availableInstances))
    copy(sortedInstances, availableInstances)
    for i := 0; i < len(sortedInstances)-1; i++ {
        for j := i+1; j < len(sortedInstances); j++ {
            priceI := sortedInstances[i].OnDemandPrice
            priceJ := sortedInstances[j].OnDemandPrice
            if useSpot {
                priceI = sortedInstances[i].SpotPrice
                priceJ = sortedInstances[j].SpotPrice
            }
            if priceI > priceJ {
                sortedInstances[i], sortedInstances[j] = sortedInstances[j], sortedInstances[i]
            }
        }
    }

    // Track remaining capacity per instance (we simulate a single instance for simplicity)
    bestInstance := EC2Instance{}
    bestPodCount := 0
    bestUtilVCPU := int64(0)
    bestUtilMem := int64(0)

    for _, instance := range sortedInstances {
        remainingVCPU := instance.VCPU
        remainingMem := instance.MemoryMiB
        podCount := 0
        utilVCPU := int64(0)
        utilMem := int64(0)

        for _, pod := range sortedPods {
            if pod.VCPU <= remainingVCPU && pod.MemoryMiB <= remainingMem {
                remainingVCPU -= pod.VCPU
                remainingMem -= pod.MemoryMiB
                podCount++
                utilVCPU += pod.VCPU
                utilMem += pod.MemoryMiB
            }
        }

        // Prefer instance that fits more pods, then lower price
        if podCount > bestPodCount || (podCount == bestPodCount && (instance.OnDemandPrice < bestInstance.OnDemandPrice)) {
            bestInstance = instance
            bestPodCount = podCount
            bestUtilVCPU = utilVCPU
            bestUtilMem = utilMem
        }
    }

    if bestPodCount == 0 {
        return nil, fmt.Errorf("no instance can fit any pending pods")
    }

    price := bestInstance.OnDemandPrice
    if useSpot {
        price = bestInstance.SpotPrice
    }
    costPerPod := math.Round((price / float64(bestPodCount)) * 100) / 100

    return &BinpackResult{
        Instance:      bestInstance,
        PodCount:      bestPodCount,
        UtilizedVCPU:  bestUtilVCPU,
        UtilizedMemMiB: bestUtilMem,
        CostPerPod:    costPerPod,
    }, nil
}

// PodRequest represents a pending pod's resource requirements
type PodRequest struct {
    Name      string
    VCPU      int64
    MemoryMiB int64
}

func main() {
    // Mock EC2 instance data (real data fetched from AWS EC2 API in production)
    availableInstances := []EC2Instance{
        {InstanceType: "m5.large", VCPU: 2, MemoryMiB: 8192, OnDemandPrice: 0.096, SpotPrice: 0.028, StorageGB: 20},
        {InstanceType: "m5.xlarge", VCPU: 4, MemoryMiB: 16384, OnDemandPrice: 0.192, SpotPrice: 0.056, StorageGB: 40},
        {InstanceType: "c6i.2xlarge", VCPU: 8, MemoryMiB: 16384, OnDemandPrice: 0.34, SpotPrice: 0.10, StorageGB: 40},
        {InstanceType: "r6i.large", VCPU: 2, MemoryMiB: 16384, OnDemandPrice: 0.126, SpotPrice: 0.038, StorageGB: 20},
    }

    // Mock pending pods (from Karpenter's pod watcher)
    pendingPods := []PodRequest{
        {Name: "batch-pod-1", VCPU: 2, MemoryMiB: 4096},
        {Name: "batch-pod-2", VCPU: 2, MemoryMiB: 4096},
        {Name: "batch-pod-3", VCPU: 2, MemoryMiB: 4096},
        {Name: "web-pod-1", VCPU: 1, MemoryMiB: 2048},
    }

    // Run binpacker with spot instances
    result, err := KarpenterBinpacker(pendingPods, availableInstances, true)
    if err != nil {
        log.Fatalf("Binpacking failed: %v", err)
    }

    fmt.Printf("Optimal Instance: %s\n", result.Instance.InstanceType)
    fmt.Printf("Fits %d pods\n", result.PodCount)
    fmt.Printf("VCPU Utilization: %d/%d (%.2f%%)\n", result.UtilizedVCPU, result.Instance.VCPU, float64(result.UtilizedVCPU)/float64(result.Instance.VCPU)*100)
    fmt.Printf("Memory Utilization: %d/%d MiB (%.2f%%)\n", result.UtilizedMemMiB, result.Instance.MemoryMiB, float64(result.UtilizedMemMiB)/float64(result.Instance.MemoryMiB)*100)
    fmt.Printf("Cost per pod (spot): $%.2f\n", result.CostPerPod)

    // Output as JSON for integration
    jsonResult, _ := json.MarshalIndent(result, "", "  ")
    fmt.Println(string(jsonResult))
}

package main

import (
    "context"
    "encoding/base64"
    "fmt"
    "log"
    "strings"

    "github.com/aws/aws-sdk-go/aws"
    "github.com/aws/aws-sdk-go/aws/session"
    "github.com/aws/aws-sdk-go/service/ec2"
)

// KarpenterUserData generates the cloud-init userdata for Karpenter-provisioned nodes
// Matches logic from pkg/cloudprovider/aws/launch_template.go
// Source reference: https://github.com/kubernetes-sigs/karpenter/blob/main/pkg/cloudprovider/aws/launch_template.go
func KarpenterUserData(clusterName string, kubeletArgs []string, nodeLabels map[string]string) string {
    labelsStr := ""
    for k, v := range nodeLabels {
        labelsStr += fmt.Sprintf("    - \"%s=%s\"\n", k, v)
    }

    kubeletArgsStr := ""
    for _, arg := range kubeletArgs {
        kubeletArgsStr += fmt.Sprintf("    - \"%s\"\n", arg)
    }

    userData := fmt.Sprintf(`#cloud-config
package_update: true
packages:
  - awscli
  - kubelet
  - kubectl
  - containerd

write_files:
  - path: /etc/kubernetes/kubelet.conf
    content: |
      apiVersion: kubelet.config.k8s.io/v1beta1
      kind: KubeletConfiguration
      clusterDNS:
        - "10.100.0.10"
      clusterDomain: "cluster.local"
      nodeLabels:
%s
      kubeletArgs:
%s

  - path: /etc/systemd/system/kubelet.service
    content: |
      [Unit]
      Description=Kubernetes Kubelet
      After=containerd.service

      [Service]
      ExecStart=/usr/bin/kubelet --config=/etc/kubernetes/kubelet.conf --kubeconfig=/etc/kubernetes/kubelet.kubeconfig
      Restart=always

      [Install]
      WantedBy=multi-user.target

runcmd:
  - systemctl enable containerd
  - systemctl start containerd
  - systemctl enable kubelet
  - systemctl start kubelet
  - aws ec2 create-tags --resources $(curl -s http://169.254.169.254/latest/meta-data/instance-id) --tags Key=karpenter.sh/cluster,Value=%s Key=karpenter.sh/node-pool,Value=batch-nodepool
`, labelsStr, kubeletArgsStr, clusterName)

    return base64.StdEncoding.EncodeToString([]byte(userData))
}

// ProvisionEC2Node replicates Karpenter's EC2 provisioning logic from pkg/cloudprovider/aws/cloud_provider.go
// Source reference: https://github.com/kubernetes-sigs/karpenter/blob/main/pkg/cloudprovider/aws/cloud_provider.go
func ProvisionEC2Node(ctx context.Context, sess *session.Session, input *ec2.RunInstancesInput) (*ec2.Instance, error) {
    svc := ec2.New(sess)

    // Validate input
    if input.InstanceType == nil || *input.InstanceType == "" {
        return nil, fmt.Errorf("instance type is required")
    }
    if input.MinCount == nil || *input.MinCount < 1 {
        return nil, fmt.Errorf("min count must be at least 1")
    }

    // Add Karpenter-specific tags to all instances
    if input.TagSpecifications == nil {
        input.TagSpecifications = []*ec2.TagSpecification{}
    }
    input.TagSpecifications = append(input.TagSpecifications, &ec2.TagSpecification{
        ResourceType: aws.String("instance"),
        Tags: []*ec2.Tag{
            {Key: aws.String("karpenter.sh/cluster"), Value: aws.String("my-eks-cluster")},
            {Key: aws.String("karpenter.sh/managed"), Value: aws.String("true")},
            {Key: aws.String("Name"), Value: aws.String("karpenter-provisioned-node")},
        },
    })

    // Run instances
    result, err := svc.RunInstancesWithContext(ctx, input)
    if err != nil {
        return nil, fmt.Errorf("failed to run EC2 instances: %w", err)
    }

    if len(result.Instances) == 0 {
        return nil, fmt.Errorf("no instances returned from RunInstances")
    }

    // Wait for instance to be running (simplified, production uses waiter)
    describeInput := &ec2.DescribeInstancesInput{
        InstanceIds: []*string{result.Instances[0].InstanceId},
    }
    describeResult, err := svc.DescribeInstancesWithContext(ctx, describeInput)
    if err != nil {
        return nil, fmt.Errorf("failed to describe instance: %w", err)
    }

    return describeResult.Reservations[0].Instances[0], nil
}

func main() {
    // Initialize AWS session
    sess, err := session.NewSession(&aws.Config{
        Region: aws.String("us-east-1"),
    })
    if err != nil {
        log.Fatalf("Failed to create AWS session: %v", err)
    }

    // Generate userdata
    userData := KarpenterUserData(
        "my-eks-cluster",
        []string{"--max-pods=110", "--cgroup-driver=systemd"},
        map[string]string{
            "karpenter.sh/cluster": "my-eks-cluster",
            "workload-type": "batch",
            "node.kubernetes.io/instance-type": "m5.xlarge",
        },
    )

    // Define RunInstances input
    runInput := &ec2.RunInstancesInput{
        InstanceType:     aws.String("m5.xlarge"),
        MinCount:         aws.Int64(1),
        MaxCount:         aws.Int64(1),
        ImageId:          aws.String("ami-0abcdef1234567890"), // EKS optimized AMI
        SubnetId:         aws.String("subnet-0123456789abcdef0"),
        SecurityGroupIds: []*string{aws.String("sg-0123456789abcdef0")},
        UserData:         aws.String(userData),
        BlockDeviceMappings: []*ec2.BlockDeviceMapping{
            {
                DeviceName: aws.String("/dev/sda1"),
                Ebs: &ec2.EbsBlockDevice{
                    VolumeSize: aws.Int64(100),
                    VolumeType: aws.String("gp3"),
                    Iops:       aws.Int64(3000),
                },
            },
        },
    }

    // Provision node
    instance, err := ProvisionEC2Node(context.Background(), sess, runInput)
    if err != nil {
        log.Fatalf("Provisioning failed: %v", err)
    }

    fmt.Printf("Provisioned EC2 instance: %s\n", *instance.InstanceId)
    fmt.Printf("Instance state: %s\n", *instance.State.Name)
    fmt.Printf("Private IP: %s\n", *instance.PrivateIpAddress)
}

apiVersion: karpenter.sh/v1beta1
kind: NodePool
metadata:
  name: web-nodepool
spec:
  disruption:
    maxUnavailable: "5%"
    consolidationPolicy: WhenEmpty
  template:
    spec:
      requirements:
        - key: karpenter.sh/capacity-type
          operator: In
          values: ["on-demand"]
        - key: node.kubernetes.io/instance-family
          operator: In
          values: ["m5", "c6i"]
      nodeClassRef:
        name: web-nodeclass

// Fetch Spot Placement Score for m5.xlarge in us-east-1a
func GetSpotPlacementScore(sess *session.Session, instanceType string) (int, error) {
    svc := ec2.New(sess)
    input := &ec2.GetSpotPlacementScoresInput{
        InstanceTypes: []*string{aws.String(instanceType)},
        TargetRegion:  aws.String("us-east-1"),
        AvailabilityZone: aws.String("us-east-1a"),
    }
    result, err := svc.GetSpotPlacementScores(input)
    if err != nil {
        return 0, err
    }
    if len(result.SpotPlacementScores) == 0 {
        return 0, fmt.Errorf("no spot placement score for %s", instanceType)
    }
    return int(*result.SpotPlacementScores[0].Score), nil
}

# Prometheus query for p99 instance launch time over 7 days
histogram_quantile(0.99, 
  sum(rate(karpenter_cloudprovider_instance_launch_time_seconds_bucket[7d])) by (le, nodepool)
)