feature(scheduler): Custom pod selection/ordering in DefaultPreemption

The current behavior is to select only based on pod priority, where any
pod with higher priority can preempt any pod with lower priority.

In our case, we have multiple priority classes but where only a subset
of those are considered preemptible. Here's roughly how this looks:

- High priority: higher in the scheduling queue, not preemptible
- Low priority: lower in the scheduling queue, not preemptible
- Preemptible priority: lowest in the scheduling queue, preemptible

This PR allows the preemption selection to be configured against the
`DefaultPreemption` plugin, rather than needing to reimplement the
plugin itself. The structure used here mimics [what's currently being
done for the `Evaluator`](https://github.com/kubernetes/kubernetes/blob/release-1.32/pkg/scheduler/framework/preemption/preemption.go#L161-L165),
where a `PreemptPod` callback may be overridden to customize what
happens when performing a preemption.

This PR also updates the plugin's tests to call the updated `New()`
function rather than constructing `DefaultPreemption` directly, so
that `New()` is now being exercised in tests.

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

@@ -54,6 +54,14 @@ type DefaultPreemption struct {
 	podLister corelisters.PodLister
 	pdbLister policylisters.PodDisruptionBudgetLister
 	Evaluator *preemption.Evaluator
+
+	// EligiblePods returns victim pods which are allowed to be preempted by the provided preemptor.
+	// The default behavior is to allow any pods of lower priority to be preempted by any pods of higher priority.
+	EligiblePods func(nodeInfo *framework.NodeInfo, preemptor *v1.Pod) []*framework.PodInfo
+
+	// OrderedPods sorts eligible victims in-place in descending order of highest to lowest priority.
+	// Pods at the start of the slice are less likely to be preempted.
+	OrderedPods func(eligible []*framework.PodInfo)
 }
 
 var _ framework.PostFilterPlugin = &DefaultPreemption{}
@@ -86,6 +94,23 @@ func New(_ context.Context, dpArgs runtime.Object, fh framework.Handle, fts feat
 	}
 	pl.Evaluator = preemption.NewEvaluator(Name, fh, &pl, fts.EnableAsyncPreemption)
 
+	// Default behavior: Any pods of lower priority may be preempted by any pods of higher priority.
+	pl.EligiblePods = func(nodeInfo *framework.NodeInfo, preemptor *v1.Pod) []*framework.PodInfo {
+		var eligible []*framework.PodInfo
+		podPriority := corev1helpers.PodPriority(preemptor)
+		for _, pi := range nodeInfo.Pods {
+			if corev1helpers.PodPriority(pi.Pod) < podPriority {
+				eligible = append(eligible, pi)
+			}
+		}
+		return eligible
+	}
+
+	// Default behavior: Sort by descending priority, then by descending runtime duration as secondary ordering.
+	pl.OrderedPods = func(eligible []*framework.PodInfo) {
+		sort.Slice(eligible, func(i, j int) bool { return util.MoreImportantPod(eligible[i].Pod, eligible[j].Pod) })
+	}
+
 	return &pl, nil
 }
 
