HPA support for pod-level resource specifications (kubernetes#132430)

* HPA support for pod-level resource specifications

* Add e2e tests for HPA support for pod-level resource specifications

Author: laoj2

pkg/controller/podautoscaler/replica_calculator.go

@@ -27,9 +27,12 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/apiserver/pkg/util/feature"
 	corelisters "k8s.io/client-go/listers/core/v1"
+	resourcehelpers "k8s.io/component-helpers/resource"
 	podutil "k8s.io/kubernetes/pkg/api/v1/pod"
 	metricsclient "k8s.io/kubernetes/pkg/controller/podautoscaler/metrics"
+	"k8s.io/kubernetes/pkg/features"
 )
 
 const (
@@ -94,7 +97,7 @@ func (c *ReplicaCalculator) GetResourceReplicas(ctx context.Context, currentRepl
 		return 0, 0, 0, time.Time{}, fmt.Errorf("did not receive metrics for targeted pods (pods might be unready)")
 	}
 
-	requests, err := calculatePodRequests(podList, container, resource)
+	requests, err := calculateRequests(podList, container, resource)
 	if err != nil {
 		return 0, 0, 0, time.Time{}, err
 	}
@@ -449,31 +452,76 @@ func groupPods(pods []*v1.Pod, metrics metricsclient.PodMetricsInfo, resource v1
 	return
 }
 
-func calculatePodRequests(pods []*v1.Pod, container string, resource v1.ResourceName) (map[string]int64, error) {
+// calculateRequests computes the request value for each pod for the specified
+// resource.
+// If container is non-empty, it uses the request of that specific container.
+// If container is empty, it uses pod-level requests if pod-level requests are
+// set on the pod. Otherwise, it sums the requests of all containers in the pod
+// (including restartable init containers).
+// It returns a map of pod names to their calculated request values.
+func calculateRequests(pods []*v1.Pod, container string, resource v1.ResourceName) (map[string]int64, error) {
+	podLevelResourcesEnabled := feature.DefaultFeatureGate.Enabled(features.PodLevelResources)
 	requests := make(map[string]int64, len(pods))
 	for _, pod := range pods {
-		podSum := int64(0)
-		// Calculate all regular containers and restartable init containers requests.
-		containers := append([]v1.Container{}, pod.Spec.Containers...)
-		for _, c := range pod.Spec.InitContainers {
-			if c.RestartPolicy != nil && *c.RestartPolicy == v1.ContainerRestartPolicyAlways {
-				containers = append(containers, c)
-			}
+		var request int64
+		var err error
+		// Determine if we should use pod-level requests: see KEP-2837
+		// https://github.com/kubernetes/enhancements/blob/master/keps/sig-node/2837-pod-level-resource-spec/README.md
+		usePodLevelRequests := podLevelResourcesEnabled &&
+			resourcehelpers.IsPodLevelRequestsSet(pod) &&
+			// If a container name is specified in the HPA, it takes precedence over
+			// the pod-level requests.
+			container == ""
+
+		if usePodLevelRequests {
+			request, err = calculatePodLevelRequests(pod, resource)
+		} else {
+			request, err = calculatePodRequestsFromContainers(pod, container, resource)
 		}
-		for _, c := range containers {
-			if container == "" || container == c.Name {
-				if containerRequest, ok := c.Resources.Requests[resource]; ok {
-					podSum += containerRequest.MilliValue()
-				} else {
-					return nil, fmt.Errorf("missing request for %s in container %s of Pod %s", resource, c.Name, pod.ObjectMeta.Name)
-				}
-			}
+		if err != nil {
+			return nil, err
 		}
-		requests[pod.Name] = podSum
+		requests[pod.Name] = request
 	}
 	return requests, nil
 }
 
+// calculatePodLevelRequests computes the requests for the specific resource at
+// the pod level.
+func calculatePodLevelRequests(pod *v1.Pod, resource v1.ResourceName) (int64, error) {
+	podLevelRequests := resourcehelpers.PodRequests(pod, resourcehelpers.PodResourcesOptions{})
+	podRequest, ok := podLevelRequests[resource]
+	if !ok {
+		return 0, fmt.Errorf("missing pod-level request for %s in Pod %s", resource, pod.Name)
+	}
+	return podRequest.MilliValue(), nil
+}
+
+// calculatePodRequestsFromContainers computes the requests for the specified
+// resource by summing requests from all containers in the pod.
+// If a container name is specified, it uses only that container.
+func calculatePodRequestsFromContainers(pod *v1.Pod, container string, resource v1.ResourceName) (int64, error) {
+	// Calculate all regular containers and restartable init containers requests.
+	containers := append([]v1.Container{}, pod.Spec.Containers...)
+	for _, c := range pod.Spec.InitContainers {
+		if c.RestartPolicy != nil && *c.RestartPolicy == v1.ContainerRestartPolicyAlways {
+			containers = append(containers, c)
+		}
+	}
+
+	request := int64(0)
+	for _, c := range containers {
+		if container == "" || container == c.Name {
+			containerRequest, ok := c.Resources.Requests[resource]
+			if !ok {
+				return 0, fmt.Errorf("missing request for %s in container %s of Pod %s", resource, c.Name, pod.Name)
+			}
+			request += containerRequest.MilliValue()
+		}
+	}
+	return request, nil
+}
+
 func removeMetricsForPods(metrics metricsclient.PodMetricsInfo, pods sets.Set[string]) {
 	for _, pod := range pods.UnsortedList() {
 		delete(metrics, pod)

pkg/controller/podautoscaler/replica_calculator_test.go

@@ -31,13 +31,16 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/sets"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/client-go/informers"
 	"k8s.io/client-go/kubernetes/fake"
 	core "k8s.io/client-go/testing"
 	"k8s.io/client-go/tools/cache"
+	featuregatetesting "k8s.io/component-base/featuregate/testing"
 	"k8s.io/kubernetes/pkg/api/legacyscheme"
 	"k8s.io/kubernetes/pkg/controller"
 	metricsclient "k8s.io/kubernetes/pkg/controller/podautoscaler/metrics"
+	"k8s.io/kubernetes/pkg/features"
 	cmapi "k8s.io/metrics/pkg/apis/custom_metrics/v1beta2"
 	emapi "k8s.io/metrics/pkg/apis/external_metrics/v1beta1"
 	metricsapi "k8s.io/metrics/pkg/apis/metrics/v1beta1"
@@ -2225,17 +2228,18 @@ func TestGroupPods(t *testing.T) {
 	}
 }
 
-func TestCalculatePodRequests(t *testing.T) {
+func TestCalculateRequests(t *testing.T) {
 	containerRestartPolicyAlways := v1.ContainerRestartPolicyAlways
 	testPod := "test-pod"
 
 	tests := []struct {
-		name             string
-		pods             []*v1.Pod
-		container        string
-		resource         v1.ResourceName
-		expectedRequests map[string]int64
-		expectedError    error
+		name                    string
+		pods                    []*v1.Pod
+		container               string
+		resource                v1.ResourceName
+		enablePodLevelResources bool
+		expectedRequests        map[string]int64
+		expectedError           error
 	}{
 		{
 			name:             "void",
@@ -2246,7 +2250,7 @@ func TestCalculatePodRequests(t *testing.T) {
 			expectedError:    nil,
 		},
 		{
-			name: "pod with regular containers",
+			name: "Sum container requests if pod-level feature is disabled",
 			pods: []*v1.Pod{{
 				ObjectMeta: metav1.ObjectMeta{
 					Name:      testPod,
@@ -2265,7 +2269,8 @@ func TestCalculatePodRequests(t *testing.T) {
 			expectedError:    nil,
 		},
 		{
-			name: "calculate requests with special container",
+			name:                    "Pod-level resources are enabled, but not set: fallback to sum container requests",
+			enablePodLevelResources: true,
 			pods: []*v1.Pod{{
 				ObjectMeta: metav1.ObjectMeta{
 					Name:      testPod,
@@ -2278,13 +2283,37 @@ func TestCalculatePodRequests(t *testing.T) {
 					},
 				},
 			}},
-			container:        "container1",
+			container:        "",
 			resource:         v1.ResourceCPU,
-			expectedRequests: map[string]int64{testPod: 100},
+			expectedRequests: map[string]int64{testPod: 150},
+			expectedError:    nil,
+		},
+		{
+			name:                    "Pod-level resources override container requests when feature enabled and pod resources specified",
+			enablePodLevelResources: true,
+			pods: []*v1.Pod{{
+
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      testPod,
+					Namespace: testNamespace,
+				},
+				Spec: v1.PodSpec{
+					Resources: &v1.ResourceRequirements{
+						Requests: v1.ResourceList{v1.ResourceCPU: *resource.NewMilliQuantity(800, resource.DecimalSI)},
+					},
+					Containers: []v1.Container{
+						{Name: "container1", Resources: v1.ResourceRequirements{Requests: v1.ResourceList{v1.ResourceCPU: *resource.NewMilliQuantity(100, resource.DecimalSI)}}},
+						{Name: "container2", Resources: v1.ResourceRequirements{Requests: v1.ResourceList{v1.ResourceCPU: *resource.NewMilliQuantity(50, resource.DecimalSI)}}},
+					},
+				},
+			}},
+			container:        "",
+			resource:         v1.ResourceCPU,
+			expectedRequests: map[string]int64{testPod: 800},
 			expectedError:    nil,
 		},
 		{
-			name: "container missing requests",
+			name: "Fail if at least one of the containers is missing requests and pod-level feature/requests are not set",
 			pods: []*v1.Pod{{
 				ObjectMeta: metav1.ObjectMeta{
 					Name:      testPod,
@@ -2293,6 +2322,7 @@ func TestCalculatePodRequests(t *testing.T) {
 				Spec: v1.PodSpec{
 					Containers: []v1.Container{
 						{Name: "container1"},
+						{Name: "container2", Resources: v1.ResourceRequirements{Requests: v1.ResourceList{v1.ResourceCPU: *resource.NewMilliQuantity(50, resource.DecimalSI)}}},
 					},
 				},
 			}},
@@ -2301,6 +2331,71 @@ func TestCalculatePodRequests(t *testing.T) {
 			expectedRequests: nil,
 			expectedError:    fmt.Errorf("missing request for %s in container %s of Pod %s", v1.ResourceCPU, "container1", testPod),
 		},
+		{
+			name:                    "Pod-level resources override missing container requests when feature enabled and pod resources specified",
+			enablePodLevelResources: true,
+			pods: []*v1.Pod{{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      testPod,
+					Namespace: testNamespace,
+				},
+				Spec: v1.PodSpec{
+					Resources: &v1.ResourceRequirements{
+						Requests: v1.ResourceList{v1.ResourceCPU: *resource.NewMilliQuantity(800, resource.DecimalSI)},
+					},
+					Containers: []v1.Container{
+						{Name: "container1"},
+						{Name: "container2", Resources: v1.ResourceRequirements{Requests: v1.ResourceList{v1.ResourceCPU: *resource.NewMilliQuantity(50, resource.DecimalSI)}}},
+					},
+				},
+			}},
+			container:        "",
+			resource:         v1.ResourceCPU,
+			expectedRequests: map[string]int64{testPod: 800},
+			expectedError:    nil,
+		},
+		{
+			name: "Container: if a container name is specified, calculate requests only for that container",
+			pods: []*v1.Pod{{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      testPod,
+					Namespace: testNamespace,
+				},
+				Spec: v1.PodSpec{
+					Containers: []v1.Container{
+						{Name: "container1", Resources: v1.ResourceRequirements{Requests: v1.ResourceList{v1.ResourceCPU: *resource.NewMilliQuantity(100, resource.DecimalSI)}}},
+						{Name: "container2", Resources: v1.ResourceRequirements{Requests: v1.ResourceList{v1.ResourceCPU: *resource.NewMilliQuantity(50, resource.DecimalSI)}}},
+					},
+				},
+			}},
+			container:        "container1",
+			resource:         v1.ResourceCPU,
+			expectedRequests: map[string]int64{testPod: 100},
+			expectedError:    nil,
+		},
+		{
+			name:                    "Container: if a container name is specified, calculate requests only for that container and ignore pod-level requests",
+			enablePodLevelResources: true,
+			pods: []*v1.Pod{{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      testPod,
+					Namespace: testNamespace,
+				},
+				Spec: v1.PodSpec{
+					Resources: &v1.ResourceRequirements{
+						Requests: v1.ResourceList{v1.ResourceCPU: *resource.NewMilliQuantity(800, resource.DecimalSI)},
+					},
+					Containers: []v1.Container{
+						{Name: "container1", Resources: v1.ResourceRequirements{Requests: v1.ResourceList{v1.ResourceCPU: *resource.NewMilliQuantity(100, resource.DecimalSI)}}},
+						{Name: "container2", Resources: v1.ResourceRequirements{Requests: v1.ResourceList{v1.ResourceCPU: *resource.NewMilliQuantity(50, resource.DecimalSI)}}},
+					},
+				},
+			}},
+			container:        "container1",
+			resource:         v1.ResourceCPU,
+			expectedRequests: map[string]int64{testPod: 100},
+			expectedError:    nil,
+		},
 		{
 			name: "pod with restartable init containers",
 			pods: []*v1.Pod{{
@@ -2327,7 +2422,9 @@ func TestCalculatePodRequests(t *testing.T) {
 
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
-			requests, err := calculatePodRequests(tc.pods, tc.container, tc.resource)
+			featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.PodLevelResources, tc.enablePodLevelResources)
+
+			requests, err := calculateRequests(tc.pods, tc.container, tc.resource)
 			assert.Equal(t, tc.expectedRequests, requests, "requests should be as expected")
 			assert.Equal(t, tc.expectedError, err, "error should be as expected")
 		})

test/e2e/autoscaling/autoscaling_timer.go

@@ -86,7 +86,7 @@ var _ = SIGDescribe(feature.ClusterSizeAutoscalingScaleUp, framework.WithSlow(),
 				nodeMemoryMB := (&nodeMemoryBytes).Value() / 1024 / 1024
 				memRequestMB := nodeMemoryMB / 10 // Ensure each pod takes not more than 10% of node's allocatable memory.
 				replicas := 1
-				resourceConsumer := e2eautoscaling.NewDynamicResourceConsumer(ctx, "resource-consumer", f.Namespace.Name, e2eautoscaling.KindDeployment, replicas, 0, 0, 0, cpuRequestMillis, memRequestMB, f.ClientSet, f.ScalesGetter, e2eautoscaling.Disable, e2eautoscaling.Idle)
+				resourceConsumer := e2eautoscaling.NewDynamicResourceConsumer(ctx, "resource-consumer", f.Namespace.Name, e2eautoscaling.KindDeployment, replicas, 0, 0, 0, cpuRequestMillis, memRequestMB, f.ClientSet, f.ScalesGetter, e2eautoscaling.Disable, e2eautoscaling.Idle, nil)
 				ginkgo.DeferCleanup(resourceConsumer.CleanUp)
 				resourceConsumer.WaitForReplicas(ctx, replicas, 1*time.Minute) // Should finish ~immediately, so 1 minute is more than enough.