diff --git a/pkg/scheduler/framework/plugins/defaultpreemption/default_preemption.go b/pkg/scheduler/framework/plugins/defaultpreemption/default_preemption.go index 922693bde43..6fc08b83663 100644 --- a/pkg/scheduler/framework/plugins/defaultpreemption/default_preemption.go +++ b/pkg/scheduler/framework/plugins/defaultpreemption/default_preemption.go @@ -46,6 +46,23 @@ import ( // Name of the plugin used in the plugin registry and configurations. const Name = names.DefaultPreemption +// IsEligiblePodFunc is a function which may be assigned to the DefaultPreemption plugin. +// This may implement rules/filtering around preemption eligibility, which is in addition to +// the internal requirement that the victim pod have lower priority than the preemptor pod. +// Any customizations should always allow system services to preempt normal pods, to avoid +// problems if system pods are unable to find space. +type IsEligiblePodFunc func(nodeInfo *framework.NodeInfo, victim *framework.PodInfo, preemptor *v1.Pod) bool + +// MoreImportantPodFunc is a function which may be assigned to the DefaultPreemption plugin. +// Implementations should return true if the first pod is more important than the second pod +// and the second one should be considered for preemption before the first one. +// For performance reasons, the search for nodes eligible for preemption is done by omitting all +// eligible victims from a node then checking whether the preemptor fits on the node without them, +// before adding back victims (starting from the most important) that still fit with the preemptor. +// The default behavior is to not consider pod affinity between the preemptor and the victims, +// as affinity between pods that are eligible to preempt each other isn't recommended. +type MoreImportantPodFunc func(pod1, pod2 *v1.Pod) bool + // DefaultPreemption is a PostFilter plugin implements the preemption logic. type DefaultPreemption struct { fh framework.Handle @@ -54,6 +71,18 @@ type DefaultPreemption struct { podLister corelisters.PodLister pdbLister policylisters.PodDisruptionBudgetLister Evaluator *preemption.Evaluator + + // IsEligiblePod returns whether a victim pod is allowed to be preempted by a preemptor pod. + // This filtering is in addition to the internal requirement that the victim pod have lower + // priority than the preemptor pod. Any customizations should always allow system services + // to preempt normal pods, to avoid problems if system pods are unable to find space. + IsEligiblePod IsEligiblePodFunc + + // MoreImportantPod is used to sort eligible victims in-place in descending order of highest to + // lowest importance. Pods with higher importance are less likely to be preempted. + // The default behavior is to order pods by descending priority, then descending runtime duration + // for pods with equal priority. + MoreImportantPod MoreImportantPodFunc } var _ framework.PostFilterPlugin = &DefaultPreemption{} @@ -64,8 +93,8 @@ func (pl *DefaultPreemption) Name() string { return Name } -// New initializes a new plugin and returns it. -func New(_ context.Context, dpArgs runtime.Object, fh framework.Handle, fts feature.Features) (framework.Plugin, error) { +// New initializes a new plugin and returns it. The plugin type is retained to allow modification. +func New(_ context.Context, dpArgs runtime.Object, fh framework.Handle, fts feature.Features) (*DefaultPreemption, error) { args, ok := dpArgs.(*config.DefaultPreemptionArgs) if !ok { return nil, fmt.Errorf("got args of type %T, want *DefaultPreemptionArgs", dpArgs) @@ -86,6 +115,15 @@ func New(_ context.Context, dpArgs runtime.Object, fh framework.Handle, fts feat } pl.Evaluator = preemption.NewEvaluator(Name, fh, &pl, fts.EnableAsyncPreemption) + // Default behavior: No additional filtering, beyond the internal requirement that the victim pod + // have lower priority than the preemptor pod. + pl.IsEligiblePod = func(nodeInfo *framework.NodeInfo, victim *framework.PodInfo, preemptor *v1.Pod) bool { + return true + } + + // Default behavior: Sort by descending priority, then by descending runtime duration as secondary ordering. + pl.MoreImportantPod = util.MoreImportantPod + return &pl, nil } @@ -202,11 +240,10 @@ 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) + // As the first step, remove all pods eligible for preemption from the node and + // check if the given pod can be scheduled without them present. for _, pi := range nodeInfo.Pods { - if corev1helpers.PodPriority(pi.Pod) < podPriority { + if pl.isPreemptionAllowed(nodeInfo, pi, pod) { potentialVictims = append(potentialVictims, pi) if err := removePod(pi); err != nil { return nil, 0, framework.AsStatus(err) @@ -219,23 +256,23 @@ 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 // support this case for performance reasons. Having affinity to lower - // priority pods is not a recommended configuration anyway. + // importance (priority) pods is not a recommended configuration anyway. 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. + sort.Slice(potentialVictims, func(i, j int) bool { return pl.MoreImportantPod(potentialVictims[i].Pod, potentialVictims[j].Pod) }) // 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. + // from the highest importance victims. violatingVictims, nonViolatingVictims := filterPodsWithPDBViolation(potentialVictims, pdbs) reprievePod := func(pi *framework.PodInfo) (bool, error) { if err := addPod(pi); err != nil { @@ -247,9 +284,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 } @@ -267,11 +303,15 @@ 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. + // Sort victims after reprieving pods to keep the pods in the victims sorted in order of importance 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]) }) + sort.Slice(victims, func(i, j int) bool { return pl.MoreImportantPod(victims[i].Pod, victims[j].Pod) }) } - 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 @@ -297,9 +337,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) { + if pl.isPreemptionAllowed(nodeInfo, p, pod) && 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." } @@ -314,6 +353,12 @@ func (pl *DefaultPreemption) OrderedScoreFuncs(ctx context.Context, nodesToVicti return nil } +// isPreemptionAllowed returns whether the victim residing on nodeInfo can be preempted by the preemptor +func (pl *DefaultPreemption) isPreemptionAllowed(nodeInfo *framework.NodeInfo, victim *framework.PodInfo, preemptor *v1.Pod) bool { + // The victim must have lower priority than the preemptor, in addition to any filtering implemented by IsEligiblePod + return corev1helpers.PodPriority(victim.Pod) < corev1helpers.PodPriority(preemptor) && pl.IsEligiblePod(nodeInfo, victim, preemptor) +} + // podTerminatingByPreemption returns true if the pod is in the termination state caused by scheduler preemption. func podTerminatingByPreemption(p *v1.Pod) bool { if p.DeletionTimestamp == nil { diff --git a/pkg/scheduler/framework/plugins/defaultpreemption/default_preemption_test.go b/pkg/scheduler/framework/plugins/defaultpreemption/default_preemption_test.go index cb0b04a71e0..5cc3ce976f0 100644 --- a/pkg/scheduler/framework/plugins/defaultpreemption/default_preemption_test.go +++ b/pkg/scheduler/framework/plugins/defaultpreemption/default_preemption_test.go @@ -43,6 +43,7 @@ import ( clientsetfake "k8s.io/client-go/kubernetes/fake" clienttesting "k8s.io/client-go/testing" "k8s.io/client-go/tools/events" + corev1helpers "k8s.io/component-helpers/scheduling/corev1" "k8s.io/klog/v2" "k8s.io/klog/v2/ktesting" kubeschedulerconfigv1 "k8s.io/kube-scheduler/config/v1" @@ -57,7 +58,6 @@ 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" @@ -67,6 +67,7 @@ import ( "k8s.io/kubernetes/pkg/scheduler/metrics" st "k8s.io/kubernetes/pkg/scheduler/testing" tf "k8s.io/kubernetes/pkg/scheduler/testing/framework" + "k8s.io/kubernetes/pkg/scheduler/util" ) var ( @@ -436,13 +437,10 @@ 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(), + p, err := New(ctx, getDefaultDefaultPreemptionArgs(), f, feature.Features{}) + if err != nil { + t.Fatal(err) } - p.Evaluator = preemption.NewEvaluator(names.DefaultPreemption, f, &p, false) state := framework.NewCycleState() // Ensure is populated. @@ -1189,11 +1187,9 @@ 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, + pl, err := New(ctx, tt.args, fwk, feature.Features{}) + if err != nil { + t.Fatal(err) } // Using 4 as a seed source to test getOffsetAndNumCandidates() deterministically. @@ -1207,15 +1203,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 +1414,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,24 +1431,15 @@ 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(), - } - pe := preemption.Evaluator{ - PluginName: names.DefaultPreemption, - Handler: pl.fh, - PodLister: pl.podLister, - PdbLister: pl.pdbLister, - Interface: pl, + pl, err := New(ctx, getDefaultDefaultPreemptionArgs(), fwk, feature.Features{}) + if err != nil { + t.Fatal(err) } 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 + t.Fatalf("expected any node in %v, but candidate is missing", tt.expected) } found := false for _, nodeName := range tt.expected { @@ -1474,6 +1455,368 @@ func TestSelectBestCandidate(t *testing.T) { } } +func TestCustomSelection(t *testing.T) { + podLabelIsEligible := func(key, val string) IsEligiblePodFunc { + return func(nodeInfo *framework.NodeInfo, victim *framework.PodInfo, preemptor *v1.Pod) bool { + pval, ok := victim.Pod.Labels[key] + if !ok { + return false + } + return pval == val + } + } + nodeNameIsEligible := func(name string) IsEligiblePodFunc { + return func(nodeInfo *framework.NodeInfo, victim *framework.PodInfo, preemptor *v1.Pod) bool { + return nodeInfo.Node().Name == name + } + } + priorityBelowThresholdCannotPreempt := func(minPreempting int32) IsEligiblePodFunc { + return func(nodeInfo *framework.NodeInfo, victim *framework.PodInfo, preemptor *v1.Pod) bool { + return corev1helpers.PodPriority(preemptor) >= minPreempting + } + } + priorityAboveThresholdCannotBePreempted := func(maxPreemptible int32) IsEligiblePodFunc { + return func(nodeInfo *framework.NodeInfo, victim *framework.PodInfo, preemptor *v1.Pod) bool { + return corev1helpers.PodPriority(victim.Pod) <= maxPreemptible + } + } + + tests := []struct { + name string + eligiblePods IsEligiblePodFunc + nodeNames []string + pod *v1.Pod + pods []*v1.Pod + expected map[string][]string + }{ + { + name: "filter for matching pod label: high priority", + eligiblePods: podLabelIsEligible("preemptible", "yes"), + nodeNames: []string{"node1", "node2", "node3", "node4"}, + pod: st.MakePod().Name("p1").UID("p1").Priority(highPriority).Req(largeRes).Obj(), + pods: []*v1.Pod{ + st.MakePod().Name("v1").UID("v1").Label("preemptible", "no").Node("node1").Priority(midPriority).Req(largeRes).StartTime(epochTime).Obj(), + st.MakePod().Name("v2").UID("v2").Label("preemptible", "yes").Node("node2").Priority(midPriority).Req(largeRes).StartTime(epochTime).Obj(), + st.MakePod().Name("v3").UID("v3").Label("preemptible", "yes").Node("node3").Priority(lowPriority).Req(largeRes).StartTime(epochTime).Obj(), + st.MakePod().Name("v4").UID("v4").Node("node4").Priority(lowPriority).Req(largeRes).StartTime(epochTime).Obj(), + }, + expected: map[string][]string{"node2": {"v2"}, "node3": {"v3"}}, + }, + { + name: "filter for matching pod label: mid priority", + eligiblePods: podLabelIsEligible("preemptible", "yes"), + nodeNames: []string{"node1", "node2", "node3", "node4"}, + pod: st.MakePod().Name("p2").UID("p2").Priority(midPriority).Req(largeRes).Obj(), + pods: []*v1.Pod{ + st.MakePod().Name("v1").UID("v1").Label("preemptible", "no").Node("node1").Priority(midPriority).Req(largeRes).StartTime(epochTime).Obj(), + st.MakePod().Name("v2").UID("v2").Label("preemptible", "yes").Node("node2").Priority(midPriority).Req(largeRes).StartTime(epochTime).Obj(), + st.MakePod().Name("v3").UID("v3").Label("preemptible", "yes").Node("node3").Priority(lowPriority).Req(largeRes).StartTime(epochTime).Obj(), + st.MakePod().Name("v4").UID("v4").Node("node4").Priority(lowPriority).Req(largeRes).StartTime(epochTime).Obj(), + }, + expected: map[string][]string{"node3": {"v3"}}, + }, + { + name: "filter for matching pod label: low priority", + eligiblePods: podLabelIsEligible("preemptible", "yes"), + nodeNames: []string{"node1", "node2", "node3", "node4"}, + pod: st.MakePod().Name("p3").UID("p3").Priority(lowPriority).Req(largeRes).Obj(), + pods: []*v1.Pod{ + st.MakePod().Name("v1").UID("v1").Label("preemptible", "no").Node("node1").Priority(midPriority).Req(largeRes).StartTime(epochTime).Obj(), + st.MakePod().Name("v2").UID("v2").Label("preemptible", "yes").Node("node2").Priority(midPriority).Req(largeRes).StartTime(epochTime).Obj(), + st.MakePod().Name("v3").UID("v3").Label("preemptible", "yes").Node("node3").Priority(lowPriority).Req(largeRes).StartTime(epochTime).Obj(), + st.MakePod().Name("v4").UID("v4").Node("node4").Priority(lowPriority).Req(largeRes).StartTime(epochTime).Obj(), + }, + expected: map[string][]string{}, + }, + { + name: "filter for matching victim node: high priority", + eligiblePods: nodeNameIsEligible("node1"), + nodeNames: []string{"node1", "node2", "node3"}, + pod: st.MakePod().Name("p3").UID("p3").Priority(highPriority).Req(largeRes).Obj(), + pods: []*v1.Pod{ + st.MakePod().Name("v1").UID("v1").Node("node1").Priority(highPriority).Req(mediumRes).StartTime(epochTime).Obj(), + st.MakePod().Name("v2").UID("v2").Node("node1").Priority(midPriority).Req(smallRes).StartTime(epochTime).Obj(), + st.MakePod().Name("v3").UID("v3").Node("node1").Priority(lowPriority).Req(smallRes).StartTime(epochTime).Obj(), + st.MakePod().Name("v4").UID("v4").Node("node2").Priority(midPriority).Req(largeRes).StartTime(epochTime).Obj(), + st.MakePod().Name("v5").UID("v5").Node("node3").Priority(lowPriority).Req(largeRes).StartTime(epochTime).Obj(), + }, + expected: map[string][]string{"node1": {"v2", "v3"}}, + }, + { + name: "filter for matching victim node: mid priority", + eligiblePods: nodeNameIsEligible("node1"), + nodeNames: []string{"node1", "node2", "node3"}, + pod: st.MakePod().Name("p3").UID("p3").Priority(midPriority).Req(largeRes).Obj(), + pods: []*v1.Pod{ + st.MakePod().Name("v1").UID("v1").Node("node1").Priority(highPriority).Req(smallRes).StartTime(epochTime).Obj(), + st.MakePod().Name("v2").UID("v2").Node("node1").Priority(midPriority).Req(smallRes).StartTime(epochTime).Obj(), + st.MakePod().Name("v3").UID("v3").Node("node1").Priority(lowPriority).Req(smallRes).StartTime(epochTime).Obj(), + st.MakePod().Name("v4").UID("v4").Node("node2").Priority(midPriority).Req(largeRes).StartTime(epochTime).Obj(), + st.MakePod().Name("v5").UID("v5").Node("node3").Priority(lowPriority).Req(largeRes).StartTime(epochTime).Obj(), + }, + expected: map[string][]string{"node1": {"v3"}}, + }, + { + name: "filter for matching victim node: low priority", + eligiblePods: nodeNameIsEligible("node1"), + nodeNames: []string{"node1", "node2", "node3"}, + pod: st.MakePod().Name("p3").UID("p3").Priority(lowPriority).Req(largeRes).Obj(), + pods: []*v1.Pod{ + st.MakePod().Name("v1").UID("v1").Node("node1").Priority(highPriority).Req(smallRes).StartTime(epochTime).Obj(), + st.MakePod().Name("v2").UID("v2").Node("node1").Priority(midPriority).Req(smallRes).StartTime(epochTime).Obj(), + st.MakePod().Name("v3").UID("v3").Node("node1").Priority(lowPriority).Req(smallRes).StartTime(epochTime).Obj(), + st.MakePod().Name("v4").UID("v4").Node("node2").Priority(midPriority).Req(largeRes).StartTime(epochTime).Obj(), + st.MakePod().Name("v5").UID("v5").Node("node3").Priority(lowPriority).Req(largeRes).StartTime(epochTime).Obj(), + }, + expected: map[string][]string{}, + }, + { + name: "only pods at or above specified priority can preempted: high priority", + eligiblePods: priorityBelowThresholdCannotPreempt(highPriority), + nodeNames: []string{"node1", "node2", "node3"}, + pod: st.MakePod().Name("p1").UID("p1").Priority(highPriority).Req(largeRes).Obj(), + pods: []*v1.Pod{ + st.MakePod().Name("v1").UID("v1").Node("node1").Priority(highPriority).Req(largeRes).StartTime(epochTime).Obj(), + st.MakePod().Name("v2").UID("v2").Node("node2").Priority(midPriority).Req(largeRes).StartTime(epochTime).Obj(), + st.MakePod().Name("v3").UID("v3").Node("node3").Priority(lowPriority).Req(largeRes).StartTime(epochTime).Obj(), + }, + // highPriority can preempt anything, but not other highPriority + expected: map[string][]string{"node2": {"v2"}, "node3": {"v3"}}, + }, + { + name: "only pods at or above specified priority can preempted: mid priority", + eligiblePods: priorityBelowThresholdCannotPreempt(highPriority), + nodeNames: []string{"node1", "node2", "node3"}, + pod: st.MakePod().Name("p2").UID("p2").Priority(midPriority).Req(largeRes).Obj(), + pods: []*v1.Pod{ + st.MakePod().Name("v1").UID("v1").Node("node1").Priority(highPriority).Req(largeRes).StartTime(epochTime).Obj(), + st.MakePod().Name("v2").UID("v2").Node("node2").Priority(midPriority).Req(largeRes).StartTime(epochTime).Obj(), + st.MakePod().Name("v3").UID("v3").Node("node3").Priority(lowPriority).Req(largeRes).StartTime(epochTime).Obj(), + }, + // midPriority can't preempt anything + expected: map[string][]string{}, + }, + { + name: "only pods at or below specified priority can be preempted: high priority", + eligiblePods: priorityAboveThresholdCannotBePreempted(midPriority), + nodeNames: []string{"node1", "node2", "node3"}, + pod: st.MakePod().Name("p1").UID("p1").Priority(highPriority).Req(largeRes).Obj(), + pods: []*v1.Pod{ + st.MakePod().Name("v1").UID("v1").Node("node1").Priority(highPriority).Req(largeRes).StartTime(epochTime).Obj(), + st.MakePod().Name("v2").UID("v2").Node("node2").Priority(midPriority).Req(largeRes).StartTime(epochTime).Obj(), + st.MakePod().Name("v3").UID("v3").Node("node3").Priority(lowPriority).Req(largeRes).StartTime(epochTime).Obj(), + }, + // the lowPriority pod can be preempted but not the midPriority pod + expected: map[string][]string{"node2": {"v2"}, "node3": {"v3"}}, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + nodes := make([]*v1.Node, len(tt.nodeNames)) + for i, nodeName := range tt.nodeNames { + nodes[i] = st.MakeNode().Name(nodeName).Capacity(veryLargeRes).Obj() + } + + var objs []runtime.Object + objs = append(objs, tt.pod) + for _, pod := range tt.pods { + objs = append(objs, pod) + } + cs := clientsetfake.NewClientset(objs...) + informerFactory := informers.NewSharedInformerFactory(cs, 0) + snapshot := internalcache.NewSnapshot(tt.pods, nodes) + logger, ctx := ktesting.NewTestContext(t) + ctx, cancel := context.WithCancel(ctx) + defer cancel() + fwk, err := tf.NewFramework( + ctx, + []tf.RegisterPluginFunc{ + tf.RegisterPluginAsExtensions(noderesources.Name, nodeResourcesFitFunc, "Filter", "PreFilter"), + tf.RegisterQueueSortPlugin(queuesort.Name, queuesort.New), + tf.RegisterBindPlugin(defaultbinder.Name, defaultbinder.New), + }, + "", + frameworkruntime.WithPodNominator(internalqueue.NewSchedulingQueue(nil, informerFactory)), + frameworkruntime.WithSnapshotSharedLister(snapshot), + frameworkruntime.WithInformerFactory(informerFactory), + frameworkruntime.WithLogger(logger), + ) + if err != nil { + t.Fatal(err) + } + + state := framework.NewCycleState() + // Some tests rely on PreFilter plugin to compute its CycleState. + if _, status, _ := fwk.RunPreFilterPlugins(ctx, state, tt.pod); !status.IsSuccess() { + t.Errorf("Unexpected PreFilter Status: %v", status) + } + nodeInfos, err := snapshot.NodeInfos().List() + if err != nil { + t.Fatal(err) + } + + pl, err := New(ctx, getDefaultDefaultPreemptionArgs(), fwk, feature.Features{}) + if err != nil { + t.Fatal(err) + } + // Override eligibility logic + if tt.eligiblePods != nil { + pl.IsEligiblePod = tt.eligiblePods + } + offset, numCandidates := pl.GetOffsetAndNumCandidates(int32(len(nodeInfos))) + candidates, _, _ := pl.Evaluator.DryRunPreemption(ctx, state, tt.pod, nodeInfos, nil, offset, numCandidates) + // check that the candidates match what's expected + if len(tt.expected) != len(candidates) { + candidateNames := []string{} + for _, c := range candidates { + candidateNames = append(candidateNames, c.Name()) + } + t.Fatalf("expected %d candidates (%+v) but got %d: %+v", len(tt.expected), tt.expected, len(candidates), candidateNames) + } + for len(candidates) > 0 { + selected := pl.Evaluator.SelectCandidate(ctx, candidates) + + expectVictims, ok := tt.expected[selected.Name()] + if !ok { + t.Fatalf("got unexpected candidate %+v, when expected is %+v", selected, tt.expected) + } + + gotVictims := []string{} + for _, p := range selected.Victims().Pods { + gotVictims = append(gotVictims, p.Name) + } + if diff := cmp.Diff(expectVictims, gotVictims); diff != "" { + t.Errorf("Unexpected victims on node %s (-want,+got):\n%s", selected.Name(), diff) + } + + // remove selected from candidates + notSelected := []preemption.Candidate{} + for _, c := range candidates { + if c.Name() != selected.Name() { + notSelected = append(notSelected, c) + } + } + candidates = notSelected + } + }) + } +} + +func TestCustomOrdering(t *testing.T) { + // Two arbitrary examples of custom selection ordering to check that they behave as expected + orderByOldestStart := func(pod1, pod2 *v1.Pod) bool { + return util.GetPodStartTime(pod1).Before(util.GetPodStartTime(pod2)) + } + orderByPodName := func(pod1, pod2 *v1.Pod) bool { + return pod1.Name < pod2.Name + } + + tests := []struct { + name string + orderPods MoreImportantPodFunc + nodeNames []string + pod *v1.Pod + pods []*v1.Pod + expectedPods []string + }{ + { + name: "select newest pods", + orderPods: orderByOldestStart, + nodeNames: []string{"node1"}, + pod: st.MakePod().Name("p2").UID("p2").Priority(highPriority).Req(largeRes).Obj(), + // size victims to require at least two to be preempted + pods: []*v1.Pod{ + st.MakePod().Name("v1").UID("v1").Node("node1").Priority(lowPriority).Req(mediumRes).StartTime(epochTime2).Obj(), + st.MakePod().Name("v2").UID("v2").Node("node1").Priority(lowPriority).Req(mediumRes).StartTime(epochTime).Obj(), + st.MakePod().Name("v3").UID("v3").Node("node1").Priority(midPriority).Req(mediumRes).StartTime(epochTime1).Obj(), + }, + // the newest two pods are selected, despite one with higher priority + expectedPods: []string{"v3", "v1"}, + }, + { + name: "select alphabetically-last pods", + orderPods: orderByPodName, + nodeNames: []string{"node1"}, + pod: st.MakePod().Name("p2").UID("p2").Priority(highPriority).Req(largeRes).Obj(), + // size victims to require at least two to be preempted + pods: []*v1.Pod{ + st.MakePod().Name("foo").UID("v1").Node("node1").Priority(lowPriority).Req(mediumRes).StartTime(epochTime).Obj(), + st.MakePod().Name("bar").UID("v2").Node("node1").Priority(lowPriority).Req(mediumRes).StartTime(epochTime).Obj(), + st.MakePod().Name("baz").UID("v3").Node("node1").Priority(midPriority).Req(mediumRes).StartTime(epochTime).Obj(), + }, + // the last pods in alphabetic order are selected, despite one with higher priority + expectedPods: []string{"baz", "foo"}, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + nodes := make([]*v1.Node, len(tt.nodeNames)) + for i, nodeName := range tt.nodeNames { + nodes[i] = st.MakeNode().Name(nodeName).Capacity(veryLargeRes).Obj() + } + + var objs []runtime.Object + objs = append(objs, tt.pod) + for _, pod := range tt.pods { + objs = append(objs, pod) + } + cs := clientsetfake.NewClientset(objs...) + informerFactory := informers.NewSharedInformerFactory(cs, 0) + snapshot := internalcache.NewSnapshot(tt.pods, nodes) + logger, ctx := ktesting.NewTestContext(t) + ctx, cancel := context.WithCancel(ctx) + defer cancel() + fwk, err := tf.NewFramework( + ctx, + []tf.RegisterPluginFunc{ + tf.RegisterPluginAsExtensions(noderesources.Name, nodeResourcesFitFunc, "Filter", "PreFilter"), + tf.RegisterQueueSortPlugin(queuesort.Name, queuesort.New), + tf.RegisterBindPlugin(defaultbinder.Name, defaultbinder.New), + }, + "", + frameworkruntime.WithPodNominator(internalqueue.NewSchedulingQueue(nil, informerFactory)), + frameworkruntime.WithSnapshotSharedLister(snapshot), + frameworkruntime.WithInformerFactory(informerFactory), + frameworkruntime.WithLogger(logger), + ) + if err != nil { + t.Fatal(err) + } + + state := framework.NewCycleState() + // Some tests rely on PreFilter plugin to compute its CycleState. + if _, status, _ := fwk.RunPreFilterPlugins(ctx, state, tt.pod); !status.IsSuccess() { + t.Errorf("Unexpected PreFilter Status: %v", status) + } + nodeInfos, err := snapshot.NodeInfos().List() + if err != nil { + t.Fatal(err) + } + + pl, err := New(ctx, getDefaultDefaultPreemptionArgs(), fwk, feature.Features{}) + if err != nil { + t.Fatal(err) + } + // Override ordering logic + if tt.orderPods != nil { + pl.MoreImportantPod = tt.orderPods + } + offset, numCandidates := pl.GetOffsetAndNumCandidates(int32(len(nodeInfos))) + candidates, _, _ := pl.Evaluator.DryRunPreemption(ctx, state, tt.pod, nodeInfos, nil, offset, numCandidates) + if len(candidates) != 1 { + t.Fatalf("expected exactly one node but got %+v", candidates) + } + podNames := []string{} + for _, p := range candidates[0].Victims().Pods { + podNames = append(podNames, p.Name) + } + if diff := cmp.Diff(tt.expectedPods, podNames); diff != "" { + t.Errorf("expect pods %+v, but got pods %+v", tt.expectedPods, podNames) + } + }) + } +} + func TestPodEligibleToPreemptOthers(t *testing.T) { tests := []struct { name string @@ -1534,18 +1877,29 @@ 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} + pl, err := New(ctx, getDefaultDefaultPreemptionArgs(), f, feature.Features{}) + if err != nil { + t.Fatal(err) + } 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 +2199,13 @@ 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, + } + pl, err := New(ctx, getDefaultDefaultPreemptionArgs(), fwk, features) + if err != nil { + t.Fatal(err) } - - 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 +2215,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 +2288,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/framework/runtime/registry.go b/pkg/scheduler/framework/runtime/registry.go index b6fca3c29cd..8c5901ae52a 100644 --- a/pkg/scheduler/framework/runtime/registry.go +++ b/pkg/scheduler/framework/runtime/registry.go @@ -31,11 +31,11 @@ import ( type PluginFactory = func(ctx context.Context, configuration runtime.Object, f framework.Handle) (framework.Plugin, error) // PluginFactoryWithFts is a function that builds a plugin with certain feature gates. -type PluginFactoryWithFts func(context.Context, runtime.Object, framework.Handle, plfeature.Features) (framework.Plugin, error) +type PluginFactoryWithFts[T framework.Plugin] func(context.Context, runtime.Object, framework.Handle, plfeature.Features) (T, error) // FactoryAdapter can be used to inject feature gates for a plugin that needs // them when the caller expects the older PluginFactory method. -func FactoryAdapter(fts plfeature.Features, withFts PluginFactoryWithFts) PluginFactory { +func FactoryAdapter[T framework.Plugin](fts plfeature.Features, withFts PluginFactoryWithFts[T]) PluginFactory { return func(ctx context.Context, plArgs runtime.Object, fh framework.Handle) (framework.Plugin, error) { return withFts(ctx, plArgs, fh, fts) } 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)