github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2026-07-18 13:57:38 +00:00
Code Issues Projects Releases Wiki Activity

Merge pull request #129983 from nickbp/master

Browse Source 
feature(scheduler): Customizable pod selection and ordering in DefaultPreemption plugin
This commit is contained in:
Kubernetes Prow Robot
2025-05-20 05:09:15 -07:00
committed by GitHub
parent b20fca7484 7bccb1acb5
commit 0afe2b839d
4 changed files with 468 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files

87
pkg/scheduler/framework/plugins/defaultpreemption/default_preemption.go
View File

@@ -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 {

443
pkg/scheduler/framework/plugins/defaultpreemption/default_preemption_test.go
View File

@@ -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 <state> 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())
}

4
pkg/scheduler/framework/runtime/registry.go
View File

@@ -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)
}

5
pkg/scheduler/util/utils.go
View File

@@ -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)