[serve] Check if proxy is ready to serve traffic

python/ray/serve/_private/client.py

@@ -20,6 +20,7 @@
 from ray.serve._private.constants import (
     CLIENT_CHECK_CREATION_POLLING_INTERVAL_S,
     CLIENT_POLLING_INTERVAL_S,
+    HTTP_PROXY_TIMEOUT,
     MAX_CACHED_HANDLES,
     SERVE_DEFAULT_APP_NAME,
     SERVE_LOGGER_NAME,
@@ -288,6 +289,30 @@ def _wait_for_application_running(self, name: str, timeout_s: int = -1):
                 f"Application {name} did not become RUNNING after {timeout_s}s."
             )
 
+    @_ensure_connected
+    def wait_for_proxies_serving(self) -> None:
+        """Wait for the proxies to be ready to serve requests."""
+        proxy_handles = ray.get(self._controller.get_proxies.remote())
+        proxy_states = {
+            handle: ray.util.state.get_actor(id=handle._actor_id.hex())
+            for handle in proxy_handles.values()
+        }
+        alive_proxy_handles = [
+            handle
+            for handle, state in proxy_states.items()
+            if state and state.state != "DEAD"
+        ]
+        if len(alive_proxy_handles) > 0:
+            try:
+                ray.get(
+                    [handle.serving.remote() for handle in alive_proxy_handles],
+                    timeout=HTTP_PROXY_TIMEOUT,
+                )
+            except ray.exceptions.GetTimeoutError:
+                raise TimeoutError(
+                    f"Proxies not available after {HTTP_PROXY_TIMEOUT}s."
+                )
+
     @_ensure_connected
     def deploy_applications(
         self,

python/ray/serve/_private/proxy.py

@@ -1055,6 +1055,11 @@ async def ready(self) -> str:
         """
         pass
 
+    @abstractmethod
+    async def serving(self) -> None:
+        """Wait for the proxy to be ready to serve requests."""
+        pass
+
     @abstractmethod
     async def update_draining(
         self, draining: bool, _after: Optional[Any] = None
@@ -1308,6 +1313,10 @@ async def ready(self) -> str:
             ]
         )
 
+    async def serving(self) -> None:
+        """Wait for the proxy to be ready to serve requests."""
+        return
+
     async def update_draining(self, draining: bool, _after: Optional[Any] = None):
         """Update the draining status of the HTTP and gRPC proxies.
 

python/ray/serve/api.py

@@ -608,12 +608,15 @@ def _run_many(
         # Record after Ray has been started.
         ServeUsageTag.API_VERSION.record("v2")
 
-        return client.deploy_applications(
+        handles = client.deploy_applications(
             built_apps,
             wait_for_ingress_deployment_creation=wait_for_ingress_deployment_creation,
             wait_for_applications_running=wait_for_applications_running,
         )
 
+        client.wait_for_proxies_serving()
+        return handles
+
 
 @PublicAPI(stability="stable")
 def _run(