Remove group priority lookups and nil-lister forks from preemption (#142951)

* Remove group priority lookups and nil-lister forks from preemption

* Address comments from code review
This commit is contained in:
Antoni Zawodny authored and GitHub committed 2026-10-20 00:33:08 +00:00
1 parent 8820bafbb1
commit c905fddc59
14 files changed
+140 -576

No files matched your search

@@ -75,8 +75,6 @@ type DefaultPreemption struct {
Executor fwk.PreemptionExecutor
Evaluator *preemption.Evaluator
pgLister fwk.PodGroupLister
pgSnapshotLister fwk.PodGroupLister
cpgSnapshotLister fwk.CompositePodGroupLister
podGroupEvaluator podGroupEvaluator
// IsEligiblePod returns whether a victim (individual pod/pod group) is allowed to be preempted by a preemptor pod.
@@ -124,14 +122,9 @@ func New(_ context.Context, dpArgs runtime.Object, fh fwk.Handle, fts feature.Fe
if pl.fts.EnableGenericWorkload {
pl.pgLister = fh.PodGroupManager().PodGroups()
pl.pgSnapshotLister = fh.SnapshotSharedLister().PodGroups()
pl.podGroupEvaluator = preemption.NewPodGroupEvaluator(fh)
}
if pl.fts.EnableCompositePodGroup {
pl.cpgSnapshotLister = fh.SnapshotSharedLister().CompositePodGroups()
}
// Default behavior: No additional filtering, beyond the internal requirement that the victim
// has lower priority than the preemptor.
pl.IsEligiblePod = func(nodeInfo fwk.NodeInfo, victim preemption.Victim, preemptor *v1.Pod) bool {
@@ -447,7 +440,7 @@ func (pl *DefaultPreemption) PodEligibleToPreemptOthers(_ context.Context, pod *
if nodeInfo, _ := nodeInfos.Get(nomNodeName); nodeInfo != nil {
for _, p := range nodeInfo.GetPods() {
victim := preemption.NewPodVictim(p, pl.pgSnapshotLister, pl.cpgSnapshotLister)
victim := preemption.NewPodVictim(p)
if pl.isPreemptionAllowed(nodeInfo, victim, pod) && preemption.PodTerminatingByPreemption(p.GetPod()) {
// There is a terminating pod on the nominated node.
return false, "not eligible due to a terminating pod on the nominated node."
@@ -2479,6 +2479,22 @@ func TestPodEligibleToPreemptOthers(t *testing.T) {
nodes: []string{"node1"},
expected: false,
},
{
name: "pod terminating by preemption on the nominated node with higher priority than the preemptor",
pod: st.MakePod().Name("p_with_nominated_node").UID("p").Priority(highPriority).NominatedNodeName("node1").Obj(),
pods: []*v1.Pod{st.MakePod().Name("p1").UID("p1").Priority(veryHighPriority).Node("node1").Terminating().
Condition(v1.DisruptionTarget, v1.ConditionTrue, v1.PodReasonPreemptionByScheduler).Obj()},
nodes: []string{"node1"},
expected: true,
},
{
name: "pod terminating by preemption on the nominated node with the same priority as the preemptor",
pod: st.MakePod().Name("p_with_nominated_node").UID("p").Priority(highPriority).NominatedNodeName("node1").Obj(),
pods: []*v1.Pod{st.MakePod().Name("p1").UID("p1").Priority(highPriority).Node("node1").Terminating().
Condition(v1.DisruptionTarget, v1.ConditionTrue, v1.PodReasonPreemptionByScheduler).Obj()},
nodes: []string{"node1"},
expected: true,
},
{
name: "non-victim Pods terminating",
pod: st.MakePod().Name("p_with_nominated_node").UID("p").Priority(highPriority).NominatedNodeName("node1").Obj(),
@@ -2487,10 +2503,10 @@ func TestPodEligibleToPreemptOthers(t *testing.T) {
expected: true,
},
{
name: "Preemptor with SchedulingGroup, victim priority overridden by PodGroup in victim namespace",
name: "Preemptor with SchedulingGroup, terminating victim is judged by its own priority, which matches its PodGroup's",
pod: st.MakePod().Name("p").UID("p").Namespace("ns1").Priority(highPriority).PodGroupName("pg1").NominatedNodeName("node1").Obj(),
pods: []*v1.Pod{
st.MakePod().Name("v1").UID("v1").Namespace("ns2").Node("node1").Priority(veryHighPriority).PodGroupName("pg1").Terminating().
st.MakePod().Name("v1").UID("v1").Namespace("ns2").Node("node1").Priority(lowPriority).PodGroupName("pg1").Terminating().
Condition(v1.DisruptionTarget, v1.ConditionTrue, v1.PodReasonPreemptionByScheduler).Obj(),
},
podGroups: []*v1beta1.PodGroup{
@@ -2500,66 +2516,6 @@ func TestPodEligibleToPreemptOthers(t *testing.T) {
features: feature.Features{EnableGenericWorkload: true},
expected: false,
},
{
name: "Preemptor with SchedulingGroup, victim priority overridden by CompositePodGroup in victim namespace",
pod: st.MakePod().Name("p").UID("p").Namespace("ns1").Priority(highPriority).PodGroupName("pg1").NominatedNodeName("node1").Obj(),
pods: []*v1.Pod{
st.MakePod().Name("v1").UID("v1").Namespace("ns2").Node("node1").Priority(veryHighPriority).PodGroupName("pg1").Terminating().
Condition(v1.DisruptionTarget, v1.ConditionTrue, v1.PodReasonPreemptionByScheduler).Obj(),
},
podGroups: []*v1beta1.PodGroup{
st.MakePodGroup().Name("pg1").UID("pg1").Namespace("ns2").ParentCompositePodGroup("cpg1").Priority(lowPriority).Obj(),
},
compositePodGroups: []*v1alpha3.CompositePodGroup{
st.MakeCompositePodGroup().Name("cpg1").UID("cpg1").Namespace("ns2").Priority(lowPriority).Obj(),
},
nodes: []string{"node1"},
features: feature.Features{EnableGenericWorkload: true, EnableCompositePodGroup: true},
expected: false,
},
{
name: "Preemptor with SchedulingGroup, CompositePodGroup not found in victim namespace, fallback to PodGroup priority",
pod: st.MakePod().Name("p").UID("p").Namespace("ns1").Priority(highPriority).PodGroupName("pg1").NominatedNodeName("node1").Obj(),
pods: []*v1.Pod{
st.MakePod().Name("v1").UID("v1").Namespace("ns2").Node("node1").Priority(veryHighPriority).PodGroupName("pg1").Terminating().
Condition(v1.DisruptionTarget, v1.ConditionTrue, v1.PodReasonPreemptionByScheduler).Obj(),
},
podGroups: []*v1beta1.PodGroup{
st.MakePodGroup().Name("pg1").UID("pg1").Namespace("ns2").ParentCompositePodGroup("cpg1").Priority(lowPriority).Obj(),
},
nodes: []string{"node1"},
features: feature.Features{EnableGenericWorkload: true, EnableCompositePodGroup: true},
expected: false,
},
{
name: "Preemptor with SchedulingGroup, PodGroup not found in victim namespace, fallback to pod priority when CompositePodGroup is enabled",
pod: st.MakePod().Name("p").UID("p").Namespace("ns1").Priority(highPriority).PodGroupName("pg1").NominatedNodeName("node1").Obj(),
pods: []*v1.Pod{
st.MakePod().Name("v1").UID("v1").Namespace("ns2").Node("node1").Priority(veryHighPriority).PodGroupName("pg1").Terminating().
Condition(v1.DisruptionTarget, v1.ConditionTrue, v1.PodReasonPreemptionByScheduler).Obj(),
},
nodes: []string{"node1"},
features: feature.Features{EnableGenericWorkload: true, EnableCompositePodGroup: true},
expected: true,
},
{
name: "Preemptor with SchedulingGroup, victim priority overridden by root CompositePodGroup in multi-level CPG hierarchy",
pod: st.MakePod().Name("p").UID("p").Namespace("ns1").Priority(highPriority).PodGroupName("pg1").NominatedNodeName("node1").Obj(),
pods: []*v1.Pod{
st.MakePod().Name("v1").UID("v1").Namespace("ns2").Node("node1").Priority(veryHighPriority).PodGroupName("leaf-pg").Terminating().
Condition(v1.DisruptionTarget, v1.ConditionTrue, v1.PodReasonPreemptionByScheduler).Obj(),
},
podGroups: []*v1beta1.PodGroup{
st.MakePodGroup().Name("leaf-pg").UID("leaf-pg").Namespace("ns2").ParentCompositePodGroup("child-cpg").Priority(veryHighPriority).Obj(),
},
compositePodGroups: []*v1alpha3.CompositePodGroup{
st.MakeCompositePodGroup().Name("child-cpg").UID("child-cpg").Namespace("ns2").ParentCompositePodGroup("root-cpg").Priority(veryHighPriority).Obj(),
st.MakeCompositePodGroup().Name("root-cpg").UID("root-cpg").Namespace("ns2").Priority(lowPriority).Obj(),
},
nodes: []string{"node1"},
features: feature.Features{EnableGenericWorkload: true, EnableCompositePodGroup: true},
expected: false,
},
}
for _, test := range tests {
@@ -385,18 +385,13 @@ func (e *Executor) IsPodGroupWaitingForVictims(pgInfo fwk.PodGroupInfo) bool {
preemptorPriority := pgInfo.GetPriority()
nodeLister := e.fh.SnapshotSharedLister().NodeInfos()
pgs := e.fh.SnapshotSharedLister().PodGroups()
var cpgs fwk.CompositePodGroupLister
if e.fts.EnableCompositePodGroup {
cpgs = e.fh.SnapshotSharedLister().CompositePodGroups()
}
for nomNodeName := range nominatedNodes {
nodeInfo, err := nodeLister.Get(nomNodeName)
if err != nil || nodeInfo == nil {
continue
}
for _, p := range nodeInfo.GetPods() {
if getPodPriority(p.GetPod(), pgs, cpgs) < preemptorPriority && PodTerminatingByPreemption(p.GetPod()) {
if corev1helpers.PodPriority(p.GetPod()) < preemptorPriority && PodTerminatingByPreemption(p.GetPod()) {
return true
}
}
@@ -2263,13 +2263,11 @@ func TestIsPodGroupWaitingForVictims(t *testing.T) {
}
tests := []struct {
name string
enableCompositePodGroup bool
initPods []*v1.Pod
initPodGroups []*schedulingv1beta1.PodGroup
initCompositePodGroups []*schedulingv1alpha3.CompositePodGroup
preemptor fwk.PodGroupInfo
want bool
name string
initPods []*v1.Pod
initPodGroups []*schedulingv1beta1.PodGroup
preemptor fwk.PodGroupInfo
want bool
}{
{
name: "returns true when member has nominated node with lower-priority pod terminating by preemption",
@@ -2340,9 +2338,9 @@ func TestIsPodGroupWaitingForVictims(t *testing.T) {
want: false,
},
{
name: "returns true when terminating pod on nominated node belongs to lower-priority PodGroup with DisruptionModeAll",
name: "returns true when terminating pod on nominated node belongs to a lower-priority PodGroup with DisruptionModeAll",
initPods: []*v1.Pod{
makeTerminatingVictim("victim", "node1", highPriority).PodGroupName("victim-pg").Obj(),
makeTerminatingVictim("victim", "node1", lowPriority).PodGroupName("victim-pg").Obj(),
},
initPodGroups: []*schedulingv1beta1.PodGroup{
st.MakePodGroup().Name("victim-pg").Namespace("default").UID("victim-pg").Priority(lowPriority).DisruptionModeAll().Obj(),
@@ -2351,9 +2349,9 @@ func TestIsPodGroupWaitingForVictims(t *testing.T) {
want: true,
},
{
name: "returns true when terminating pod on nominated node belongs to lower-priority PodGroup with DisruptionModeSingle",
name: "returns true when terminating pod on nominated node belongs to a lower-priority PodGroup with DisruptionModeSingle",
initPods: []*v1.Pod{
makeTerminatingVictim("victim", "node1", highPriority).PodGroupName("victim-pg").Obj(),
makeTerminatingVictim("victim", "node1", lowPriority).PodGroupName("victim-pg").Obj(),
},
initPodGroups: []*schedulingv1beta1.PodGroup{
st.MakePodGroup().Name("victim-pg").Namespace("default").UID("victim-pg").Priority(lowPriority).DisruptionModeSingle().Obj(),
@@ -2362,9 +2360,9 @@ func TestIsPodGroupWaitingForVictims(t *testing.T) {
want: true,
},
{
name: "returns false when terminating pod on nominated node belongs to higher-priority PodGroup with DisruptionModeAll",
name: "returns false when terminating pod on nominated node belongs to a higher-priority PodGroup with DisruptionModeAll",
initPods: []*v1.Pod{
makeTerminatingVictim("victim", "node1", lowPriority).PodGroupName("victim-pg").Obj(),
makeTerminatingVictim("victim", "node1", highPriority).PodGroupName("victim-pg").Obj(),
},
initPodGroups: []*schedulingv1beta1.PodGroup{
st.MakePodGroup().Name("victim-pg").Namespace("default").UID("victim-pg").Priority(highPriority).DisruptionModeAll().Obj(),
@@ -2373,9 +2371,9 @@ func TestIsPodGroupWaitingForVictims(t *testing.T) {
want: false,
},
{
name: "returns false when terminating pod on nominated node belongs to higher-priority PodGroup with DisruptionModeSingle",
name: "returns false when terminating pod on nominated node belongs to a higher-priority PodGroup with DisruptionModeSingle",
initPods: []*v1.Pod{
makeTerminatingVictim("victim", "node1", lowPriority).PodGroupName("victim-pg").Obj(),
makeTerminatingVictim("victim", "node1", highPriority).PodGroupName("victim-pg").Obj(),
},
initPodGroups: []*schedulingv1beta1.PodGroup{
st.MakePodGroup().Name("victim-pg").Namespace("default").UID("victim-pg").Priority(highPriority).DisruptionModeSingle().Obj(),
@@ -2383,63 +2381,16 @@ func TestIsPodGroupWaitingForVictims(t *testing.T) {
preemptor: makePreemptor(midPriority, "node1"),
want: false,
},
{
name: "returns true when victim uses its hierarchy's lower priority (CompositePodGroup root overrides higher pod and child PodGroup priority)",
enableCompositePodGroup: true,
initPods: []*v1.Pod{
makeTerminatingVictim("victim", "node1", highPriority).PodGroupName("victim-pg").Obj(),
},
initPodGroups: []*schedulingv1beta1.PodGroup{
st.MakePodGroup().Name("victim-pg").Namespace("default").UID("victim-pg").Priority(highPriority).ParentCompositePodGroup("victim-cpg").Obj(),
},
initCompositePodGroups: []*schedulingv1alpha3.CompositePodGroup{
st.MakeCompositePodGroup().Name("victim-cpg").Namespace("default").UID("victim-cpg").Priority(lowPriority).Obj(),
},
preemptor: makePreemptor(midPriority, "node1"),
want: true,
},
{
name: "returns false when victim uses its hierarchy's higher priority (CompositePodGroup root overrides lower pod and child PodGroup priority)",
enableCompositePodGroup: true,
initPods: []*v1.Pod{
makeTerminatingVictim("victim", "node1", lowPriority).PodGroupName("victim-pg").Obj(),
},
initPodGroups: []*schedulingv1beta1.PodGroup{
st.MakePodGroup().Name("victim-pg").Namespace("default").UID("victim-pg").Priority(lowPriority).ParentCompositePodGroup("victim-cpg").Obj(),
},
initCompositePodGroups: []*schedulingv1alpha3.CompositePodGroup{
st.MakeCompositePodGroup().Name("victim-cpg").Namespace("default").UID("victim-cpg").Priority(highPriority).Obj(),
},
preemptor: makePreemptor(midPriority, "node1"),
want: false,
},
{
name: "returns false when CompositePodGroup is disabled and child PodGroup has higher priority than preemptor",
enableCompositePodGroup: false,
initPods: []*v1.Pod{
makeTerminatingVictim("victim", "node1", highPriority).PodGroupName("victim-pg").Obj(),
},
initPodGroups: []*schedulingv1beta1.PodGroup{
st.MakePodGroup().Name("victim-pg").Namespace("default").UID("victim-pg").Priority(highPriority).ParentCompositePodGroup("victim-cpg").Obj(),
},
initCompositePodGroups: []*schedulingv1alpha3.CompositePodGroup{
st.MakeCompositePodGroup().Name("victim-cpg").Namespace("default").UID("victim-cpg").Priority(lowPriority).Obj(),
},
preemptor: makePreemptor(midPriority, "node1"),
want: false,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
featuregatetesting.SetFeatureGatesDuringTest(t, utilfeature.DefaultFeatureGate, featuregatetesting.FeatureOverrides{
features.GenericWorkload: true,
features.TopologyAwareWorkloadScheduling: true,
features.CompositePodGroup: tt.enableCompositePodGroup,
features.GenericWorkload: true,
})
client := clientsetfake.NewSimpleClientset()
informerFactory := informers.NewSharedInformerFactory(client, 0)
snapshot := internalcache.NewTestSnapshotWithPodGroups(tt.initPods, nodes, tt.initPodGroups, tt.initCompositePodGroups)
snapshot := internalcache.NewTestSnapshotWithPodGroups(tt.initPods, nodes, tt.initPodGroups, nil)
fh := &fakeHandleForLister{
informerFactory: informerFactory,
snapshot: snapshot,
@@ -61,12 +61,7 @@ func (m *defaultPreemptionManager) GenerateVictims(ctx context.Context, pgInfo f
return nil, fwk.NewStatus(fwk.Unschedulable, msg)
}
podGroupSnapshot := m.snapshot.PodGroups()
var compositePodGroupSnapshot fwk.CompositePodGroupLister
if m.fts.EnableCompositePodGroup {
compositePodGroupSnapshot = m.snapshot.CompositePodGroups()
}
victims, err := getWorkloadPreemptionVictims(logger, m.snapshot, podGroupSnapshot, compositePodGroupSnapshot)
victims, err := getWorkloadPreemptionVictims(logger, m.snapshot)
if err != nil {
return nil, fwk.AsStatus(fmt.Errorf("failed to get victims: %w", err))
}
@@ -264,9 +264,9 @@ func TestDefaultPreemptionManager_GenerateVictims(t *testing.T) {
},
},
{
name: "ordering by priority: group hierarchy root priority overrides individual pod priority",
name: "ordering by priority: pod group members are ordered by their own priority, which matches their group's",
initPods: []*v1.Pod{
makeGroupPod("low-pod-in-mid-pg", lowPriority, "pg-mid"),
makeGroupPod("mid-pod-in-mid-pg", midPriority, "pg-mid"),
makePod("standalone-low", lowPriority),
},
initPodGroups: []*schedulingv1beta1.PodGroup{
@@ -275,25 +275,7 @@ func TestDefaultPreemptionManager_GenerateVictims(t *testing.T) {
preemptor: makePreemptor(highPriority),
wantVictims: []wantVictim{
wantVic("standalone-low"),
wantVic("low-pod-in-mid-pg"),
},
},
{
name: "ordering by priority: CompositePodGroup root priority overrides child PodGroup and pod priority",
initPods: []*v1.Pod{
makeGroupPod("pod-in-cpg", lowPriority, "pg-low"),
makePod("standalone-low", lowPriority),
},
initPodGroups: []*schedulingv1beta1.PodGroup{
makePG("pg-low", lowPriority, false, "cpg-mid"),
},
initCompositePodGroups: []*schedulingv1alpha3.CompositePodGroup{
makeCPG("cpg-mid", midPriority, false),
},
preemptor: makePreemptor(highPriority),
wantVictims: []wantVictim{
wantVic("standalone-low"),
wantVic("pod-in-cpg"),
wantVic("mid-pod-in-mid-pg"),
},
},
{
@@ -346,7 +328,7 @@ func TestDefaultPreemptionManager_GenerateVictims(t *testing.T) {
},
},
{
name: "feature gates: CompositePodGroup disabled ignores parent CompositePodGroup disruption mode and priority",
name: "feature gates: CompositePodGroup disabled ignores parent CompositePodGroup disruption mode",
featureGates: featuregatetesting.FeatureOverrides{
features.CompositePodGroup: false,
},
@@ -355,13 +337,12 @@ func TestDefaultPreemptionManager_GenerateVictims(t *testing.T) {
makeGroupPod("p2", lowPriority, "pg-child"),
},
initPodGroups: []*schedulingv1beta1.PodGroup{
// pg-child has lowPriority (eligible for preemption) and DisruptionModeSingle (individual victims).
// pg-child has DisruptionModeSingle, so its pods are individual victims.
makePG("pg-child", lowPriority, false, "cpg-root"),
},
initCompositePodGroups: []*schedulingv1alpha3.CompositePodGroup{
// If enabled, cpg-root's highPriority would protect p1/p2 from highPriority preemptor,
// and DisruptionModeAll would group them together.
makeCPG("cpg-root", highPriority, true),
// If enabled, cpg-root's DisruptionModeAll would group p1/p2 into one atomic victim.
makeCPG("cpg-root", lowPriority, true),
},
preemptor: makePreemptor(highPriority),
wantVictims: []wantVictim{
@@ -436,7 +436,7 @@ func TestPodGroupEvaluator_Preempt_Victims(t *testing.T) {
st.MakeNode().Name("node2").Obj(),
},
initPods: []*v1.Pod{
st.MakePod().Name("victim").UID("v1").Node("node1").Priority(lowPriority).PodGroupName("victim-pg").Condition(v1.DisruptionTarget, v1.ConditionTrue, v1.PodReasonPreemptionByScheduler).Terminating().Obj(),
st.MakePod().Name("victim").UID("v1").Node("node1").Priority(highPriority).PodGroupName("victim-pg").Condition(v1.DisruptionTarget, v1.ConditionTrue, v1.PodReasonPreemptionByScheduler).Terminating().Obj(),
st.MakePod().Name("other-victim").UID("v2").Node("node1").Priority(lowPriority).Obj(),
},
initPodGroups: []*schedulingv1beta1.PodGroup{
@@ -469,7 +469,7 @@ func TestPodGroupEvaluator_Preempt_Victims(t *testing.T) {
st.MakeNode().Name("node2").Obj(),
},
initPods: []*v1.Pod{
st.MakePod().Name("victim").UID("v1").Node("node1").Priority(lowPriority).PodGroupName("victim-pg").Condition(v1.DisruptionTarget, v1.ConditionTrue, v1.PodReasonPreemptionByScheduler).Terminating().Obj(),
st.MakePod().Name("victim").UID("v1").Node("node1").Priority(highPriority).PodGroupName("victim-pg").Condition(v1.DisruptionTarget, v1.ConditionTrue, v1.PodReasonPreemptionByScheduler).Terminating().Obj(),
st.MakePod().Name("other-victim").UID("v2").Node("node1").Priority(lowPriority).Obj(),
},
initPodGroups: []*schedulingv1beta1.PodGroup{
@@ -887,67 +887,6 @@ func TestPodGroupEvaluator_Preempt_Victims(t *testing.T) {
expectedVictims: []string{},
expectedStatus: fwk.NewStatus(fwk.Unschedulable),
},
{
name: "Priority divergence: candidate victim PodGroup has lower priority than the Pods from that group",
nodes: []*v1.Node{
st.MakeNode().Name("node1").Obj(),
st.MakeNode().Name("node2").Obj(),
},
initPods: []*v1.Pod{
st.MakePod().Name("p1").UID("v1").Node("node1").Priority(highPriority).PodGroupName("pg1").Obj(),
st.MakePod().Name("p2").UID("v2").Node("node2").Priority(highPriority).PodGroupName("pg1").Obj(),
},
initPodGroups: []*schedulingv1beta1.PodGroup{
st.MakePodGroup().Name("pg1").UID("pg1").DisruptionModeAll().Priority(midPriority).Obj(),
},
preemptor: makePodGroupPreemptor(
st.MakePodGroup().Name("preemptor-pg").Priority(highPriority).Obj(),
[]*v1.Pod{st.MakePod().Name("p").UID("p").Priority(highPriority).Obj()},
),
nodeCapacities: []nodeCapacity{
{
nodeName: "node1",
capacity: 1,
},
{
nodeName: "node2",
capacity: 1,
},
},
expectedVictims: []string{"p1", "p2"},
expectedStatus: fwk.NewStatus(fwk.Success),
expectedNumPodGroupDisruptions: 1,
},
{
name: "Priority divergence: candidate victim PodGroup has higher priority than the Pods from that group",
nodes: []*v1.Node{
st.MakeNode().Name("node1").Obj(),
st.MakeNode().Name("node2").Obj(),
},
initPods: []*v1.Pod{
st.MakePod().Name("p1").UID("v1").Node("node1").Priority(lowPriority).PodGroupName("pg1").Obj(),
st.MakePod().Name("p2").UID("v2").Node("node2").Priority(lowPriority).PodGroupName("pg1").Obj(),
},
initPodGroups: []*schedulingv1beta1.PodGroup{
st.MakePodGroup().Name("pg1").UID("pg1").DisruptionModeAll().Priority(midPriority).Obj(),
},
preemptor: makePodGroupPreemptor(
st.MakePodGroup().Name("preemptor-pg").Priority(midPriority).Obj(),
[]*v1.Pod{st.MakePod().Name("p").UID("p").Priority(midPriority).Obj()},
),
nodeCapacities: []nodeCapacity{
{
nodeName: "node1",
capacity: 1,
},
{
nodeName: "node2",
capacity: 1,
},
},
expectedVictims: []string{},
expectedStatus: fwk.NewStatus(fwk.Unschedulable),
},
{
name: "Gang scheduling: schedule as many pods as possible without preempting higher priority pods, but still more than minCount",
nodes: []*v1.Node{
@@ -1613,7 +1552,7 @@ func TestPodGroupEvaluator_Preempt(t *testing.T) {
st.MakeNode().Name("node1").Obj(),
},
initPods: []*v1.Pod{
st.MakePod().Name("victim").UID("v1").Node("node1").Priority(highPriority).PodGroupName("victim-pg").Condition(v1.DisruptionTarget, v1.ConditionTrue, v1.PodReasonPreemptionByScheduler).Terminating().Obj(),
st.MakePod().Name("victim").UID("v1").Node("node1").Priority(lowPriority).PodGroupName("victim-pg").Condition(v1.DisruptionTarget, v1.ConditionTrue, v1.PodReasonPreemptionByScheduler).Terminating().Obj(),
},
initPodGroups: []*schedulingv1beta1.PodGroup{
st.MakePodGroup().Name("victim-pg").UID("victim-pg").Priority(lowPriority).DisruptionModeAll().Obj(),
@@ -1635,7 +1574,7 @@ func TestPodGroupEvaluator_Preempt(t *testing.T) {
st.MakeNode().Name("node1").Obj(),
},
initPods: []*v1.Pod{
st.MakePod().Name("victim").UID("v1").Node("node1").Priority(highPriority).PodGroupName("victim-pg").Condition(v1.DisruptionTarget, v1.ConditionTrue, v1.PodReasonPreemptionByScheduler).Terminating().Obj(),
st.MakePod().Name("victim").UID("v1").Node("node1").Priority(lowPriority).PodGroupName("victim-pg").Condition(v1.DisruptionTarget, v1.ConditionTrue, v1.PodReasonPreemptionByScheduler).Terminating().Obj(),
},
initPodGroups: []*schedulingv1beta1.PodGroup{
st.MakePodGroup().Name("victim-pg").UID("victim-pg").Priority(lowPriority).DisruptionModeSingle().Obj(),
@@ -65,12 +65,10 @@ type Interface interface {
// Evaluator is a preemption evaluator. It runs preemption logic with a given Interface.
type Evaluator struct {
PluginName string
Handler fwk.Handle
PodLister corelisters.PodLister
PdbLister policylisters.PodDisruptionBudgetLister
podGroupSnapshot fwk.PodGroupLister
compositePodGroupSnapshot fwk.CompositePodGroupLister
PluginName string
Handler fwk.Handle
PodLister corelisters.PodLister
PdbLister policylisters.PodDisruptionBudgetLister
Interface
executor fwk.PreemptionExecutor
@@ -79,7 +77,7 @@ type Evaluator struct {
// NewEvaluator creates a new Evaluator.
func NewEvaluator(pluginName string, fh fwk.Handle, i Interface, fts feature.Features) *Evaluator {
ev := &Evaluator{
return &Evaluator{
PluginName: pluginName,
Handler: fh,
PodLister: fh.SharedInformerFactory().Core().V1().Pods().Lister(),
@@ -88,13 +86,6 @@ func NewEvaluator(pluginName string, fh fwk.Handle, i Interface, fts feature.Fea
executor: fh.PreemptionManager().Executor(),
fts: fts,
}
if fts.EnableGenericWorkload {
ev.podGroupSnapshot = fh.MutableSnapshotSharedLister().PodGroups()
}
if fts.EnableCompositePodGroup {
ev.compositePodGroupSnapshot = fh.MutableSnapshotSharedLister().CompositePodGroups()
}
return ev
}
// evaluate determines victims for preemption, without actuation.
@@ -530,22 +521,21 @@ func (ev *Evaluator) DryRunPreemption(ctx context.Context, state fwk.CycleState,
// (with disruption mode All) across the cluster into a single victim.
// If GenericWorkload is disabled, it treats each pod on the node as an individual victim.
func (ev *Evaluator) GetVictimsOnNode(ctx context.Context, nodeInfo fwk.NodeInfo) ([]*DomainVictim, error) {
fh := ev.Handler
snapshot := ev.Handler.MutableSnapshotSharedLister()
// If GenericWorkload is disabled, we treat each pod on the node
// as an individual preemption victim. We bypass getCrossNodesVictims and call NewPodVictim
// with a nil pgLister to force using the pod's own priority. Otherwise, if we used the
// pgLister, it would resolve to the PodGroup's priority, which might be nil (treated as 0)
// when GenericWorkload is disabled.
// With GenericWorkload disabled, pod groups are not a scheduling concept: every pod on the node
// is an individual PodKeyType victim, so skip the hierarchy lookups altogether.
if !ev.fts.EnableGenericWorkload {
var victims []Victim
for _, podInfo := range nodeInfo.GetPods() {
victims = append(victims, NewPodVictim(podInfo, nil, nil))
// NewVictim only fails for an empty pod list or key type, neither of which can happen here.
victim, _ := NewVictim([]fwk.PodInfo{podInfo}, corev1helpers.PodPriority(podInfo.GetPod()), fwk.PodKeyType)
victims = append(victims, victim)
}
return createDomainVictims(fh.MutableSnapshotSharedLister(), victims)
return createDomainVictims(snapshot, victims)
}
logger := klog.FromContext(ctx)
victims := getCrossNodesVictims(logger, fh.MutableSnapshotSharedLister(), ev.podGroupSnapshot, ev.compositePodGroupSnapshot, []fwk.NodeInfo{nodeInfo})
return createDomainVictims(fh.SnapshotSharedLister(), victims)
victims := getCrossNodesVictims(logger, snapshot, []fwk.NodeInfo{nodeInfo})
return createDomainVictims(snapshot, victims)
}
@@ -838,9 +838,9 @@ func TestGetVictimsOnNode(t *testing.T) {
st.MakePodGroup().Name("pg1").Namespace("default").Priority(midPriority).DisruptionModeSingle().Obj(),
},
pods: []*v1.Pod{
st.MakePod().Name("p1").UID("p1").Namespace("default").Node("node1").PodGroupName("pg1").Priority(lowPriority).Obj(),
st.MakePod().Name("p2").UID("p2").Namespace("default").Node("node2").PodGroupName("pg1").Priority(lowPriority).Obj(),
st.MakePod().Name("p3").UID("p3").Namespace("default").Node("node1").PodGroupName("pg1").Priority(lowPriority).Obj(),
st.MakePod().Name("p1").UID("p1").Namespace("default").Node("node1").PodGroupName("pg1").Priority(midPriority).Obj(),
st.MakePod().Name("p2").UID("p2").Namespace("default").Node("node2").PodGroupName("pg1").Priority(midPriority).Obj(),
st.MakePod().Name("p3").UID("p3").Namespace("default").Node("node1").PodGroupName("pg1").Priority(midPriority).Obj(),
st.MakePod().Name("p4").UID("p4").Namespace("default").Node("node1").Priority(lowPriority).Obj(),
},
targetNode: "node1",
@@ -849,7 +849,7 @@ func TestGetVictimsOnNode(t *testing.T) {
Victim: &victim{
priority: midPriority,
pods: []fwk.PodInfo{
newPodInfo(st.MakePod().Name("p1").UID("p1").Namespace("default").Node("node1").PodGroupName("pg1").Priority(lowPriority).Obj()),
newPodInfo(st.MakePod().Name("p1").UID("p1").Namespace("default").Node("node1").PodGroupName("pg1").Priority(midPriority).Obj()),
},
keyType: fwk.PodGroupKeyType,
},
@@ -858,7 +858,7 @@ func TestGetVictimsOnNode(t *testing.T) {
Victim: &victim{
priority: midPriority,
pods: []fwk.PodInfo{
newPodInfo(st.MakePod().Name("p3").UID("p3").Namespace("default").Node("node1").PodGroupName("pg1").Priority(lowPriority).Obj()),
newPodInfo(st.MakePod().Name("p3").UID("p3").Namespace("default").Node("node1").PodGroupName("pg1").Priority(midPriority).Obj()),
},
keyType: fwk.PodGroupKeyType,
},
@@ -888,8 +888,8 @@ func TestGetVictimsOnNode(t *testing.T) {
pods: []*v1.Pod{
st.MakePod().Name("p-all-1").UID("p-all-1").Namespace("default").Node("node1").PodGroupName("pg-all").Priority(highPriority).Obj(),
st.MakePod().Name("p-all-2").UID("p-all-2").Namespace("default").Node("node2").PodGroupName("pg-all").Priority(highPriority).Obj(),
st.MakePod().Name("p-single-1").UID("p-single-1").Namespace("default").Node("node1").PodGroupName("pg-single").Priority(lowPriority).Obj(),
st.MakePod().Name("p-single-2").UID("p-single-2").Namespace("default").Node("node2").PodGroupName("pg-single").Priority(lowPriority).Obj(),
st.MakePod().Name("p-single-1").UID("p-single-1").Namespace("default").Node("node1").PodGroupName("pg-single").Priority(midPriority).Obj(),
st.MakePod().Name("p-single-2").UID("p-single-2").Namespace("default").Node("node2").PodGroupName("pg-single").Priority(midPriority).Obj(),
st.MakePod().Name("p-ind").UID("p-ind").Namespace("default").Node("node1").Priority(lowPriority).Obj(),
},
targetNode: "node1",
@@ -908,7 +908,7 @@ func TestGetVictimsOnNode(t *testing.T) {
Victim: &victim{
priority: midPriority,
pods: []fwk.PodInfo{
newPodInfo(st.MakePod().Name("p-single-1").UID("p-single-1").Namespace("default").Node("node1").PodGroupName("pg-single").Priority(lowPriority).Obj()),
newPodInfo(st.MakePod().Name("p-single-1").UID("p-single-1").Namespace("default").Node("node1").PodGroupName("pg-single").Priority(midPriority).Obj()),
},
keyType: fwk.PodGroupKeyType,
},
@@ -936,12 +936,12 @@ func TestGetVictimsOnNode(t *testing.T) {
st.MakeCompositePodGroup().Name("cpg1").Namespace("default").Priority(midPriority).DisruptionModeAll().Obj(),
},
podGroups: []*schedulingv1beta1.PodGroup{
st.MakePodGroup().Name("pg1").Namespace("default").ParentCompositePodGroup("cpg1").Priority(lowPriority).Obj(),
st.MakePodGroup().Name("pg2").Namespace("default").ParentCompositePodGroup("cpg1").Priority(lowPriority).Obj(),
st.MakePodGroup().Name("pg1").Namespace("default").ParentCompositePodGroup("cpg1").Priority(midPriority).Obj(),
st.MakePodGroup().Name("pg2").Namespace("default").ParentCompositePodGroup("cpg1").Priority(midPriority).Obj(),
},
pods: []*v1.Pod{
st.MakePod().Name("p1").UID("p1").Namespace("default").Node("node1").PodGroupName("pg1").Priority(lowPriority).Obj(),
st.MakePod().Name("p2").UID("p2").Namespace("default").Node("node2").PodGroupName("pg2").Priority(lowPriority).Obj(),
st.MakePod().Name("p1").UID("p1").Namespace("default").Node("node1").PodGroupName("pg1").Priority(midPriority).Obj(),
st.MakePod().Name("p2").UID("p2").Namespace("default").Node("node2").PodGroupName("pg2").Priority(midPriority).Obj(),
},
targetNode: "node1",
expectedVictims: []*DomainVictim{
@@ -949,8 +949,8 @@ func TestGetVictimsOnNode(t *testing.T) {
Victim: &victim{
priority: midPriority,
pods: []fwk.PodInfo{
newPodInfo(st.MakePod().Name("p1").UID("p1").Namespace("default").Node("node1").PodGroupName("pg1").Priority(lowPriority).Obj()),
newPodInfo(st.MakePod().Name("p2").UID("p2").Namespace("default").Node("node2").PodGroupName("pg2").Priority(lowPriority).Obj()),
newPodInfo(st.MakePod().Name("p1").UID("p1").Namespace("default").Node("node1").PodGroupName("pg1").Priority(midPriority).Obj()),
newPodInfo(st.MakePod().Name("p2").UID("p2").Namespace("default").Node("node2").PodGroupName("pg2").Priority(midPriority).Obj()),
},
keyType: fwk.CompositePodGroupKeyType,
},
@@ -969,12 +969,12 @@ func TestGetVictimsOnNode(t *testing.T) {
st.MakeCompositePodGroup().Name("cpg1").Namespace("default").Priority(midPriority).DisruptionModeSingle().Obj(),
},
podGroups: []*schedulingv1beta1.PodGroup{
st.MakePodGroup().Name("pg1").Namespace("default").ParentCompositePodGroup("cpg1").Priority(lowPriority).Obj(),
st.MakePodGroup().Name("pg2").Namespace("default").ParentCompositePodGroup("cpg1").Priority(lowPriority).Obj(),
st.MakePodGroup().Name("pg1").Namespace("default").ParentCompositePodGroup("cpg1").Priority(midPriority).Obj(),
st.MakePodGroup().Name("pg2").Namespace("default").ParentCompositePodGroup("cpg1").Priority(midPriority).Obj(),
},
pods: []*v1.Pod{
st.MakePod().Name("p1").UID("p1").Namespace("default").Node("node1").PodGroupName("pg1").Priority(lowPriority).Obj(),
st.MakePod().Name("p2").UID("p2").Namespace("default").Node("node2").PodGroupName("pg2").Priority(lowPriority).Obj(),
st.MakePod().Name("p1").UID("p1").Namespace("default").Node("node1").PodGroupName("pg1").Priority(midPriority).Obj(),
st.MakePod().Name("p2").UID("p2").Namespace("default").Node("node2").PodGroupName("pg2").Priority(midPriority).Obj(),
},
targetNode: "node1",
expectedVictims: []*DomainVictim{
@@ -982,7 +982,7 @@ func TestGetVictimsOnNode(t *testing.T) {
Victim: &victim{
priority: midPriority,
pods: []fwk.PodInfo{
newPodInfo(st.MakePod().Name("p1").UID("p1").Namespace("default").Node("node1").PodGroupName("pg1").Priority(lowPriority).Obj()),
newPodInfo(st.MakePod().Name("p1").UID("p1").Namespace("default").Node("node1").PodGroupName("pg1").Priority(midPriority).Obj()),
},
keyType: fwk.PodGroupKeyType,
},
@@ -998,7 +998,7 @@ func TestGetVictimsOnNode(t *testing.T) {
st.MakeNode().Name("node2").Obj(),
},
compositePodGroups: []*schedulingv1alpha3.CompositePodGroup{
st.MakeCompositePodGroup().Name("cpg1").Namespace("default").Priority(midPriority).DisruptionModeAll().Obj(),
st.MakeCompositePodGroup().Name("cpg1").Namespace("default").Priority(lowPriority).DisruptionModeAll().Obj(),
},
podGroups: []*schedulingv1beta1.PodGroup{
st.MakePodGroup().Name("pg1").Namespace("default").ParentCompositePodGroup("cpg1").Priority(lowPriority).DisruptionModeSingle().Obj(),
@@ -1074,18 +1074,12 @@ func TestGetVictimsOnNode(t *testing.T) {
t.Fatal(err)
}
var cpgSnapshot fwk.CompositePodGroupLister
if tt.enableCompositePodGroup {
cpgSnapshot = fw.MutableSnapshotSharedLister().CompositePodGroups()
}
fts := feature.NewSchedulerFeaturesFromGates(utilfeature.DefaultFeatureGate)
pe := Evaluator{
PluginName: "TestPlugin",
Handler: fw,
executor: nil,
fts: fts,
podGroupSnapshot: fw.MutableSnapshotSharedLister().PodGroups(),
compositePodGroupSnapshot: cpgSnapshot,
PluginName: "TestPlugin",
Handler: fw,
executor: nil,
fts: fts,
}
_ = informerFactory.Scheduling().V1beta1().PodGroups().Informer()
+11 -11
View File
@@ -21,6 +21,7 @@ import (
"sync/atomic"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
corev1helpers "k8s.io/component-helpers/scheduling/corev1"
extenderv1 "k8s.io/kube-scheduler/extender/v1"
fwk "k8s.io/kube-scheduler/framework"
schedutil "k8s.io/kubernetes/pkg/scheduler/util"
@@ -30,9 +31,9 @@ import (
// It abstracts individual Pods and PodGroup, ensuring that
// atomic entities are treated as a single unit during eviction.
type Victim interface {
// Priority returns the priority of the preemption unit.
// For a single Pod, this is the Pod's priority.
// For a PodGroup, this is the priority of the PodGroup.
// Priority returns the priority of the preemption unit, which is the priority of its Pods.
// For a PodGroup or CompositePodGroup this is also the group's priority: the scheduling cycle
// guarantees that every Pod in a hierarchy shares the priority of the hierarchy's root.
Priority() int32
// Pods returns the list of all Pods that belong to this preemption unit.
@@ -64,9 +65,9 @@ func (v *victim) Pods() []fwk.PodInfo {
return v.pods
}
// Priority returns the priority of the preemption unit.
// For a single Pod, this is the Pod's priority.
// For a PodGroup, this is the priority of the PodGroup.
// Priority returns the priority of the preemption unit, which is the priority of its Pods.
// For a PodGroup or CompositePodGroup this is also the group's priority: the scheduling cycle
// guarantees that every Pod in a hierarchy shares the priority of the hierarchy's root.
func (v *victim) Priority() int32 {
return v.priority
}
@@ -82,7 +83,6 @@ func (v *victim) Type() fwk.EntityKeyType {
}
// NewPodVictim creates a new Victim representing a single Pod.
// It calculates the priority of the pod, taking into account its scheduling group if applicable.
// It ignores the error from NewVictim internally as it is guaranteed to succeed for a single valid pod.
// TODO: what we do below is not ideal:
// - From the victim's importance POV, individual Pods should always have the PodKeyType.
@@ -92,13 +92,13 @@ func (v *victim) Type() fwk.EntityKeyType {
// Pod belongs to.
//
// We should fix this on the occasion of adding support for CompositePodGroup WAP-related metrics.
func NewPodVictim(podInfo fwk.PodInfo, pgLister fwk.PodGroupLister, cpgLister fwk.CompositePodGroupLister) Victim {
priority := getPodPriority(podInfo.GetPod(), pgLister, cpgLister)
func NewPodVictim(podInfo fwk.PodInfo) Victim {
pod := podInfo.GetPod()
keyType := fwk.PodKeyType
if podInfo.GetPod().Spec.SchedulingGroup != nil && pgLister != nil {
if pod.Spec.SchedulingGroup != nil {
keyType = fwk.PodGroupKeyType
}
vi, _ := NewVictim([]fwk.PodInfo{podInfo}, priority, keyType)
vi, _ := NewVictim([]fwk.PodInfo{podInfo}, corev1helpers.PodPriority(pod), keyType)
return vi
}
+2 -29
View File
@@ -26,7 +26,6 @@ import (
"k8s.io/apimachinery/pkg/labels"
"k8s.io/apimachinery/pkg/util/sets"
policylisters "k8s.io/client-go/listers/policy/v1"
corev1helpers "k8s.io/component-helpers/scheduling/corev1"
fwk "k8s.io/kube-scheduler/framework"
)
@@ -109,8 +108,8 @@ func victimRank(vi Victim) int {
// traverseHierarchyUp traverses the hierarchy of PodGroups/CompositePodGroups upward from startKey.
// If the next parent is missing, or cannot be listed, the traversal stops.
// This method assumes that GenericWorkload feature gate is enabled.
// The cpgLister should be non-nil only if the CompositePodGroup feature gate is enabled.
// Feature gating is left to the listers: when GenericWorkload or CompositePodGroup is disabled,
// the corresponding lister rejects lookups, so the traversal ends there.
// TODO: log/return an error if there is a gap in the hierarchy.
func traverseHierarchyUp(
namespace string,
@@ -151,9 +150,6 @@ func traverseHierarchyUp(
return
}
if cpgLister == nil {
return
}
parentKey, hasParent := gpg.GetParentKey()
if !hasParent {
return
@@ -163,29 +159,6 @@ func traverseHierarchyUp(
}
}
// getPodPriority returns the effective preemption priority of a pod. If the pod belongs to
// a pod group or composite pod group hierarchy, it returns the priority of the root group of the hierarchy.
// If podGroupLister is nil or the pod does not belong to a pod group, it returns the pod's own priority.
// If compositePodGroupLister is nil or a parent composite pod group in the hierarchy is not found,
// it falls back to the priority of the last successfully resolved group (or the pod's own priority).
// TODO: log/return an error if there is a gap in the hierarchy.
func getPodPriority(p *v1.Pod, podGroupLister fwk.PodGroupLister, compositePodGroupLister fwk.CompositePodGroupLister) int32 {
if p.Spec.SchedulingGroup == nil || podGroupLister == nil {
return corev1helpers.PodPriority(p)
}
startKey := fwk.PodGroupKey(p.Namespace, *p.Spec.SchedulingGroup.PodGroupName)
var lastGPG *fwk.GenericPodGroup
for gpg := range traverseHierarchyUp(p.Namespace, startKey, podGroupLister, compositePodGroupLister) {
lastGPG = gpg
}
if lastGPG != nil {
return lastGPG.GetPriority()
}
return corev1helpers.PodPriority(p)
}
// FilterVictimsWithPDBViolation groups "victims" into "violatingVictims"
// and "nonViolatingVictims" based on whether their PDBs will be violated if they are
// preempted.
@@ -878,177 +878,6 @@ func TestPodTerminatingByPreemption(t *testing.T) {
}
}
func TestGetPodPriority(t *testing.T) {
tests := []struct {
name string
pod *v1.Pod
podGroupLister fwk.PodGroupLister
compositePodGroupLister fwk.CompositePodGroupLister
expectedPriority int32
}{
{
name: "pod without PodGroup returns pod priority",
pod: st.MakePod().Name("p1").Priority(10).Obj(),
podGroupLister: nil,
compositePodGroupLister: nil,
expectedPriority: 10,
},
{
name: "pod with PodGroup returns PodGroup priority instead of pod priority",
pod: st.MakePod().Name("p2").Priority(10).PodGroupName("pg1").Obj(),
podGroupLister: &mockPodGroupLister{
podGroups: map[string]*schedulingv1beta1.PodGroup{
"pg1": st.MakePodGroup().Name("pg1").Priority(50).Obj(),
},
},
compositePodGroupLister: nil,
expectedPriority: 50,
},
{
name: "pod with PodGroup but nil PodGroupLister falls back to pod priority",
pod: st.MakePod().Name("p4").Priority(20).PodGroupName("pg1").Obj(),
podGroupLister: nil,
compositePodGroupLister: nil,
expectedPriority: 20,
},
{
name: "pod with PodGroup but PodGroup not found in lister falls back to pod priority",
pod: st.MakePod().Name("p5").Priority(30).PodGroupName("missing-pg").Obj(),
podGroupLister: &mockPodGroupLister{podGroups: map[string]*schedulingv1beta1.PodGroup{}},
compositePodGroupLister: nil,
expectedPriority: 30,
},
{
name: "pod with nil priority and without PodGroup returns zero priority",
pod: st.MakePod().Name("p6").Obj(),
podGroupLister: nil,
compositePodGroupLister: nil,
expectedPriority: 0,
},
{
name: "pod with PodGroup and parent CPG, returns CPG priority",
pod: st.MakePod().Name("p7").Priority(10).PodGroupName("pg1").Obj(),
podGroupLister: &mockPodGroupLister{
podGroups: map[string]*schedulingv1beta1.PodGroup{
"pg1": {
ObjectMeta: metav1.ObjectMeta{Name: "pg1"},
Spec: schedulingv1beta1.PodGroupSpec{
ParentCompositePodGroupName: new("cpg1"),
},
},
},
},
compositePodGroupLister: &mockCompositePodGroupLister{
compositePodGroups: map[string]*schedulingv1alpha3.CompositePodGroup{
"cpg1": {
ObjectMeta: metav1.ObjectMeta{Name: "cpg1"},
Spec: schedulingv1alpha3.CompositePodGroupSpec{
Priority: new(int32(150)),
},
},
},
},
expectedPriority: 150,
},
{
name: "pod with PodGroup and grandparent CPG, returns grandparent CPG priority",
pod: st.MakePod().Name("p8").Priority(10).PodGroupName("pg1").Obj(),
podGroupLister: &mockPodGroupLister{
podGroups: map[string]*schedulingv1beta1.PodGroup{
"pg1": {
ObjectMeta: metav1.ObjectMeta{Name: "pg1"},
Spec: schedulingv1beta1.PodGroupSpec{
ParentCompositePodGroupName: new("cpg1"),
},
},
},
},
compositePodGroupLister: &mockCompositePodGroupLister{
compositePodGroups: map[string]*schedulingv1alpha3.CompositePodGroup{
"cpg1": {
ObjectMeta: metav1.ObjectMeta{Name: "cpg1"},
Spec: schedulingv1alpha3.CompositePodGroupSpec{
ParentCompositePodGroupName: new("cpg2"),
},
},
"cpg2": {
ObjectMeta: metav1.ObjectMeta{Name: "cpg2"},
Spec: schedulingv1alpha3.CompositePodGroupSpec{
Priority: new(int32(250)),
},
},
},
},
expectedPriority: 250,
},
{
name: "pod with PodGroup and parent CPG, but nil compositePodGroupLister, falls back to PodGroup priority",
pod: st.MakePod().Name("p9").Priority(10).PodGroupName("pg1").Obj(),
podGroupLister: &mockPodGroupLister{
podGroups: map[string]*schedulingv1beta1.PodGroup{
"pg1": st.MakePodGroup().Name("pg1").Priority(50).ParentCompositePodGroup("cpg1").Obj(),
},
},
compositePodGroupLister: nil,
expectedPriority: 50,
},
{
name: "pod with PodGroup and parent CPG, but parent CPG not found in lister, falls back to PodGroup priority",
pod: st.MakePod().Name("p10").Priority(10).PodGroupName("pg1").Obj(),
podGroupLister: &mockPodGroupLister{
podGroups: map[string]*schedulingv1beta1.PodGroup{
"pg1": st.MakePodGroup().Name("pg1").Priority(50).ParentCompositePodGroup("cpg1").Obj(),
},
},
compositePodGroupLister: &mockCompositePodGroupLister{
compositePodGroups: map[string]*schedulingv1alpha3.CompositePodGroup{},
},
expectedPriority: 50,
},
{
name: "pod with PodGroup, parent CPG and grandparent CPG, but grandparent CPG not found in lister, falls back to parent CPG priority",
pod: st.MakePod().Name("p11").Priority(10).PodGroupName("pg1").Obj(),
podGroupLister: &mockPodGroupLister{
podGroups: map[string]*schedulingv1beta1.PodGroup{
"pg1": st.MakePodGroup().Name("pg1").Priority(50).ParentCompositePodGroup("cpg1").Obj(),
},
},
compositePodGroupLister: &mockCompositePodGroupLister{
compositePodGroups: map[string]*schedulingv1alpha3.CompositePodGroup{
"cpg1": {
ObjectMeta: metav1.ObjectMeta{Name: "cpg1"},
Spec: schedulingv1alpha3.CompositePodGroupSpec{
Priority: new(int32(150)),
ParentCompositePodGroupName: new("missing-cpg2"),
},
},
},
},
expectedPriority: 150,
},
{
name: "pod with PodGroup, but PodGroup not found in lister and non-nil compositePodGroupLister, falls back to pod priority",
pod: st.MakePod().Name("p12").Priority(50).PodGroupName("missing-pg").Obj(),
podGroupLister: &mockPodGroupLister{
podGroups: map[string]*schedulingv1beta1.PodGroup{},
},
compositePodGroupLister: &mockCompositePodGroupLister{
compositePodGroups: map[string]*schedulingv1alpha3.CompositePodGroup{},
},
expectedPriority: 50,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got := getPodPriority(tt.pod, tt.podGroupLister, tt.compositePodGroupLister)
if got != tt.expectedPriority {
t.Errorf("getPodPriority() = %v, want %v", got, tt.expectedPriority)
}
})
}
}
func TestTraverseHierarchyUp(t *testing.T) {
namespace := metav1.NamespaceDefault
@@ -1060,32 +889,6 @@ func TestTraverseHierarchyUp(t *testing.T) {
stopAt string
expectedVisitedKeys []fwk.EntityKey
}{
{
name: "nil compositePodGroupLister",
startKey: fwk.PodGroupKey(namespace, "pg1"),
podGroupLister: &mockPodGroupLister{
podGroups: map[string]*schedulingv1beta1.PodGroup{
"pg1": st.MakePodGroup().Namespace(namespace).Name("pg1").Obj(),
},
},
compositePodGroupLister: nil,
expectedVisitedKeys: []fwk.EntityKey{
fwk.PodGroupKey(namespace, "pg1"),
},
},
{
name: "nil compositePodGroupLister with PG pointing to a parent CPG",
startKey: fwk.PodGroupKey(namespace, "pg1"),
podGroupLister: &mockPodGroupLister{
podGroups: map[string]*schedulingv1beta1.PodGroup{
"pg1": st.MakePodGroup().Namespace(namespace).Name("pg1").ParentCompositePodGroup("cpg").Obj(),
},
},
compositePodGroupLister: nil,
expectedVisitedKeys: []fwk.EntityKey{
fwk.PodGroupKey(namespace, "pg1"),
},
},
{
name: "unsupported key type",
startKey: fwk.PodKey(namespace, "p1"),
+19 -30
View File
@@ -19,6 +19,7 @@ package preemption
import (
v1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/util/sets"
corev1helpers "k8s.io/component-helpers/scheduling/corev1"
"k8s.io/klog/v2"
fwk "k8s.io/kube-scheduler/framework"
"k8s.io/kubernetes/pkg/scheduler/framework"
@@ -28,15 +29,15 @@ import (
// getHighestAllAncestor returns the key of the highest ancestor in the hierarchy that has disruption mode All.
// It returns (key, true) if found, or (empty, false) if not found.
// TODO: log/return an error if there is a gap in the hierarchy.
func getHighestAllAncestor(pod *v1.Pod, pgLister fwk.PodGroupLister, cpgLister fwk.CompositePodGroupLister) (fwk.EntityKey, bool) {
if pod.Spec.SchedulingGroup == nil || pgLister == nil {
func getHighestAllAncestor(pod *v1.Pod, snapshot fwk.SharedLister) (fwk.EntityKey, bool) {
if pod.Spec.SchedulingGroup == nil {
return fwk.EntityKey{}, false
}
startKey := fwk.PodGroupKey(pod.Namespace, *pod.Spec.SchedulingGroup.PodGroupName)
var highestAllKey fwk.EntityKey
var hasAll bool
for gpg := range traverseHierarchyUp(pod.Namespace, startKey, pgLister, cpgLister) {
for gpg := range traverseHierarchyUp(pod.Namespace, startKey, snapshot.PodGroups(), snapshot.CompositePodGroups()) {
if gpg.HasDisruptionModeAll() {
highestAllKey = gpg.GetKey()
hasAll = true
@@ -64,40 +65,41 @@ func createDomainVictims(snapshot fwk.SharedLister, victims []Victim) ([]*Domain
// getWorkloadPreemptionVictims discovers all candidate victims across the cluster snapshot for workload preemption.
// Pods that are part of a pod group or composite pod group with disruption mode All are grouped
// together into a single victim. Otherwise, they are treated as individual victims.
// In both cases, the priority of the victim is determined by the pod group or composite pod group priority.
func getWorkloadPreemptionVictims(logger klog.Logger, snapshot fwk.SharedLister, podGroupSnapshot fwk.PodGroupLister, compositePodGroupSnapshot fwk.CompositePodGroupLister) ([]Victim, error) {
// In both cases, the priority of the victim is the pods' own priority, which the scheduling cycle
// guarantees to match the priority of the pod group or composite pod group hierarchy they belong to.
func getWorkloadPreemptionVictims(logger klog.Logger, snapshot fwk.SharedLister) ([]Victim, error) {
nodes, err := snapshot.NodeInfos().List()
if err != nil {
return nil, err
}
return getCrossNodesVictims(logger, snapshot, podGroupSnapshot, compositePodGroupSnapshot, nodes), nil
return getCrossNodesVictims(logger, snapshot, nodes), nil
}
// getCrossNodesVictims aggregates pods across the provided nodes into cluster-wide preemption candidates.
// When a pod belongs to a group hierarchy where any ancestor has disruption mode All, it groups all scheduled pods of that hierarchy
// across the cluster into a single victim so that preemption evaluates the total cluster-wide cost
// and blast radius of evicting the entire group.
func getCrossNodesVictims(logger klog.Logger, snapshot fwk.SharedLister, podGroupSnapshot fwk.PodGroupLister, compositePodGroupSnapshot fwk.CompositePodGroupLister, nodes []fwk.NodeInfo) []Victim {
func getCrossNodesVictims(logger klog.Logger, snapshot fwk.SharedLister, nodes []fwk.NodeInfo) []Victim {
existing := sets.New[fwk.EntityKey]()
var victims []Victim
for _, node := range nodes {
for _, podInfo := range node.GetPods() {
p := podInfo.GetPod()
highestAllKey, hasAll := getHighestAllAncestor(p, podGroupSnapshot, compositePodGroupSnapshot)
highestAllKey, hasAll := getHighestAllAncestor(p, snapshot)
if !hasAll {
itemKey := fwk.PodKey(p.Namespace, p.Name)
if !existing.Has(itemKey) {
existing.Insert(itemKey)
victims = append(victims, NewPodVictim(podInfo, podGroupSnapshot, compositePodGroupSnapshot))
victims = append(victims, NewPodVictim(podInfo))
}
continue
}
if !existing.Has(highestAllKey) {
existing.Insert(highestAllKey)
victims = append(victims, searchCrossNodesVictimPods(logger, highestAllKey, podInfo, snapshot, podGroupSnapshot, compositePodGroupSnapshot))
victims = append(victims, searchCrossNodesVictimPods(logger, highestAllKey, podInfo, snapshot))
}
}
}
@@ -113,27 +115,15 @@ func searchCrossNodesVictimPods(
pgKey fwk.EntityKey,
podInfo fwk.PodInfo,
snapshot fwk.SharedLister,
podGroupSnapshot fwk.PodGroupLister,
compositePodGroupSnapshot fwk.CompositePodGroupLister,
) Victim {
var allPods []*v1.Pod
if compositePodGroupSnapshot == nil {
pgState, err := snapshot.PodGroupStates().Get(podInfo.GetPod().GetNamespace(), pgKey.Name)
for pgState, err := range helper.GetPodGroupStates(snapshot, pgKey) {
if err != nil {
// Assuming this is guaranteed to succeed if feature is on and pods exist.
// If it fails, we keep the local pods only.
return NewPodVictim(podInfo, podGroupSnapshot, compositePodGroupSnapshot)
logger.Error(err, "Failed to get pod group state during preemption victim collection", "rootKey", pgKey)
continue
}
allPods = append(allPods, pgState.ScheduledPods()...)
} else {
for pgState, err := range helper.GetPodGroupStates(snapshot, pgKey) {
if err != nil {
logger.Error(err, "Failed to get pod group state during preemption victim collection", "rootKey", pgKey)
continue
}
if pgState != nil {
allPods = append(allPods, pgState.ScheduledPods()...)
}
if pgState != nil {
allPods = append(allPods, pgState.ScheduledPods()...)
}
}
@@ -141,7 +131,7 @@ func searchCrossNodesVictimPods(
// This should never happen since podInfo is sourced from a node in the snapshot and
// should be present in ScheduledPods() of its leaf PodGroup. If leaf traversal or
// PodGroupState snapshot lookups fail, treat the pod as an individual victim.
return NewPodVictim(podInfo, podGroupSnapshot, compositePodGroupSnapshot)
return NewPodVictim(podInfo)
}
podInfos := make([]fwk.PodInfo, len(allPods))
@@ -150,8 +140,7 @@ func searchCrossNodesVictimPods(
podInfos[i], _ = framework.NewPodInfo(p)
}
priority := getPodPriority(podInfo.GetPod(), podGroupSnapshot, compositePodGroupSnapshot)
// It can only return an error for empty podInfos, which is guaranteed not to be empty here.
victim, _ := NewVictim(podInfos, priority, pgKey.Type)
victim, _ := NewVictim(podInfos, corev1helpers.PodPriority(podInfo.GetPod()), pgKey.Type)
return victim
}
@@ -175,12 +175,22 @@ func TestGetHighestAllAncestor(t *testing.T) {
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
pgLister := &mockPodGroupLister{podGroups: tt.podGroups}
var cpgLister fwk.CompositePodGroupLister
if tt.compositePodGroups != nil {
cpgLister = &mockCompositePodGroupLister{compositePodGroups: tt.compositePodGroups}
// The snapshot's listers enforce the feature gates; this test exercises the traversal itself.
featuregatetesting.SetFeatureGatesDuringTest(t, utilfeature.DefaultFeatureGate, featuregatetesting.FeatureOverrides{
features.GenericWorkload: true,
features.TopologyAwareWorkloadScheduling: true,
features.CompositePodGroup: true,
})
var pgs []*schedulingv1beta1.PodGroup
for _, pg := range tt.podGroups {
pgs = append(pgs, pg)
}
gotKey, gotHasAll := getHighestAllAncestor(tt.pod, pgLister, cpgLister)
var cpgs []*schedulingv1alpha3.CompositePodGroup
for _, cpg := range tt.compositePodGroups {
cpgs = append(cpgs, cpg)
}
snapshot := internalcache.NewTestSnapshotWithPodGroups(nil, nil, pgs, cpgs)
gotKey, gotHasAll := getHighestAllAncestor(tt.pod, snapshot)
if gotHasAll != tt.wantHasAll {
t.Errorf("hasAll mismatch: got %v, want %v", gotHasAll, tt.wantHasAll)
}
@@ -238,8 +248,8 @@ func TestGetWorkloadPreemptionVictims(t *testing.T) {
st.MakeNode().Name("node2").Obj(),
},
pods: []*v1.Pod{
st.MakePod().Name("p1").UID("p1").Node("node1").PodGroupName("pg1").Priority(10).Obj(),
st.MakePod().Name("p2").UID("p2").Node("node2").PodGroupName("pg1").Priority(10).Obj(),
st.MakePod().Name("p1").UID("p1").Node("node1").PodGroupName("pg1").Priority(50).Obj(),
st.MakePod().Name("p2").UID("p2").Node("node2").PodGroupName("pg1").Priority(50).Obj(),
},
podGroups: map[string]*schedulingv1beta1.PodGroup{
"pg1": st.MakePodGroup().Name("pg1").UID("pg1").DisruptionModeAll().Priority(50).Obj(),
@@ -255,8 +265,8 @@ func TestGetWorkloadPreemptionVictims(t *testing.T) {
st.MakeNode().Name("node2").Obj(),
},
pods: []*v1.Pod{
st.MakePod().Name("p1").UID("p1").Node("node1").PodGroupName("pg1").Priority(10).Obj(),
st.MakePod().Name("p2").UID("p2").Node("node2").PodGroupName("pg1").Priority(20).Obj(),
st.MakePod().Name("p1").UID("p1").Node("node1").PodGroupName("pg1").Priority(50).Obj(),
st.MakePod().Name("p2").UID("p2").Node("node2").PodGroupName("pg1").Priority(50).Obj(),
},
podGroups: map[string]*schedulingv1beta1.PodGroup{
"pg1": st.MakePodGroup().Name("pg1").UID("pg1").DisruptionModeSingle().Priority(50).Obj(),
@@ -273,9 +283,9 @@ func TestGetWorkloadPreemptionVictims(t *testing.T) {
st.MakeNode().Name("node2").Obj(),
},
pods: []*v1.Pod{
st.MakePod().Name("p1").UID("p1").Node("node1").PodGroupName("pg1").Priority(10).Obj(),
st.MakePod().Name("p2").UID("p2").Node("node2").PodGroupName("pg1").Priority(10).Obj(),
st.MakePod().Name("p3").UID("p3").Node("node1").PodGroupName("pg2").Priority(20).Obj(),
st.MakePod().Name("p1").UID("p1").Node("node1").PodGroupName("pg1").Priority(50).Obj(),
st.MakePod().Name("p2").UID("p2").Node("node2").PodGroupName("pg1").Priority(50).Obj(),
st.MakePod().Name("p3").UID("p3").Node("node1").PodGroupName("pg2").Priority(60).Obj(),
st.MakePod().Name("p4").UID("p4").Node("node2").Priority(30).Obj(),
},
podGroups: map[string]*schedulingv1beta1.PodGroup{
@@ -780,12 +790,7 @@ func TestGetWorkloadPreemptionVictims(t *testing.T) {
if err := cache.UpdateSnapshot(logger, snapshot); err != nil {
t.Fatalf("Failed to update snapshot: %v", err)
}
pgLister := &mockPodGroupLister{podGroups: tt.podGroups}
var cpgLister fwk.CompositePodGroupLister
if tt.enableCompositePodGroup {
cpgLister = &mockCompositePodGroupLister{compositePodGroups: tt.compositePodGroups}
}
victims, err := getWorkloadPreemptionVictims(logger, snapshot, pgLister, cpgLister)
victims, err := getWorkloadPreemptionVictims(logger, snapshot)
if err != nil {
t.Fatalf("Failed to create victims: %v", err)
}