fix: correct the test name in raycluster_controller_unit_test.go, change the logic which same as ValidateRayClusterSpec in raycluster_controller_unit_test.go and add additional unit test

Author: CheyuWu

ray-operator/apis/ray/v1/raycluster_webhook.go

@@ -94,21 +94,66 @@ func (r *RayCluster) validateWorkerGroups() *field.Error {
 }
 
 func (r *RayCluster) ValidateRayClusterSpec() *field.Error {
-	if r.Annotations[RayFTEnabledAnnotationKey] == "false" && r.Spec.GcsFaultToleranceOptions != nil {
+	if len(r.Spec.HeadGroupSpec.Template.Spec.Containers) == 0 {
 		return field.Invalid(
-			field.NewPath("spec").Child("gcsFaultToleranceOptions"),
-			r.Spec.GcsFaultToleranceOptions,
-			fmt.Sprintf("GcsFaultToleranceOptions should be nil when %s annotation is set to false", RayFTEnabledAnnotationKey),
+			field.NewPath("spec").Child("headGroupSpec").Child("template").Child("spec").Child("containers"),
+			r.Spec.HeadGroupSpec.Template.Spec.Containers,
+			"headGroupSpec should have at least one container",
 		)
 	}
-	if r.Annotations[RayFTEnabledAnnotationKey] != "true" && len(r.Spec.HeadGroupSpec.Template.Spec.Containers) > 0 {
+
+	for _, workerGroup := range r.Spec.WorkerGroupSpecs {
+		if len(workerGroup.Template.Spec.Containers) == 0 {
+			return field.Invalid(
+				field.NewPath("spec").Child("workerGroupSpecs"),
+				r.Spec.WorkerGroupSpecs,
+				"workerGroupSpec should have at least one container",
+			)
+		}
+	}
+
+	if r.Annotations[RayFTEnabledAnnotationKey] != "" && r.Spec.GcsFaultToleranceOptions != nil {
+		return field.Invalid(
+			field.NewPath("metadata").Child("annotations").Child(RayFTEnabledAnnotationKey),
+			r.Annotations[RayFTEnabledAnnotationKey],
+			fmt.Sprintf("%s annotation and GcsFaultToleranceOptions are both set. "+
+				"Please use only GcsFaultToleranceOptions to configure GCS fault tolerance", RayFTEnabledAnnotationKey),
+		)
+	}
+
+	if !IsGCSFaultToleranceEnabled(*r) {
 		if EnvVarExists(RAY_REDIS_ADDRESS, r.Spec.HeadGroupSpec.Template.Spec.Containers[RayContainerIndex].Env) {
 			return field.Invalid(
-				field.NewPath("spec").Child("headGroupSpec").Child("template").Child("spec").Child("containers").Index(0).Child("env"),
-				RAY_REDIS_ADDRESS,
-				fmt.Sprintf("%s should not be set when %s is disabled", RAY_REDIS_ADDRESS, RayFTEnabledAnnotationKey),
+				field.NewPath("spec").Child("headGroupSpec").Child("template").Child("spec").Child("containers").Index(RayContainerIndex).Child("env"),
+				r.Spec.HeadGroupSpec.Template.Spec.Containers[RayContainerIndex].Env,
+				fmt.Sprintf("%s is set which implicitly enables GCS fault tolerance, "+
+					"but GcsFaultToleranceOptions is not set. Please set GcsFaultToleranceOptions "+
+					"to enable GCS fault tolerance", RAY_REDIS_ADDRESS),
 			)
 		}
 	}
+
+	if r.Spec.GcsFaultToleranceOptions != nil {
+		if redisPassword := r.Spec.HeadGroupSpec.RayStartParams["redis-password"]; redisPassword != "" {
+			return field.Invalid(
+				field.NewPath("spec").Child("headGroupSpec").Child("rayStartParams"),
+				r.Spec.HeadGroupSpec.RayStartParams,
+				"cannot set `redis-password` in rayStartParams when GcsFaultToleranceOptions is enabled - use GcsFaultToleranceOptions.RedisPassword instead",
+			)
+		}
+
+		headContainer := r.Spec.HeadGroupSpec.Template.Spec.Containers[RayContainerIndex]
+		if EnvVarExists(REDIS_PASSWORD, headContainer.Env) {
+			return field.Invalid(
+				field.NewPath("spec").Child("headGroupSpec").Child("template").Child("spec").Child("containers").Index(RayContainerIndex).Child("env"),
+				headContainer.Env,
+				"cannot set `REDIS_PASSWORD` env var in head Pod when GcsFaultToleranceOptions is enabled - use GcsFaultToleranceOptions.RedisPassword instead",
+			)
+		}
+	}
+
+	// TODO (kevin85421): If GcsFaultToleranceOptions is set, users should use `GcsFaultToleranceOptions.RedisAddress` instead of `RAY_REDIS_ADDRESS`.
+	// TODO (kevin85421): If GcsFaultToleranceOptions is set, users should use `GcsFaultToleranceOptions.ExternalStorageNamespace` instead of
+	// the annotation `ray.io/external-storage-namespace`.
 	return nil
 }