diff --git a/.changes/unreleased/Fixed-20260722-035021.yaml b/.changes/unreleased/Fixed-20260722-035021.yaml new file mode 100644 index 000000000..4e63ceebf --- /dev/null +++ b/.changes/unreleased/Fixed-20260722-035021.yaml @@ -0,0 +1,3 @@ +kind: Fixed +body: |- + Pipeline-only allocation no longer leaks simulation fit errors onto pending jobs, preserving authoritative scheduling diagnostics. [#1948](https://github.com/kai-scheduler/KAI-Scheduler/issues/1948) diff --git a/pkg/scheduler/actions/common/allocate.go b/pkg/scheduler/actions/common/allocate.go index 41ce228c6..ace91ca60 100644 --- a/pkg/scheduler/actions/common/allocate.go +++ b/pkg/scheduler/actions/common/allocate.go @@ -140,7 +140,10 @@ func allocateTasksOnNodeSet(ssn *framework.Session, stmt *framework.Statement, n for index, task := range tasksToAllocate { success := allocateTask(ssn, stmt, nodes, task, isPipelineOnly) if !success { - handleFailedTaskAllocation(job, task, index) + // Skip job fit errors during pipeline-only simulations; see allocateTask (#1948). + if !isPipelineOnly { + handleFailedTaskAllocation(job, task, index) + } return false } } @@ -159,9 +162,15 @@ func allocateTask(ssn *framework.Session, stmt *framework.Statement, nodes []*no log.InfraLogger.V(6).Infof("pre-predicates failed on task %s/%s. Error: %v", task.Namespace, task.Name, err) - fitErrors := common_info.NewFitErrors() - fitErrors.SetError(err.Error()) - job.AddTaskFitErrors(task, fitErrors) + // Pipeline-only allocation runs inside solver simulations that are rolled back. + // Statement.Rollback does not restore fit errors, so recording them here would leak + // simulation-only diagnostics onto the live job and mask the authoritative Allocate + // reason (#1948). Mirror FittingNode's writeFittingDelta=!isPipelineOnly behavior. + if !isPipelineOnly { + fitErrors := common_info.NewFitErrors() + fitErrors.SetError(err.Error()) + job.AddTaskFitErrors(task, fitErrors) + } return false } diff --git a/pkg/scheduler/actions/integration_tests/reclaim/reclaim_fit_errors_test.go b/pkg/scheduler/actions/integration_tests/reclaim/reclaim_fit_errors_test.go new file mode 100644 index 000000000..59b8e6332 --- /dev/null +++ b/pkg/scheduler/actions/integration_tests/reclaim/reclaim_fit_errors_test.go @@ -0,0 +1,160 @@ +// Copyright 2026 NVIDIA CORPORATION +// SPDX-License-Identifier: Apache-2.0 + +package reclaim + +import ( + "errors" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.uber.org/mock/gomock" + "gopkg.in/h2non/gock.v1" + + "github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/api/common_info" + "github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/api/node_info" + "github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/api/pod_info" + "github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/api/pod_status" + "github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/api/podgroup_info" + "github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/constants" + "github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/test_utils" + "github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/test_utils/jobs_fake" + "github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/test_utils/nodes_fake" + "github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/test_utils/tasks_fake" +) + +func TestSchedulingCyclePreservesAllocateFitErrors(t *testing.T) { + defer gock.Off() + + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + ssn := test_utils.BuildSession(buildFailedGangReclaimTopology(), ctrl) + ssn.AddNodeOrderFn(func(_ *pod_info.PodInfo, node *node_info.NodeInfo) (float64, error) { + if node.Name == "node0" { + return 1000, nil + } + return 0, nil + }) + failedPrePredicateCalls := 0 + ssn.AddPrePredicateFn(func( + task *pod_info.PodInfo, + _ *podgroup_info.PodGroupInfo, + ) error { + if task.Job != common_info.PodGroupID("pending-gang") || task.Name != "pending-gang-0" { + return nil + } + releasingTasks := 0 + for _, job := range ssn.ClusterInfo.PodGroupInfos { + releasingTasks += len(job.PodStatusIndex[pod_status.Releasing]) + } + if releasingTasks < 2 { + return nil + } + failedPrePredicateCalls++ + return errors.New("simulated pre-predicate failure") + }) + onJobSolutionStartCalls := 0 + ssn.AddOnJobSolutionStartFn(func() { + onJobSolutionStartCalls++ + }) + actions := manySingleGPUJobsSchedulingCycleActions() + actions[0].Execute(ssn) + + job := ssn.ClusterInfo.PodGroupInfos[common_info.PodGroupID("pending-gang")] + require.NotNil(t, job) + require.NotEmpty(t, job.TasksFitErrors) + require.NotEmpty(t, job.JobFitErrors) + require.NotContains(t, job.TasksFitErrors, common_info.PodID("pending-gang-0")) + require.Zero(t, failedPrePredicateCalls) + + allocateTaskErrors := taskFitErrorMessages(job) + allocateJobErrors := jobFitErrorMessages(job) + firstTaskFitErrorMutation := "" + firstJobFitErrorMutation := "" + for _, action := range actions[1:] { + action.Execute(ssn) + currentJob := ssn.ClusterInfo.PodGroupInfos[common_info.PodGroupID("pending-gang")] + if firstTaskFitErrorMutation == "" && + !assert.ObjectsAreEqual(allocateTaskErrors, taskFitErrorMessages(currentJob)) { + firstTaskFitErrorMutation = string(action.Name()) + } + if firstJobFitErrorMutation == "" && + !assert.ObjectsAreEqual(allocateJobErrors, jobFitErrorMessages(currentJob)) { + firstJobFitErrorMutation = string(action.Name()) + } + } + + job = ssn.ClusterInfo.PodGroupInfos[common_info.PodGroupID("pending-gang")] + assert.Equalf(t, allocateTaskErrors, taskFitErrorMessages(job), + "task fit errors produced by Allocate were first changed by %s", firstTaskFitErrorMutation) + assert.Equalf(t, allocateJobErrors, jobFitErrorMessages(job), + "job fit errors produced by Allocate were first changed by %s", firstJobFitErrorMutation) + require.Len(t, job.PodStatusIndex[pod_status.Pending], 2) + require.NotZero(t, onJobSolutionStartCalls) + require.NotZero(t, failedPrePredicateCalls) +} + +func buildFailedGangReclaimTopology() test_utils.TestTopologyBasic { + return test_utils.TestTopologyBasic{ + Name: "failed gang reclaim preserves allocate fit errors", + Nodes: map[string]nodes_fake.TestNodeBasic{ + "node0": {GPUs: 1}, + "node1": {GPUs: 1}, + }, + Jobs: []*jobs_fake.TestJobBasic{ + { + Name: "running-victim", + RequiredGPUsPerTask: 1, + Priority: constants.PriorityTrainNumber, + QueueName: "running-queue", + Tasks: []*tasks_fake.TestTaskBasic{{ + NodeName: "node0", + State: pod_status.Releasing, + }}, + }, + { + Name: "running-victim-2", + RequiredGPUsPerTask: 1, + Priority: constants.PriorityTrainNumber, + QueueName: "running-queue", + Tasks: []*tasks_fake.TestTaskBasic{{ + NodeName: "node1", + State: pod_status.Running, + }}, + }, + { + Name: "pending-gang", + RequiredGPUsPerTask: 1, + Priority: constants.PriorityTrainNumber, + QueueName: "pending-queue", + Tasks: []*tasks_fake.TestTaskBasic{ + {State: pod_status.Pending}, + {State: pod_status.Pending}, + }, + }, + }, + Queues: []test_utils.TestQueueBasic{ + {Name: "running-queue", DeservedGPUs: 0, GPUOverQuotaWeight: 0}, + {Name: "pending-queue", DeservedGPUs: 2, GPUOverQuotaWeight: 0}, + }, + Mocks: &test_utils.TestMock{CacheRequirements: &test_utils.CacheMocking{}}, + } +} + +func taskFitErrorMessages(job *podgroup_info.PodGroupInfo) map[common_info.PodID]string { + messages := make(map[common_info.PodID]string, len(job.TasksFitErrors)) + for taskID, fitErrors := range job.TasksFitErrors { + messages[taskID] = fitErrors.Error() + } + return messages +} + +func jobFitErrorMessages(job *podgroup_info.PodGroupInfo) []string { + messages := make([]string, 0, len(job.JobFitErrors)) + for _, fitError := range job.JobFitErrors { + messages = append(messages, fitError.DetailedMessage()) + } + return messages +}