Merge pull request #115583 from lianghao208/scheduling-gate

Optimization on running prePreEnqueuePlugins before adding pods into activeQ

pkg/scheduler/internal/queue/scheduling_queue.go

@@ -521,7 +521,9 @@ func (p *PriorityQueue) flushBackoffQCompleted() {
 			klog.ErrorS(err, "Unable to pop pod from backoff queue despite backoff completion", "pod", klog.KObj(pod))
 			break
 		}
-		if added, _ := p.addToActiveQ(pInfo); added {
+		if err := p.activeQ.Add(pInfo); err != nil {
+			klog.ErrorS(err, "Error adding pod to the active queue", "pod", klog.KObj(pInfo.Pod))
+		} else {
 			klog.V(5).InfoS("Pod moved to an internal scheduling queue", "pod", klog.KObj(pod), "event", BackoffComplete, "queue", activeQName)
 			metrics.SchedulerQueueIncomingPods.WithLabelValues("active", BackoffComplete).Inc()
 			activated = true

pkg/scheduler/internal/queue/scheduling_queue_test.go

@@ -447,6 +447,8 @@ func TestPriorityQueue_Activate(t *testing.T) {
 }
 
 type preEnqueuePlugin struct {
+	// called counts the number of calling PreEnqueue()
+	called     int
 	allowlists []string
 }
 
@@ -455,6 +457,7 @@ func (pl *preEnqueuePlugin) Name() string {
 }
 
 func (pl *preEnqueuePlugin) PreEnqueue(ctx context.Context, p *v1.Pod) *framework.Status {
+	pl.called++
 	for _, allowed := range pl.allowlists {
 		for label := range p.Labels {
 			if label == allowed {
@@ -536,6 +539,47 @@ func TestPriorityQueue_addToActiveQ(t *testing.T) {
 	}
 }
 
+func TestPriorityQueue_flushBackoffQCompleted(t *testing.T) {
+	tests := []struct {
+		name                       string
+		plugin                     framework.PreEnqueuePlugin
+		pod                        *v1.Pod
+		operations                 []operation
+		wantPreEnqueuePluginCalled int
+	}{
+		{
+			name:   "preEnqueue plugin registered, not running preEnqueue plugin when backoff completed",
+			plugin: &preEnqueuePlugin{},
+			pod:    st.MakePod().Name("foo").Label("foo", "").Obj(),
+			operations: []operation{
+				addPodBackoffQ,
+				flushBackoffQ,
+			},
+			wantPreEnqueuePluginCalled: 0,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+
+			m := map[string][]framework.PreEnqueuePlugin{"": {tt.plugin}}
+			c := testingclock.NewFakeClock(time.Now())
+			q := NewTestQueueWithObjects(ctx, newDefaultQueueSort(), []runtime.Object{tt.pod}, WithPreEnqueuePluginMap(m),
+				WithPodInitialBackoffDuration(time.Second*1), WithPodMaxBackoffDuration(time.Second*60), WithClock(c))
+			pInfo := newQueuedPodInfoForLookup(tt.pod)
+			pInfo.Gated = true
+			for _, op := range tt.operations {
+				op(q, pInfo)
+			}
+			if tt.wantPreEnqueuePluginCalled != tt.plugin.(*preEnqueuePlugin).called {
+				t.Errorf("Unexpected number of calling preEnqueue: want %v, but got %v", tt.wantPreEnqueuePluginCalled, tt.plugin.(*preEnqueuePlugin).called)
+			}
+		})
+	}
+}
+
 func BenchmarkMoveAllToActiveOrBackoffQueue(b *testing.B) {
 	tests := []struct {
 		name      string