Skip to content

Commit 1b80aed

Browse files
gshaibigithub-actions[bot]
authored andcommitted
fix(podgrouper): respect minReplicas in segmented elastic PyTorchJob (#1971)
Signed-off-by: gshaibi <gshaibi@nvidia.com> (cherry picked from commit 4ca07f4)
1 parent f9c97c0 commit 1b80aed

3 files changed

Lines changed: 53 additions & 1 deletion

File tree

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,3 @@
1+
kind: Fixed
2+
body: |-
3+
Segmented elastic PyTorchJob now respects minReplicas instead of requiring all worker segments

pkg/podgrouper/podgrouper/plugins/kubeflow/pytorch/pytorch_grouper.go

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -191,6 +191,9 @@ func buildWorkerSubGroups(
191191
Name: strings.ToLower(replicaTypeWorker),
192192
MinSubGroup: ptr.To(int32(numSegments)),
193193
}}
194+
// Segments whose first pod index is within workerMinAvailable are mandatory;
195+
// the rest are elastic and must not block scheduling.
196+
mandatorySegments := (int(workerMinAvailable) + segmentSize - 1) / segmentSize
194197
for i := range numSegments {
195198
subGroup := &podgroup.SubGroupMetadata{
196199
Name: fmt.Sprintf("worker-%d", i),
@@ -201,7 +204,9 @@ func buildWorkerSubGroups(
201204
if i == segmentIndex {
202205
subGroup.PodsReferences = podReferences
203206
}
204-
if partialSegmentSize != 0 && i == numSegments-1 {
207+
if i >= mandatorySegments {
208+
subGroup.MinAvailable = 0
209+
} else if partialSegmentSize != 0 && i == numSegments-1 {
205210
subGroup.MinAvailable = int32(partialSegmentSize)
206211
}
207212
subGroups = append(subGroups, subGroup)

pkg/podgrouper/podgrouper/plugins/kubeflow/pytorch/pytorch_grouper_test.go

Lines changed: 44 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -511,6 +511,50 @@ func TestGetPodGroupMetadata_Segments_5Workers_2PerSegment(t *testing.T) {
511511
assert.Equal(t, "test-job-worker-4", workerSegment2.PodsReferences[0])
512512
}
513513

514+
func TestGetPodGroupMetadata_Segments_ElasticMinReplicas(t *testing.T) {
515+
// 1 master + 8 workers, minReplicas=5, segment size 2 → 4 segments, 2 mandatory
516+
pytorchJob := getPytorchJobWithSegments(1, 8, "2")
517+
err := unstructured.SetNestedField(pytorchJob.Object, int64(5), "spec", "elasticPolicy", "minReplicas")
518+
assert.Nil(t, err)
519+
grouper := newTestPyTorchGrouper()
520+
521+
workerPod := &v1.Pod{
522+
ObjectMeta: metav1.ObjectMeta{
523+
Name: "test-job-worker-0",
524+
Namespace: "test_namespace",
525+
Labels: map[string]string{
526+
replicaTypeLabel: "worker",
527+
"training.kubeflow.org/replica-index": "0",
528+
},
529+
Annotations: map[string]string{
530+
"kai.scheduler/segment-size": "2",
531+
},
532+
},
533+
}
534+
535+
metadata, err := grouper.GetPodGroupMetadata(pytorchJob, workerPod)
536+
assert.Nil(t, err)
537+
assert.Equal(t, int32(5), metadata.MinAvailable)
538+
539+
workerParent := findSubGroupByName(metadata.SubGroups, strings.ToLower(replicaTypeWorker))
540+
assert.NotNil(t, workerParent)
541+
assert.Equal(t, ptr.To(int32(4)), workerParent.MinSubGroup)
542+
543+
for _, tc := range []struct {
544+
name string
545+
minAvailable int32
546+
}{
547+
{"worker-0", 2},
548+
{"worker-1", 2},
549+
{"worker-2", 0},
550+
{"worker-3", 0},
551+
} {
552+
sg := findSubGroupByName(metadata.SubGroups, tc.name)
553+
assert.NotNil(t, sg, "segment %s not found", tc.name)
554+
assert.Equal(t, tc.minAvailable, sg.MinAvailable, "segment %s MinAvailable", tc.name)
555+
}
556+
}
557+
514558
func TestGetPodGroupMetadata_Segments_MalformedAnnotation(t *testing.T) {
515559
pytorchJob := getPytorchJobWithSegments(1, 4, "invalid")
516560
grouper := newTestPyTorchGrouper()

0 commit comments

Comments
 (0)