Skip to content

Commit 87107b1

Browse files
committed
feat(scheduler): gate NvFractions on node readiness
Signed-off-by: davidLif <davidshani12@gmail.com>
1 parent 2fbf3a3 commit 87107b1

13 files changed

Lines changed: 476 additions & 28 deletions

File tree

cmd/scheduler/app/options/options.go

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -76,6 +76,7 @@ type ServerOption struct {
7676
GPUWorkerNodeLabelKey string
7777
MIGWorkerNodeLabelKey string
7878
QueueLabelKey string
79+
GpuSharingMode string
7980
Namspace string
8081

8182
JSONLog bool
@@ -100,6 +101,7 @@ func (s *ServerOption) AddFlags(fs *pflag.FlagSet) {
100101
fs.BoolVar(&s.RestrictSchedulingNodes, "restrict-node-scheduling", false, "kai-scheduler will allocate jobs only to restricted nodes")
101102
fs.StringVar(&s.NodePoolLabelKey, "nodepool-label-key", constants.DefaultNodePoolLabelKey, "The label key by which to filter scheduling nodepool")
102103
fs.StringVar(&s.QueueLabelKey, "queue-label-key", constants.DefaultQueueLabel, "The label key for the queue")
104+
fs.StringVar(&s.GpuSharingMode, "gpu-sharing-mode", "", "The effective GPU-sharing mode resolved from the KAI config")
103105
fs.StringVar(&s.NodePoolLabelValue, "partition-label-value", "", "The label value by which to filter scheduling partition")
104106
fs.StringVar(&s.SchedulerConf, "scheduler-conf", "", "The absolute path of scheduler configuration file")
105107
fs.DurationVar(&s.SchedulePeriod, "schedule-period", defaultSchedulerPeriod, "The period between each scheduling cycle")

cmd/scheduler/app/server.go

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -47,6 +47,7 @@ import (
4747

4848
"github.com/kai-scheduler/KAI-scheduler/cmd/scheduler/app/options"
4949
"github.com/kai-scheduler/KAI-scheduler/cmd/scheduler/profiling"
50+
kaiv1common "github.com/kai-scheduler/KAI-scheduler/pkg/apis/kai/v1/common"
5051
"github.com/kai-scheduler/KAI-scheduler/pkg/scheduler"
5152
"github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/actions"
5253
"github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/conf"
@@ -96,7 +97,7 @@ func BuildSchedulerParams(opt *options.ServerOption) *conf.SchedulerParams {
9697
NodePoolLabelValue: opt.NodePoolLabelValue,
9798
}
9899

99-
return &conf.SchedulerParams{
100+
params := &conf.SchedulerParams{
100101
SchedulerName: opt.SchedulerName,
101102
RestrictSchedulingNodes: opt.RestrictSchedulingNodes,
102103
PartitionParams: schedulingPartitionParams,
@@ -113,6 +114,11 @@ func BuildSchedulerParams(opt *options.ServerOption) *conf.SchedulerParams {
113114
UpdatePodEvictionCondition: opt.UpdatePodEvictionCondition,
114115
QueueLabelKey: opt.QueueLabelKey,
115116
}
117+
if opt.GpuSharingMode != "" {
118+
gpuSharingMode := kaiv1common.GpuSharingMode(opt.GpuSharingMode)
119+
params.GpuSharingMode = &gpuSharingMode
120+
}
121+
return params
116122
}
117123

118124
func RunApp() error {

pkg/common/constants/constants.go

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -82,6 +82,9 @@ const (
8282
NvFractionsMemoryRequestSuffix = ".gpu-memory.request"
8383
NvFractionsMemoryLimitSuffix = ".gpu-memory.limit"
8484
NvFractionsVisibleDevicesSuffix = ".gpus.devices"
85+
86+
// gpu-sharing operator statuses
87+
NvFractionNodeReadyConditionType = "gpu-sharing.nvidia.com/Ready"
8588
)
8689

8790
// QueueValidatedVersions returns the list of queue versions that we validate with a webhook. This will be used by the

pkg/operator/operands/scheduler/resources_for_shard.go

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -363,6 +363,10 @@ func buildArgsList(
363363
fmt.Sprintf("--%s=%s", "queue-label-key", *kaiConfig.Spec.Global.QueueLabelKey),
364364
}
365365

366+
if kaiConfig.Spec.Global.GpuSharingMode != nil {
367+
args = append(args, fmt.Sprintf("--%s=%s", "gpu-sharing-mode", *kaiConfig.Spec.Global.GpuSharingMode))
368+
}
369+
366370
if kaiConfig.Spec.Scheduler.SchedulerService.Port != nil {
367371
portNumberString := strconv.Itoa(*kaiConfig.Spec.Scheduler.SchedulerService.Port)
368372
args = append(args, fmt.Sprintf("--%s=:%s", "listen-address", portNumberString))

pkg/operator/operands/scheduler/resources_test.go

Lines changed: 26 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,7 @@ import (
1515

1616
"github.com/kai-scheduler/KAI-scheduler/cmd/scheduler/app/options"
1717
kaiv1 "github.com/kai-scheduler/KAI-scheduler/pkg/apis/kai/v1"
18-
"github.com/kai-scheduler/KAI-scheduler/pkg/apis/kai/v1/common"
18+
kaiv1common "github.com/kai-scheduler/KAI-scheduler/pkg/apis/kai/v1/common"
1919
kaiprometheus "github.com/kai-scheduler/KAI-scheduler/pkg/apis/kai/v1/prometheus"
2020
kaiv1qc "github.com/kai-scheduler/KAI-scheduler/pkg/apis/kai/v1/queue_controller"
2121
kaiv1scheduler "github.com/kai-scheduler/KAI-scheduler/pkg/apis/kai/v1/scheduler"
@@ -434,6 +434,30 @@ func TestBuildArgsList(t *testing.T) {
434434
"log-json": "true",
435435
},
436436
},
437+
{
438+
name: "with gpu sharing mode from global config",
439+
config: &kaiv1.Config{
440+
Spec: kaiv1.ConfigSpec{
441+
Global: &kaiv1.GlobalConfig{
442+
SchedulerName: ptr.To("test-scheduler"),
443+
GpuSharingMode: ptr.To(kaiv1common.GpuSharingModeNvFractions),
444+
},
445+
Namespace: "kai-system",
446+
Scheduler: &kaiv1scheduler.Scheduler{
447+
Replicas: ptr.To(int32(1)),
448+
},
449+
},
450+
},
451+
shard: &kaiv1.SchedulingShard{
452+
Spec: kaiv1.SchedulingShardSpec{},
453+
},
454+
expected: map[string]string{
455+
"scheduler-conf": "config.yaml",
456+
"scheduler-name": "test-scheduler",
457+
"namespace": "kai-system",
458+
"gpu-sharing-mode": "NvFractions",
459+
},
460+
},
437461
}
438462

439463
for _, tt := range tests {
@@ -1498,7 +1522,7 @@ func TestPodDisruptionBudgetForShard(t *testing.T) {
14981522
config := &kaiv1.Config{}
14991523
config.Spec.SetDefaultsWhereNeeded()
15001524
config.Spec.Scheduler.Replicas = ptr.To(tt.replicas)
1501-
config.Spec.Scheduler.Service.PodDisruptionBudget = &common.PodDisruptionBudget{
1525+
config.Spec.Scheduler.Service.PodDisruptionBudget = &kaiv1common.PodDisruptionBudget{
15021526
Enabled: ptr.To(tt.pdbEnabled),
15031527
MaxUnavailable: ptr.To(tt.maxUnavailable),
15041528
}

pkg/scheduler/actions/allocate/allocateFractionalGpu_test.go

Lines changed: 73 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,11 @@ import (
88

99
. "go.uber.org/mock/gomock"
1010
"gopkg.in/h2non/gock.v1"
11+
v1 "k8s.io/api/core/v1"
12+
"k8s.io/utils/ptr"
1113

14+
kaiv1common "github.com/kai-scheduler/KAI-scheduler/pkg/apis/kai/v1/common"
15+
commonconstants "github.com/kai-scheduler/KAI-scheduler/pkg/common/constants"
1216
"github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/actions/allocate"
1317
"github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/actions/integration_tests/integration_tests_utils"
1418
"github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/api/pod_status"
@@ -38,6 +42,75 @@ func TestHandleFractionalGPUAllocation(t *testing.T) {
3842
}
3943
}
4044

45+
func TestFractionalGPUAllocationUsesNodeConditionOverride(t *testing.T) {
46+
test_utils.InitTestingInfrastructure()
47+
controller := NewController(t)
48+
defer controller.Finish()
49+
defer gock.Off()
50+
51+
notReadyConditions := []v1.NodeCondition{
52+
{
53+
Type: v1.NodeConditionType(commonconstants.NvFractionNodeReadyConditionType),
54+
Status: v1.ConditionFalse,
55+
Reason: "DevicePluginNotReady",
56+
},
57+
}
58+
topology := test_utils.TestTopologyBasic{
59+
Name: "fractional pod cannot allocate on gpu sharing unready node",
60+
Jobs: []*jobs_fake.TestJobBasic{
61+
{
62+
Name: "pending_job0",
63+
RequiredGpuMemory: 50,
64+
Priority: constants.PriorityTrainNumber,
65+
QueueName: "queue0",
66+
Tasks: []*tasks_fake.TestTaskBasic{
67+
{
68+
State: pod_status.Pending,
69+
},
70+
},
71+
},
72+
},
73+
Nodes: map[string]nodes_fake.TestNodeBasic{
74+
"node0": {
75+
GPUs: 1,
76+
Conditions: ptr.To(notReadyConditions),
77+
},
78+
"node1": {
79+
GPUs: 1,
80+
Conditions: ptr.To(notReadyConditions),
81+
},
82+
},
83+
Queues: []test_utils.TestQueueBasic{
84+
{
85+
Name: "queue0",
86+
DeservedGPUs: 1,
87+
},
88+
},
89+
JobExpectedResults: map[string]test_utils.TestExpectedResultBasic{
90+
"pending_job0": {
91+
Status: pod_status.Pending,
92+
ExpectedErrorMessage: "\nPodSchedulingErrors.\nResources were not found for pod /pending_job0-0 due to: " +
93+
"no nodes with enough resources were found: 2 node is not ready for fractional GPU scheduling. " +
94+
"Condition gpu-sharing.nvidia.com/Ready is False. Reason: DevicePluginNotReady. Message: ..",
95+
},
96+
},
97+
Mocks: &test_utils.TestMock{
98+
SchedulerConf: &conf.SchedulerConfiguration{
99+
Actions: "allocate, consolidation, reclaim, preempt, stalegangeviction",
100+
},
101+
CacheRequirements: &test_utils.CacheMocking{
102+
NumberOfCacheBinds: 0,
103+
},
104+
},
105+
}
106+
107+
ssn := test_utils.BuildSession(topology, controller)
108+
ssn.SchedulerParams.GpuSharingMode = ptr.To(kaiv1common.GpuSharingModeNvFractions)
109+
allocate.New().Execute(ssn)
110+
111+
test_utils.MatchExpectedAndRealTasks(t, 0, topology, ssn)
112+
}
113+
41114
func getFractionalGPUTestsMetadata() []integration_tests_utils.TestTopologyMetadata {
42115
return []integration_tests_utils.TestTopologyMetadata{
43116
{

pkg/scheduler/conf/scheduler_conf.go

Lines changed: 17 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -26,25 +26,27 @@ import (
2626
"k8s.io/apimachinery/pkg/selection"
2727

2828
kaiv1 "github.com/kai-scheduler/KAI-scheduler/pkg/apis/kai/v1"
29+
kaiv1common "github.com/kai-scheduler/KAI-scheduler/pkg/apis/kai/v1/common"
2930
usagedbapi "github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/cache/usagedb/api"
3031
)
3132

3233
type SchedulerParams struct {
33-
SchedulerName string `json:"schedulerName,omitempty"`
34-
RestrictSchedulingNodes bool `json:"restrictSchedulingNodes,omitempty"`
35-
PartitionParams *SchedulingNodePoolParams `json:"partitionParams,omitempty"`
36-
MaxNumberConsolidationPreemptees int `json:"maxNumberConsolidationPreemptees,omitempty"`
37-
ScheduleCSIStorage bool `json:"scheduleCSIStorage,omitempty"`
38-
UseSchedulingSignatures bool `json:"useSchedulingSignatures,omitempty"`
39-
FullHierarchyFairness bool `json:"fullHierarchyFairness,omitempty"`
40-
AllowConsolidatingReclaim bool `json:"allowConsolidatingReclaim,omitempty"`
41-
NumOfStatusRecordingWorkers int `json:"numOfStatusRecordingWorkers,omitempty"`
42-
GlobalDefaultStalenessGracePeriod time.Duration `json:"globalDefaultStalenessGracePeriod,omitempty"`
43-
SchedulePeriod time.Duration `json:"schedulePeriod,omitempty"`
44-
StuckInReleasingThreshold time.Duration `json:"stuckInReleasingThreshold,omitempty"`
45-
DetailedFitErrors bool `json:"detailedFitErrors,omitempty"`
46-
UpdatePodEvictionCondition bool `json:"updatePodEvictionCondition,omitempty"`
47-
QueueLabelKey string `json:"queueLabelKey,omitempty"`
34+
SchedulerName string `json:"schedulerName,omitempty"`
35+
RestrictSchedulingNodes bool `json:"restrictSchedulingNodes,omitempty"`
36+
PartitionParams *SchedulingNodePoolParams `json:"partitionParams,omitempty"`
37+
GpuSharingMode *kaiv1common.GpuSharingMode `json:"gpuSharingMode,omitempty"`
38+
MaxNumberConsolidationPreemptees int `json:"maxNumberConsolidationPreemptees,omitempty"`
39+
ScheduleCSIStorage bool `json:"scheduleCSIStorage,omitempty"`
40+
UseSchedulingSignatures bool `json:"useSchedulingSignatures,omitempty"`
41+
FullHierarchyFairness bool `json:"fullHierarchyFairness,omitempty"`
42+
AllowConsolidatingReclaim bool `json:"allowConsolidatingReclaim,omitempty"`
43+
NumOfStatusRecordingWorkers int `json:"numOfStatusRecordingWorkers,omitempty"`
44+
GlobalDefaultStalenessGracePeriod time.Duration `json:"globalDefaultStalenessGracePeriod,omitempty"`
45+
SchedulePeriod time.Duration `json:"schedulePeriod,omitempty"`
46+
StuckInReleasingThreshold time.Duration `json:"stuckInReleasingThreshold,omitempty"`
47+
DetailedFitErrors bool `json:"detailedFitErrors,omitempty"`
48+
UpdatePodEvictionCondition bool `json:"updatePodEvictionCondition,omitempty"`
49+
QueueLabelKey string `json:"queueLabelKey,omitempty"`
4850
}
4951

5052
// SchedulerConfiguration defines the configuration of scheduler.
Lines changed: 136 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,136 @@
1+
// Copyright 2025 NVIDIA CORPORATION
2+
// SPDX-License-Identifier: Apache-2.0
3+
4+
package gpusharingnodevalidation
5+
6+
import (
7+
"fmt"
8+
"strconv"
9+
10+
v1 "k8s.io/api/core/v1"
11+
12+
kaiv1common "github.com/kai-scheduler/KAI-scheduler/pkg/apis/kai/v1/common"
13+
"github.com/kai-scheduler/KAI-scheduler/pkg/common/constants"
14+
"github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/api"
15+
"github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/api/common_info"
16+
"github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/api/node_info"
17+
"github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/api/pod_info"
18+
"github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/api/resource_info"
19+
"github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/framework"
20+
"github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/gpu_sharing"
21+
)
22+
23+
const (
24+
fractionalGPUReadyConditionType = v1.NodeConditionType(constants.NvFractionNodeReadyConditionType)
25+
fractionalGPUReadyReasonMissing = "ConditionNotFound"
26+
)
27+
28+
func Validate(task *pod_info.PodInfo, node *node_info.NodeInfo, ssn *framework.Session) error {
29+
if err := checkMaxPodsWithGpuGroupReservation(task, node, ssn); err != nil {
30+
return err
31+
}
32+
33+
if err := checkNvFractionalGPUReadyCondition(task, node, isNvFractionsMode(ssn)); err != nil {
34+
return err
35+
}
36+
37+
return nil
38+
}
39+
40+
// Check if the gpu-sharing operator set this node as ready. This is relevant only for NvFractions mode.
41+
func checkNvFractionalGPUReadyCondition(task *pod_info.PodInfo, node *node_info.NodeInfo, nvFractionsMode bool) error {
42+
if !nvFractionsMode || !task.IsSharedGPURequest() {
43+
return nil
44+
}
45+
46+
conditionStatus := v1.ConditionUnknown
47+
conditionReason := fractionalGPUReadyReasonMissing
48+
for _, condition := range node.Node.Status.Conditions {
49+
if condition.Type != fractionalGPUReadyConditionType {
50+
continue
51+
}
52+
if condition.Status == v1.ConditionTrue {
53+
return nil
54+
}
55+
conditionStatus = condition.Status
56+
if condition.Reason != "" {
57+
conditionReason = condition.Reason
58+
}
59+
break
60+
}
61+
62+
return common_info.NewFitError(task.Name, task.Namespace, node.Name, fmt.Sprintf(
63+
"node is not ready for fractional GPU scheduling. Condition %s is %s. Reason: %s",
64+
fractionalGPUReadyConditionType, conditionStatus, conditionReason,
65+
))
66+
}
67+
68+
func isNvFractionsMode(ssn *framework.Session) bool {
69+
return ssn != nil && ssn.SchedulerParams.GpuSharingMode != nil &&
70+
*ssn.SchedulerParams.GpuSharingMode == kaiv1common.GpuSharingModeNvFractions
71+
}
72+
73+
func checkMaxPodsWithGpuGroupReservation(
74+
task *pod_info.PodInfo, node *node_info.NodeInfo, ssn *framework.Session) error {
75+
availablePods := node.IdleVector.Get(resource_info.PodsIndex) + node.ReleasingVector.Get(resource_info.PodsIndex)
76+
77+
if !task.IsSharedGPURequest() {
78+
if availablePods > 0 {
79+
return nil
80+
}
81+
return common_info.NewFitError(task.Name, task.Namespace, node.Name, api.NodePodNumberExceeded)
82+
}
83+
84+
needsNewGpuGroup := willCreateNewGpuGroup(task, node, ssn)
85+
if !needsNewGpuGroup {
86+
return nil
87+
}
88+
89+
if availablePods < 2 {
90+
return common_info.NewFitError(task.Name, task.Namespace, node.Name, api.NodePodNumberExceeded)
91+
}
92+
93+
return nil
94+
}
95+
96+
// willCreateNewGpuGroup determines if allocating this task will create a new GPU group
97+
// and thus require a new reservation pod.
98+
func willCreateNewGpuGroup(task *pod_info.PodInfo, node *node_info.NodeInfo, ssn *framework.Session) bool {
99+
if ssn == nil {
100+
return true
101+
}
102+
103+
fittingGPUs := ssn.FittingGPUs(node, task)
104+
gpuForSharingImmediate := gpu_sharing.GetNodePreferableGpuForSharing(fittingGPUs, node, task, false)
105+
106+
if gpuForSharingImmediate != nil && !gpuForSharingImmediate.IsReleasing {
107+
return containsNewGpuGroup(gpuForSharingImmediate.GroupIDs())
108+
}
109+
110+
gpuForSharingPipelined := gpu_sharing.GetNodePreferableGpuForSharing(fittingGPUs, node, task, true)
111+
112+
if gpuForSharingPipelined != nil {
113+
return containsNewGpuGroup(gpuForSharingPipelined.GroupIDs())
114+
}
115+
116+
// No GPU assignment possible - conservatively assume new group would be needed
117+
return true
118+
}
119+
120+
// containsNewGpuGroup checks if any of the GPU groups is a newly created one (UUID format).
121+
func containsNewGpuGroup(groups []string) bool {
122+
for _, gpuGroup := range groups {
123+
if isNewGpuGroup(gpuGroup) {
124+
return true
125+
}
126+
}
127+
return false
128+
}
129+
130+
// isNewGpuGroup determines if a GPU group ID represents a new group (UUID) vs an existing one (numeric).
131+
func isNewGpuGroup(gpuGroup string) bool {
132+
// New GPU groups are UUIDs (e.g., "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx")
133+
// Existing GPU groups are numeric strings ("0", "1", "2", etc.)
134+
_, err := strconv.Atoi(gpuGroup)
135+
return err != nil // If not a number, it's a UUID = new group
136+
}

0 commit comments

Comments
 (0)