diff --git a/internal/common/resource/resource.go b/internal/common/resource/resource.go index 1ef9144867e..ed2e4819fff 100644 --- a/internal/common/resource/resource.go +++ b/internal/common/resource/resource.go @@ -264,6 +264,13 @@ func TotalPodResourceRequest(podSpec *v1.PodSpec) ComputeResources { totalResources.Max(containerResource) } } + + // Pod-level resources (KEP-2837): the effective request is + // max(sum of container requests, pod-level request) per resource. Inert when + // podSpec.Resources is unset. + if podSpec.Resources != nil { + totalResources.Max(FromResourceList(podSpec.Resources.Requests)) + } return totalResources } diff --git a/internal/common/resource/resource_test.go b/internal/common/resource/resource_test.go index b0d45c90242..b8a4c80da61 100644 --- a/internal/common/resource/resource_test.go +++ b/internal/common/resource/resource_test.go @@ -8,6 +8,7 @@ import ( "github.com/stretchr/testify/require" v1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/resource" + "k8s.io/utils/ptr" ) func TestComputeResources_String(t *testing.T) { @@ -148,6 +149,52 @@ func TestTotalResourceRequest_ShouldCombineMaxInitContainerResourcesWithSummedCo assert.Equal(t, result, FromResourceList(expectedResult)) } +// Pod-level resources (KEP-2837): TotalPodResourceRequest uses the effective +// request max(sum(containers), pod-level). +func TestTotalResourceRequest_PodLevelResources(t *testing.T) { + tests := map[string]struct { + containerResources []*v1.ResourceList + podLevelResources *v1.ResourceList + expected v1.ResourceList + }{ + "pod-level only (empty containers) is accounted at the pod-level value": { + containerResources: []*v1.ResourceList{}, + podLevelResources: ptr.To(makeContainerResource(4, 16)), + expected: makeContainerResource(4, 16), + }, + // cpu: max(2, 4) = 4 ; memory: max(4, 2) = 4 + "effective request is max of container-sum and pod-level": { + containerResources: []*v1.ResourceList{ptr.To(makeContainerResource(2, 4))}, + podLevelResources: ptr.To(makeContainerResource(4, 2)), + expected: makeContainerResource(4, 4), + }, + "container-sum wins when it exceeds pod-level": { + containerResources: []*v1.ResourceList{ptr.To(makeContainerResource(8, 8))}, + podLevelResources: ptr.To(makeContainerResource(4, 4)), + expected: makeContainerResource(8, 8), + }, + "nil pod-level leaves upstream behaviour unchanged": { + containerResources: []*v1.ResourceList{ptr.To(makeContainerResource(2, 4))}, + podLevelResources: nil, + expected: makeContainerResource(2, 4), + }, + } + for name, tc := range tests { + t.Run(name, func(t *testing.T) { + pod := makePodWithResource(tc.containerResources, []*v1.ResourceList{}) + if tc.podLevelResources != nil { + pod.Spec.Resources = &v1.ResourceRequirements{ + Requests: *tc.podLevelResources, + Limits: *tc.podLevelResources, + } + } + + result := TotalPodResourceRequest(&pod.Spec) + assert.Equal(t, FromResourceList(tc.expected), result) + }) + } +} + func TestTotalResourceRequest_NativeSidecarsShouldBeSummed(t *testing.T) { mainResource := makeContainerResource(2, 1) sidecarResource := makeContainerResource(1, 1) diff --git a/internal/lookoutingester/instructions/instructions.go b/internal/lookoutingester/instructions/instructions.go index 734eff6d45d..feb1c48d0a6 100644 --- a/internal/lookoutingester/instructions/instructions.go +++ b/internal/lookoutingester/instructions/instructions.go @@ -562,18 +562,21 @@ func getJobResources(job *api.Job) jobResources { podSpec := job.GetMainPodSpec() - for _, container := range podSpec.Containers { - resources.Cpu += getResource(container, v1.ResourceCPU, true) - resources.Memory += getResource(container, v1.ResourceMemory, false) - resources.EphemeralStorage += getResource(container, v1.ResourceEphemeralStorage, false) - resources.Gpu += getResource(container, "nvidia.com/gpu", false) - } + // Use the canonical effective-request computation so the reported footprint + // matches what the scheduler bills: sum of main containers + native sidecars, + // max over classic init containers, and max with the pod-level block (KEP-2837). + // This also fixes the prior undercount that summed only main containers. + requests := api.SchedulingResourceRequirementsFromPodSpec(podSpec).Requests + resources.Cpu = getResourceFromList(requests, v1.ResourceCPU, true) + resources.Memory = getResourceFromList(requests, v1.ResourceMemory, false) + resources.EphemeralStorage = getResourceFromList(requests, v1.ResourceEphemeralStorage, false) + resources.Gpu = getResourceFromList(requests, "nvidia.com/gpu", false) return resources } -func getResource(container v1.Container, resourceName v1.ResourceName, useMillis bool) int64 { - resource, ok := container.Resources.Requests[resourceName] +func getResourceFromList(rl v1.ResourceList, resourceName v1.ResourceName, useMillis bool) int64 { + resource, ok := rl[resourceName] if !ok { return 0 } diff --git a/internal/scheduler/api.go b/internal/scheduler/api.go index 25825233306..6ac6ec8694d 100644 --- a/internal/scheduler/api.go +++ b/internal/scheduler/api.go @@ -238,6 +238,12 @@ func (srv *ExecutorApi) dropDisallowedResources(pod *v1.PodSpec) { } srv.dropDisallowedResourcesFromContainers(pod.InitContainers) srv.dropDisallowedResourcesFromContainers(pod.Containers) + // Pod-level resources (KEP-2837) must be filtered by the same allow-list, else + // a disallowed resource could bypass it via the pod-level block. + if pod.Resources != nil { + removeDisallowedKeys(pod.Resources.Limits, srv.allowedResources) + removeDisallowedKeys(pod.Resources.Requests, srv.allowedResources) + } } func (srv *ExecutorApi) dropDisallowedResourcesFromContainers(containers []v1.Container) { diff --git a/internal/server/configuration/types.go b/internal/server/configuration/types.go index 38d88f43fce..6dc1bab4abb 100644 --- a/internal/server/configuration/types.go +++ b/internal/server/configuration/types.go @@ -99,6 +99,11 @@ type SubmissionConfig struct { AddGangIdLabel bool // Controls whether custom service names are allowed AllowCustomServiceNames bool + // When true, honour Kubernetes pod-level resources (KEP-2837, podSpec.Resources): + // a container may omit its own resources if the pod-level block is set, and + // accounting uses max(sum(container requests), podLevel.Requests). Default false + // preserves container-only behaviour. + PodLevelResources bool } // TODO: we can probably just typedef this to map[string]string diff --git a/internal/server/submit/conversion/post_process.go b/internal/server/submit/conversion/post_process.go index 4fce47f45e0..47fc64f3470 100644 --- a/internal/server/submit/conversion/post_process.go +++ b/internal/server/submit/conversion/post_process.go @@ -30,6 +30,7 @@ var ( addGangIdLabel, } podLevelProcessors = []podProcessor{ + dropPodLevelResourcesIfDisabled, defaultActiveDeadlineSeconds, defaultPriorityClass, defaultResource, @@ -110,6 +111,14 @@ func defaultPriorityClass(spec *v1.PodSpec, config configuration.SubmissionConfi } } +// Clears the pod-level resources block (KEP-2837) unless the feature is enabled, +// so all downstream accounting ignores it when the feature is off. +func dropPodLevelResourcesIfDisabled(spec *v1.PodSpec, config configuration.SubmissionConfig) { + if !config.PodLevelResources { + spec.Resources = nil + } +} + // Adds resources defined in config.DefaultJobLimits to all containers in the podspec if that container is missing // requests/limits for that particular resource. This can be used to e.g. ensure that all jobs define at least some // ephemeral storage. diff --git a/internal/server/submit/conversion/post_process_test.go b/internal/server/submit/conversion/post_process_test.go index a03c8370083..1092bb8db30 100644 --- a/internal/server/submit/conversion/post_process_test.go +++ b/internal/server/submit/conversion/post_process_test.go @@ -756,3 +756,36 @@ func submitMsgFromAnnotations(annotations map[string]string) *armadaevents.Submi }, } } + +func TestDropPodLevelResourcesIfDisabled(t *testing.T) { + podLevel := &v1.ResourceRequirements{ + Requests: v1.ResourceList{"cpu": resource.MustParse("2")}, + Limits: v1.ResourceList{"cpu": resource.MustParse("2")}, + } + + tests := map[string]struct { + initialResources *v1.ResourceRequirements + enabled bool + expectedResources *v1.ResourceRequirements + }{ + "disabled clears the pod-level block": { + initialResources: podLevel, + enabled: false, + }, + "enabled preserves the pod-level block": { + initialResources: podLevel, + enabled: true, + expectedResources: podLevel, + }, + "unset pod-level block is left unset": { + enabled: true, + }, + } + for name, tc := range tests { + t.Run(name, func(t *testing.T) { + spec := &v1.PodSpec{Resources: tc.initialResources.DeepCopy()} + dropPodLevelResourcesIfDisabled(spec, configuration.SubmissionConfig{PodLevelResources: tc.enabled}) + assert.Equal(t, tc.expectedResources, spec.Resources) + }) + } +} diff --git a/internal/server/submit/validation/submit_request.go b/internal/server/submit/validation/submit_request.go index ccb7d527c21..b85485e2376 100644 --- a/internal/server/submit/validation/submit_request.go +++ b/internal/server/submit/validation/submit_request.go @@ -5,6 +5,7 @@ import ( "strings" "github.com/pkg/errors" + v1 "k8s.io/api/core/v1" "k8s.io/component-helpers/scheduling/corev1/nodeaffinity" "github.com/armadaproject/armada/internal/common/constants" @@ -252,8 +253,21 @@ func validateResources(j *api.JobSubmitRequestItem, config configuration.Submiss if maxOversubscriptionByResource == nil { maxOversubscriptionByResource = map[string]float64{} } + // Pod-level resources (KEP-2837): when enabled and a pod-level block is set, a + // container may omit its own resources. Containers that do set resources are + // still validated below, as is the pod-level block itself. + podLevelResourcesEnabled := config.PodLevelResources && spec.Resources != nil + if podLevelResourcesEnabled { + if err := validatePodLevelResources(spec, maxOversubscriptionByResource, config); err != nil { + return err + } + } for _, container := range armadaslices.Concatenate(spec.Containers, spec.InitContainers) { if len(container.Resources.Requests) == 0 && len(container.Resources.Limits) == 0 { + if podLevelResourcesEnabled { + // Budget is supplied at the pod level; nothing to validate for this container. + continue + } return fmt.Errorf("container %v has no resources specified", container.Name) } @@ -308,6 +322,62 @@ func validateResources(j *api.JobSubmitRequestItem, config configuration.Submiss return nil } +// validatePodLevelResources validates a pod-level resources block (KEP-2837), +// mirroring the per-container checks: requests and limits must be non-negative, +// cover the same resource set, satisfy limit >= request within the +// max-oversubscription ratio, and meet MinJobResources. +func validatePodLevelResources( + spec *v1.PodSpec, + maxOversubscriptionByResource map[string]float64, + config configuration.SubmissionConfig, +) error { + resources := spec.Resources + if len(resources.Requests) == 0 && len(resources.Limits) == 0 { + return fmt.Errorf("pod-level resources block is empty") + } + if len(resources.Requests) != len(resources.Limits) { + return fmt.Errorf("pod-level resources define different resources for requests and limits") + } + for resourceName, request := range resources.Requests { + if request.Sign() < 0 { + return fmt.Errorf("pod-level resources define negative request (%s) for resource %s", request.String(), resourceName) + } + } + for resourceName, limit := range resources.Limits { + if limit.Sign() < 0 { + return fmt.Errorf("pod-level resources define negative limit (%s) for resource %s", limit.String(), resourceName) + } + } + for resourceName, request := range resources.Requests { + limit, ok := resources.Limits[resourceName] + if !ok { + return fmt.Errorf("pod-level resources define %s for requests but not limits", resourceName) + } + if limit.MilliValue() < request.MilliValue() { + return fmt.Errorf("pod-level resources define %s with limits smaller than requests", resourceName) + } + maxOversubscription, ok := maxOversubscriptionByResource[resourceName.String()] + if !ok { + maxOversubscription = 1.0 + } + if float64(limit.MilliValue()) > maxOversubscription*float64(request.MilliValue()) { + return fmt.Errorf("pod-level resources define %s with limits greater than %.2f*requests", resourceName, maxOversubscription) + } + } + // MinJobResources is checked against the effective request + // (max of the pod-level request and the summed container requests), since that + // is what the scheduler reserves; checking the raw pod-level value alone would + // wrongly reject a pod whose container total already meets the minimum. + effective := api.SchedulingResourceRequirementsFromPodSpec(spec).Requests + for rc, serverRsc := range config.MinJobResources { + eff := effective[rc] + if eff.Value() < serverRsc.Value() { + return fmt.Errorf("effective %s requests (%s) below server minimum (%s)", rc, &eff, &serverRsc) + } + } + return nil +} + // jobAdapter turns JobSubmitRequestItem into a MinimalJob // This is needed for gang information to be extracted type jobAdapter struct { diff --git a/internal/server/submit/validation/submit_request_test.go b/internal/server/submit/validation/submit_request_test.go index 5b83dbd9085..ee3d351146b 100644 --- a/internal/server/submit/validation/submit_request_test.go +++ b/internal/server/submit/validation/submit_request_test.go @@ -1150,6 +1150,122 @@ func TestValidateResources(t *testing.T) { } } +// TestValidateResources_PodLevel covers Kubernetes pod-level resources (KEP-2837), +// gated by SubmissionConfig.PodLevelResources. +func TestValidateResources_PodLevel(t *testing.T) { + oneCpu := v1.ResourceList{v1.ResourceCPU: resource.MustParse("1")} + twoCpu := v1.ResourceList{v1.ResourceCPU: resource.MustParse("2")} + negativeCpu := v1.ResourceList{v1.ResourceCPU: resource.MustParse("-1")} + + // Container that declares no resources of its own. + emptyContainer := []v1.Container{{Name: "main"}} + + req := func(podResources *v1.ResourceRequirements, containers []v1.Container) *api.JobSubmitRequestItem { + return &api.JobSubmitRequestItem{ + PodSpec: &v1.PodSpec{ + Containers: containers, + Resources: podResources, + }, + } + } + + tests := map[string]struct { + req *api.JobSubmitRequestItem + podLevelEnabled bool + expectSuccess bool + }{ + "pod-level-only accepted when feature enabled": { + req: req(&v1.ResourceRequirements{Requests: oneCpu, Limits: oneCpu}, emptyContainer), + podLevelEnabled: true, + expectSuccess: true, + }, + "pod-level-only rejected when feature disabled": { + req: req(&v1.ResourceRequirements{Requests: oneCpu, Limits: oneCpu}, emptyContainer), + podLevelEnabled: false, + expectSuccess: false, + }, + "empty container with NO pod-level still rejected even when enabled": { + req: req(nil, emptyContainer), + podLevelEnabled: true, + expectSuccess: false, + }, + "pod-level with limits < requests rejected": { + req: req(&v1.ResourceRequirements{Requests: twoCpu, Limits: oneCpu}, emptyContainer), + podLevelEnabled: true, + expectSuccess: false, + }, + "pod-level negative request rejected": { + req: req(&v1.ResourceRequirements{Requests: negativeCpu, Limits: negativeCpu}, emptyContainer), + podLevelEnabled: true, + expectSuccess: false, + }, + "pod-level empty block rejected": { + req: req(&v1.ResourceRequirements{}, emptyContainer), + podLevelEnabled: true, + expectSuccess: false, + }, + "container-only still valid with feature enabled": { + req: req(nil, []v1.Container{{ + Name: "main", + Resources: v1.ResourceRequirements{Requests: oneCpu, Limits: oneCpu}, + }}), + podLevelEnabled: true, + expectSuccess: true, + }, + } + + for name, tc := range tests { + t.Run(name, func(t *testing.T) { + cfg := configuration.SubmissionConfig{PodLevelResources: tc.podLevelEnabled} + err := validateResources(tc.req, cfg) + if tc.expectSuccess { + assert.NoError(t, err) + } else { + assert.Error(t, err) + } + }) + } +} + +// MinJobResources is checked against the effective request (max of pod-level and +// summed container requests), not the raw pod-level value. +func TestValidateResources_PodLevelMinJobResources(t *testing.T) { + fourCpuMin := v1.ResourceList{v1.ResourceCPU: resource.MustParse("4")} + cpu := func(s string) v1.ResourceList { return v1.ResourceList{v1.ResourceCPU: resource.MustParse(s)} } + + req := func(pod *v1.ResourceRequirements, containers []v1.Container) *api.JobSubmitRequestItem { + return &api.JobSubmitRequestItem{PodSpec: &v1.PodSpec{Containers: containers, Resources: pod}} + } + container5 := []v1.Container{{Name: "main", Resources: v1.ResourceRequirements{Requests: cpu("5"), Limits: cpu("5")}}} + + tests := map[string]struct { + req *api.JobSubmitRequestItem + expectSuccess bool + }{ + // container 5cpu + pod-level 2cpu -> effective 5cpu >= 4cpu min: accepted. + "container total meets minimum, low pod-level ok": { + req: req(&v1.ResourceRequirements{Requests: cpu("2"), Limits: cpu("2")}, container5), + expectSuccess: true, + }, + // pod-level 2cpu, empty container -> effective 2cpu < 4cpu min: rejected. + "pod-level below minimum with empty container rejected": { + req: req(&v1.ResourceRequirements{Requests: cpu("2"), Limits: cpu("2")}, []v1.Container{{Name: "main"}}), + expectSuccess: false, + }, + } + for name, tc := range tests { + t.Run(name, func(t *testing.T) { + cfg := configuration.SubmissionConfig{PodLevelResources: true, MinJobResources: fourCpuMin} + err := validateResources(tc.req, cfg) + if tc.expectSuccess { + assert.NoError(t, err) + } else { + assert.Error(t, err) + } + }) + } +} + func TestValidateTerminationGracePeriod(t *testing.T) { defaultMinPeriod := 30 * time.Second defaultMaxPeriod := 300 * time.Second diff --git a/pkg/api/util.go b/pkg/api/util.go index 1fb0347c412..44505d9d8a3 100644 --- a/pkg/api/util.go +++ b/pkg/api/util.go @@ -69,6 +69,13 @@ func SchedulingResourceRequirementsFromPodSpec(podSpec *v1.PodSpec) *v1.Resource maxResourcesToList(rv.Limits, c.Resources.Limits) } } + + // Pod-level resources (KEP-2837): max with the pod-level request/limit so the + // scheduler reserves the pod-level budget. Inert when podSpec.Resources is unset. + if podSpec.Resources != nil { + maxResourcesToList(rv.Requests, podSpec.Resources.Requests) + maxResourcesToList(rv.Limits, podSpec.Resources.Limits) + } return &rv } diff --git a/pkg/api/util_test.go b/pkg/api/util_test.go index 6de1198c78b..55ffae4ae1e 100644 --- a/pkg/api/util_test.go +++ b/pkg/api/util_test.go @@ -373,6 +373,39 @@ func TestSchedulingResourceRequirementsFromPodSpec(t *testing.T) { }, }, }, + // Pod-level resources (KEP-2837). + "pod-level only (empty containers) uses the pod-level value": { + input: &v1.PodSpec{ + Containers: []v1.Container{{}}, + Resources: &v1.ResourceRequirements{ + Requests: v1.ResourceList{"cpu": QuantityWithMilliValue(4000)}, + Limits: v1.ResourceList{"cpu": QuantityWithMilliValue(4000)}, + }, + }, + expected: &v1.ResourceRequirements{ + Requests: v1.ResourceList{"cpu": QuantityWithMilliValue(4000)}, + Limits: v1.ResourceList{"cpu": QuantityWithMilliValue(4000)}, + }, + }, + "pod-level is max'd with the container sum": { + input: &v1.PodSpec{ + Containers: []v1.Container{{ + Resources: v1.ResourceRequirements{ + Requests: v1.ResourceList{"cpu": QuantityWithMilliValue(1000)}, + Limits: v1.ResourceList{"cpu": QuantityWithMilliValue(1000)}, + }, + }}, + Resources: &v1.ResourceRequirements{ + Requests: v1.ResourceList{"cpu": QuantityWithMilliValue(4000)}, + Limits: v1.ResourceList{"cpu": QuantityWithMilliValue(4000)}, + }, + }, + // max(container-sum 1000, pod-level 4000) = 4000 + expected: &v1.ResourceRequirements{ + Requests: v1.ResourceList{"cpu": QuantityWithMilliValue(4000)}, + Limits: v1.ResourceList{"cpu": QuantityWithMilliValue(4000)}, + }, + }, } for name, tc := range tests { t.Run(name, func(t *testing.T) {