From 71fd4b19e3e97ef2b3cbc5ca1e80c76969f0dbe7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Maciej=20Skocze=C5=84?= Date: Thu, 5 Mar 2026 13:35:44 +0000 Subject: [PATCH] Rework podGroupAlgorithmResult to use fwk.Status and capture errors correctly --- pkg/scheduler/metrics/profile_metrics.go | 6 + pkg/scheduler/schedule_one_podgroup.go | 193 +++++++++------- pkg/scheduler/schedule_one_podgroup_test.go | 211 +++++++++++++++--- .../kube-scheduler/framework/interface.go | 16 ++ 4 files changed, 312 insertions(+), 114 deletions(-) diff --git a/pkg/scheduler/metrics/profile_metrics.go b/pkg/scheduler/metrics/profile_metrics.go index f88bcf28e91..16db37c9e6f 100644 --- a/pkg/scheduler/metrics/profile_metrics.go +++ b/pkg/scheduler/metrics/profile_metrics.go @@ -66,6 +66,12 @@ func PodGroupWaitingOnPreemption(profile string, duration float64) { observePodGroupScheduleAttemptAndLatency(WaitingOnPreemptionResult, profile, duration) } +// PodGroupScheduleError records a pod group scheduling attempt that had an error, and the +// duration since `start`. +func PodGroupScheduleError(profile string, duration float64) { + observePodGroupScheduleAttemptAndLatency(ErrorResult, profile, duration) +} + func observePodGroupScheduleAttemptAndLatency(result, profile string, duration float64) { podGroupSchedulingLatency.WithLabelValues(result, profile).Observe(duration) podGroupScheduleAttempts.WithLabelValues(result, profile).Inc() diff --git a/pkg/scheduler/schedule_one_podgroup.go b/pkg/scheduler/schedule_one_podgroup.go index d88fefd45aa..c73c010844d 100644 --- a/pkg/scheduler/schedule_one_podgroup.go +++ b/pkg/scheduler/schedule_one_podgroup.go @@ -36,7 +36,7 @@ import ( ) // errPodGroupUnschedulable is used to describe that the pod group is unschedulable. -var errPodGroupUnschedulable = fmt.Errorf("other pods from a pod group are unschedulable") +var errPodGroupUnschedulable = fmt.Errorf("pod group is unschedulable") // scheduleOnePodGroup does the entire workload-aware scheduling workflow for a single pod group. func (sched *Scheduler) scheduleOnePodGroup(ctx context.Context, podGroupInfo *framework.QueuedPodGroupInfo) { @@ -58,7 +58,7 @@ func (sched *Scheduler) scheduleOnePodGroup(ctx context.Context, podGroupInfo *f sched.SchedulingQueue.Done(podInfo.Pod.UID) return } - sched.FailureHandler(ctx, podFwk, podInfo, fwk.AsStatus(err), nil, time.Now()) + sched.FailureHandler(ctx, podFwk, podInfo, fwk.AsStatus(err), clearNominatedNode, time.Now()) } return } @@ -175,7 +175,7 @@ type podSchedulingContext struct { } // initPodSchedulingContext initializes the scheduling context of a single pod for pod group scheduling cycle. -func (sched *Scheduler) initPodSchedulingContext(ctx context.Context, pod *v1.Pod) *podSchedulingContext { +func initPodSchedulingContext(ctx context.Context, pod *v1.Pod) *podSchedulingContext { logger := klog.FromContext(ctx) // TODO(knelasevero): Remove duplicated keys from log entry calls // When contextualized logging hits GA @@ -205,22 +205,19 @@ func (sched *Scheduler) podGroupCycle(ctx context.Context, schedFwk framework.Fr // Synchronously attempt to find a fit for the pod group. start := time.Now() - podGroupCycleCtx, cancel := context.WithCancel(ctx) - defer cancel() - logger := klog.FromContext(podGroupCycleCtx) - + logger := klog.FromContext(ctx) if err := sched.Cache.UpdateSnapshot(logger, sched.nodeInfoSnapshot); err != nil { - logger.Error(err, "Error updating snapshot") - for _, pInfo := range podGroupInfo.QueuedPodInfos { - sched.FailureHandler(podGroupCycleCtx, schedFwk, pInfo, fwk.AsStatus(err), nil, start) + logger.Error(err, "Error updating snapshot", "podGroup", klog.KObj(podGroupInfo)) + result := podGroupAlgorithmResult{ + status: fwk.AsStatus(err), } + sched.submitPodGroupAlgorithmResult(ctx, schedFwk, podGroupInfo, result, start) return } - result := sched.podGroupSchedulingAlgorithm(podGroupCycleCtx, schedFwk, podGroupInfo) + result := sched.podGroupSchedulingAlgorithm(ctx, schedFwk, podGroupInfo) metrics.PodGroupSchedulingAlgorithmLatency.Observe(metrics.SinceInSeconds(start)) - // submitPodGroupAlgorithmResult can dispatch binding goroutines, so should be called with the noncancelable ctx. sched.submitPodGroupAlgorithmResult(ctx, schedFwk, podGroupInfo, result, start) } @@ -241,31 +238,28 @@ type algorithmResult struct { permitStatus *fwk.Status } -// podGroupAlgorithmStatus is a status of a pod group scheduling algorithm. -type podGroupAlgorithmStatus string - -const ( - // podGroupFeasible means that the pod group is schedulable, doesn't require any preemption - // and its feasible pods should be moved to the binding cycle. - // Should be set when the pod group is feasible and preemption is not required for any member pod. - podGroupFeasible podGroupAlgorithmStatus = "feasible" - // podGroupUnschedulable means that the pod group is unschedulable - // and all its pods should be moved back to the scheduling queue as unschedulable. - podGroupUnschedulable podGroupAlgorithmStatus = "unschedulable" - // podGroupWaitingOnPreemption means that the pod group requires preemption or is waiting for it to complete, - // so all its pods should be moved back to the scheduling queue, - // waiting for resources to be released. - // Should be set when the pod group would be feasible, but any member pod requires preemption. - podGroupWaitingOnPreemption podGroupAlgorithmStatus = "waiting_on_preemption" -) - // podGroupAlgorithmResult stores the scheduling pod scheduling results for a pod group // and any information needed to act on these results. type podGroupAlgorithmResult struct { // podResults is the list of scheduling results for each pod in the group. + // Only in the case of a pod group-wide Unschedulable or Error status can it contain fewer pods. podResults []algorithmResult // status is the final status of the pod group algorithm. - status podGroupAlgorithmStatus + // + // Success code indicates that the pod group is schedulable and does not require any preemption. + // Its feasible pods should be moved to the binding cycle. + // This should only be set when the pod group is feasible and `waitingOnPreemption` is false. + // + // Unschedulable code indicates that the pod group is unschedulable, + // and all its pods should be moved back to the scheduling queue as unschedulable. + // Result with `waitingOnPreemption` set to true should have the Unschedulable status. + // + // Error code means that pod group scheduling failed due to an unexpected error, + // and no pods will be scheduled this attempt. + status *fwk.Status + // waitingOnPreemption indicates whether this pod group requires or is waiting for preemption to complete. + // This can only be set to true when the status is Unschedulable. + waitingOnPreemption bool } // podGroupSchedulingDefaultAlgorithm runs the default algorithm for scheduling a pod group. @@ -274,8 +268,9 @@ type podGroupAlgorithmResult struct { // treat that pod as already scheduled on that node with victims being already removed in memory. func (sched *Scheduler) podGroupSchedulingDefaultAlgorithm(ctx context.Context, schedFwk framework.Framework, podGroupInfo *framework.QueuedPodGroupInfo) podGroupAlgorithmResult { result := podGroupAlgorithmResult{ - podResults: make([]algorithmResult, 0, len(podGroupInfo.QueuedPodInfos)), - status: podGroupUnschedulable, + podResults: make([]algorithmResult, 0, len(podGroupInfo.QueuedPodInfos)), + status: fwk.NewStatus(fwk.Unschedulable).WithError(errPodGroupUnschedulable), + waitingOnPreemption: false, } logger := klog.FromContext(ctx) @@ -287,9 +282,17 @@ func (sched *Scheduler) podGroupSchedulingDefaultAlgorithm(ctx context.Context, result.podResults = append(result.podResults, podResult) if !podResult.status.IsSuccess() && !podResult.requiresPreemption { // When a pod is not feasible and doesn't require preemption, it means that it failed scheduling. - // However, the pod group can still be schedulable as long as the permit check can succeed. - continue + if podResult.status.IsRejected() { + // If the pod is rejected, the pod group can still be schedulable as long as the permit check can succeed. + continue + } + // When the algorithm returns error or unexpected status, stop evaluating the rest of the pods. + result.status = fwk.AsStatus(fmt.Errorf("failed to schedule other pod from a pod group: %w", podResult.status.AsError())) + // Clear the waiting on preemption flag that could have been set by previous pods. + result.waitingOnPreemption = false + break } + // At this point, the pod has passed the scheduling algorithm with the Permit status being either Success or Wait. // We unreserve the pod at the end of the whole algorithm (via defer) because it should be ultimately returned to the queue, // without binding it yet. We only assumed the pod to check feasibility of subsequent pods in the group. defer revertFn() @@ -299,9 +302,11 @@ func (sched *Scheduler) podGroupSchedulingDefaultAlgorithm(ctx context.Context, // When the permit returns success for any pod, the pod group is schedulable. if requiresPreemption { // If any preemption is required, the whole pod group requires it to be feasible. - result.status = podGroupWaitingOnPreemption + result.status = fwk.NewStatus(fwk.Unschedulable, "pod group is waiting for preemption to complete").WithError(errPodGroupUnschedulable) + // Set the waitingOnPreemption to true iff the pod group is feasible (Permit returned Success) and requires preemption. + result.waitingOnPreemption = true } else { - result.status = podGroupFeasible + result.status = nil // Success } } } @@ -313,7 +318,7 @@ func (sched *Scheduler) podGroupSchedulingDefaultAlgorithm(ctx context.Context, // It returns the algorithm result and, if successful or the preemption is required, the permit status together with the revert function. func (sched *Scheduler) podGroupPodSchedulingAlgorithm(ctx context.Context, schedFwk framework.Framework, podGroupInfo *framework.QueuedPodGroupInfo, podInfo *framework.QueuedPodInfo) (algorithmResult, func()) { pod := podInfo.Pod - podCtx := sched.initPodSchedulingContext(ctx, pod) + podCtx := initPodSchedulingContext(ctx, pod) logger := podCtx.logger ctx = klog.NewContext(ctx, logger) start := time.Now() @@ -396,22 +401,42 @@ func (sched *Scheduler) podGroupPodSchedulingAlgorithm(ctx context.Context, sche // for the next pod group scheduling cycle. // If the preemption is required for this pod group, all pods are moved back to the scheduling queue // and require the next pod group scheduling cycle to verify the preemption outcome. -func (sched *Scheduler) submitPodGroupAlgorithmResult(ctx context.Context, schedFwk framework.Framework, podGroupInfo *framework.QueuedPodGroupInfo, result podGroupAlgorithmResult, start time.Time) { +func (sched *Scheduler) submitPodGroupAlgorithmResult(ctx context.Context, schedFwk framework.Framework, podGroupInfo *framework.QueuedPodGroupInfo, podGroupResult podGroupAlgorithmResult, start time.Time) { var scheduledPods, unschedulablePods int - for i, podResult := range result.podResults { - pInfo := podGroupInfo.QueuedPodInfos[i] + for i, pInfo := range podGroupInfo.QueuedPodInfos { + var podResult algorithmResult + if len(podGroupResult.podResults) > i { + podResult = podGroupResult.podResults[i] + } else { + // In pod group-level unschedulable or error cases, podResult may not be defined. + // Initialize it now to handle pod failure correctly. + podResult = algorithmResult{ + podCtx: initPodSchedulingContext(ctx, pInfo.Pod), + status: podGroupResult.status.Clone(), + } + } podCtx := podResult.podCtx ctx := klog.NewContext(ctx, podCtx.logger) // To be consistent with pod-by-pod scheduling, construct pod scheduling start time as `now - scheduling duration`. podSchedulingStart := time.Now().Add(-podResult.schedulingDuration) + if podGroupResult.status.IsError() { + if podResult.status.IsError() { + // If this exact pod failed with an error, use its status instead. + sched.FailureHandler(ctx, schedFwk, pInfo, podResult.status, clearNominatedNode, podSchedulingStart) + continue + } + // Pod group failed with an error. Reject all pods with its status. + sched.FailureHandler(ctx, schedFwk, pInfo, podGroupResult.status, clearNominatedNode, podSchedulingStart) + continue + } if podResult.status.IsSuccess() { nominatingInfo := &fwk.NominatingInfo{ NominatingMode: fwk.ModeOverride, NominatedNodeName: podResult.scheduleResult.SuggestedHost, } - switch result.status { - case podGroupFeasible: + switch { + case podGroupResult.status.IsSuccess(): // Pod no longer needs a pod group scheduling cycle. Setting it to false to disable any checks in further functions. pInfo.NeedsPodGroupScheduling = false // Schedule result is applied for pod and its binding cycle executes. @@ -424,27 +449,26 @@ func (sched *Scheduler) submitPodGroupAlgorithmResult(ctx context.Context, sched } go sched.runBindingCycle(ctx, podCtx.state, schedFwk, podResult.scheduleResult, assumedPodInfo, podSchedulingStart, podCtx.podsToActivate) scheduledPods++ - case podGroupUnschedulable: - // Pod group is unschedulable, so the pod has to be marked as unschedulable. - // Its rejection status is set to its permit status message. - status := fwk.NewStatus(fwk.Unschedulable, podResult.permitStatus.Message()).WithError(errPodGroupUnschedulable) - sched.FailureHandler(ctx, schedFwk, pInfo, status, clearNominatedNode, podSchedulingStart) - unschedulablePods++ - case podGroupWaitingOnPreemption: - // Pod has to come back to the scheduling queue as unschedulable, waiting for preemption to complete. - status := fwk.NewStatus(fwk.Unschedulable, "preemption is required for other pods from a pod group").WithError(errPodGroupUnschedulable) - sched.FailureHandler(ctx, schedFwk, pInfo, status, nominatingInfo, podSchedulingStart) + case podGroupResult.status.IsRejected(): + if podGroupResult.waitingOnPreemption { + // Pod has to come back to the scheduling queue as unschedulable, waiting for preemption to complete. + sched.FailureHandler(ctx, schedFwk, pInfo, podGroupResult.status, nominatingInfo, podSchedulingStart) + } else { + // Pod group is unschedulable, so the pod has to be marked as unschedulable. + // Its rejection status is set to its permit status message. + status := fwk.NewStatus(fwk.Unschedulable, podResult.permitStatus.Message()).WithError(errPodGroupUnschedulable) + sched.FailureHandler(ctx, schedFwk, pInfo, status, clearNominatedNode, podSchedulingStart) + } unschedulablePods++ default: - err := fmt.Errorf("got unexpected pod group scheduling algorithm result status: %q", result.status) - utilruntime.HandleErrorWithLogger(podCtx.logger, err, "Unexpected error", "podGroup", klog.KObj(podGroupInfo), "pod", klog.KObj(pInfo.Pod)) - sched.FailureHandler(ctx, schedFwk, pInfo, fwk.AsStatus(err), nominatingInfo, podSchedulingStart) + err := fmt.Errorf("received unexpected pod group scheduling algorithm status code: %s", podGroupResult.status.Code()) + sched.FailureHandler(ctx, schedFwk, pInfo, fwk.AsStatus(err), clearNominatedNode, podSchedulingStart) unschedulablePods++ } } else { // TBD: Add a message to status if the pod used features for which finding a placement cannot be guaranteed, // such as heterogeneous pod group or using inter-pod dependencies. - if podResult.scheduleResult.nominatingInfo != nil && podResult.scheduleResult.nominatingInfo.NominatedNodeName != "" && result.status == podGroupUnschedulable { + if podResult.requiresPreemption && !podGroupResult.waitingOnPreemption { // Pod group is unschedulable, so the pod has to be marked as unschedulable, even if it just required preemption. // Its rejection status is set to its permit status message, as the preemption message is no longer relevant. status := fwk.NewStatus(fwk.Unschedulable, podResult.permitStatus.Message()).WithError(errPodGroupUnschedulable) @@ -457,16 +481,21 @@ func (sched *Scheduler) submitPodGroupAlgorithmResult(ctx context.Context, sched } } logger := klog.FromContext(ctx) - switch result.status { - case podGroupFeasible: + switch { + case podGroupResult.status.IsSuccess(): logger.V(2).Info("Successfully scheduled a pod group", "podGroup", klog.KObj(podGroupInfo), "scheduledPods", scheduledPods, "unschedulablePods", unschedulablePods) metrics.PodGroupScheduled(schedFwk.ProfileName(), metrics.SinceInSeconds(start)) - case podGroupUnschedulable: - logger.V(2).Info("Unable to schedule a pod group", "podGroup", klog.KObj(podGroupInfo), "unschedulablePods", unschedulablePods) - metrics.PodGroupUnschedulable(schedFwk.ProfileName(), metrics.SinceInSeconds(start)) - case podGroupWaitingOnPreemption: - logger.V(2).Info("Pod group is waiting for preemption", "podGroup", klog.KObj(podGroupInfo), "unschedulablePods", unschedulablePods) - metrics.PodGroupWaitingOnPreemption(schedFwk.ProfileName(), metrics.SinceInSeconds(start)) + case podGroupResult.status.IsRejected(): + if podGroupResult.waitingOnPreemption { + logger.V(2).Info("Pod group is waiting for preemption", "podGroup", klog.KObj(podGroupInfo), "unschedulablePods", unschedulablePods) + metrics.PodGroupWaitingOnPreemption(schedFwk.ProfileName(), metrics.SinceInSeconds(start)) + } else { + logger.V(2).Info("Unable to schedule a pod group", "podGroup", klog.KObj(podGroupInfo), "unschedulablePods", unschedulablePods) + metrics.PodGroupUnschedulable(schedFwk.ProfileName(), metrics.SinceInSeconds(start)) + } + default: + utilruntime.HandleErrorWithContext(ctx, podGroupResult.status.AsError(), "Error scheduling pod group", "podGroup", klog.KObj(podGroupInfo), "errorPods", len(podGroupInfo.QueuedPodInfos)) + metrics.PodGroupScheduleError(schedFwk.ProfileName(), metrics.SinceInSeconds(start)) } } @@ -486,7 +515,9 @@ func (sched *Scheduler) podGroupSchedulingPlacementAlgorithm(ctx context.Context logger := klog.FromContext(ctx) allNodes, err := sched.nodeInfoSnapshot.NodeInfos().List() if err != nil { - return sched.podGroupAlgorithmFailure(ctx, podGroupInfo, fwk.AsStatus(err)) + return podGroupAlgorithmResult{ + status: fwk.AsStatus(fmt.Errorf("failed to list node infos: %w", err)), + } } // TODO: kubernetes/enhancements#5732 - run placement generator plugins to get the set of placements @@ -503,17 +534,22 @@ func (sched *Scheduler) podGroupSchedulingPlacementAlgorithm(ctx context.Context logger.V(4).Info("Assuming placement in snapshot", "placement", placement.Name) err := sched.nodeInfoSnapshot.AssumePlacement(placement) if err != nil { - return sched.podGroupAlgorithmFailure(ctx, podGroupInfo, fwk.AsStatus(err)) + return podGroupAlgorithmResult{ + status: fwk.AsStatus(fmt.Errorf("failed to assume pod group placement: %w", err)), + } } result := sched.podGroupSchedulingDefaultAlgorithm(ctx, schedFwk, podGroupInfo) sched.nodeInfoSnapshot.ForgetPlacement() + if result.status.IsError() { + return result + } results[i] = placementResult{ podGroupAlgorithmResult: result, placement: placement, } - if result.status != podGroupUnschedulable { + if result.status.IsSuccess() || result.waitingOnPreemption { successfulResults = append(successfulResults, results[i]) } } @@ -528,25 +564,14 @@ func (sched *Scheduler) podGroupSchedulingPlacementAlgorithm(ctx context.Context return successfulResults[0].podGroupAlgorithmResult } -// podGroupAlgorithmFailure creates podGroupAlgorithmResult in cases where a podgroup-wide error occurred. -func (sched *Scheduler) podGroupAlgorithmFailure(ctx context.Context, podGroupInfo *framework.QueuedPodGroupInfo, status *fwk.Status) podGroupAlgorithmResult { - schedulingResult := podGroupAlgorithmResult{ - status: podGroupUnschedulable, - } - for _, podInfo := range podGroupInfo.QueuedPodInfos { - schedulingResult.podResults = append(schedulingResult.podResults, algorithmResult{ - podCtx: sched.initPodSchedulingContext(ctx, podInfo.Pod), - status: status, - }) - } - return schedulingResult -} - // podGroupSchedulingAlgorithm attempts to schedule pods in the pod group according to the policy and constraints and returns the scheduling result for each pod in the pod group. func (sched *Scheduler) podGroupSchedulingAlgorithm(ctx context.Context, schedFwk framework.Framework, podGroupInfo *framework.QueuedPodGroupInfo) podGroupAlgorithmResult { + podGroupCycleCtx, cancel := context.WithCancel(ctx) + defer cancel() + if utilfeature.DefaultFeatureGate.Enabled(features.TopologyAwareWorkloadScheduling) { - return sched.podGroupSchedulingPlacementAlgorithm(ctx, schedFwk, podGroupInfo) + return sched.podGroupSchedulingPlacementAlgorithm(podGroupCycleCtx, schedFwk, podGroupInfo) } else { - return sched.podGroupSchedulingDefaultAlgorithm(ctx, schedFwk, podGroupInfo) + return sched.podGroupSchedulingDefaultAlgorithm(podGroupCycleCtx, schedFwk, podGroupInfo) } } diff --git a/pkg/scheduler/schedule_one_podgroup_test.go b/pkg/scheduler/schedule_one_podgroup_test.go index 66db7079b5c..e680292d210 100644 --- a/pkg/scheduler/schedule_one_podgroup_test.go +++ b/pkg/scheduler/schedule_one_podgroup_test.go @@ -412,12 +412,13 @@ func TestPodGroupSchedulingAlgorithm(t *testing.T) { } tests := []struct { - name string - plugin *fakePodGroupPlugin - expectedGroupStatus podGroupAlgorithmStatus - expectedPodStatus map[string]*fwk.Status - expectedPreemption map[string]bool - skipForTAS bool + name string + plugin *fakePodGroupPlugin + expectedGroupStatusCode fwk.Code + expectedGroupWaitingOnPreemption bool + expectedPodStatus map[string]*fwk.Status + expectedPreemption map[string]bool + skipForTAS bool }{ { name: "All pods feasible", @@ -433,7 +434,7 @@ func TestPodGroupSchedulingAlgorithm(t *testing.T) { "p3": nil, }, }, - expectedGroupStatus: podGroupFeasible, + expectedGroupStatusCode: fwk.Success, expectedPodStatus: map[string]*fwk.Status{ "p1": nil, "p2": nil, @@ -454,7 +455,7 @@ func TestPodGroupSchedulingAlgorithm(t *testing.T) { "p3": nil, }, }, - expectedGroupStatus: podGroupFeasible, + expectedGroupStatusCode: fwk.Success, expectedPodStatus: map[string]*fwk.Status{ "p1": nil, "p2": nil, @@ -475,7 +476,7 @@ func TestPodGroupSchedulingAlgorithm(t *testing.T) { "p3": fwk.NewStatus(fwk.Wait), }, }, - expectedGroupStatus: podGroupUnschedulable, + expectedGroupStatusCode: fwk.Unschedulable, expectedPodStatus: map[string]*fwk.Status{ "p1": nil, "p2": nil, @@ -496,7 +497,7 @@ func TestPodGroupSchedulingAlgorithm(t *testing.T) { "p3": fwk.NewStatus(fwk.Wait), }, }, - expectedGroupStatus: podGroupFeasible, + expectedGroupStatusCode: fwk.Success, expectedPodStatus: map[string]*fwk.Status{ "p1": nil, "p2": nil, @@ -517,7 +518,7 @@ func TestPodGroupSchedulingAlgorithm(t *testing.T) { "p3": nil, }, }, - expectedGroupStatus: podGroupFeasible, + expectedGroupStatusCode: fwk.Success, expectedPodStatus: map[string]*fwk.Status{ "p1": nil, "p2": fwk.NewStatus(fwk.Unschedulable), @@ -548,7 +549,8 @@ func TestPodGroupSchedulingAlgorithm(t *testing.T) { "p3": nil, }, }, - expectedGroupStatus: podGroupWaitingOnPreemption, + expectedGroupStatusCode: fwk.Unschedulable, + expectedGroupWaitingOnPreemption: true, expectedPodStatus: map[string]*fwk.Status{ "p1": fwk.NewStatus(fwk.Unschedulable), "p2": fwk.NewStatus(fwk.Unschedulable), @@ -585,7 +587,7 @@ func TestPodGroupSchedulingAlgorithm(t *testing.T) { "p3": fwk.NewStatus(fwk.Wait), }, }, - expectedGroupStatus: podGroupUnschedulable, + expectedGroupStatusCode: fwk.Unschedulable, expectedPodStatus: map[string]*fwk.Status{ "p1": fwk.NewStatus(fwk.Unschedulable), "p2": fwk.NewStatus(fwk.Unschedulable), @@ -619,7 +621,8 @@ func TestPodGroupSchedulingAlgorithm(t *testing.T) { "p3": nil, }, }, - expectedGroupStatus: podGroupWaitingOnPreemption, + expectedGroupStatusCode: fwk.Unschedulable, + expectedGroupWaitingOnPreemption: true, expectedPodStatus: map[string]*fwk.Status{ "p1": fwk.NewStatus(fwk.Unschedulable), "p2": nil, @@ -654,7 +657,8 @@ func TestPodGroupSchedulingAlgorithm(t *testing.T) { "p3": nil, }, }, - expectedGroupStatus: podGroupWaitingOnPreemption, + expectedGroupStatusCode: fwk.Unschedulable, + expectedGroupWaitingOnPreemption: true, expectedPodStatus: map[string]*fwk.Status{ "p1": fwk.NewStatus(fwk.Unschedulable), "p2": fwk.NewStatus(fwk.Unschedulable), @@ -677,13 +681,103 @@ func TestPodGroupSchedulingAlgorithm(t *testing.T) { "p3": fwk.NewStatus(fwk.Unschedulable), }, }, - expectedGroupStatus: podGroupUnschedulable, + expectedGroupStatusCode: fwk.Unschedulable, expectedPodStatus: map[string]*fwk.Status{ "p1": fwk.NewStatus(fwk.Unschedulable), "p2": fwk.NewStatus(fwk.Unschedulable), "p3": fwk.NewStatus(fwk.Unschedulable), }, }, + { + name: "Any filter returned Error", + plugin: &fakePodGroupPlugin{ + filterStatus: map[string]*fwk.Status{ + "p1": nil, + "p2": fwk.NewStatus(fwk.Error), + "p3": nil, + }, + permitStatus: map[string]*fwk.Status{ + "p1": nil, + "p2": nil, + "p3": nil, + }, + }, + expectedGroupStatusCode: fwk.Error, + expectedPodStatus: map[string]*fwk.Status{ + "p1": nil, + "p2": fwk.NewStatus(fwk.Error), + // The algorithm stopped evaluating the pods after an error occurred, so a "p3" status is not expected. + }, + }, + { + name: "Any permit returned Error", + plugin: &fakePodGroupPlugin{ + filterStatus: map[string]*fwk.Status{ + "p1": nil, + "p2": nil, + "p3": nil, + }, + permitStatus: map[string]*fwk.Status{ + "p1": nil, + "p2": fwk.NewStatus(fwk.Error), + "p3": nil, + }, + }, + expectedGroupStatusCode: fwk.Error, + expectedPodStatus: map[string]*fwk.Status{ + "p1": nil, + "p2": fwk.NewStatus(fwk.Error), + // The algorithm stopped evaluating the pods after an error occurred, so a "p3" status is not expected. + }, + }, + { + name: "Any filter returned Error while waiting on preemption", + plugin: &fakePodGroupPlugin{ + filterStatus: map[string]*fwk.Status{ + "p1": fwk.NewStatus(fwk.Unschedulable), + "p2": fwk.NewStatus(fwk.Error), + "p3": nil, + }, + permitStatus: map[string]*fwk.Status{ + "p1": nil, + "p2": nil, + "p3": nil, + }, + postFilterResult: map[string]*fwk.PostFilterResult{ + "p1": {NominatingInfo: &fwk.NominatingInfo{NominatedNodeName: "node1", NominatingMode: fwk.ModeOverride}}, + }, + }, + expectedGroupStatusCode: fwk.Error, + expectedPodStatus: map[string]*fwk.Status{ + "p1": fwk.NewStatus(fwk.Unschedulable), + "p2": fwk.NewStatus(fwk.Error), + // The algorithm stopped evaluating the pods after an error occurred, so a "p3" status is not expected. + }, + }, + { + name: "Any permit returned Error while waiting on preemption", + plugin: &fakePodGroupPlugin{ + filterStatus: map[string]*fwk.Status{ + "p1": fwk.NewStatus(fwk.Unschedulable), + "p2": nil, + "p3": nil, + }, + permitStatus: map[string]*fwk.Status{ + "p1": nil, + "p2": fwk.NewStatus(fwk.Error), + "p3": nil, + }, + postFilterResult: map[string]*fwk.PostFilterResult{ + "p1": {NominatingInfo: &fwk.NominatingInfo{NominatedNodeName: "node1", NominatingMode: fwk.ModeOverride}}, + }, + }, + expectedGroupStatusCode: fwk.Error, + expectedPodStatus: map[string]*fwk.Status{ + "p1": fwk.NewStatus(fwk.Unschedulable), + "p2": fwk.NewStatus(fwk.Error), + // The algorithm stopped evaluating the pods after an error occurred, so a "p3" status is not expected. + }, + }, } for _, tasEnabled := range []bool{true, false} { @@ -750,8 +844,14 @@ func TestPodGroupSchedulingAlgorithm(t *testing.T) { result := sched.podGroupSchedulingAlgorithm(ctx, schedFwk, pgInfo) - if result.status != tt.expectedGroupStatus { - t.Errorf("Expected group status: %v, got: %v", tt.expectedGroupStatus, result.status) + if result.status.Code() != tt.expectedGroupStatusCode { + t.Errorf("Expected group status code: %v, got: %v", tt.expectedGroupStatusCode, result.status.Code()) + } + if result.waitingOnPreemption != tt.expectedGroupWaitingOnPreemption { + t.Errorf("Expected group waiting on preemption: %v, got: %v", tt.expectedGroupWaitingOnPreemption, result.waitingOnPreemption) + } + if len(tt.expectedPodStatus) != len(result.podResults) { + t.Errorf("Expected %d pod results, got %d", len(tt.expectedPodStatus), len(result.podResults)) } for i, podResult := range result.podResults { podName := pgInfo.QueuedPodInfos[i].Pod.Name @@ -759,6 +859,8 @@ func TestPodGroupSchedulingAlgorithm(t *testing.T) { if podResult.status.Code() != expected.Code() { t.Errorf("Expected pod %s status code: %v, got: %v", podName, expected.Code(), podResult.status.Code()) } + } else { + t.Errorf("Got result for unexpected pod %s: %v", podName, podResult.status.Code()) } if podResult.status.IsSuccess() || podResult.requiresPreemption { if podResult.scheduleResult.SuggestedHost != "node1" { @@ -817,7 +919,7 @@ func TestSubmitPodGroupAlgorithmResult(t *testing.T) { { name: "All pods feasible", algorithmResult: podGroupAlgorithmResult{ - status: podGroupFeasible, + status: nil, podResults: []algorithmResult{{ scheduleResult: ScheduleResult{SuggestedHost: "node1"}, status: nil, @@ -837,7 +939,7 @@ func TestSubmitPodGroupAlgorithmResult(t *testing.T) { { name: "All pods feasible, but all waiting", algorithmResult: podGroupAlgorithmResult{ - status: podGroupUnschedulable, + status: fwk.NewStatus(fwk.Unschedulable), podResults: []algorithmResult{{ scheduleResult: ScheduleResult{SuggestedHost: "node1"}, status: nil, @@ -857,7 +959,7 @@ func TestSubmitPodGroupAlgorithmResult(t *testing.T) { { name: "All pods feasible, but last waiting", algorithmResult: podGroupAlgorithmResult{ - status: podGroupFeasible, + status: nil, podResults: []algorithmResult{{ scheduleResult: ScheduleResult{SuggestedHost: "node1"}, status: nil, @@ -877,7 +979,7 @@ func TestSubmitPodGroupAlgorithmResult(t *testing.T) { { name: "All pods feasible, one waiting, one unschedulable", algorithmResult: podGroupAlgorithmResult{ - status: podGroupFeasible, + status: nil, podResults: []algorithmResult{{ scheduleResult: ScheduleResult{SuggestedHost: "node1"}, status: nil, @@ -897,7 +999,8 @@ func TestSubmitPodGroupAlgorithmResult(t *testing.T) { { name: "All pods require preemption", algorithmResult: podGroupAlgorithmResult{ - status: podGroupWaitingOnPreemption, + status: fwk.NewStatus(fwk.Unschedulable), + waitingOnPreemption: true, podResults: []algorithmResult{{ scheduleResult: ScheduleResult{ SuggestedHost: "node1", @@ -929,7 +1032,7 @@ func TestSubmitPodGroupAlgorithmResult(t *testing.T) { { name: "All pods require preemption, but waiting", algorithmResult: podGroupAlgorithmResult{ - status: podGroupUnschedulable, + status: fwk.NewStatus(fwk.Unschedulable), podResults: []algorithmResult{{ scheduleResult: ScheduleResult{ SuggestedHost: "node1", @@ -961,7 +1064,8 @@ func TestSubmitPodGroupAlgorithmResult(t *testing.T) { { name: "One pod requires preemption, but waiting, two are feasible", algorithmResult: podGroupAlgorithmResult{ - status: podGroupWaitingOnPreemption, + status: fwk.NewStatus(fwk.Unschedulable), + waitingOnPreemption: true, podResults: []algorithmResult{{ scheduleResult: ScheduleResult{ SuggestedHost: "node1", @@ -985,7 +1089,8 @@ func TestSubmitPodGroupAlgorithmResult(t *testing.T) { { name: "One pod unschedulable, one requires preemption, one feasible", algorithmResult: podGroupAlgorithmResult{ - status: podGroupWaitingOnPreemption, + status: fwk.NewStatus(fwk.Unschedulable), + waitingOnPreemption: true, podResults: []algorithmResult{{ scheduleResult: ScheduleResult{ SuggestedHost: "node1", @@ -1008,7 +1113,7 @@ func TestSubmitPodGroupAlgorithmResult(t *testing.T) { { name: "All pods unschedulable", algorithmResult: podGroupAlgorithmResult{ - status: podGroupUnschedulable, + status: fwk.NewStatus(fwk.Unschedulable), podResults: []algorithmResult{{ scheduleResult: ScheduleResult{SuggestedHost: "", nominatingInfo: clearNominatedNode}, status: fwk.NewStatus(fwk.Unschedulable), @@ -1026,7 +1131,7 @@ func TestSubmitPodGroupAlgorithmResult(t *testing.T) { { name: "All pods unschedulable with nil nominatingInfo", algorithmResult: podGroupAlgorithmResult{ - status: podGroupUnschedulable, + status: fwk.NewStatus(fwk.Unschedulable), podResults: []algorithmResult{{ scheduleResult: ScheduleResult{SuggestedHost: "", nominatingInfo: nil}, status: fwk.NewStatus(fwk.Unschedulable), @@ -1041,6 +1146,51 @@ func TestSubmitPodGroupAlgorithmResult(t *testing.T) { expectBound: sets.New[string](), expectFailed: sets.New("p1", "p2", "p3"), }, + { + name: "Unschedulable for the entire pod group", + algorithmResult: podGroupAlgorithmResult{ + status: fwk.NewStatus(fwk.Unschedulable), + podResults: []algorithmResult{}, + }, + expectBound: sets.New[string](), + expectFailed: sets.New("p1", "p2", "p3"), + }, + { + name: "Error for one pod", + algorithmResult: podGroupAlgorithmResult{ + status: fwk.NewStatus(fwk.Error), + podResults: []algorithmResult{{ + scheduleResult: ScheduleResult{SuggestedHost: "node1"}, + status: nil, + permitStatus: nil, + }, { + scheduleResult: ScheduleResult{SuggestedHost: "", nominatingInfo: clearNominatedNode}, + status: fwk.NewStatus(fwk.Error), + }}, + }, + expectBound: sets.New[string](), + expectFailed: sets.New("p1", "p2", "p3"), + }, + { + name: "Error for one pod while waiting on preemption", + algorithmResult: podGroupAlgorithmResult{ + status: fwk.NewStatus(fwk.Error), + podResults: []algorithmResult{{ + scheduleResult: ScheduleResult{ + SuggestedHost: "node1", + nominatingInfo: &fwk.NominatingInfo{NominatedNodeName: "node1", NominatingMode: fwk.ModeOverride}, + }, + status: fwk.NewStatus(fwk.Unschedulable), + requiresPreemption: true, + permitStatus: nil, + }, { + scheduleResult: ScheduleResult{SuggestedHost: "", nominatingInfo: clearNominatedNode}, + status: fwk.NewStatus(fwk.Error), + }}, + }, + expectBound: sets.New[string](), + expectFailed: sets.New("p1", "p2", "p3"), + }, } for _, tt := range tests { @@ -1109,8 +1259,9 @@ func TestSubmitPodGroupAlgorithmResult(t *testing.T) { }, } - for i := range pgInfo.QueuedPodInfos { - podCtx := sched.initPodSchedulingContext(ctx, p1) + for i := range tt.algorithmResult.podResults { + pod := pgInfo.QueuedPodInfos[i].Pod + podCtx := initPodSchedulingContext(ctx, pod) tt.algorithmResult.podResults[i].podCtx = podCtx } diff --git a/staging/src/k8s.io/kube-scheduler/framework/interface.go b/staging/src/k8s.io/kube-scheduler/framework/interface.go index a3dd45d5af9..b89c14351ff 100644 --- a/staging/src/k8s.io/kube-scheduler/framework/interface.go +++ b/staging/src/k8s.io/kube-scheduler/framework/interface.go @@ -22,6 +22,7 @@ import ( "context" "errors" "math" + "slices" "strings" "time" @@ -192,6 +193,11 @@ func (s *Status) IsRejected() bool { return code == Unschedulable || code == UnschedulableAndUnresolvable || code == Pending } +// IsError returns true if and only if "Status" is non-nil and its Code is "Error". +func (s *Status) IsError() bool { + return s.Code() == Error +} + // AsError returns nil if the status is a success, a wait or a skip; otherwise returns an "error" object // with a concatenated message on reasons of the Status. func (s *Status) AsError() error { @@ -226,6 +232,16 @@ func (s *Status) String() string { return s.Message() } +// Clone clones the entire Status and returns a copy. +func (s *Status) Clone() *Status { + return &Status{ + code: s.code, + reasons: slices.Clone(s.reasons), + err: s.err, + plugin: s.plugin, + } +} + // NewStatus makes a Status out of the given arguments and returns its pointer. func NewStatus(code Code, reasons ...string) *Status { s := &Status{