Change backoffQ less function to order pods by priority in windows

Author: macsko

pkg/scheduler/backend/queue/backoff_queue.go

@@ -27,10 +27,18 @@ import (
 	"k8s.io/utils/clock"
 )
 
+// backoffQOrderingWindowDuration is a duration of an ordering window in the podBackoffQ.
+// In each window, represented as a whole second, pods are ordered by priority.
+// It is the same as interval of flushing the pods from the podBackoffQ to the activeQ, to flush the whole windows there.
+// This works only if PopFromBackoffQ feature is enabled.
+// See the KEP-5142 (http://kep.k8s.io/5142) for rationale.
+const backoffQOrderingWindowDuration = time.Second
+
 // backoffQueuer is a wrapper for backoffQ related operations.
 type backoffQueuer interface {
 	// isPodBackingoff returns true if a pod is still waiting for its backoff timer.
 	// If this returns true, the pod should not be re-tried.
+	// If the pod backoff time is in the actual ordering window, it should still be backing off.
 	isPodBackingoff(podInfo *framework.QueuedPodInfo) bool
 	// popEachBackoffCompleted run fn for all pods from podBackoffQ and podErrorBackoffQ that completed backoff while popping them.
 	popEachBackoffCompleted(logger klog.Logger, fn func(pInfo *framework.QueuedPodInfo))
@@ -39,6 +47,11 @@ type backoffQueuer interface {
 	podInitialBackoffDuration() time.Duration
 	// podMaxBackoffDuration returns maximum backoff duration that pod can get.
 	podMaxBackoffDuration() time.Duration
+	// waitUntilAlignedWithOrderingWindow waits until the time reaches a multiple of backoffQOrderingWindowDuration.
+	// It then runs the f function at the backoffQOrderingWindowDuration interval using a ticker.
+	// It's important to align the flushing time, because podBackoffQ's ordering is based on the windows
+	// and whole windows have to be flushed at one time without a visible latency.
+	waitUntilAlignedWithOrderingWindow(f func(), stopCh <-chan struct{})
 
 	// add adds the pInfo to backoffQueue.
 	// The event should show which event triggered this addition and is used for the metric recording.
@@ -54,15 +67,15 @@ type backoffQueuer interface {
 	// has inform if pInfo exists in the queue.
 	has(pInfo *framework.QueuedPodInfo) bool
 	// list returns all pods that are in the queue.
-	list() []*framework.QueuedPodInfo
+	list() []*v1.Pod
 	// len returns length of the queue.
 	len() int
 }
 
 // backoffQueue implements backoffQueuer and wraps two queues inside,
 // providing seamless access as if it were one queue.
 type backoffQueue struct {
-	clock clock.Clock
+	clock clock.WithTicker
 
 	// podBackoffQ is a heap ordered by backoff expiry. Pods which have completed backoff
 	// are popped from this heap before the scheduler looks at activeQ
@@ -73,15 +86,27 @@ type backoffQueue struct {
 
 	podInitialBackoff time.Duration
 	podMaxBackoff     time.Duration
+	// activeQLessFn is used as an eventual less function if two backoff times are equal,
+	// when the SchedulerPopFromBackoffQ feature is enabled.
+	activeQLessFn framework.LessFunc
+
+	// isPopFromBackoffQEnabled indicates whether the feature gate SchedulerPopFromBackoffQ is enabled.
+	isPopFromBackoffQEnabled bool
 }
 
-func newBackoffQueue(clock clock.Clock, podInitialBackoffDuration time.Duration, podMaxBackoffDuration time.Duration) *backoffQueue {
+func newBackoffQueue(clock clock.WithTicker, podInitialBackoffDuration time.Duration, podMaxBackoffDuration time.Duration, activeQLessFn framework.LessFunc, popFromBackoffQEnabled bool) *backoffQueue {
 	bq := &backoffQueue{
-		clock:             clock,
-		podInitialBackoff: podInitialBackoffDuration,
-		podMaxBackoff:     podMaxBackoffDuration,
+		clock:                    clock,
+		podInitialBackoff:        podInitialBackoffDuration,
+		podMaxBackoff:            podMaxBackoffDuration,
+		isPopFromBackoffQEnabled: popFromBackoffQEnabled,
+		activeQLessFn:            activeQLessFn,
 	}
-	bq.podBackoffQ = heap.NewWithRecorder(podInfoKeyFunc, bq.lessBackoffCompleted, metrics.NewBackoffPodsRecorder())
+	podBackoffQLessFn := bq.lessBackoffCompleted
+	if popFromBackoffQEnabled {
+		podBackoffQLessFn = bq.lessBackoffCompletedWithPriority
+	}
+	bq.podBackoffQ = heap.NewWithRecorder(podInfoKeyFunc, podBackoffQLessFn, metrics.NewBackoffPodsRecorder())
 	bq.podErrorBackoffQ = heap.NewWithRecorder(podInfoKeyFunc, bq.lessBackoffCompleted, metrics.NewBackoffPodsRecorder())
 
 	return bq
@@ -97,7 +122,70 @@ func (bq *backoffQueue) podMaxBackoffDuration() time.Duration {
 	return bq.podMaxBackoff
 }
 
-// lessBackoffCompleted is a less function of podBackoffQ and podErrorBackoffQ.
+// alignToWindow truncates the provided time to the podBackoffQ ordering window.
+// It returns the lowest possible timestamp in the window.
+func (bq *backoffQueue) alignToWindow(t time.Time) time.Time {
+	if !bq.isPopFromBackoffQEnabled {
+		return t
+	}
+	return t.Truncate(backoffQOrderingWindowDuration)
+}
+
+// waitUntilAlignedWithOrderingWindow waits until the time reaches a multiple of backoffQOrderingWindowDuration.
+// It then runs the f function at the backoffQOrderingWindowDuration interval using a ticker.
+// It's important to align the flushing time, because podBackoffQ's ordering is based on the windows
+// and whole windows have to be flushed at one time without a visible latency.
+func (bq *backoffQueue) waitUntilAlignedWithOrderingWindow(f func(), stopCh <-chan struct{}) {
+	now := bq.clock.Now()
+	// Wait until the time reaches the multiple of backoffQOrderingWindowDuration.
+	durationToNextWindow := bq.alignToWindow(now.Add(backoffQOrderingWindowDuration)).Sub(now)
+	timer := bq.clock.NewTimer(durationToNextWindow)
+	select {
+	case <-stopCh:
+		timer.Stop()
+		return
+	case <-timer.C():
+	}
+
+	// Run a ticker to make sure the invocations of f function
+	// are aligned with the backoffQ's ordering window.
+	ticker := bq.clock.NewTicker(backoffQOrderingWindowDuration)
+	for {
+		select {
+		case <-stopCh:
+			return
+		default:
+		}
+
+		f()
+
+		// NOTE: b/c there is no priority selection in golang
+		// it is possible for this to race, meaning we could
+		// trigger ticker.C and stopCh, and ticker.C select falls through.
+		// In order to mitigate we re-check stopCh at the beginning
+		// of every loop to prevent extra executions of f().
+		select {
+		case <-stopCh:
+			ticker.Stop()
+			return
+		case <-ticker.C():
+		}
+	}
+}
+
+// lessBackoffCompletedWithPriority is a less function of podBackoffQ if PopFromBackoffQ feature is enabled.
+// It orders the pods in the same BackoffOrderingWindow the same as the activeQ will do to improve popping order from backoffQ when activeQ is empty.
+func (bq *backoffQueue) lessBackoffCompletedWithPriority(pInfo1, pInfo2 *framework.QueuedPodInfo) bool {
+	bo1 := bq.getBackoffTime(pInfo1)
+	bo2 := bq.getBackoffTime(pInfo2)
+	if !bo1.Equal(bo2) {
+		return bo1.Before(bo2)
+	}
+	// If the backoff time is the same, sort the pod in the same manner as activeQ does.
+	return bq.activeQLessFn(pInfo1, pInfo2)
+}
+
+// lessBackoffCompleted is a less function of podErrorBackoffQ.
 func (bq *backoffQueue) lessBackoffCompleted(pInfo1, pInfo2 *framework.QueuedPodInfo) bool {
 	bo1 := bq.getBackoffTime(pInfo1)
 	bo2 := bq.getBackoffTime(pInfo2)
@@ -106,9 +194,11 @@ func (bq *backoffQueue) lessBackoffCompleted(pInfo1, pInfo2 *framework.QueuedPod
 
 // isPodBackingoff returns true if a pod is still waiting for its backoff timer.
 // If this returns true, the pod should not be re-tried.
+// If the pod backoff time is in the actual ordering window, it should still be backing off.
 func (bq *backoffQueue) isPodBackingoff(podInfo *framework.QueuedPodInfo) bool {
 	boTime := bq.getBackoffTime(podInfo)
-	return boTime.After(bq.clock.Now())
+	// Don't use After, because in case of windows equality we want to return true.
+	return !boTime.Before(bq.alignToWindow(bq.clock.Now()))
 }
 
 // getBackoffTime returns the time that podInfo completes backoff.
@@ -117,9 +207,14 @@ func (bq *backoffQueue) isPodBackingoff(podInfo *framework.QueuedPodInfo) bool {
 // because of the fact that the backoff time is calculated based on podInfo.Attempts,
 // which doesn't get changed until the pod's scheduling is retried.
 func (bq *backoffQueue) getBackoffTime(podInfo *framework.QueuedPodInfo) time.Time {
+	if podInfo.Attempts == 0 {
+		// Don't store backoff expiration if the duration is 0
+		// to correctly handle isPodBackingoff, if pod should skip backoff, when it wasn't tried at all.
+		return time.Time{}
+	}
 	if podInfo.BackoffExpiration.IsZero() {
 		duration := bq.calculateBackoffDuration(podInfo)
-		podInfo.BackoffExpiration = podInfo.Timestamp.Add(duration)
+		podInfo.BackoffExpiration = bq.alignToWindow(podInfo.Timestamp.Add(duration))
 	}
 	return podInfo.BackoffExpiration
 }
@@ -238,8 +333,15 @@ func (bq *backoffQueue) has(pInfo *framework.QueuedPodInfo) bool {
 }
 
 // list returns all pods that are in the queue.
-func (bq *backoffQueue) list() []*framework.QueuedPodInfo {
-	return append(bq.podBackoffQ.List(), bq.podErrorBackoffQ.List()...)
+func (bq *backoffQueue) list() []*v1.Pod {
+	var result []*v1.Pod
+	for _, pInfo := range bq.podBackoffQ.List() {
+		result = append(result, pInfo.Pod)
+	}
+	for _, pInfo := range bq.podErrorBackoffQ.List() {
+		result = append(result, pInfo.Pod)
+	}
+	return result
 }
 
 // len returns length of the queue.

pkg/scheduler/backend/queue/backoff_queue_test.go

@@ -17,6 +17,7 @@ limitations under the License.
 package queue
 
 import (
+	"fmt"
 	"math"
 	"testing"
 	"time"
@@ -69,7 +70,7 @@ func TestBackoffQueue_calculateBackoffDuration(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			bq := newBackoffQueue(clock.RealClock{}, tt.initialBackoffDuration, tt.maxBackoffDuration)
+			bq := newBackoffQueue(clock.RealClock{}, tt.initialBackoffDuration, tt.maxBackoffDuration, newDefaultQueueSort(), true)
 			if got := bq.calculateBackoffDuration(tt.podInfo); got != tt.want {
 				t.Errorf("backoffQueue.calculateBackoffDuration() = %v, want %v", got, tt.want)
 			}
@@ -84,7 +85,7 @@ func TestBackoffQueue_popEachBackoffCompleted(t *testing.T) {
 			PodInfo: &framework.PodInfo{
 				Pod: st.MakePod().Name("pod0").Obj(),
 			},
-			Timestamp:            fakeClock.Now().Add(-time.Second),
+			Timestamp:            fakeClock.Now().Add(-2 * time.Second),
 			Attempts:             1,
 			UnschedulablePlugins: sets.New("plugin"),
 		},
@@ -100,7 +101,7 @@ func TestBackoffQueue_popEachBackoffCompleted(t *testing.T) {
 			PodInfo: &framework.PodInfo{
 				Pod: st.MakePod().Name("pod2").Obj(),
 			},
-			Timestamp: fakeClock.Now().Add(-time.Second),
+			Timestamp: fakeClock.Now().Add(-2 * time.Second),
 			Attempts:  1,
 		},
 		"pod3": {
@@ -147,12 +148,105 @@ func TestBackoffQueue_popEachBackoffCompleted(t *testing.T) {
 			wantPods:      nil,
 		},
 	}
+	for _, tt := range tests {
+		for _, popFromBackoffQEnabled := range []bool{true, false} {
+			t.Run(fmt.Sprintf("%s popFromBackoffQEnabled(%v)", tt.name, popFromBackoffQEnabled), func(t *testing.T) {
+				logger, _ := ktesting.NewTestContext(t)
+				bq := newBackoffQueue(fakeClock, DefaultPodInitialBackoffDuration, DefaultPodMaxBackoffDuration, newDefaultQueueSort(), popFromBackoffQEnabled)
+				for _, podName := range tt.podsInBackoff {
+					bq.add(logger, podInfos[podName], framework.EventUnscheduledPodAdd.Label())
+				}
+				var gotPods []string
+				bq.popEachBackoffCompleted(logger, func(pInfo *framework.QueuedPodInfo) {
+					gotPods = append(gotPods, pInfo.Pod.Name)
+				})
+				if diff := cmp.Diff(tt.wantPods, gotPods); diff != "" {
+					t.Errorf("Unexpected pods moved (-want, +got):\n%s", diff)
+				}
+				podsToStayInBackoff := len(tt.podsInBackoff) - len(tt.wantPods)
+				if bq.len() != podsToStayInBackoff {
+					t.Errorf("Expected %v pods to stay in backoffQ, but got: %v", podsToStayInBackoff, bq.len())
+				}
+			})
+		}
+	}
+}
+
+func TestBackoffQueueOrdering(t *testing.T) {
+	// Align the fake clock with ordering window.
+	fakeClock := testingclock.NewFakeClock(time.Now().Truncate(backoffQOrderingWindowDuration))
+	podInfos := []*framework.QueuedPodInfo{
+		{
+			PodInfo: &framework.PodInfo{
+				Pod: st.MakePod().Name("pod0").Priority(1).Obj(),
+			},
+			Timestamp:            fakeClock.Now(),
+			Attempts:             1,
+			UnschedulablePlugins: sets.New("plugin"),
+		},
+		{
+			PodInfo: &framework.PodInfo{
+				Pod: st.MakePod().Name("pod1").Priority(1).Obj(),
+			},
+			Timestamp:            fakeClock.Now().Add(-time.Second),
+			Attempts:             1,
+			UnschedulablePlugins: sets.New("plugin"),
+		},
+		{
+			PodInfo: &framework.PodInfo{
+				Pod: st.MakePod().Name("pod2").Priority(2).Obj(),
+			},
+			Timestamp:            fakeClock.Now().Add(-2*time.Second + time.Millisecond),
+			Attempts:             1,
+			UnschedulablePlugins: sets.New("plugin"),
+		},
+		{
+			PodInfo: &framework.PodInfo{
+				Pod: st.MakePod().Name("pod3").Priority(1).Obj(),
+			},
+			Timestamp:            fakeClock.Now().Add(-2 * time.Second),
+			Attempts:             1,
+			UnschedulablePlugins: sets.New("plugin"),
+		},
+		{
+			PodInfo: &framework.PodInfo{
+				Pod: st.MakePod().Name("pod4").Priority(2).Obj(),
+			},
+			Timestamp:            fakeClock.Now().Add(-2 * time.Second),
+			Attempts:             1,
+			UnschedulablePlugins: sets.New("plugin"),
+		},
+		{
+			PodInfo: &framework.PodInfo{
+				Pod: st.MakePod().Name("pod5").Priority(1).Obj(),
+			},
+			Timestamp:            fakeClock.Now().Add(-3 * time.Second),
+			Attempts:             1,
+			UnschedulablePlugins: sets.New("plugin"),
+		},
+	}
+	tests := []struct {
+		name                   string
+		popFromBackoffQEnabled bool
+		wantPods               []string
+	}{
+		{
+			name:                   "Pods with the same window are ordered by priority if PopFromBackoffQ is enabled",
+			popFromBackoffQEnabled: true,
+			wantPods:               []string{"pod5", "pod4", "pod2", "pod3"},
+		},
+		{
+			name:                   "Pods priority doesn't matter if PopFromBackoffQ is disabled",
+			popFromBackoffQEnabled: false,
+			wantPods:               []string{"pod5", "pod3", "pod4", "pod2"},
+		},
+	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			logger, _ := ktesting.NewTestContext(t)
-			bq := newBackoffQueue(fakeClock, DefaultPodInitialBackoffDuration, DefaultPodMaxBackoffDuration)
-			for _, podName := range tt.podsInBackoff {
-				bq.add(logger, podInfos[podName], framework.EventUnscheduledPodAdd.Label())
+			bq := newBackoffQueue(fakeClock, DefaultPodInitialBackoffDuration, DefaultPodMaxBackoffDuration, newDefaultQueueSort(), tt.popFromBackoffQEnabled)
+			for _, podInfo := range podInfos {
+				bq.add(logger, podInfo, framework.EventUnscheduledPodAdd.Label())
 			}
 			var gotPods []string
 			bq.popEachBackoffCompleted(logger, func(pInfo *framework.QueuedPodInfo) {
@@ -161,10 +255,6 @@ func TestBackoffQueue_popEachBackoffCompleted(t *testing.T) {
 			if diff := cmp.Diff(tt.wantPods, gotPods); diff != "" {
 				t.Errorf("Unexpected pods moved (-want, +got):\n%s", diff)
 			}
-			podsToStayInBackoff := len(tt.podsInBackoff) - len(tt.wantPods)
-			if bq.len() != podsToStayInBackoff {
-				t.Errorf("Expected %v pods to stay in backoffQ, but got: %v", podsToStayInBackoff, bq.len())
-			}
 		})
 	}
 }