Merge pull request #127180 from sanposhiho/general-gate

feat: introduce pInfo.GatingPlugin to filter out events more generally

pkg/scheduler/backend/queue/active_queue.go

@@ -306,6 +306,8 @@ func (aq *activeQueue) unlockedPop(logger klog.Logger) (*framework.QueuedPodInfo
 	}
 	pInfo.UnschedulablePlugins.Clear()
 	pInfo.PendingPlugins.Clear()
+	pInfo.GatingPlugin = ""
+	pInfo.GatingPluginEvents = nil
 
 	return pInfo, nil
 }

pkg/scheduler/backend/queue/scheduling_queue.go

@@ -48,7 +48,6 @@ import (
 	"k8s.io/kubernetes/pkg/scheduler/backend/heap"
 	"k8s.io/kubernetes/pkg/scheduler/framework"
 	"k8s.io/kubernetes/pkg/scheduler/framework/plugins/interpodaffinity"
-	"k8s.io/kubernetes/pkg/scheduler/framework/plugins/names"
 	"k8s.io/kubernetes/pkg/scheduler/framework/plugins/podtopologyspread"
 	"k8s.io/kubernetes/pkg/scheduler/metrics"
 	"k8s.io/kubernetes/pkg/scheduler/util"
@@ -179,10 +178,12 @@ type PriorityQueue struct {
 	// TODO: this will be removed after SchedulingQueueHint goes to stable and the feature gate is removed.
 	moveRequestCycle int64
 
-	// preEnqueuePluginMap is keyed with profile name, valued with registered preEnqueue plugins.
-	preEnqueuePluginMap map[string][]framework.PreEnqueuePlugin
+	// preEnqueuePluginMap is keyed with profile and plugin name, valued with registered preEnqueue plugins.
+	preEnqueuePluginMap map[string]map[string]framework.PreEnqueuePlugin
 	// queueingHintMap is keyed with profile name, valued with registered queueing hint functions.
 	queueingHintMap QueueingHintMapPerProfile
+	// pluginToEventsMap shows which plugin is interested in which events.
+	pluginToEventsMap map[string][]framework.ClusterEvent
 
 	nsLister listersv1.NamespaceLister
 
@@ -219,7 +220,7 @@ type priorityQueueOptions struct {
 	podLister                         listersv1.PodLister
 	metricsRecorder                   metrics.MetricAsyncRecorder
 	pluginMetricsSamplePercent        int
-	preEnqueuePluginMap               map[string][]framework.PreEnqueuePlugin
+	preEnqueuePluginMap               map[string]map[string]framework.PreEnqueuePlugin
 	queueingHintMap                   QueueingHintMapPerProfile
 }
 
@@ -275,7 +276,7 @@ func WithQueueingHintMapPerProfile(m QueueingHintMapPerProfile) Option {
 }
 
 // WithPreEnqueuePluginMap sets preEnqueuePluginMap for PriorityQueue.
-func WithPreEnqueuePluginMap(m map[string][]framework.PreEnqueuePlugin) Option {
+func WithPreEnqueuePluginMap(m map[string]map[string]framework.PreEnqueuePlugin) Option {
 	return func(o *priorityQueueOptions) {
 		o.preEnqueuePluginMap = m
 	}
@@ -341,6 +342,7 @@ func NewPriorityQueue(
 		unschedulablePods:                 newUnschedulablePods(metrics.NewUnschedulablePodsRecorder(), metrics.NewGatedPodsRecorder()),
 		preEnqueuePluginMap:               options.preEnqueuePluginMap,
 		queueingHintMap:                   options.queueingHintMap,
+		pluginToEventsMap:                 buildEventMap(options.queueingHintMap),
 		metricsRecorder:                   options.metricsRecorder,
 		pluginMetricsSamplePercent:        options.pluginMetricsSamplePercent,
 		moveRequestCycle:                  -1,
@@ -358,6 +360,20 @@ func NewPriorityQueue(
 	return pq
 }
 
+func buildEventMap(qHintMap QueueingHintMapPerProfile) map[string][]framework.ClusterEvent {
+	eventMap := make(map[string][]framework.ClusterEvent)
+
+	for _, hintMap := range qHintMap {
+		for event, qHints := range hintMap {
+			for _, qHint := range qHints {
+				eventMap[qHint.PluginName] = append(eventMap[qHint.PluginName], event)
+			}
+		}
+	}
+
+	return eventMap
+}
+
 // Run starts the goroutine to pump from backoffQ to activeQ
 func (p *PriorityQueue) Run(logger klog.Logger) {
 	go p.backoffQ.waitUntilAlignedWithOrderingWindow(func() {
@@ -515,13 +531,11 @@ func queueingHintToLabel(hint framework.QueueingHint, err error) string {
 	return ""
 }
 
-// runPreEnqueuePlugins iterates PreEnqueue function in each registered PreEnqueuePlugin.
-// It returns true if all PreEnqueue function run successfully; otherwise returns false
-// upon the first failure.
+// runPreEnqueuePlugins iterates PreEnqueue function in each registered PreEnqueuePlugin,
+// and updates pInfo.GatingPlugin and pInfo.UnschedulablePlugins.
 // Note: we need to associate the failed plugin to `pInfo`, so that the pod can be moved back
 // to activeQ by related cluster event.
-func (p *PriorityQueue) runPreEnqueuePlugins(ctx context.Context, pInfo *framework.QueuedPodInfo) bool {
-	logger := klog.FromContext(ctx)
+func (p *PriorityQueue) runPreEnqueuePlugins(ctx context.Context, pInfo *framework.QueuedPodInfo) {
 	var s *framework.Status
 	pod := pInfo.Pod
 	startTime := p.clock.Now()
@@ -530,30 +544,54 @@ func (p *PriorityQueue) runPreEnqueuePlugins(ctx context.Context, pInfo *framewo
 	}()
 
 	shouldRecordMetric := rand.Intn(100) < p.pluginMetricsSamplePercent
+	logger := klog.FromContext(ctx)
+	gatingPlugin := pInfo.GatingPlugin
+	if gatingPlugin != "" {
+		// Run the gating plugin first
+		s := p.runPreEnqueuePlugin(ctx, logger, p.preEnqueuePluginMap[pod.Spec.SchedulerName][gatingPlugin], pInfo, shouldRecordMetric)
+		if !s.IsSuccess() {
+			// No need to iterate other plugins
+			return
+		}
+	}
+
 	for _, pl := range p.preEnqueuePluginMap[pod.Spec.SchedulerName] {
-		s = p.runPreEnqueuePlugin(ctx, pl, pod, shouldRecordMetric)
-		if s.IsSuccess() {
+		if gatingPlugin != "" && pl.Name() == gatingPlugin {
+			// should be run already above.
 			continue
 		}
-		pInfo.UnschedulablePlugins.Insert(pl.Name())
-		metrics.UnschedulableReason(pl.Name(), pod.Spec.SchedulerName).Inc()
-		if s.Code() == framework.Error {
-			logger.Error(s.AsError(), "Unexpected error running PreEnqueue plugin", "pod", klog.KObj(pod), "plugin", pl.Name())
-		} else {
-			logger.V(4).Info("Status after running PreEnqueue plugin", "pod", klog.KObj(pod), "plugin", pl.Name(), "status", s)
+		s := p.runPreEnqueuePlugin(ctx, logger, pl, pInfo, shouldRecordMetric)
+		if !s.IsSuccess() {
+			// No need to iterate other plugins
+			return
 		}
-		return false
 	}
-	return true
+	// all plugins passed
+	pInfo.GatingPlugin = ""
 }
 
-func (p *PriorityQueue) runPreEnqueuePlugin(ctx context.Context, pl framework.PreEnqueuePlugin, pod *v1.Pod, shouldRecordMetric bool) *framework.Status {
-	if !shouldRecordMetric {
-		return pl.PreEnqueue(ctx, pod)
-	}
+// runPreEnqueuePlugin runs the PreEnqueue plugin and update pInfo's fields accordingly if needed.
+func (p *PriorityQueue) runPreEnqueuePlugin(ctx context.Context, logger klog.Logger, pl framework.PreEnqueuePlugin, pInfo *framework.QueuedPodInfo, shouldRecordMetric bool) *framework.Status {
+	pod := pInfo.Pod
 	startTime := p.clock.Now()
 	s := pl.PreEnqueue(ctx, pod)
-	p.metricsRecorder.ObservePluginDurationAsync(preEnqueue, pl.Name(), s.Code().String(), p.clock.Since(startTime).Seconds())
+	if shouldRecordMetric {
+		p.metricsRecorder.ObservePluginDurationAsync(preEnqueue, pl.Name(), s.Code().String(), p.clock.Since(startTime).Seconds())
+	}
+	if s.IsSuccess() {
+		// No need to change GatingPlugin; it's overwritten by the next PreEnqueue plugin if they gate this pod, or it's overwritten with an empty string if all PreEnqueue plugins pass.
+		return s
+	}
+	pInfo.UnschedulablePlugins.Insert(pl.Name())
+	metrics.UnschedulableReason(pl.Name(), pod.Spec.SchedulerName).Inc()
+	pInfo.GatingPlugin = pl.Name()
+	pInfo.GatingPluginEvents = p.pluginToEventsMap[pInfo.GatingPlugin]
+	if s.Code() == framework.Error {
+		logger.Error(s.AsError(), "Unexpected error running PreEnqueue plugin", "pod", klog.KObj(pod), "plugin", pl.Name())
+	} else {
+		logger.V(4).Info("Status after running PreEnqueue plugin", "pod", klog.KObj(pod), "plugin", pl.Name(), "status", s)
+	}
+
 	return s
 }
 
@@ -561,17 +599,17 @@ func (p *PriorityQueue) runPreEnqueuePlugin(ctx context.Context, pl framework.Pr
 // If the pod doesn't pass PreEnqueue plugins, it gets added to unschedulablePods instead.
 // It returns a boolean flag to indicate whether the pod is added successfully.
 func (p *PriorityQueue) moveToActiveQ(logger klog.Logger, pInfo *framework.QueuedPodInfo, event string) bool {
-	gatedBefore := pInfo.Gated
+	gatedBefore := pInfo.Gated()
 	// If SchedulerPopFromBackoffQ feature gate is enabled,
 	// PreEnqueue plugins were called when the pod was added to the backoffQ.
 	// Don't need to repeat it here when the pod is directly moved from the backoffQ.
 	if !p.isPopFromBackoffQEnabled || event != framework.BackoffComplete {
-		pInfo.Gated = !p.runPreEnqueuePlugins(context.Background(), pInfo)
+		p.runPreEnqueuePlugins(context.Background(), pInfo)
 	}
 
 	added := false
 	p.activeQ.underLock(func(unlockedActiveQ unlockedActiveQueuer) {
-		if pInfo.Gated {
+		if pInfo.Gated() {
 			// Add the Pod to unschedulablePods if it's not passing PreEnqueuePlugins.
 			if unlockedActiveQ.has(pInfo) {
 				return
@@ -612,8 +650,8 @@ func (p *PriorityQueue) moveToBackoffQ(logger klog.Logger, pInfo *framework.Queu
 	// PreEnqueue plugins are called on inserting pods to the backoffQ,
 	// not to call them again on popping out.
 	if p.isPopFromBackoffQEnabled {
-		pInfo.Gated = !p.runPreEnqueuePlugins(context.Background(), pInfo)
-		if pInfo.Gated {
+		p.runPreEnqueuePlugins(context.Background(), pInfo)
+		if pInfo.Gated() {
 			if p.unschedulablePods.get(pInfo.Pod) == nil {
 				p.unschedulablePods.addOrUpdate(pInfo, event)
 				logger.V(5).Info("Pod moved to an internal scheduling queue", "pod", klog.KObj(pInfo.Pod), "event", event, "queue", unschedulablePods)
@@ -958,7 +996,7 @@ func (p *PriorityQueue) Update(logger klog.Logger, oldPod, newPod *v1.Pod) {
 	if pInfo := p.unschedulablePods.get(newPod); pInfo != nil {
 		_ = pInfo.Update(newPod)
 		p.UpdateNominatedPod(logger, oldPod, pInfo.PodInfo)
-		gated := pInfo.Gated
+		gated := pInfo.Gated()
 		if p.isSchedulingQueueHintEnabled {
 			// When unscheduled Pods are updated, we check with QueueingHint
 			// whether the update may make the pods schedulable.
@@ -1022,7 +1060,7 @@ func (p *PriorityQueue) Delete(pod *v1.Pod) {
 		return
 	}
 	if pInfo = p.unschedulablePods.get(pod); pInfo != nil {
-		p.unschedulablePods.delete(pod, pInfo.Gated)
+		p.unschedulablePods.delete(pod, pInfo.Gated())
 	}
 }
 
@@ -1120,22 +1158,9 @@ func (p *PriorityQueue) movePodsToActiveOrBackoffQueue(logger klog.Logger, podIn
 
 	activated := false
 	for _, pInfo := range podInfoList {
-		// When handling events takes time, a scheduling throughput gets impacted negatively
-		// because of a shared lock within PriorityQueue, which Pop() also requires.
-		//
-		// Scheduling-gated Pods never get schedulable with any events,
-		// except the Pods themselves got updated, which isn't handled by movePodsToActiveOrBackoffQueue.
-		// So, we can skip them early here so that they don't go through isPodWorthRequeuing,
-		// which isn't fast enough to keep a sufficient scheduling throughput
-		// when the number of scheduling-gated Pods in unschedulablePods is large.
-		// https://github.com/kubernetes/kubernetes/issues/124384
-		// This is a hotfix for this issue, which might be changed
-		// once we have a better general solution for the shared lock issue.
-		//
-		// Note that we cannot skip all pInfo.Gated Pods here
-		// because PreEnqueue plugins apart from the scheduling gate plugin may change the gating status
-		// with these events.
-		if pInfo.Gated && pInfo.UnschedulablePlugins.Has(names.SchedulingGates) {
+		if pInfo.Gated() && !event.MatchAny(pInfo.GatingPluginEvents) {
+			// This event doesn't interest the gating plugin of this Pod,
+			// which means this event never moves this Pod to activeQ.
 			continue
 		}
 
@@ -1146,7 +1171,7 @@ func (p *PriorityQueue) movePodsToActiveOrBackoffQueue(logger klog.Logger, podIn
 			continue
 		}
 
-		p.unschedulablePods.delete(pInfo.Pod, pInfo.Gated)
+		p.unschedulablePods.delete(pInfo.Pod, pInfo.Gated())
 		queue := p.requeuePodViaQueueingHint(logger, pInfo, schedulingHint, event.Label())
 		logger.V(4).Info("Pod moved to an internal scheduling queue", "pod", klog.KObj(pInfo.Pod), "event", event.Label(), "queue", queue, "hint", schedulingHint)
 		if queue == activeQ || (p.isPopFromBackoffQEnabled && queue == backoffQ) {
@@ -1341,9 +1366,9 @@ type UnschedulablePods struct {
 func (u *UnschedulablePods) addOrUpdate(pInfo *framework.QueuedPodInfo, event string) {
 	podID := u.keyFunc(pInfo.Pod)
 	if _, exists := u.podInfoMap[podID]; !exists {
-		if pInfo.Gated && u.gatedRecorder != nil {
+		if pInfo.Gated() && u.gatedRecorder != nil {
 			u.gatedRecorder.Inc()
-		} else if !pInfo.Gated && u.unschedulableRecorder != nil {
+		} else if !pInfo.Gated() && u.unschedulableRecorder != nil {
 			u.unschedulableRecorder.Inc()
 		}
 		metrics.SchedulerQueueIncomingPods.WithLabelValues("unschedulable", event).Inc()

pkg/scheduler/backend/queue/scheduling_queue_test.go

@@ -107,8 +107,11 @@ func init() {
 	metrics.Register()
 }
 
-func setQueuedPodInfoGated(queuedPodInfo *framework.QueuedPodInfo) *framework.QueuedPodInfo {
-	queuedPodInfo.Gated = true
+func setQueuedPodInfoGated(queuedPodInfo *framework.QueuedPodInfo, gatingPlugin string, gatingPluginEvents []framework.ClusterEvent) *framework.QueuedPodInfo {
+	queuedPodInfo.GatingPlugin = gatingPlugin
+	// GatingPlugin should also be registered in UnschedulablePlugins.
+	queuedPodInfo.UnschedulablePlugins = sets.New(gatingPlugin)
+	queuedPodInfo.GatingPluginEvents = gatingPluginEvents
 	return queuedPodInfo
 }
 
@@ -1591,7 +1594,10 @@ func TestPriorityQueue_moveToActiveQ(t *testing.T) {
 				ctx, cancel := context.WithCancel(ctx)
 				defer cancel()
 
-				m := map[string][]framework.PreEnqueuePlugin{"": tt.plugins}
+				m := map[string]map[string]framework.PreEnqueuePlugin{"": make(map[string]framework.PreEnqueuePlugin, len(tt.plugins))}
+				for _, plugin := range tt.plugins {
+					m[""][plugin.Name()] = plugin
+				}
 				q := NewTestQueueWithObjects(ctx, newDefaultQueueSort(), []runtime.Object{tt.pod}, WithPreEnqueuePluginMap(m),
 					WithPodInitialBackoffDuration(time.Second*30), WithPodMaxBackoffDuration(time.Second*60))
 				got := q.moveToActiveQ(logger, q.newQueuedPodInfo(tt.pod), tt.event)
@@ -1684,7 +1690,10 @@ func TestPriorityQueue_moveToBackoffQ(t *testing.T) {
 				ctx, cancel := context.WithCancel(ctx)
 				defer cancel()
 
-				m := map[string][]framework.PreEnqueuePlugin{"": tt.plugins}
+				m := map[string]map[string]framework.PreEnqueuePlugin{"": make(map[string]framework.PreEnqueuePlugin, len(tt.plugins))}
+				for _, plugin := range tt.plugins {
+					m[""][plugin.Name()] = plugin
+				}
 				q := NewTestQueueWithObjects(ctx, newDefaultQueueSort(), []runtime.Object{tt.pod}, WithPreEnqueuePluginMap(m),
 					WithPodInitialBackoffDuration(time.Second*30), WithPodMaxBackoffDuration(time.Second*60))
 				pInfo := q.newQueuedPodInfo(tt.pod)
@@ -1836,7 +1845,7 @@ func BenchmarkMoveAllToActiveOrBackoffQueue(b *testing.B) {
 
 func TestPriorityQueue_MoveAllToActiveOrBackoffQueueWithQueueingHint(t *testing.T) {
 	now := time.Now()
-	p := st.MakePod().Name("pod1").Namespace("ns1").UID("1").Obj()
+	p := st.MakePod().Name("pod1").Namespace("ns1").UID("1").Label("foo", "bar").Obj()
 	tests := []struct {
 		name    string
 		podInfo *framework.QueuedPodInfo
@@ -1872,22 +1881,27 @@ func TestPriorityQueue_MoveAllToActiveOrBackoffQueueWithQueueingHint(t *testing.
 			expectedQ: unschedulablePods,
 		},
 		{
-			name:    "QueueHintFunction is not called when Pod is gated by SchedulingGates plugin",
-			podInfo: setQueuedPodInfoGated(&framework.QueuedPodInfo{PodInfo: mustNewPodInfo(p), UnschedulablePlugins: sets.New(names.SchedulingGates, "foo")}),
+			name:    "QueueHintFunction is not called when Pod is gated by the plugin that isn't interested in the event",
+			podInfo: setQueuedPodInfoGated(&framework.QueuedPodInfo{PodInfo: mustNewPodInfo(p)}, names.SchedulingGates, []framework.ClusterEvent{framework.EventUnscheduledPodUpdate}),
+			// The hintFn should not be called as the pod is gated by SchedulingGates plugin,
+			// the scheduling gate isn't interested in the node add event,
+			// and the queue should keep this Pod in the unschedQ without calling the hintFn.
 			hint: func(logger klog.Logger, pod *v1.Pod, oldObj, newObj interface{}) (framework.QueueingHint, error) {
 				return framework.Queue, fmt.Errorf("QueueingHintFn should not be called as pod is gated")
 			},
 			expectedQ: unschedulablePods,
 		},
 		{
-			name:      "QueueHintFunction is called when Pod is gated by a plugin other than SchedulingGates",
-			podInfo:   setQueuedPodInfoGated(&framework.QueuedPodInfo{PodInfo: mustNewPodInfo(p), UnschedulablePlugins: sets.New("foo")}),
-			hint:      queueHintReturnQueue,
+			name:    "QueueHintFunction is called when Pod is gated by the plugin that is interested in the event",
+			podInfo: setQueuedPodInfoGated(&framework.QueuedPodInfo{PodInfo: mustNewPodInfo(p)}, "foo", []framework.ClusterEvent{nodeAdd}),
+			// In this case, the hintFn should be called as the pod is gated by foo plugin that is interested in the NodeAdd event.
+			hint: queueHintReturnQueue,
+			// and, as a result, this pod should be queued to activeQ.
 			expectedQ: activeQ,
 		},
 		{
 			name:      "Pod that experienced a scheduling failure before should be queued to backoffQ after un-gated",
-			podInfo:   setQueuedPodInfoGated(&framework.QueuedPodInfo{PodInfo: mustNewPodInfo(p), Attempts: 1, UnschedulablePlugins: sets.New("foo")}),
+			podInfo:   setQueuedPodInfoGated(&framework.QueuedPodInfo{PodInfo: mustNewPodInfo(p), Attempts: 1}, "foo", []framework.ClusterEvent{nodeAdd}),
 			hint:      queueHintReturnQueue,
 			expectedQ: backoffQ,
 		},
@@ -1903,8 +1917,19 @@ func TestPriorityQueue_MoveAllToActiveOrBackoffQueueWithQueueingHint(t *testing.
 					QueueingHintFn: test.hint,
 				},
 			}
+			m[""][framework.EventUnscheduledPodUpdate] = []*QueueingHintFunction{
+				{
+					PluginName:     names.SchedulingGates,
+					QueueingHintFn: queueHintReturnQueue,
+				},
+			}
 			cl := testingclock.NewFakeClock(now)
-			q := NewTestQueue(ctx, newDefaultQueueSort(), WithQueueingHintMapPerProfile(m), WithClock(cl))
+			plugin, _ := schedulinggates.New(ctx, nil, nil, plfeature.Features{})
+			preEnqM := map[string]map[string]framework.PreEnqueuePlugin{"": {
+				names.SchedulingGates: plugin.(framework.PreEnqueuePlugin),
+				"foo":                 &preEnqueuePlugin{allowlists: []string{"foo"}},
+			}}
+			q := NewTestQueue(ctx, newDefaultQueueSort(), WithQueueingHintMapPerProfile(m), WithClock(cl), WithPreEnqueuePluginMap(preEnqM))
 			q.Add(logger, test.podInfo.Pod)
 			if p, err := q.Pop(logger); err != nil || p.Pod != test.podInfo.Pod {
 				t.Errorf("Expected: %v after Pop, but got: %v", test.podInfo.Pod.Name, p.Pod.Name)
@@ -3165,7 +3190,7 @@ var (
 		})
 	}
 	addPodUnschedulablePods = func(t *testing.T, logger klog.Logger, queue *PriorityQueue, pInfo *framework.QueuedPodInfo) {
-		if !pInfo.Gated {
+		if !pInfo.Gated() {
 			// Update pod condition to unschedulable.
 			podutil.UpdatePodCondition(&pInfo.Pod.Status, &v1.PodCondition{
 				Type:    v1.PodScheduled,
@@ -3297,6 +3322,8 @@ func TestPodTimestamp(t *testing.T) {
 // TestPendingPodsMetric tests Prometheus metrics related with pending pods
 func TestPendingPodsMetric(t *testing.T) {
 	timestamp := time.Now()
+	preenqueuePluginName := "preEnqueuePlugin"
+	metrics.Register()
 	total := 60
 	queueableNum := 50
 	queueable, failme := "queueable", "failme"
@@ -3306,7 +3333,7 @@ func TestPendingPodsMetric(t *testing.T) {
 	gated := makeQueuedPodInfos(total-queueableNum, "y", failme, timestamp)
 	// Manually mark them as gated=true.
 	for _, pInfo := range gated {
-		setQueuedPodInfoGated(pInfo)
+		setQueuedPodInfoGated(pInfo, preenqueuePluginName, []framework.ClusterEvent{framework.EventUnscheduledPodUpdate})
 	}
 	pInfos = append(pInfos, gated...)
 	totalWithDelay := 20
@@ -3561,9 +3588,16 @@ scheduler_plugin_execution_duration_seconds_count{extension_point="PreEnqueue",p
 			resetMetrics()
 			resetPodInfos()
 
-			m := map[string][]framework.PreEnqueuePlugin{"": {&preEnqueuePlugin{allowlists: []string{queueable}}}}
+			m := makeEmptyQueueingHintMapPerProfile()
+			m[""][framework.EventUnscheduledPodUpdate] = []*QueueingHintFunction{
+				{
+					PluginName:     preenqueuePluginName,
+					QueueingHintFn: queueHintReturnQueue,
+				},
+			}
+			preenq := map[string]map[string]framework.PreEnqueuePlugin{"": {(&preEnqueuePlugin{}).Name(): &preEnqueuePlugin{allowlists: []string{queueable}}}}
 			recorder := metrics.NewMetricsAsyncRecorder(3, 20*time.Microsecond, ctx.Done())
-			queue := NewTestQueue(ctx, newDefaultQueueSort(), WithClock(testingclock.NewFakeClock(timestamp)), WithPreEnqueuePluginMap(m), WithPluginMetricsSamplePercent(test.pluginMetricsSamplePercent), WithMetricsRecorder(*recorder))
+			queue := NewTestQueue(ctx, newDefaultQueueSort(), WithClock(testingclock.NewFakeClock(timestamp)), WithPreEnqueuePluginMap(preenq), WithPluginMetricsSamplePercent(test.pluginMetricsSamplePercent), WithMetricsRecorder(*recorder), WithQueueingHintMapPerProfile(m))
 			for i, op := range test.operations {
 				for _, pInfo := range test.operands[i] {
 					op(t, logger, queue, pInfo)
@@ -3652,7 +3686,7 @@ func TestPerPodSchedulingMetrics(t *testing.T) {
 			name: "A gated pod is created and scheduled after lifting gate",
 			perPodSchedulingMetricsScenario: func(c *testingclock.FakeClock, queue *PriorityQueue, pod *v1.Pod) {
 				// Create a queue with PreEnqueuePlugin
-				queue.preEnqueuePluginMap = map[string][]framework.PreEnqueuePlugin{"": {&preEnqueuePlugin{allowlists: []string{"foo"}}}}
+				queue.preEnqueuePluginMap = map[string]map[string]framework.PreEnqueuePlugin{"": {(&preEnqueuePlugin{}).Name(): &preEnqueuePlugin{allowlists: []string{"foo"}}}}
 				queue.pluginMetricsSamplePercent = 0
 				queue.Add(logger, pod)
 				// Check pod is added to the unschedulablePods queue.
@@ -4301,13 +4335,13 @@ func Test_isPodWorthRequeuing(t *testing.T) {
 func Test_queuedPodInfo_gatedSetUponCreationAndUnsetUponUpdate(t *testing.T) {
 	logger, ctx := ktesting.NewTestContext(t)
 	plugin, _ := schedulinggates.New(ctx, nil, nil, plfeature.Features{})
-	m := map[string][]framework.PreEnqueuePlugin{"": {plugin.(framework.PreEnqueuePlugin)}}
+	m := map[string]map[string]framework.PreEnqueuePlugin{"": {names.SchedulingGates: plugin.(framework.PreEnqueuePlugin)}}
 	q := NewTestQueue(ctx, newDefaultQueueSort(), WithPreEnqueuePluginMap(m))
 
 	gatedPod := st.MakePod().SchedulingGates([]string{"hello world"}).Obj()
 	q.Add(logger, gatedPod)
 
-	if !q.unschedulablePods.get(gatedPod).Gated {
+	if !q.unschedulablePods.get(gatedPod).Gated() {
 		t.Error("Expected pod to be gated")
 	}
 
@@ -4316,7 +4350,7 @@ func Test_queuedPodInfo_gatedSetUponCreationAndUnsetUponUpdate(t *testing.T) {
 	q.Update(logger, gatedPod, ungatedPod)
 
 	ungatedPodInfo, _ := q.Pop(logger)
-	if ungatedPodInfo.Gated {
+	if ungatedPodInfo.Gated() {
 		t.Error("Expected pod to be ungated")
 	}
 }

pkg/scheduler/framework/interface.go

@@ -476,9 +476,14 @@ type QueueSortPlugin interface {
 // This is because such temporal errors cannot be resolved by specific cluster events,
 // and we have no choose but keep retrying scheduling until the failure is resolved.
 //
-// Plugins that make pod unschedulable (PreEnqueue, PreFilter, Filter, Reserve, and Permit plugins) should implement this interface,
+// Plugins that make pod unschedulable (PreEnqueue, PreFilter, Filter, Reserve, and Permit plugins) must implement this interface,
 // otherwise the default implementation will be used, which is less efficient in requeueing Pods rejected by the plugin.
-// And, if plugins other than above extension points support this interface, they are just ignored.
+//
+// Also, if EventsToRegister returns an empty list, that means the Pods failed by the plugin are not requeued by any events,
+// which doesn't make sense in most cases (very likely misuse)
+// since the pods rejected by the plugin could be stuck in the unschedulable pod pool forever.
+//
+// If plugins other than above extension points support this interface, they are just ignored.
 type EnqueueExtensions interface {
 	Plugin
 	// EventsToRegister returns a series of possible events that may cause a Pod

pkg/scheduler/framework/plugins/defaultpreemption/default_preemption.go

@@ -116,11 +116,29 @@ func (pl *DefaultPreemption) PreEnqueue(ctx context.Context, p *v1.Pod) *framewo
 // EventsToRegister returns the possible events that may make a Pod
 // failed by this plugin schedulable.
 func (pl *DefaultPreemption) EventsToRegister(_ context.Context) ([]framework.ClusterEventWithHint, error) {
-	// The plugin moves the preemptor Pod to acviteQ/backoffQ once the preemption API calls are all done,
-	// and we don't need to move the Pod with any events.
+	if pl.fts.EnableAsyncPreemption {
+		return []framework.ClusterEventWithHint{
+			// We need to register the event to tell the scheduling queue that the pod could be un-gated after some Pods' deletion.
+			{Event: framework.ClusterEvent{Resource: framework.Pod, ActionType: framework.Delete}, QueueingHintFn: pl.isPodSchedulableAfterPodDeletion},
+		}, nil
+	}
+
+	// When the async preemption is disabled, PreEnqueue always returns nil, and hence pods never get rejected by this plugin.
 	return nil, nil
 }
 
+// isPodSchedulableAfterPodDeletion returns the queueing hint for the pod after the pod deletion event,
+// which always return Skip.
+// The default preemption plugin is a bit tricky;
+// the pods rejected by it are the ones that have run/are running the preemption asynchronously.
+// And, those pods should always have the other plugins in pInfo.UnschedulablePlugins
+// which failure will be resolved by the preemption.
+// The reason why we return Skip here is that the preemption plugin should not make the decision of when to requeueing Pods,
+// and rather, those plugins should be responsible for that.
+func (pl *DefaultPreemption) isPodSchedulableAfterPodDeletion(logger klog.Logger, pod *v1.Pod, oldObj, newObj interface{}) (framework.QueueingHint, error) {
+	return framework.QueueSkip, nil
+}
+
 // calculateNumCandidates returns the number of candidates the FindCandidates
 // method must produce from dry running based on the constraints given by
 // <minCandidateNodesPercentage> and <minCandidateNodesAbsolute>. The number of

pkg/scheduler/framework/types.go

@@ -19,6 +19,7 @@ package framework
 import (
 	"errors"
 	"fmt"
+	"slices"
 	"sort"
 	"strings"
 	"sync/atomic"
@@ -341,6 +342,15 @@ func (r EventResource) match(resource EventResource) bool {
 		r == Pod && (resource == assignedPod || resource == unschedulablePod)
 }
 
+func (ce ClusterEvent) MatchAny(events []ClusterEvent) bool {
+	for _, e := range events {
+		if e.Match(ce) {
+			return true
+		}
+	}
+	return false
+}
+
 func UnrollWildCardResource() []ClusterEventWithHint {
 	return []ClusterEventWithHint{
 		{Event: ClusterEvent{Resource: Pod, ActionType: All}},
@@ -385,8 +395,16 @@ type QueuedPodInfo struct {
 	UnschedulablePlugins sets.Set[string]
 	// PendingPlugins records the plugin names that the Pod failed with Pending status.
 	PendingPlugins sets.Set[string]
-	// Whether the Pod is scheduling gated (by PreEnqueuePlugins) or not.
-	Gated bool
+	// GatingPlugin records the plugin name that gated the Pod at PreEnqueue.
+	GatingPlugin string
+	// GatingPluginEvents records the events registered by the plugin that gated the Pod at PreEnqueue.
+	// We have it as a cache purpose to avoid re-computing which event(s) might ungate the Pod.
+	GatingPluginEvents []ClusterEvent
+}
+
+// Gated returns true if the pod is gated by any plugin.
+func (pqi *QueuedPodInfo) Gated() bool {
+	return pqi.GatingPlugin != ""
 }
 
 // DeepCopy returns a deep copy of the QueuedPodInfo object.
@@ -397,8 +415,9 @@ func (pqi *QueuedPodInfo) DeepCopy() *QueuedPodInfo {
 		Attempts:                pqi.Attempts,
 		InitialAttemptTimestamp: pqi.InitialAttemptTimestamp,
 		UnschedulablePlugins:    pqi.UnschedulablePlugins.Clone(),
+		GatingPlugin:            pqi.GatingPlugin,
+		GatingPluginEvents:      slices.Clone(pqi.GatingPluginEvents),
 		PendingPlugins:          pqi.PendingPlugins.Clone(),
-		Gated:                   pqi.Gated,
 	}
 }
 

pkg/scheduler/scheduler.go

@@ -352,11 +352,16 @@ func New(ctx context.Context,
 		return nil, errors.New("at least one profile is required")
 	}
 
-	preEnqueuePluginMap := make(map[string][]framework.PreEnqueuePlugin)
+	preEnqueuePluginMap := make(map[string]map[string]framework.PreEnqueuePlugin)
 	queueingHintsPerProfile := make(internalqueue.QueueingHintMapPerProfile)
 	var returnErr error
 	for profileName, profile := range profiles {
-		preEnqueuePluginMap[profileName] = profile.PreEnqueuePlugins()
+		plugins := profile.PreEnqueuePlugins()
+		preEnqueuePluginMap[profileName] = make(map[string]framework.PreEnqueuePlugin, len(plugins))
+		for _, plugin := range plugins {
+			preEnqueuePluginMap[profileName][plugin.Name()] = plugin
+		}
+
 		queueingHintsPerProfile[profileName], err = buildQueueingHintMap(ctx, profile.EnqueueExtensions())
 		if err != nil {
 			returnErr = errors.Join(returnErr, err)

test/integration/scheduler/plugins/plugins_test.go

@@ -41,6 +41,7 @@ import (
 	listersv1 "k8s.io/client-go/listers/core/v1"
 	featuregatetesting "k8s.io/component-base/featuregate/testing"
 	corev1helpers "k8s.io/component-helpers/scheduling/corev1"
+	"k8s.io/klog/v2"
 	configv1 "k8s.io/kube-scheduler/config/v1"
 	"k8s.io/kubernetes/pkg/features"
 	"k8s.io/kubernetes/pkg/scheduler"
@@ -2753,6 +2754,7 @@ func (pl *SchedulingGatesPluginWithEvents) Name() string {
 
 func (pl *SchedulingGatesPluginWithEvents) PreEnqueue(ctx context.Context, p *v1.Pod) *framework.Status {
 	pl.called++
+	klog.FromContext(ctx).Info("PreEnqueue is called", "pod", klog.KObj(p), "count", pl.called)
 	return pl.SchedulingGates.PreEnqueue(ctx, p)
 }
 
@@ -2773,6 +2775,7 @@ func (pl *SchedulingGatesPluginWOEvents) Name() string {
 
 func (pl *SchedulingGatesPluginWOEvents) PreEnqueue(ctx context.Context, p *v1.Pod) *framework.Status {
 	pl.called++
+	klog.FromContext(ctx).Info("PreEnqueue is called", "pod", klog.KObj(p), "count", pl.called)
 	return pl.SchedulingGates.PreEnqueue(ctx, p)
 }
 
@@ -2782,8 +2785,6 @@ func (pl *SchedulingGatesPluginWOEvents) EventsToRegister(_ context.Context) ([]
 
 // This test helps to verify registering nil events for PreEnqueue plugin works as expected.
 func TestPreEnqueuePluginEventsToRegister(t *testing.T) {
-	testContext := testutils.InitTestAPIServer(t, "preenqueue-plugin", nil)
-
 	num := func(pl framework.Plugin) int {
 		switch item := pl.(type) {
 		case *SchedulingGatesPluginWithEvents:
@@ -2830,6 +2831,7 @@ func TestPreEnqueuePluginEventsToRegister(t *testing.T) {
 			t.Run(tt.name+fmt.Sprintf(" queueHint(%v)", queueHintEnabled), func(t *testing.T) {
 				featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.SchedulerQueueingHints, queueHintEnabled)
 
+				testContext := testutils.InitTestAPIServer(t, "preenqueue-plugin", nil)
 				// use new plugin every time to clear counts
 				var plugin framework.PreEnqueuePlugin
 				if tt.withEvents {
@@ -2865,6 +2867,8 @@ func TestPreEnqueuePluginEventsToRegister(t *testing.T) {
 				)
 				defer teardown()
 
+				t.Log("Create the gated pod")
+
 				// Create a pod with schedulingGates.
 				gatedPod := st.MakePod().Name("p").Namespace(testContext.NS.Name).
 					SchedulingGates([]string{"foo"}).
@@ -2885,6 +2889,8 @@ func TestPreEnqueuePluginEventsToRegister(t *testing.T) {
 					return
 				}
 
+				t.Log("Create the pause pod")
+
 				// Create a best effort pod.
 				pausePod, err := testutils.CreatePausePod(testCtx.ClientSet, testutils.InitPausePod(&testutils.PausePodConfig{
 					Name:      "pause-pod",
@@ -2902,6 +2908,8 @@ func TestPreEnqueuePluginEventsToRegister(t *testing.T) {
 					return
 				}
 
+				t.Log("Update the pause pod")
+
 				// Update the pod which will trigger the requeue logic if plugin registers the events.
 				pausePod, err = testCtx.ClientSet.CoreV1().Pods(pausePod.Namespace).Get(testCtx.Ctx, pausePod.Name, metav1.GetOptions{})
 				if err != nil {
@@ -2925,6 +2933,8 @@ func TestPreEnqueuePluginEventsToRegister(t *testing.T) {
 					return
 				}
 
+				t.Log("Remove the scheduling gate")
+
 				// Remove gated pod's scheduling gates.
 				gatedPod, err = testCtx.ClientSet.CoreV1().Pods(gatedPod.Namespace).Get(testCtx.Ctx, gatedPod.Name, metav1.GetOptions{})
 				if err != nil {