[release-1.14] Correct NominatedNodeName lifecycle and preempt eligibility

pkg/scheduler/actions/preempt/preempt.go

@@ -295,7 +295,7 @@ func (pmpt *Action) preempt(
 	filter func(*api.TaskInfo) bool,
 	predicateHelper util.PredicateHelper,
 ) (bool, error) {
-	// Check whether the task is eligible to preempt others, e.g., check preemptionPolicy is `Never` or not
+	// Policy-only eligibility (e.g. preemptionPolicy=Never) does not depend on predicates.
 	if err := pmpt.taskEligibleToPreempt(preemptor); err != nil {
 		return false, err
 	}
@@ -304,6 +304,11 @@ func (pmpt *Action) preempt(
 		return false, fmt.Errorf("PrePredicate for task %s/%s failed for: %v", preemptor.Namespace, preemptor.Name, err)
 	}
 
+	// Nominated-node eligibility depends on predicates, so it must run after PrePredicateFn.
+	if err := pmpt.nominatedNodeEligibleToPreempt(preemptor); err != nil {
+		return false, err
+	}
+
 	// we should filter out those nodes that are UnschedulableAndUnresolvable status got in allocate action
 	allNodes := ssn.FilterOutUnschedulableAndUnresolvableNodesForTask(preemptor)
 	predicateNodes, _ := predicateHelper.PredicateNodes(preemptor, allNodes, ssn.PredicateForPreemptAction, pmpt.enablePredicateErrorCache, ssn.NodesInShard)
@@ -429,40 +434,50 @@ func preemptorFitsOnNode(ssn *framework.Session, queue *api.QueueInfo, preemptor
 		ssn.PredicateFn(preemptor, node) == nil
 }
 
+// taskEligibleToPreempt checks policy-only preconditions that do not depend on predicates.
 func (pmpt *Action) taskEligibleToPreempt(preemptor *api.TaskInfo) error {
 	if preemptor.Pod.Spec.PreemptionPolicy != nil && *preemptor.Pod.Spec.PreemptionPolicy == v1.PreemptNever {
 		return fmt.Errorf("not eligible to preempt other tasks due to preemptionPolicy is Never")
 	}
+	return nil
+}
 
+// nominatedNodeEligibleToPreempt decides whether a task carrying a NominatedNodeName
+// should still enter the preempt flow. It relies on PredicateFn and must run after
+// PrePredicateFn has initialized the task's pre-predicate state.
+func (pmpt *Action) nominatedNodeEligibleToPreempt(preemptor *api.TaskInfo) error {
 	nomNodeName := preemptor.Pod.Status.NominatedNodeName
-	if len(nomNodeName) > 0 {
-		nodeInfo, ok := pmpt.ssn.Nodes[nomNodeName]
-		if !ok {
-			return fmt.Errorf("not eligible due to the pod's nominated node is not found in the session")
-		}
+	if len(nomNodeName) == 0 {
+		return nil
+	}
 
-		err := pmpt.ssn.PredicateFn(preemptor, nodeInfo)
-		if err == nil {
-			return fmt.Errorf("not eligible due to the pod's nominated node is already schedulable, which should not happen as preemption means no node is schedulable")
-		}
+	nodeInfo, ok := pmpt.ssn.Nodes[nomNodeName]
+	if !ok {
+		// The nominated node may have been deleted before the preemptor bound.
+		// Its nomination is no longer useful, so allow normal preemption to
+		// select another candidate node.
+		return nil
+	}
 
+	err := pmpt.ssn.PredicateFn(preemptor, nodeInfo)
+	if err != nil {
 		fitError, ok := err.(*api.FitError)
 		if !ok {
 			return fmt.Errorf("not eligible due to the predicate returned a non-FitError error, the error is: %v", err)
 		}
 
-		// If the pod's nominated node is considered as UnschedulableAndUnresolvable by the predicate,
-		// then the pod should be considered for preempting again.
+		// If the nominated node is UnschedulableAndUnresolvable, the pod should be considered for preempting again.
 		if fitError.Status.ContainsUnschedulableAndUnresolvable() {
 			return nil
 		}
+	}
 
-		preemptorPodPriority := PodPriority(preemptor.Pod)
-		for _, p := range nodeInfo.Pods() {
-			if PodPriority(p) < preemptorPodPriority && podTerminatingByPreemption(p) {
-				// There is a terminating pod on the nominated node.
-				return fmt.Errorf("not eligible due to a terminating pod caused by preemption on the nominated node")
-			}
+	// Predicate success does not bypass this guard: if a lower-priority victim is already
+	// terminating on the nominated node, do not trigger another round of preemption there.
+	preemptorPodPriority := PodPriority(preemptor.Pod)
+	for _, p := range nodeInfo.Pods() {
+		if PodPriority(p) < preemptorPodPriority && podTerminatingByPreemption(p) {
+			return fmt.Errorf("not eligible due to a terminating pod caused by preemption on the nominated node")
 		}
 	}
 	return nil

pkg/scheduler/actions/preempt/preempt_test.go

@@ -313,6 +313,47 @@ func TestPreempt(t *testing.T) {
 			ExpectEvictNum: 0,
 			ExpectEvicted:  []string{}, // no victims should be reclaimed
 		},
+		{
+			Name: "continue preempt flow when nominated node is already schedulable",
+			PodGroups: []*schedulingv1beta1.PodGroup{
+				util.BuildPodGroup("pg1", "c1", "q1", 1, nil, schedulingv1beta1.PodGroupRunning),
+				util.BuildPodGroup("pg2", "c1", "q1", 1, nil, schedulingv1beta1.PodGroupInqueue),
+			},
+			Pods: []*v1.Pod{
+				util.BuildPod("c1", "preemptee1", "n1", v1.PodRunning, api.BuildResourceList("1", "1G"), "pg1", map[string]string{schedulingv1beta1.PodPreemptable: "true"}, make(map[string]string)),
+				util.BuildPodWithNominatedNodeName("c1", "preemptor1", "", v1.PodPending, api.BuildResourceList("1", "1G"), "pg2", "n1", make(map[string]string), make(map[string]string)),
+			},
+			Nodes: []*v1.Node{
+				util.BuildNode("n1", api.BuildResourceList("4", "4G", []api.ScalarResource{{Name: "pods", Value: "10"}}...), make(map[string]string)),
+			},
+			Queues: []*schedulingv1beta1.Queue{
+				util.BuildQueue("q1", 1, nil),
+			},
+			ExpectPipeLined: map[string][]string{
+				"c1/pg2": {"n1"},
+			},
+			ExpectEvictNum: 0,
+			ExpectEvicted:  []string{},
+		},
+		{
+			Name: "preempt when nominated node no longer exists",
+			PodGroups: []*schedulingv1beta1.PodGroup{
+				util.BuildPodGroupWithPrio("pg1", "c1", "q1", 0, map[string]int32{}, schedulingv1beta1.PodGroupRunning, "low-priority"),
+				util.BuildPodGroupWithPrio("pg2", "c1", "q1", 1, map[string]int32{"": 1}, schedulingv1beta1.PodGroupInqueue, "high-priority"),
+			},
+			Pods: []*v1.Pod{
+				util.BuildPod("c1", "preemptee1", "n1", v1.PodRunning, api.BuildResourceList("2", "2G"), "pg1", map[string]string{schedulingv1beta1.PodPreemptable: "true"}, make(map[string]string)),
+				util.BuildPodWithNominatedNodeName("c1", "preemptor1", "", v1.PodPending, api.BuildResourceList("2", "2G"), "pg2", "deleted-node", make(map[string]string), make(map[string]string)),
+			},
+			Nodes: []*v1.Node{
+				util.BuildNode("n1", api.BuildResourceList("2", "2G", []api.ScalarResource{{Name: "pods", Value: "10"}}...), make(map[string]string)),
+			},
+			Queues: []*schedulingv1beta1.Queue{
+				util.BuildQueue("q1", 1, nil),
+			},
+			ExpectEvicted:  []string{"c1/preemptee1"},
+			ExpectEvictNum: 1,
+		},
 		{
 			// Regression test for the preemptorTasks overwrite issue in multi-queue preemption.
 			//
@@ -548,6 +589,109 @@ func benchmarkPreemptor(ssn *framework.Session) *api.TaskInfo {
 	return nil
 }
 
+// TestNominatedNodeEligibleToPreempt verifies that a schedulable nominated node
+// no longer short-circuits eligibility: the task proceeds when no victim is
+// terminating, but predicate success still does not bypass the terminating-victim
+// guard. The guard is evaluated after PrePredicateFn, mirroring the action.
+func TestNominatedNodeEligibleToPreempt(t *testing.T) {
+	plugins := map[string]framework.PluginBuilder{
+		conformance.PluginName: conformance.New,
+		gang.PluginName:        gang.New,
+		priority.PluginName:    priority.New,
+		proportion.PluginName:  proportion.New,
+		predicates.PluginName:  predicates.New,
+	}
+	trueValue := true
+	tiers := []conf.Tier{{
+		Plugins: []conf.PluginOption{
+			{Name: conformance.PluginName, EnabledPreemptable: &trueValue},
+			{Name: gang.PluginName, EnabledPreemptable: &trueValue, EnabledJobPipelined: &trueValue, EnabledJobStarving: &trueValue},
+			{Name: priority.PluginName, EnabledTaskOrder: &trueValue, EnabledJobOrder: &trueValue, EnabledPreemptable: &trueValue, EnabledJobPipelined: &trueValue, EnabledJobStarving: &trueValue},
+			{Name: proportion.PluginName, EnabledOverused: &trueValue, EnabledAllocatable: &trueValue, EnabledQueueOrder: &trueValue, EnabledPredicate: &trueValue},
+			{Name: predicates.PluginName, EnabledPreemptable: &trueValue, EnabledPredicate: &trueValue},
+		},
+	}}
+
+	lowPrio, highPrio := int32(10), int32(100000)
+	newVictim := func(terminating bool) *v1.Pod {
+		victim := util.BuildPod("c1", "victim", "n1", v1.PodRunning, api.BuildResourceList("1", "1G"), "pg1", map[string]string{schedulingv1beta1.PodPreemptable: "true"}, make(map[string]string))
+		victim.Spec.Priority = &lowPrio
+		if terminating {
+			now := metav1.Now()
+			victim.DeletionTimestamp = &now
+			victim.Status.Conditions = append(victim.Status.Conditions, v1.PodCondition{
+				Type:   v1.DisruptionTarget,
+				Status: v1.ConditionTrue,
+				Reason: v1.PodReasonPreemptionByScheduler,
+			})
+		}
+		return victim
+	}
+	newPreemptor := func() *v1.Pod {
+		preemptor := util.BuildPodWithNominatedNodeName("c1", "preemptor", "", v1.PodPending, api.BuildResourceList("1", "1G"), "pg2", "n1", make(map[string]string), make(map[string]string))
+		preemptor.Spec.Priority = &highPrio
+		return preemptor
+	}
+
+	cases := []struct {
+		name        string
+		terminating bool
+		wantErr     bool
+	}{
+		{name: "schedulable nominated node without terminating victim is eligible", terminating: false, wantErr: false},
+		{name: "schedulable nominated node with terminating victim is ineligible", terminating: true, wantErr: true},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			test := uthelper.TestCommonStruct{
+				Name:    tc.name,
+				Plugins: plugins,
+				PodGroups: []*schedulingv1beta1.PodGroup{
+					util.BuildPodGroup("pg1", "c1", "q1", 1, nil, schedulingv1beta1.PodGroupRunning),
+					util.BuildPodGroup("pg2", "c1", "q1", 1, nil, schedulingv1beta1.PodGroupInqueue),
+				},
+				Pods: []*v1.Pod{
+					newVictim(tc.terminating),
+					newPreemptor(),
+				},
+				Nodes: []*v1.Node{
+					util.BuildNode("n1", api.BuildResourceList("4", "4G", []api.ScalarResource{{Name: "pods", Value: "10"}}...), make(map[string]string)),
+				},
+				Queues: []*schedulingv1beta1.Queue{
+					util.BuildQueue("q1", 1, nil),
+				},
+			}
+
+			ssn := test.RegisterSession(tiers, []conf.Configuration{{Name: New().Name(),
+				Arguments: map[string]interface{}{EnableTopologyAwarePreemptionKey: false}}})
+			defer test.Close()
+
+			preemptor := benchmarkPreemptor(ssn)
+			if preemptor == nil {
+				t.Fatal("preemptor task not found in session")
+			}
+
+			pmpt := New()
+			pmpt.ssn = ssn
+
+			// Mirror the action: PrePredicateFn initializes pre-predicate state before
+			// the nominated-node eligibility check runs.
+			if err := ssn.PrePredicateFn(preemptor); err != nil {
+				t.Fatalf("PrePredicateFn failed: %v", err)
+			}
+
+			err := pmpt.nominatedNodeEligibleToPreempt(preemptor)
+			if tc.wantErr && err == nil {
+				t.Errorf("expected ineligible (error), got nil")
+			}
+			if !tc.wantErr && err != nil {
+				t.Errorf("expected eligible (nil), got error: %v", err)
+			}
+		})
+	}
+}
+
 func TestTopologyAwarePreempt(t *testing.T) {
 	plugins := map[string]framework.PluginBuilder{
 		capacity.PluginName:    capacity.New,

pkg/scheduler/api/job_info.go

@@ -850,8 +850,10 @@ func (ji *JobInfo) TaskSchedulingReason(tid TaskID) (reason, msg, nominatedNodeN
 		return PodReasonSchedulable, msg, ""
 	case Pipelined:
 		msg = fmt.Sprintf("Pod %s/%s can possibly be assigned to %s, once resource is released and minAvailable is satisfied", taskInfo.Namespace, taskInfo.Name, ctx.NodeName)
-		if ctx.EvictionOccurred {
-			nominatedNodeName = ctx.NodeName
+		// Derive the nomination from the live transaction, not the LastTransaction snapshot used for the message above.
+		liveCtx := taskInfo.GetTransactionContext()
+		if liveCtx.Status == Pipelined && liveCtx.NodeName != "" && liveCtx.EvictionOccurred {
+			nominatedNodeName = liveCtx.NodeName
 		}
 		return PodReasonUnschedulable, msg, nominatedNodeName
 	case Pending:

pkg/scheduler/api/job_info_test.go

@@ -397,6 +397,48 @@ func TestTaskSchedulingReason(t *testing.T) {
 	}
 }
 
+// TestTaskSchedulingReasonNomination checks that the nomination is published
+// from the live transaction, so a rolled-back gang attempt does not leak a
+// stale NominatedNodeName while a committed preemption still keeps it.
+func TestTaskSchedulingReasonNomination(t *testing.T) {
+	tests := []struct {
+		name           string
+		live           TransactionContext
+		lastTx         *TransactionContext
+		wantNomination string
+	}{
+		{
+			name:           "committed preemption retains nomination",
+			live:           TransactionContext{Status: Pipelined, NodeName: "node1", EvictionOccurred: true},
+			lastTx:         nil,
+			wantNomination: "node1",
+		},
+		{
+			name:           "discarded attempt publishes no nomination",
+			live:           TransactionContext{Status: Pending},
+			lastTx:         &TransactionContext{Status: Pipelined, NodeName: "node1", EvictionOccurred: true},
+			wantNomination: "",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			pod := buildPod("ns1", "task-1", "", v1.PodPending, BuildResourceList("1", "1G"), nil, make(map[string]string))
+			job := NewJobInfo(JobID("job1"))
+			ti := NewTaskInfo(pod)
+			job.AddTaskInfo(ti)
+
+			ti.TransactionContext = tt.live
+			ti.LastTransaction = tt.lastTx
+
+			_, _, nominated := job.TaskSchedulingReason(ti.UID)
+			if nominated != tt.wantNomination {
+				t.Errorf("nominatedNodeName = %q, want %q", nominated, tt.wantNomination)
+			}
+		})
+	}
+}
+
 func TestJobInfo(t *testing.T) {
 	newTaskFunc := func(uid, jobUid types.UID, status TaskStatus, resources *Resource) *TaskInfo {
 		isBestEffort := resources.IsEmpty()

pkg/scheduler/framework/statement.go

@@ -254,6 +254,7 @@ func (s *Statement) UnPipeline(task *api.TaskInfo) error {
 		}
 	}
 	task.NodeName = ""
+	task.EvictionOccurred = false
 	task.JobAllocatedHyperNode = ""
 
 	return nil