[cherry-pick for release-1.9] Update NominatedNodeName for pipelined task

pkg/scheduler/api/job_info.go

@@ -640,7 +640,7 @@ func (ji *JobInfo) FitError() string {
 	// Stat histogram for pending tasks only
 	reasons = make(map[string]int)
 	for uid := range ji.TaskStatusIndex[Pending] {
-		reason, _ := ji.TaskSchedulingReason(uid)
+		reason, _, _ := ji.TaskSchedulingReason(uid)
 		reasons[reason]++
 	}
 	if len(reasons) > 0 {
@@ -651,10 +651,10 @@ func (ji *JobInfo) FitError() string {
 
 // TaskSchedulingReason get detailed reason and message of the given task
 // It returns detailed reason and message for tasks based on last scheduling transaction.
-func (ji *JobInfo) TaskSchedulingReason(tid TaskID) (reason string, msg string) {
+func (ji *JobInfo) TaskSchedulingReason(tid TaskID) (reason, msg, nominatedNodeName string) {
 	taskInfo, exists := ji.Tasks[tid]
 	if !exists {
-		return "", ""
+		return "", "", ""
 	}
 
 	// Get detailed scheduling reason based on LastTransaction
@@ -668,19 +668,19 @@ func (ji *JobInfo) TaskSchedulingReason(tid TaskID) (reason string, msg string)
 	case Allocated:
 		// Pod is schedulable
 		msg = fmt.Sprintf("Pod %s/%s can possibly be assigned to %s", taskInfo.Namespace, taskInfo.Name, ctx.NodeName)
-		return PodReasonSchedulable, msg
+		return PodReasonSchedulable, msg, ""
 	case Pipelined:
 		msg = fmt.Sprintf("Pod %s/%s can possibly be assigned to %s, once resource is released", taskInfo.Namespace, taskInfo.Name, ctx.NodeName)
-		return PodReasonUnschedulable, msg
+		return PodReasonUnschedulable, msg, ctx.NodeName
 	case Pending:
 		if fe := ji.NodesFitErrors[tid]; fe != nil {
 			// Pod is unschedulable
-			return PodReasonUnschedulable, fe.Error()
+			return PodReasonUnschedulable, fe.Error(), ""
 		}
 		// Pod is not scheduled yet, keep UNSCHEDULABLE as the reason to support cluster autoscaler
-		return PodReasonUnschedulable, msg
+		return PodReasonUnschedulable, msg, ""
 	default:
-		return status.String(), msg
+		return status.String(), msg, ""
 	}
 }
 

pkg/scheduler/api/job_info_test.go

@@ -285,7 +285,7 @@ func TestTaskSchedulingReason(t *testing.T) {
 		for uid, exp := range test.expected {
 			msg := job.JobFitErrors
 			if uid != "pg" {
-				_, msg = job.TaskSchedulingReason(TaskID(uid))
+				_, msg, _ = job.TaskSchedulingReason(TaskID(uid))
 			}
 			t.Logf("case #%d, task %v, result: %s", i, uid, msg)
 			if msg != exp {

pkg/scheduler/cache/cache.go

@@ -278,16 +278,16 @@ func podConditionHaveUpdate(status *v1.PodStatus, condition *v1.PodCondition) bo
 	return !isEqual
 }
 
-// UpdatePodCondition will Update pod with podCondition
-func (su *defaultStatusUpdater) UpdatePodCondition(pod *v1.Pod, condition *v1.PodCondition) (*v1.Pod, error) {
-	klog.V(3).Infof("Updating pod condition for %s/%s to (%s==%s)", pod.Namespace, pod.Name, condition.Type, condition.Status)
-	if podutil.UpdatePodCondition(&pod.Status, condition) {
-		return su.kubeclient.CoreV1().Pods(pod.Namespace).UpdateStatus(context.TODO(), pod, metav1.UpdateOptions{})
-	}
-	return pod, nil
+func podNominatedNodeNameNeedUpdate(status *v1.PodStatus, nodeName string) bool {
+	return status.NominatedNodeName != nodeName
+}
+
+// UpdatePodStatus will Update pod status
+func (su *defaultStatusUpdater) UpdatePodStatus(pod *v1.Pod) (*v1.Pod, error) {
+	return su.kubeclient.CoreV1().Pods(pod.Namespace).UpdateStatus(context.TODO(), pod, metav1.UpdateOptions{})
 }
 
-// UpdatePodGroup will Update pod with podCondition
+// UpdatePodGroup will Update PodGroup
 func (su *defaultStatusUpdater) UpdatePodGroup(pg *schedulingapi.PodGroup) (*schedulingapi.PodGroup, error) {
 	podgroup := &vcv1beta1.PodGroup{}
 	if err := schedulingscheme.Scheme.Convert(&pg.PodGroup, podgroup, nil); err != nil {
@@ -982,7 +982,7 @@ func (sc *SchedulerCache) EventRecorder() record.EventRecorder {
 }
 
 // taskUnschedulable updates pod status of pending task
-func (sc *SchedulerCache) taskUnschedulable(task *schedulingapi.TaskInfo, reason, message string) error {
+func (sc *SchedulerCache) taskUnschedulable(task *schedulingapi.TaskInfo, reason, message, nominatedNodeName string) error {
 	pod := task.Pod
 
 	condition := &v1.PodCondition{
@@ -992,14 +992,28 @@ func (sc *SchedulerCache) taskUnschedulable(task *schedulingapi.TaskInfo, reason
 		Message: message,
 	}
 
-	if podConditionHaveUpdate(&pod.Status, condition) {
+	updateCond := podConditionHaveUpdate(&pod.Status, condition)
+	updateNomiNode := podNominatedNodeNameNeedUpdate(&pod.Status, nominatedNodeName)
+
+	if updateCond || updateNomiNode {
 		pod = pod.DeepCopy()
 
+		if updateCond && podutil.UpdatePodCondition(&pod.Status, condition) {
+			klog.V(3).Infof("Updating pod condition for %s/%s to (%s==%s)", pod.Namespace, pod.Name, condition.Type, condition.Status)
+		}
+
+		// if nominatedNode field changed, we should update it to the pod status, for k8s
+		// autoscaler will check this field and ignore this pod when scale up.
+		if updateNomiNode {
+			klog.V(3).Infof("Updating pod nominatedNodeName for %s/%s from (%s) to (%s)", pod.Namespace, pod.Name, pod.Status.NominatedNodeName, nominatedNodeName)
+			pod.Status.NominatedNodeName = nominatedNodeName
+		}
+
 		// The reason field in 'Events' should be "FailedScheduling", there is not constants defined for this in
 		// k8s core, so using the same string here.
 		// The reason field in PodCondition can be "Unschedulable"
 		sc.Recorder.Eventf(pod, v1.EventTypeWarning, "FailedScheduling", message)
-		if _, err := sc.StatusUpdater.UpdatePodCondition(pod, condition); err != nil {
+		if _, err := sc.StatusUpdater.UpdatePodStatus(pod); err != nil {
 			return err
 		}
 	} else {
@@ -1434,13 +1448,13 @@ func (sc *SchedulerCache) RecordJobStatusEvent(job *schedulingapi.JobInfo, updat
 	// Update podCondition for tasks Allocated and Pending before job discarded
 	for _, status := range []schedulingapi.TaskStatus{schedulingapi.Allocated, schedulingapi.Pending, schedulingapi.Pipelined} {
 		for _, taskInfo := range job.TaskStatusIndex[status] {
-			reason, msg := job.TaskSchedulingReason(taskInfo.UID)
+			reason, msg, nominatedNodeName := job.TaskSchedulingReason(taskInfo.UID)
 			if len(msg) == 0 {
 				msg = baseErrorMessage
 			}
-			if err := sc.taskUnschedulable(taskInfo, reason, msg); err != nil {
-				klog.Errorf("Failed to update unschedulable task status <%s/%s>: %v",
-					taskInfo.Namespace, taskInfo.Name, err)
+			if err := sc.taskUnschedulable(taskInfo, reason, msg, nominatedNodeName); err != nil {
+				klog.ErrorS(err, "Failed to update unschedulable task status", "task", klog.KRef(taskInfo.Namespace, taskInfo.Name),
+					"reason", reason, "message", msg)
 			}
 		}
 	}

pkg/scheduler/cache/interface.go

@@ -109,7 +109,7 @@ type Evictor interface {
 
 // StatusUpdater updates pod with given PodCondition
 type StatusUpdater interface {
-	UpdatePodCondition(pod *v1.Pod, podCondition *v1.PodCondition) (*v1.Pod, error)
+	UpdatePodStatus(pod *v1.Pod) (*v1.Pod, error)
 	UpdatePodGroup(pg *api.PodGroup) (*api.PodGroup, error)
 	UpdateQueueStatus(queue *api.QueueInfo) error
 }

pkg/scheduler/util/test_utils.go

@@ -306,13 +306,13 @@ func (fe *FakeEvictor) Evict(p *v1.Pod, reason string) error {
 type FakeStatusUpdater struct {
 }
 
-// UpdatePodCondition is a empty function
-func (ftsu *FakeStatusUpdater) UpdatePodCondition(pod *v1.Pod, podCondition *v1.PodCondition) (*v1.Pod, error) {
+// UpdatePodStatus is an empty function
+func (ftsu *FakeStatusUpdater) UpdatePodStatus(pod *v1.Pod) (*v1.Pod, error) {
 	// do nothing here
 	return nil, nil
 }
 
-// UpdatePodGroup is a empty function
+// UpdatePodGroup is an empty function
 func (ftsu *FakeStatusUpdater) UpdatePodGroup(pg *api.PodGroup) (*api.PodGroup, error) {
 	// do nothing here
 	return nil, nil