[serve] Make serve.shutdown() shut down remote Serve applications (#…

…23476)

python/ray/serve/api.py

@@ -132,6 +132,10 @@ def internal_get_global_client(
         _health_check_controller (bool): If True, run a health check on the
             cached controller if it exists. If the check fails, try reconnecting
             to the controller.
+
+    Raises:
+        RayServeException: if there is no Serve controller actor in the
+            expected namespace.
     """
 
     try:
@@ -812,6 +816,10 @@ def _connect(_override_controller_namespace: Optional[str] = None) -> Client:
         _override_controller_namespace (Optional[str]): The namespace to use
             when looking for the controller. If None, Serve recalculates the
             controller's namespace using _get_controller_namespace().
+
+    Raises:
+        RayServeException: if there is no Serve controller actor in the
+            expected namespace.
     """
 
     # Initialize ray if needed.
@@ -858,10 +866,17 @@ def shutdown() -> None:
     Shuts down all processes and deletes all state associated with the
     instance.
     """
-    if _global_client is None:
+
+    try:
+        client = internal_get_global_client()
+    except RayServeException:
+        logger.info(
+            "Nothing to shut down. There's no Serve application "
+            "running on this Ray cluster."
+        )
         return
 
-    internal_get_global_client().shutdown()
+    client.shutdown()
     _set_global_client(None)
 
 

python/ray/serve/tests/test_standalone2.py

@@ -1,9 +1,12 @@
 import sys
+import os
+import subprocess
+from tempfile import NamedTemporaryFile
+import requests
 
 import pytest
 from ray.cluster_utils import AutoscalingCluster
 from ray.exceptions import RayActorError
-import requests
 
 import ray
 import ray.state
@@ -22,6 +25,13 @@ def shutdown_ray():
         ray.shutdown()
 
 
+@pytest.fixture
+def start_and_shutdown_ray_cli():
+    subprocess.check_output(["ray", "start", "--head"])
+    yield
+    subprocess.check_output(["ray", "stop", "--force"])
+
+
 def test_standalone_actor_outside_serve():
     # https://github.com/ray-project/ray/issues/20066
 
@@ -116,6 +126,55 @@ def controller_died(handle):
     ray.shutdown()
 
 
+def test_shutdown_remote(start_and_shutdown_ray_cli):
+    """Check that serve.shutdown() works on a remote Ray cluster."""
+
+    deploy_serve_script = (
+        "import ray\n"
+        "from ray import serve\n"
+        "\n"
+        'ray.init(address="auto", namespace="x")\n'
+        "serve.start(detached=True)\n"
+        "\n"
+        "@serve.deployment\n"
+        "def f(*args):\n"
+        '   return "got f"\n'
+        "\n"
+        "f.deploy()\n"
+    )
+
+    shutdown_serve_script = (
+        "import ray\n"
+        "from ray import serve\n"
+        "\n"
+        'ray.init(address="auto", namespace="x")\n'
+        "serve.shutdown()\n"
+    )
+
+    # Cannot use context manager due to tmp file's delete flag issue in Windows
+    # https://stackoverflow.com/a/15590253
+    deploy_file = NamedTemporaryFile(mode="w+", delete=False, suffix=".py")
+    shutdown_file = NamedTemporaryFile(mode="w+", delete=False, suffix=".py")
+
+    try:
+        deploy_file.write(deploy_serve_script)
+        deploy_file.close()
+
+        shutdown_file.write(shutdown_serve_script)
+        shutdown_file.close()
+
+        # Ensure Serve can be restarted and shutdown with for loop
+        for _ in range(2):
+            subprocess.check_output(["python", deploy_file.name])
+            assert requests.get("http://localhost:8000/f").text == "got f"
+            subprocess.check_output(["python", shutdown_file.name])
+            with pytest.raises(requests.exceptions.ConnectionError):
+                requests.get("http://localhost:8000/f")
+    finally:
+        os.unlink(deploy_file.name)
+        os.unlink(shutdown_file.name)
+
+
 def test_autoscaler_shutdown_node_http_everynode(
     shutdown_ray, call_ray_stop_only  # noqa: F811
 ):