From 260c856fc296564677adfa9d211460162e9b6f01 Mon Sep 17 00:00:00 2001 From: bishal7679 Date: Thu, 12 Mar 2026 15:03:55 +0530 Subject: [PATCH] fix: deterministic CPU load distribution for HPAConfigurableTolerance e2e test --- .../horizontal_pod_autoscaling_behavior.go | 5 +- .../autoscaling/autoscaling_utils.go | 80 ++++++++++++++++--- .../controller.go | 8 +- 3 files changed, 78 insertions(+), 15 deletions(-) diff --git a/test/e2e/autoscaling/horizontal_pod_autoscaling_behavior.go b/test/e2e/autoscaling/horizontal_pod_autoscaling_behavior.go index 18cfab03901..36ba5664a89 100644 --- a/test/e2e/autoscaling/horizontal_pod_autoscaling_behavior.go +++ b/test/e2e/autoscaling/horizontal_pod_autoscaling_behavior.go @@ -576,8 +576,11 @@ var _ = SIGDescribe(feature.HPA, framework.WithSlow(), framework.WithFeatureGate ginkgo.By("waiting for deployment to start initial pods") rc.WaitForReplicas(ctx, initPods, waitDeadline) - // Set initial stable load using even per-pod distribution. + // Set initial stable load using even per-pod distribution and wait for + // HPA to observe it before triggering scale-up. This ensures HPA has a + // baseline measurement before we push load above the scale-up tolerance. rc.ConsumeCPUPerPod(initCPUUsageTotal) + rc.EnsureDesiredReplicasInRange(ctx, initPods, initPods, maxHPAReactionTime+maxResourceConsumerDelay, hpa.Name) ginkgo.By("trying to trigger scale up to 11 replicas") rc.ConsumeCPUPerPod(usageForReplicasWithRequest(11, podCPURequest, targetCPUUtilizationPercent)) diff --git a/test/e2e/framework/autoscaling/autoscaling_utils.go b/test/e2e/framework/autoscaling/autoscaling_utils.go index 1e35d61824a..60f167a0df6 100644 --- a/test/e2e/framework/autoscaling/autoscaling_utils.go +++ b/test/e2e/framework/autoscaling/autoscaling_utils.go @@ -655,31 +655,91 @@ func (rc *ResourceConsumer) makeConsumeCPUPerPodRequests(ctx context.Context) { rc.stopWaitGroup.Add(1) defer rc.stopWaitGroup.Done() tick := time.After(time.Duration(0)) - milliCoresPerPod := 0 + millicoresTotal := 0 for { select { - case milliCoresPerPod = <-rc.cpuPerPod: - if milliCoresPerPod != 0 { - framework.Logf("RC %s setting per-pod CPU to %v millicores", rc.name, milliCoresPerPod) + case millicoresTotal = <-rc.cpuPerPod: + if millicoresTotal != 0 { + framework.Logf("RC %s: setting per-pod CPU to %v millicores total", rc.name, millicoresTotal) } else { - framework.Logf("RC %s disabling per-pod CPU consumption", rc.name) + framework.Logf("RC %s: disabling per-pod CPU consumption", rc.name) } case <-tick: - if milliCoresPerPod != 0 { - framework.Logf("RC %s sending per-pod CPU request: %d millicores total", rc.name, milliCoresPerPod) - rc.sendConsumeCPURequest(ctx, milliCoresPerPod) + if millicoresTotal != 0 { + framework.Logf("RC %s: sending per-pod CPU request: %d millicores total", rc.name, millicoresTotal) + rc.sendConsumeCPUPerPodRequest(ctx, millicoresTotal) } tick = time.After(rc.sleepTime) case <-ctx.Done(): - framework.Logf("RC %s stopping per-pod CPU consumer: %v", rc.name, ctx.Err()) + framework.Logf("RC %s: stopping per-pod CPU consumer: %v", rc.name, ctx.Err()) return case <-rc.stopCPUPerPod: - framework.Logf("RC %s stopping per-pod CPU consumer", rc.name) + framework.Logf("RC %s: stopping per-pod CPU consumer", rc.name) return } } } +// sendConsumeCPUPerPodRequest distributes CPU load evenly across all running +// pods by sending requests directly via the Kubernetes pod proxy API. This +// bypasses kube-proxy load balancing, guaranteeing each pod receives exactly +// its share. Falls back to sendConsumeCPURequest if pod listing fails. +func (rc *ResourceConsumer) sendConsumeCPUPerPodRequest(ctx context.Context, millicoresTotal int) { + pods, err := rc.clientSet.CoreV1().Pods(rc.nsName).List(ctx, metav1.ListOptions{ + LabelSelector: fmt.Sprintf("name=%s", rc.name), + }) + if err != nil { + framework.Logf("ConsumeCPUPerPod: failed to list pods: %v, falling back to service proxy", err) + rc.sendConsumeCPURequest(ctx, millicoresTotal) + return + } + + var readyPods []string + for i := range pods.Items { + if pods.Items[i].Status.Phase == v1.PodRunning { + readyPods = append(readyPods, pods.Items[i].Name) + } + } + if len(readyPods) == 0 { + framework.Logf("ConsumeCPUPerPod: no running pods, falling back to service proxy") + rc.sendConsumeCPURequest(ctx, millicoresTotal) + return + } + + perPodMillicores := millicoresTotal / len(readyPods) + if perPodMillicores == 0 { + perPodMillicores = 1 + } + + framework.Logf("ConsumeCPUPerPod: distributing %d millicores across %d pods (%d per pod)", + millicoresTotal, len(readyPods), perPodMillicores) + + var wg sync.WaitGroup + for _, podName := range readyPods { + wg.Add(1) + go func(name string) { + defer wg.Done() + // Pod proxy URL: /api/v1/namespaces/{ns}/pods/{podname}:{port}/proxy/{path} + // Both service and pod proxy support the name:port format. Without an explicit + // port the API server defaults to port 80, but resource-consumer listens on + // targetPort (8080), so the port must be specified. + _, podErr := rc.clientSet.CoreV1().RESTClient().Post(). + Resource("pods"). + Namespace(rc.nsName). + Name(fmt.Sprintf("%s:%d", name, targetPort)). + SubResource("proxy"). + Suffix("ConsumeCPU"). + Param("millicores", strconv.Itoa(perPodMillicores)). + Param("durationSec", strconv.Itoa(rc.consumptionTimeInSeconds)). + DoRaw(ctx) + if podErr != nil { + framework.Logf("ConsumeCPUPerPod: error sending to pod %s: %v", name, podErr) + } + }(podName) + } + wg.Wait() +} + // GetReplicas get the replicas func (rc *ResourceConsumer) GetReplicas(ctx context.Context) (int, error) { switch rc.kind { diff --git a/test/images/agnhost/resource-consumer-controller/controller.go b/test/images/agnhost/resource-consumer-controller/controller.go index d94e7d6106c..f0241c857a3 100644 --- a/test/images/agnhost/resource-consumer-controller/controller.go +++ b/test/images/agnhost/resource-consumer-controller/controller.go @@ -157,7 +157,7 @@ func (c *controller) handleConsumeCPU(w http.ResponseWriter, query url.Values) { if perPodMillicores == 0 { perPodMillicores = 1 } - fmt.Fprintf(w, "RC manager (per-pod): sending %d millicores to each of %d pods via headless DNS\n", + _, _ = fmt.Fprintf(w, "RC manager (per-pod): sending %d millicores to each of %d pods via headless DNS\n", perPodMillicores, len(podAddrs)) c.waitGroup.Add(len(podAddrs)) for _, addr := range podAddrs { @@ -175,7 +175,7 @@ func (c *controller) handleConsumeCPU(w http.ResponseWriter, query url.Values) { log.Printf("headless DNS lookup for %s failed (%v), using ClusterIP fallback", headlessDNS, err) count := millicores / requestSizeInMillicores rest := millicores - count*requestSizeInMillicores - fmt.Fprintf(w, "RC manager: sending %v requests to consume %v millicores each and 1 request to consume %v millicores\n", + _, _ = fmt.Fprintf(w, "RC manager: sending %v requests to consume %v millicores each and 1 request to consume %v millicores\n", count, requestSizeInMillicores, rest) if count > 0 { c.waitGroup.Add(count) @@ -275,10 +275,10 @@ func (c *controller) sendOneCPURequestToURL(w http.ResponseWriter, podURL string if err != nil { c.responseWriterLock.Lock() defer c.responseWriterLock.Unlock() - fmt.Fprintf(w, "Failed to send to pod at %s: %v\n", podURL, err) + _, _ = fmt.Fprintf(w, "Failed to send to pod at %s: %v\n", podURL, err) return } - defer resp.Body.Close() + defer func() { _ = resp.Body.Close() }() } func (c *controller) sendConsumeMemRequests(w http.ResponseWriter, requests, megabytes, durationSec int) {