[serve] configure http options in controller (#54331)

We set a bunch of defaults for http and grpc options in the proxy, but
we should set them in the controller.

---------

Signed-off-by: Cindy Zhang <[email protected]>

Author: zcin

python/ray/serve/_private/controller.py

@@ -36,6 +36,10 @@
 from ray.serve._private.deployment_info import DeploymentInfo
 from ray.serve._private.deployment_state import DeploymentStateManager
 from ray.serve._private.endpoint_state import EndpointState
+from ray.serve._private.grpc_util import set_proxy_default_grpc_options
+from ray.serve._private.http_util import (
+    configure_http_options_with_defaults,
+)
 from ray.serve._private.logging_utils import (
     configure_component_cpu_profiler,
     configure_component_logger,
@@ -156,13 +160,16 @@ async def __init__(
         self.cluster_node_info_cache = create_cluster_node_info_cache(self.gcs_client)
         self.cluster_node_info_cache.update()
 
+        # Configure proxy default HTTP and gRPC options.
         self.proxy_state_manager = ProxyStateManager(
-            http_options=http_options,
+            http_options=configure_http_options_with_defaults(http_options),
             head_node_id=self._controller_node_id,
             cluster_node_info_cache=self.cluster_node_info_cache,
             logging_config=self.global_logging_config,
-            grpc_options=grpc_options,
+            grpc_options=set_proxy_default_grpc_options(grpc_options),
         )
+        # We modify the HTTP and gRPC options above, so delete them to avoid
+        del http_options, grpc_options
 
         self.endpoint_state = EndpointState(self.kv_store, self.long_poll_host)
 

python/ray/serve/_private/grpc_util.py

@@ -1,13 +1,18 @@
 import asyncio
 import logging
+from copy import deepcopy
 from typing import Callable, List, Optional, Sequence, Tuple
 from unittest.mock import Mock
 
 import grpc
 from grpc.aio._server import Server
 
 from ray.exceptions import RayActorError, RayTaskError
-from ray.serve._private.constants import DEFAULT_GRPC_SERVER_OPTIONS, SERVE_LOGGER_NAME
+from ray.serve._private.constants import (
+    DEFAULT_GRPC_SERVER_OPTIONS,
+    RAY_SERVE_REQUEST_PROCESSING_TIMEOUT_S,
+    SERVE_LOGGER_NAME,
+)
 from ray.serve._private.proxy_request_response import ResponseStatus
 from ray.serve.config import gRPCOptions
 from ray.serve.exceptions import BackPressureError, DeploymentUnavailableError
@@ -155,3 +160,14 @@ def set_grpc_code_and_details(
         context.set_code(status.code)
     if not context.details():
         context.set_details(status.message)
+
+
+def set_proxy_default_grpc_options(grpc_options) -> gRPCOptions:
+    grpc_options = deepcopy(grpc_options) or gRPCOptions()
+
+    if grpc_options.request_timeout_s or RAY_SERVE_REQUEST_PROCESSING_TIMEOUT_S:
+        grpc_options.request_timeout_s = (
+            grpc_options.request_timeout_s or RAY_SERVE_REQUEST_PROCESSING_TIMEOUT_S
+        )
+
+    return grpc_options

python/ray/serve/_private/http_util.py

@@ -809,12 +809,19 @@ def configure_http_options_with_defaults(http_options: HTTPOptions) -> HTTPOptio
         http_options.keep_alive_timeout_s = RAY_SERVE_HTTP_KEEP_ALIVE_TIMEOUT_S
 
     # TODO: Deprecate SERVE_REQUEST_PROCESSING_TIMEOUT_S env var
-    http_options.request_timeout_s = (
-        http_options.request_timeout_s or RAY_SERVE_REQUEST_PROCESSING_TIMEOUT_S
-    )
+    if http_options.request_timeout_s or RAY_SERVE_REQUEST_PROCESSING_TIMEOUT_S:
+        http_options.request_timeout_s = (
+            http_options.request_timeout_s or RAY_SERVE_REQUEST_PROCESSING_TIMEOUT_S
+        )
 
     http_options.middlewares = http_options.middlewares or []
 
+    return http_options
+
+
+def configure_http_middlewares(http_options: HTTPOptions) -> HTTPOptions:
+    http_options = deepcopy(http_options)
+
     # Add environment variable middleware
     if RAY_SERVE_HTTP_PROXY_CALLBACK_IMPORT_PATH:
         logger.info(

python/ray/serve/_private/proxy.py

@@ -6,7 +6,6 @@
 import pickle
 import time
 from abc import ABC, abstractmethod
-from copy import deepcopy
 from typing import Any, Callable, Dict, Generator, Optional, Set, Tuple
 
 import grpc
@@ -30,7 +29,6 @@
     RAY_SERVE_ENABLE_PROXY_GC_OPTIMIZATIONS,
     RAY_SERVE_PROXY_GC_THRESHOLD,
     RAY_SERVE_REQUEST_PATH_LOG_BUFFER_SIZE,
-    RAY_SERVE_REQUEST_PROCESSING_TIMEOUT_S,
     REQUEST_LATENCY_BUCKETS_MS,
     SERVE_CONTROLLER_NAME,
     SERVE_HTTP_REQUEST_ID_HEADER,
@@ -46,7 +44,7 @@
 )
 from ray.serve._private.http_util import (
     MessageQueue,
-    configure_http_options_with_defaults,
+    configure_http_middlewares,
     convert_object_to_asgi_messages,
     get_http_response_status,
     receive_http_body,
@@ -1007,40 +1005,22 @@ async def send_request_to_replica(
         yield status
 
 
-def _set_proxy_default_grpc_options(grpc_options) -> gRPCOptions:
-    grpc_options = deepcopy(grpc_options) or gRPCOptions()
-
-    grpc_options.request_timeout_s = (
-        grpc_options.request_timeout_s or RAY_SERVE_REQUEST_PROCESSING_TIMEOUT_S
-    )
-
-    return grpc_options
-
-
 @ray.remote(num_cpus=0)
 class ProxyActor:
     def __init__(
         self,
         http_options: HTTPOptions,
+        grpc_options: gRPCOptions,
         *,
-        grpc_options: Optional[gRPCOptions] = None,
         node_id: NodeId,
         node_ip_address: str,
         logging_config: LoggingConfig,
         long_poll_client: Optional[LongPollClient] = None,
     ):  # noqa: F821
         self._node_id = node_id
         self._node_ip_address = node_ip_address
-
-        # Configure proxy default HTTP and gRPC options.
-        http_options = configure_http_options_with_defaults(http_options)
-        grpc_options = _set_proxy_default_grpc_options(grpc_options)
-        self._http_options = http_options
+        self._http_options = configure_http_middlewares(http_options)
         self._grpc_options = grpc_options
-
-        # We modify the HTTP and gRPC options above, so delete them to avoid
-        del http_options, grpc_options
-
         grpc_enabled = is_grpc_enabled(self._grpc_options)
 
         event_loop = get_or_create_event_loop()

python/ray/serve/tests/test_callback.py

@@ -14,7 +14,7 @@
 from ray.exceptions import RayActorError
 from ray.serve._private.test_utils import get_application_url
 from ray.serve._private.utils import call_function_from_import_path
-from ray.serve.config import HTTPOptions
+from ray.serve.config import HTTPOptions, gRPCOptions
 from ray.serve.context import _get_global_client
 from ray.serve.schema import LoggingConfig, ProxyStatus, ServeInstanceDetails
 
@@ -170,6 +170,7 @@ def test_callback_fail(ray_instance):
     actor_def = ray.serve._private.proxy.ProxyActor
     handle = actor_def.remote(
         http_options=HTTPOptions(host="http_proxy", root_path="/", port=123),
+        grpc_options=gRPCOptions(),
         node_ip_address="127.0.0.1",
         node_id="123",
         logging_config=LoggingConfig(),
@@ -182,6 +183,7 @@ def test_callback_fail(ray_instance):
     actor_def = ray.actor._make_actor(serve_controller, {})
     handle = actor_def.remote(
         http_options=HTTPOptions(),
+        grpc_options=gRPCOptions(),
         global_logging_config=LoggingConfig(),
     )
     with pytest.raises(RayActorError, match="cannot be imported"):
@@ -203,6 +205,7 @@ def test_http_proxy_return_aribitary_objects(ray_instance):
     actor_def = ray.serve._private.proxy.ProxyActor
     handle = actor_def.remote(
         http_options=HTTPOptions(host="http_proxy", root_path="/", port=123),
+        grpc_options=gRPCOptions(),
         node_ip_address="127.0.0.1",
         node_id="123",
         logging_config=LoggingConfig(),

python/ray/serve/tests/unit/test_http_util.py

@@ -13,6 +13,7 @@
 from ray.serve._private.http_util import (
     ASGIReceiveProxy,
     MessageQueue,
+    configure_http_middlewares,
     configure_http_options_with_defaults,
 )
 
@@ -359,7 +360,7 @@ async def dispatch(self, request, call_next):
         ]  # Return list of wrapped middleware
 
         # Act
-        result = configure_http_options_with_defaults(base_http_options)
+        result = configure_http_middlewares(base_http_options)
 
         # Assert
         mock_call_function.assert_called_once_with(