feat: enable Ray cluster head pod persistency

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

Author: kramaranya

src/codeflare_sdk/ray/cluster/build_ray_cluster.py

@@ -170,6 +170,31 @@ def build_ray_cluster(cluster: "codeflare_sdk.ray.cluster.Cluster"):
         },
     }
 
+    if cluster.config.enable_gcs_ft:
+        if not cluster.config.redis_address:
+            raise ValueError(
+                "redis_address must be provided when enable_gcs_ft is True"
+            )
+
+        gcs_ft_options = {"redisAddress": cluster.config.redis_address}
+
+        if cluster.config.external_storage_namespace:
+            gcs_ft_options[
+                "externalStorageNamespace"
+            ] = cluster.config.external_storage_namespace
+
+        if cluster.config.redis_password_secret:
+            gcs_ft_options["redisPassword"] = {
+                "valueFrom": {
+                    "secretKeyRef": {
+                        "name": cluster.config.redis_password_secret["name"],
+                        "key": cluster.config.redis_password_secret["key"],
+                    }
+                }
+            }
+
+        resource["spec"]["gcsFaultToleranceOptions"] = gcs_ft_options
+
     config_check()
     k8s_client = get_api_client() or client.ApiClient()
 

src/codeflare_sdk/ray/cluster/config.py

@@ -142,13 +142,38 @@ class ClusterConfiguration:
     annotations: Dict[str, str] = field(default_factory=dict)
     volumes: list[V1Volume] = field(default_factory=list)
     volume_mounts: list[V1VolumeMount] = field(default_factory=list)
+    enable_gcs_ft: bool = False
+    redis_address: Optional[str] = None
+    redis_password_secret: Optional[Dict[str, str]] = None
+    external_storage_namespace: Optional[str] = None
 
     def __post_init__(self):
         if not self.verify_tls:
             print(
                 "Warning: TLS verification has been disabled - Endpoint checks will be bypassed"
             )
 
+        if self.enable_gcs_ft:
+            if not self.redis_address:
+                raise ValueError(
+                    "redis_address must be provided when enable_gcs_ft is True"
+                )
+
+            if self.redis_password_secret and not isinstance(
+                self.redis_password_secret, dict
+            ):
+                raise ValueError(
+                    "redis_password_secret must be a dictionary with 'name' and 'key' fields"
+                )
+
+            if self.redis_password_secret and (
+                "name" not in self.redis_password_secret
+                or "key" not in self.redis_password_secret
+            ):
+                raise ValueError(
+                    "redis_password_secret must contain both 'name' and 'key' fields"
+                )
+
         self._validate_types()
         self._memory_to_resource()
         self._memory_to_string()
@@ -283,10 +308,13 @@ def check_type(value, expected_type):
                 else:
                     return True
             if origin_type is dict:
-                return all(
-                    check_type(k, args[0]) and check_type(v, args[1])
-                    for k, v in value.items()
-                )
+                if value is not None:
+                    return all(
+                        check_type(k, args[0]) and check_type(v, args[1])
+                        for k, v in value.items()
+                    )
+                else:
+                    return True
             if origin_type is tuple:
                 return all(check_type(elem, etype) for elem, etype in zip(value, args))
             if expected_type is int: