[serve] Add ControllerOptions for configurable controller runtime_env (#63352)

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: kouroshHakha

ci/lint/pydoclint-baseline.txt

@@ -1477,8 +1477,6 @@ python/ray/serve/api.py
     DOC103: Function `start`: Docstring arguments are different from function arguments. (Or could be other formatting issues: https://jsh9.github.io/pydoclint/violation_codes.html#notes-on-doc103 ). Arguments in the function signature but not in the docstring: [**kwargs: ].
     DOC201: Function `get_replica_context` does not have a return section in docstring
     DOC201: Function `ingress` does not have a return section in docstring
-    DOC101: Function `run_many`: Docstring contains fewer arguments than in function signature.
-    DOC103: Function `run_many`: Docstring arguments are different from function arguments. (Or could be other formatting issues: https://jsh9.github.io/pydoclint/violation_codes.html#notes-on-doc103 ). Arguments in the function signature but not in the docstring: [_local_testing_mode: bool].
     DOC101: Function `run`: Docstring contains fewer arguments than in function signature.
     DOC103: Function `run`: Docstring arguments are different from function arguments. (Or could be other formatting issues: https://jsh9.github.io/pydoclint/violation_codes.html#notes-on-doc103 ). Arguments in the function signature but not in the docstring: [_local_testing_mode: bool].
     DOC101: Function `multiplexed`: Docstring contains fewer arguments than in function signature.

doc/source/serve/api/index.md

@@ -94,6 +94,7 @@ See the [model composition guide](serve-model-composition) for how to update cod
    :toctree: doc/
    :template: autosummary/autopydantic.rst
 
+   serve.config.ControllerOptions
    serve.config.gRPCOptions
    serve.config.HTTPOptions
    serve.config.AutoscalingConfig

python/ray/dashboard/modules/serve/serve_head.py

@@ -170,6 +170,7 @@ async def put_all_applications(self, req: Request) -> Response:
                 http_options=full_http_options,
                 grpc_options=grpc_options,
                 global_logging_config=config.logging_config,
+                controller_options=config.controller_options,
             )
 
         # Serve ignores HTTP options if it was already running when

python/ray/serve/__init__.py

@@ -26,7 +26,7 @@
         status,
     )
     from ray.serve.batching import batch
-    from ray.serve.config import HTTPOptions
+    from ray.serve.config import ControllerOptions, HTTPOptions
     from ray.serve.utils import get_trace_context
 
 except ModuleNotFoundError as e:
@@ -48,6 +48,7 @@
     "_run_many",
     "batch",
     "start",
+    "ControllerOptions",
     "HTTPOptions",
     "get_replica_context",
     "get_deployment_actor",

python/ray/serve/_private/api.py

@@ -15,7 +15,7 @@
     SERVE_NAMESPACE,
 )
 from ray.serve._private.default_impl import get_controller_impl
-from ray.serve.config import HTTPOptions, gRPCOptions
+from ray.serve.config import ControllerOptions, HTTPOptions, gRPCOptions
 from ray.serve.context import (
     _check_cached_client_alive,
     _get_global_client,
@@ -28,6 +28,22 @@
 logger = logging.getLogger(SERVE_LOGGER_NAME)
 
 
+def _coerce_controller_options(
+    controller_options: Union[None, dict, ControllerOptions],
+) -> ControllerOptions:
+    """Normalize an optional dict / model into a validated ControllerOptions."""
+    if controller_options is None:
+        return ControllerOptions()
+    if isinstance(controller_options, ControllerOptions):
+        return controller_options
+    if isinstance(controller_options, dict):
+        return ControllerOptions.model_validate(controller_options)
+    raise TypeError(
+        "controller_options must be a dict, ControllerOptions, or None; got "
+        f"{type(controller_options).__name__}."
+    )
+
+
 def _check_http_options(
     client: ServeControllerClient, http_options: Union[dict, HTTPOptions]
 ) -> None:
@@ -56,6 +72,7 @@ def _start_controller(
     http_options: Union[None, dict, HTTPOptions] = None,
     grpc_options: Union[None, dict, gRPCOptions] = None,
     global_logging_config: Union[None, dict, LoggingConfig] = None,
+    controller_options: Union[None, dict, ControllerOptions] = None,
     **kwargs,
 ) -> ActorHandle:
     """Start Ray Serve controller.
@@ -95,7 +112,8 @@ def _start_controller(
     elif isinstance(global_logging_config, dict):
         global_logging_config = LoggingConfig(**global_logging_config)
 
-    controller_impl = get_controller_impl()
+    controller_options = _coerce_controller_options(controller_options)
+    controller_impl = get_controller_impl(controller_options=controller_options)
     controller = controller_impl.remote(
         http_options=http_options,
         grpc_options=grpc_options,
@@ -120,6 +138,7 @@ async def serve_start_async(
     http_options: Union[None, dict, HTTPOptions] = None,
     grpc_options: Union[None, dict, gRPCOptions] = None,
     global_logging_config: Union[None, dict, LoggingConfig] = None,
+    controller_options: Union[None, dict, ControllerOptions] = None,
     **kwargs,
 ) -> ServeControllerClient:
     """Initialize a serve instance asynchronously.
@@ -134,6 +153,10 @@ async def serve_start_async(
 
     usage_lib.record_library_usage("serve")
 
+    # Validate eagerly in the caller so a bad ``controller_options`` raises
+    # locally rather than from the ``_start_controller`` Ray task.
+    controller_options = _coerce_controller_options(controller_options)
+
     client, _ = _check_cached_client_alive()
     if client is None:
         try:
@@ -143,7 +166,7 @@ async def serve_start_async(
     if client is not None:
         logger.info(
             f'Connecting to existing Serve app in namespace "{SERVE_NAMESPACE}".'
-            " New http options will not be applied."
+            " New http_options/controller_options will not be applied."
         )
         if http_options:
             _check_http_options(client, http_options)
@@ -152,7 +175,13 @@ async def serve_start_async(
     controller = (
         await ray.remote(_start_controller)
         .options(num_cpus=0)
-        .remote(http_options, grpc_options, global_logging_config, **kwargs)
+        .remote(
+            http_options,
+            grpc_options,
+            global_logging_config,
+            controller_options=controller_options,
+            **kwargs,
+        )
     )
 
     client = ServeControllerClient(
@@ -167,6 +196,7 @@ def serve_start(
     http_options: Union[None, dict, HTTPOptions] = None,
     grpc_options: Union[None, dict, gRPCOptions] = None,
     global_logging_config: Union[None, dict, LoggingConfig] = None,
+    controller_options: Union[None, dict, ControllerOptions] = None,
     **kwargs,
 ) -> ServeControllerClient:
     """Initialize a serve instance.
@@ -207,10 +237,17 @@ def serve_start(
             - grpc_servicer_functions(list): List of import paths for gRPC
                 `add_servicer_to_server` functions to add to Serve's gRPC proxy.
                 Default empty list, meaning not to start the gRPC server.
+        controller_options: Optional ``ControllerOptions`` (or dict) for the
+            Serve controller actor. Currently only ``runtime_env.env_vars``
+            is honored; see ``ray.serve.config.ControllerOptions``. Only
+            applied on first controller creation -- ignored if the controller
+            is already running in this Ray cluster (a log line is emitted).
     """
 
     usage_lib.record_library_usage("serve")
 
+    controller_options = _coerce_controller_options(controller_options)
+
     client, _ = _check_cached_client_alive()
     if client is None:
         try:
@@ -220,14 +257,18 @@ def serve_start(
     if client is not None:
         logger.info(
             f'Connecting to existing Serve app in namespace "{SERVE_NAMESPACE}".'
-            " New http options will not be applied."
+            " New http_options/controller_options will not be applied."
         )
         if http_options:
             _check_http_options(client, http_options)
         return client
 
     controller = _start_controller(
-        http_options, grpc_options, global_logging_config, **kwargs
+        http_options,
+        grpc_options,
+        global_logging_config,
+        controller_options=controller_options,
+        **kwargs,
     )
 
     client = ServeControllerClient(

python/ray/serve/_private/default_impl.py

@@ -42,6 +42,7 @@
     inside_ray_client_context,
     resolve_deployment_response,
 )
+from ray.serve.config import ControllerOptions
 from ray.util.placement_group import PlacementGroup
 
 # NOTE: Please read carefully before changing!
@@ -242,10 +243,17 @@ def get_proxy_handle(endpoint: DeploymentID, info: EndpointInfo):
     )
 
 
-def get_controller_impl():
+def get_controller_impl(controller_options: Optional[ControllerOptions] = None):
+    """Build the Ray actor class for the Serve controller.
+
+    ``controller_options`` is the validated ``ControllerOptions`` model from
+    ``serve.start`` / ``serve.run`` / the YAML schema. Today only its
+    ``runtime_env`` field is consumed; future fields (num_cpus, resources,
+    max_concurrency overrides) slot in here.
+    """
     from ray.serve._private.controller import ServeController
 
-    controller_impl = ray.remote(
+    actor_options = dict(
         name=SERVE_CONTROLLER_NAME,
         namespace=SERVE_NAMESPACE,
         num_cpus=0,
@@ -255,9 +263,13 @@ def get_controller_impl():
         resources={HEAD_NODE_RESOURCE_NAME: 0.001},
         max_concurrency=CONTROLLER_MAX_CONCURRENCY,
         enable_task_events=RAY_SERVE_ENABLE_TASK_EVENTS,
-    )(ServeController)
+    )
+    if controller_options is not None and controller_options.runtime_env:
+        # The validator on ControllerOptions guarantees this is a dict
+        # containing only the ``env_vars`` key with str->str entries.
+        actor_options["runtime_env"] = controller_options.runtime_env
 
-    return controller_impl
+    return ray.remote(**actor_options)(ServeController)
 
 
 def get_proxy_actor_class():

python/ray/serve/api.py

@@ -41,6 +41,7 @@
 )
 from ray.serve.config import (
     AutoscalingConfig,
+    ControllerOptions,
     DeploymentActorConfig,
     GangSchedulingConfig,
     HTTPOptions,
@@ -77,6 +78,7 @@ def start(
     http_options: Union[None, dict, HTTPOptions] = None,
     grpc_options: Union[None, dict, gRPCOptions] = None,
     logging_config: Union[None, dict, LoggingConfig] = None,
+    controller_options: Union[None, dict, ControllerOptions] = None,
     **kwargs,
 ):
     """Start Serve on the cluster.
@@ -102,12 +104,19 @@ def start(
           class See `gRPCOptions` for supported options.
         logging_config: logging config options for the serve component (
             controller & proxy).
+        controller_options: [EXPERIMENTAL] Options for the Serve controller actor.
+          Currently scoped to a strictly-validated ``runtime_env.env_vars``
+          (other ``runtime_env`` keys are rejected). See
+          ``ray.serve.config.ControllerOptions``. Only applied on first
+          controller creation -- ignored if a Serve controller is already
+          running in this Ray cluster.
     """
     http_options = prepare_imperative_http_options(proxy_location, http_options)
     _private_api.serve_start(
         http_options=http_options,
         grpc_options=grpc_options,
         global_logging_config=logging_config,
+        controller_options=controller_options,
         **kwargs,
     )
 
@@ -727,6 +736,7 @@ def _run_many(
     wait_for_ingress_deployment_creation: bool = True,
     wait_for_applications_running: bool = True,
     _local_testing_mode: bool = False,
+    controller_options: Union[None, dict, ControllerOptions] = None,
 ) -> List[DeploymentHandle]:
     """Run many applications and return the handles to their ingress deployments.
 
@@ -787,6 +797,7 @@ def _run_many(
         client = _private_api.serve_start(
             http_options={"location": "EveryNode"},
             global_logging_config=None,
+            controller_options=controller_options,
         )
 
         # Record after Ray has been started.
@@ -814,6 +825,7 @@ def _run(
     logging_config: Optional[Union[Dict, LoggingConfig]] = None,
     _local_testing_mode: bool = False,
     external_scaler_enabled: bool = False,
+    controller_options: Union[None, dict, ControllerOptions] = None,
 ) -> DeploymentHandle:
     """Run an application and return a handle to its ingress deployment.
 
@@ -832,6 +844,7 @@ def _run(
         ],
         wait_for_applications_running=_blocking,
         _local_testing_mode=_local_testing_mode,
+        controller_options=controller_options,
     )[0]
 
 
@@ -842,6 +855,7 @@ def run_many(
     wait_for_ingress_deployment_creation: bool = True,
     wait_for_applications_running: bool = True,
     _local_testing_mode: bool = False,
+    controller_options: Union[None, dict, ControllerOptions] = None,
 ) -> List[DeploymentHandle]:
     """Run many applications and return the handles to their ingress deployments.
 
@@ -858,6 +872,13 @@ def run_many(
             `wait_for_ingress_deployment_creation=True`,
             because the ingress deployments must be created
             before the applications can be running.
+        _local_testing_mode: Internal flag enabling in-process local testing
+            mode. Not part of the public API.
+        controller_options: [EXPERIMENTAL] Options for the Serve controller
+            actor (e.g. ``runtime_env.env_vars`` for HAProxy / controller-side
+            tunables). See ``ray.serve.config.ControllerOptions``. Only applied
+            on first controller creation -- ignored if a Serve controller is
+            already running in this Ray cluster.
 
     Returns:
         List[DeploymentHandle]: A list of handles that can be used
@@ -868,6 +889,7 @@ def run_many(
         wait_for_ingress_deployment_creation=wait_for_ingress_deployment_creation,
         wait_for_applications_running=wait_for_applications_running,
         _local_testing_mode=_local_testing_mode,
+        controller_options=controller_options,
     )
 
     if blocking:
@@ -885,6 +907,7 @@ def run(
     logging_config: Optional[Union[Dict, LoggingConfig]] = None,
     _local_testing_mode: bool = False,
     external_scaler_enabled: bool = False,
+    controller_options: Union[None, dict, ControllerOptions] = None,
 ) -> DeploymentHandle:
     """Run an application and return a handle to its ingress deployment.
 
@@ -910,6 +933,11 @@ def run(
             be applied to all deployments which doesn't have logging config.
         external_scaler_enabled: Whether external autoscaling is enabled for
             this application.
+        controller_options: [EXPERIMENTAL] Options for the Serve controller
+            actor (e.g. ``runtime_env.env_vars`` for HAProxy / controller-side
+            tunables). See ``ray.serve.config.ControllerOptions``. Only applied
+            on first controller creation -- ignored if a Serve controller is
+            already running in this Ray cluster.
 
     Returns:
         DeploymentHandle: A handle that can be used to call the application.
@@ -921,6 +949,7 @@ def run(
         logging_config=logging_config,
         _local_testing_mode=_local_testing_mode,
         external_scaler_enabled=external_scaler_enabled,
+        controller_options=controller_options,
     )
 
     if blocking: