v0.4 - add runtime class support (#399)

* add pod overhead resources to total requirements

* adding e2e test with runtime class

* update changelog

* add failure message due to overhead resources

* update message due to pod overhead

Author: enoodle

CHANGELOG.md

@@ -4,6 +4,11 @@ All notable changes to this project will be documented in this file.
 
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
+## [Unreleased]
+
+### Fixed
+- kai-scheduler will not ignore pod spec.overhead field
+
 ## [v0.4.12] - 2025-07-18
 
 ### Fixed

pkg/scheduler/api/common_info/errors.go

@@ -19,6 +19,7 @@ const (
 	ResourcesWereNotFoundMsg = "no nodes with enough resources were found"
 	DefaultPodgroupError     = "Unable to schedule podgroup"
 	DefaultPodError          = "Unable to schedule pod"
+	OverheadMessage          = "Not enough resources due to pod overhead resources"
 )
 
 type FitError struct {
@@ -60,7 +61,7 @@ func NewFitErrorByReasons(name, namespace, nodeName string, err error, reasons .
 func NewFitErrorInsufficientResource(
 	name, namespace, nodeName string,
 	resourceRequested *resource_info.ResourceRequirements, usedResource, capacityResource *resource_info.Resource,
-	capacityGpuMemory int64, gangSchedulingJob bool,
+	capacityGpuMemory int64, gangSchedulingJob bool, messageSuffix string,
 ) *FitError {
 	availableResource := capacityResource.Clone()
 	availableResource.Sub(usedResource)
@@ -146,6 +147,15 @@ func NewFitErrorInsufficientResource(
 		}
 	}
 
+	if len(messageSuffix) > 0 {
+		for i, msg := range shortMessages {
+			shortMessages[i] = fmt.Sprintf("%s. %s", msg, messageSuffix)
+		}
+		for i, msg := range detailedMessages {
+			detailedMessages[i] = fmt.Sprintf("%s. %s", msg, messageSuffix)
+		}
+	}
+
 	return NewFitErrorWithDetailedMessage(name, namespace, nodeName, shortMessages, detailedMessages...)
 }
 

pkg/scheduler/api/common_info/errors_test.go

@@ -51,6 +51,7 @@ func TestNewFitErrorInsufficientResource(t *testing.T) {
 		capacityResource  *resource_info.Resource
 		capacityGpuMemory int64
 		gangSchedulingJob bool
+		suffix            string
 	}
 	tests := []struct {
 		name string
@@ -163,12 +164,33 @@ func TestNewFitErrorInsufficientResource(t *testing.T) {
 				DetailedReasons: []string{"Node didn't have enough resources: Each gpu on the node has a gpu memory capacity of 1000 Mib. 2000 Mib of gpu memory has been requested."},
 			},
 		},
+		{
+			name: "Not enough cpu due to pod overhead",
+			args: args{
+				name:              "t1",
+				namespace:         "n1",
+				nodeName:          "node1",
+				resourceRequested: resource_info.NewResourceRequirements(0, 1500, 1000),
+				usedResource:      BuildResource("500m", "1M"),
+				capacityResource:  BuildResource("1000m", "2M"),
+				capacityGpuMemory: 0,
+				gangSchedulingJob: false,
+				suffix:            "Message suffix",
+			},
+			want: &FitError{
+				taskName:        "t1",
+				taskNamespace:   "n1",
+				NodeName:        "node1",
+				Reasons:         []string{"node(s) didn't have enough resources: CPU cores. Message suffix"},
+				DetailedReasons: []string{"Node didn't have enough resources: CPU cores, requested: 1.5, used: 0.5, capacity: 1. Message suffix"},
+			},
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			if got := NewFitErrorInsufficientResource(tt.args.name, tt.args.namespace, tt.args.nodeName,
 				tt.args.resourceRequested, tt.args.usedResource, tt.args.capacityResource, tt.args.capacityGpuMemory,
-				tt.args.gangSchedulingJob); !reflect.DeepEqual(got, tt.want) {
+				tt.args.gangSchedulingJob, tt.args.suffix); !reflect.DeepEqual(got, tt.want) {
 				t.Errorf("NewFitErrorInsufficientResource() = %v, want %v", got, tt.want)
 			}
 		})

pkg/scheduler/api/node_info/mig_node_info_test.go

@@ -58,7 +58,7 @@ func TestIsTaskAllocatable_Mig(t *testing.T) {
 		t.Run(testName, func(t *testing.T) {
 			runAllocatableTest(
 				t, testData, testName,
-				func(ni *NodeInfo, task *pod_info.PodInfo) (bool, error) {
+				func(ni *NodeInfo, task *pod_info.PodInfo) (bool, *common_info.FitError) {
 					allocatable, err := ni.IsTaskAllocatable(task), ni.FittingError(task, false)
 					return allocatable, err
 				},

pkg/scheduler/api/node_info/node_info.go

@@ -24,6 +24,7 @@ import (
 	"github.com/NVIDIA/KAI-scheduler/pkg/scheduler/api/resource_info"
 	sc_info "github.com/NVIDIA/KAI-scheduler/pkg/scheduler/api/storagecapacity_info"
 	"github.com/NVIDIA/KAI-scheduler/pkg/scheduler/api/storageclaim_info"
+	"github.com/NVIDIA/KAI-scheduler/pkg/scheduler/k8s_utils"
 	"github.com/NVIDIA/KAI-scheduler/pkg/scheduler/log"
 )
 
@@ -243,9 +244,23 @@ func (ni *NodeInfo) FittingError(task *pod_info.PodInfo, isGangTask bool) *commo
 				task.ResReq.GetNumOfGpuDevices(), ni.getResourceGpuPortion(task.ResReq), requestedResources.GpuMemory())
 		}
 
-		return common_info.NewFitErrorInsufficientResource(
+		messageSuffix := ""
+		if len(task.Pod.Spec.Overhead) > 0 {
+			// Adding to node idle instead of subtracting from pod requested resources
+			idleResourcesWithOverhead := ni.Idle.Clone()
+			idleResourcesWithOverhead.Add(resource_info.ResourceFromResourceList(task.Pod.Spec.Overhead))
+			enoughResourcesWithoutOverhead := ni.lessEqualTaskToNodeResources(task.ResReq, idleResourcesWithOverhead)
+			if enoughResourcesWithoutOverhead {
+				messageSuffix = fmt.Sprintf("%s. The overhead resources are %v", common_info.OverheadMessage,
+					k8s_utils.StringResourceList(task.Pod.Spec.Overhead))
+			}
+		}
+
+		fitError := common_info.NewFitErrorInsufficientResource(
 			task.Name, task.Namespace, ni.Name, task.ResReq, totalUsed, totalCapability, ni.MemoryOfEveryGpuOnNode,
-			isGangTask)
+			isGangTask, messageSuffix)
+
+		return fitError
 	}
 
 	allocatable, err := ni.isTaskStorageAllocatable(task)

pkg/scheduler/api/node_info/node_info_test.go

@@ -445,10 +445,12 @@ type allocatableTestData struct {
 	podsResources           []v1.ResourceList
 	podResourcesToAllocate  v1.ResourceList
 	podAnnotations          map[string]string
+	podOverhead             v1.ResourceList
 	expected                bool
 	expectedMessageContains []string
+	expectOverheadMessage   bool
 }
-type allocatableTestFunction func(ni *NodeInfo, task *pod_info.PodInfo) (bool, error)
+type allocatableTestFunction func(ni *NodeInfo, task *pod_info.PodInfo) (bool, *common_info.FitError)
 
 func TestIsTaskAllocatable(t *testing.T) {
 	nodeCapacityDifferent := common_info.BuildNode("n2", common_info.BuildResourceList("1000m", "1G"))
@@ -519,13 +521,49 @@ func TestIsTaskAllocatable(t *testing.T) {
 			expected:                true,
 			expectedMessageContains: []string{},
 		},
+		"pod with overhead that fits without overhead but not with overhead": {
+			node:                    common_info.BuildNode("n1", common_info.BuildResourceList("2000m", "2G")),
+			podsResources:           []v1.ResourceList{common_info.BuildResourceList("1000m", "1G")},
+			podResourcesToAllocate:  common_info.BuildResourceList("500m", "500M"),
+			podOverhead:             common_info.BuildResourceList("600m", "600M"),
+			expected:                false,
+			expectedMessageContains: []string{"CPU cores", "memory"},
+			expectOverheadMessage:   true,
+		},
+		"pod with overhead that doesn't fit even without overhead": {
+			node:                    common_info.BuildNode("n1", common_info.BuildResourceList("2000m", "2G")),
+			podsResources:           []v1.ResourceList{common_info.BuildResourceList("1000m", "1G")},
+			podResourcesToAllocate:  common_info.BuildResourceList("1500m", "1500M"),
+			podOverhead:             common_info.BuildResourceList("100m", "100M"),
+			expected:                false,
+			expectedMessageContains: []string{"CPU cores", "memory"},
+			expectOverheadMessage:   false,
+		},
+		"pod without overhead that doesn't fit": {
+			node:                    common_info.BuildNode("n1", common_info.BuildResourceList("2000m", "2G")),
+			podsResources:           []v1.ResourceList{common_info.BuildResourceList("1000m", "1G")},
+			podResourcesToAllocate:  common_info.BuildResourceList("1500m", "1500M"),
+			podOverhead:             v1.ResourceList{},
+			expected:                false,
+			expectedMessageContains: []string{"CPU cores", "memory"},
+			expectOverheadMessage:   false,
+		},
+		"pod with overhead that fits with overhead": {
+			node:                    common_info.BuildNode("n1", common_info.BuildResourceList("2000m", "2G")),
+			podsResources:           []v1.ResourceList{common_info.BuildResourceList("1000m", "1G")},
+			podResourcesToAllocate:  common_info.BuildResourceList("500m", "500M"),
+			podOverhead:             common_info.BuildResourceList("100m", "100M"),
+			expected:                true,
+			expectedMessageContains: []string{},
+			expectOverheadMessage:   false,
+		},
 	}
 
 	for testName, testData := range tests {
 		t.Run(testName, func(t *testing.T) {
 			runAllocatableTest(
 				t, testData, testName,
-				func(ni *NodeInfo, task *pod_info.PodInfo) (bool, error) {
+				func(ni *NodeInfo, task *pod_info.PodInfo) (bool, *common_info.FitError) {
 					return ni.IsTaskAllocatable(task), ni.FittingError(task, false)
 				},
 			)
@@ -561,7 +599,7 @@ func TestIsTaskAllocatableOnReleasingOrIdle(t *testing.T) {
 		t.Run(testName, func(t *testing.T) {
 			runAllocatableTest(
 				t, testData, testName,
-				func(ni *NodeInfo, task *pod_info.PodInfo) (bool, error) {
+				func(ni *NodeInfo, task *pod_info.PodInfo) (bool, *common_info.FitError) {
 					return ni.IsTaskAllocatableOnReleasingOrIdle(task), nil
 				},
 			)
@@ -592,6 +630,11 @@ func runAllocatableTest(
 		"podToAllocate", "p1", "n1", v1.PodRunning, testData.podResourcesToAllocate,
 		[]metav1.OwnerReference{}, make(map[string]string), testData.podAnnotations)
 	addJobAnnotation(pod)
+
+	if len(testData.podOverhead) > 0 {
+		pod.Spec.Overhead = testData.podOverhead
+	}
+
 	task := pod_info.NewTaskInfo(pod)
 	allocatable, fitErr := testedFunction(ni, task)
 	if allocatable != testData.expected {
@@ -603,6 +646,20 @@ func runAllocatableTest(
 				t.Errorf("%s: expected error message to contain %s, got %s", testName, expectedMessage, fitErr.Error())
 			}
 		}
+
+		if testData.expectOverheadMessage {
+			if !strings.Contains(fitErr.Error(), "Not enough resources due to pod overhead resources") {
+				t.Errorf("%s: expected overhead message, got %s", testName, fitErr.Error())
+			}
+		} else {
+			fitErrMessage := fitErr.Error()
+			if strings.Contains(fitErrMessage, "Not enough resources due to pod overhead resources") {
+				t.Errorf("%s: unexpected overhead message, got %s", testName, fitErrMessage)
+			}
+		}
+	} else if len(testData.expectedMessageContains) > 0 {
+		// If we expected an error but got none, that's a test failure
+		t.Errorf("%s: expected error but got none", testName)
 	}
 }
 

pkg/scheduler/api/pod_info/pod_info.go

@@ -307,6 +307,11 @@ func getPodResourceRequest(pod *v1.Pod) *resource_info.ResourceRequirements {
 		}
 	}
 
+	if pod.Spec.Overhead != nil {
+		overheadReq := resource_info.RequirementsFromResourceList(pod.Spec.Overhead)
+		result.Add(&overheadReq.BaseResource)
+	}
+
 	return result
 }
 

pkg/scheduler/api/pod_info/pod_info_test.go

@@ -109,6 +109,27 @@ func TestGetPodResourceRequest(t *testing.T) {
 			},
 			expectedResource: resource_info.NewResourceRequirements(1, 3000, 5000000000),
 		},
+		{
+			name: "pod with overhead resources",
+			pod: &v1.Pod{
+				Spec: v1.PodSpec{
+					Containers: []v1.Container{
+						{
+							Resources: v1.ResourceRequirements{
+								Requests: common_info.BuildResourceListWithGPU("1000m", "1G", "1"),
+							},
+						},
+						{
+							Resources: v1.ResourceRequirements{
+								Requests: common_info.BuildResourceList("2000m", "1G"),
+							},
+						},
+					},
+					Overhead: common_info.BuildResourceList("1000m", "1G"),
+				},
+			},
+			expectedResource: resource_info.NewResourceRequirements(1, 4000, 3000000000),
+		},
 	}
 	for i, test := range tests {
 		req := getPodResourceRequest(test.pod)

pkg/scheduler/k8s_utils/k8s_utils.go

@@ -51,3 +51,12 @@ func annotationAndLabelsPatchBytes(annotations, labels map[string]interface{}) (
 		"annotations": annotations,
 	}})
 }
+
+func StringResourceList(resources v1.ResourceList) string {
+	output := ""
+	for name, value := range resources {
+		output += fmt.Sprintf(" %s: %s", name, value.String())
+	}
+
+	return output
+}

test/e2e/suites/integrations/k8s_native/k8s_native_test.go

@@ -13,6 +13,7 @@ import (
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 	v1 "k8s.io/api/core/v1"
+	nodev1 "k8s.io/api/node/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/utils/ptr"
@@ -48,20 +49,65 @@ var _ = Describe("K8S Native object integrations", Ordered, func() {
 		testCtx.ClusterCleanup(ctx)
 	})
 
-	It("Pod", func(ctx context.Context) {
-		pod := rd.CreatePodObject(testCtx.Queues[0], v1.ResourceRequirements{})
+	Context("Pods", func() {
+		It("schedules simple pods", func(ctx context.Context) {
+			pod := rd.CreatePodObject(testCtx.Queues[0], v1.ResourceRequirements{})
 
-		_, err := rd.CreatePod(ctx, testCtx.KubeClientset, pod)
-		if err != nil {
-			Expect(err).NotTo(HaveOccurred(), "Failed to create pod-job")
-		}
+			_, err := rd.CreatePod(ctx, testCtx.KubeClientset, pod)
+			if err != nil {
+				Expect(err).NotTo(HaveOccurred(), "Failed to create pod-job")
+			}
 
-		defer func() {
-			err = testCtx.KubeClientset.CoreV1().Pods(pod.Namespace).Delete(ctx, pod.Name, metav1.DeleteOptions{})
-			Expect(err).To(Succeed())
-		}()
+			defer func() {
+				err = testCtx.KubeClientset.CoreV1().Pods(pod.Namespace).Delete(ctx, pod.Name, metav1.DeleteOptions{})
+				Expect(err).To(Succeed())
+			}()
+
+			wait.ForPodScheduled(ctx, testCtx.ControllerClient, pod)
+		})
+
+		It("considers pod Overhead from runtimeclass", func(ctx context.Context) {
+			runtimeClassName := "my-runtime-class-" + utils.GenerateRandomK8sName(5)
+			limitedQueue := queue.CreateQueueObject("limited-"+utils.GenerateRandomK8sName(10), testCtx.Queues[1].Name)
+			limitedQueue.Spec.Resources.CPU.Limit = 1
+			testCtx.AddQueues(ctx, []*v2.Queue{limitedQueue})
+
+			defer func() {
+				Expect(testCtx.ControllerClient.Delete(ctx, limitedQueue)).To(Succeed())
+			}()
+
+			pod := rd.CreatePodObject(limitedQueue, v1.ResourceRequirements{})
+			pod.Spec.RuntimeClassName = &runtimeClassName
 
-		wait.ForPodScheduled(ctx, testCtx.ControllerClient, pod)
+			runtimeClass := &nodev1.RuntimeClass{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: runtimeClassName,
+				},
+				Handler: "runc",
+				Overhead: &nodev1.Overhead{
+					PodFixed: v1.ResourceList{
+						v1.ResourceCPU: resource.MustParse("2"),
+					},
+				},
+			}
+
+			Expect(testCtx.ControllerClient.Create(ctx, runtimeClass)).To(Succeed())
+
+			defer func() {
+				Expect(testCtx.ControllerClient.Delete(ctx, runtimeClass)).To(Succeed())
+			}()
+
+			_, err := rd.CreatePod(ctx, testCtx.KubeClientset, pod)
+			if err != nil {
+				Expect(err).NotTo(HaveOccurred(), "Failed to create pod-job")
+			}
+
+			defer func() {
+				Expect(testCtx.ControllerClient.Delete(ctx, pod)).To(Succeed())
+			}()
+
+			wait.ForPodUnschedulable(ctx, testCtx.ControllerClient, pod)
+		})
 	})
 
 	It("ReplicaSet", func(ctx context.Context) {