From 47d296d62d74e6bcb41881c5564d06edcaaf0b9d Mon Sep 17 00:00:00 2001 From: Kensei Nakada Date: Fri, 6 Sep 2024 11:19:49 +0900 Subject: [PATCH 1/2] feat: introduce pInfo.GatingPlugin to filter out events more generally --- pkg/scheduler/backend/queue/active_queue.go | 2 + .../backend/queue/scheduling_queue.go | 129 +++++++++++------- .../backend/queue/scheduling_queue_test.go | 74 +++++++--- pkg/scheduler/framework/interface.go | 9 +- .../defaultpreemption/default_preemption.go | 22 ++- pkg/scheduler/framework/types.go | 25 +++- pkg/scheduler/scheduler.go | 9 +- .../scheduler/plugins/plugins_test.go | 14 +- 8 files changed, 202 insertions(+), 82 deletions(-) diff --git a/pkg/scheduler/backend/queue/active_queue.go b/pkg/scheduler/backend/queue/active_queue.go index 14c734d7561..60cc7f87a24 100644 --- a/pkg/scheduler/backend/queue/active_queue.go +++ b/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 } diff --git a/pkg/scheduler/backend/queue/scheduling_queue.go b/pkg/scheduler/backend/queue/scheduling_queue.go index 69eae17fd16..284ae25ef92 100644 --- a/pkg/scheduler/backend/queue/scheduling_queue.go +++ b/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,56 @@ 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() { + // This plugin passed, remove it from the unschedulable plugins. + // 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. + pInfo.UnschedulablePlugins.Delete(pl.Name()) + 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 +601,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 +652,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 +998,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 +1062,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 +1160,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 +1173,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 +1368,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() diff --git a/pkg/scheduler/backend/queue/scheduling_queue_test.go b/pkg/scheduler/backend/queue/scheduling_queue_test.go index d221ff2617e..97e300a2f14 100644 --- a/pkg/scheduler/backend/queue/scheduling_queue_test.go +++ b/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") } } diff --git a/pkg/scheduler/framework/interface.go b/pkg/scheduler/framework/interface.go index f52a2a2abdb..c32ced4a2a2 100644 --- a/pkg/scheduler/framework/interface.go +++ b/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 diff --git a/pkg/scheduler/framework/plugins/defaultpreemption/default_preemption.go b/pkg/scheduler/framework/plugins/defaultpreemption/default_preemption.go index c36528530f9..922693bde43 100644 --- a/pkg/scheduler/framework/plugins/defaultpreemption/default_preemption.go +++ b/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 // and . The number of diff --git a/pkg/scheduler/framework/types.go b/pkg/scheduler/framework/types.go index f580dfca473..a0d2cc73274 100644 --- a/pkg/scheduler/framework/types.go +++ b/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, } } diff --git a/pkg/scheduler/scheduler.go b/pkg/scheduler/scheduler.go index 0fd972fbc73..cae4922205b 100644 --- a/pkg/scheduler/scheduler.go +++ b/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) diff --git a/test/integration/scheduler/plugins/plugins_test.go b/test/integration/scheduler/plugins/plugins_test.go index 1c511feda41..c560fffd1be 100644 --- a/test/integration/scheduler/plugins/plugins_test.go +++ b/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 { From d28c8cd48879e31486bba3a5e91c32fcc5b2c7c6 Mon Sep 17 00:00:00 2001 From: Kensei Nakada Date: Wed, 7 May 2025 14:08:23 +0200 Subject: [PATCH 2/2] fix: not removing the plugin from the unsched plugins after PreEnqueue --- pkg/scheduler/backend/queue/scheduling_queue.go | 2 -- 1 file changed, 2 deletions(-) diff --git a/pkg/scheduler/backend/queue/scheduling_queue.go b/pkg/scheduler/backend/queue/scheduling_queue.go index 284ae25ef92..84a63f4768d 100644 --- a/pkg/scheduler/backend/queue/scheduling_queue.go +++ b/pkg/scheduler/backend/queue/scheduling_queue.go @@ -579,9 +579,7 @@ func (p *PriorityQueue) runPreEnqueuePlugin(ctx context.Context, logger klog.Log p.metricsRecorder.ObservePluginDurationAsync(preEnqueue, pl.Name(), s.Code().String(), p.clock.Since(startTime).Seconds()) } if s.IsSuccess() { - // This plugin passed, remove it from the unschedulable plugins. // 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. - pInfo.UnschedulablePlugins.Delete(pl.Name()) return s } pInfo.UnschedulablePlugins.Insert(pl.Name())