github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2026-07-17 20:00:07 +00:00
Code Issues Projects Releases Wiki Activity

fix: deterministic CPU load distribution for HPAConfigurableTolerance e2e test

Browse Source
This commit is contained in:
bishal7679
2026-03-12 15:03:55 +05:30
parent 3a56126347
commit 260c856fc2
3 changed files with 78 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
test/e2e/autoscaling/horizontal_pod_autoscaling_behavior.go
View File

@@ -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))

80
test/e2e/framework/autoscaling/autoscaling_utils.go
View File

@@ -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 {

8
test/images/agnhost/resource-consumer-controller/controller.go
View File

@@ -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) {