Merge branch 'master' into elliot-barn/upgrading-data-ci-tests-py310

Author: aslonnie

.buildkite/build.rayci.yml

@@ -97,7 +97,6 @@ steps:
       - raycudabase
       - raycpubase
     matrix:
-      - "3.9"
       - "3.10"
       - "3.11"
       - "3.12"
@@ -123,7 +122,6 @@ steps:
       - raycpubaseextra
       - raycudabaseextra
     matrix:
-      - "3.9"
       - "3.10"
       - "3.11"
       - "3.12"

.buildkite/cicd.rayci.yml

@@ -7,11 +7,11 @@ steps:
           //ci/ray_ci/... //release/... //ci/pipeline/... ci
           --only-tags=release_unit,ci_unit
           --cache-test-results --parallelism-per-worker 2
-          --build-name oss-ci-base_test
+          --build-name oss-ci-base_test-py3.10
           --build-type skip
     instance_type: small
     depends_on:
-      - oss-ci-base_test
+      - oss-ci-base_test-multipy
       - forge
     tags: tools
   - label: ":coral: reef: raydepsets tests"
@@ -20,30 +20,30 @@ steps:
       - bazel run //ci/ray_ci:test_in_docker --
           //ci/raydepsets/... ci
           --cache-test-results
-          --build-name oss-ci-base_test
+          --build-name oss-ci-base_test-py3.10
           --build-type skip
     instance_type: small
     depends_on:
-      - oss-ci-base_test
+      - oss-ci-base_test-multipy
       - forge
     tags: tools
   - label: ":coral: reef: privileged container tests"
     commands:
       - bazel run //ci/ray_ci:test_in_docker --
           //ci/ray_ci:test_privileged ci
           --cache-test-results
-          --build-name oss-ci-base_test
+          --build-name oss-ci-base_test-py3.10
           --build-type cgroup
           --privileged
     instance_type: small
     depends_on:
-      - oss-ci-base_test
+      - oss-ci-base_test-multipy
       - forge
     tags: tools
   - label: ":coral: reef: iwyu tests"
     commands:
       - bazel test --config iwyu //bazel/tests/cpp:example_test
     instance_type: small
-    depends_on: oss-ci-base_build
-    job_env: oss-ci-base_build
+    depends_on: oss-ci-base_build-multipy
+    job_env: oss-ci-base_build-py3.10
     tags: tools

doc/source/cluster/kubernetes/k8s-ecosystem/yunikorn.md

@@ -34,7 +34,7 @@ helm install kuberay-operator kuberay/kuberay-operator --version 1.4.2 --set bat
 
 ## Step 4: Use Apache YuniKorn for gang scheduling
 
-This example uses gang scheduling with Apache YuniKorn and KubeRay.
+This example demonstrates gang scheduling of RayCluster custom resources with Apache YuniKorn and KubeRay. Starting with KubeRay 1.5.0, KubeRay also supports gang scheduling for RayJob custom resources.
 
 First, create a queue with a capacity of 4 CPUs and 6Gi of RAM by editing the ConfigMap:
 

python/ray/_private/authentication/authentication_constants.py

@@ -21,3 +21,5 @@
     "The authentication token you provided is invalid or incorrect.\n\n"
     + TOKEN_SETUP_INSTRUCTIONS
 )
+
+AUTHORIZATION_HEADER_NAME = "authorization"

python/ray/_private/authentication/http_token_authentication.py

