From c9a272fa55fa89fce9de534b52a1d8b6c408b961 Mon Sep 17 00:00:00 2001 From: maishivamhoo123 Date: Sat, 18 Apr 2026 20:03:37 +0000 Subject: [PATCH 01/14] fix/nodelock-time-parsec-panic Signed-off-by: maishivamhoo123 --- pkg/scheduler/scheduler.go | 28 +++++- pkg/scheduler/scheduler_test.go | 49 +++++++++++ pkg/scheduler/score.go | 31 ++++++- pkg/scheduler/score_test.go | 147 ++++++++++++++++++++++++++++++++ pkg/scheduler/webhook.go | 55 ++++++++++-- pkg/scheduler/webhook_test.go | 83 ++++++++++++++++++ 6 files changed, 382 insertions(+), 11 deletions(-) diff --git a/pkg/scheduler/scheduler.go b/pkg/scheduler/scheduler.go index 57d40a5ac..772638d48 100644 --- a/pkg/scheduler/scheduler.go +++ b/pkg/scheduler/scheduler.go @@ -724,12 +724,32 @@ ReleaseNodeLocks: func (s *Scheduler) Filter(args extenderv1.ExtenderArgs) (*extenderv1.ExtenderFilterResult, error) { klog.InfoS("Starting schedule filter process", "pod", args.Pod.Name, "uuid", args.Pod.UID, "namespace", args.Pod.Namespace) resourceReqs := device.Resourcereqs(args.Pod) - resourceReqTotal := 0 - for _, n := range resourceReqs { - for _, k := range n { - resourceReqTotal += int(k.Nums) + + initReqTotal := 0 + appReqTotal := 0 + numInitContainers := len(args.Pod.Spec.InitContainers) + + for i := range numInitContainers { + currentInitReq := 0 + for _, k := range resourceReqs[i] { + currentInitReq += int(k.Nums) + } + if currentInitReq > initReqTotal { + initReqTotal = currentInitReq + } + } + + // 2. Calculate the SUM of requests among Regular Containers + for i := numInitContainers; i < len(resourceReqs); i++ { + for _, k := range resourceReqs[i] { + appReqTotal += int(k.Nums) } } + + // 3. The effective total request is the MAX of (InitContainers, RegularContainers) + resourceReqTotal := appReqTotal + resourceReqTotal = max(resourceReqTotal, initReqTotal) + if resourceReqTotal == 0 { klog.V(1).InfoS("Pod does not request any resources", "pod", args.Pod.Name) diff --git a/pkg/scheduler/scheduler_test.go b/pkg/scheduler/scheduler_test.go index f1ca3bcfa..b3bce9f0c 100644 --- a/pkg/scheduler/scheduler_test.go +++ b/pkg/scheduler/scheduler_test.go @@ -499,6 +499,55 @@ func Test_Filter(t *testing.T) { }, wantPodAnnotationDeviceID: "device4", }, + { + name: "pod with init containers fits correctly using max resource logic (Binpack)", + args: extenderv1.ExtenderArgs{ + Pod: &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-init-containers", + UID: "test-init-uid", + Annotations: map[string]string{ + util.GPUSchedulerPolicyAnnotationKey: util.GPUSchedulerPolicyBinpack.String(), + util.NodeSchedulerPolicyAnnotationKey: util.NodeSchedulerPolicyBinpack.String(), + }, + }, + Spec: corev1.PodSpec{ + InitContainers: []corev1.Container{ + { + Name: "init-1", + Image: "busybox", + Resources: corev1.ResourceRequirements{ + Limits: corev1.ResourceList{ + "hami.io/gpu": *resource.NewQuantity(1, resource.BinarySI), + "hami.io/gpucores": *resource.NewQuantity(20, resource.BinarySI), + "hami.io/gpumem": *resource.NewQuantity(5000, resource.BinarySI), + }, + }, + }, + }, + Containers: []corev1.Container{ + { + Name: "app-1", + Image: "chrstnhntschl/gpu_burn", + Resources: corev1.ResourceRequirements{ + Limits: corev1.ResourceList{ + "hami.io/gpu": *resource.NewQuantity(1, resource.BinarySI), + "hami.io/gpucores": *resource.NewQuantity(20, resource.BinarySI), + "hami.io/gpumem": *resource.NewQuantity(4000, resource.BinarySI), + }, + }, + }, + }, + }, + }, + NodeNames: &[]string{"node1", "node2"}, + }, + wantErr: nil, + want: &extenderv1.ExtenderFilterResult{ + NodeNames: &[]string{"node2"}, + }, + wantPodAnnotationDeviceID: "device3", + }, { name: "node use binpack gpu use spread policy", args: extenderv1.ExtenderArgs{ diff --git a/pkg/scheduler/score.go b/pkg/scheduler/score.go index 236f3fedb..3720a7db3 100644 --- a/pkg/scheduler/score.go +++ b/pkg/scheduler/score.go @@ -16,6 +16,7 @@ limitations under the License. package scheduler import ( + "encoding/json" "fmt" "sort" "strings" @@ -140,6 +141,21 @@ func (s *Scheduler) calcScore(nodes *map[string]*NodeUsage, resourceReqs device. // requests, which should be schedulable on any node. ctrfit := true deviceType := "" + + numInitContainers := len(task.Spec.InitContainers) + + // Snapshot the node's initial state so init containers don't drain the pool + initialNodeBytes, err := json.Marshal(node) + if err != nil { + klog.ErrorS(err, "Failed to marshal node state for cloning", "nodeID", nodeID) + errCh <- err + return + } + + // This shared state will be drained by regular app containers + appContainersNode := &NodeUsage{} + json.Unmarshal(initialNodeBytes, appContainersNode) + //This loop is for different container request for ctrid, n := range resourceReqs { sums := 0 @@ -153,7 +169,20 @@ func (s *Scheduler) calcScore(nodes *map[string]*NodeUsage, resourceReqs device. continue } klog.V(5).InfoS("fitInDevices", "pod", klog.KObj(task), "node", nodeID) - fit, reason := fitInDevices(node, n, task, nodeInfo, &score.Devices) + + // Determine which node state to pass to fitInDevices + var workingNode *NodeUsage + if ctrid < numInitContainers { + // Init containers get a fresh reset of the node's capacity + workingNode = &NodeUsage{} + json.Unmarshal(initialNodeBytes, workingNode) + } else { + // Regular containers deduct from the shared app capacity + workingNode = appContainersNode + } + + fit, reason := fitInDevices(workingNode, n, task, nodeInfo, &score.Devices) + // found certain deviceType, fill missing empty allocation for containers before this for idx := range score.Devices { deviceType = idx diff --git a/pkg/scheduler/score_test.go b/pkg/scheduler/score_test.go index fa670ef7a..165bfd8ed 100644 --- a/pkg/scheduler/score_test.go +++ b/pkg/scheduler/score_test.go @@ -2444,6 +2444,153 @@ func Test_calcScore(t *testing.T) { err: nil, }, }, + { + name: "one node two devices, pod has one init container (uses 1 device) and one regular container (uses 1 device) - tests InitContainer capacity reset", + args: struct { + nodes *map[string]*NodeUsage + nums device.PodDeviceRequests + annos map[string]string + task *corev1.Pod + }{ + nodes: &map[string]*NodeUsage{ + "node1": { + Node: &corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: "node1"}}, + Devices: policy.DeviceUsageList{ + Policy: util.NodeSchedulerPolicyBinpack.String(), + DeviceLists: []*policy.DeviceListsScore{ + { + Device: &device.DeviceUsage{ + ID: "uuid1", + Index: 0, + Used: 0, + Count: 10, + Usedmem: 0, + Totalmem: 8000, + Totalcore: 100, + Usedcores: 0, + Numa: 0, + Type: nvidia.NvidiaGPUDevice, + Health: true, + }, + }, + { + Device: &device.DeviceUsage{ + ID: "uuid2", + Index: 0, + Used: 0, + Count: 10, + Usedmem: 0, + Totalmem: 8000, + Totalcore: 100, + Usedcores: 0, + Numa: 0, + Type: nvidia.NvidiaGPUDevice, + Health: true, + }, + }, + }, + }, + }, + }, + nums: device.PodDeviceRequests{ + // Index 0: InitContainer Request + { + nvidia.NvidiaGPUDevice: device.ContainerDeviceRequest{ + Nums: 1, + Type: nvidia.NvidiaGPUDevice, + Memreq: 1000, + Coresreq: 30, + }, + }, + // Index 1: Regular Container Request + { + nvidia.NvidiaGPUDevice: device.ContainerDeviceRequest{ + Nums: 1, + Type: nvidia.NvidiaGPUDevice, + Memreq: 1000, + Coresreq: 30, + }, + }, + }, + annos: make(map[string]string), + task: &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-init-container", + }, + Spec: corev1.PodSpec{ + InitContainers: []corev1.Container{ + { + Name: "init-gpu-burn", + Image: "chrstnhntschl/gpu_burn", + Args: []string{"6000"}, + Resources: corev1.ResourceRequirements{ + Limits: corev1.ResourceList{ + "hami.io/gpu": *resource.NewQuantity(1, resource.BinarySI), + "hami.io/gpucores": *resource.NewQuantity(30, resource.BinarySI), + "hami.io/gpumem": *resource.NewQuantity(1000, resource.BinarySI), + }, + }, + }, + }, + Containers: []corev1.Container{ + { + Name: "app-gpu-burn", + Image: "chrstnhntschl/gpu_burn", + Args: []string{"6000"}, + Resources: corev1.ResourceRequirements{ + Limits: corev1.ResourceList{ + "hami.io/gpu": *resource.NewQuantity(1, resource.BinarySI), + "hami.io/gpucores": *resource.NewQuantity(30, resource.BinarySI), + "hami.io/gpumem": *resource.NewQuantity(1000, resource.BinarySI), + }, + }, + }, + }, + }, + }, + }, + wants: struct { + want *policy.NodeScoreList + failedNodes map[string]string + err error + }{ + want: &policy.NodeScoreList{ + Policy: util.NodeSchedulerPolicyBinpack.String(), + NodeList: []*policy.NodeScore{ + { + NodeID: "node1", + Devices: device.PodDevices{ + "NVIDIA": device.PodSingleDevice{ + // Index 0: InitContainer successfully allocated to uuid1 + { + { + Idx: 0, + UUID: "uuid2", + Type: nvidia.NvidiaGPUDevice, + Usedcores: 30, + Usedmem: 1000, + }, + }, + // Index 1: Regular Container successfully allocated to uuid1 + // (Proves that the node state reset correctly after the InitContainer) + { + { + Idx: 0, + UUID: "uuid2", + Type: nvidia.NvidiaGPUDevice, + Usedcores: 30, + Usedmem: 1000, + }, + }, + }, + }, + Score: 0, + }, + }, + }, + err: nil, + }, + }, { name: "two node per node having one device one pod two container use one device", args: struct { diff --git a/pkg/scheduler/webhook.go b/pkg/scheduler/webhook.go index f3ab758ae..545615d3c 100644 --- a/pkg/scheduler/webhook.go +++ b/pkg/scheduler/webhook.go @@ -69,6 +69,27 @@ func (h *webhook) Handle(_ context.Context, req admission.Request) admission.Res } klog.V(5).Infof(template, pod.Namespace, pod.Name, pod.UID) hasResource := false + + // 1. Process InitContainers + for idx, ctr := range pod.Spec.InitContainers { + c := &pod.Spec.InitContainers[idx] + if ctr.SecurityContext != nil { + if ctr.SecurityContext.Privileged != nil && *ctr.SecurityContext.Privileged { + klog.Warningf(template+" - Denying admission as init container %s is privileged", pod.Namespace, pod.Name, pod.UID, c.Name) + continue + } + } + for _, val := range device.GetDevices() { + found, err := val.MutateAdmission(c, pod) + if err != nil { + klog.Errorf("validating pod failed:%s", err.Error()) + return admission.Errored(http.StatusInternalServerError, err) + } + hasResource = hasResource || found + } + } + + // 2. Process Regular Containers (Keep your existing loop here) for idx, ctr := range pod.Spec.Containers { c := &pod.Spec.Containers[idx] if ctr.SecurityContext != nil { @@ -110,7 +131,6 @@ func (h *webhook) Handle(_ context.Context, req admission.Request) admission.Res func fitResourceQuota(pod *corev1.Pod) bool { for deviceName, dev := range device.GetDevices() { - // Only supports NVIDIA if deviceName != nvidia.NvidiaGPUDevice { continue } @@ -119,8 +139,7 @@ func fitResourceQuota(pod *corev1.Pod) bool { resourceName := corev1.ResourceName(resourceNames.ResourceCountName) memResourceName := corev1.ResourceName(resourceNames.ResourceMemoryName) coreResourceName := corev1.ResourceName(resourceNames.ResourceCoreName) - var memoryReq int64 = 0 - var coresReq int64 = 0 + getRequest := func(ctr *corev1.Container, resName corev1.ResourceName) (int64, bool) { v, ok := ctr.Resources.Limits[resName] if !ok { @@ -133,24 +152,48 @@ func fitResourceQuota(pod *corev1.Pod) bool { } return 0, false } + + var initMemoryReq, initCoresReq int64 + var appMemoryReq, appCoresReq int64 + + // 1. Calculate the MAX request among InitContainers + for _, ctr := range pod.Spec.InitContainers { + req, ok := getRequest(&ctr, resourceName) + if ok && req == 1 { + if memReq, ok := getRequest(&ctr, memResourceName); ok { + // --- FIX 2: Use modern max() --- + initMemoryReq = max(initMemoryReq, memReq) + } + if coreReq, ok := getRequest(&ctr, coreResourceName); ok { + initCoresReq = max(initCoresReq, coreReq) + } + } + } + + // 2. Calculate the SUM of requests among Regular Containers for _, ctr := range pod.Spec.Containers { req, ok := getRequest(&ctr, resourceName) if ok && req == 1 { if memReq, ok := getRequest(&ctr, memResourceName); ok { - memoryReq += memReq + appMemoryReq += memReq } if coreReq, ok := getRequest(&ctr, coreResourceName); ok { - coresReq += coreReq + appCoresReq += coreReq } } } + + memoryReq := max(appMemoryReq, initMemoryReq) + coresReq := max(appCoresReq, initCoresReq) + if memoryFactor > 1 { oriMemReq := memoryReq memoryReq = memoryReq * int64(memoryFactor) klog.V(5).Infof("Adjusting memory request for quota check: oriMemReq %d, memoryReq %d, factor %d", oriMemReq, memoryReq, memoryFactor) } + if !device.GetLocalCache().FitQuota(pod.Namespace, memoryReq, memoryFactor, coresReq, deviceName) { - klog.Infof(template+" - Denying admission", pod.Namespace, pod.Name, pod.UID) + klog.Infof("Namespace %s, Pod %s, UID %s - Denying admission", pod.Namespace, pod.Name, pod.UID) return false } } diff --git a/pkg/scheduler/webhook_test.go b/pkg/scheduler/webhook_test.go index 00f356ca5..af6354096 100644 --- a/pkg/scheduler/webhook_test.go +++ b/pkg/scheduler/webhook_test.go @@ -358,6 +358,89 @@ func TestFitResourceQuota(t *testing.T) { }, fit: true, }, + { + name: "InitContainers run sequentially: max init fits quota, but simple sum would exceed", + pod: &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-pod-init-fit", + Namespace: "default", + }, + Spec: corev1.PodSpec{ + SchedulerName: "hami-scheduler", + // Two init containers each asking for 800. Max = 800. + InitContainers: []corev1.Container{ + { + Name: "init1", + Resources: corev1.ResourceRequirements{ + Limits: corev1.ResourceList{ + "nvidia.com/gpu": resource.MustParse("1"), + "nvidia.com/gpumem": resource.MustParse("800"), + }, + }, + }, + { + Name: "init2", + Resources: corev1.ResourceRequirements{ + Limits: corev1.ResourceList{ + "nvidia.com/gpu": resource.MustParse("1"), + "nvidia.com/gpumem": resource.MustParse("800"), + }, + }, + }, + }, + // App container asking for 500. Total effective requirement = max(800, 500) = 800. + // 800 is less than the available 1000 limit, so it should fit. + // If the quota manager wrongly sums everything (800+800+500=2100), this test will catch it by failing. + Containers: []corev1.Container{ + { + Name: "app1", + Resources: corev1.ResourceRequirements{ + Limits: corev1.ResourceList{ + "nvidia.com/gpu": resource.MustParse("1"), + "nvidia.com/gpumem": resource.MustParse("500"), + }, + }, + }, + }, + }, + }, + fit: true, + }, + { + name: "InitContainer request exceeds quota directly", + pod: &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-pod-init-fail", + Namespace: "default", + }, + Spec: corev1.PodSpec{ + SchedulerName: "hami-scheduler", + InitContainers: []corev1.Container{ + { + Name: "init-massive", + Resources: corev1.ResourceRequirements{ + Limits: corev1.ResourceList{ + "nvidia.com/gpu": resource.MustParse("1"), + "nvidia.com/gpumem": resource.MustParse("1500"), // 1500 > 1000 available limit + }, + }, + }, + }, + Containers: []corev1.Container{ + { + Name: "app1", + Resources: corev1.ResourceRequirements{ + Limits: corev1.ResourceList{ + "nvidia.com/gpu": resource.MustParse("1"), + "nvidia.com/gpumem": resource.MustParse("100"), + }, + }, + }, + }, + }, + }, + fit: false, + }, { name: "request ascend", pod: &corev1.Pod{ From f6049484e1b66c8234cbb87b5f30306188c4e1d2 Mon Sep 17 00:00:00 2001 From: maishivamhoo123 Date: Sat, 18 Apr 2026 20:24:22 +0000 Subject: [PATCH 02/14] added all the suggestions Signed-off-by: maishivamhoo123 --- pkg/scheduler/score.go | 43 +++++++++++++++++++++++++++++------------- 1 file changed, 30 insertions(+), 13 deletions(-) diff --git a/pkg/scheduler/score.go b/pkg/scheduler/score.go index 3720a7db3..0e62f3452 100644 --- a/pkg/scheduler/score.go +++ b/pkg/scheduler/score.go @@ -120,6 +120,7 @@ func (s *Scheduler) calcScore(nodes *map[string]*NodeUsage, resourceReqs device. failedNodesMutex := sync.Mutex{} failureReason := make(map[string][]string) errCh := make(chan error, len(*nodes)) + for nodeID, node := range *nodes { wg.Add(1) go func(nodeID string, node *NodeUsage) { @@ -137,24 +138,35 @@ func (s *Scheduler) calcScore(nodes *map[string]*NodeUsage, resourceReqs device. return } - // Assume the node is a fit by default. This handles pods with no device - // requests, which should be schedulable on any node. ctrfit := true deviceType := "" numInitContainers := len(task.Spec.InitContainers) - // Snapshot the node's initial state so init containers don't drain the pool - initialNodeBytes, err := json.Marshal(node) - if err != nil { - klog.ErrorS(err, "Failed to marshal node state for cloning", "nodeID", nodeID) - errCh <- err - return - } + var initialNodeBytes []byte + var appContainersNode *NodeUsage + + if numInitContainers > 0 { + var err error + // Snapshot the node's initial state so init containers don't drain the pool + initialNodeBytes, err = json.Marshal(node) + if err != nil { + klog.ErrorS(err, "Failed to marshal node state for cloning", "nodeID", nodeID) + errCh <- err + return + } - // This shared state will be drained by regular app containers - appContainersNode := &NodeUsage{} - json.Unmarshal(initialNodeBytes, appContainersNode) + // This shared state will be drained by regular app containers + appContainersNode = &NodeUsage{} + if err := json.Unmarshal(initialNodeBytes, appContainersNode); err != nil { + klog.ErrorS(err, "Failed to unmarshal node state", "nodeID", nodeID) + errCh <- err + return + } + } else { + // Avoid expensive JSON operations if there are no init containers + appContainersNode = node + } //This loop is for different container request for ctrid, n := range resourceReqs { @@ -175,7 +187,12 @@ func (s *Scheduler) calcScore(nodes *map[string]*NodeUsage, resourceReqs device. if ctrid < numInitContainers { // Init containers get a fresh reset of the node's capacity workingNode = &NodeUsage{} - json.Unmarshal(initialNodeBytes, workingNode) + + if err := json.Unmarshal(initialNodeBytes, workingNode); err != nil { + klog.ErrorS(err, "Failed to unmarshal node state for init container", "nodeID", nodeID) + errCh <- err + return + } } else { // Regular containers deduct from the shared app capacity workingNode = appContainersNode From 07e98689bc8c169db5526a2326194b943e0ac884 Mon Sep 17 00:00:00 2001 From: maishivamhoo123 Date: Sat, 18 Apr 2026 20:56:56 +0000 Subject: [PATCH 03/14] writing cleaner code Signed-off-by: maishivamhoo123 --- pkg/scheduler/scheduler_test.go | 28 +++++++++++++++++----------- 1 file changed, 17 insertions(+), 11 deletions(-) diff --git a/pkg/scheduler/scheduler_test.go b/pkg/scheduler/scheduler_test.go index b3bce9f0c..ee6a00e56 100644 --- a/pkg/scheduler/scheduler_test.go +++ b/pkg/scheduler/scheduler_test.go @@ -20,6 +20,7 @@ import ( "context" "fmt" "maps" + "slices" "strings" "sync/atomic" "testing" @@ -456,11 +457,11 @@ func Test_Filter(t *testing.T) { } tests := []struct { - name string - args extenderv1.ExtenderArgs - want *extenderv1.ExtenderFilterResult - wantPodAnnotationDeviceID string - wantErr error + name string + args extenderv1.ExtenderArgs + want *extenderv1.ExtenderFilterResult + wantPodAnnotationDeviceIDs []string + wantErr error }{ { name: "node use binpack gpu use binpack policy", @@ -497,7 +498,7 @@ func Test_Filter(t *testing.T) { want: &extenderv1.ExtenderFilterResult{ NodeNames: &[]string{"node2"}, }, - wantPodAnnotationDeviceID: "device4", + wantPodAnnotationDeviceIDs: []string{"device4"}, }, { name: "pod with init containers fits correctly using max resource logic (Binpack)", @@ -546,7 +547,7 @@ func Test_Filter(t *testing.T) { want: &extenderv1.ExtenderFilterResult{ NodeNames: &[]string{"node2"}, }, - wantPodAnnotationDeviceID: "device3", + wantPodAnnotationDeviceIDs: []string{"device3"}, }, { name: "node use binpack gpu use spread policy", @@ -583,7 +584,7 @@ func Test_Filter(t *testing.T) { want: &extenderv1.ExtenderFilterResult{ NodeNames: &[]string{"node2"}, }, - wantPodAnnotationDeviceID: "device3", + wantPodAnnotationDeviceIDs: []string{"device3", "device4"}, // Both are acceptable due to tie }, { name: "node use spread gpu use binpack policy", @@ -620,7 +621,7 @@ func Test_Filter(t *testing.T) { want: &extenderv1.ExtenderFilterResult{ NodeNames: &[]string{"node1"}, }, - wantPodAnnotationDeviceID: "device1", + wantPodAnnotationDeviceIDs: []string{"device1"}, }, { name: "node use spread gpu use spread policy", @@ -657,7 +658,7 @@ func Test_Filter(t *testing.T) { want: &extenderv1.ExtenderFilterResult{ NodeNames: &[]string{"node1"}, }, - wantPodAnnotationDeviceID: "device2", + wantPodAnnotationDeviceIDs: []string{"device2"}, }, } @@ -670,7 +671,12 @@ func Test_Filter(t *testing.T) { assert.DeepEqual(t, test.want, got) getPod, _ := client.KubeClient.CoreV1().Pods(test.args.Pod.Namespace).Get(context.Background(), test.args.Pod.Name, metav1.GetOptions{}) podDevices, _ := device.DecodePodDevices(device.SupportDevices, getPod.Annotations) - assert.DeepEqual(t, test.wantPodAnnotationDeviceID, podDevices["NVIDIA"][0][0].UUID) + + actualUUID := podDevices["NVIDIA"][0][0].UUID + + if !slices.Contains(test.wantPodAnnotationDeviceIDs, actualUUID) { + t.Errorf("expected one of %v, got %s", test.wantPodAnnotationDeviceIDs, actualUUID) + } }) } } From 385e32ce506d940e006b02ab41dcf33b3da69754 Mon Sep 17 00:00:00 2001 From: maishivamhoo123 Date: Thu, 30 Apr 2026 01:33:04 +0000 Subject: [PATCH 04/14] adding the file Signed-off-by: maishivamhoo123 --- pkg/scheduler/score.go | 135 ++++++++++++++++++++++++++++++------ pkg/scheduler/score_test.go | 125 ++++++++++++++++++++++++++++++--- 2 files changed, 228 insertions(+), 32 deletions(-) diff --git a/pkg/scheduler/score.go b/pkg/scheduler/score.go index 0e62f3452..e53979e13 100644 --- a/pkg/scheduler/score.go +++ b/pkg/scheduler/score.go @@ -33,6 +33,13 @@ import ( "github.com/Project-HAMi/HAMi/pkg/util" ) +// containerResourceSummary holds both the GPU count and total memory +// for a group of containers, so we can compare them on both dimensions. +type containerResourceSummary struct { + nums int + memreq int32 +} + func viewStatus(usage NodeUsage) { klog.V(5).Info("devices status") for _, val := range usage.Devices.DeviceLists { @@ -51,16 +58,15 @@ func getNodeResources(list NodeUsage, t string) []*device.DeviceUsage { } func fitInDevices(node *NodeUsage, requests device.ContainerDeviceRequests, pod *corev1.Pod, nodeInfo *device.NodeInfo, devinput *device.PodDevices) (bool, string) { - //devmap := make(map[string]device.ContainerDevices) devs := device.ContainerDevices{} total, totalCore, totalMem := int32(0), int32(0), int32(0) free, freeCore, freeMem := int32(0), int32(0), int32(0) sums := 0 - // computer all device score for one node + // compute all device scores for one node for index := range node.Devices.DeviceLists { node.Devices.DeviceLists[index].ComputeScore(requests) } - //This loop is for requests for different devices + // This loop is for requests for different devices for _, k := range requests { sums += int(k.Nums) if int(k.Nums) > len(node.Devices.DeviceLists) { @@ -76,7 +82,7 @@ func fitInDevices(node *NodeUsage, requests device.ContainerDeviceRequests, pod if fit { for idx, val := range tmpDevs[k.Type] { for nidx, v := range node.Devices.DeviceLists { - //bc node.Devices has been sorted, so we should find out the correct device + // bc node.Devices has been sorted, so we should find out the correct device if v.Device.ID != val.UUID { continue } @@ -103,6 +109,40 @@ func fitInDevices(node *NodeUsage, requests device.ContainerDeviceRequests, pod return true, "" } +// podInitContainerMaxRequest returns the maximum single-container device +// request across all init containers, considering both GPU count AND memory. +// Per Kubernetes semantics, init containers run sequentially, so the node +// only needs to reserve capacity for the largest one at any moment. +func podInitContainerMaxRequest(resourceReqs device.PodDeviceRequests, numInitContainers int) containerResourceSummary { + maxReq := containerResourceSummary{} + for i := range numInitContainers { + var nums int + var mem int32 + for _, k := range resourceReqs[i] { + nums += int(k.Nums) + mem += k.Memreq + } + if nums > maxReq.nums || (nums == maxReq.nums && mem > maxReq.memreq) { + maxReq.nums = nums + maxReq.memreq = mem + } + } + return maxReq +} + +// podAppContainerTotalRequest returns the sum of device requests across all +// regular (non-init) containers, considering both GPU count AND memory. +func podAppContainerTotalRequest(resourceReqs device.PodDeviceRequests, numInitContainers int) containerResourceSummary { + total := containerResourceSummary{} + for i := numInitContainers; i < len(resourceReqs); i++ { + for _, k := range resourceReqs[i] { + total.nums += int(k.Nums) + total.memreq += k.Memreq + } + } + return total +} + func (s *Scheduler) calcScore(nodes *map[string]*NodeUsage, resourceReqs device.PodDeviceRequests, task *corev1.Pod, failedNodes map[string]string) (*policy.NodeScoreList, error) { userNodePolicy := config.NodeSchedulerPolicy if task.GetAnnotations() != nil { @@ -121,6 +161,22 @@ func (s *Scheduler) calcScore(nodes *map[string]*NodeUsage, resourceReqs device. failureReason := make(map[string][]string) errCh := make(chan error, len(*nodes)) + // Pre-compute container counts and request totals once, outside the + // per-node goroutines, since they are identical for every node. + numInitContainers := len(task.Spec.InitContainers) + maxInitReq := podInitContainerMaxRequest(resourceReqs, numInitContainers) + appReqTotal := podAppContainerTotalRequest(resourceReqs, numInitContainers) + + // Maintainer optimization (@Shouren): + // Init containers run sequentially and exit before app containers start. + // The node only ever needs to hold max(maxInitReq, appReqTotal) free at + // any moment. We compare both GPU count AND memory to correctly detect + // when init containers need more resources than app containers. + // If needsInitClone=false, we skip fitInDevices for init containers + // entirely — the app allocation implicitly covers them. + needsInitClone := numInitContainers > 0 && + (maxInitReq.nums > appReqTotal.nums || maxInitReq.memreq > appReqTotal.memreq) + for nodeID, node := range *nodes { wg.Add(1) go func(nodeID string, node *NodeUsage) { @@ -141,14 +197,21 @@ func (s *Scheduler) calcScore(nodes *map[string]*NodeUsage, resourceReqs device. ctrfit := true deviceType := "" - numInitContainers := len(task.Spec.InitContainers) + // appContainersNode is the NodeUsage that regular containers + // will deduct from. + var appContainersNode *NodeUsage + // initialNodeBytes holds the JSON snapshot of the node's full + // capacity. It is only populated when needsInitClone is true, + // so that each init container can be evaluated against a fresh + // copy of the node without draining the pool for app containers. var initialNodeBytes []byte - var appContainersNode *NodeUsage - if numInitContainers > 0 { - var err error - // Snapshot the node's initial state so init containers don't drain the pool + if needsInitClone { + // Init containers request MORE than app containers (by count + // or memory), so we must evaluate them against full node + // capacity. Marshal once; each init container gets its own + // Unmarshal. initialNodeBytes, err = json.Marshal(node) if err != nil { klog.ErrorS(err, "Failed to marshal node state for cloning", "nodeID", nodeID) @@ -156,51 +219,78 @@ func (s *Scheduler) calcScore(nodes *map[string]*NodeUsage, resourceReqs device. return } - // This shared state will be drained by regular app containers + // App containers share a separate copy that starts at full + // capacity and is drained as each app container is scheduled. appContainersNode = &NodeUsage{} if err := json.Unmarshal(initialNodeBytes, appContainersNode); err != nil { - klog.ErrorS(err, "Failed to unmarshal node state", "nodeID", nodeID) + klog.ErrorS(err, "Failed to unmarshal node state for app containers", "nodeID", nodeID) errCh <- err return } } else { - // Avoid expensive JSON operations if there are no init containers + // No init containers, or init requests are already covered + // by app container requests — use the node directly with no + // clone overhead. appContainersNode = node } - //This loop is for different container request + // This loop iterates over every container's device requests. + // resourceReqs is ordered: [initContainer_0, ..., initContainer_N-1, + // appContainer_0, ..., appContainer_M-1] for ctrid, n := range resourceReqs { sums := 0 for _, k := range n { sums += int(k.Nums) } - // container need no device and we have got certain deviceType + // Maintainer optimization (@Shouren): + // When init container requests are already covered by app + // container requests (!needsInitClone), skip fitInDevices + // for init containers entirely. Their device usage is + // implicitly satisfied by the app container allocation since + // init and app containers never run simultaneously — the node + // only needs max(init, app) capacity at any moment. + if ctrid < numInitContainers && !needsInitClone { + if deviceType != "" { + score.Devices[deviceType] = append(score.Devices[deviceType], device.ContainerDevices{}) + } + continue + } + + // Container needs no device but a deviceType has already been + // identified — record an empty allocation and continue. if sums == 0 && deviceType != "" { score.Devices[deviceType] = append(score.Devices[deviceType], device.ContainerDevices{}) continue } + klog.V(5).InfoS("fitInDevices", "pod", klog.KObj(task), "node", nodeID) - // Determine which node state to pass to fitInDevices + // Decide which NodeUsage view to pass to fitInDevices. var workingNode *NodeUsage - if ctrid < numInitContainers { - // Init containers get a fresh reset of the node's capacity + if needsInitClone && ctrid < numInitContainers { + // This is an init container that needs more resources than + // the app containers. Give it a fresh snapshot of the + // node's full capacity so it doesn't interfere with the + // app container pool. workingNode = &NodeUsage{} - if err := json.Unmarshal(initialNodeBytes, workingNode); err != nil { - klog.ErrorS(err, "Failed to unmarshal node state for init container", "nodeID", nodeID) + klog.ErrorS(err, "Failed to unmarshal node state for init container", + "nodeID", nodeID, "ctrid", ctrid) errCh <- err return } } else { - // Regular containers deduct from the shared app capacity + // Regular app container (or an init container whose + // request is already covered by the app allocation). + // Both deduct from the shared accumulated state. workingNode = appContainersNode } fit, reason := fitInDevices(workingNode, n, task, nodeInfo, &score.Devices) - // found certain deviceType, fill missing empty allocation for containers before this + // Fill any missing empty-allocation slots for containers + // that were processed before the first device-using container. for idx := range score.Devices { deviceType = idx for len(score.Devices[idx]) <= ctrid { @@ -210,6 +300,7 @@ func (s *Scheduler) calcScore(nodes *map[string]*NodeUsage, resourceReqs device. score.Devices[idx] = append(emptyPodSingleDevice, score.Devices[idx]...) } } + ctrfit = fit if !fit { klog.V(4).InfoS(common.NodeUnfitPod, "pod", klog.KObj(task), "node", nodeID, "reason", reason) @@ -235,7 +326,7 @@ func (s *Scheduler) calcScore(nodes *map[string]*NodeUsage, resourceReqs device. wg.Wait() close(errCh) - // only pod scheduler failure will record failure event + // Only record a filter failure event when no node could fit the pod. if len(res.NodeList) == 0 { for reasonType, failureNodes := range failureReason { sort.Strings(failureNodes) diff --git a/pkg/scheduler/score_test.go b/pkg/scheduler/score_test.go index 165bfd8ed..86b296b38 100644 --- a/pkg/scheduler/score_test.go +++ b/pkg/scheduler/score_test.go @@ -2561,7 +2561,10 @@ func Test_calcScore(t *testing.T) { NodeID: "node1", Devices: device.PodDevices{ "NVIDIA": device.PodSingleDevice{ - // Index 0: InitContainer successfully allocated to uuid1 + // Index 0: InitContainer skipped (covered by app allocation) + // since init request == app request, no clone needed. + {}, + // Index 1: Regular Container allocated normally. { { Idx: 0, @@ -2571,16 +2574,118 @@ func Test_calcScore(t *testing.T) { Usedmem: 1000, }, }, - // Index 1: Regular Container successfully allocated to uuid1 - // (Proves that the node state reset correctly after the InitContainer) + }, + }, + Score: 0, + }, + }, + }, + err: nil, + }, + }, + { + name: "init container requests MORE than app container - needsInitClone=true path", + args: struct { + nodes *map[string]*NodeUsage + nums device.PodDeviceRequests + annos map[string]string + task *corev1.Pod + }{ + nodes: &map[string]*NodeUsage{ + "node1": { + Node: &corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: "node1"}}, + Devices: policy.DeviceUsageList{ + Policy: util.NodeSchedulerPolicyBinpack.String(), + DeviceLists: []*policy.DeviceListsScore{ + { + Device: &device.DeviceUsage{ + ID: "uuid1", Index: 0, Used: 0, Count: 10, + Usedmem: 0, Totalmem: 8000, Totalcore: 100, + Usedcores: 0, Numa: 0, + Type: nvidia.NvidiaGPUDevice, Health: true, + }, + }, + { + Device: &device.DeviceUsage{ + ID: "uuid2", Index: 0, Used: 0, Count: 10, + Usedmem: 0, Totalmem: 8000, Totalcore: 100, + Usedcores: 0, Numa: 0, + Type: nvidia.NvidiaGPUDevice, Health: true, + }, + }, + }, + }, + }, + }, + nums: device.PodDeviceRequests{ + // Index 0: InitContainer requests 2 GPUs (MORE than app) + { + nvidia.NvidiaGPUDevice: device.ContainerDeviceRequest{ + Nums: 2, Type: nvidia.NvidiaGPUDevice, + Memreq: 1000, Coresreq: 30, + }, + }, + // Index 1: App container requests only 1 GPU (LESS than init) + { + nvidia.NvidiaGPUDevice: device.ContainerDeviceRequest{ + Nums: 1, Type: nvidia.NvidiaGPUDevice, + Memreq: 1000, Coresreq: 30, + }, + }, + }, + annos: make(map[string]string), + task: &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: "test-init-bigger"}, + Spec: corev1.PodSpec{ + InitContainers: []corev1.Container{ + { + Name: "init-heavy", + Image: "chrstnhntschl/gpu_burn", + Resources: corev1.ResourceRequirements{ + Limits: corev1.ResourceList{ + "hami.io/gpu": *resource.NewQuantity(2, resource.BinarySI), + "hami.io/gpucores": *resource.NewQuantity(30, resource.BinarySI), + "hami.io/gpumem": *resource.NewQuantity(1000, resource.BinarySI), + }, + }, + }, + }, + Containers: []corev1.Container{ + { + Name: "app-light", + Image: "chrstnhntschl/gpu_burn", + Resources: corev1.ResourceRequirements{ + Limits: corev1.ResourceList{ + "hami.io/gpu": *resource.NewQuantity(1, resource.BinarySI), + "hami.io/gpucores": *resource.NewQuantity(30, resource.BinarySI), + "hami.io/gpumem": *resource.NewQuantity(1000, resource.BinarySI), + }, + }, + }, + }, + }, + }, + }, + wants: struct { + want *policy.NodeScoreList + failedNodes map[string]string + err error + }{ + want: &policy.NodeScoreList{ + Policy: util.NodeSchedulerPolicyBinpack.String(), + NodeList: []*policy.NodeScore{ + { + NodeID: "node1", + Devices: device.PodDevices{ + "NVIDIA": device.PodSingleDevice{ + // Init container allocated 2 GPUs from fresh snapshot + { + {Idx: 0, UUID: "uuid2", Type: nvidia.NvidiaGPUDevice, Usedcores: 30, Usedmem: 1000}, + {Idx: 0, UUID: "uuid1", Type: nvidia.NvidiaGPUDevice, Usedcores: 30, Usedmem: 1000}, + }, + // App container allocated 1 GPU from its own fresh pool { - { - Idx: 0, - UUID: "uuid2", - Type: nvidia.NvidiaGPUDevice, - Usedcores: 30, - Usedmem: 1000, - }, + {Idx: 0, UUID: "uuid2", Type: nvidia.NvidiaGPUDevice, Usedcores: 30, Usedmem: 1000}, }, }, }, From f6cc3d62efb9ed00e7567098312c8725b29ccaf4 Mon Sep 17 00:00:00 2001 From: maishivamhoo123 Date: Wed, 6 May 2026 09:26:15 +0000 Subject: [PATCH 05/14] refactor(score): replace JSON clone with DeepCopy Signed-off-by: maishivamhoo123 --- pkg/scheduler/score.go | 59 +++--------------------------------------- 1 file changed, 3 insertions(+), 56 deletions(-) diff --git a/pkg/scheduler/score.go b/pkg/scheduler/score.go index e53979e13..f39f054dc 100644 --- a/pkg/scheduler/score.go +++ b/pkg/scheduler/score.go @@ -16,7 +16,6 @@ limitations under the License. package scheduler import ( - "encoding/json" "fmt" "sort" "strings" @@ -161,19 +160,10 @@ func (s *Scheduler) calcScore(nodes *map[string]*NodeUsage, resourceReqs device. failureReason := make(map[string][]string) errCh := make(chan error, len(*nodes)) - // Pre-compute container counts and request totals once, outside the - // per-node goroutines, since they are identical for every node. numInitContainers := len(task.Spec.InitContainers) maxInitReq := podInitContainerMaxRequest(resourceReqs, numInitContainers) appReqTotal := podAppContainerTotalRequest(resourceReqs, numInitContainers) - // Maintainer optimization (@Shouren): - // Init containers run sequentially and exit before app containers start. - // The node only ever needs to hold max(maxInitReq, appReqTotal) free at - // any moment. We compare both GPU count AND memory to correctly detect - // when init containers need more resources than app containers. - // If needsInitClone=false, we skip fitInDevices for init containers - // entirely — the app allocation implicitly covers them. needsInitClone := numInitContainers > 0 && (maxInitReq.nums > appReqTotal.nums || maxInitReq.memreq > appReqTotal.memreq) @@ -197,40 +187,14 @@ func (s *Scheduler) calcScore(nodes *map[string]*NodeUsage, resourceReqs device. ctrfit := true deviceType := "" - // appContainersNode is the NodeUsage that regular containers - // will deduct from. var appContainersNode *NodeUsage - - // initialNodeBytes holds the JSON snapshot of the node's full - // capacity. It is only populated when needsInitClone is true, - // so that each init container can be evaluated against a fresh - // copy of the node without draining the pool for app containers. - var initialNodeBytes []byte - if needsInitClone { // Init containers request MORE than app containers (by count - // or memory), so we must evaluate them against full node - // capacity. Marshal once; each init container gets its own - // Unmarshal. - initialNodeBytes, err = json.Marshal(node) - if err != nil { - klog.ErrorS(err, "Failed to marshal node state for cloning", "nodeID", nodeID) - errCh <- err - return - } - - // App containers share a separate copy that starts at full - // capacity and is drained as each app container is scheduled. - appContainersNode = &NodeUsage{} - if err := json.Unmarshal(initialNodeBytes, appContainersNode); err != nil { - klog.ErrorS(err, "Failed to unmarshal node state for app containers", "nodeID", nodeID) - errCh <- err - return - } + // or memory). DeepCopy replaces JSON marshal/unmarshal + appContainersNode = node.DeepCopy() } else { // No init containers, or init requests are already covered // by app container requests — use the node directly with no - // clone overhead. appContainersNode = node } @@ -243,10 +207,6 @@ func (s *Scheduler) calcScore(nodes *map[string]*NodeUsage, resourceReqs device. sums += int(k.Nums) } - // Maintainer optimization (@Shouren): - // When init container requests are already covered by app - // container requests (!needsInitClone), skip fitInDevices - // for init containers entirely. Their device usage is // implicitly satisfied by the app container allocation since // init and app containers never run simultaneously — the node // only needs max(init, app) capacity at any moment. @@ -269,21 +229,8 @@ func (s *Scheduler) calcScore(nodes *map[string]*NodeUsage, resourceReqs device. // Decide which NodeUsage view to pass to fitInDevices. var workingNode *NodeUsage if needsInitClone && ctrid < numInitContainers { - // This is an init container that needs more resources than - // the app containers. Give it a fresh snapshot of the - // node's full capacity so it doesn't interfere with the - // app container pool. - workingNode = &NodeUsage{} - if err := json.Unmarshal(initialNodeBytes, workingNode); err != nil { - klog.ErrorS(err, "Failed to unmarshal node state for init container", - "nodeID", nodeID, "ctrid", ctrid) - errCh <- err - return - } + workingNode = node.DeepCopy() } else { - // Regular app container (or an init container whose - // request is already covered by the app allocation). - // Both deduct from the shared accumulated state. workingNode = appContainersNode } From 99d945f2a118035d0f65e9a53ba0ec88a69be4ec Mon Sep 17 00:00:00 2001 From: maishivamhoo123 Date: Fri, 15 May 2026 04:46:05 +0000 Subject: [PATCH 06/14] fix: add per-device memory validation for init containers Signed-off-by: maishivamhoo123 --- pkg/scheduler/score.go | 56 ++++++++++++----- pkg/scheduler/score_test.go | 118 ++++++++++++++++++++++++++++++++++++ 2 files changed, 158 insertions(+), 16 deletions(-) diff --git a/pkg/scheduler/score.go b/pkg/scheduler/score.go index f39f054dc..16c75609d 100644 --- a/pkg/scheduler/score.go +++ b/pkg/scheduler/score.go @@ -142,6 +142,32 @@ func podAppContainerTotalRequest(resourceReqs device.PodDeviceRequests, numInitC return total } +// canFitInitContainer checks if the node has at least one device that can +// individually satisfy the init container's resource requirements. +// This prevents allocation failures when total capacity appears sufficient +// but no single device has enough memory/cores. +func (node *NodeUsage) canFitInitContainer(maxInitReq containerResourceSummary) bool { + if node == nil { + return true // No node means no constraints + } + + for _, deviceList := range node.Devices.DeviceLists { + device := deviceList.Device + if device == nil { + continue + } + + availableMem := device.Totalmem - device.Usedmem + availableCores := device.Totalcore - device.Usedcores + + // Check if this device individually satisfies init requirements + if availableMem >= maxInitReq.memreq && availableCores >= 0 { + return true + } + } + return false +} + func (s *Scheduler) calcScore(nodes *map[string]*NodeUsage, resourceReqs device.PodDeviceRequests, task *corev1.Pod, failedNodes map[string]string) (*policy.NodeScoreList, error) { userNodePolicy := config.NodeSchedulerPolicy if task.GetAnnotations() != nil { @@ -168,6 +194,20 @@ func (s *Scheduler) calcScore(nodes *map[string]*NodeUsage, resourceReqs device. (maxInitReq.nums > appReqTotal.nums || maxInitReq.memreq > appReqTotal.memreq) for nodeID, node := range *nodes { + // Early exit: skip nodes that can't fit init containers on any single device + if numInitContainers > 0 && !node.canFitInitContainer(maxInitReq) { + failedNodesMutex.Lock() + failedNodes[nodeID] = fmt.Sprintf( + "no single device has sufficient memory (%dMiB) for init container", + maxInitReq.memreq) + for reasonType := range map[string]bool{"InsufficientInitDeviceMemory": true} { + failureReason[reasonType] = append(failureReason[reasonType], nodeID) + } + failedNodesMutex.Unlock() + klog.V(4).InfoS("Node filtered: insufficient per-device capacity for init container", + "pod", klog.KObj(task), "node", nodeID, "required_mem", maxInitReq.memreq) + continue + } wg.Add(1) go func(nodeID string, node *NodeUsage) { defer wg.Done() @@ -189,27 +229,17 @@ func (s *Scheduler) calcScore(nodes *map[string]*NodeUsage, resourceReqs device. var appContainersNode *NodeUsage if needsInitClone { - // Init containers request MORE than app containers (by count - // or memory). DeepCopy replaces JSON marshal/unmarshal appContainersNode = node.DeepCopy() } else { - // No init containers, or init requests are already covered - // by app container requests — use the node directly with no appContainersNode = node } - // This loop iterates over every container's device requests. - // resourceReqs is ordered: [initContainer_0, ..., initContainer_N-1, - // appContainer_0, ..., appContainer_M-1] for ctrid, n := range resourceReqs { sums := 0 for _, k := range n { sums += int(k.Nums) } - // implicitly satisfied by the app container allocation since - // init and app containers never run simultaneously — the node - // only needs max(init, app) capacity at any moment. if ctrid < numInitContainers && !needsInitClone { if deviceType != "" { score.Devices[deviceType] = append(score.Devices[deviceType], device.ContainerDevices{}) @@ -217,8 +247,6 @@ func (s *Scheduler) calcScore(nodes *map[string]*NodeUsage, resourceReqs device. continue } - // Container needs no device but a deviceType has already been - // identified — record an empty allocation and continue. if sums == 0 && deviceType != "" { score.Devices[deviceType] = append(score.Devices[deviceType], device.ContainerDevices{}) continue @@ -226,7 +254,6 @@ func (s *Scheduler) calcScore(nodes *map[string]*NodeUsage, resourceReqs device. klog.V(5).InfoS("fitInDevices", "pod", klog.KObj(task), "node", nodeID) - // Decide which NodeUsage view to pass to fitInDevices. var workingNode *NodeUsage if needsInitClone && ctrid < numInitContainers { workingNode = node.DeepCopy() @@ -236,8 +263,6 @@ func (s *Scheduler) calcScore(nodes *map[string]*NodeUsage, resourceReqs device. fit, reason := fitInDevices(workingNode, n, task, nodeInfo, &score.Devices) - // Fill any missing empty-allocation slots for containers - // that were processed before the first device-using container. for idx := range score.Devices { deviceType = idx for len(score.Devices[idx]) <= ctrid { @@ -273,7 +298,6 @@ func (s *Scheduler) calcScore(nodes *map[string]*NodeUsage, resourceReqs device. wg.Wait() close(errCh) - // Only record a filter failure event when no node could fit the pod. if len(res.NodeList) == 0 { for reasonType, failureNodes := range failureReason { sort.Strings(failureNodes) diff --git a/pkg/scheduler/score_test.go b/pkg/scheduler/score_test.go index 86b296b38..2e915cf55 100644 --- a/pkg/scheduler/score_test.go +++ b/pkg/scheduler/score_test.go @@ -402,6 +402,124 @@ func Test_calcScore(t *testing.T) { err: nil, }, }, + { + name: "init container requests more memory than any single device has (should filter node)", + args: struct { + nodes *map[string]*NodeUsage + nums device.PodDeviceRequests + annos map[string]string + task *corev1.Pod + }{ + nodes: &map[string]*NodeUsage{ + "node1": { + Node: &corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: "node1"}}, + Devices: policy.DeviceUsageList{ + Policy: util.NodeSchedulerPolicyBinpack.String(), + DeviceLists: []*policy.DeviceListsScore{ + { + Device: &device.DeviceUsage{ + ID: "uuid1", + Index: 0, + Used: 0, + Count: 10, + Usedmem: 7500, // Only 500 available (8000-7500) + Totalmem: 8000, + Totalcore: 100, + Usedcores: 0, + Numa: 0, + Type: nvidia.NvidiaGPUDevice, + Health: true, + }, + }, + { + Device: &device.DeviceUsage{ + ID: "uuid2", + Index: 1, + Used: 0, + Count: 10, + Usedmem: 7500, // Only 500 available (8000-7500) + Totalmem: 8000, + Totalcore: 100, + Usedcores: 0, + Numa: 0, + Type: nvidia.NvidiaGPUDevice, + Health: true, + }, + }, + }, + }, + }, + }, + nums: device.PodDeviceRequests{ + // Index 0: InitContainer requests 4000 (more than any single device has) + { + nvidia.NvidiaGPUDevice: device.ContainerDeviceRequest{ + Nums: 1, + Type: nvidia.NvidiaGPUDevice, + Memreq: 4000, // Each device only has 500 available + Coresreq: 30, + }, + }, + // Index 1: App container requests only 100 + { + nvidia.NvidiaGPUDevice: device.ContainerDeviceRequest{ + Nums: 1, + Type: nvidia.NvidiaGPUDevice, + Memreq: 100, + Coresreq: 30, + }, + }, + }, + annos: make(map[string]string), + task: &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-init-large-mem", + }, + Spec: corev1.PodSpec{ + InitContainers: []corev1.Container{ + { + Name: "init-large", + Image: "busybox", + Resources: corev1.ResourceRequirements{ + Limits: corev1.ResourceList{ + "hami.io/gpu": *resource.NewQuantity(1, resource.BinarySI), + "hami.io/gpucores": *resource.NewQuantity(30, resource.BinarySI), + "hami.io/gpumem": *resource.NewQuantity(4000, resource.BinarySI), + }, + }, + }, + }, + Containers: []corev1.Container{ + { + Name: "app-small", + Image: "busybox", + Resources: corev1.ResourceRequirements{ + Limits: corev1.ResourceList{ + "hami.io/gpu": *resource.NewQuantity(1, resource.BinarySI), + "hami.io/gpucores": *resource.NewQuantity(30, resource.BinarySI), + "hami.io/gpumem": *resource.NewQuantity(100, resource.BinarySI), + }, + }, + }, + }, + }, + }, + }, + wants: struct { + want *policy.NodeScoreList + failedNodes map[string]string + err error + }{ + want: &policy.NodeScoreList{ + Policy: util.NodeSchedulerPolicyBinpack.String(), + NodeList: []*policy.NodeScore{}, // Node should be filtered out + }, + failedNodes: map[string]string{ + "node1": "no single device has sufficient memory (4000MiB) for init container", + }, + err: nil, + }, + }, { name: "one node two device one pod one container use one device,but having use 50%", args: struct { From a680055a309a851035f602bbfd3dc651beeae728 Mon Sep 17 00:00:00 2001 From: maishivamhoo123 Date: Fri, 15 May 2026 09:21:32 +0000 Subject: [PATCH 07/14] suggested changes in score and score_test Signed-off-by: maishivamhoo123 --- pkg/scheduler/score.go | 26 ++++++++++++++++++++++++++ pkg/scheduler/score_test.go | 12 +++++++++--- 2 files changed, 35 insertions(+), 3 deletions(-) diff --git a/pkg/scheduler/score.go b/pkg/scheduler/score.go index 16c75609d..2ae9071ee 100644 --- a/pkg/scheduler/score.go +++ b/pkg/scheduler/score.go @@ -286,6 +286,32 @@ func (s *Scheduler) calcScore(nodes *map[string]*NodeUsage, resourceReqs device. } } + // when needsInitClone=false, so the device plugin knows which devices to use + if numInitContainers > 0 && !needsInitClone { + for deviceType := range score.Devices { + containerList := score.Devices[deviceType] + // We need to fill empty init slots with app container allocations + + if len(containerList) > numInitContainers { + firstAppAllocation := containerList[numInitContainers] + + // Updated to modern Go range-over-int syntax + for initIdx := range numInitContainers { + if initIdx < len(containerList) { + containerList[initIdx] = firstAppAllocation + klog.V(4).InfoS( + "Copied app container device allocation to init container slot", + "pod", klog.KObj(task), + "init_idx", initIdx, + "device_type", deviceType, + "allocation", firstAppAllocation, + ) + } + } + } + } + } + if ctrfit { fitNodesMutex.Lock() res.NodeList = append(res.NodeList, &score) diff --git a/pkg/scheduler/score_test.go b/pkg/scheduler/score_test.go index 2e915cf55..ca83a93eb 100644 --- a/pkg/scheduler/score_test.go +++ b/pkg/scheduler/score_test.go @@ -2679,9 +2679,15 @@ func Test_calcScore(t *testing.T) { NodeID: "node1", Devices: device.PodDevices{ "NVIDIA": device.PodSingleDevice{ - // Index 0: InitContainer skipped (covered by app allocation) - // since init request == app request, no clone needed. - {}, + { + { + Idx: 0, + UUID: "uuid2", + Type: nvidia.NvidiaGPUDevice, + Usedcores: 30, + Usedmem: 1000, + }, + }, // Index 1: Regular Container allocated normally. { { From 1de5b181f7d238a13b61d52e17ffd54a3162c245 Mon Sep 17 00:00:00 2001 From: maishivamhoo123 Date: Sun, 28 Jun 2026 10:55:42 +0000 Subject: [PATCH 08/14] Added and tested Signed-off-by: maishivamhoo123 --- pkg/scheduler/score.go | 107 +++++++++++++++++---------------- pkg/scheduler/score_test.go | 116 ++++++++++++++++++++++++++++++++++++ 2 files changed, 173 insertions(+), 50 deletions(-) diff --git a/pkg/scheduler/score.go b/pkg/scheduler/score.go index 2ae9071ee..151770257 100644 --- a/pkg/scheduler/score.go +++ b/pkg/scheduler/score.go @@ -35,8 +35,9 @@ import ( // containerResourceSummary holds both the GPU count and total memory // for a group of containers, so we can compare them on both dimensions. type containerResourceSummary struct { - nums int - memreq int32 + nums int + memreq int32 + coresreq int32 } func viewStatus(usage NodeUsage) { @@ -117,13 +118,16 @@ func podInitContainerMaxRequest(resourceReqs device.PodDeviceRequests, numInitCo for i := range numInitContainers { var nums int var mem int32 + var cores int32 for _, k := range resourceReqs[i] { nums += int(k.Nums) mem += k.Memreq + cores += k.Coresreq } if nums > maxReq.nums || (nums == maxReq.nums && mem > maxReq.memreq) { maxReq.nums = nums maxReq.memreq = mem + maxReq.coresreq = cores } } return maxReq @@ -148,7 +152,7 @@ func podAppContainerTotalRequest(resourceReqs device.PodDeviceRequests, numInitC // but no single device has enough memory/cores. func (node *NodeUsage) canFitInitContainer(maxInitReq containerResourceSummary) bool { if node == nil { - return true // No node means no constraints + return false // nil node cannot satisfy any request } for _, deviceList := range node.Devices.DeviceLists { @@ -161,7 +165,7 @@ func (node *NodeUsage) canFitInitContainer(maxInitReq containerResourceSummary) availableCores := device.Totalcore - device.Usedcores // Check if this device individually satisfies init requirements - if availableMem >= maxInitReq.memreq && availableCores >= 0 { + if availableMem >= maxInitReq.memreq && availableCores >= maxInitReq.coresreq { return true } } @@ -200,9 +204,7 @@ func (s *Scheduler) calcScore(nodes *map[string]*NodeUsage, resourceReqs device. failedNodes[nodeID] = fmt.Sprintf( "no single device has sufficient memory (%dMiB) for init container", maxInitReq.memreq) - for reasonType := range map[string]bool{"InsufficientInitDeviceMemory": true} { - failureReason[reasonType] = append(failureReason[reasonType], nodeID) - } + failureReason["InsufficientInitDeviceMemory"] = append(failureReason["InsufficientInitDeviceMemory"], nodeID) failedNodesMutex.Unlock() klog.V(4).InfoS("Node filtered: insufficient per-device capacity for init container", "pod", klog.KObj(task), "node", nodeID, "required_mem", maxInitReq.memreq) @@ -224,9 +226,6 @@ func (s *Scheduler) calcScore(nodes *map[string]*NodeUsage, resourceReqs device. return } - ctrfit := true - deviceType := "" - var appContainersNode *NodeUsage if needsInitClone { appContainersNode = node.DeepCopy() @@ -234,68 +233,76 @@ func (s *Scheduler) calcScore(nodes *map[string]*NodeUsage, resourceReqs device. appContainersNode = node } + ctrfit := true for ctrid, n := range resourceReqs { sums := 0 for _, k := range n { sums += int(k.Nums) } - if ctrid < numInitContainers && !needsInitClone { - if deviceType != "" { - score.Devices[deviceType] = append(score.Devices[deviceType], device.ContainerDevices{}) - } - continue - } - - if sums == 0 && deviceType != "" { - score.Devices[deviceType] = append(score.Devices[deviceType], device.ContainerDevices{}) - continue - } - - klog.V(5).InfoS("fitInDevices", "pod", klog.KObj(task), "node", nodeID) - - var workingNode *NodeUsage - if needsInitClone && ctrid < numInitContainers { - workingNode = node.DeepCopy() - } else { - workingNode = appContainersNode + // Capture the set of device types that existed BEFORE processing this container + existingBefore := make(map[string]bool) + for typ := range score.Devices { + existingBefore[typ] = true } - fit, reason := fitInDevices(workingNode, n, task, nodeInfo, &score.Devices) + // Skip device allocation for init containers if they don't need cloning. + if ctrid < numInitContainers && !needsInitClone { + // Do nothing. Let it fall through to normalization. + } else if sums > 0 { + klog.V(5).InfoS("fitInDevices", "pod", klog.KObj(task), "node", nodeID) + + var workingNode *NodeUsage + if needsInitClone && ctrid < numInitContainers { + workingNode = node.DeepCopy() + } else { + workingNode = appContainersNode + } - for idx := range score.Devices { - deviceType = idx - for len(score.Devices[idx]) <= ctrid { - emptyContainerDevices := device.ContainerDevices{} - emptyPodSingleDevice := device.PodSingleDevice{} - emptyPodSingleDevice = append(emptyPodSingleDevice, emptyContainerDevices) - score.Devices[idx] = append(emptyPodSingleDevice, score.Devices[idx]...) + fit, reason := fitInDevices(workingNode, n, task, nodeInfo, &score.Devices) + ctrfit = fit + if !fit { + klog.V(4).InfoS(common.NodeUnfitPod, "pod", klog.KObj(task), "node", nodeID, "reason", reason) + failedNodesMutex.Lock() + failedNodes[nodeID] = common.NodeUnfitPod + for reasonType := range common.ParseReason(reason) { + failureReason[reasonType] = append(failureReason[reasonType], nodeID) + } + failedNodesMutex.Unlock() + break } } - ctrfit = fit - if !fit { - klog.V(4).InfoS(common.NodeUnfitPod, "pod", klog.KObj(task), "node", nodeID, "reason", reason) - failedNodesMutex.Lock() - failedNodes[nodeID] = common.NodeUnfitPod - for reasonType := range common.ParseReason(reason) { - failureReason[reasonType] = append(failureReason[reasonType], nodeID) + // NORMALIZATION: Ensure every device slice has exactly ctrid+1 entries, + // keeping previous allocations at their original indices. + for typ, devSlice := range score.Devices { + if len(devSlice) < ctrid+1 { + if !existingBefore[typ] { + // Brand new type for this container: allocation is at devSlice[0], + // shift it to index ctrid by prepending non‑nil empty slices. + pad := make(device.PodSingleDevice, 0, ctrid) + for i := 0; i < ctrid; i++ { + pad = append(pad, device.ContainerDevices{}) + } + score.Devices[typ] = append(pad, devSlice...) + } else { + // Type existed before: just append empty (non‑nil) slices to reach the needed length. + for len(score.Devices[typ]) < ctrid+1 { + score.Devices[typ] = append(score.Devices[typ], device.ContainerDevices{}) + } + } } - failedNodesMutex.Unlock() - break } } - // when needsInitClone=false, so the device plugin knows which devices to use + // When needsInitClone=false, copy the first app container's allocation + // into the init container slots so the device plugin knows which devices to use. if numInitContainers > 0 && !needsInitClone { for deviceType := range score.Devices { containerList := score.Devices[deviceType] - // We need to fill empty init slots with app container allocations - if len(containerList) > numInitContainers { firstAppAllocation := containerList[numInitContainers] - // Updated to modern Go range-over-int syntax for initIdx := range numInitContainers { if initIdx < len(containerList) { containerList[initIdx] = firstAppAllocation diff --git a/pkg/scheduler/score_test.go b/pkg/scheduler/score_test.go index ca83a93eb..c6531fe23 100644 --- a/pkg/scheduler/score_test.go +++ b/pkg/scheduler/score_test.go @@ -2820,6 +2820,122 @@ func Test_calcScore(t *testing.T) { err: nil, }, }, + { + name: "init container requests more cores than any single device has (should filter node)", + args: struct { + nodes *map[string]*NodeUsage + nums device.PodDeviceRequests + annos map[string]string + task *corev1.Pod + }{ + nodes: &map[string]*NodeUsage{ + "node1": { + Node: &corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: "node1"}}, + Devices: policy.DeviceUsageList{ + Policy: util.NodeSchedulerPolicyBinpack.String(), + DeviceLists: []*policy.DeviceListsScore{ + { + Device: &device.DeviceUsage{ + ID: "uuid1", + Index: 0, + Used: 0, + Count: 10, + Usedmem: 0, + Totalmem: 8000, + Totalcore: 100, + Usedcores: 80, // only 20 cores free + Numa: 0, + Type: nvidia.NvidiaGPUDevice, + Health: true, + }, + }, + { + Device: &device.DeviceUsage{ + ID: "uuid2", + Index: 1, + Used: 0, + Count: 10, + Usedmem: 0, + Totalmem: 8000, + Totalcore: 100, + Usedcores: 80, // same, only 20 cores free + Numa: 0, + Type: nvidia.NvidiaGPUDevice, + Health: true, + }, + }, + }, + }, + }, + }, + nums: device.PodDeviceRequests{ + // Index 0: InitContainer requests 30 cores (more than any single device's 20 free) + { + nvidia.NvidiaGPUDevice: device.ContainerDeviceRequest{ + Nums: 1, + Type: nvidia.NvidiaGPUDevice, + Memreq: 1000, + Coresreq: 30, + }, + }, + // Index 1: App container requests 10 cores (fits, but init should block the node) + { + nvidia.NvidiaGPUDevice: device.ContainerDeviceRequest{ + Nums: 1, + Type: nvidia.NvidiaGPUDevice, + Memreq: 1000, + Coresreq: 10, + }, + }, + }, + annos: make(map[string]string), + task: &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: "test-init-large-cores"}, + Spec: corev1.PodSpec{ + InitContainers: []corev1.Container{ + { + Name: "init-core-heavy", + Image: "busybox", + Resources: corev1.ResourceRequirements{ + Limits: corev1.ResourceList{ + "hami.io/gpu": *resource.NewQuantity(1, resource.BinarySI), + "hami.io/gpucores": *resource.NewQuantity(30, resource.BinarySI), + "hami.io/gpumem": *resource.NewQuantity(1000, resource.BinarySI), + }, + }, + }, + }, + Containers: []corev1.Container{ + { + Name: "app-light", + Image: "busybox", + Resources: corev1.ResourceRequirements{ + Limits: corev1.ResourceList{ + "hami.io/gpu": *resource.NewQuantity(1, resource.BinarySI), + "hami.io/gpucores": *resource.NewQuantity(10, resource.BinarySI), + "hami.io/gpumem": *resource.NewQuantity(1000, resource.BinarySI), + }, + }, + }, + }, + }, + }, + }, + wants: struct { + want *policy.NodeScoreList + failedNodes map[string]string + err error + }{ + want: &policy.NodeScoreList{ + Policy: util.NodeSchedulerPolicyBinpack.String(), + NodeList: []*policy.NodeScore{}, + }, + failedNodes: map[string]string{ + "node1": "no single device has sufficient memory (1000MiB) for init container", + }, + err: nil, + }, + }, { name: "two node per node having one device one pod two container use one device", args: struct { From b5fdd2913c87edd01a30675f768e3f579bcc1b99 Mon Sep 17 00:00:00 2001 From: maishivamhoo123 Date: Sun, 28 Jun 2026 12:56:45 +0000 Subject: [PATCH 09/14] Removed lint Signed-off-by: maishivamhoo123 --- pkg/scheduler/score.go | 59 ++++++++++++++++++++++++++++++++++-------- 1 file changed, 48 insertions(+), 11 deletions(-) diff --git a/pkg/scheduler/score.go b/pkg/scheduler/score.go index 151770257..c0f9b5714 100644 --- a/pkg/scheduler/score.go +++ b/pkg/scheduler/score.go @@ -141,6 +141,7 @@ func podAppContainerTotalRequest(resourceReqs device.PodDeviceRequests, numInitC for _, k := range resourceReqs[i] { total.nums += int(k.Nums) total.memreq += k.Memreq + total.coresreq += k.Coresreq } } return total @@ -195,19 +196,49 @@ func (s *Scheduler) calcScore(nodes *map[string]*NodeUsage, resourceReqs device. appReqTotal := podAppContainerTotalRequest(resourceReqs, numInitContainers) needsInitClone := numInitContainers > 0 && - (maxInitReq.nums > appReqTotal.nums || maxInitReq.memreq > appReqTotal.memreq) + (maxInitReq.nums > appReqTotal.nums || + maxInitReq.memreq > appReqTotal.memreq || + maxInitReq.coresreq > appReqTotal.coresreq) for nodeID, node := range *nodes { // Early exit: skip nodes that can't fit init containers on any single device if numInitContainers > 0 && !node.canFitInitContainer(maxInitReq) { failedNodesMutex.Lock() - failedNodes[nodeID] = fmt.Sprintf( - "no single device has sufficient memory (%dMiB) for init container", - maxInitReq.memreq) - failureReason["InsufficientInitDeviceMemory"] = append(failureReason["InsufficientInitDeviceMemory"], nodeID) + // FIXED: Check whether failure is due to memory or cores, and provide appropriate error message + var failureMsg string + var failureType string + hasDeviceWithEnoughMem := false + for _, deviceList := range node.Devices.DeviceLists { + device := deviceList.Device + if device == nil { + continue + } + availableMem := device.Totalmem - device.Usedmem + if availableMem >= maxInitReq.memreq { + hasDeviceWithEnoughMem = true + break + } + } + + if hasDeviceWithEnoughMem { + // Failure is due to cores, not memory + failureMsg = fmt.Sprintf( + "no single device has sufficient cores (%d) for init container", + maxInitReq.coresreq) + failureType = "InsufficientInitDeviceCores" + } else { + // Failure is due to memory + failureMsg = fmt.Sprintf( + "no single device has sufficient memory (%dMiB) for init container", + maxInitReq.memreq) + failureType = "InsufficientInitDeviceMemory" + } + + failedNodes[nodeID] = failureMsg + failureReason[failureType] = append(failureReason[failureType], nodeID) failedNodesMutex.Unlock() klog.V(4).InfoS("Node filtered: insufficient per-device capacity for init container", - "pod", klog.KObj(task), "node", nodeID, "required_mem", maxInitReq.memreq) + "pod", klog.KObj(task), "node", nodeID, "required_mem", maxInitReq.memreq, "required_cores", maxInitReq.coresreq) continue } wg.Add(1) @@ -281,7 +312,7 @@ func (s *Scheduler) calcScore(nodes *map[string]*NodeUsage, resourceReqs device. // Brand new type for this container: allocation is at devSlice[0], // shift it to index ctrid by prepending non‑nil empty slices. pad := make(device.PodSingleDevice, 0, ctrid) - for i := 0; i < ctrid; i++ { + for range ctrid { pad = append(pad, device.ContainerDevices{}) } score.Devices[typ] = append(pad, devSlice...) @@ -297,21 +328,27 @@ func (s *Scheduler) calcScore(nodes *map[string]*NodeUsage, resourceReqs device. // When needsInitClone=false, copy the first app container's allocation // into the init container slots so the device plugin knows which devices to use. + // FIXED: Aggregate all app container allocations instead of just copying the first app container if numInitContainers > 0 && !needsInitClone { for deviceType := range score.Devices { containerList := score.Devices[deviceType] if len(containerList) > numInitContainers { - firstAppAllocation := containerList[numInitContainers] + // Aggregate allocations from all app containers + aggregatedAllocation := device.ContainerDevices{} + for appIdx := numInitContainers; appIdx < len(containerList); appIdx++ { + aggregatedAllocation = append(aggregatedAllocation, containerList[appIdx]...) + } + // Copy aggregated allocation to all init container slots for initIdx := range numInitContainers { if initIdx < len(containerList) { - containerList[initIdx] = firstAppAllocation + containerList[initIdx] = aggregatedAllocation klog.V(4).InfoS( - "Copied app container device allocation to init container slot", + "Copied aggregated app container device allocation to init container slot", "pod", klog.KObj(task), "init_idx", initIdx, "device_type", deviceType, - "allocation", firstAppAllocation, + "allocation", aggregatedAllocation, ) } } From b28b13523e288baa4d5ab45bc5b95481d3dec77f Mon Sep 17 00:00:00 2001 From: maishivamhoo123 Date: Wed, 1 Jul 2026 04:40:08 +0000 Subject: [PATCH 10/14] Added all the changes Signed-off-by: maishivamhoo123 --- pkg/scheduler/score.go | 94 +++++++++++++++++++++++++++---------- pkg/scheduler/score_test.go | 2 +- 2 files changed, 69 insertions(+), 27 deletions(-) diff --git a/pkg/scheduler/score.go b/pkg/scheduler/score.go index c0f9b5714..43c3b58be 100644 --- a/pkg/scheduler/score.go +++ b/pkg/scheduler/score.go @@ -201,13 +201,13 @@ func (s *Scheduler) calcScore(nodes *map[string]*NodeUsage, resourceReqs device. maxInitReq.coresreq > appReqTotal.coresreq) for nodeID, node := range *nodes { - // Early exit: skip nodes that can't fit init containers on any single device if numInitContainers > 0 && !node.canFitInitContainer(maxInitReq) { failedNodesMutex.Lock() - // FIXED: Check whether failure is due to memory or cores, and provide appropriate error message + var failureMsg string var failureType string hasDeviceWithEnoughMem := false + for _, deviceList := range node.Devices.DeviceLists { device := deviceList.Device if device == nil { @@ -241,6 +241,7 @@ func (s *Scheduler) calcScore(nodes *map[string]*NodeUsage, resourceReqs device. "pod", klog.KObj(task), "node", nodeID, "required_mem", maxInitReq.memreq, "required_cores", maxInitReq.coresreq) continue } + wg.Add(1) go func(nodeID string, node *NodeUsage) { defer wg.Done() @@ -271,15 +272,12 @@ func (s *Scheduler) calcScore(nodes *map[string]*NodeUsage, resourceReqs device. sums += int(k.Nums) } - // Capture the set of device types that existed BEFORE processing this container existingBefore := make(map[string]bool) for typ := range score.Devices { existingBefore[typ] = true } - // Skip device allocation for init containers if they don't need cloning. if ctrid < numInitContainers && !needsInitClone { - // Do nothing. Let it fall through to normalization. } else if sums > 0 { klog.V(5).InfoS("fitInDevices", "pod", klog.KObj(task), "node", nodeID) @@ -304,8 +302,6 @@ func (s *Scheduler) calcScore(nodes *map[string]*NodeUsage, resourceReqs device. } } - // NORMALIZATION: Ensure every device slice has exactly ctrid+1 entries, - // keeping previous allocations at their original indices. for typ, devSlice := range score.Devices { if len(devSlice) < ctrid+1 { if !existingBefore[typ] { @@ -326,32 +322,78 @@ func (s *Scheduler) calcScore(nodes *map[string]*NodeUsage, resourceReqs device. } } - // When needsInitClone=false, copy the first app container's allocation - // into the init container slots so the device plugin knows which devices to use. - // FIXED: Aggregate all app container allocations instead of just copying the first app container if numInitContainers > 0 && !needsInitClone { for deviceType := range score.Devices { containerList := score.Devices[deviceType] - if len(containerList) > numInitContainers { - // Aggregate allocations from all app containers - aggregatedAllocation := device.ContainerDevices{} - for appIdx := numInitContainers; appIdx < len(containerList); appIdx++ { - aggregatedAllocation = append(aggregatedAllocation, containerList[appIdx]...) + + if len(containerList) <= numInitContainers { + continue + } + + for initIdx := range numInitContainers { + if initIdx >= len(resourceReqs) { + // Shouldn't happen, but safety check + continue + } + + // ContainerDeviceRequests is map[string]ContainerDeviceRequest keyed + // by device type, so this is a direct lookup, not a search. + initDeviceReq, found := resourceReqs[initIdx][deviceType] + if !found { + containerList[initIdx] = device.ContainerDevices{} + continue } - // Copy aggregated allocation to all init container slots - for initIdx := range numInitContainers { - if initIdx < len(containerList) { - containerList[initIdx] = aggregatedAllocation - klog.V(4).InfoS( - "Copied aggregated app container device allocation to init container slot", - "pod", klog.KObj(task), - "init_idx", initIdx, - "device_type", deviceType, - "allocation", aggregatedAllocation, - ) + selectedAllocation := device.ContainerDevices{} + devicesNeeded := int(initDeviceReq.Nums) + memNeeded := initDeviceReq.Memreq + coresNeeded := initDeviceReq.Coresreq + + outerLoop: + for appIdx := numInitContainers; appIdx < len(containerList); appIdx++ { + appAllocation := containerList[appIdx] + + for deviceIdx := range appAllocation { + if devicesNeeded == 0 { + break outerLoop + } + + dev := appAllocation[deviceIdx] + + if dev.Usedmem >= memNeeded && dev.Usedcores >= coresNeeded { + selectedAllocation = append(selectedAllocation, dev) + devicesNeeded-- + } } } + + // Verify we found enough valid devices before assigning + if len(selectedAllocation) == int(initDeviceReq.Nums) { + containerList[initIdx] = selectedAllocation + klog.V(4).InfoS( + "Assigned validated app devices to init container", + "pod", klog.KObj(task), + "init_index", initIdx, + "device_type", deviceType, + "devices_assigned", len(selectedAllocation), + "required_mem", memNeeded, + "required_cores", coresNeeded, + ) + } else { + klog.V(3).InfoS( + "ERROR: Insufficient validated devices for init container after canFitInitContainer passed", + "pod", klog.KObj(task), + "init_index", initIdx, + "device_type", deviceType, + "devices_needed", int(initDeviceReq.Nums), + "devices_found", len(selectedAllocation), + "required_mem", memNeeded, + "required_cores", coresNeeded, + ) + // Still assign what we found (partial allocation) + // The device plugin will handle the error downstream + containerList[initIdx] = selectedAllocation + } } } } diff --git a/pkg/scheduler/score_test.go b/pkg/scheduler/score_test.go index c6531fe23..81686c5e0 100644 --- a/pkg/scheduler/score_test.go +++ b/pkg/scheduler/score_test.go @@ -2931,7 +2931,7 @@ func Test_calcScore(t *testing.T) { NodeList: []*policy.NodeScore{}, }, failedNodes: map[string]string{ - "node1": "no single device has sufficient memory (1000MiB) for init container", + "node1": "no single device has sufficient cores (30) for init container", }, err: nil, }, From 55556da250b5d35035e943dc5e6e0a45c4e26dc5 Mon Sep 17 00:00:00 2001 From: maishivamhoo123 Date: Mon, 6 Jul 2026 08:00:25 +0000 Subject: [PATCH 11/14] modiefied the sequemsital logic for initcontainer and added the test Signed-off-by: maishivamhoo123 --- pkg/device/common/common.go | 2 + pkg/scheduler/scheduler.go | 74 ++-- pkg/scheduler/scheduler_test.go | 4 +- pkg/scheduler/score.go | 585 ++++++++++++++++++-------------- pkg/scheduler/score_test.go | 148 +++++++- 5 files changed, 512 insertions(+), 301 deletions(-) diff --git a/pkg/device/common/common.go b/pkg/device/common/common.go index 6405dc8dc..174be2bf1 100644 --- a/pkg/device/common/common.go +++ b/pkg/device/common/common.go @@ -18,6 +18,7 @@ package common import ( "fmt" + "sort" "strings" ) @@ -45,6 +46,7 @@ func GenReason(reasons map[string]int, cards int) string { for r, cnt := range reasons { reason = append(reason, fmt.Sprintf("%d/%d %s", cnt, cards, r)) } + sort.Strings(reason) return strings.Join(reason, ", ") } diff --git a/pkg/scheduler/scheduler.go b/pkg/scheduler/scheduler.go index a3a947c84..e3ca3e791 100644 --- a/pkg/scheduler/scheduler.go +++ b/pkg/scheduler/scheduler.go @@ -94,8 +94,6 @@ func NewScheduler() *Scheduler { s.nodeManager = newNodeManager() s.podManager = device.NewPodManager() s.quotaManager = device.NewQuotaManager() - // Use dummy leader manager when leaderElect is disabled - // This ensures IsLeader() always returns true and synced will not be set to false s.leaderManager = leaderelection.NewDummyLeaderManager(true) if config.LeaderElect { callbacks := leaderelection.LeaderCallbacks{ @@ -160,6 +158,22 @@ func (s *Scheduler) onAddPod(obj any) { klog.ErrorS(err, "failed to decode pod devices", "pod", klog.KObj(pod)) return } + + numInit := len(pod.Spec.InitContainers) + if numInit > 0 { + allInitDone := len(pod.Status.InitContainerStatuses) >= numInit + for _, cs := range pod.Status.InitContainerStatuses { + if cs.State.Terminated == nil || cs.State.Terminated.ExitCode != 0 { + allInitDone = false + break + } + } + if allInitDone { + + podDev = stripInitContainerAliasSlots(pod, nil, podDev) + } + } + if s.podManager.AddPod(pod, nodeID, podDev) { s.quotaManager.AddUsage(pod, podDev) } @@ -403,7 +417,14 @@ func (s *Scheduler) register(labelSelector labels.Selector, printedLog map[strin klog.V(5).InfoS("Skipping device cleanup for vendor not present in scheduler cache", "nodeName", val.Name, "deviceVendor", devhandsk) continue } - klog.Warning("Device is unhealthy, cleaning up node", "nodeName", val.Name, "deviceVendor", devhandsk) + // klog.Warning does plain fmt.Print-style concatenation of its arguments - + // klog v2 has no structured WarningS variant. Passing alternating + // "key", value pairs to it (as if it were InfoS/ErrorS) produces a garbled, + // unstructured log line instead of the intended structured fields. Use + // ErrorS (nil error is fine here; this is a detected condition, not a Go + // error) to match the structured logging used throughout the rest of this + // file. + klog.ErrorS(nil, "Device is unhealthy, cleaning up node", "nodeName", val.Name, "deviceVendor", devhandsk) err := devInstance.NodeCleanUp(val.Name) if err != nil { klog.ErrorS(err, "Node cleanup failed", "nodeName", val.Name, "deviceVendor", devhandsk) @@ -739,41 +760,24 @@ func (s *Scheduler) Filter(args extenderv1.ExtenderArgs) (*extenderv1.ExtenderFi klog.InfoS("Starting schedule filter process", "pod", args.Pod.Name, "uuid", args.Pod.UID, "namespace", args.Pod.Namespace) resourceReqs := device.Resourcereqs(args.Pod) - initReqTotal := 0 - appReqTotal := 0 - numInitContainers := len(args.Pod.Spec.InitContainers) + hasHAMiResource := false - for i := range numInitContainers { - currentInitReq := 0 - for _, k := range resourceReqs[i] { - currentInitReq += int(k.Nums) - } - if currentInitReq > initReqTotal { - initReqTotal = currentInitReq - } - } - - // 2. Calculate the SUM of requests among Regular Containers - for i := numInitContainers; i < len(resourceReqs); i++ { - for _, k := range resourceReqs[i] { - appReqTotal += int(k.Nums) + for _, reqMap := range resourceReqs { + if len(reqMap) > 0 { + hasHAMiResource = true + break } } - // 3. The effective total request is the MAX of (InitContainers, RegularContainers) - resourceReqTotal := appReqTotal - resourceReqTotal = max(resourceReqTotal, initReqTotal) - - if resourceReqTotal == 0 { - klog.V(1).InfoS("Pod does not request any resources", - "pod", args.Pod.Name) - s.recordScheduleFilterResultEvent(args.Pod, EventReasonFilteringFailed, "", fmt.Errorf("does not request any resource")) + if !hasHAMiResource { + klog.V(1).InfoS("Pod does not request any resources", "pod", args.Pod.Name) return &extenderv1.ExtenderFilterResult{ NodeNames: args.NodeNames, FailedNodes: nil, Error: "", }, nil } + if pi, ok := s.podManager.TakeAndDeletePod(args.Pod); ok { s.quotaManager.RmUsage(args.Pod, pi.Devices) } @@ -783,8 +787,7 @@ func (s *Scheduler) Filter(args extenderv1.ExtenderArgs) (*extenderv1.ExtenderFi return nil, err } if len(failedNodes) != 0 { - klog.V(5).InfoS("Nodes failed during usage retrieval", - "nodes", failedNodes) + klog.V(5).InfoS("Nodes failed during usage retrieval", "nodes", failedNodes) } nodeScores, err := s.calcScore(nodeUsage, resourceReqs, args.Pod, failedNodes) if err != nil { @@ -793,8 +796,7 @@ func (s *Scheduler) Filter(args extenderv1.ExtenderArgs) (*extenderv1.ExtenderFi return nil, err } if len((*nodeScores).NodeList) == 0 { - klog.V(4).InfoS("No available nodes meet the required scores", - "pod", args.Pod.Name) + klog.V(4).InfoS("No available nodes meet the required scores", "pod", args.Pod.Name) s.recordScheduleFilterResultEvent(args.Pod, EventReasonFilteringFailed, "", fmt.Errorf("no available node, %d nodes do not meet", len(*args.NodeNames))) return &extenderv1.ExtenderFilterResult{ FailedNodes: failedNodes, @@ -816,16 +818,18 @@ func (s *Scheduler) Filter(args extenderv1.ExtenderArgs) (*extenderv1.ExtenderFi val.PatchAnnotations(args.Pod, &annotations, m.Devices) } - added := s.podManager.AddPod(args.Pod, m.NodeID, m.Devices) + usageDevices := stripInitContainerAliasSlots(args.Pod, resourceReqs, m.Devices) + + added := s.podManager.AddPod(args.Pod, m.NodeID, usageDevices) if added { - s.quotaManager.AddUsage(args.Pod, m.Devices) + s.quotaManager.AddUsage(args.Pod, usageDevices) } err = util.PatchPodAnnotations(args.Pod, annotations) if err != nil { s.recordScheduleFilterResultEvent(args.Pod, EventReasonFilteringFailed, "", err) if added { - s.quotaManager.RmUsage(args.Pod, m.Devices) + s.quotaManager.RmUsage(args.Pod, usageDevices) } s.podManager.DelPod(args.Pod) return nil, err diff --git a/pkg/scheduler/scheduler_test.go b/pkg/scheduler/scheduler_test.go index afdfd7965..15bdacb84 100644 --- a/pkg/scheduler/scheduler_test.go +++ b/pkg/scheduler/scheduler_test.go @@ -1287,7 +1287,7 @@ func Test_ResourceQuota(t *testing.T) { wantErr: nil, want: &extenderv1.ExtenderFilterResult{ FailedNodes: map[string]string{ - "node1": "NodeUnfitPod", + "node1": "1/4 AllocatedCardsInsufficientRequest, 3/4 ResourceQuotaNotFit", }, }, }, @@ -1375,7 +1375,7 @@ func Test_ResourceQuota(t *testing.T) { wantErr: nil, want: &extenderv1.ExtenderFilterResult{ FailedNodes: map[string]string{ - "node1": "NodeUnfitPod", + "node1": "4/4 ResourceQuotaNotFit", }, }, }, diff --git a/pkg/scheduler/score.go b/pkg/scheduler/score.go index 43c3b58be..fdc987aeb 100644 --- a/pkg/scheduler/score.go +++ b/pkg/scheduler/score.go @@ -32,8 +32,7 @@ import ( "github.com/Project-HAMi/HAMi/pkg/util" ) -// containerResourceSummary holds both the GPU count and total memory -// for a group of containers, so we can compare them on both dimensions. +// containerResourceSummary holds both the GPU count and total memory for a group of containers, so we can compare them on both dimensions. type containerResourceSummary struct { nums int memreq int32 @@ -47,130 +46,230 @@ func viewStatus(usage NodeUsage) { } } +// getNodeResources returns all devices of the exact given type from the node. func getNodeResources(list NodeUsage, t string) []*device.DeviceUsage { l := []*device.DeviceUsage{} for _, val := range list.Devices.DeviceLists { - if strings.Contains(val.Device.Type, t) { + if val.Device == nil { + continue + } + // Exact match or prefix match (e.g., "NVIDIA" matches "NVIDIA A100-SXM4-40GB") + if val.Device.Type == t || strings.HasPrefix(val.Device.Type, t+" ") { l = append(l, val.Device) } } return l } +// getDeviceBaseType maps a device model string (e.g., "NVIDIA A100-SXM4-40GB") to the base type key used in the device registry (e.g., "NVIDIA"). +func getDeviceBaseType(model string) string { + bestName := "" + bestWordLen := -1 + for name, dev := range device.GetDevices() { + word := dev.CommonWord() + if !strings.HasPrefix(model, word) { + continue + } + if len(word) > bestWordLen || (len(word) == bestWordLen && name < bestName) { + bestName = name + bestWordLen = len(word) + } + } + if bestWordLen == -1 { + return model + } + return bestName +} + +// nodeDeviceBaseTypes returns the set of base device types physically present on the node, computed once so that callers don't repeatedly re-derive it (which, combined with any residual +// ambiguity in getDeviceBaseType, previously risked producing different results across calls within the same scheduling attempt). +func nodeDeviceBaseTypes(list policy.DeviceUsageList) map[string]struct{} { + types := make(map[string]struct{}) + for _, dl := range list.DeviceLists { + if dl.Device != nil { + types[getDeviceBaseType(dl.Device.Type)] = struct{}{} + } + } + return types +} + +// fitInDevices tries to allocate a single container's device requests on the given node. func fitInDevices(node *NodeUsage, requests device.ContainerDeviceRequests, pod *corev1.Pod, nodeInfo *device.NodeInfo, devinput *device.PodDevices) (bool, string) { - devs := device.ContainerDevices{} - total, totalCore, totalMem := int32(0), int32(0), int32(0) - free, freeCore, freeMem := int32(0), int32(0), int32(0) - sums := 0 - // compute all device scores for one node + // Snapshot the entire node state before any modifications. + type devSnapshot struct { + idx int + used int32 + usedcores int32 + usedmem int32 + } + saved := make([]devSnapshot, len(node.Devices.DeviceLists)) + for i := range node.Devices.DeviceLists { + d := node.Devices.DeviceLists[i].Device + saved[i] = devSnapshot{ + idx: i, + used: d.Used, + usedcores: d.Usedcores, + usedmem: d.Usedmem, + } + } + + // Global rollback function restores the node to its original state. + rollbackAll := func() { + for i := range saved { + dev := node.Devices.DeviceLists[i].Device + dev.Used = saved[i].used + dev.Usedcores = saved[i].usedcores + dev.Usedmem = saved[i].usedmem + } + } + for index := range node.Devices.DeviceLists { node.Devices.DeviceLists[index].ComputeScore(requests) } - // This loop is for requests for different devices + + // Process each device type in the request. for _, k := range requests { - sums += int(k.Nums) - if int(k.Nums) > len(node.Devices.DeviceLists) { - klog.V(5).InfoS(common.NodeInsufficientDevice, "pod", klog.KObj(pod), "request devices nums", k.Nums, "node device nums", len(node.Devices.DeviceLists)) - return false, common.NodeInsufficientDevice - } + // Sort devices by score (best fit first). sort.Sort(node.Devices) - _, ok := device.GetDevices()[k.Type] + + devPlugin, ok := device.GetDevices()[k.Type] if !ok { - return false, "Device type not found" + rollbackAll() + errMsg := "Device type not found" + klog.ErrorS(nil, errMsg, "pod", klog.KObj(pod), "type", k.Type, "node", node.Node.Name) + return false, errMsg } - fit, tmpDevs, reason := device.GetDevices()[k.Type].Fit(getNodeResources(*node, k.Type), k, pod, nodeInfo, devinput) - if fit { - for idx, val := range tmpDevs[k.Type] { - for nidx, v := range node.Devices.DeviceLists { - // bc node.Devices has been sorted, so we should find out the correct device - if v.Device.ID != val.UUID { - continue - } - total += v.Device.Count - totalCore += v.Device.Totalcore - totalMem += v.Device.Totalmem - free += v.Device.Count - v.Device.Used - freeCore += v.Device.Totalcore - v.Device.Usedcores - freeMem += v.Device.Totalmem - v.Device.Usedmem - err := device.GetDevices()[k.Type].AddResourceUsage(pod, node.Devices.DeviceLists[nidx].Device, &tmpDevs[k.Type][idx]) - if err != nil { - klog.Errorf("AddResourceUsage failed:%s", err.Error()) - return false, "AddResourceUsage failed" - } - klog.V(5).Infoln("After AddResourceUsage:", node.Devices.DeviceLists[nidx].Device) + + typeDevices := getNodeResources(*node, k.Type) + if int(k.Nums) > len(typeDevices) { + klog.V(5).InfoS(common.NodeInsufficientDevice, "pod", klog.KObj(pod), + "request devices nums", k.Nums, "node device nums (type)", len(typeDevices), "type", k.Type) + rollbackAll() + return false, common.NodeInsufficientDevice + } + + fit, tmpDevs, reason := devPlugin.Fit(typeDevices, k, pod, nodeInfo, devinput) + if !fit { + rollbackAll() + return false, reason + } + + type usageSnapshot struct { + idx int + used int32 + usedcores int32 + usedmem int32 + } + modified := make([]usageSnapshot, 0, len(tmpDevs[k.Type])) + + for idx, val := range tmpDevs[k.Type] { + targetIdx := -1 + for nidx, v := range node.Devices.DeviceLists { + if v.Device.ID == val.UUID { + targetIdx = nidx + break } } - devs = append(devs, tmpDevs[k.Type]...) - } else { - return false, reason + if targetIdx == -1 { + rollbackAll() + errMsg := fmt.Sprintf("Device with UUID %q not found on node %s after Fit", val.UUID, node.Node.Name) + klog.ErrorS(nil, errMsg, "pod", klog.KObj(pod)) + return false, errMsg + } + d := node.Devices.DeviceLists[targetIdx].Device + + snap := usageSnapshot{ + idx: targetIdx, + used: d.Used, + usedcores: d.Usedcores, + usedmem: d.Usedmem, + } + + err := devPlugin.AddResourceUsage(pod, d, &tmpDevs[k.Type][idx]) + if err != nil { + klog.Errorf("AddResourceUsage failed for device %s: %v, rolling back all changes", d.ID, err) + for _, m := range modified { + dev := node.Devices.DeviceLists[m.idx].Device + dev.Used = m.used + dev.Usedcores = m.usedcores + dev.Usedmem = m.usedmem + } + rollbackAll() + errMsg := fmt.Sprintf("AddResourceUsage failed for device %s: %v", d.ID, err) + return false, errMsg + } + modified = append(modified, snap) + klog.V(5).Infof("Allocated device %s: used=%d, cores=%d, mem=%d", + d.ID, d.Used, d.Usedcores, d.Usedmem) } - (*devinput)[k.Type] = append((*devinput)[k.Type], devs) + + (*devinput)[k.Type] = append((*devinput)[k.Type], tmpDevs[k.Type]) } + return true, "" } -// podInitContainerMaxRequest returns the maximum single-container device -// request across all init containers, considering both GPU count AND memory. -// Per Kubernetes semantics, init containers run sequentially, so the node -// only needs to reserve capacity for the largest one at any moment. -func podInitContainerMaxRequest(resourceReqs device.PodDeviceRequests, numInitContainers int) containerResourceSummary { - maxReq := containerResourceSummary{} +// podInitContainerMaxRequest returns a map of max requirements per device type. +func podInitContainerMaxRequest(resourceReqs device.PodDeviceRequests, numInitContainers int) map[string]containerResourceSummary { + maxReqs := make(map[string]containerResourceSummary) for i := range numInitContainers { - var nums int - var mem int32 - var cores int32 - for _, k := range resourceReqs[i] { - nums += int(k.Nums) - mem += k.Memreq - cores += k.Coresreq + if i >= len(resourceReqs) { + break } - if nums > maxReq.nums || (nums == maxReq.nums && mem > maxReq.memreq) { - maxReq.nums = nums - maxReq.memreq = mem - maxReq.coresreq = cores + for devType, req := range resourceReqs[i] { + existing := maxReqs[devType] + existing.nums = max(existing.nums, int(req.Nums)) + existing.memreq = max(existing.memreq, req.Memreq) + existing.coresreq = max(existing.coresreq, req.Coresreq) + maxReqs[devType] = existing } } - return maxReq + return maxReqs } -// podAppContainerTotalRequest returns the sum of device requests across all -// regular (non-init) containers, considering both GPU count AND memory. -func podAppContainerTotalRequest(resourceReqs device.PodDeviceRequests, numInitContainers int) containerResourceSummary { - total := containerResourceSummary{} +// podAppContainerTotalRequest returns the sum of device requests per device type. +func podAppContainerTotalRequest(resourceReqs device.PodDeviceRequests, numInitContainers int) map[string]containerResourceSummary { + totals := make(map[string]containerResourceSummary) for i := numInitContainers; i < len(resourceReqs); i++ { - for _, k := range resourceReqs[i] { - total.nums += int(k.Nums) - total.memreq += k.Memreq - total.coresreq += k.Coresreq + for devType, req := range resourceReqs[i] { + current := totals[devType] + current.nums += int(req.Nums) + current.memreq += req.Memreq + current.coresreq += req.Coresreq + totals[devType] = current } } - return total + return totals } -// canFitInitContainer checks if the node has at least one device that can -// individually satisfy the init container's resource requirements. -// This prevents allocation failures when total capacity appears sufficient -// but no single device has enough memory/cores. -func (node *NodeUsage) canFitInitContainer(maxInitReq containerResourceSummary) bool { - if node == nil { - return false // nil node cannot satisfy any request +// stripInitContainerAliasSlots removes the device allocations of init containers from the +// PodDevices structure, keeping only regular (non-init) container allocations. +func stripInitContainerAliasSlots(pod *corev1.Pod, resourceReqs device.PodDeviceRequests, devices device.PodDevices) device.PodDevices { + numInitContainers := len(pod.Spec.InitContainers) + if numInitContainers == 0 || len(devices) == 0 { + return devices + } + + expectedTotal := -1 + if resourceReqs != nil { + expectedTotal = len(resourceReqs) } - for _, deviceList := range node.Devices.DeviceLists { - device := deviceList.Device - if device == nil { + result := make(device.PodDevices, len(devices)) + for devType, containerList := range devices { + if expectedTotal >= 0 && len(containerList) != expectedTotal { + klog.ErrorS(nil, "device slot count does not match pod container count, skipping alias-slot strip for this device type to avoid misaligned data", + "pod", klog.KObj(pod), "deviceType", devType, "slots", len(containerList), "expectedContainers", expectedTotal) + result[devType] = containerList continue } - - availableMem := device.Totalmem - device.Usedmem - availableCores := device.Totalcore - device.Usedcores - - // Check if this device individually satisfies init requirements - if availableMem >= maxInitReq.memreq && availableCores >= maxInitReq.coresreq { - return true + if len(containerList) <= numInitContainers { + result[devType] = device.PodSingleDevice{} + } else { + result[devType] = containerList[numInitContainers:] } } - return false + return result } func (s *Scheduler) calcScore(nodes *map[string]*NodeUsage, resourceReqs device.PodDeviceRequests, task *corev1.Pod, failedNodes map[string]string) (*policy.NodeScoreList, error) { @@ -192,219 +291,187 @@ func (s *Scheduler) calcScore(nodes *map[string]*NodeUsage, resourceReqs device. errCh := make(chan error, len(*nodes)) numInitContainers := len(task.Spec.InitContainers) - maxInitReq := podInitContainerMaxRequest(resourceReqs, numInitContainers) - appReqTotal := podAppContainerTotalRequest(resourceReqs, numInitContainers) - - needsInitClone := numInitContainers > 0 && - (maxInitReq.nums > appReqTotal.nums || - maxInitReq.memreq > appReqTotal.memreq || - maxInitReq.coresreq > appReqTotal.coresreq) for nodeID, node := range *nodes { - if numInitContainers > 0 && !node.canFitInitContainer(maxInitReq) { - failedNodesMutex.Lock() - - var failureMsg string - var failureType string - hasDeviceWithEnoughMem := false - - for _, deviceList := range node.Devices.DeviceLists { - device := deviceList.Device - if device == nil { - continue - } - availableMem := device.Totalmem - device.Usedmem - if availableMem >= maxInitReq.memreq { - hasDeviceWithEnoughMem = true - break - } - } - - if hasDeviceWithEnoughMem { - // Failure is due to cores, not memory - failureMsg = fmt.Sprintf( - "no single device has sufficient cores (%d) for init container", - maxInitReq.coresreq) - failureType = "InsufficientInitDeviceCores" - } else { - // Failure is due to memory - failureMsg = fmt.Sprintf( - "no single device has sufficient memory (%dMiB) for init container", - maxInitReq.memreq) - failureType = "InsufficientInitDeviceMemory" - } - - failedNodes[nodeID] = failureMsg - failureReason[failureType] = append(failureReason[failureType], nodeID) - failedNodesMutex.Unlock() - klog.V(4).InfoS("Node filtered: insufficient per-device capacity for init container", - "pod", klog.KObj(task), "node", nodeID, "required_mem", maxInitReq.memreq, "required_cores", maxInitReq.coresreq) - continue - } - wg.Add(1) go func(nodeID string, node *NodeUsage) { defer wg.Done() viewStatus(*node) - score := policy.NodeScore{NodeID: nodeID, Node: node.Node, Devices: make(device.PodDevices), Score: 0} - score.ComputeDefaultScore(node.Devices) - snapshot := score.SnapshotDevice(node.Devices) - nodeInfo, err := s.GetNode(nodeID) if err != nil { klog.ErrorS(err, "Failed to get node", "nodeID", nodeID) + failedNodesMutex.Lock() + failedNodes[nodeID] = fmt.Sprintf("failed to fetch node info: %v", err) + failedNodesMutex.Unlock() errCh <- err return } - var appContainersNode *NodeUsage - if needsInitClone { - appContainersNode = node.DeepCopy() - } else { - appContainersNode = node + baseTypes := nodeDeviceBaseTypes(node.Devices) + + peakUsage := make([]struct { + used int32 + usedcores int32 + usedmem int32 + }, len(node.Devices.DeviceLists)) + for i, dl := range node.Devices.DeviceLists { + peakUsage[i].used = dl.Device.Used + peakUsage[i].usedcores = dl.Device.Usedcores + peakUsage[i].usedmem = dl.Device.Usedmem } - ctrfit := true - for ctrid, n := range resourceReqs { - sums := 0 - for _, k := range n { - sums += int(k.Nums) - } + //Check init containers (they run sequentially, each on a fresh copy) + var initAllocs device.PodDevices + if numInitContainers > 0 { + initAllocs = make(device.PodDevices) - existingBefore := make(map[string]bool) - for typ := range score.Devices { - existingBefore[typ] = true - } - - if ctrid < numInitContainers && !needsInitClone { - } else if sums > 0 { - klog.V(5).InfoS("fitInDevices", "pod", klog.KObj(task), "node", nodeID) - - var workingNode *NodeUsage - if needsInitClone && ctrid < numInitContainers { - workingNode = node.DeepCopy() - } else { - workingNode = appContainersNode + initFit := true + for i, req := range resourceReqs { + if i >= numInitContainers { + break } - - fit, reason := fitInDevices(workingNode, n, task, nodeInfo, &score.Devices) - ctrfit = fit + // Pad previous slots for all types to maintain index alignment + for typ := range baseTypes { + for len(initAllocs[typ]) < i { + initAllocs[typ] = append(initAllocs[typ], device.ContainerDevices{}) + } + } + if len(req) == 0 { + for typ := range baseTypes { + initAllocs[typ] = append(initAllocs[typ], device.ContainerDevices{}) + } + continue + } + nodeCopy := node.DeepCopy() + fit, reason := fitInDevices(nodeCopy, req, task, nodeInfo, &initAllocs) if !fit { - klog.V(4).InfoS(common.NodeUnfitPod, "pod", klog.KObj(task), "node", nodeID, "reason", reason) + klog.V(4).InfoS("Init container does not fit", + "pod", klog.KObj(task), "node", nodeID, "containerIndex", i, "reason", reason) failedNodesMutex.Lock() - failedNodes[nodeID] = common.NodeUnfitPod + failedNodes[nodeID] = reason for reasonType := range common.ParseReason(reason) { failureReason[reasonType] = append(failureReason[reasonType], nodeID) } failedNodesMutex.Unlock() + initFit = false break } - } - - for typ, devSlice := range score.Devices { - if len(devSlice) < ctrid+1 { - if !existingBefore[typ] { - // Brand new type for this container: allocation is at devSlice[0], - // shift it to index ctrid by prepending non‑nil empty slices. - pad := make(device.PodSingleDevice, 0, ctrid) - for range ctrid { - pad = append(pad, device.ContainerDevices{}) - } - score.Devices[typ] = append(pad, devSlice...) - } else { - // Type existed before: just append empty (non‑nil) slices to reach the needed length. - for len(score.Devices[typ]) < ctrid+1 { - score.Devices[typ] = append(score.Devices[typ], device.ContainerDevices{}) - } + // Record how much this init container used, at its peak, per device. + for pi, dl := range nodeCopy.Devices.DeviceLists { + if dl.Device.Used > peakUsage[pi].used { + peakUsage[pi].used = dl.Device.Used + } + if dl.Device.Usedcores > peakUsage[pi].usedcores { + peakUsage[pi].usedcores = dl.Device.Usedcores + } + if dl.Device.Usedmem > peakUsage[pi].usedmem { + peakUsage[pi].usedmem = dl.Device.Usedmem + } + } + // Ensure every type has an entry for this container (even if empty) + for typ := range baseTypes { + if len(initAllocs[typ]) == i { + initAllocs[typ] = append(initAllocs[typ], device.ContainerDevices{}) } } } + if !initFit { + return + } } - if numInitContainers > 0 && !needsInitClone { - for deviceType := range score.Devices { - containerList := score.Devices[deviceType] - - if len(containerList) <= numInitContainers { - continue - } - - for initIdx := range numInitContainers { - if initIdx >= len(resourceReqs) { - // Shouldn't happen, but safety check - continue - } - - // ContainerDeviceRequests is map[string]ContainerDeviceRequest keyed - // by device type, so this is a direct lookup, not a search. - initDeviceReq, found := resourceReqs[initIdx][deviceType] - if !found { - containerList[initIdx] = device.ContainerDevices{} - continue - } - - selectedAllocation := device.ContainerDevices{} - devicesNeeded := int(initDeviceReq.Nums) - memNeeded := initDeviceReq.Memreq - coresNeeded := initDeviceReq.Coresreq - - outerLoop: - for appIdx := numInitContainers; appIdx < len(containerList); appIdx++ { - appAllocation := containerList[appIdx] + // Allocate app containers (they run concurrently, cumulative) + appNodeCopy := node.DeepCopy() + score := policy.NodeScore{ + NodeID: nodeID, + Node: node.Node, + Devices: make(device.PodDevices), + Score: 0, + } + score.ComputeDefaultScore(appNodeCopy.Devices) + snapshot := score.SnapshotDevice(appNodeCopy.Devices) - for deviceIdx := range appAllocation { - if devicesNeeded == 0 { - break outerLoop - } + appIndex := 0 + ctrfit := true - dev := appAllocation[deviceIdx] + for ctrid, n := range resourceReqs { + if ctrid < numInitContainers { + continue + } + for typ := range baseTypes { + for len(score.Devices[typ]) < appIndex { + score.Devices[typ] = append(score.Devices[typ], device.ContainerDevices{}) + } + } + sums := 0 + for _, k := range n { + sums += int(k.Nums) + } + if sums == 0 { + for typ := range baseTypes { + score.Devices[typ] = append(score.Devices[typ], device.ContainerDevices{}) + } + appIndex++ + continue + } + fit, reason := fitInDevices(appNodeCopy, n, task, nodeInfo, &score.Devices) + ctrfit = fit + if !fit { + klog.V(4).InfoS(common.NodeUnfitPod, "pod", klog.KObj(task), "node", nodeID, "reason", reason) + failedNodesMutex.Lock() + failedNodes[nodeID] = reason + for reasonType := range common.ParseReason(reason) { + failureReason[reasonType] = append(failureReason[reasonType], nodeID) + } + failedNodesMutex.Unlock() + break + } + // Ensure every type has an entry for this container + for typ := range baseTypes { + if len(score.Devices[typ]) == appIndex { + score.Devices[typ] = append(score.Devices[typ], device.ContainerDevices{}) + } + } + appIndex++ + } - if dev.Usedmem >= memNeeded && dev.Usedcores >= coresNeeded { - selectedAllocation = append(selectedAllocation, dev) - devicesNeeded-- - } - } - } + if !ctrfit { + return + } - // Verify we found enough valid devices before assigning - if len(selectedAllocation) == int(initDeviceReq.Nums) { - containerList[initIdx] = selectedAllocation - klog.V(4).InfoS( - "Assigned validated app devices to init container", - "pod", klog.KObj(task), - "init_index", initIdx, - "device_type", deviceType, - "devices_assigned", len(selectedAllocation), - "required_mem", memNeeded, - "required_cores", coresNeeded, - ) - } else { - klog.V(3).InfoS( - "ERROR: Insufficient validated devices for init container after canFitInitContainer passed", - "pod", klog.KObj(task), - "init_index", initIdx, - "device_type", deviceType, - "devices_needed", int(initDeviceReq.Nums), - "devices_found", len(selectedAllocation), - "required_mem", memNeeded, - "required_cores", coresNeeded, - ) - // Still assign what we found (partial allocation) - // The device plugin will handle the error downstream - containerList[initIdx] = selectedAllocation - } - } + // Prepend init container allocations + if numInitContainers > 0 && initAllocs != nil { + for devType, initConList := range initAllocs { + score.Devices[devType] = append(initConList, score.Devices[devType]...) } } - if ctrfit { - fitNodesMutex.Lock() - res.NodeList = append(res.NodeList, &score) - fitNodesMutex.Unlock() - score.OverrideScore(snapshot, userNodePolicy) - klog.V(4).InfoS(common.NodeFitPod, "pod", klog.KObj(task), "node", nodeID, "score", score.Score) + // Commit the successful allocations to the original node. + for i := range node.Devices.DeviceLists { + finalUsed := appNodeCopy.Devices.DeviceLists[i].Device.Used + finalCores := appNodeCopy.Devices.DeviceLists[i].Device.Usedcores + finalMem := appNodeCopy.Devices.DeviceLists[i].Device.Usedmem + if peakUsage[i].used > finalUsed { + finalUsed = peakUsage[i].used + } + if peakUsage[i].usedcores > finalCores { + finalCores = peakUsage[i].usedcores + } + if peakUsage[i].usedmem > finalMem { + finalMem = peakUsage[i].usedmem + } + node.Devices.DeviceLists[i].Device.Used = finalUsed + node.Devices.DeviceLists[i].Device.Usedcores = finalCores + node.Devices.DeviceLists[i].Device.Usedmem = finalMem } + + score.OverrideScore(snapshot, userNodePolicy) + fitNodesMutex.Lock() + res.NodeList = append(res.NodeList, &score) + fitNodesMutex.Unlock() + + klog.V(4).InfoS(common.NodeFitPod, "pod", klog.KObj(task), "node", nodeID, "score", score.Score) }(nodeID, node) } wg.Wait() diff --git a/pkg/scheduler/score_test.go b/pkg/scheduler/score_test.go index 81686c5e0..ae403a367 100644 --- a/pkg/scheduler/score_test.go +++ b/pkg/scheduler/score_test.go @@ -515,7 +515,7 @@ func Test_calcScore(t *testing.T) { NodeList: []*policy.NodeScore{}, // Node should be filtered out }, failedNodes: map[string]string{ - "node1": "no single device has sufficient memory (4000MiB) for init container", + "node1": "2/2 CardInsufficientMemory", }, err: nil, }, @@ -1692,7 +1692,7 @@ func Test_calcScore(t *testing.T) { NodeList: []*policy.NodeScore{}, }, failedNodes: map[string]string{ - "node1": common.NodeUnfitPod, + "node1": "1/1 CardInsufficientMemory", }, err: nil, }, @@ -1798,8 +1798,146 @@ func Test_calcScore(t *testing.T) { NodeList: []*policy.NodeScore{}, }, failedNodes: map[string]string{ - "node1": common.NodeUnfitPod, - "node2": common.NodeUnfitPod, + "node1": "1/1 CardInsufficientMemory", + "node2": "1/1 CardInsufficientMemory", + }, + err: nil, + }, + }, + { + name: "two init containers, first has no device request, second does - tests slot alignment (issue #1667)", + args: struct { + nodes *map[string]*NodeUsage + nums device.PodDeviceRequests + annos map[string]string + task *corev1.Pod + }{ + nodes: &map[string]*NodeUsage{ + "node1": { + Node: &corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: "node1"}}, + Devices: policy.DeviceUsageList{ + Policy: util.NodeSchedulerPolicyBinpack.String(), + DeviceLists: []*policy.DeviceListsScore{ + { + Device: &device.DeviceUsage{ + ID: "uuid1", + Index: 0, + Used: 0, + Count: 10, + Usedmem: 0, + Totalmem: 8000, + Totalcore: 100, + Usedcores: 0, + Numa: 0, + Type: nvidia.NvidiaGPUDevice, + Health: true, + }, + }, + }, + }, + }, + }, + nums: device.PodDeviceRequests{ + // Index 0: InitContainer 0 - no device request (e.g. a setup/download step) + {}, + // Index 1: InitContainer 1 - requests a device + { + nvidia.NvidiaGPUDevice: device.ContainerDeviceRequest{ + Nums: 1, + Type: nvidia.NvidiaGPUDevice, + Memreq: 1000, + Coresreq: 30, + }, + }, + // Index 2: App container - requests a device + { + nvidia.NvidiaGPUDevice: device.ContainerDeviceRequest{ + Nums: 1, + Type: nvidia.NvidiaGPUDevice, + Memreq: 1000, + Coresreq: 30, + }, + }, + }, + annos: make(map[string]string), + task: &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-init-heterogeneous", + }, + Spec: corev1.PodSpec{ + InitContainers: []corev1.Container{ + { + Name: "init-no-device", + Image: "busybox", + Resources: corev1.ResourceRequirements{}, + }, + { + Name: "init-gpu-check", + Image: "chrstnhntschl/gpu_burn", + Resources: corev1.ResourceRequirements{ + Limits: corev1.ResourceList{ + "hami.io/gpu": *resource.NewQuantity(1, resource.BinarySI), + "hami.io/gpucores": *resource.NewQuantity(30, resource.BinarySI), + "hami.io/gpumem": *resource.NewQuantity(1000, resource.BinarySI), + }, + }, + }, + }, + Containers: []corev1.Container{ + { + Name: "app-gpu-burn", + Image: "chrstnhntschl/gpu_burn", + Resources: corev1.ResourceRequirements{ + Limits: corev1.ResourceList{ + "hami.io/gpu": *resource.NewQuantity(1, resource.BinarySI), + "hami.io/gpucores": *resource.NewQuantity(30, resource.BinarySI), + "hami.io/gpumem": *resource.NewQuantity(1000, resource.BinarySI), + }, + }, + }, + }, + }, + }, + }, + wants: struct { + want *policy.NodeScoreList + failedNodes map[string]string + err error + }{ + want: &policy.NodeScoreList{ + Policy: util.NodeSchedulerPolicyBinpack.String(), + NodeList: []*policy.NodeScore{ + { + NodeID: "node1", + Devices: device.PodDevices{ + "NVIDIA": device.PodSingleDevice{ + // Index 0: init-no-device -> untouched placeholder slot + {}, + // Index 1: init-gpu-check -> real allocation + { + { + Idx: 0, + UUID: "uuid1", + Type: nvidia.NvidiaGPUDevice, + Usedcores: 30, + Usedmem: 1000, + }, + }, + // Index 2: app-gpu-burn -> real allocation + { + { + Idx: 0, + UUID: "uuid1", + Type: nvidia.NvidiaGPUDevice, + Usedcores: 30, + Usedmem: 1000, + }, + }, + }, + }, + Score: 0, + }, + }, }, err: nil, }, @@ -2931,7 +3069,7 @@ func Test_calcScore(t *testing.T) { NodeList: []*policy.NodeScore{}, }, failedNodes: map[string]string{ - "node1": "no single device has sufficient cores (30) for init container", + "node1": "2/2 CardInsufficientCore", }, err: nil, }, From 6be1ae25329db51bc617431283f9b1093baa7166 Mon Sep 17 00:00:00 2001 From: maishivamhoo123 Date: Mon, 6 Jul 2026 08:38:16 +0000 Subject: [PATCH 12/14] Fixed the linter issue Signed-off-by: maishivamhoo123 --- pkg/scheduler/webhook.go | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/pkg/scheduler/webhook.go b/pkg/scheduler/webhook.go index 538bba464..35a639584 100644 --- a/pkg/scheduler/webhook.go +++ b/pkg/scheduler/webhook.go @@ -158,10 +158,9 @@ func fitResourceQuota(pod *corev1.Pod) bool { // 1. Calculate the MAX request among InitContainers for _, ctr := range pod.Spec.InitContainers { - req, ok := getRequest(&ctr, resourceName) - if ok && req == 1 { + _, ok := getRequest(&ctr, resourceName) + if ok { if memReq, ok := getRequest(&ctr, memResourceName); ok { - // --- FIX 2: Use modern max() --- initMemoryReq = max(initMemoryReq, memReq) } if coreReq, ok := getRequest(&ctr, coreResourceName); ok { @@ -172,7 +171,7 @@ func fitResourceQuota(pod *corev1.Pod) bool { // 2. Calculate the SUM of requests among Regular Containers for _, ctr := range pod.Spec.Containers { - req, ok := getRequest(&ctr, resourceName) + _, ok := getRequest(&ctr, resourceName) if ok { if memReq, ok := getRequest(&ctr, memResourceName); ok { appMemoryReq += memReq From 25d36021ce66321ee8029121b855b3cdec4ebb4f Mon Sep 17 00:00:00 2001 From: maishivamhoo123 Date: Mon, 6 Jul 2026 08:57:00 +0000 Subject: [PATCH 13/14] fixed e2e test Signed-off-by: maishivamhoo123 --- pkg/scheduler/score.go | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/pkg/scheduler/score.go b/pkg/scheduler/score.go index fdc987aeb..fb8951ec1 100644 --- a/pkg/scheduler/score.go +++ b/pkg/scheduler/score.go @@ -46,15 +46,13 @@ func viewStatus(usage NodeUsage) { } } -// getNodeResources returns all devices of the exact given type from the node. func getNodeResources(list NodeUsage, t string) []*device.DeviceUsage { l := []*device.DeviceUsage{} for _, val := range list.Devices.DeviceLists { if val.Device == nil { continue } - // Exact match or prefix match (e.g., "NVIDIA" matches "NVIDIA A100-SXM4-40GB") - if val.Device.Type == t || strings.HasPrefix(val.Device.Type, t+" ") { + if getDeviceBaseType(val.Device.Type) == t { l = append(l, val.Device) } } From db04eaa06ba6c97b97479994a12b698e8f1c42e1 Mon Sep 17 00:00:00 2001 From: maishivamhoo123 Date: Mon, 6 Jul 2026 09:47:06 +0000 Subject: [PATCH 14/14] Added the suggestion Signed-off-by: maishivamhoo123 --- pkg/scheduler/scheduler.go | 4 +- pkg/scheduler/score.go | 110 +++++++++++++++++++++------------- pkg/scheduler/webhook.go | 14 ++--- pkg/scheduler/webhook_test.go | 7 ++- 4 files changed, 81 insertions(+), 54 deletions(-) diff --git a/pkg/scheduler/scheduler.go b/pkg/scheduler/scheduler.go index a8975b62b..df71fb8ad 100644 --- a/pkg/scheduler/scheduler.go +++ b/pkg/scheduler/scheduler.go @@ -170,8 +170,8 @@ func (s *Scheduler) onAddPod(obj any) { } } if allInitDone { - - podDev = stripInitContainerAliasSlots(pod, nil, podDev) + resourceReqs := device.Resourcereqs(pod) + podDev = stripInitContainerAliasSlots(pod, resourceReqs, podDev) } } diff --git a/pkg/scheduler/score.go b/pkg/scheduler/score.go index fb8951ec1..245bf5d43 100644 --- a/pkg/scheduler/score.go +++ b/pkg/scheduler/score.go @@ -95,7 +95,7 @@ func nodeDeviceBaseTypes(list policy.DeviceUsageList) map[string]struct{} { func fitInDevices(node *NodeUsage, requests device.ContainerDeviceRequests, pod *corev1.Pod, nodeInfo *device.NodeInfo, devinput *device.PodDevices) (bool, string) { // Snapshot the entire node state before any modifications. type devSnapshot struct { - idx int + dev *device.DeviceUsage used int32 usedcores int32 usedmem int32 @@ -104,20 +104,17 @@ func fitInDevices(node *NodeUsage, requests device.ContainerDeviceRequests, pod for i := range node.Devices.DeviceLists { d := node.Devices.DeviceLists[i].Device saved[i] = devSnapshot{ - idx: i, + dev: d, used: d.Used, usedcores: d.Usedcores, usedmem: d.Usedmem, } } - - // Global rollback function restores the node to its original state. rollbackAll := func() { - for i := range saved { - dev := node.Devices.DeviceLists[i].Device - dev.Used = saved[i].used - dev.Usedcores = saved[i].usedcores - dev.Usedmem = saved[i].usedmem + for _, s := range saved { + s.dev.Used = s.used + s.dev.Usedcores = s.usedcores + s.dev.Usedmem = s.usedmem } } @@ -308,18 +305,22 @@ func (s *Scheduler) calcScore(nodes *map[string]*NodeUsage, resourceReqs device. baseTypes := nodeDeviceBaseTypes(node.Devices) - peakUsage := make([]struct { + type peakUsageSnapshot struct { used int32 usedcores int32 usedmem int32 - }, len(node.Devices.DeviceLists)) - for i, dl := range node.Devices.DeviceLists { - peakUsage[i].used = dl.Device.Used - peakUsage[i].usedcores = dl.Device.Usedcores - peakUsage[i].usedmem = dl.Device.Usedmem + } + peakUsage := make(map[string]peakUsageSnapshot) + for _, dl := range node.Devices.DeviceLists { + id := dl.Device.ID + peakUsage[id] = peakUsageSnapshot{ + used: dl.Device.Used, + usedcores: dl.Device.Usedcores, + usedmem: dl.Device.Usedmem, + } } - //Check init containers (they run sequentially, each on a fresh copy) + // 1) Check init containers (they run sequentially, each on a fresh copy) var initAllocs device.PodDevices if numInitContainers > 0 { initAllocs = make(device.PodDevices) @@ -355,16 +356,26 @@ func (s *Scheduler) calcScore(nodes *map[string]*NodeUsage, resourceReqs device. initFit = false break } - // Record how much this init container used, at its peak, per device. - for pi, dl := range nodeCopy.Devices.DeviceLists { - if dl.Device.Used > peakUsage[pi].used { - peakUsage[pi].used = dl.Device.Used - } - if dl.Device.Usedcores > peakUsage[pi].usedcores { - peakUsage[pi].usedcores = dl.Device.Usedcores - } - if dl.Device.Usedmem > peakUsage[pi].usedmem { - peakUsage[pi].usedmem = dl.Device.Usedmem + // Record peak usage for this init container per device (by device ID). + for _, dl := range nodeCopy.Devices.DeviceLists { + id := dl.Device.ID + if p, ok := peakUsage[id]; ok { + if dl.Device.Used > p.used { + p.used = dl.Device.Used + } + if dl.Device.Usedcores > p.usedcores { + p.usedcores = dl.Device.Usedcores + } + if dl.Device.Usedmem > p.usedmem { + p.usedmem = dl.Device.Usedmem + } + peakUsage[id] = p + } else { + peakUsage[id] = peakUsageSnapshot{ + used: dl.Device.Used, + usedcores: dl.Device.Usedcores, + usedmem: dl.Device.Usedmem, + } } } // Ensure every type has an entry for this container (even if empty) @@ -379,7 +390,7 @@ func (s *Scheduler) calcScore(nodes *map[string]*NodeUsage, resourceReqs device. } } - // Allocate app containers (they run concurrently, cumulative) + // 2) Allocate app containers (they run concurrently, cumulative) appNodeCopy := node.DeepCopy() score := policy.NodeScore{ NodeID: nodeID, @@ -438,30 +449,45 @@ func (s *Scheduler) calcScore(nodes *map[string]*NodeUsage, resourceReqs device. return } - // Prepend init container allocations + // 3) Prepend init container allocations if numInitContainers > 0 && initAllocs != nil { for devType, initConList := range initAllocs { score.Devices[devType] = append(initConList, score.Devices[devType]...) } } - // Commit the successful allocations to the original node. - for i := range node.Devices.DeviceLists { - finalUsed := appNodeCopy.Devices.DeviceLists[i].Device.Used - finalCores := appNodeCopy.Devices.DeviceLists[i].Device.Usedcores - finalMem := appNodeCopy.Devices.DeviceLists[i].Device.Usedmem - if peakUsage[i].used > finalUsed { - finalUsed = peakUsage[i].used + for _, dl := range node.Devices.DeviceLists { + id := dl.Device.ID + // Find the corresponding device in the app copy + var appDev *device.DeviceUsage + for _, adl := range appNodeCopy.Devices.DeviceLists { + if adl.Device.ID == id { + appDev = adl.Device + break + } } - if peakUsage[i].usedcores > finalCores { - finalCores = peakUsage[i].usedcores + finalUsed := int32(0) + finalCores := int32(0) + finalMem := int32(0) + if appDev != nil { + finalUsed = appDev.Used + finalCores = appDev.Usedcores + finalMem = appDev.Usedmem } - if peakUsage[i].usedmem > finalMem { - finalMem = peakUsage[i].usedmem + if p, ok := peakUsage[id]; ok { + if p.used > finalUsed { + finalUsed = p.used + } + if p.usedcores > finalCores { + finalCores = p.usedcores + } + if p.usedmem > finalMem { + finalMem = p.usedmem + } } - node.Devices.DeviceLists[i].Device.Used = finalUsed - node.Devices.DeviceLists[i].Device.Usedcores = finalCores - node.Devices.DeviceLists[i].Device.Usedmem = finalMem + dl.Device.Used = finalUsed + dl.Device.Usedcores = finalCores + dl.Device.Usedmem = finalMem } score.OverrideScore(snapshot, userNodePolicy) diff --git a/pkg/scheduler/webhook.go b/pkg/scheduler/webhook.go index 35a639584..9439ab875 100644 --- a/pkg/scheduler/webhook.go +++ b/pkg/scheduler/webhook.go @@ -156,28 +156,26 @@ func fitResourceQuota(pod *corev1.Pod) bool { var initMemoryReq, initCoresReq int64 var appMemoryReq, appCoresReq int64 - // 1. Calculate the MAX request among InitContainers for _, ctr := range pod.Spec.InitContainers { - _, ok := getRequest(&ctr, resourceName) + req, ok := getRequest(&ctr, resourceName) if ok { if memReq, ok := getRequest(&ctr, memResourceName); ok { - initMemoryReq = max(initMemoryReq, memReq) + initMemoryReq = max(initMemoryReq, memReq*req) } if coreReq, ok := getRequest(&ctr, coreResourceName); ok { - initCoresReq = max(initCoresReq, coreReq) + initCoresReq = max(initCoresReq, coreReq*req) } } } - // 2. Calculate the SUM of requests among Regular Containers for _, ctr := range pod.Spec.Containers { - _, ok := getRequest(&ctr, resourceName) + req, ok := getRequest(&ctr, resourceName) if ok { if memReq, ok := getRequest(&ctr, memResourceName); ok { - appMemoryReq += memReq + appMemoryReq += memReq * req } if coreReq, ok := getRequest(&ctr, coreResourceName); ok { - appCoresReq += coreReq + appCoresReq += coreReq * req } } } diff --git a/pkg/scheduler/webhook_test.go b/pkg/scheduler/webhook_test.go index 3d168d35f..462be4221 100644 --- a/pkg/scheduler/webhook_test.go +++ b/pkg/scheduler/webhook_test.go @@ -262,12 +262,15 @@ func TestFitResourceQuota(t *testing.T) { klog.Fatalf("Failed to initialize devices with config: %v", err) } - qm := device.NewQuotaManager() ns := "default" memName := "nvidia.com/gpumem" coreName := "nvidia.com/gpucores" - qm.Quotas[ns] = &device.DeviceQuota{ + cache := device.GetLocalCache() + if cache.Quotas == nil { + cache.Quotas = make(map[string]*device.DeviceQuota) + } + cache.Quotas[ns] = &device.DeviceQuota{ memName: &device.Quota{Used: 1000, Limit: 2000}, coreName: &device.Quota{Used: 200, Limit: 400}, }