[RayCluster] yunikorn batchscheduler respect gang scheduling (#4075)

Signed-off-by: fscnick <[email protected]>

Author: fscnick

ray-operator/controllers/ray/batchscheduler/yunikorn/yunikorn_scheduler_test.go

@@ -142,11 +142,12 @@ func TestPopulateGangSchedulingAnnotations(t *testing.T) {
 	//   memory: 10Gi
 	//   nvidia.com/gpu: 1
 	addWorkerPodSpec(rayClusterWithGangScheduling,
-		"worker-group-1", 1, 1, 2, v1.ResourceList{
+		"worker-group-1", 2, 2, 2, v1.ResourceList{
 			v1.ResourceCPU:    resource.MustParse("2"),
 			v1.ResourceMemory: resource.MustParse("10Gi"),
 			"nvidia.com/gpu":  resource.MustParse("1"),
 		})
+	rayClusterWithGangScheduling.Spec.WorkerGroupSpecs[0].NumOfHosts = 3
 
 	// gang-scheduling enabled case, the plugin should populate the taskGroup annotation to the app
 	rayPod := createPod("ray-pod", "default")
@@ -173,7 +174,7 @@ func TestPopulateGangSchedulingAnnotations(t *testing.T) {
 	// verify the correctness of worker group
 	workerGroup := taskGroups.getTaskGroup("worker-group-1")
 	assert.NotNil(t, workerGroup)
-	assert.Equal(t, int32(1), workerGroup.MinMember)
+	assert.Equal(t, int32(6), workerGroup.MinMember)
 	assert.Equal(t, resource.MustParse("2"), workerGroup.MinResource[v1.ResourceCPU.String()])
 	assert.Equal(t, resource.MustParse("10Gi"), workerGroup.MinResource[v1.ResourceMemory.String()])
 	assert.Equal(t, resource.MustParse("1"), workerGroup.MinResource["nvidia.com/gpu"])

ray-operator/controllers/ray/batchscheduler/yunikorn/yunikorn_task_groups.go

@@ -52,11 +52,11 @@ func newTaskGroupsFromApp(app *v1.RayCluster) *TaskGroups {
 	// worker groups
 	for _, workerGroupSpec := range app.Spec.WorkerGroupSpecs {
 		workerMinResource := utils.CalculatePodResource(workerGroupSpec.Template.Spec)
-		minWorkers := workerGroupSpec.MinReplicas
+		minWorkers := (*workerGroupSpec.MinReplicas) * workerGroupSpec.NumOfHosts
 		taskGroups.addTaskGroup(
 			TaskGroup{
 				Name:         workerGroupSpec.GroupName,
-				MinMember:    *minWorkers,
+				MinMember:    minWorkers,
 				MinResource:  utils.ConvertResourceListToMapString(workerMinResource),
 				NodeSelector: workerGroupSpec.Template.Spec.NodeSelector,
 				Tolerations:  workerGroupSpec.Template.Spec.Tolerations,