✨ add run launcher and executor for KubeRay

docs/api/core.md

@@ -26,15 +26,15 @@ The `LocalRay` can be used to connect to a local Ray cluster.
 
 ::: dagster_ray.core.run_launcher.RayRunLauncher
     options:
-      members: true
+      members: false
 
 ---
 
 ### Executor
 
 ::: dagster_ray.core.executor.ray_executor
     options:
-      members: true
+      members: false
 
 ---
 

docs/api/kuberay.md

@@ -4,6 +4,18 @@ KubeRay integration components for running Ray on Kubernetes.  Learn how to use
 
 ---
 
+## Run Launcher
+
+::: dagster_ray.kuberay.run_launcher.KubeRayRunLauncher
+
+---
+
+# Executor
+
+::: dagster_ray.kuberay.executor.kuberay_executor
+
+---
+
 ## Client Mode Resources
 
 These resources initialize Ray client connection with a remote cluster.

justfile

@@ -14,7 +14,7 @@ docs-build:
     uv run --group docs mkdocs build --clean --strict
 
 docs-serve:
-    uv run --group docs mkdocs serve
+    uv run --group docs mkdocs serve --clean
 
 docs-publish:
     uv run --group docs --all-extras mike deploy --push --update-aliases $(uv run dunamai from any --style pep440)

src/dagster_ray/_base/utils.py

@@ -1,12 +1,17 @@
 from __future__ import annotations
 
 import dagster as dg
+from dagster._core.executor.step_delegating import (
+    StepHandlerContext,
+)
+from dagster._core.launcher.base import LaunchRunContext
 
 from dagster_ray._base.constants import DEFAULT_DEPLOYMENT_NAME
+from dagster_ray.types import AnyDagsterContext
 
 
 def get_dagster_tags(
-    context: dg.InitResourceContext | dg.OpExecutionContext | dg.AssetExecutionContext,
+    context: dg.InitResourceContext | AnyDagsterContext | StepHandlerContext | LaunchRunContext,
     extra_tags: dict[str, str] | None = None,
 ) -> dict[str, str]:
     """
@@ -29,9 +34,24 @@ def get_dagster_tags(
                 # inject the resource key used for this KubeRay resource
                 # this enables e.g reusing the same `RayCluster` across Dagster steps that require it without recreating the cluster
                 labels["dagster/resource-key"] = resource_key
-    else:
+
+    elif isinstance(context, StepHandlerContext):
+        labels.update(
+            **context.dagster_run.dagster_execution_info,
+        )
+    elif isinstance(context, LaunchRunContext):
+        labels.update(
+            **context.dagster_run.dagster_execution_info,
+        )
+    elif isinstance(context, dg.OpExecutionContext):
+        labels.update(
+            **context.run.dagster_execution_info,
+        )
+    elif isinstance(context, dg.AssetExecutionContext):
         labels.update(
             **context.run.dagster_execution_info,
         )
+    else:
+        raise ValueError(f"Unexpected context type: {type(context)}")
 
     return labels

src/dagster_ray/core/executor.py

@@ -1,5 +1,3 @@
-from __future__ import annotations
-
 from collections.abc import Iterator
 from typing import TYPE_CHECKING, Any, cast
 
@@ -23,14 +21,14 @@
 except ImportError:
     # for new versions of dagster > 1.11.6
     from dagster._core.remote_origin import RemoteJobOrigin  # pyright: ignore[reportMissingImports]
+
 from dagster._utils.merger import merge_dicts
 from packaging.version import Version
 from pydantic import Field
 
 from dagster_ray.configs import RayExecutionConfig, RayJobSubmissionClientConfig
 from dagster_ray.core.run_launcher import RayRunLauncher
-from dagster_ray.kuberay.utils import get_k8s_object_name
-from dagster_ray.utils import resolve_env_vars_list
+from dagster_ray.utils import get_k8s_object_name, resolve_env_vars_list
 
 if TYPE_CHECKING:
     from ray.job_submission import JobSubmissionClient
@@ -129,7 +127,7 @@ def name(self):
 
     def __init__(
         self,
-        client: JobSubmissionClient,
+        client: "JobSubmissionClient",
         env_vars: list[str] | None,
         runtime_env: dict[str, Any] | None,
         num_cpus: float | None,