@@ -0,0 +1,93 @@
+import logging
+from types import ModuleType
+from typing import Dict, Optional
+
+from ray._private.authentication import authentication_constants
+from ray.dashboard import authentication_utils as auth_utils
+
+logger = logging.getLogger(__name__)
+
+
+def get_token_auth_middleware(aiohttp_module: ModuleType):
+    """Internal helper to create token auth middleware with provided modules.
+
+    Args:
+        aiohttp_module: The aiohttp module to use
+    Returns:
+        An aiohttp middleware function
+    """
+
+    @aiohttp_module.web.middleware
+    async def token_auth_middleware(request, handler):
+        """Middleware to validate bearer tokens when token authentication is enabled.
+
+        In minimal Ray installations (without ray._raylet), this middleware is a no-op
+        and passes all requests through without authentication.
+        """
+        # No-op if token auth is not enabled or raylet is not available
+        if not auth_utils.is_token_auth_enabled():
+            return await handler(request)
+
+        auth_header = request.headers.get(
+            authentication_constants.AUTHORIZATION_HEADER_NAME, ""
+        )
+        if not auth_header:
+            return aiohttp_module.web.Response(
+                status=401, text="Unauthorized: Missing authentication token"
+            )
+
+        if not auth_utils.validate_request_token(auth_header):
+            return aiohttp_module.web.Response(
+                status=403, text="Forbidden: Invalid authentication token"
+            )
+
+        return await handler(request)
+
+    return token_auth_middleware
+
+
+def get_auth_headers_if_auth_enabled(user_headers: Dict[str, str]) -> Dict[str, str]:
+
+    if not auth_utils.is_token_auth_enabled():
+        return {}
+
+    from ray._raylet import AuthenticationTokenLoader
+
+    # Check if user provided their own Authorization header (case-insensitive)
+    has_user_auth = any(
+        key.lower() == authentication_constants.AUTHORIZATION_HEADER_NAME
+        for key in user_headers.keys()
+    )
+    if has_user_auth:
+        # User has provided their own auth header, don't override
+        return {}
+
+    token_loader = AuthenticationTokenLoader.instance()
+    auth_headers = token_loader.get_token_for_http_header()
+
+    if not auth_headers:
+        # Token auth enabled but no token found
+        logger.warning(
+            "Token authentication is enabled but no token was found. "
+            "Requests to authenticated clusters will fail."
+        )
+
+    return auth_headers
+
+
+def format_authentication_http_error(status: int, body: str) -> Optional[str]:
+    """Return a user-friendly authentication error message, if applicable."""
+
+    if status == 401:
+        return "Authentication required: {body}\n\n{details}".format(
+            body=body,
+            details=authentication_constants.HTTP_REQUEST_MISSING_TOKEN_ERROR_MESSAGE,
+        )
+
+    if status == 403:
+        return "Authentication failed: {body}\n\n{details}".format(
+            body=body,
+            details=authentication_constants.HTTP_REQUEST_INVALID_TOKEN_ERROR_MESSAGE,
+        )
+
+    return None

python/ray/_private/runtime_env/agent/main.py

@@ -8,6 +8,9 @@
     get_or_create_event_loop,
 )
 from ray._private import logging_utils
+from ray._private.authentication.http_token_authentication import (
+    get_token_auth_middleware,
+)
 from ray._private.process_watcher import create_check_raylet_task
 from ray._raylet import GcsClient
 from ray.core.generated import (
@@ -23,6 +26,7 @@ def import_libs():
 
 import_libs()
 
+import aiohttp  # noqa: E402
 import runtime_env_consts  # noqa: E402
 from aiohttp import web  # noqa: E402
 from runtime_env_agent import RuntimeEnvAgent  # noqa: E402
@@ -194,7 +198,7 @@ async def get_runtime_envs_info(request: web.Request) -> web.Response:
             body=reply.SerializeToString(), content_type="application/octet-stream"
         )
 
