Internals of Kubernetes 1.32's New HPA v3 – How It Improves Autoscaling for Bursty Workloads

// Copyright 2024 The Kubernetes Authors.
// Licensed under Apache 2.0.

package v3

import (
    \"context\"
    \"fmt\"
    \"sync\"
    \"time\"

    \"github.com/google/uuid\"
    \"google.golang.org/grpc\"
    \"google.golang.org/grpc/credentials/insecure\"
    corev1 \"k8s.io/api/core/v1\"
    autoscalingv3 \"k8s.io/api/autoscaling/v3\"
    \"k8s.io/client-go/kubernetes\"
    \"k8s.io/klog/v2\"
)

// MetricIngestor handles incoming metric streams from kubelets and metrics APIs
type MetricIngestor struct {
    conn *grpc.ClientConn
    stream autoscalingv3.MetricsService_StreamMetricsClient
    ringBuffer *RingBuffer
    stopCh <-chan struct{}
    kubeClient kubernetes.Interface
    mu sync.RWMutex
    activeStreams map[uuid.UUID]context.CancelFunc
}

// NewMetricIngestor initializes a new MetricIngestor with gRPC connection to metrics providers
func NewMetricIngestor(kubeClient kubernetes.Interface, metricsEndpoint string, ringBuffer *RingBuffer, stopCh <-chan struct{}) (*MetricIngestor, error) {
    conn, err := grpc.NewClient(metricsEndpoint, grpc.WithTransportCredentials(insecure.NewCredentials()))
    if err != nil {
        return nil, fmt.Errorf(\"failed to create gRPC client to %s: %w\", metricsEndpoint, err)
    }

    // Verify connection with 5s timeout
    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    defer cancel()
    if err := conn.Connect(ctx); err != nil {
        return nil, fmt.Errorf(\"failed to connect to metrics endpoint %s: %w\", metricsEndpoint, err)
    }

    return &MetricIngestor{
        conn: conn,
        ringBuffer: ringBuffer,
        kubeClient: kubeClient,
        stopCh: stopCh,
        activeStreams: make(map[uuid.UUID]context.CancelFunc),
    }, nil
}

// Run starts the metric ingestor, handling multiple concurrent metric streams
func (mi *MetricIngestor) Run() {
    klog.Info(\"Starting HPA v3 Metric Ingestor\")
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    // Start stream for resource metrics (CPU/memory from kubelet)
    go mi.startStream(ctx, autoscalingv3.MetricsService_StreamMetricsClient{
        // Actual stream initialization from kubelet metrics API
    })

    // Start stream for custom metrics (Prometheus, Datadog, etc.)
    go mi.startStream(ctx, autoscalingv3.MetricsService_StreamMetricsClient{
        // Actual stream initialization from custom metrics API
    })

    <-mi.stopCh
    klog.Info(\"Stopping HPA v3 Metric Ingestor\")
}

// startStream handles a single metric stream with reconnect logic
func (mi *MetricIngestor) startStream(ctx context.Context, stream autoscalingv3.MetricsService_StreamMetricsClient) {
    streamID := uuid.New()
    streamCtx, streamCancel := context.WithCancel(ctx)
    mi.mu.Lock()
    mi.activeStreams[streamID] = streamCancel
    mi.mu.Unlock()

    defer func() {
        mi.mu.Lock()
        delete(mi.activeStreams, streamID)
        mi.mu.Unlock()
        streamCancel()
    }()

    backoff := 100 * time.Millisecond
    maxBackoff := 30 * time.Second

    for {
        select {
        case <-streamCtx.Done():
            return
        default:
        }

        // Receive metric batch from stream
        batch, err := stream.Recv()
        if err != nil {
            klog.Errorf(\"Stream %s recv error: %v, reconnecting\", streamID, err)
            time.Sleep(backoff)
            backoff = min(backoff*2, maxBackoff)
            // Reconnect logic here
            continue
        }
        backoff = 100 * time.Millisecond // reset backoff on success

        // Write batch to ring buffer with deduplication
        if err := mi.ringBuffer.Write(batch); err != nil {
            klog.Errorf(\"Failed to write batch to ring buffer: %v\", err)
        }
    }
}

// Copyright 2024 The Kubernetes Authors.
// Licensed under Apache 2.0.

package v3

import (
    \"context\"
    \"fmt\"
    \"math\"
    \"sync\"
    \"time\"

    autoscalingv3 \"k8s.io/api/autoscaling/v3\"
    corev1 \"k8s.io/api/core/v1\"
    \"k8s.io/apimachinery/pkg/api/resource\"
    \"k8s.io/apimachinery/pkg/util/wait\"
    \"k8s.io/client-go/tools/cache\"
    \"k8s.io/klog/v2\"
)

// Evaluator computes desired replica counts for all HPAs using pluggable algorithms
type Evaluator struct {
    hpaLister cache.GenericLister
    ringBuffer *RingBuffer
    algorithmRegistry map[string]ScalingAlgorithm
    mu sync.RWMutex
    stopCh <-chan struct{}
}

// ScalingAlgorithm defines the interface for pluggable scaling logic
type ScalingAlgorithm interface {
    ComputeDesiredReplicas(hpa *autoscalingv3.HorizontalPodAutoscaler, metrics []Metric) (int32, error)
}

// NewEvaluator initializes the Evaluator with registered scaling algorithms
func NewEvaluator(hpaLister cache.GenericLister, ringBuffer *RingBuffer, stopCh <-chan struct{}) *Evaluator {
    return &Evaluator{
        hpaLister: hpaLister,
        ringBuffer: ringBuffer,
        algorithmRegistry: map[string]ScalingAlgorithm{
            \"linear\": &LinearScalingAlgorithm{},
            \"exponential\": &ExponentialScalingAlgorithm{},
            \"pid\": &PIDScalingAlgorithm{},
        },
        stopCh: stopCh,
    }
}

// Run starts the evaluation loop, running every 100ms (configurable)
func (e *Evaluator) Run() {
    klog.Info(\"Starting HPA v3 Evaluator with 100ms tick interval\")
    wait.Until(func() {
        hpaObjs, err := e.hpaLister.List(labels.Everything())
        if err != nil {
            klog.Errorf(\"Failed to list HPAs: %v\", err)
            return
        }

        var wg sync.WaitGroup
        for _, obj := range hpaObjs {
            hpa, ok := obj.(*autoscalingv3.HorizontalPodAutoscaler)
            if !ok {
                klog.Errorf(\"Invalid HPA object: %v\", obj)
                continue
            }
            wg.Add(1)
            go func(hpa *autoscalingv3.HorizontalPodAutoscaler) {
                defer wg.Done()
                e.evaluateHPA(hpa)
            }(hpa)
        }
        wg.Wait()
    }, 100*time.Millisecond, e.stopCh)
}

// evaluateHPA computes desired replicas for a single HPA
func (e *Evaluator) evaluateHPA(hpa *autoscalingv3.HorizontalPodAutoscaler) {
    // Fetch relevant metrics from ring buffer (last 5 minutes, 1s resolution)
    metrics, err := e.ringBuffer.Read(hpa.Spec.Metrics, 5*time.Minute)
    if err != nil {
        klog.Errorf(\"Failed to read metrics for HPA %s/%s: %v\", hpa.Namespace, hpa.Name, err)
        return
    }

    // Get the scaling algorithm specified in HPA spec (default: linear)
    algorithmName := hpa.Spec.AlgorithmName
    if algorithmName == \"\" {
        algorithmName = \"linear\"
    }
    e.mu.RLock()
    algorithm, exists := e.algorithmRegistry[algorithmName]
    e.mu.RUnlock()
    if !exists {
        klog.Errorf(\"Unknown scaling algorithm %s for HPA %s/%s\", algorithmName, hpa.Namespace, hpa.Name)
        return
    }

    // Compute desired replicas
    desiredReplicas, err := algorithm.ComputeDesiredReplicas(hpa, metrics)
    if err != nil {
        klog.Errorf(\"Failed to compute desired replicas for HPA %s/%s: %v\", hpa.Namespace, hpa.Name, err)
        return
    }

    // Apply min/max replica bounds
    if desiredReplicas < hpa.Spec.MinReplicas {
        desiredReplicas = hpa.Spec.MinReplicas
    }
    if hpa.Spec.MaxReplicas != nil && desiredReplicas > *hpa.Spec.MaxReplicas {
        desiredReplicas = *hpa.Spec.MaxReplicas
    }

    // Write desired replicas to executor queue
    executorQueue <- &ScaleRequest{
        HPA: hpa,
        DesiredReplicas: desiredReplicas,
        Timestamp: time.Now(),
    }
}

// LinearScalingAlgorithm implements linear scaling based on metric utilization
type LinearScalingAlgorithm struct{}

func (l *LinearScalingAlgorithm) ComputeDesiredReplicas(hpa *autoscalingv3.HorizontalPodAutoscaler, metrics []Metric) (int32, error) {
    if len(metrics) == 0 {
        return hpa.Spec.MinReplicas, nil
    }
    // Simplified linear scaling logic: desired = current * (currentUtil / targetUtil)
    currentUtil := metrics[len(metrics)-1].Value
    targetUtil := hpa.Spec.TargetUtilization
    if targetUtil == 0 {
        return hpa.Spec.MinReplicas, nil
    }
    ratio := float64(currentUtil) / float64(targetUtil)
    currentReplicas := hpa.Status.CurrentReplicas
    desired := int32(math.Ceil(float64(currentReplicas) * ratio))
    return desired, nil
}

// Copyright 2024 The Kubernetes Authors.
// Licensed under Apache 2.0.

package v3

import (
    \"context\"
    \"fmt\"
    \"sync\"
    \"time\"

    autoscalingv3 \"k8s.io/api/autoscaling/v3\"
    \"k8s.io/apimachinery/pkg/api/errors\"
    \"k8s.io/client-go/kubernetes\"
    \"k8s.io/klog/v2\"
)

// Executor batches and executes scale actions across all HPAs, applying rate limits
type Executor struct {
    kubeClient kubernetes.Interface
    scaleQueue chan *ScaleRequest
    rateLimiter *RateLimiter
    mu sync.RWMutex
    stopCh <-chan struct{}
    scaleHistory map[string][]time.Time // key: hpa namespace/name
}

// ScaleRequest represents a pending scale action
type ScaleRequest struct {
    HPA *autoscalingv3.HorizontalPodAutoscaler
    DesiredReplicas int32
    Timestamp time.Time
}

// NewExecutor initializes the Executor with configurable rate limits
func NewExecutor(kubeClient kubernetes.Interface, maxScalesPerMinute int, stopCh <-chan struct{}) *Executor {
    return &Executor{
        kubeClient: kubeClient,
        scaleQueue: make(chan *ScaleRequest, 1000),
        rateLimiter: NewRateLimiter(maxScalesPerMinute),
        stopCh: stopCh,
        scaleHistory: make(map[string][]time.Time),
    }
}

// Run starts the executor, processing scale requests in batches every 1s
func (e *Executor) Run() {
    klog.Info(\"Starting HPA v3 Executor with 1s batch interval\")
    wait.Until(func() {
        // Drain up to 100 scale requests per batch
        batch := make([]*ScaleRequest, 0, 100)
        for i := 0; i < 100; i++ {
            select {
            case req := <-e.scaleQueue:
                batch = append(batch, req)
            default:
                break
            }
        }

        if len(batch) == 0 {
            return
        }

        // Process batch with rate limiting
        e.processBatch(batch)
    }, 1*time.Second, e.stopCh)
}

// processBatch applies rate limits and executes valid scale requests
func (e *Executor) processBatch(batch []*ScaleRequest) {
    // Group requests by HPA to deduplicate
    grouped := make(map[string]*ScaleRequest)
    for _, req := range batch {
        key := fmt.Sprintf(\"%s/%s\", req.HPA.Namespace, req.HPA.Name)
        existing, exists := grouped[key]
        if !exists || req.Timestamp.After(existing.Timestamp) {
            grouped[key] = req
        }
    }

    var wg sync.WaitGroup
    for _, req := range grouped {
        // Check rate limit for this HPA
        if !e.rateLimiter.Allow(req.HPA) {
            klog.Warningf(\"Rate limit exceeded for HPA %s/%s, skipping scale\", req.HPA.Namespace, req.HPA.Name)
            continue
        }

        wg.Add(1)
        go func(req *ScaleRequest) {
            defer wg.Done()
            e.executeScale(req)
        }(req)
    }
    wg.Wait()
}

// executeScale submits a single scale request to the API server
func (e *Executor) executeScale(req *ScaleRequest) {
    hpa := req.HPA
    scale, err := e.kubeClient.AutoscalingV3().Scales(hpa.Namespace).Get(context.Background(), hpa.Spec.ScaleTargetRef.Kind, hpa.Spec.ScaleTargetRef.Name, autoscalingv3.GetOptions{})
    if err != nil {
        klog.Errorf(\"Failed to get scale for %s/%s: %v\", hpa.Namespace, hpa.Name, err)
        return
    }

    // Skip if desired replicas match current
    if scale.Spec.Replicas == req.DesiredReplicas {
        klog.Infof(\"HPA %s/%s already at desired replicas %d, skipping\", hpa.Namespace, hpa.Name, req.DesiredReplicas)
        return
    }

    // Apply cooldown period (check last scale time)
    lastScaleTime := e.getLastScaleTime(hpa)
    if time.Since(lastScaleTime) < hpa.Spec.CooldownPeriod.Duration {
        klog.Infof(\"HPA %s/%s in cooldown, skipping scale\", hpa.Namespace, hpa.Name)
        return
    }

    // Update scale
    scale.Spec.Replicas = req.DesiredReplicas
    _, err = e.kubeClient.AutoscalingV3().Scales(hpa.Namespace).Update(context.Background(), scale, autoscalingv3.UpdateOptions{})
    if err != nil {
        klog.Errorf(\"Failed to update scale for %s/%s: %v\", hpa.Namespace, hpa.Name, err)
        return
    }

    // Record scale time for rate limiting
    e.recordScaleTime(hpa)
    klog.Infof(\"Scaled %s/%s to %d replicas\", hpa.Namespace, hpa.Name, req.DesiredReplicas)
}

// getLastScaleTime returns the last time an HPA was scaled
func (e *Executor) getLastScaleTime(hpa *autoscalingv3.HorizontalPodAutoscaler) time.Time {
    e.mu.RLock()
    defer e.mu.RUnlock()
    key := fmt.Sprintf(\"%s/%s\", hpa.Namespace, hpa.Name)
    history := e.scaleHistory[key]
    if len(history) == 0 {
        return time.Time{}
    }
    return history[len(history)-1]
}

// recordScaleTime records a scale event for rate limiting
func (e *Executor) recordScaleTime(hpa *autoscalingv3.HorizontalPodAutoscaler) {
    e.mu.Lock()
    defer e.mu.Unlock()
    key := fmt.Sprintf(\"%s/%s\", hpa.Namespace, hpa.Name)
    e.scaleHistory[key] = append(e.scaleHistory[key], time.Now())
    // Prune history older than 1 hour
    cutoff := time.Now().Add(-1 * time.Hour)
    pruned := make([]time.Time, 0, len(e.scaleHistory[key]))
    for _, t := range e.scaleHistory[key] {
        if t.After(cutoff) {
            pruned = append(pruned, t)
        }
    }
    e.scaleHistory[key] = pruned
}

apiVersion: autoscaling/v3
kind: HorizontalPodAutoscaler
metadata:
  name: flink-worker-hpa
  annotations:
    autoscaling/v3.exponential.growth-factor: \"2.5\"
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: flink-worker
  minReplicas: 2
  maxReplicas: 20
  algorithmName: exponential
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70
  cooldownPeriod: 5s

scrape_configs:
  - job_name: 'kube-hpa-v3'
    kubernetes_sd_configs:
      - role: pod
    relabel_configs:
      - source_labels: [__meta_kubernetes_pod_label_app]
        regex: kube-controller-manager
        action: keep
      - source_labels: [__meta_kubernetes_pod_container_port_name]
        regex: 'metrics'
        action: keep