diff --git a/pkg/scheduler/framework/plugins/defaultpreemption/default_preemption.go b/pkg/scheduler/framework/plugins/defaultpreemption/default_preemption.go index c36528530f9..8c9abbe9fb1 100644 --- a/pkg/scheduler/framework/plugins/defaultpreemption/default_preemption.go +++ b/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." } diff --git a/pkg/scheduler/framework/plugins/defaultpreemption/default_preemption_test.go b/pkg/scheduler/framework/plugins/defaultpreemption/default_preemption_test.go index cb0b04a71e0..d8d470e19a2 100644 --- a/pkg/scheduler/framework/plugins/defaultpreemption/default_preemption_test.go +++ b/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 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()) } diff --git a/pkg/scheduler/util/utils.go b/pkg/scheduler/util/utils.go index e14b4e77453..c565aa97295 100644 --- a/pkg/scheduler/util/utils.go +++ b/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)