Repository: ray-project/ray
Pull request: [serve] Add request timeout sec for grpc
<!-- Thank you for your contribution! Please review https://github.com/ray-project/ray/blob/master/CONTRIBUTING.rst before opening a pull request. -->

<!-- Please add a reviewer to the assignee section when you create a PR. If you don't have the access to it, we will shortly find a reviewer and assign them to your PR. -->

## Why are these changes needed?

Add `request_timeout_s` in `grpc_options`, We can explicitly specify the timeout.

The env var RAY_SERVE_REQUEST_PROCESSING_TIMEOUT_S will be deprecated in the future release.

<!-- Please give a short summary of the change and the problem this solves. -->

## Related issue number

Closes #51761 

## Checks

- [x] I've signed off every commit(by using the -s flag, i.e., `git commit -s`) in this PR.
- [x] I've run `scripts/format.sh` to lint the changes in this PR.
- [x] I've included any doc changes needed for https://docs.ray.io/en/master/.
    - [ ] I've added any new APIs to the API Reference. For example, if I added a
           method in Tune, I've added it in `doc/source/tune/api/` under the
           corresponding `.rst` file.
- [x] I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/
- Testing Strategy
   - [x] Unit tests
   - [ ] Release tests
   - [ ] This PR is not tested :(


Problem statement:
[Ray Serve] request timeout sec for grpc
### Description

We don't have request_timeout_s for gRPC like we do for HTTP. Currently, gRPC relies on an environment variable for request timeouts. We should deprecate the environment variable and introduce a new attribute in the gRPC options schema to configure request timeouts explicitly.

### Use case

_No response_

Hints from issue discussion:
Hey, @abrarsheikh ! I want to work on this, could you assign to me?
> Hey, [@abrarsheikh](https://github.com/abrarsheikh) ! I want to work on this, could you assign to me?

@Ziy1-Tan appreciate it. Feel free to open a PR when you find time.

Commit message: [Ray Serve] Add request timeout sec for grpc

Signed-off-by: Ziy1-Tan <ajb459684460@gmail.com>

Patch to review:
diff --git a/doc/source/serve/api/index.md b/doc/source/serve/api/index.md
index bbc13d79def6a..8996ddb427fbf 100644
--- a/doc/source/serve/api/index.md
+++ b/doc/source/serve/api/index.md
@@ -212,7 +212,8 @@ Content-Type: application/json
     },
     "grpc_options": {
         "port": 9000,
-        "grpc_servicer_functions": []
+        "grpc_servicer_functions": [],
+        "request_timeout_s": null
     },
     "proxies": {
         "cef533a072b0f03bf92a6b98cb4eb9153b7b7c7b7f15954feb2f38ec": {
diff --git a/doc/source/serve/production-guide/config.md b/doc/source/serve/production-guide/config.md
index a9ca9d8d60fa3..8e3a367876e53 100644
--- a/doc/source/serve/production-guide/config.md
+++ b/doc/source/serve/production-guide/config.md
@@ -28,6 +28,7 @@ http_options:
 grpc_options:
   port: ...
   grpc_servicer_functions: ...
+  request_timeout_s: ...
 
 logging_config:
   log_level: ...
@@ -66,6 +67,7 @@ The `grpc_options` are as follows. Note that the gRPC config is global to your R
 - **`port`**: The port that the gRPC proxies listen on. These are optional settings and can be omitted. By default, the port is
   set to `9000`.
 - **`grpc_servicer_functions`**: List of import paths for gRPC `add_servicer_to_server` functions to add to Serve's gRPC proxy. The servicer functions need to be importable from the context of where Serve is running. This defaults to an empty list, which means the gRPC server isn't started.
+- **`request_timeout_s`**: Allows you to set the end-to-end timeout for a request before terminating and retrying at another replica. By default, the Serve gRPC proxy retries up to `10` times when a response is not received due to failures (for example, network disconnect, request timeout, etc.) By default, there is no request timeout.
 
 The `logging_config` is global config, you can configure controller & proxy & replica logs. Note that you can also set application and deployment level logging config, which will take precedence over the global config. See logging config API [here](../../serve/api/doc/ray.serve.schema.LoggingConfig.rst) for more details.
 
@@ -191,6 +193,3 @@ deployments:
       user_config:
         threshold: 1.5
 ```
-
-
-
diff --git a/python/ray/serve/_private/proxy.py b/python/ray/serve/_private/proxy.py
index 82ef285ff9798..a4bfe3cd0c311 100644
--- a/python/ray/serve/_private/proxy.py
+++ b/python/ray/serve/_private/proxy.py
@@ -1198,13 +1198,27 @@ def __init__(
             proxy_router=self.proxy_router,
             request_timeout_s=self._http_options.request_timeout_s,
         )
+
+        if (
+            RAY_SERVE_REQUEST_PROCESSING_TIMEOUT_S is not None
+            and self._grpc_options.request_timeout_s is None
+        ):
+            self._grpc_options.request_timeout_s = (
+                RAY_SERVE_REQUEST_PROCESSING_TIMEOUT_S
+            )
+            logger.warning(
+                "The `SERVE_REQUEST_PROCESSING_TIMEOUT_S` environment variable will be deprecated "
+                "in the future release. Please set `request_timeout_s` in your Serve config's `grpc_options` "
+                "field instead."
+            )
+
         self.grpc_proxy = (
             gRPCProxy(
                 node_id=self._node_id,
                 node_ip_address=self._node_ip_address,
                 is_head=is_head,
                 proxy_router=self.proxy_router,
-                request_timeout_s=RAY_SERVE_REQUEST_PROCESSING_TIMEOUT_S,
+                request_timeout_s=self._grpc_options.request_timeout_s,
             )
             if grpc_enabled
             else None
diff --git a/python/ray/serve/config.py b/python/ray/serve/config.py
index b2b0fbdbefa9a..f295a242dd47b 100644
--- a/python/ray/serve/config.py
+++ b/python/ray/serve/config.py
@@ -304,10 +304,12 @@ class gRPCOptions(BaseModel):
             Serve's gRPC proxy. Default to empty list, which means no gRPC methods will
             be added and no gRPC server will be started. The servicer functions need to
             be importable from the context of where Serve is running.
+        request_timeout_s: End-to-end timeout for gRPC requests.
     """
 
     port: int = DEFAULT_GRPC_PORT
     grpc_servicer_functions: List[str] = []
+    request_timeout_s: Optional[float] = None
 
     @property
     def grpc_servicer_func_callable(self) -> List[Callable]:
diff --git a/python/ray/serve/schema.py b/python/ray/serve/schema.py
index 2814f1e849005..1fe12d7a796e5 100644
--- a/python/ray/serve/schema.py
+++ b/python/ray/serve/schema.py
@@ -659,6 +659,10 @@ class gRPCOptionsSchema(BaseModel):
             "need to be importable from the context of where Serve is running."
         ),
     )
+    request_timeout_s: float = Field(
+        default=None,
+        description="The timeout for gRPC requests. Defaults to no timeout.",
+    )
 
 
 @PublicAPI(stability="stable")
diff --git a/python/ray/serve/tests/test_grpc.py b/python/ray/serve/tests/test_grpc.py
index b9ce564615635..6da6b65189782 100644
--- a/python/ray/serve/tests/test_grpc.py
+++ b/python/ray/serve/tests/test_grpc.py
@@ -354,15 +354,6 @@ def _check():
     ping_grpc_healthz(worker_node_channel, test_draining=True)
 
 
-@pytest.mark.parametrize(
-    "ray_instance",
-    [
-        {
-            "RAY_SERVE_REQUEST_PROCESSING_TIMEOUT_S": "0.1",
-        },
-    ],
-    indirect=True,
-)
 @pytest.mark.parametrize("streaming", [False, True])
 def test_grpc_proxy_timeouts(ray_instance, ray_shutdown, streaming: bool):
     """Test gRPC request timed out.
@@ -375,11 +366,13 @@ def test_grpc_proxy_timeouts(ray_instance, ray_shutdown, streaming: bool):
         "ray.serve.generated.serve_pb2_grpc.add_UserDefinedServiceServicer_to_server",
         "ray.serve.generated.serve_pb2_grpc.add_FruitServiceServicer_to_server",
     ]
+    grpc_request_timeout_s = 0.1
 
     serve.start(
         grpc_options=gRPCOptions(
             port=grpc_port,
             grpc_servicer_functions=grpc_servicer_functions,
+            request_timeout_s=grpc_request_timeout_s,
         ),
     )
 
diff --git a/python/ray/serve/tests/test_metrics.py b/python/ray/serve/tests/test_metrics.py
index 11a9e017f50c6..0f2a3fbd268b3 100644
--- a/python/ray/serve/tests/test_metrics.py
+++ b/python/ray/serve/tests/test_metrics.py
@@ -45,8 +45,6 @@
 def serve_start_shutdown(request):
     param = request.param if hasattr(request, "param") else None
     request_timeout_s = param if param else None
-    if request_timeout_s:
-        os.environ["RAY_SERVE_REQUEST_PROCESSING_TIMEOUT_S"] = str(request_timeout_s)
     """Fixture provides a fresh Ray cluster to prevent metrics state sharing."""
     ray.init(
         _metrics_export_port=TEST_METRICS_EXPORT_PORT,
@@ -64,6 +62,7 @@ def serve_start_shutdown(request):
         grpc_options=gRPCOptions(
             port=grpc_port,
             grpc_servicer_functions=grpc_servicer_functions,
+            request_timeout_s=request_timeout_s,
         ),
         http_options=HTTPOptions(
             request_timeout_s=request_timeout_s,
@@ -72,8 +71,6 @@ def serve_start_shutdown(request):
     serve.shutdown()
     ray.shutdown()
     ray._private.utils.reset_ray_address()
-    if request_timeout_s:
-        os.environ.pop("RAY_SERVE_REQUEST_PROCESSING_TIMEOUT_S")
 
 
 def extract_tags(line: str) -> Dict[str, str]:
diff --git a/python/ray/serve/tests/unit/test_config.py b/python/ray/serve/tests/unit/test_config.py
index 4a0ccbc6fa8af..011053c1c34d6 100644
--- a/python/ray/serve/tests/unit/test_config.py
+++ b/python/ray/serve/tests/unit/test_config.py
@@ -590,20 +590,24 @@ def test_grpc_options():
     assert default_grpc_options.port == DEFAULT_GRPC_PORT
     assert default_grpc_options.grpc_servicer_functions == []
     assert default_grpc_options.grpc_servicer_func_callable == []
+    assert default_grpc_options.request_timeout_s is None
 
     port = 9001
     grpc_servicer_functions = [
         "ray.serve.generated.serve_pb2_grpc.add_UserDefinedServiceServicer_to_server",
     ]
+    request_timeout_s = 1
     grpc_options = gRPCOptions(
         port=port,
         grpc_servicer_functions=grpc_servicer_functions,
+        request_timeout_s=request_timeout_s,
     )
     assert grpc_options.port == port
     assert grpc_options.grpc_servicer_functions == grpc_servicer_functions
     assert grpc_options.grpc_servicer_func_callable == [
         add_UserDefinedServiceServicer_to_server
     ]
+    assert grpc_options.request_timeout_s == request_timeout_s
 
     # Import not found should raise ModuleNotFoundError.
     grpc_servicer_functions = ["fake.service.that.does.not.exist"]


Code review comments:
diff_hunk: @@ -1198,13 +1198,27 @@ def __init__(
             proxy_router=self.proxy_router,
             request_timeout_s=self._http_options.request_timeout_s,
         )
+
+        if (
+            RAY_SERVE_REQUEST_PROCESSING_TIMEOUT_S is not None
+            and self._grpc_options.request_timeout_s is None
+        ):
+            self._grpc_options.request_timeout_s = (
+                RAY_SERVE_REQUEST_PROCESSING_TIMEOUT_S
+            )
+            logger.warning(
+                "The `SERVE_REQUEST_PROCESSING_TIMEOUT_S` environment variable will be deprecated "
+                "in the future release. Please set `request_timeout_s` in your Serve config's `grpc_options` "
+                "field instead."
+            )
+
path: python/ray/serve/_private/proxy.py
text: @user1:
nit: move this into a new function called `_set_proxy_default_grpc_options` similar to how its done for http. This will help reduce some code clutter in this function.