feat: add schedulerName to pod for second scheduler

Signed-off-by: Cheyu Wu <[email protected]>

Author: CheyuWu

ray-operator/controllers/ray/batchscheduler/scheduler-plugins/scheduler_plugins.go

@@ -99,8 +99,7 @@ func (k *KubeScheduler) AddMetadataToPod(_ context.Context, app *rayv1.RayCluste
 	if k.isGangSchedulingEnabled(app) {
 		pod.Labels[kubeSchedulerPodGroupLabelKey] = app.Name
 	}
-	// TODO(kevin85421): Currently, we only support "single scheduler" mode. If we want to support
-	// "second scheduler" mode, we need to add `schedulerName` to the pod spec.
+	pod.Spec.SchedulerName = k.Name()
 }
 
 func (k *KubeScheduler) isGangSchedulingEnabled(app *rayv1.RayCluster) bool {

ray-operator/controllers/ray/batchscheduler/scheduler-plugins/scheduler_plugins_test.go

@@ -96,3 +96,54 @@ func TestCreatePodGroup(t *testing.T) {
 	// 1 head and 2 workers
 	a.Equal(int32(3), podGroup.Spec.MinMember)
 }
+
+func TestAddMetadataToPod(t *testing.T) {
+	tests := []struct {
+		name             string
+		enableGang       bool
+		podHasLabels     bool
+		expectedPodGroup bool
+	}{
+		{"GangEnabled_WithLabels", true, true, true},
+		{"GangDisabled_WithLabels", false, true, false},
+		{"GangDisabled_WithoutLabels", false, false, false},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			a := assert.New(t)
+			cluster := createTestRayCluster(1)
+			if cluster.Labels == nil {
+				cluster.Labels = make(map[string]string)
+			}
+			if tt.enableGang {
+				cluster.Labels["ray.io/gang-scheduling-enabled"] = "true"
+			} else {
+				delete(cluster.Labels, "ray.io/gang-scheduling-enabled")
+			}
+
+			var pod *corev1.Pod
+			if tt.podHasLabels {
+				pod = &corev1.Pod{
+					ObjectMeta: metav1.ObjectMeta{
+						Labels: map[string]string{},
+					},
+				}
+			} else {
+				pod = &corev1.Pod{}
+			}
+
+			scheduler := &KubeScheduler{}
+			scheduler.AddMetadataToPod(context.TODO(), &cluster, "worker", pod)
+
+			if tt.expectedPodGroup {
+				a.Equal(cluster.Name, pod.Labels[kubeSchedulerPodGroupLabelKey])
+			} else {
+				_, exists := pod.Labels[kubeSchedulerPodGroupLabelKey]
+				a.False(exists)
+			}
+
+			a.Equal(scheduler.Name(), pod.Spec.SchedulerName)
+		})
+	}
+}