From 75e70ca98906acc8db523fbd0dc1b206956bb93f Mon Sep 17 00:00:00 2001 From: yxxhero Date: Sun, 24 May 2026 17:31:47 +0800 Subject: [PATCH 1/2] fix: replace klog.Fatalf with error returns to prevent process crash klog.Fatalf calls os.Exit on transient API errors (informer list failures, label patch failures, config loading), which kills the scheduler process instead of allowing retry. Replace with: - updateSchedulerLabel(): klog.ErrorS + return instead of klog.Fatalf - InitDevices(): return error instead of klog.Fatalf - InitDefaultDevices(): return error instead of klog.Fatalf Update all callers to handle the returned errors. Signed-off-by: yxxhero --- cmd/scheduler/main.go | 4 +++- pkg/scheduler/config/config.go | 18 +++++++++--------- pkg/scheduler/config/config_test.go | 3 ++- pkg/scheduler/nodes_test.go | 2 +- 4 files changed, 15 insertions(+), 12 deletions(-) diff --git a/cmd/scheduler/main.go b/cmd/scheduler/main.go index f7c68d1585..d61dd42855 100644 --- a/cmd/scheduler/main.go +++ b/cmd/scheduler/main.go @@ -119,7 +119,9 @@ func start() error { client.WithTimeout(config.Timeout), ) - config.InitDevices() + if err := config.InitDevices(); err != nil { + return fmt.Errorf("failed to initialize devices: %w", err) + } var err error config.HostName, err = os.Hostname() diff --git a/pkg/scheduler/config/config.go b/pkg/scheduler/config/config.go index 701a0c1bba..dbac14cf52 100644 --- a/pkg/scheduler/config/config.go +++ b/pkg/scheduler/config/config.go @@ -278,24 +278,25 @@ func validateConfig(config *Config) error { return fmt.Errorf("all configurations are empty") } -func InitDevices() { +func InitDevices() error { if len(device.DevicesMap) > 0 { klog.Info("Devices are already initialized, skipping initialization") - return + return nil } klog.Infof("Loading device configuration from file: %s", configFile) config, err := LoadConfig(configFile) if err != nil { - klog.Fatalf("Failed to load device config file %s: %v", configFile, err) + return fmt.Errorf("failed to load device config file %s: %w", configFile, err) } klog.Infof("Loaded config: %v", config) err = InitDevicesWithConfig(config) if err != nil { - klog.Fatalf("Failed to initialize devices: %v", err) + return fmt.Errorf("failed to initialize devices: %w", err) } + return nil } -func InitDefaultDevices() { +func InitDefaultDevices() error { configMapdata := ` nvidia: resourceCountName: "nvidia.com/gpu" @@ -454,14 +455,13 @@ vnpus: var yamlData Config err := yaml.Unmarshal([]byte(configMapdata), &yamlData) if err != nil { - klog.Fatalf("Failed to unmarshal default config: %v", err) - return + return fmt.Errorf("failed to unmarshal default config: %w", err) } - // Initialize devices with configuration if err := InitDevicesWithConfig(&yamlData); err != nil { - klog.Fatalf("Failed to initialize devices with default config: %v", err) + return fmt.Errorf("failed to initialize devices with default config: %w", err) } + return nil } func GlobalFlagSet() *flag.FlagSet { diff --git a/pkg/scheduler/config/config_test.go b/pkg/scheduler/config/config_test.go index ad1b35ac90..b561c3f4e8 100644 --- a/pkg/scheduler/config/config_test.go +++ b/pkg/scheduler/config/config_test.go @@ -479,7 +479,8 @@ func Test_GetDevices(t *testing.T) { } func Test_InitDefaultDevices(t *testing.T) { - InitDefaultDevices() + err := InitDefaultDevices() + assert.NilError(t, err, "Expected InitDefaultDevices to succeed") assert.Assert(t, len(device.DevicesMap) > 0, "Expected devicesMap to be populated") assert.Assert(t, len(device.DevicesToHandle) > 0, "Expected DevicesToHandle to be populated") } diff --git a/pkg/scheduler/nodes_test.go b/pkg/scheduler/nodes_test.go index 4118957222..3e5178f6b3 100644 --- a/pkg/scheduler/nodes_test.go +++ b/pkg/scheduler/nodes_test.go @@ -128,7 +128,7 @@ func Test_addNode_ListNodes(t *testing.T) { err: nil, }, } - config.InitDefaultDevices() + assert.NilError(t, config.InitDefaultDevices(), "Expected InitDefaultDevices to succeed") for _, test := range tests { t.Run(test.name, func(t *testing.T) { m := nodeManager{ From 4ac00e105d7427fcb1719b468aa2c7b97ffcd8e3 Mon Sep 17 00:00:00 2001 From: yxxhero Date: Sat, 8 Aug 2026 17:39:35 +0800 Subject: [PATCH 2/2] fix(scheduler): serialize Filter device selection to prevent double GPU allocation Under concurrent pod creation (e.g. a multi-replica Deployment), two Filter extender calls can run in separate goroutines and both observe the same free GPU before either commits its reservation to the pod cache. Both pods then get the same card id annotated, so the card id in the pod annotation no longer matches the card the pod actually uses: the shared card gets over-allocated (causing OOM) while other idle cards cannot be handed out. This is issue #2232. The scheduler extender is served over HTTP with one goroutine per request and provides no cross-request serialization guarantee, so the read-pick-commit sequence in Filter (read node usage -> pick devices -> commit to cache) must be made atomic. Add a filterLock and move the in-memory critical section into a selectAndCommitDevice helper that owns the lock through a single deferred Unlock (so a future early return can never leak it). The lock covers only device selection and cache commit; the PatchPodAnnotations API call and the on-failure rollback run outside it, so concurrent Filters are not blocked on network I/O. The rollback is safe because the pod/quota managers have their own locks. The simulation path does not reserve devices and is unchanged. In the normal kube-scheduler flow scheduling cycles are already serialized, so this lock is uncontended there; it only closes the window for concurrent extender requests. Tests: - Test_Filter_ConcurrentNoDoubleAllocation fires many pods at Filter simultaneously and asserts each is annotated with a distinct device id. It reliably reproduces the double allocation without the lock and passes with it. - Test_Filter_RollbackOnPatchFailure verifies the cache reservation is rolled back when the annotation patch fails. Signed-off-by: yxxhero --- pkg/scheduler/scheduler.go | 100 +++++++++++++------- pkg/scheduler/scheduler_test.go | 158 ++++++++++++++++++++++++++++++++ 2 files changed, 226 insertions(+), 32 deletions(-) diff --git a/pkg/scheduler/scheduler.go b/pkg/scheduler/scheduler.go index 2b65789459..0b632dbade 100644 --- a/pkg/scheduler/scheduler.go +++ b/pkg/scheduler/scheduler.go @@ -78,6 +78,10 @@ type Scheduler struct { lock sync.RWMutex synced bool + + // filterLock serializes device selection in Filter so concurrent requests + // cannot reserve the same device (issue #2232). + filterLock sync.Mutex } func NewScheduler() *Scheduler { @@ -1030,13 +1034,62 @@ func (s *Scheduler) Filter(args extenderv1.ExtenderArgs) (*extenderv1.ExtenderFi if args.Nodes != nil { return s.filterSimulation(args, resourceReqs) } + klog.V(2).InfoS("Choosing live filter path", + "pod", klog.KObj(args.Pod), + "reason", "request does not contain full nodes", + "nodeNamesLen", nodeNamesLen(args.NodeNames)) + // selectAndCommitDevice holds filterLock during selection; the annotation + // patch below runs outside it (issue #2232). + selection, err := s.selectAndCommitDevice(args, resourceReqs) + if err != nil { + s.recordScheduleFilterResultEvent(args.Pod, EventReasonFilteringFailed, "", err) + return nil, err + } + if selection.chosen == nil { + s.recordScheduleFilterResultEvent(args.Pod, EventReasonFilteringFailed, "", fmt.Errorf("no available node, %d nodes do not meet", len(*args.NodeNames))) + return &extenderv1.ExtenderFilterResult{ + FailedNodes: selection.failedNodes, + }, nil + } + m := selection.chosen + // Patch the annotation outside the lock; roll back the cache on failure. + if err = util.PatchPodAnnotations(args.Pod, selection.annotations); err != nil { + s.recordScheduleFilterResultEvent(args.Pod, EventReasonFilteringFailed, "", err) + if selection.added { + s.quotaManager.RmUsage(args.Pod, selection.effectiveDevices) + } + s.podManager.DelPod(args.Pod) + return nil, err + } + successMsg := genSuccessMsg(len(*args.NodeNames), m.NodeID, selection.nodeList) + s.recordScheduleFilterResultEvent(args.Pod, EventReasonFilteringSucceed, successMsg, nil) + res := extenderv1.ExtenderFilterResult{NodeNames: &[]string{m.NodeID}} + return &res, nil +} + +// filterSelection holds a Filter reservation; chosen is nil when no node fits +// (failedNodes then carries the per-node reasons). +type filterSelection struct { + chosen *policy.NodeScore + annotations map[string]string + added bool + nodeList []*policy.NodeScore + failedNodes map[string]string + effectiveDevices device.PodDevices +} +// selectAndCommitDevice selects a device and commits it to the cache under +// filterLock. It does not publish the pod annotation; the caller patches it. +func (s *Scheduler) selectAndCommitDevice(args extenderv1.ExtenderArgs, resourceReqs device.PodDeviceRequests) (*filterSelection, error) { + s.filterLock.Lock() + defer s.filterLock.Unlock() + + selection := &filterSelection{} if pi, ok := s.podManager.TakeAndDeletePod(args.Pod); ok { s.quotaManager.RmUsage(args.Pod, pi.Devices) } nodeUsage, _, failedNodes, err := s.getNodesUsage(args.NodeNames, args.Pod) if err != nil { - s.recordScheduleFilterResultEvent(args.Pod, EventReasonFilteringFailed, "", err) return nil, err } if len(failedNodes) != 0 { @@ -1044,16 +1097,12 @@ func (s *Scheduler) Filter(args extenderv1.ExtenderArgs) (*extenderv1.ExtenderFi } nodeScores, err := s.calcScore(nodeUsage, resourceReqs, args.Pod, failedNodes) if err != nil { - err := fmt.Errorf("calcScore failed %v for pod %v", err, args.Pod.Name) - s.recordScheduleFilterResultEvent(args.Pod, EventReasonFilteringFailed, "", err) - return nil, err + return nil, fmt.Errorf("calcScore failed %v for pod %v", err, args.Pod.Name) } if len((*nodeScores).NodeList) == 0 { 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, - }, nil + selection.failedNodes = failedNodes + return selection, nil } klog.V(4).Infoln("nodeScores_len=", len((*nodeScores).NodeList)) sort.Sort(nodeScores) @@ -1066,33 +1115,20 @@ func (s *Scheduler) Filter(args extenderv1.ExtenderArgs) (*extenderv1.ExtenderFi annotations := make(map[string]string) annotations[util.AssignedNodeAnnotations] = m.NodeID annotations[util.AssignedTimeAnnotations] = strconv.FormatInt(time.Now().Unix(), 10) - for _, val := range device.GetDevices() { val.PatchAnnotations(args.Pod, &annotations, m.Devices) } - - rawDevices := m.Devices - effectiveDevices := device.CollapseInitContainerUsage(args.Pod, rawDevices) - if args.Nodes == nil { - added := s.podManager.AddPod(args.Pod, m.NodeID, effectiveDevices) - if added { - s.quotaManager.AddUsage(args.Pod, effectiveDevices) // use collapsed - } - err = util.PatchPodAnnotations(args.Pod, annotations) - if err != nil { - s.recordScheduleFilterResultEvent(args.Pod, EventReasonFilteringFailed, "", err) - if added { - s.quotaManager.RmUsage(args.Pod, effectiveDevices) - } - s.podManager.DelPod(args.Pod) - return nil, err - } - } - - successMsg := genSuccessMsg(len(*args.NodeNames), m.NodeID, nodeScores.NodeList) - s.recordScheduleFilterResultEvent(args.Pod, EventReasonFilteringSucceed, successMsg, nil) - res := extenderv1.ExtenderFilterResult{NodeNames: &[]string{m.NodeID}} - return &res, nil + // Collapse init-container usage so the cache reflects the effective footprint. + effectiveDevices := device.CollapseInitContainerUsage(args.Pod, m.Devices) + selection.chosen = m + selection.annotations = annotations + selection.nodeList = nodeScores.NodeList + selection.effectiveDevices = effectiveDevices + selection.added = s.podManager.AddPod(args.Pod, m.NodeID, effectiveDevices) + if selection.added { + s.quotaManager.AddUsage(args.Pod, effectiveDevices) + } + return selection, nil } func (s *Scheduler) filterSimulation(args extenderv1.ExtenderArgs, resourceReqs device.PodDeviceRequests) (*extenderv1.ExtenderFilterResult, error) { diff --git a/pkg/scheduler/scheduler_test.go b/pkg/scheduler/scheduler_test.go index 23383a9a08..a9d742eabb 100644 --- a/pkg/scheduler/scheduler_test.go +++ b/pkg/scheduler/scheduler_test.go @@ -22,6 +22,7 @@ import ( "maps" "slices" "strings" + "sync" "sync/atomic" "testing" "time" @@ -1584,6 +1585,163 @@ func Test_Filter_EvictsStaleEntry(t *testing.T) { } } +// Test_Filter_ConcurrentNoDoubleAllocation asserts that concurrent Filter +// calls do not reserve the same device (issue #2232). Each device holds one +// pod, so the 8 pods must each get a distinct device id. +func Test_Filter_ConcurrentNoDoubleAllocation(t *testing.T) { + const numPods = 8 + + require.NoError(t, config.InitDevicesWithConfig(&config.Config{ + NvidiaConfig: nvidia.NvidiaConfig{ + ResourceCountName: "hami.io/gpu", + ResourceMemoryName: "hami.io/gpumem", + ResourceMemoryPercentageName: "hami.io/gpumem-percentage", + ResourceCoreName: "hami.io/gpucores", + DefaultGPUNum: 1, + }, + })) + + client.KubeClient = fake.NewClientset() + t.Cleanup(func() { client.KubeClient = nil }) + s := NewScheduler() + s.kubeClient = client.KubeClient + informerFactory := informers.NewSharedInformerFactoryWithOptions(client.KubeClient, time.Hour) + s.podLister = informerFactory.Core().V1().Pods().Lister() + s.nodeLister = informerFactory.Core().V1().Nodes().Lister() + informerFactory.Start(s.stopCh) + informerFactory.WaitForCacheSync(s.stopCh) + + // One node with numPods devices; each device fits exactly one pod. + deviceInfos := make([]device.DeviceInfo, 0, numPods) + for i := range numPods { + deviceInfos = append(deviceInfos, device.DeviceInfo{ + ID: fmt.Sprintf("device-%d", i), + Index: uint(i), + Count: 10, + Devmem: 8000, + Devcore: 100, + Mode: "hami", + Type: nvidia.NvidiaGPUDevice, + Health: true, + DeviceVendor: nvidia.NvidiaGPUDevice, + }) + } + s.addNode("node1", &device.NodeInfo{ + ID: "node1", + Node: &corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: "node1"}}, + Devices: map[string][]device.DeviceInfo{ + nvidia.NvidiaGPUDevice: deviceInfos, + }, + }) + + // Create the pods in the fake apiserver so PatchPodAnnotations succeeds. + pods := make([]*corev1.Pod, numPods) + for i := range numPods { + pod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: fmt.Sprintf("pod-%d", i), + UID: types.UID(fmt.Sprintf("uid-%d", i)), + Annotations: map[string]string{ + util.GPUSchedulerPolicyAnnotationKey: util.GPUSchedulerPolicyBinpack.String(), + util.NodeSchedulerPolicyAnnotationKey: util.NodeSchedulerPolicyBinpack.String(), + }, + }, + Spec: corev1.PodSpec{ + Containers: []corev1.Container{{ + Name: "gpu-burn", + Resources: corev1.ResourceRequirements{Limits: corev1.ResourceList{ + "hami.io/gpu": *resource.NewQuantity(1, resource.BinarySI), + "hami.io/gpumem": *resource.NewQuantity(5000, resource.BinarySI), + "hami.io/gpucores": *resource.NewQuantity(10, resource.BinarySI), + }}, + }}, + }, + } + pods[i] = pod + _, err := client.KubeClient.CoreV1().Pods(pod.Namespace).Create(t.Context(), pod, metav1.CreateOptions{}) + require.NoError(t, err) + } + + nodeNames := []string{"node1"} + + // Barrier so all goroutines enter Filter simultaneously. + start := make(chan struct{}) + var wg sync.WaitGroup + for i := range numPods { + wg.Add(1) + go func(pod *corev1.Pod) { + defer wg.Done() + <-start + _, _ = s.Filter(extenderv1.ExtenderArgs{Pod: pod, NodeNames: &nodeNames}) + }(pods[i]) + } + close(start) + wg.Wait() + + // Every pod must have been annotated with a distinct device id. + seen := make(map[string]string, numPods) // deviceID -> podName + for _, pod := range pods { + refreshed, err := client.KubeClient.CoreV1().Pods(pod.Namespace).Get(t.Context(), pod.Name, metav1.GetOptions{}) + require.NoError(t, err) + allocated, err := device.DecodePodDevices(device.InRequestDevices, refreshed.Annotations) + require.NoError(t, err, "decoding devices for pod %s", pod.Name) + single, ok := allocated[nvidia.NvidiaGPUDevice] + require.True(t, ok && len(single) > 0 && len(single[0]) > 0, "pod %s has no allocated device", pod.Name) + deviceID := single[0][0].UUID + require.NotEmpty(t, deviceID, "pod %s has empty device id", pod.Name) + if other, dup := seen[deviceID]; dup { + t.Fatalf("device %s double-allocated to pod %s and pod %s", deviceID, other, pod.Name) + } + seen[deviceID] = pod.Name + } + require.Len(t, seen, numPods, "expected each pod on a distinct device") +} + +// Test_Filter_RollbackOnPatchFailure verifies that when the annotation patch +// fails, Filter rolls back the reservation it just committed to the cache. +func Test_Filter_RollbackOnPatchFailure(t *testing.T) { + require.NoError(t, config.InitDevicesWithConfig(&config.Config{ + NvidiaConfig: nvidia.NvidiaConfig{ + ResourceCountName: "hami.io/gpu", + ResourceMemoryName: "hami.io/gpumem", + ResourceCoreName: "hami.io/gpucores", + DefaultGPUNum: 1, + }, + })) + // Empty clientset: the pod is unknown to the apiserver, so PatchPodAnnotations fails. + client.KubeClient = fake.NewClientset() + t.Cleanup(func() { client.KubeClient = nil }) + + s := NewScheduler() + s.addNode("node1", &device.NodeInfo{ + ID: "node1", + Node: &corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: "node1"}}, + Devices: map[string][]device.DeviceInfo{ + nvidia.NvidiaGPUDevice: {{ + ID: "device1", Index: 0, Count: 10, Devmem: 8000, Devcore: 100, + Mode: "hami", Type: nvidia.NvidiaGPUDevice, Health: true, DeviceVendor: nvidia.NvidiaGPUDevice, + }}, + }, + }) + + pod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: "pod-fail", UID: "uid-fail"}, + Spec: corev1.PodSpec{Containers: []corev1.Container{{ + Name: "c", + Resources: corev1.ResourceRequirements{Limits: corev1.ResourceList{ + "hami.io/gpu": *resource.NewQuantity(1, resource.BinarySI), + "hami.io/gpumem": *resource.NewQuantity(1000, resource.BinarySI), + }}, + }}}, + } + + _, err := s.Filter(extenderv1.ExtenderArgs{Pod: pod, NodeNames: &[]string{"node1"}}) + require.Error(t, err, "expected patch failure to surface as an error") + + _, inCache := s.podManager.GetPod(pod) + require.False(t, inCache, "reservation must be rolled back when the annotation patch fails") +} + func TestFilterUsesTemplateNodesWithoutSideEffects(t *testing.T) { s := NewScheduler() client.KubeClient = fake.NewSimpleClientset()