-    app = web.Application()
+    app = web.Application(middlewares=[get_token_auth_middleware(aiohttp)])
 
     app.router.add_post("/get_or_create_runtime_env", get_or_create_runtime_env)
     app.router.add_post(

python/ray/dashboard/authentication_utils.py

@@ -1,8 +1,14 @@
-from ray._raylet import (
-    AuthenticationMode,
-    get_authentication_mode,
-    validate_authentication_token,
-)
+try:
+    from ray._raylet import (
+        AuthenticationMode,
+        get_authentication_mode,
+        validate_authentication_token,
+    )
+
+    _RAYLET_AVAILABLE = True
+except ImportError:
+    # ray._raylet not available during doc builds
+    _RAYLET_AVAILABLE = False
 
 
 def is_token_auth_enabled() -> bool:
@@ -11,6 +17,8 @@ def is_token_auth_enabled() -> bool:
     Returns:
         bool: True if auth_mode is set to "token", False otherwise
     """
+    if not _RAYLET_AVAILABLE:
+        return False
     return get_authentication_mode() == AuthenticationMode.TOKEN
 
 
@@ -23,7 +31,7 @@ def validate_request_token(auth_header: str) -> bool:
     Returns:
         bool: True if token is valid, False otherwise
     """
-    if not auth_header:
+    if not _RAYLET_AVAILABLE or not auth_header:
         return False
 
     # validate_authentication_token expects full "Bearer <token>" format

python/ray/dashboard/http_server_head.py

@@ -20,7 +20,9 @@
 from ray._common.network_utils import build_address, parse_address
 from ray._common.usage.usage_lib import TagKey, record_extra_usage_tag
 from ray._common.utils import get_or_create_event_loop
-from ray.dashboard import authentication_utils as auth_utils
+from ray._private.authentication.http_token_authentication import (
+    get_token_auth_middleware,
+)
 from ray.dashboard.dashboard_metrics import DashboardPrometheusMetrics
 from ray.dashboard.head import DashboardHeadModule
 
@@ -164,30 +166,6 @@ def get_address(self):
         assert self.http_host and self.http_port
         return self.http_host, self.http_port
 
-    @aiohttp.web.middleware
-    async def auth_middleware(self, request, handler):
-        """Authenticate requests when token auth is enabled."""
-
-        # Skip if auth not enabled
-        if not auth_utils.is_token_auth_enabled():
-            return await handler(request)
-
-        # Extract and validate token
-        auth_header = request.headers.get("Authorization", "")
-
-        if not auth_header:
-            return aiohttp.web.Response(
-                status=401, text="Unauthorized: Missing authentication token"
-            )
-
-        # Validate token
-        if not auth_utils.validate_request_token(auth_header):
-            return aiohttp.web.Response(
-                status=403, text="Forbidden: Invalid authentication token"
-            )
-
-        return await handler(request)
-
     @aiohttp.web.middleware
     async def path_clean_middleware(self, request, handler):
         if request.path.startswith("/static") or request.path.startswith("/logs"):
@@ -273,11 +251,12 @@ async def run(
 
         # Http server should be initialized after all modules loaded.
         # working_dir uploads for job submission can be up to 100MiB.
+
         app = aiohttp.web.Application(
             client_max_size=ray_constants.DASHBOARD_CLIENT_MAX_SIZE,
             middlewares=[
                 self.metrics_middleware,
-                self.auth_middleware,
+                get_token_auth_middleware(aiohttp),
                 self.path_clean_middleware,
                 self.browsers_no_post_put_middleware,
                 self.cache_control_static_middleware,

python/ray/dashboard/modules/dashboard_sdk.py

@@ -12,7 +12,10 @@
 import yaml
 
 import ray
-from ray._private.authentication import authentication_constants
+from ray._private.authentication.http_token_authentication import (
+    format_authentication_http_error,
+    get_auth_headers_if_auth_enabled,
+)
 from ray._private.runtime_env.packaging import (
     create_package,
     get_uri_for_directory,
@@ -21,9 +24,7 @@
 from ray._private.runtime_env.py_modules import upload_py_modules_if_needed
 from ray._private.runtime_env.working_dir import upload_working_dir_if_needed
 from ray._private.utils import split_address
-from ray._raylet import AuthenticationTokenLoader
 from ray.autoscaler._private.cli_logger import cli_logger
-from ray.dashboard.authentication_utils import is_token_auth_enabled
 from ray.dashboard.modules.job.common import uri_to_http_components
 from ray.util.annotations import DeveloperAPI, PublicAPI
 
@@ -226,7 +227,7 @@ def __init__(
         # Headers used for all requests sent to job server, optional and only
         # needed for cases like authentication to remote cluster.
         self._headers = cluster_info.headers or {}
-        self._headers.update(**self._get_auth_headers())
+        self._headers.update(**get_auth_headers_if_auth_enabled(self._headers))
 
         # Set SSL verify parameter for the requests library and create an ssl_context
         # object when needed for the aiohttp library.
@@ -247,36 +248,6 @@ def __init__(
             else:
                 self._ssl_context = None
 
-    def _get_auth_headers(self) -> Dict[str, str]:
-        """Get authentication headers if token auth is enabled.
-
-        Returns:
-            dict: Authentication headers to merge with request headers.
-                  Empty dict if no auth needed or token unavailable.
-        """
-        if not is_token_auth_enabled():
-            return {}
-
-        # Check if user provided their own Authorization header (case-insensitive)
-        has_user_auth = any(
-            key.lower() == "authorization" for key in self._headers.keys()
-        )
-        if has_user_auth:
-            # User has provided their own auth header, don't override
-            return {}
-
-        token_loader = AuthenticationTokenLoader.instance()
-        auth_headers = token_loader.get_token_for_http_header()
-
-        if not auth_headers:
-            # Token auth enabled but no token found
-            logger.warning(
-                "Token authentication is enabled but no token was found. "
-                "Requests to authenticated clusters will fail."
-            )
-
-        return auth_headers
-
     def _check_connection_and_version(
         self, min_version: str = "1.9", version_error_message: str = None
     ):
@@ -348,18 +319,11 @@ def _do_request(
         )
 
         # Check for authentication errors and provide helpful messages
-        if response.status_code == 401:
-            # Unauthorized - missing or no token provided
-            raise RuntimeError(
-                f"Authentication required: {response.text}\n\n"
-                + authentication_constants.HTTP_REQUEST_MISSING_TOKEN_ERROR_MESSAGE
-            )
-        elif response.status_code == 403:
-            # Forbidden - invalid token
-            raise RuntimeError(
-                f"Authentication failed: {response.text}\n\n"
-                + authentication_constants.HTTP_REQUEST_INVALID_TOKEN_ERROR_MESSAGE
-            )
+        formatted_error = format_authentication_http_error(
+            response.status_code, response.text
+        )
+        if formatted_error:
+            raise RuntimeError(formatted_error)
 
         return response