[Feature] [scheduler-plugins] Support second scheduler mode

helm-chart/kuberay-operator/values.yaml

@@ -75,7 +75,7 @@ batchScheduler:
   # Deprecated. This option will be removed in the future.
   # Note, for backwards compatibility. When it sets to true, it enables volcano scheduler integration.
   enabled: false
-  # Set the customized scheduler name, supported values are "volcano" or "yunikorn", do not set
+  # Set the customized scheduler name, supported values are "volcano", "yunikorn" or "scheduler-plugins", do not set
   # "batchScheduler.enabled=true" at the same time as it will override this option.
   name: ""
 

ray-operator/apis/config/v1alpha1/config_utils_test.go

@@ -6,6 +6,7 @@ import (
 	"github.com/go-logr/logr"
 	"github.com/go-logr/logr/testr"
 
+	schedulerPlugins "github.com/ray-project/kuberay/ray-operator/controllers/ray/batchscheduler/scheduler-plugins"
 	"github.com/ray-project/kuberay/ray-operator/controllers/ray/batchscheduler/volcano"
 	"github.com/ray-project/kuberay/ray-operator/controllers/ray/batchscheduler/yunikorn"
 )
@@ -60,6 +61,16 @@ func TestValidateBatchSchedulerConfig(t *testing.T) {
 			},
 			wantErr: false,
 		},
+		{
+			name: "valid option, batch-scheduler=scheduler-plugins",
+			args: args{
+				logger: testr.New(t),
+				config: Configuration{
+					BatchScheduler: schedulerPlugins.GetPluginName(),
+				},
+			},
+			wantErr: false,
+		},
 		{
 			name: "invalid option, invalid scheduler name",
 			args: args{

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

@@ -21,7 +21,13 @@ import (
 )
 
 const (
-	schedulerName                 string = "scheduler-plugins"
+	// This is the batchScheduler name used in the Ray Operator.
+	// We use this name because it is easier to understand and remember.
+	// It is also consistent with the name used in the Helm chart values.yaml.
+	schedulerName string = "scheduler-plugins"
+	// The default scheduler plugins name is "scheduler-plugins-scheduler".
+	// https://github.com/kubernetes-sigs/scheduler-plugins/blob/b3127ba4cc420430ca5322740103220043697eec/manifests/install/charts/as-a-second-scheduler/values.yaml#L6C9-L6C36
+	schedulerInstanceName         string = "scheduler-plugins-scheduler"
 	kubeSchedulerPodGroupLabelKey string = "scheduling.x-k8s.io/pod-group"
 )
 
@@ -36,7 +42,7 @@ func GetPluginName() string {
 }
 
 func (k *KubeScheduler) Name() string {
-	return GetPluginName()
+	return schedulerInstanceName
 }
 
 func createPodGroup(ctx context.Context, app *rayv1.RayCluster) *v1alpha1.PodGroup {
@@ -90,8 +96,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

@@ -116,3 +116,54 @@ func TestCreatePodGroupWithMultipleHosts(t *testing.T) {
 	// 1 head and 4 workers
 	a.Equal(int32(5), podGroup.Spec.MinMember)
 }
+
+func TestAddMetadataToPod(t *testing.T) {
+	tests := []struct {
+		name         string
+		enableGang   bool
+		podHasLabels bool
+	}{
+		{"GangEnabled_WithLabels", true, true},
+		{"GangDisabled_WithLabels", false, true},
+		{"GangDisabled_WithoutLabels", false, false},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			a := assert.New(t)
+			cluster := createTestRayCluster(1)
+			cluster.Labels = make(map[string]string)
+
+			if tt.enableGang {
+				cluster.Labels["ray.io/gang-scheduling-enabled"] = "true"
+			}
+
+			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.enableGang {
+				a.Equal(cluster.Name, pod.Labels[kubeSchedulerPodGroupLabelKey])
+			} else {
+				_, exists := pod.Labels[kubeSchedulerPodGroupLabelKey]
+				a.False(exists)
+			}
+
+			a.Equal(scheduler.Name(), pod.Spec.SchedulerName)
+			// The default scheduler plugins name is "scheduler-plugins-scheduler"
+			// The batchScheduler name is "scheduler-plugins"
+			// This is to ensure batchScheduler and default scheduler plugins name are not the same.
+			a.NotEqual(scheduler.Name(), GetPluginName())
+		})
+	}
+}