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 1eb4801d3..df71fb8ad 100644 --- a/pkg/scheduler/scheduler.go +++ b/pkg/scheduler/scheduler.go @@ -92,8 +92,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{ @@ -161,6 +159,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 { + resourceReqs := device.Resourcereqs(pod) + podDev = stripInitContainerAliasSlots(pod, resourceReqs, podDev) + } + } + if s.podManager.AddPod(pod, nodeID, podDev) { s.quotaManager.AddUsage(pod, podDev) } @@ -400,7 +414,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) @@ -741,22 +762,25 @@ 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) + + hasHAMiResource := false + + for _, reqMap := range resourceReqs { + if len(reqMap) > 0 { + hasHAMiResource = true + break } } - 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) } @@ -766,8 +790,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 { @@ -776,8 +799,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, @@ -799,16 +821,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 953f13a72..0f99ec969 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" @@ -459,11 +460,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", @@ -500,7 +501,56 @@ 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)", + 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"}, + }, + wantPodAnnotationDeviceIDs: []string{"device3"}, }, { name: "node use binpack gpu use spread policy", @@ -537,7 +587,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", @@ -574,7 +624,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", @@ -611,7 +661,7 @@ func Test_Filter(t *testing.T) { want: &extenderv1.ExtenderFilterResult{ NodeNames: &[]string{"node1"}, }, - wantPodAnnotationDeviceID: "device2", + wantPodAnnotationDeviceIDs: []string{"device2"}, }, } @@ -624,7 +674,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) + } }) } } @@ -1232,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", }, }, }, @@ -1320,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 236f3fedb..245bf5d43 100644 --- a/pkg/scheduler/score.go +++ b/pkg/scheduler/score.go @@ -32,6 +32,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 + coresreq int32 +} + func viewStatus(usage NodeUsage) { klog.V(5).Info("devices status") for _, val := range usage.Devices.DeviceLists { @@ -42,66 +49,224 @@ func viewStatus(usage NodeUsage) { 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 + } + if getDeviceBaseType(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) { - //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 + // Snapshot the entire node state before any modifications. + type devSnapshot struct { + dev *device.DeviceUsage + 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{ + dev: d, + used: d.Used, + usedcores: d.Usedcores, + usedmem: d.Usedmem, + } + } + rollbackAll := func() { + for _, s := range saved { + s.dev.Used = s.used + s.dev.Usedcores = s.usedcores + s.dev.Usedmem = s.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" - } - 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) + rollbackAll() + errMsg := "Device type not found" + klog.ErrorS(nil, errMsg, "pod", klog.KObj(pod), "type", k.Type, "node", node.Node.Name) + return false, errMsg + } + + 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 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 { + if i >= len(resourceReqs) { + break + } + 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 maxReqs +} + +// 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 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 totals +} + +// 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) + } + + 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 + } + if len(containerList) <= numInitContainers { + result[devType] = device.PodSingleDevice{} + } else { + result[devType] = containerList[numInitContainers:] + } + } + return result +} + 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 { @@ -119,77 +284,223 @@ func (s *Scheduler) calcScore(nodes *map[string]*NodeUsage, resourceReqs device. failedNodesMutex := sync.Mutex{} failureReason := make(map[string][]string) errCh := make(chan error, len(*nodes)) + + numInitContainers := len(task.Spec.InitContainers) + for nodeID, node := range *nodes { 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 } - // Assume the node is a fit by default. This handles pods with no device - // requests, which should be schedulable on any node. + baseTypes := nodeDeviceBaseTypes(node.Devices) + + type peakUsageSnapshot struct { + used int32 + usedcores int32 + usedmem int32 + } + 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, + } + } + + // 1) Check init containers (they run sequentially, each on a fresh copy) + var initAllocs device.PodDevices + if numInitContainers > 0 { + initAllocs = make(device.PodDevices) + + initFit := true + for i, req := range resourceReqs { + if i >= numInitContainers { + break + } + // 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("Init container does not fit", + "pod", klog.KObj(task), "node", nodeID, "containerIndex", i, "reason", reason) + failedNodesMutex.Lock() + failedNodes[nodeID] = reason + for reasonType := range common.ParseReason(reason) { + failureReason[reasonType] = append(failureReason[reasonType], nodeID) + } + failedNodesMutex.Unlock() + initFit = false + break + } + // 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) + for typ := range baseTypes { + if len(initAllocs[typ]) == i { + initAllocs[typ] = append(initAllocs[typ], device.ContainerDevices{}) + } + } + } + if !initFit { + return + } + } + + // 2) 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) + + appIndex := 0 ctrfit := true - deviceType := "" - //This loop is for different container request + 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) } - - // container need no device and we have got certain deviceType - 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) - fit, reason := fitInDevices(node, n, task, nodeInfo, &score.Devices) - // found certain deviceType, fill missing empty allocation for containers before this - 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]...) + 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] = common.NodeUnfitPod + 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 !ctrfit { + return + } + + // 3) 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) + 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 + } + } + finalUsed := int32(0) + finalCores := int32(0) + finalMem := int32(0) + if appDev != nil { + finalUsed = appDev.Used + finalCores = appDev.Usedcores + finalMem = appDev.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 + } + } + dl.Device.Used = finalUsed + dl.Device.Usedcores = finalCores + dl.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() close(errCh) - // only pod scheduler failure will record failure event 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 fa670ef7a..ae403a367 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": "2/2 CardInsufficientMemory", + }, + err: nil, + }, + }, { name: "one node two device one pod one container use one device,but having use 50%", args: struct { @@ -1574,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, }, @@ -1680,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, }, @@ -2444,6 +2700,380 @@ 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{ + { + { + Idx: 0, + UUID: "uuid2", + Type: nvidia.NvidiaGPUDevice, + Usedcores: 30, + Usedmem: 1000, + }, + }, + // Index 1: Regular Container allocated normally. + { + { + Idx: 0, + UUID: "uuid2", + Type: nvidia.NvidiaGPUDevice, + Usedcores: 30, + Usedmem: 1000, + }, + }, + }, + }, + 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}, + }, + }, + }, + Score: 0, + }, + }, + }, + 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": "2/2 CardInsufficientCore", + }, + 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 00b4578e7..9439ab875 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,45 @@ func fitResourceQuota(pod *corev1.Pod) bool { } return 0, false } + + var initMemoryReq, initCoresReq int64 + var appMemoryReq, appCoresReq int64 + + for _, ctr := range pod.Spec.InitContainers { + req, ok := getRequest(&ctr, resourceName) + if ok { + if memReq, ok := getRequest(&ctr, memResourceName); ok { + initMemoryReq = max(initMemoryReq, memReq*req) + } + if coreReq, ok := getRequest(&ctr, coreResourceName); ok { + initCoresReq = max(initCoresReq, coreReq*req) + } + } + } + for _, ctr := range pod.Spec.Containers { req, ok := getRequest(&ctr, resourceName) if ok { if memReq, ok := getRequest(&ctr, memResourceName); ok { - memoryReq += memReq * req + appMemoryReq += memReq * req } if coreReq, ok := getRequest(&ctr, coreResourceName); ok { - coresReq += coreReq * req + appCoresReq += coreReq * req } } } + + 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 bb5e9f94d..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}, } @@ -385,6 +388,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{