@@ -165,7 +190,6 @@ func (pl *DefaultPreemption) SelectVictimsOnNode(
 	nodeInfo *framework.NodeInfo,
 	pdbs []*policy.PodDisruptionBudget) ([]*v1.Pod, int, *framework.Status) {
 	logger := klog.FromContext(ctx)
-	var potentialVictims []*framework.PodInfo
 	removePod := func(rpi *framework.PodInfo) error {
 		if err := nodeInfo.RemovePod(logger, rpi.Pod); err != nil {
 			return err
@@ -184,15 +208,12 @@ func (pl *DefaultPreemption) SelectVictimsOnNode(
 		}
 		return nil
 	}
-	// As the first step, remove all the lower priority pods from the node and
-	// check if the given pod can be scheduled.
-	podPriority := corev1helpers.PodPriority(pod)
-	for _, pi := range nodeInfo.Pods {
-		if corev1helpers.PodPriority(pi.Pod) < podPriority {
-			potentialVictims = append(potentialVictims, pi)
-			if err := removePod(pi); err != nil {
-				return nil, 0, framework.AsStatus(err)
-			}
+	// As the first step, remove all pods eligible for preemption from the node and
+	// check if the given pod could be scheduled without them present.
+	potentialVictims := pl.EligiblePods(nodeInfo, pod)
+	for _, pi := range potentialVictims {
+		if err := removePod(pi); err != nil {
+			return nil, 0, framework.AsStatus(err)
 		}
 	}
 
@@ -201,7 +222,7 @@ func (pl *DefaultPreemption) SelectVictimsOnNode(
 		return nil, 0, framework.NewStatus(framework.UnschedulableAndUnresolvable, "No preemption victims found for incoming pod")
 	}
 
-	// If the new pod does not fit after removing all the lower priority pods,
+	// If the new pod does not fit after removing all the eligible pods,
 	// we are almost done and this node is not suitable for preemption. The only
 	// condition that we could check is if the "pod" is failing to schedule due to
 	// inter-pod affinity to one or more victims, but we have decided not to
@@ -210,11 +231,11 @@ func (pl *DefaultPreemption) SelectVictimsOnNode(
 	if status := pl.fh.RunFilterPluginsWithNominatedPods(ctx, state, pod, nodeInfo); !status.IsSuccess() {
 		return nil, 0, status
 	}
-	var victims []*v1.Pod
+	var victims []*framework.PodInfo
 	numViolatingVictim := 0
-	// Sort potentialVictims by pod priority from high to low, which ensures to
-	// reprieve higher priority pods first.
-	sort.Slice(potentialVictims, func(i, j int) bool { return util.MoreImportantPod(potentialVictims[i].Pod, potentialVictims[j].Pod) })
+	// Sort potentialVictims by descending importance, which ensures reprieve of
+	// higher importance pods first.
+	pl.OrderedPods(potentialVictims)
 	// Try to reprieve as many pods as possible. We first try to reprieve the PDB
 	// violating victims and then other non-violating ones. In both cases, we start
 	// from the highest priority victims.
@@ -229,9 +250,8 @@ func (pl *DefaultPreemption) SelectVictimsOnNode(
 			if err := removePod(pi); err != nil {
 				return false, err
 			}
-			rpi := pi.Pod
-			victims = append(victims, rpi)
-			logger.V(5).Info("Pod is a potential preemption victim on node", "pod", klog.KObj(rpi), "node", klog.KObj(nodeInfo.Node()))
+			victims = append(victims, pi)
+			logger.V(5).Info("Pod is a potential preemption victim on node", "pod", klog.KObj(pi.Pod), "node", klog.KObj(nodeInfo.Node()))
 		}
 		return fits, nil
 	}
@@ -251,9 +271,13 @@ func (pl *DefaultPreemption) SelectVictimsOnNode(
 
 	// Sort victims after reprieving pods to keep the pods in the victims sorted in order of priority from high to low.
 	if len(violatingVictims) != 0 && len(nonViolatingVictims) != 0 {
-		sort.Slice(victims, func(i, j int) bool { return util.MoreImportantPod(victims[i], victims[j]) })
+		pl.OrderedPods(victims)
 	}
-	return victims, numViolatingVictim, framework.NewStatus(framework.Success)
+	var victimPods []*v1.Pod
+	for _, pi := range victims {
+		victimPods = append(victimPods, pi.Pod)
+	}
+	return victimPods, numViolatingVictim, framework.NewStatus(framework.Success)
 }
 
 // PodEligibleToPreemptOthers returns one bool and one string. The bool
@@ -279,9 +303,8 @@ func (pl *DefaultPreemption) PodEligibleToPreemptOthers(_ context.Context, pod *
 		}
 
 		if nodeInfo, _ := nodeInfos.Get(nomNodeName); nodeInfo != nil {
-			podPriority := corev1helpers.PodPriority(pod)
-			for _, p := range nodeInfo.Pods {
-				if corev1helpers.PodPriority(p.Pod) < podPriority && podTerminatingByPreemption(p.Pod) {
+			for _, p := range pl.EligiblePods(nodeInfo, pod) {
+				if podTerminatingByPreemption(p.Pod) {
 					// There is a terminating pod on the nominated node.
 					return false, "not eligible due to a terminating pod on the nominated node."
 				}

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

@@ -57,12 +57,10 @@ import (
 	"k8s.io/kubernetes/pkg/scheduler/framework/plugins/defaultbinder"
 	"k8s.io/kubernetes/pkg/scheduler/framework/plugins/feature"
 	"k8s.io/kubernetes/pkg/scheduler/framework/plugins/interpodaffinity"
-	"k8s.io/kubernetes/pkg/scheduler/framework/plugins/names"
 	"k8s.io/kubernetes/pkg/scheduler/framework/plugins/noderesources"
 	"k8s.io/kubernetes/pkg/scheduler/framework/plugins/podtopologyspread"
 	"k8s.io/kubernetes/pkg/scheduler/framework/plugins/queuesort"
 	"k8s.io/kubernetes/pkg/scheduler/framework/plugins/tainttoleration"
-	"k8s.io/kubernetes/pkg/scheduler/framework/preemption"
 	frameworkruntime "k8s.io/kubernetes/pkg/scheduler/framework/runtime"
 	"k8s.io/kubernetes/pkg/scheduler/metrics"
 	st "k8s.io/kubernetes/pkg/scheduler/testing"
@@ -436,13 +434,14 @@ func TestPostFilter(t *testing.T) {
 			if err != nil {
 				t.Fatal(err)
 			}
-			p := DefaultPreemption{
-				fh:        f,
-				podLister: informerFactory.Core().V1().Pods().Lister(),
-				pdbLister: getPDBLister(informerFactory),
-				args:      *getDefaultDefaultPreemptionArgs(),
+			pi, err := New(context.Background(), getDefaultDefaultPreemptionArgs(), f, feature.Features{})
+			if err != nil {
+				t.Fatal(err)
+			}
+			p, ok := pi.(*DefaultPreemption)
+			if !ok {
+				t.Fatal("Unable to cast to DefaultPreemption")
 			}
-			p.Evaluator = preemption.NewEvaluator(names.DefaultPreemption, f, &p, false)
 
 			state := framework.NewCycleState()
 			// Ensure <state> is populated.
@@ -1189,11 +1188,13 @@ func TestDryRunPreemption(t *testing.T) {
 			if tt.args == nil {
 				tt.args = getDefaultDefaultPreemptionArgs()
 			}
-			pl := &DefaultPreemption{
-				fh:        fwk,
-				podLister: informerFactory.Core().V1().Pods().Lister(),
-				pdbLister: getPDBLister(informerFactory),
-				args:      *tt.args,
+			pi, err := New(context.Background(), tt.args, fwk, feature.Features{})
+			if err != nil {
+				t.Fatal(err)
+			}
+			pl, ok := pi.(*DefaultPreemption)
+			if !ok {
+				t.Fatal("Unable to cast to DefaultPreemption")
 			}
 
 			// Using 4 as a seed source to test getOffsetAndNumCandidates() deterministically.
@@ -1207,15 +1208,8 @@ func TestDryRunPreemption(t *testing.T) {
 				if _, status, _ := fwk.RunPreFilterPlugins(ctx, state, pod); !status.IsSuccess() {
 					t.Errorf("cycle %d: Unexpected PreFilter Status: %v", cycle, status)
 				}
-				pe := preemption.Evaluator{
-					PluginName: names.DefaultPreemption,
-					Handler:    pl.fh,
-					PodLister:  pl.podLister,
-					PdbLister:  pl.pdbLister,
-					Interface:  pl,
-				}
 				offset, numCandidates := pl.GetOffsetAndNumCandidates(int32(len(nodeInfos)))
-				got, _, _ := pe.DryRunPreemption(ctx, state, pod, nodeInfos, tt.pdbs, offset, numCandidates)
+				got, _, _ := pl.Evaluator.DryRunPreemption(ctx, state, pod, nodeInfos, tt.pdbs, offset, numCandidates)
 				// Sort the values (inner victims) and the candidate itself (by its NominatedNodeName).
 				for i := range got {
 					victims := got[i].Victims().Pods
@@ -1425,6 +1419,7 @@ func TestSelectBestCandidate(t *testing.T) {
 				"",
 				frameworkruntime.WithPodNominator(internalqueue.NewSchedulingQueue(nil, informerFactory)),
 				frameworkruntime.WithSnapshotSharedLister(snapshot),
+				frameworkruntime.WithInformerFactory(informerFactory),
 				frameworkruntime.WithLogger(logger),
 			)
 			if err != nil {
@@ -1441,22 +1436,17 @@ func TestSelectBestCandidate(t *testing.T) {
 				t.Errorf("Unexpected PreFilter Status: %v", status)
 			}
 
-			pl := &DefaultPreemption{
-				fh:        fwk,
-				podLister: informerFactory.Core().V1().Pods().Lister(),
-				pdbLister: getPDBLister(informerFactory),
-				args:      *getDefaultDefaultPreemptionArgs(),
+			pi, err := New(context.Background(), getDefaultDefaultPreemptionArgs(), fwk, feature.Features{})
+			if err != nil {
+				t.Fatal(err)
 			}
-			pe := preemption.Evaluator{
-				PluginName: names.DefaultPreemption,
-				Handler:    pl.fh,
-				PodLister:  pl.podLister,
-				PdbLister:  pl.pdbLister,
-				Interface:  pl,
+			pl, ok := pi.(*DefaultPreemption)
+			if !ok {
+				t.Fatal("Unable to cast to DefaultPreemption")
 			}
 			offset, numCandidates := pl.GetOffsetAndNumCandidates(int32(len(nodeInfos)))
-			candidates, _, _ := pe.DryRunPreemption(ctx, state, tt.pod, nodeInfos, nil, offset, numCandidates)
-			s := pe.SelectCandidate(ctx, candidates)
+			candidates, _, _ := pl.Evaluator.DryRunPreemption(ctx, state, tt.pod, nodeInfos, nil, offset, numCandidates)
+			s := pl.Evaluator.SelectCandidate(ctx, candidates)
 			if s == nil || len(s.Name()) == 0 {
 				return
 			}
@@ -1534,18 +1524,33 @@ func TestPodEligibleToPreemptOthers(t *testing.T) {
 			for _, n := range test.nodes {
 				nodes = append(nodes, st.MakeNode().Name(n).Obj())
 			}
+			var pods []runtime.Object
+			pods = append(pods, test.pod)
+			for _, pod := range test.pods {
+				pods = append(pods, pod)
+			}
+			cs := clientsetfake.NewClientset(pods...)
+			informerFactory := informers.NewSharedInformerFactory(cs, 0)
 			registeredPlugins := []tf.RegisterPluginFunc{
 				tf.RegisterQueueSortPlugin(queuesort.Name, queuesort.New),
 				tf.RegisterBindPlugin(defaultbinder.Name, defaultbinder.New),
 			}
 			f, err := tf.NewFramework(ctx, registeredPlugins, "",
 				frameworkruntime.WithSnapshotSharedLister(internalcache.NewSnapshot(test.pods, nodes)),
+				frameworkruntime.WithInformerFactory(informerFactory),
 				frameworkruntime.WithLogger(logger),
 			)
 			if err != nil {
 				t.Fatal(err)
 			}
-			pl := DefaultPreemption{fh: f, fts: test.fts}
+			pi, err := New(context.Background(), getDefaultDefaultPreemptionArgs(), f, feature.Features{})
+			if err != nil {
+				t.Fatal(err)
+			}
+			pl, ok := pi.(*DefaultPreemption)
+			if !ok {
+				t.Fatal("Unable to cast to DefaultPreemption")
+			}
 			if got, _ := pl.PodEligibleToPreemptOthers(ctx, test.pod, test.nominatedNodeStatus); got != test.expected {
 				t.Errorf("expected %t, got %t for pod: %s", test.expected, got, test.pod.Name)
 			}
@@ -1845,14 +1850,17 @@ func TestPreempt(t *testing.T) {
 					t.Errorf("Unexpected preFilterStatus: %v", s)
 				}
 				// Call preempt and check the expected results.
-				pl := DefaultPreemption{
-					fh:        fwk,
-					podLister: informerFactory.Core().V1().Pods().Lister(),
-					pdbLister: getPDBLister(informerFactory),
-					args:      *getDefaultDefaultPreemptionArgs(),
+				features := feature.Features{
+					EnableAsyncPreemption: asyncPreemptionEnabled,
+				}
+				pi, err := New(context.Background(), getDefaultDefaultPreemptionArgs(), fwk, features)
+				if err != nil {
+					t.Fatal(err)
+				}
+				pl, ok := pi.(*DefaultPreemption)
+				if !ok {
+					t.Fatal("Unable to cast to DefaultPreemption")
 				}
-
-				pe := preemption.NewEvaluator(names.DefaultPreemption, pl.fh, &pl, asyncPreemptionEnabled)
 
 				// so that these nodes are eligible for preemption, we set their status
 				// to Unschedulable.
@@ -1862,7 +1870,7 @@ func TestPreempt(t *testing.T) {
 					nodeToStatusMap.Set(n.Name, framework.NewStatus(framework.Unschedulable))
 				}
 
-				res, status := pe.Preempt(ctx, state, testPod, nodeToStatusMap)
+				res, status := pl.Evaluator.Preempt(ctx, state, testPod, nodeToStatusMap)
 				if !status.IsSuccess() && !status.IsRejected() {
 					t.Errorf("unexpected error in preemption: %v", status.AsError())
 				}
@@ -1935,7 +1943,7 @@ func TestPreempt(t *testing.T) {
 				mu.RUnlock()
 
 				// Call preempt again and make sure it doesn't preempt any more pods.
-				res, status = pe.Preempt(ctx, state, testPod, framework.NewDefaultNodeToStatus())
+				res, status = pl.Evaluator.Preempt(ctx, state, testPod, framework.NewDefaultNodeToStatus())
 				if !status.IsSuccess() && !status.IsRejected() {
 					t.Errorf("unexpected error in preemption: %v", status.AsError())
 				}

pkg/scheduler/util/utils.go

@@ -82,9 +82,8 @@ func GetEarliestPodStartTime(victims *extenderv1.Victims) *metav1.Time {
 }
 
 // MoreImportantPod return true when priority of the first pod is higher than
-// the second one. If two pods' priorities are equal, compare their StartTime.
-// It takes arguments of the type "interface{}" to be used with SortableList,
-// but expects those arguments to be *v1.Pod.
+// the second one. If two pods' priorities are equal, compare their StartTime,
+// treating the older pod as more important.
 func MoreImportantPod(pod1, pod2 *v1.Pod) bool {
 	p1 := corev1helpers.PodPriority(pod1)
 	p2 := corev1helpers.PodPriority(pod2)