wait until RayCluster is ready

README.md

@@ -142,7 +142,9 @@ def my_asset(context: AssetExecutionContext, pipes_rayjob_client: PipesRayJobCli
     pipes_rayjob_client.run(
         context=context,
         ray_job={
-            # RayJob manifest goes here, only .metadata.name is not required and will be generated if not provided
+            # RayJob manifest goes here
+            # .metadata.name is not required and will be generated if not provided
+            # *.container.image is not required and will be set to the current `dagster/image` tag if not provided
             # full reference: https://ray-project.github.io/kuberay/reference/api/#rayjob
             ...
         },

dagster_ray/kuberay/configs.py

@@ -13,8 +13,8 @@
     "env": [],
     "envFrom": [],
     "resources": {
-        "limits": {"cpu": "1000m", "memory": "1Gi"},
-        "requests": {"cpu": "1000m", "memory": "1Gi"},
+        "limits": {"cpu": "50m", "memory": "0.1Gi"},
+        "requests": {"cpu": "50m", "memory": "0.1Gi"},
     },
 }
 DEFAULT_HEAD_GROUP_SPEC = {

dagster_ray/kuberay/pipes.py

@@ -172,6 +172,15 @@ def _start(self, context: OpExecutionContext, ray_job: Dict[str, Any]) -> Dict[s
             namespace=ray_job["metadata"]["namespace"],
         )
 
+        self.client.ray_cluster_client.wait_until_ready(
+            name=self.client.get_ray_cluster_name(
+                name=name,
+                namespace=namespace,
+            ),
+            namespace=namespace,
+            timeout=self.timeout,
+        )
+
         self.client.wait_until_running(
             name=name,
             namespace=namespace,