From dc47d8840d349455add5bc8d1413a0d2fac28051 Mon Sep 17 00:00:00 2001 From: Samarth Verma Date: Sun, 12 Apr 2026 11:14:51 -0400 Subject: [PATCH 1/3] scheduler: fix inFlightPods leak when pod is recreated during scheduling failure handleSchedulingFailure can refresh podInfo from the informer before AddUnschedulableIfNotPresent. A delete and recreate with the same name may change the Pod UID while inFlightPods still tracks the UID from Pop, so Done and queueing-hint lookups must use that in-flight UID. Add an explicit in-flight UID parameter, thread it through queueing-hint lookups, cover the same-name recreation case with a regression test, and check the returned error in updated test call sites. --- pkg/scheduler/backend/queue/active_queue.go | 17 ++- .../backend/queue/scheduling_queue.go | 25 +++- .../backend/queue/scheduling_queue_test.go | 121 +++++++++++------- pkg/scheduler/eventhandlers_test.go | 2 +- pkg/scheduler/schedule_one.go | 6 +- pkg/scheduler/schedule_one_podgroup_test.go | 2 +- .../scheduler/queueing/queue_test.go | 7 +- 7 files changed, 120 insertions(+), 60 deletions(-) diff --git a/pkg/scheduler/backend/queue/active_queue.go b/pkg/scheduler/backend/queue/active_queue.go index 060ee8c3640..5d1fdb78bc2 100644 --- a/pkg/scheduler/backend/queue/active_queue.go +++ b/pkg/scheduler/backend/queue/active_queue.go @@ -46,7 +46,9 @@ type activeQueuer interface { movePodToInFlight(pInfo *framework.QueuedPodInfo) error listInFlightEvents() []interface{} listInFlightPods() []*v1.Pod - clusterEventsForPod(logger klog.Logger, pInfo *framework.QueuedPodInfo) ([]*clusterEvent, error) + // clusterEventsForPod collects cluster events recorded while the pod registered under inFlightListUID + // was in flight. If inFlightListUID is empty, pInfo.Pod.UID is used. + clusterEventsForPod(logger klog.Logger, pInfo *framework.QueuedPodInfo, inFlightListUID types.UID) ([]*clusterEvent, error) addEventsIfPodInFlight(oldPod, newPod *v1.Pod, events []fwk.ClusterEvent) bool addEventIfAnyInFlight(oldObj, newObj interface{}, event fwk.ClusterEvent) bool @@ -396,16 +398,21 @@ func (aq *activeQueue) listInFlightPods() []*v1.Pod { return pods } -// clusterEventsForPod gets all cluster events that have happened during pod for pInfo is being scheduled. -func (aq *activeQueue) clusterEventsForPod(logger klog.Logger, pInfo *framework.QueuedPodInfo) ([]*clusterEvent, error) { +// clusterEventsForPod gets all cluster events that have happened while the scheduling attempt +// for the pod keyed by inFlightListUID (in inFlightPods) was in flight. +func (aq *activeQueue) clusterEventsForPod(logger klog.Logger, pInfo *framework.QueuedPodInfo, inFlightListUID types.UID) ([]*clusterEvent, error) { aq.lock.RLock() defer aq.lock.RUnlock() logger.V(5).Info("Checking events for in-flight pod", "pod", klog.KObj(pInfo.Pod), "unschedulablePlugins", pInfo.UnschedulablePlugins, "inFlightEventsSize", aq.inFlightEvents.Len(), "inFlightPodsSize", len(aq.inFlightPods)) + listUID := inFlightListUID + if listUID == "" { + listUID = pInfo.Pod.UID + } // AddUnschedulableIfNotPresent is called with the Pod at the end of scheduling or binding. // So, given pInfo should have been Pop()ed before, - // we can assume pInfo must be recorded in inFlightPods and thus inFlightEvents. - inFlightPod, ok := aq.inFlightPods[pInfo.Pod.UID] + // we can assume listUID must be recorded in inFlightPods and thus inFlightEvents. + inFlightPod, ok := aq.inFlightPods[listUID] if !ok { return nil, fmt.Errorf("in flight Pod isn't found in the scheduling queue. If you see this error log, it's likely a bug in the scheduler") } diff --git a/pkg/scheduler/backend/queue/scheduling_queue.go b/pkg/scheduler/backend/queue/scheduling_queue.go index 81c5488cb44..aacb7394ecc 100644 --- a/pkg/scheduler/backend/queue/scheduling_queue.go +++ b/pkg/scheduler/backend/queue/scheduling_queue.go @@ -108,7 +108,11 @@ type SchedulingQueue interface { // AddUnschedulableIfNotPresent adds an unschedulable pod back to scheduling queue. // The podSchedulingCycle represents the current scheduling cycle number which can be // returned by calling SchedulingCycle(). - AddUnschedulableIfNotPresent(logger klog.Logger, pod *framework.QueuedPodInfo, podSchedulingCycle int64) error + // schedulingCycleInFlightUID is the UID under which the pod is registered in inFlightPods (from Pop). + // Use the empty string to mean pInfo.Pod.UID. When the pod object is refreshed from the API/informer + // after a delete+recreate with the same name, pInfo.Pod.UID may differ; pass the pre-refresh UID so + // Done and queueing-hint logic use the correct in-flight entry (see kubernetes/kubernetes#138316). + AddUnschedulableIfNotPresent(logger klog.Logger, pod *framework.QueuedPodInfo, podSchedulingCycle int64, schedulingCycleInFlightUID types.UID) error // SchedulingCycle returns the current number of scheduling cycle which is // cached by scheduling queue. Normally, incrementing this number whenever // a pod is popped (e.g. called Pop()) is enough. @@ -815,7 +819,8 @@ func (p *PriorityQueue) SchedulingCycle() int64 { // determineSchedulingHintForInFlightPod looks at the unschedulable plugins of the given Pod // and determines the scheduling hint for this Pod while checking the events that happened during in-flight. -func (p *PriorityQueue) determineSchedulingHintForInFlightPod(logger klog.Logger, pInfo *framework.QueuedPodInfo) queueingStrategy { +// inFlightListUID is the key in inFlightPods for this scheduling attempt (empty means pInfo.Pod.UID). +func (p *PriorityQueue) determineSchedulingHintForInFlightPod(logger klog.Logger, pInfo *framework.QueuedPodInfo, inFlightListUID types.UID) queueingStrategy { if len(pInfo.UnschedulablePlugins) == 0 && len(pInfo.PendingPlugins) == 0 { // No failed plugins are associated with this Pod. // Meaning something unusual (a temporal failure on kube-apiserver, etc) happened and this Pod gets moved back to the queue. @@ -829,7 +834,7 @@ func (p *PriorityQueue) determineSchedulingHintForInFlightPod(logger klog.Logger return queueAfterBackoff } - events, err := p.activeQ.clusterEventsForPod(logger, pInfo) + events, err := p.activeQ.clusterEventsForPod(logger, pInfo, inFlightListUID) if err != nil { utilruntime.HandleErrorWithLogger(logger, err, "Error getting cluster events for pod", "pod", klog.KObj(pInfo.Pod)) return queueAfterBackoff @@ -902,12 +907,18 @@ func (p *PriorityQueue) addUnschedulableWithoutQueueingHint(logger klog.Logger, // the queue, unless it is already in the queue. Normally, PriorityQueue puts // unschedulable pods in `unschedulablePods`. But if there has been a recent move // request, then the pod is put in `backoffQ`. -func (p *PriorityQueue) AddUnschedulableIfNotPresent(logger klog.Logger, pInfo *framework.QueuedPodInfo, podSchedulingCycle int64) error { +func (p *PriorityQueue) AddUnschedulableIfNotPresent(logger klog.Logger, pInfo *framework.QueuedPodInfo, podSchedulingCycle int64, schedulingCycleInFlightUID types.UID) error { p.lock.Lock() defer p.lock.Unlock() - // In any case, this Pod will be moved back to the queue and we should call Done. - defer p.Done(pInfo.Pod.UID) + doneUID := pInfo.Pod.UID + if schedulingCycleInFlightUID != "" { + doneUID = schedulingCycleInFlightUID + } + // In any case, this scheduling attempt will be moved back to the queue and we should call Done + // for the UID registered in inFlightPods at Pop time (doneUID), which may differ from pInfo.Pod.UID + // if the pod was refreshed from the informer after a same-name recreate. + defer p.Done(doneUID) pod := pInfo.Pod if p.unschedulablePods.get(pod) != nil { @@ -951,7 +962,7 @@ func (p *PriorityQueue) AddUnschedulableIfNotPresent(logger klog.Logger, pInfo * } // We check whether this Pod may change its scheduling result by any of events that happened during scheduling. - schedulingHint := p.determineSchedulingHintForInFlightPod(logger, pInfo) + schedulingHint := p.determineSchedulingHintForInFlightPod(logger, pInfo, doneUID) // In this case, we try to requeue this Pod to activeQ/backoffQ. queue := p.requeuePodWithQueueingStrategy(logger, pInfo, schedulingHint, framework.ScheduleAttemptFailure) diff --git a/pkg/scheduler/backend/queue/scheduling_queue_test.go b/pkg/scheduler/backend/queue/scheduling_queue_test.go index 13da015ad27..2655c375677 100644 --- a/pkg/scheduler/backend/queue/scheduling_queue_test.go +++ b/pkg/scheduler/backend/queue/scheduling_queue_test.go @@ -617,7 +617,7 @@ func Test_InFlightPods(t *testing.T) { logger, _ := ktesting.NewTestContext(t) // This pod will be requeued to backoffQ immediately because no plugin is registered as unschedulable plugin, // which means the pod encountered an unexpected error (e.g., a network error). - err := q.AddUnschedulableIfNotPresent(logger, poppedPod, q.SchedulingCycle()) + err := q.AddUnschedulableIfNotPresent(logger, poppedPod, q.SchedulingCycle(), "") if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -626,7 +626,7 @@ func Test_InFlightPods(t *testing.T) { logger, _ := ktesting.NewTestContext(t) poppedPod2.UnschedulablePlugins = sets.New("fooPlugin2", "fooPlugin3") poppedPod2.PendingPlugins = sets.New("fooPlugin1") - err := q.AddUnschedulableIfNotPresent(logger, poppedPod2, q.SchedulingCycle()) + err := q.AddUnschedulableIfNotPresent(logger, poppedPod2, q.SchedulingCycle(), "") if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -670,7 +670,7 @@ func Test_InFlightPods(t *testing.T) { // Unschedulable due to PendingPlugins. poppedPod.PendingPlugins = sets.New("fooPlugin1") poppedPod.UnschedulablePlugins = sets.New("fooPlugin2") - if err := q.AddUnschedulableIfNotPresent(logger, poppedPod, q.SchedulingCycle()); err != nil { + if err := q.AddUnschedulableIfNotPresent(logger, poppedPod, q.SchedulingCycle(), ""); err != nil { t.Errorf("Unexpected error from AddUnschedulableIfNotPresent: %v", err) } }}, @@ -689,7 +689,7 @@ func Test_InFlightPods(t *testing.T) { {callback: func(t *testing.T, q *PriorityQueue) { logger, _ := ktesting.NewTestContext(t) // Failed (i.e. no UnschedulablePlugins). Should go to backoff. - if err := q.AddUnschedulableIfNotPresent(logger, poppedPod, q.SchedulingCycle()); err != nil { + if err := q.AddUnschedulableIfNotPresent(logger, poppedPod, q.SchedulingCycle(), ""); err != nil { t.Errorf("Unexpected error from AddUnschedulableIfNotPresent: %v", err) } }}, @@ -822,7 +822,7 @@ func Test_InFlightPods(t *testing.T) { case action.eventHappens != nil: q.MoveAllToActiveOrBackoffQueue(logger, *action.eventHappens, nil, nil, nil) case action.podEnqueued != nil: - err := q.AddUnschedulableIfNotPresent(logger, action.podEnqueued, q.SchedulingCycle()) + err := q.AddUnschedulableIfNotPresent(logger, action.podEnqueued, q.SchedulingCycle(), "") if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -942,7 +942,7 @@ func TestPop(t *testing.T) { poppedPod := popPod(t, logger, q, pod) // We put register the plugin to PendingPlugins so that it's interpreted as queueImmediately and skip backoff. poppedPod.PendingPlugins = sets.New("fooPlugin1") - if err := q.AddUnschedulableIfNotPresent(logger, poppedPod, q.SchedulingCycle()); err != nil { + if err := q.AddUnschedulableIfNotPresent(logger, poppedPod, q.SchedulingCycle(), ""); err != nil { t.Errorf("Unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -976,7 +976,7 @@ func TestPriorityQueue_AddUnschedulableIfNotPresent(t *testing.T) { } q.Add(ctx, highPriNominatedPodInfo.Pod) - err := q.AddUnschedulableIfNotPresent(logger, newQueuedPodInfoForLookup(unschedulablePodInfo.Pod, "plugin"), q.SchedulingCycle()) + err := q.AddUnschedulableIfNotPresent(logger, newQueuedPodInfoForLookup(unschedulablePodInfo.Pod, "plugin"), q.SchedulingCycle(), "") if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -994,6 +994,37 @@ func TestPriorityQueue_AddUnschedulableIfNotPresent(t *testing.T) { } } +// TestPriorityQueue_AddUnschedulableIfNotPresent_PodRecreatedSameName covers the case where the +// informer returns a new Pod UID for the same namespace/name before AddUnschedulableIfNotPresent +// (see kubernetes/kubernetes#138316): Done and queueing hints must use the popped scheduling UID. +func TestPriorityQueue_AddUnschedulableIfNotPresent_PodRecreatedSameName(t *testing.T) { + featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.SchedulerQueueingHints, true) + + logger, ctx := ktesting.NewTestContext(t) + ctx, cancel := context.WithCancel(ctx) + defer cancel() + + oldUID := types.UID("popped-pod-uid") + pod := st.MakePod().Name("foo").Namespace("ns").UID(string(oldUID)).Obj() + q := NewTestQueueWithObjects(ctx, newDefaultQueueSort(), []runtime.Object{pod}) + q.Add(ctx, pod) + popped, err := q.Pop(logger) + if err != nil { + t.Fatalf("Pop: %v", err) + } + if popped.Pod.UID != oldUID { + t.Fatalf("unexpected popped UID: got %v want %v", popped.Pod.UID, oldUID) + } + + recreated := pod.DeepCopy() + recreated.UID = types.UID("apiserver-new-uid-after-recreate") + pInfo := newQueuedPodInfoForLookup(recreated, "plugin") + if err := q.AddUnschedulableIfNotPresent(logger, pInfo, q.SchedulingCycle(), oldUID); err != nil { + t.Fatalf("AddUnschedulableIfNotPresent: %v", err) + } + expectInFlightPods(t, q) +} + // TestPriorityQueue_AddUnschedulableIfNotPresent_Backoff tests the scenarios when // AddUnschedulableIfNotPresent is called asynchronously. // Pods in and before current scheduling cycle will be put back to activeQueue @@ -1043,7 +1074,7 @@ func TestPriorityQueue_AddUnschedulableIfNotPresent_Backoff(t *testing.T) { }, } - err := q.AddUnschedulableIfNotPresent(logger, newQueuedPodInfoForLookup(unschedulablePod, "plugin"), oldCycle) + err := q.AddUnschedulableIfNotPresent(logger, newQueuedPodInfoForLookup(unschedulablePod, "plugin"), oldCycle, "") if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -1490,7 +1521,7 @@ func TestPriorityQueue_UpdateWhenInflight(t *testing.T) { // test-pod got rejected by fakePlugin, // but the update event that it just got may change this scheduling result, // and hence we should put this pod to activeQ/backoffQ. - err := q.AddUnschedulableIfNotPresent(logger, newQueuedPodInfoForLookup(updatedPod, "fakePlugin"), q.SchedulingCycle()) + err := q.AddUnschedulableIfNotPresent(logger, newQueuedPodInfoForLookup(updatedPod, "fakePlugin"), q.SchedulingCycle(), "") if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -2031,7 +2062,7 @@ func BenchmarkMoveAllToActiveOrBackoffQueue(b *testing.B) { // Random case. podInfo = q.newQueuedPodInfo(ctx, p, plugins[j%len(plugins)]) } - err := q.AddUnschedulableIfNotPresent(logger, podInfo, q.SchedulingCycle()) + err := q.AddUnschedulableIfNotPresent(logger, podInfo, q.SchedulingCycle(), "") if err != nil { b.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -2143,7 +2174,7 @@ func TestPriorityQueue_MoveAllToActiveOrBackoffQueueWithQueueingHint(t *testing. t.Errorf("Unexpected pod after Pop (-want, +got):\n%s", diff) } // add to unsched pod pool - err := q.AddUnschedulableIfNotPresent(logger, test.podInfo, q.SchedulingCycle()) + err := q.AddUnschedulableIfNotPresent(logger, test.podInfo, q.SchedulingCycle(), "") if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -2195,11 +2226,11 @@ func TestPriorityQueue_MoveAllToActiveOrBackoffQueue(t *testing.T) { t.Errorf("Unexpected pod after Pop (-want, +got):\n%s", diff) } expectInFlightPods(t, q, unschedulablePodInfo.Pod.UID, highPriorityPodInfo.Pod.UID) - err := q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, unschedulablePodInfo.Pod, "fooPlugin"), q.SchedulingCycle()) + err := q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, unschedulablePodInfo.Pod, "fooPlugin"), q.SchedulingCycle(), "") if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } - err = q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, highPriorityPodInfo.Pod, "fooPlugin"), q.SchedulingCycle()) + err = q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, highPriorityPodInfo.Pod, "fooPlugin"), q.SchedulingCycle(), "") if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -2216,7 +2247,7 @@ func TestPriorityQueue_MoveAllToActiveOrBackoffQueue(t *testing.T) { // This Pod will go to backoffQ because no failure plugin is associated with it. hpp1PodInfo := q.newQueuedPodInfo(ctx, hpp1) hpp1PodInfo.UnschedulableCount++ - err = q.AddUnschedulableIfNotPresent(logger, hpp1PodInfo, q.SchedulingCycle()) + err = q.AddUnschedulableIfNotPresent(logger, hpp1PodInfo, q.SchedulingCycle(), "") if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -2231,7 +2262,7 @@ func TestPriorityQueue_MoveAllToActiveOrBackoffQueue(t *testing.T) { } expectInFlightPods(t, q, hpp2.UID) // This Pod will go to the unschedulable Pod pool. - err = q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, hpp2, "barPlugin"), q.SchedulingCycle()) + err = q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, hpp2, "barPlugin"), q.SchedulingCycle(), "") if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -2285,17 +2316,17 @@ func TestPriorityQueue_MoveAllToActiveOrBackoffQueue(t *testing.T) { highPriorityQueuedPodInfo := q.newQueuedPodInfo(ctx, highPriorityPodInfo.Pod, "fooPlugin") hpp1QueuedPodInfo := q.newQueuedPodInfo(ctx, hpp1) expectInFlightPods(t, q, medPriorityPodInfo.Pod.UID, unschedulablePodInfo.Pod.UID, highPriorityPodInfo.Pod.UID, hpp1.UID) - err = q.AddUnschedulableIfNotPresent(logger, unschedulableQueuedPodInfo, q.SchedulingCycle()) + err = q.AddUnschedulableIfNotPresent(logger, unschedulableQueuedPodInfo, q.SchedulingCycle(), "") if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } expectInFlightPods(t, q, medPriorityPodInfo.Pod.UID, highPriorityPodInfo.Pod.UID, hpp1.UID) - err = q.AddUnschedulableIfNotPresent(logger, highPriorityQueuedPodInfo, q.SchedulingCycle()) + err = q.AddUnschedulableIfNotPresent(logger, highPriorityQueuedPodInfo, q.SchedulingCycle(), "") if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } expectInFlightPods(t, q, medPriorityPodInfo.Pod.UID, hpp1.UID) - err = q.AddUnschedulableIfNotPresent(logger, hpp1QueuedPodInfo, q.SchedulingCycle()) + err = q.AddUnschedulableIfNotPresent(logger, hpp1QueuedPodInfo, q.SchedulingCycle(), "") if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -2352,11 +2383,11 @@ func TestPriorityQueue_MoveAllToActiveOrBackoffQueueWithoutQueueingHint(t *testi // To simulate the pod is failed in scheduling in the real world, Pop() the pod from activeQ before AddUnschedulableIfNotPresent()s below. q.Add(ctx, medPriorityPodInfo.Pod) - err := q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, unschedulablePodInfo.Pod, "fooPlugin"), q.SchedulingCycle()) + err := q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, unschedulablePodInfo.Pod, "fooPlugin"), q.SchedulingCycle(), "") if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } - err = q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, highPriorityPodInfo.Pod, "fooPlugin"), q.SchedulingCycle()) + err = q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, highPriorityPodInfo.Pod, "fooPlugin"), q.SchedulingCycle(), "") if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -2365,14 +2396,14 @@ func TestPriorityQueue_MoveAllToActiveOrBackoffQueueWithoutQueueingHint(t *testi // This Pod will go to backoffQ because no failure plugin is associated with it. hpp1PodInfo := q.newQueuedPodInfo(ctx, hpp1) hpp1PodInfo.UnschedulableCount++ - err = q.AddUnschedulableIfNotPresent(logger, hpp1PodInfo, q.SchedulingCycle()) + err = q.AddUnschedulableIfNotPresent(logger, hpp1PodInfo, q.SchedulingCycle(), "") if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } // Construct another Pod, and associate its scheduler failure to plugin "barPlugin". hpp2 := clonePod(highPriorityPodInfo.Pod, "hpp2") // This Pod will go to the unschedulable Pod pool. - err = q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, hpp2, "barPlugin"), q.SchedulingCycle()) + err = q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, hpp2, "barPlugin"), q.SchedulingCycle(), "") if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -2401,15 +2432,15 @@ func TestPriorityQueue_MoveAllToActiveOrBackoffQueueWithoutQueueingHint(t *testi unschedulableQueuedPodInfo := q.newQueuedPodInfo(ctx, unschedulablePodInfo.Pod, "fooPlugin") highPriorityQueuedPodInfo := q.newQueuedPodInfo(ctx, highPriorityPodInfo.Pod, "fooPlugin") hpp1QueuedPodInfo := q.newQueuedPodInfo(ctx, hpp1) - err = q.AddUnschedulableIfNotPresent(logger, unschedulableQueuedPodInfo, q.SchedulingCycle()) + err = q.AddUnschedulableIfNotPresent(logger, unschedulableQueuedPodInfo, q.SchedulingCycle(), "") if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } - err = q.AddUnschedulableIfNotPresent(logger, highPriorityQueuedPodInfo, q.SchedulingCycle()) + err = q.AddUnschedulableIfNotPresent(logger, highPriorityQueuedPodInfo, q.SchedulingCycle(), "") if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } - err = q.AddUnschedulableIfNotPresent(logger, hpp1QueuedPodInfo, q.SchedulingCycle()) + err = q.AddUnschedulableIfNotPresent(logger, hpp1QueuedPodInfo, q.SchedulingCycle(), "") if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -2556,7 +2587,7 @@ func TestPriorityQueue_AssignedPodAdded_(t *testing.T) { t.Errorf("Unexpected pod after Pop (-want, +got):\n%s", diff) } - err := q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, tt.unschedPod, tt.unschedPlugin), q.SchedulingCycle()) + err := q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, tt.unschedPod, tt.unschedPlugin), q.SchedulingCycle(), "") if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -2672,7 +2703,7 @@ func TestPriorityQueue_AssignedPodUpdated(t *testing.T) { t.Errorf("Unexpected pod after Pop (-want, +got):\n%s", diff) } - err := q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, tt.unschedPod, tt.unschedPlugin), q.SchedulingCycle()) + err := q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, tt.unschedPod, tt.unschedPlugin), q.SchedulingCycle(), "") if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -2797,11 +2828,11 @@ func TestPriorityQueue_PendingPods(t *testing.T) { t.Errorf("Unexpected pod after Pop (-want, +got):\n%s", diff) } q.Add(ctx, medPriorityPodInfo.Pod) - err := q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, unschedulablePodInfo.Pod, "plugin"), q.SchedulingCycle()) + err := q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, unschedulablePodInfo.Pod, "plugin"), q.SchedulingCycle(), "") if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } - err = q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, highPriorityPodInfo.Pod, "plugin"), q.SchedulingCycle()) + err = q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, highPriorityPodInfo.Pod, "plugin"), q.SchedulingCycle(), "") if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -2987,7 +3018,7 @@ func TestRecentlyTriedPodsGoBack(t *testing.T) { }) p1.UnschedulablePlugins = sets.New("plugin") // Put in the unschedulable queue. - err = q.AddUnschedulableIfNotPresent(logger, p1, q.SchedulingCycle()) + err = q.AddUnschedulableIfNotPresent(logger, p1, q.SchedulingCycle(), "") if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -3039,7 +3070,7 @@ func TestPodFailedSchedulingMultipleTimesDoesNotBlockNewerPod(t *testing.T) { t.Errorf("Unexpected pod after Pop (-want, +got):\n%s", diff) } // Put in the unschedulable queue - err := q.AddUnschedulableIfNotPresent(logger, newQueuedPodInfoForLookup(unschedulablePod, "plugin"), q.SchedulingCycle()) + err := q.AddUnschedulableIfNotPresent(logger, newQueuedPodInfoForLookup(unschedulablePod, "plugin"), q.SchedulingCycle(), "") if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -3071,7 +3102,7 @@ func TestPodFailedSchedulingMultipleTimesDoesNotBlockNewerPod(t *testing.T) { }) // And then, put unschedulable pod to the unschedulable queue - err = q.AddUnschedulableIfNotPresent(logger, newQueuedPodInfoForLookup(unschedulablePod, "plugin"), q.SchedulingCycle()) + err = q.AddUnschedulableIfNotPresent(logger, newQueuedPodInfoForLookup(unschedulablePod, "plugin"), q.SchedulingCycle(), "") if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -3118,7 +3149,7 @@ func TestHighPriorityBackoff(t *testing.T) { Message: "fake scheduling failure", }) // Put in the unschedulable queue. - err = q.AddUnschedulableIfNotPresent(logger, newQueuedPodInfoForLookup(p.Pod, "fooPlugin"), q.SchedulingCycle()) + err = q.AddUnschedulableIfNotPresent(logger, newQueuedPodInfoForLookup(p.Pod, "fooPlugin"), q.SchedulingCycle(), "") if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -3181,11 +3212,11 @@ func TestHighPriorityFlushUnschedulablePodsLeftover(t *testing.T) { } else if diff := cmp.Diff(midPod, p.Pod); diff != "" { t.Errorf("Unexpected pod after Pop (-want, +got):\n%s", diff) } - err := q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, highPod, "fakePlugin"), q.SchedulingCycle()) + err := q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, highPod, "fakePlugin"), q.SchedulingCycle(), "") if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } - err = q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, midPod, "fakePlugin"), q.SchedulingCycle()) + err = q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, midPod, "fakePlugin"), q.SchedulingCycle(), "") if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -3229,7 +3260,7 @@ func TestFlushUnschedulablePodsLeftoverSetsFlag(t *testing.T) { } // Add pod to unschedulablePods (simulating failed scheduling) - err = q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, pod, "fakePlugin"), q.SchedulingCycle()) + err = q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, pod, "fakePlugin"), q.SchedulingCycle(), "") if err != nil { t.Fatalf("Unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -3248,7 +3279,7 @@ func TestFlushUnschedulablePodsLeftoverSetsFlag(t *testing.T) { } // Simulate pod failing to schedule again and returning to queue - err = q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, pInfo.Pod, "fakePlugin"), q.SchedulingCycle()) + err = q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, pInfo.Pod, "fakePlugin"), q.SchedulingCycle(), "") if err != nil { t.Fatalf("Unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -3375,7 +3406,7 @@ var ( } // Simulate plugins that are waiting for some events. p.UnschedulablePlugins = unschedulablePlugins - if err := queue.AddUnschedulableIfNotPresent(logger, p, 1); err != nil { + if err := queue.AddUnschedulableIfNotPresent(logger, p, 1, ""); err != nil { tCtx.Fatalf("Unexpected error during AddUnschedulableIfNotPresent: %v", err) } } @@ -3392,7 +3423,7 @@ var ( // needs to increment it to make it backoff p.UnschedulableCount++ // When there is no known unschedulable plugin, pods always go to the backoff queue. - if err := queue.AddUnschedulableIfNotPresent(logger, p, 1); err != nil { + if err := queue.AddUnschedulableIfNotPresent(logger, p, 1, ""); err != nil { tCtx.Fatalf("Unexpected error during AddUnschedulableIfNotPresent: %v", err) } } @@ -4030,7 +4061,9 @@ func TestPerPodSchedulingMetrics(t *testing.T) { } pInfo.UnschedulablePlugins = sets.New("plugin") - queue.AddUnschedulableIfNotPresent(logger, pInfo, 1) + if err := queue.AddUnschedulableIfNotPresent(logger, pInfo, 1, ""); err != nil { + t.Fatalf("Unexpected error from AddUnschedulableIfNotPresent: %v", err) + } // Override clock to exceed the DefaultPodMaxInUnschedulablePodsDuration so that unschedulable pods // will be moved to activeQ c.SetTime(timestamp.Add(DefaultPodMaxInUnschedulablePodsDuration + 1)) @@ -4050,7 +4083,9 @@ func TestPerPodSchedulingMetrics(t *testing.T) { } pInfo.UnschedulablePlugins = sets.New("plugin") - queue.AddUnschedulableIfNotPresent(logger, pInfo, 1) + if err := queue.AddUnschedulableIfNotPresent(logger, pInfo, 1, ""); err != nil { + t.Fatalf("Unexpected error from AddUnschedulableIfNotPresent: %v", err) + } // Override clock to exceed the DefaultPodMaxInUnschedulablePodsDuration so that unschedulable pods // will be moved to activeQ updatedTimestamp := timestamp @@ -4242,7 +4277,7 @@ func TestBackOffFlow(t *testing.T) { t.Errorf("got attempts %d, want %d", podInfo.Attempts, i+1) } podInfo.UnschedulablePlugins.Insert("unsched-plugin") - err = q.AddUnschedulableIfNotPresent(logger, podInfo, int64(i)) + err = q.AddUnschedulableIfNotPresent(logger, podInfo, int64(i), "") if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -4355,7 +4390,7 @@ func TestMoveAllToActiveOrBackoffQueue_PreEnqueueChecks(t *testing.T) { t.Errorf("Unexpected pod after Pop (-want, +got):\n%s", diff) } podInfo.UnschedulablePlugins = sets.New("plugin") - err := q.AddUnschedulableIfNotPresent(logger, podInfo, q.activeQ.schedulingCycle()) + err := q.AddUnschedulableIfNotPresent(logger, podInfo, q.activeQ.schedulingCycle(), "") if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } diff --git a/pkg/scheduler/eventhandlers_test.go b/pkg/scheduler/eventhandlers_test.go index 966bab58e8b..0b98b17239f 100644 --- a/pkg/scheduler/eventhandlers_test.go +++ b/pkg/scheduler/eventhandlers_test.go @@ -194,7 +194,7 @@ func TestEventHandlers_MoveToActiveOnNominatedNodeUpdate(t *testing.T) { t.Fatalf("Pop failed: %v", err) } poppedPod.UnschedulablePlugins = sets.New("fooPlugin1") - if err := queue.AddUnschedulableIfNotPresent(logger, poppedPod, queue.SchedulingCycle()); err != nil { + if err := queue.AddUnschedulableIfNotPresent(logger, poppedPod, queue.SchedulingCycle(), ""); err != nil { t.Errorf("Unexpected error from AddUnschedulableIfNotPresent: %v", err) } } diff --git a/pkg/scheduler/schedule_one.go b/pkg/scheduler/schedule_one.go index c8318322275..012e2a163bd 100644 --- a/pkg/scheduler/schedule_one.go +++ b/pkg/scheduler/schedule_one.go @@ -1256,12 +1256,16 @@ func (sched *Scheduler) handleSchedulingFailure(ctx context.Context, podFwk fram logger.Info("Pod has been assigned to node. Abort adding it back to queue.", "pod", klog.KObj(pod), "node", cachedPod.Spec.NodeName) // We need to call DonePod here because we don't call AddUnschedulableIfNotPresent in this case. } else { + // Remember the UID under which this pod is tracked as in-flight (from Pop) before we + // replace podInfo with the informer copy. A delete+recreate with the same name can change + // the UID; Done and queueing hints must still target the popped UID (kubernetes/kubernetes#138316). + poppedInFlightUID := podInfo.Pod.UID // As is from SharedInformer, we need to do a DeepCopy() here. // ignore this err since apiserver doesn't properly validate affinity terms // and we can't fix the validation for backwards compatibility. podInfo.PodInfo, _ = framework.NewPodInfo(cachedPod.DeepCopy()) pod = podInfo.Pod - if err := sched.SchedulingQueue.AddUnschedulableIfNotPresent(logger, podInfo, sched.SchedulingQueue.SchedulingCycle()); err != nil { + if err := sched.SchedulingQueue.AddUnschedulableIfNotPresent(logger, podInfo, sched.SchedulingQueue.SchedulingCycle(), poppedInFlightUID); err != nil { utilruntime.HandleErrorWithContext(ctx, err, "Error occurred") } calledDone = true diff --git a/pkg/scheduler/schedule_one_podgroup_test.go b/pkg/scheduler/schedule_one_podgroup_test.go index 9a0347dc6ca..511f51772ff 100644 --- a/pkg/scheduler/schedule_one_podgroup_test.go +++ b/pkg/scheduler/schedule_one_podgroup_test.go @@ -177,7 +177,7 @@ func TestPodGroupInfoForPod(t *testing.T) { q := internalqueue.NewTestQueue(ctx, nil) for _, pInfo := range tt.queuePods { - err := q.AddUnschedulableIfNotPresent(logger, pInfo, 0) + err := q.AddUnschedulableIfNotPresent(logger, pInfo, 0, "") if err != nil { t.Fatalf("Failed to add unschedulable pod: %v", err) } diff --git a/test/integration/scheduler/queueing/queue_test.go b/test/integration/scheduler/queueing/queue_test.go index b11286c502b..fb13b104265 100644 --- a/test/integration/scheduler/queueing/queue_test.go +++ b/test/integration/scheduler/queueing/queue_test.go @@ -341,8 +341,11 @@ func TestCustomResourceEnqueue(t *testing.T) { testCtx.Scheduler.FailureHandler(ctx, schedFramework, podInfo, fwk.NewStatus(fwk.Unschedulable).WithError(fitError), nil, time.Now()) // Scheduling cycle is incremented from 0 to 1 after NextPod() is called, so - // pass a number larger than 1 to move Pod to unschedulablePods. - testCtx.Scheduler.SchedulingQueue.AddUnschedulableIfNotPresent(logger, podInfo, 10) + // pass a number larger than 1 to move Pod to unschedulablePods. FailureHandler above may have + // already requeued the Pod, in which case the duplicate AddUnschedulableIfNotPresent call is benign. + if err := testCtx.Scheduler.SchedulingQueue.AddUnschedulableIfNotPresent(logger, podInfo, 10, ""); err != nil && err.Error() != fmt.Sprintf("Pod %v is already present in the active queue", klog.KObj(podInfo.Pod)) { + t.Fatalf("AddUnschedulableIfNotPresent failed: %v", err) + } // Trigger a Custom Resource event. // We expect this event to trigger moving the test Pod from unschedulablePods to activeQ. From 48c8f0deb1205acb106f1f6e21ae39bffb845b64 Mon Sep 17 00:00:00 2001 From: Samarth Verma Date: Mon, 13 Apr 2026 11:53:50 -0400 Subject: [PATCH 2/3] scheduler: skip requeueing recreated pods on scheduling failure --- pkg/scheduler/backend/queue/active_queue.go | 18 +-- .../backend/queue/scheduling_queue.go | 25 ++-- .../backend/queue/scheduling_queue_test.go | 117 +++++++----------- pkg/scheduler/eventhandlers_test.go | 2 +- pkg/scheduler/schedule_one.go | 12 +- pkg/scheduler/schedule_one_podgroup_test.go | 2 +- pkg/scheduler/schedule_one_test.go | 70 +++++++++++ .../scheduler/queueing/queue_test.go | 8 -- 8 files changed, 137 insertions(+), 117 deletions(-) diff --git a/pkg/scheduler/backend/queue/active_queue.go b/pkg/scheduler/backend/queue/active_queue.go index 5d1fdb78bc2..6728f2f0e3f 100644 --- a/pkg/scheduler/backend/queue/active_queue.go +++ b/pkg/scheduler/backend/queue/active_queue.go @@ -46,9 +46,8 @@ type activeQueuer interface { movePodToInFlight(pInfo *framework.QueuedPodInfo) error listInFlightEvents() []interface{} listInFlightPods() []*v1.Pod - // clusterEventsForPod collects cluster events recorded while the pod registered under inFlightListUID - // was in flight. If inFlightListUID is empty, pInfo.Pod.UID is used. - clusterEventsForPod(logger klog.Logger, pInfo *framework.QueuedPodInfo, inFlightListUID types.UID) ([]*clusterEvent, error) + // clusterEventsForPod collects cluster events recorded while pInfo.Pod was in flight. + clusterEventsForPod(logger klog.Logger, pInfo *framework.QueuedPodInfo) ([]*clusterEvent, error) addEventsIfPodInFlight(oldPod, newPod *v1.Pod, events []fwk.ClusterEvent) bool addEventIfAnyInFlight(oldObj, newObj interface{}, event fwk.ClusterEvent) bool @@ -398,21 +397,16 @@ func (aq *activeQueue) listInFlightPods() []*v1.Pod { return pods } -// clusterEventsForPod gets all cluster events that have happened while the scheduling attempt -// for the pod keyed by inFlightListUID (in inFlightPods) was in flight. -func (aq *activeQueue) clusterEventsForPod(logger klog.Logger, pInfo *framework.QueuedPodInfo, inFlightListUID types.UID) ([]*clusterEvent, error) { +// clusterEventsForPod gets all cluster events that have happened while pInfo.Pod was in flight. +func (aq *activeQueue) clusterEventsForPod(logger klog.Logger, pInfo *framework.QueuedPodInfo) ([]*clusterEvent, error) { aq.lock.RLock() defer aq.lock.RUnlock() logger.V(5).Info("Checking events for in-flight pod", "pod", klog.KObj(pInfo.Pod), "unschedulablePlugins", pInfo.UnschedulablePlugins, "inFlightEventsSize", aq.inFlightEvents.Len(), "inFlightPodsSize", len(aq.inFlightPods)) - listUID := inFlightListUID - if listUID == "" { - listUID = pInfo.Pod.UID - } // AddUnschedulableIfNotPresent is called with the Pod at the end of scheduling or binding. // So, given pInfo should have been Pop()ed before, - // we can assume listUID must be recorded in inFlightPods and thus inFlightEvents. - inFlightPod, ok := aq.inFlightPods[listUID] + // we can assume pInfo.Pod.UID must be recorded in inFlightPods and thus inFlightEvents. + inFlightPod, ok := aq.inFlightPods[pInfo.Pod.UID] if !ok { return nil, fmt.Errorf("in flight Pod isn't found in the scheduling queue. If you see this error log, it's likely a bug in the scheduler") } diff --git a/pkg/scheduler/backend/queue/scheduling_queue.go b/pkg/scheduler/backend/queue/scheduling_queue.go index aacb7394ecc..ae324fa3184 100644 --- a/pkg/scheduler/backend/queue/scheduling_queue.go +++ b/pkg/scheduler/backend/queue/scheduling_queue.go @@ -108,11 +108,7 @@ type SchedulingQueue interface { // AddUnschedulableIfNotPresent adds an unschedulable pod back to scheduling queue. // The podSchedulingCycle represents the current scheduling cycle number which can be // returned by calling SchedulingCycle(). - // schedulingCycleInFlightUID is the UID under which the pod is registered in inFlightPods (from Pop). - // Use the empty string to mean pInfo.Pod.UID. When the pod object is refreshed from the API/informer - // after a delete+recreate with the same name, pInfo.Pod.UID may differ; pass the pre-refresh UID so - // Done and queueing-hint logic use the correct in-flight entry (see kubernetes/kubernetes#138316). - AddUnschedulableIfNotPresent(logger klog.Logger, pod *framework.QueuedPodInfo, podSchedulingCycle int64, schedulingCycleInFlightUID types.UID) error + AddUnschedulableIfNotPresent(logger klog.Logger, pod *framework.QueuedPodInfo, podSchedulingCycle int64) error // SchedulingCycle returns the current number of scheduling cycle which is // cached by scheduling queue. Normally, incrementing this number whenever // a pod is popped (e.g. called Pop()) is enough. @@ -819,8 +815,7 @@ func (p *PriorityQueue) SchedulingCycle() int64 { // determineSchedulingHintForInFlightPod looks at the unschedulable plugins of the given Pod // and determines the scheduling hint for this Pod while checking the events that happened during in-flight. -// inFlightListUID is the key in inFlightPods for this scheduling attempt (empty means pInfo.Pod.UID). -func (p *PriorityQueue) determineSchedulingHintForInFlightPod(logger klog.Logger, pInfo *framework.QueuedPodInfo, inFlightListUID types.UID) queueingStrategy { +func (p *PriorityQueue) determineSchedulingHintForInFlightPod(logger klog.Logger, pInfo *framework.QueuedPodInfo) queueingStrategy { if len(pInfo.UnschedulablePlugins) == 0 && len(pInfo.PendingPlugins) == 0 { // No failed plugins are associated with this Pod. // Meaning something unusual (a temporal failure on kube-apiserver, etc) happened and this Pod gets moved back to the queue. @@ -834,7 +829,7 @@ func (p *PriorityQueue) determineSchedulingHintForInFlightPod(logger klog.Logger return queueAfterBackoff } - events, err := p.activeQ.clusterEventsForPod(logger, pInfo, inFlightListUID) + events, err := p.activeQ.clusterEventsForPod(logger, pInfo) if err != nil { utilruntime.HandleErrorWithLogger(logger, err, "Error getting cluster events for pod", "pod", klog.KObj(pInfo.Pod)) return queueAfterBackoff @@ -907,18 +902,12 @@ func (p *PriorityQueue) addUnschedulableWithoutQueueingHint(logger klog.Logger, // the queue, unless it is already in the queue. Normally, PriorityQueue puts // unschedulable pods in `unschedulablePods`. But if there has been a recent move // request, then the pod is put in `backoffQ`. -func (p *PriorityQueue) AddUnschedulableIfNotPresent(logger klog.Logger, pInfo *framework.QueuedPodInfo, podSchedulingCycle int64, schedulingCycleInFlightUID types.UID) error { +func (p *PriorityQueue) AddUnschedulableIfNotPresent(logger klog.Logger, pInfo *framework.QueuedPodInfo, podSchedulingCycle int64) error { p.lock.Lock() defer p.lock.Unlock() - doneUID := pInfo.Pod.UID - if schedulingCycleInFlightUID != "" { - doneUID = schedulingCycleInFlightUID - } - // In any case, this scheduling attempt will be moved back to the queue and we should call Done - // for the UID registered in inFlightPods at Pop time (doneUID), which may differ from pInfo.Pod.UID - // if the pod was refreshed from the informer after a same-name recreate. - defer p.Done(doneUID) + // In any case, this scheduling attempt will be moved back to the queue and we should call Done. + defer p.Done(pInfo.Pod.UID) pod := pInfo.Pod if p.unschedulablePods.get(pod) != nil { @@ -962,7 +951,7 @@ func (p *PriorityQueue) AddUnschedulableIfNotPresent(logger klog.Logger, pInfo * } // We check whether this Pod may change its scheduling result by any of events that happened during scheduling. - schedulingHint := p.determineSchedulingHintForInFlightPod(logger, pInfo, doneUID) + schedulingHint := p.determineSchedulingHintForInFlightPod(logger, pInfo) // In this case, we try to requeue this Pod to activeQ/backoffQ. queue := p.requeuePodWithQueueingStrategy(logger, pInfo, schedulingHint, framework.ScheduleAttemptFailure) diff --git a/pkg/scheduler/backend/queue/scheduling_queue_test.go b/pkg/scheduler/backend/queue/scheduling_queue_test.go index 2655c375677..8070b3202d9 100644 --- a/pkg/scheduler/backend/queue/scheduling_queue_test.go +++ b/pkg/scheduler/backend/queue/scheduling_queue_test.go @@ -617,7 +617,7 @@ func Test_InFlightPods(t *testing.T) { logger, _ := ktesting.NewTestContext(t) // This pod will be requeued to backoffQ immediately because no plugin is registered as unschedulable plugin, // which means the pod encountered an unexpected error (e.g., a network error). - err := q.AddUnschedulableIfNotPresent(logger, poppedPod, q.SchedulingCycle(), "") + err := q.AddUnschedulableIfNotPresent(logger, poppedPod, q.SchedulingCycle()) if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -626,7 +626,7 @@ func Test_InFlightPods(t *testing.T) { logger, _ := ktesting.NewTestContext(t) poppedPod2.UnschedulablePlugins = sets.New("fooPlugin2", "fooPlugin3") poppedPod2.PendingPlugins = sets.New("fooPlugin1") - err := q.AddUnschedulableIfNotPresent(logger, poppedPod2, q.SchedulingCycle(), "") + err := q.AddUnschedulableIfNotPresent(logger, poppedPod2, q.SchedulingCycle()) if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -670,7 +670,7 @@ func Test_InFlightPods(t *testing.T) { // Unschedulable due to PendingPlugins. poppedPod.PendingPlugins = sets.New("fooPlugin1") poppedPod.UnschedulablePlugins = sets.New("fooPlugin2") - if err := q.AddUnschedulableIfNotPresent(logger, poppedPod, q.SchedulingCycle(), ""); err != nil { + if err := q.AddUnschedulableIfNotPresent(logger, poppedPod, q.SchedulingCycle()); err != nil { t.Errorf("Unexpected error from AddUnschedulableIfNotPresent: %v", err) } }}, @@ -689,7 +689,7 @@ func Test_InFlightPods(t *testing.T) { {callback: func(t *testing.T, q *PriorityQueue) { logger, _ := ktesting.NewTestContext(t) // Failed (i.e. no UnschedulablePlugins). Should go to backoff. - if err := q.AddUnschedulableIfNotPresent(logger, poppedPod, q.SchedulingCycle(), ""); err != nil { + if err := q.AddUnschedulableIfNotPresent(logger, poppedPod, q.SchedulingCycle()); err != nil { t.Errorf("Unexpected error from AddUnschedulableIfNotPresent: %v", err) } }}, @@ -822,7 +822,7 @@ func Test_InFlightPods(t *testing.T) { case action.eventHappens != nil: q.MoveAllToActiveOrBackoffQueue(logger, *action.eventHappens, nil, nil, nil) case action.podEnqueued != nil: - err := q.AddUnschedulableIfNotPresent(logger, action.podEnqueued, q.SchedulingCycle(), "") + err := q.AddUnschedulableIfNotPresent(logger, action.podEnqueued, q.SchedulingCycle()) if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -942,7 +942,7 @@ func TestPop(t *testing.T) { poppedPod := popPod(t, logger, q, pod) // We put register the plugin to PendingPlugins so that it's interpreted as queueImmediately and skip backoff. poppedPod.PendingPlugins = sets.New("fooPlugin1") - if err := q.AddUnschedulableIfNotPresent(logger, poppedPod, q.SchedulingCycle(), ""); err != nil { + if err := q.AddUnschedulableIfNotPresent(logger, poppedPod, q.SchedulingCycle()); err != nil { t.Errorf("Unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -976,7 +976,7 @@ func TestPriorityQueue_AddUnschedulableIfNotPresent(t *testing.T) { } q.Add(ctx, highPriNominatedPodInfo.Pod) - err := q.AddUnschedulableIfNotPresent(logger, newQueuedPodInfoForLookup(unschedulablePodInfo.Pod, "plugin"), q.SchedulingCycle(), "") + err := q.AddUnschedulableIfNotPresent(logger, newQueuedPodInfoForLookup(unschedulablePodInfo.Pod, "plugin"), q.SchedulingCycle()) if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -994,37 +994,6 @@ func TestPriorityQueue_AddUnschedulableIfNotPresent(t *testing.T) { } } -// TestPriorityQueue_AddUnschedulableIfNotPresent_PodRecreatedSameName covers the case where the -// informer returns a new Pod UID for the same namespace/name before AddUnschedulableIfNotPresent -// (see kubernetes/kubernetes#138316): Done and queueing hints must use the popped scheduling UID. -func TestPriorityQueue_AddUnschedulableIfNotPresent_PodRecreatedSameName(t *testing.T) { - featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.SchedulerQueueingHints, true) - - logger, ctx := ktesting.NewTestContext(t) - ctx, cancel := context.WithCancel(ctx) - defer cancel() - - oldUID := types.UID("popped-pod-uid") - pod := st.MakePod().Name("foo").Namespace("ns").UID(string(oldUID)).Obj() - q := NewTestQueueWithObjects(ctx, newDefaultQueueSort(), []runtime.Object{pod}) - q.Add(ctx, pod) - popped, err := q.Pop(logger) - if err != nil { - t.Fatalf("Pop: %v", err) - } - if popped.Pod.UID != oldUID { - t.Fatalf("unexpected popped UID: got %v want %v", popped.Pod.UID, oldUID) - } - - recreated := pod.DeepCopy() - recreated.UID = types.UID("apiserver-new-uid-after-recreate") - pInfo := newQueuedPodInfoForLookup(recreated, "plugin") - if err := q.AddUnschedulableIfNotPresent(logger, pInfo, q.SchedulingCycle(), oldUID); err != nil { - t.Fatalf("AddUnschedulableIfNotPresent: %v", err) - } - expectInFlightPods(t, q) -} - // TestPriorityQueue_AddUnschedulableIfNotPresent_Backoff tests the scenarios when // AddUnschedulableIfNotPresent is called asynchronously. // Pods in and before current scheduling cycle will be put back to activeQueue @@ -1074,7 +1043,7 @@ func TestPriorityQueue_AddUnschedulableIfNotPresent_Backoff(t *testing.T) { }, } - err := q.AddUnschedulableIfNotPresent(logger, newQueuedPodInfoForLookup(unschedulablePod, "plugin"), oldCycle, "") + err := q.AddUnschedulableIfNotPresent(logger, newQueuedPodInfoForLookup(unschedulablePod, "plugin"), oldCycle) if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -1521,7 +1490,7 @@ func TestPriorityQueue_UpdateWhenInflight(t *testing.T) { // test-pod got rejected by fakePlugin, // but the update event that it just got may change this scheduling result, // and hence we should put this pod to activeQ/backoffQ. - err := q.AddUnschedulableIfNotPresent(logger, newQueuedPodInfoForLookup(updatedPod, "fakePlugin"), q.SchedulingCycle(), "") + err := q.AddUnschedulableIfNotPresent(logger, newQueuedPodInfoForLookup(updatedPod, "fakePlugin"), q.SchedulingCycle()) if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -2062,7 +2031,7 @@ func BenchmarkMoveAllToActiveOrBackoffQueue(b *testing.B) { // Random case. podInfo = q.newQueuedPodInfo(ctx, p, plugins[j%len(plugins)]) } - err := q.AddUnschedulableIfNotPresent(logger, podInfo, q.SchedulingCycle(), "") + err := q.AddUnschedulableIfNotPresent(logger, podInfo, q.SchedulingCycle()) if err != nil { b.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -2174,7 +2143,7 @@ func TestPriorityQueue_MoveAllToActiveOrBackoffQueueWithQueueingHint(t *testing. t.Errorf("Unexpected pod after Pop (-want, +got):\n%s", diff) } // add to unsched pod pool - err := q.AddUnschedulableIfNotPresent(logger, test.podInfo, q.SchedulingCycle(), "") + err := q.AddUnschedulableIfNotPresent(logger, test.podInfo, q.SchedulingCycle()) if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -2226,11 +2195,11 @@ func TestPriorityQueue_MoveAllToActiveOrBackoffQueue(t *testing.T) { t.Errorf("Unexpected pod after Pop (-want, +got):\n%s", diff) } expectInFlightPods(t, q, unschedulablePodInfo.Pod.UID, highPriorityPodInfo.Pod.UID) - err := q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, unschedulablePodInfo.Pod, "fooPlugin"), q.SchedulingCycle(), "") + err := q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, unschedulablePodInfo.Pod, "fooPlugin"), q.SchedulingCycle()) if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } - err = q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, highPriorityPodInfo.Pod, "fooPlugin"), q.SchedulingCycle(), "") + err = q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, highPriorityPodInfo.Pod, "fooPlugin"), q.SchedulingCycle()) if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -2247,7 +2216,7 @@ func TestPriorityQueue_MoveAllToActiveOrBackoffQueue(t *testing.T) { // This Pod will go to backoffQ because no failure plugin is associated with it. hpp1PodInfo := q.newQueuedPodInfo(ctx, hpp1) hpp1PodInfo.UnschedulableCount++ - err = q.AddUnschedulableIfNotPresent(logger, hpp1PodInfo, q.SchedulingCycle(), "") + err = q.AddUnschedulableIfNotPresent(logger, hpp1PodInfo, q.SchedulingCycle()) if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -2262,7 +2231,7 @@ func TestPriorityQueue_MoveAllToActiveOrBackoffQueue(t *testing.T) { } expectInFlightPods(t, q, hpp2.UID) // This Pod will go to the unschedulable Pod pool. - err = q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, hpp2, "barPlugin"), q.SchedulingCycle(), "") + err = q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, hpp2, "barPlugin"), q.SchedulingCycle()) if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -2316,17 +2285,17 @@ func TestPriorityQueue_MoveAllToActiveOrBackoffQueue(t *testing.T) { highPriorityQueuedPodInfo := q.newQueuedPodInfo(ctx, highPriorityPodInfo.Pod, "fooPlugin") hpp1QueuedPodInfo := q.newQueuedPodInfo(ctx, hpp1) expectInFlightPods(t, q, medPriorityPodInfo.Pod.UID, unschedulablePodInfo.Pod.UID, highPriorityPodInfo.Pod.UID, hpp1.UID) - err = q.AddUnschedulableIfNotPresent(logger, unschedulableQueuedPodInfo, q.SchedulingCycle(), "") + err = q.AddUnschedulableIfNotPresent(logger, unschedulableQueuedPodInfo, q.SchedulingCycle()) if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } expectInFlightPods(t, q, medPriorityPodInfo.Pod.UID, highPriorityPodInfo.Pod.UID, hpp1.UID) - err = q.AddUnschedulableIfNotPresent(logger, highPriorityQueuedPodInfo, q.SchedulingCycle(), "") + err = q.AddUnschedulableIfNotPresent(logger, highPriorityQueuedPodInfo, q.SchedulingCycle()) if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } expectInFlightPods(t, q, medPriorityPodInfo.Pod.UID, hpp1.UID) - err = q.AddUnschedulableIfNotPresent(logger, hpp1QueuedPodInfo, q.SchedulingCycle(), "") + err = q.AddUnschedulableIfNotPresent(logger, hpp1QueuedPodInfo, q.SchedulingCycle()) if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -2383,11 +2352,11 @@ func TestPriorityQueue_MoveAllToActiveOrBackoffQueueWithoutQueueingHint(t *testi // To simulate the pod is failed in scheduling in the real world, Pop() the pod from activeQ before AddUnschedulableIfNotPresent()s below. q.Add(ctx, medPriorityPodInfo.Pod) - err := q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, unschedulablePodInfo.Pod, "fooPlugin"), q.SchedulingCycle(), "") + err := q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, unschedulablePodInfo.Pod, "fooPlugin"), q.SchedulingCycle()) if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } - err = q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, highPriorityPodInfo.Pod, "fooPlugin"), q.SchedulingCycle(), "") + err = q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, highPriorityPodInfo.Pod, "fooPlugin"), q.SchedulingCycle()) if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -2396,14 +2365,14 @@ func TestPriorityQueue_MoveAllToActiveOrBackoffQueueWithoutQueueingHint(t *testi // This Pod will go to backoffQ because no failure plugin is associated with it. hpp1PodInfo := q.newQueuedPodInfo(ctx, hpp1) hpp1PodInfo.UnschedulableCount++ - err = q.AddUnschedulableIfNotPresent(logger, hpp1PodInfo, q.SchedulingCycle(), "") + err = q.AddUnschedulableIfNotPresent(logger, hpp1PodInfo, q.SchedulingCycle()) if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } // Construct another Pod, and associate its scheduler failure to plugin "barPlugin". hpp2 := clonePod(highPriorityPodInfo.Pod, "hpp2") // This Pod will go to the unschedulable Pod pool. - err = q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, hpp2, "barPlugin"), q.SchedulingCycle(), "") + err = q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, hpp2, "barPlugin"), q.SchedulingCycle()) if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -2432,15 +2401,15 @@ func TestPriorityQueue_MoveAllToActiveOrBackoffQueueWithoutQueueingHint(t *testi unschedulableQueuedPodInfo := q.newQueuedPodInfo(ctx, unschedulablePodInfo.Pod, "fooPlugin") highPriorityQueuedPodInfo := q.newQueuedPodInfo(ctx, highPriorityPodInfo.Pod, "fooPlugin") hpp1QueuedPodInfo := q.newQueuedPodInfo(ctx, hpp1) - err = q.AddUnschedulableIfNotPresent(logger, unschedulableQueuedPodInfo, q.SchedulingCycle(), "") + err = q.AddUnschedulableIfNotPresent(logger, unschedulableQueuedPodInfo, q.SchedulingCycle()) if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } - err = q.AddUnschedulableIfNotPresent(logger, highPriorityQueuedPodInfo, q.SchedulingCycle(), "") + err = q.AddUnschedulableIfNotPresent(logger, highPriorityQueuedPodInfo, q.SchedulingCycle()) if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } - err = q.AddUnschedulableIfNotPresent(logger, hpp1QueuedPodInfo, q.SchedulingCycle(), "") + err = q.AddUnschedulableIfNotPresent(logger, hpp1QueuedPodInfo, q.SchedulingCycle()) if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -2587,7 +2556,7 @@ func TestPriorityQueue_AssignedPodAdded_(t *testing.T) { t.Errorf("Unexpected pod after Pop (-want, +got):\n%s", diff) } - err := q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, tt.unschedPod, tt.unschedPlugin), q.SchedulingCycle(), "") + err := q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, tt.unschedPod, tt.unschedPlugin), q.SchedulingCycle()) if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -2703,7 +2672,7 @@ func TestPriorityQueue_AssignedPodUpdated(t *testing.T) { t.Errorf("Unexpected pod after Pop (-want, +got):\n%s", diff) } - err := q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, tt.unschedPod, tt.unschedPlugin), q.SchedulingCycle(), "") + err := q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, tt.unschedPod, tt.unschedPlugin), q.SchedulingCycle()) if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -2828,11 +2797,11 @@ func TestPriorityQueue_PendingPods(t *testing.T) { t.Errorf("Unexpected pod after Pop (-want, +got):\n%s", diff) } q.Add(ctx, medPriorityPodInfo.Pod) - err := q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, unschedulablePodInfo.Pod, "plugin"), q.SchedulingCycle(), "") + err := q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, unschedulablePodInfo.Pod, "plugin"), q.SchedulingCycle()) if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } - err = q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, highPriorityPodInfo.Pod, "plugin"), q.SchedulingCycle(), "") + err = q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, highPriorityPodInfo.Pod, "plugin"), q.SchedulingCycle()) if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -3018,7 +2987,7 @@ func TestRecentlyTriedPodsGoBack(t *testing.T) { }) p1.UnschedulablePlugins = sets.New("plugin") // Put in the unschedulable queue. - err = q.AddUnschedulableIfNotPresent(logger, p1, q.SchedulingCycle(), "") + err = q.AddUnschedulableIfNotPresent(logger, p1, q.SchedulingCycle()) if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -3070,7 +3039,7 @@ func TestPodFailedSchedulingMultipleTimesDoesNotBlockNewerPod(t *testing.T) { t.Errorf("Unexpected pod after Pop (-want, +got):\n%s", diff) } // Put in the unschedulable queue - err := q.AddUnschedulableIfNotPresent(logger, newQueuedPodInfoForLookup(unschedulablePod, "plugin"), q.SchedulingCycle(), "") + err := q.AddUnschedulableIfNotPresent(logger, newQueuedPodInfoForLookup(unschedulablePod, "plugin"), q.SchedulingCycle()) if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -3102,7 +3071,7 @@ func TestPodFailedSchedulingMultipleTimesDoesNotBlockNewerPod(t *testing.T) { }) // And then, put unschedulable pod to the unschedulable queue - err = q.AddUnschedulableIfNotPresent(logger, newQueuedPodInfoForLookup(unschedulablePod, "plugin"), q.SchedulingCycle(), "") + err = q.AddUnschedulableIfNotPresent(logger, newQueuedPodInfoForLookup(unschedulablePod, "plugin"), q.SchedulingCycle()) if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -3149,7 +3118,7 @@ func TestHighPriorityBackoff(t *testing.T) { Message: "fake scheduling failure", }) // Put in the unschedulable queue. - err = q.AddUnschedulableIfNotPresent(logger, newQueuedPodInfoForLookup(p.Pod, "fooPlugin"), q.SchedulingCycle(), "") + err = q.AddUnschedulableIfNotPresent(logger, newQueuedPodInfoForLookup(p.Pod, "fooPlugin"), q.SchedulingCycle()) if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -3212,11 +3181,11 @@ func TestHighPriorityFlushUnschedulablePodsLeftover(t *testing.T) { } else if diff := cmp.Diff(midPod, p.Pod); diff != "" { t.Errorf("Unexpected pod after Pop (-want, +got):\n%s", diff) } - err := q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, highPod, "fakePlugin"), q.SchedulingCycle(), "") + err := q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, highPod, "fakePlugin"), q.SchedulingCycle()) if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } - err = q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, midPod, "fakePlugin"), q.SchedulingCycle(), "") + err = q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, midPod, "fakePlugin"), q.SchedulingCycle()) if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -3260,7 +3229,7 @@ func TestFlushUnschedulablePodsLeftoverSetsFlag(t *testing.T) { } // Add pod to unschedulablePods (simulating failed scheduling) - err = q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, pod, "fakePlugin"), q.SchedulingCycle(), "") + err = q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, pod, "fakePlugin"), q.SchedulingCycle()) if err != nil { t.Fatalf("Unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -3279,7 +3248,7 @@ func TestFlushUnschedulablePodsLeftoverSetsFlag(t *testing.T) { } // Simulate pod failing to schedule again and returning to queue - err = q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, pInfo.Pod, "fakePlugin"), q.SchedulingCycle(), "") + err = q.AddUnschedulableIfNotPresent(logger, q.newQueuedPodInfo(ctx, pInfo.Pod, "fakePlugin"), q.SchedulingCycle()) if err != nil { t.Fatalf("Unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -3406,7 +3375,7 @@ var ( } // Simulate plugins that are waiting for some events. p.UnschedulablePlugins = unschedulablePlugins - if err := queue.AddUnschedulableIfNotPresent(logger, p, 1, ""); err != nil { + if err := queue.AddUnschedulableIfNotPresent(logger, p, 1); err != nil { tCtx.Fatalf("Unexpected error during AddUnschedulableIfNotPresent: %v", err) } } @@ -3423,7 +3392,7 @@ var ( // needs to increment it to make it backoff p.UnschedulableCount++ // When there is no known unschedulable plugin, pods always go to the backoff queue. - if err := queue.AddUnschedulableIfNotPresent(logger, p, 1, ""); err != nil { + if err := queue.AddUnschedulableIfNotPresent(logger, p, 1); err != nil { tCtx.Fatalf("Unexpected error during AddUnschedulableIfNotPresent: %v", err) } } @@ -4061,7 +4030,7 @@ func TestPerPodSchedulingMetrics(t *testing.T) { } pInfo.UnschedulablePlugins = sets.New("plugin") - if err := queue.AddUnschedulableIfNotPresent(logger, pInfo, 1, ""); err != nil { + if err := queue.AddUnschedulableIfNotPresent(logger, pInfo, 1); err != nil { t.Fatalf("Unexpected error from AddUnschedulableIfNotPresent: %v", err) } // Override clock to exceed the DefaultPodMaxInUnschedulablePodsDuration so that unschedulable pods @@ -4083,7 +4052,7 @@ func TestPerPodSchedulingMetrics(t *testing.T) { } pInfo.UnschedulablePlugins = sets.New("plugin") - if err := queue.AddUnschedulableIfNotPresent(logger, pInfo, 1, ""); err != nil { + if err := queue.AddUnschedulableIfNotPresent(logger, pInfo, 1); err != nil { t.Fatalf("Unexpected error from AddUnschedulableIfNotPresent: %v", err) } // Override clock to exceed the DefaultPodMaxInUnschedulablePodsDuration so that unschedulable pods @@ -4277,7 +4246,7 @@ func TestBackOffFlow(t *testing.T) { t.Errorf("got attempts %d, want %d", podInfo.Attempts, i+1) } podInfo.UnschedulablePlugins.Insert("unsched-plugin") - err = q.AddUnschedulableIfNotPresent(logger, podInfo, int64(i), "") + err = q.AddUnschedulableIfNotPresent(logger, podInfo, int64(i)) if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } @@ -4390,7 +4359,7 @@ func TestMoveAllToActiveOrBackoffQueue_PreEnqueueChecks(t *testing.T) { t.Errorf("Unexpected pod after Pop (-want, +got):\n%s", diff) } podInfo.UnschedulablePlugins = sets.New("plugin") - err := q.AddUnschedulableIfNotPresent(logger, podInfo, q.activeQ.schedulingCycle(), "") + err := q.AddUnschedulableIfNotPresent(logger, podInfo, q.activeQ.schedulingCycle()) if err != nil { t.Fatalf("unexpected error from AddUnschedulableIfNotPresent: %v", err) } diff --git a/pkg/scheduler/eventhandlers_test.go b/pkg/scheduler/eventhandlers_test.go index 0b98b17239f..966bab58e8b 100644 --- a/pkg/scheduler/eventhandlers_test.go +++ b/pkg/scheduler/eventhandlers_test.go @@ -194,7 +194,7 @@ func TestEventHandlers_MoveToActiveOnNominatedNodeUpdate(t *testing.T) { t.Fatalf("Pop failed: %v", err) } poppedPod.UnschedulablePlugins = sets.New("fooPlugin1") - if err := queue.AddUnschedulableIfNotPresent(logger, poppedPod, queue.SchedulingCycle(), ""); err != nil { + if err := queue.AddUnschedulableIfNotPresent(logger, poppedPod, queue.SchedulingCycle()); err != nil { t.Errorf("Unexpected error from AddUnschedulableIfNotPresent: %v", err) } } diff --git a/pkg/scheduler/schedule_one.go b/pkg/scheduler/schedule_one.go index 012e2a163bd..cba9ddef565 100644 --- a/pkg/scheduler/schedule_one.go +++ b/pkg/scheduler/schedule_one.go @@ -1257,15 +1257,21 @@ func (sched *Scheduler) handleSchedulingFailure(ctx context.Context, podFwk fram // We need to call DonePod here because we don't call AddUnschedulableIfNotPresent in this case. } else { // Remember the UID under which this pod is tracked as in-flight (from Pop) before we - // replace podInfo with the informer copy. A delete+recreate with the same name can change - // the UID; Done and queueing hints must still target the popped UID (kubernetes/kubernetes#138316). + // replace podInfo with the informer copy. If the pod was deleted and recreated with the + // same name, skip requeueing or updating the new pod; informer handlers will handle it (#138316). poppedInFlightUID := podInfo.Pod.UID // As is from SharedInformer, we need to do a DeepCopy() here. // ignore this err since apiserver doesn't properly validate affinity terms // and we can't fix the validation for backwards compatibility. podInfo.PodInfo, _ = framework.NewPodInfo(cachedPod.DeepCopy()) pod = podInfo.Pod - if err := sched.SchedulingQueue.AddUnschedulableIfNotPresent(logger, podInfo, sched.SchedulingQueue.SchedulingCycle(), poppedInFlightUID); err != nil { + if pod.UID != poppedInFlightUID { + logger.V(2).Info("Pod was recreated while handling scheduling failure. Skip requeueing and status updates.", "pod", klog.KObj(pod), "oldUID", poppedInFlightUID, "newUID", pod.UID) + sched.SchedulingQueue.Done(poppedInFlightUID) + calledDone = true + return + } + if err := sched.SchedulingQueue.AddUnschedulableIfNotPresent(logger, podInfo, sched.SchedulingQueue.SchedulingCycle()); err != nil { utilruntime.HandleErrorWithContext(ctx, err, "Error occurred") } calledDone = true diff --git a/pkg/scheduler/schedule_one_podgroup_test.go b/pkg/scheduler/schedule_one_podgroup_test.go index 511f51772ff..9a0347dc6ca 100644 --- a/pkg/scheduler/schedule_one_podgroup_test.go +++ b/pkg/scheduler/schedule_one_podgroup_test.go @@ -177,7 +177,7 @@ func TestPodGroupInfoForPod(t *testing.T) { q := internalqueue.NewTestQueue(ctx, nil) for _, pInfo := range tt.queuePods { - err := q.AddUnschedulableIfNotPresent(logger, pInfo, 0, "") + err := q.AddUnschedulableIfNotPresent(logger, pInfo, 0) if err != nil { t.Fatalf("Failed to add unschedulable pod: %v", err) } diff --git a/pkg/scheduler/schedule_one_test.go b/pkg/scheduler/schedule_one_test.go index d6469f9582c..e5ea5060a83 100644 --- a/pkg/scheduler/schedule_one_test.go +++ b/pkg/scheduler/schedule_one_test.go @@ -1318,6 +1318,76 @@ func TestSchedulerScheduleOne(t *testing.T) { } } +func TestHandleSchedulingFailureSkipsRecreatedPod(t *testing.T) { + logger, ctx := ktesting.NewTestContext(t) + ctx, cancel := context.WithCancel(ctx) + defer cancel() + + oldPod := st.MakePod().Name("foo").Namespace("ns").UID("old-uid").SchedulerName(testSchedulerName).Obj() + recreatedPod := oldPod.DeepCopy() + recreatedPod.UID = "new-uid" + + client := clientsetfake.NewClientset(recreatedPod) + informerFactory := informers.NewSharedInformerFactory(client, 0) + eventBroadcaster := events.NewBroadcaster(&events.EventSinkImpl{Interface: client.EventsV1()}) + + schedFramework, err := tf.NewFramework(ctx, + []tf.RegisterPluginFunc{ + tf.RegisterQueueSortPlugin(queuesort.Name, queuesort.New), + tf.RegisterBindPlugin(defaultbinder.Name, defaultbinder.New), + }, + testSchedulerName, + frameworkruntime.WithClientSet(client), + frameworkruntime.WithEventRecorder(eventBroadcaster.NewRecorder(scheme.Scheme, testSchedulerName)), + frameworkruntime.WithInformerFactory(informerFactory), + ) + if err != nil { + t.Fatal(err) + } + + ar := metrics.NewMetricsAsyncRecorder(10, time.Second, ctx.Done()) + queue := internalqueue.NewSchedulingQueue(nil, informerFactory, internalqueue.WithMetricsRecorder(ar)) + sched := &Scheduler{ + client: client, + SchedulingQueue: queue, + } + + informerFactory.Start(ctx.Done()) + informerFactory.WaitForCacheSync(ctx.Done()) + + queue.Add(ctx, oldPod) + popped, err := queue.Pop(logger) + if err != nil { + t.Fatalf("Pop: %v", err) + } + + nominatingInfo := &fwk.NominatingInfo{NominatingMode: fwk.ModeOverride, NominatedNodeName: "node1"} + sched.handleSchedulingFailure(ctx, schedFramework, popped, fwk.NewStatus(fwk.Unschedulable, "no fit"), nominatingInfo, time.Now()) + + if err := wait.PollUntilContextTimeout(ctx, time.Millisecond, wait.ForeverTestTimeout, false, func(context.Context) (bool, error) { + return len(queue.InFlightPods()) == 0, nil + }); err != nil { + t.Fatalf("in-flight pod was not cleared: %v", queue.InFlightPods()) + } + if got := queue.PodsInBackoffQ(); len(got) != 0 { + t.Fatalf("expected recreated pod to stay out of backoffQ, got %v", got) + } + if got := queue.UnschedulablePods(); len(got) != 0 { + t.Fatalf("expected recreated pod to stay out of unschedulablePods, got %v", got) + } + if got := queue.NominatedPodsForNode("node1"); len(got) != 0 { + t.Fatalf("expected recreated pod to stay out of nominated pods, got %v", got) + } + + updatedPod, err := client.CoreV1().Pods(recreatedPod.Namespace).Get(ctx, recreatedPod.Name, metav1.GetOptions{}) + if err != nil { + t.Fatalf("Get pod: %v", err) + } + if diff := cmp.Diff(recreatedPod.Status, updatedPod.Status); diff != "" { + t.Fatalf("expected recreated pod status to remain unchanged (-want,+got):\n%s", diff) + } +} + type constSigPluginConfig struct { name string signature []fwk.SignFragment diff --git a/test/integration/scheduler/queueing/queue_test.go b/test/integration/scheduler/queueing/queue_test.go index fb13b104265..404ce86aca7 100644 --- a/test/integration/scheduler/queueing/queue_test.go +++ b/test/integration/scheduler/queueing/queue_test.go @@ -304,7 +304,6 @@ func TestCustomResourceEnqueue(t *testing.T) { defer testutils.CleanupTest(t, testCtx) cs, ns, ctx := testCtx.ClientSet, testCtx.NS.Name, testCtx.Ctx - logger := klog.FromContext(ctx) // Create one Node. node := st.MakeNode().Name("fake-node").Obj() if _, err := cs.CoreV1().Nodes().Create(ctx, node, metav1.CreateOptions{}); err != nil { @@ -340,13 +339,6 @@ func TestCustomResourceEnqueue(t *testing.T) { } testCtx.Scheduler.FailureHandler(ctx, schedFramework, podInfo, fwk.NewStatus(fwk.Unschedulable).WithError(fitError), nil, time.Now()) - // Scheduling cycle is incremented from 0 to 1 after NextPod() is called, so - // pass a number larger than 1 to move Pod to unschedulablePods. FailureHandler above may have - // already requeued the Pod, in which case the duplicate AddUnschedulableIfNotPresent call is benign. - if err := testCtx.Scheduler.SchedulingQueue.AddUnschedulableIfNotPresent(logger, podInfo, 10, ""); err != nil && err.Error() != fmt.Sprintf("Pod %v is already present in the active queue", klog.KObj(podInfo.Pod)) { - t.Fatalf("AddUnschedulableIfNotPresent failed: %v", err) - } - // Trigger a Custom Resource event. // We expect this event to trigger moving the test Pod from unschedulablePods to activeQ. crdGVR := schema.GroupVersionResource{Group: fooCRD.Spec.Group, Version: fooCRD.Spec.Versions[0].Name, Resource: "foos"} From 064f0956adbcf81958e95eb7fb6f78f3d88bcde3 Mon Sep 17 00:00:00 2001 From: Samarth Verma Date: Tue, 14 Apr 2026 10:14:36 -0400 Subject: [PATCH 3/3] scheduler: address recreated pod review feedback --- pkg/scheduler/backend/queue/active_queue.go | 5 ++--- pkg/scheduler/backend/queue/scheduling_queue.go | 2 +- .../backend/queue/scheduling_queue_test.go | 8 ++------ pkg/scheduler/schedule_one.go | 14 ++++---------- pkg/scheduler/schedule_one_test.go | 3 +++ test/integration/scheduler/queueing/queue_test.go | 5 +++++ 6 files changed, 17 insertions(+), 20 deletions(-) diff --git a/pkg/scheduler/backend/queue/active_queue.go b/pkg/scheduler/backend/queue/active_queue.go index 6728f2f0e3f..060ee8c3640 100644 --- a/pkg/scheduler/backend/queue/active_queue.go +++ b/pkg/scheduler/backend/queue/active_queue.go @@ -46,7 +46,6 @@ type activeQueuer interface { movePodToInFlight(pInfo *framework.QueuedPodInfo) error listInFlightEvents() []interface{} listInFlightPods() []*v1.Pod - // clusterEventsForPod collects cluster events recorded while pInfo.Pod was in flight. clusterEventsForPod(logger klog.Logger, pInfo *framework.QueuedPodInfo) ([]*clusterEvent, error) addEventsIfPodInFlight(oldPod, newPod *v1.Pod, events []fwk.ClusterEvent) bool addEventIfAnyInFlight(oldObj, newObj interface{}, event fwk.ClusterEvent) bool @@ -397,7 +396,7 @@ func (aq *activeQueue) listInFlightPods() []*v1.Pod { return pods } -// clusterEventsForPod gets all cluster events that have happened while pInfo.Pod was in flight. +// clusterEventsForPod gets all cluster events that have happened during pod for pInfo is being scheduled. func (aq *activeQueue) clusterEventsForPod(logger klog.Logger, pInfo *framework.QueuedPodInfo) ([]*clusterEvent, error) { aq.lock.RLock() defer aq.lock.RUnlock() @@ -405,7 +404,7 @@ func (aq *activeQueue) clusterEventsForPod(logger klog.Logger, pInfo *framework. // AddUnschedulableIfNotPresent is called with the Pod at the end of scheduling or binding. // So, given pInfo should have been Pop()ed before, - // we can assume pInfo.Pod.UID must be recorded in inFlightPods and thus inFlightEvents. + // we can assume pInfo must be recorded in inFlightPods and thus inFlightEvents. inFlightPod, ok := aq.inFlightPods[pInfo.Pod.UID] if !ok { return nil, fmt.Errorf("in flight Pod isn't found in the scheduling queue. If you see this error log, it's likely a bug in the scheduler") diff --git a/pkg/scheduler/backend/queue/scheduling_queue.go b/pkg/scheduler/backend/queue/scheduling_queue.go index ae324fa3184..81c5488cb44 100644 --- a/pkg/scheduler/backend/queue/scheduling_queue.go +++ b/pkg/scheduler/backend/queue/scheduling_queue.go @@ -906,7 +906,7 @@ func (p *PriorityQueue) AddUnschedulableIfNotPresent(logger klog.Logger, pInfo * p.lock.Lock() defer p.lock.Unlock() - // In any case, this scheduling attempt will be moved back to the queue and we should call Done. + // In any case, this Pod will be moved back to the queue and we should call Done. defer p.Done(pInfo.Pod.UID) pod := pInfo.Pod diff --git a/pkg/scheduler/backend/queue/scheduling_queue_test.go b/pkg/scheduler/backend/queue/scheduling_queue_test.go index 8070b3202d9..13da015ad27 100644 --- a/pkg/scheduler/backend/queue/scheduling_queue_test.go +++ b/pkg/scheduler/backend/queue/scheduling_queue_test.go @@ -4030,9 +4030,7 @@ func TestPerPodSchedulingMetrics(t *testing.T) { } pInfo.UnschedulablePlugins = sets.New("plugin") - if err := queue.AddUnschedulableIfNotPresent(logger, pInfo, 1); err != nil { - t.Fatalf("Unexpected error from AddUnschedulableIfNotPresent: %v", err) - } + queue.AddUnschedulableIfNotPresent(logger, pInfo, 1) // Override clock to exceed the DefaultPodMaxInUnschedulablePodsDuration so that unschedulable pods // will be moved to activeQ c.SetTime(timestamp.Add(DefaultPodMaxInUnschedulablePodsDuration + 1)) @@ -4052,9 +4050,7 @@ func TestPerPodSchedulingMetrics(t *testing.T) { } pInfo.UnschedulablePlugins = sets.New("plugin") - if err := queue.AddUnschedulableIfNotPresent(logger, pInfo, 1); err != nil { - t.Fatalf("Unexpected error from AddUnschedulableIfNotPresent: %v", err) - } + queue.AddUnschedulableIfNotPresent(logger, pInfo, 1) // Override clock to exceed the DefaultPodMaxInUnschedulablePodsDuration so that unschedulable pods // will be moved to activeQ updatedTimestamp := timestamp diff --git a/pkg/scheduler/schedule_one.go b/pkg/scheduler/schedule_one.go index cba9ddef565..164eb41f3dd 100644 --- a/pkg/scheduler/schedule_one.go +++ b/pkg/scheduler/schedule_one.go @@ -1256,21 +1256,15 @@ func (sched *Scheduler) handleSchedulingFailure(ctx context.Context, podFwk fram logger.Info("Pod has been assigned to node. Abort adding it back to queue.", "pod", klog.KObj(pod), "node", cachedPod.Spec.NodeName) // We need to call DonePod here because we don't call AddUnschedulableIfNotPresent in this case. } else { - // Remember the UID under which this pod is tracked as in-flight (from Pop) before we - // replace podInfo with the informer copy. If the pod was deleted and recreated with the - // same name, skip requeueing or updating the new pod; informer handlers will handle it (#138316). - poppedInFlightUID := podInfo.Pod.UID + if cachedPod.UID != podInfo.Pod.UID { + logger.V(2).Info("Pod was recreated while handling scheduling failure. Skip requeueing and status updates.", "pod", klog.KObj(pod), "oldUID", podInfo.Pod.UID, "newUID", cachedPod.UID) + return + } // As is from SharedInformer, we need to do a DeepCopy() here. // ignore this err since apiserver doesn't properly validate affinity terms // and we can't fix the validation for backwards compatibility. podInfo.PodInfo, _ = framework.NewPodInfo(cachedPod.DeepCopy()) pod = podInfo.Pod - if pod.UID != poppedInFlightUID { - logger.V(2).Info("Pod was recreated while handling scheduling failure. Skip requeueing and status updates.", "pod", klog.KObj(pod), "oldUID", poppedInFlightUID, "newUID", pod.UID) - sched.SchedulingQueue.Done(poppedInFlightUID) - calledDone = true - return - } if err := sched.SchedulingQueue.AddUnschedulableIfNotPresent(logger, podInfo, sched.SchedulingQueue.SchedulingCycle()); err != nil { utilruntime.HandleErrorWithContext(ctx, err, "Error occurred") } diff --git a/pkg/scheduler/schedule_one_test.go b/pkg/scheduler/schedule_one_test.go index e5ea5060a83..0279d730f61 100644 --- a/pkg/scheduler/schedule_one_test.go +++ b/pkg/scheduler/schedule_one_test.go @@ -1360,6 +1360,9 @@ func TestHandleSchedulingFailureSkipsRecreatedPod(t *testing.T) { if err != nil { t.Fatalf("Pop: %v", err) } + if got := queue.InFlightPods(); !podListContainsPod(got, oldPod) { + t.Fatalf("expected popped pod to be in-flight before failure handling, got %v", got) + } nominatingInfo := &fwk.NominatingInfo{NominatingMode: fwk.ModeOverride, NominatedNodeName: "node1"} sched.handleSchedulingFailure(ctx, schedFramework, popped, fwk.NewStatus(fwk.Unschedulable, "no fit"), nominatingInfo, time.Now()) diff --git a/test/integration/scheduler/queueing/queue_test.go b/test/integration/scheduler/queueing/queue_test.go index 404ce86aca7..b11286c502b 100644 --- a/test/integration/scheduler/queueing/queue_test.go +++ b/test/integration/scheduler/queueing/queue_test.go @@ -304,6 +304,7 @@ func TestCustomResourceEnqueue(t *testing.T) { defer testutils.CleanupTest(t, testCtx) cs, ns, ctx := testCtx.ClientSet, testCtx.NS.Name, testCtx.Ctx + logger := klog.FromContext(ctx) // Create one Node. node := st.MakeNode().Name("fake-node").Obj() if _, err := cs.CoreV1().Nodes().Create(ctx, node, metav1.CreateOptions{}); err != nil { @@ -339,6 +340,10 @@ func TestCustomResourceEnqueue(t *testing.T) { } testCtx.Scheduler.FailureHandler(ctx, schedFramework, podInfo, fwk.NewStatus(fwk.Unschedulable).WithError(fitError), nil, time.Now()) + // Scheduling cycle is incremented from 0 to 1 after NextPod() is called, so + // pass a number larger than 1 to move Pod to unschedulablePods. + testCtx.Scheduler.SchedulingQueue.AddUnschedulableIfNotPresent(logger, podInfo, 10) + // Trigger a Custom Resource event. // We expect this event to trigger moving the test Pod from unschedulablePods to activeQ. crdGVR := schema.GroupVersionResource{Group: fooCRD.Spec.Group, Version: fooCRD.Spec.Versions[0].Name, Resource: "foos"}