[serve] backend and endpoint validation (#9954)

python/ray/serve/api.py

@@ -161,6 +161,17 @@ def create_endpoint(endpoint_name,
             "methods must be a list of strings, but got type {}".format(
                 type(methods)))
 
+    endpoints = list_endpoints()
+    if endpoint_name in endpoints:
+        methods_old = endpoints[endpoint_name]["methods"]
+        route_old = endpoints[endpoint_name]["route"]
+        if methods_old.sort() == methods.sort() and route_old == route:
+            raise ValueError(
+                "Route '{}' is already registered to endpoint '{}' "
+                "with methods '{}'.  To set the backend for this "
+                "endpoint, please use serve.set_traffic().".format(
+                    route, endpoint_name, methods))
+
     upper_methods = []
     for method in methods:
         if not isinstance(method, str):
@@ -260,6 +271,11 @@ def create_backend(backend_tag,
             be sent to a replica of this backend without receiving a
             response.
     """
+    if backend_tag in list_backends():
+        raise ValueError(
+            "Cannot create backend. "
+            "Backend '{}' is already registered.".format(backend_tag))
+
     if config is None:
         config = {}
     if not isinstance(config, dict):

python/ray/serve/controller.py

@@ -21,7 +21,7 @@ import numpy as np
 
 # Used for testing purposes only. If this is set, the controller will crash
 # after writing each checkpoint with the specified probability.
-_CRASH_AFTER_CHECKPOINT_PROBABILITY = 0.0
+_CRASH_AFTER_CHECKPOINT_PROBABILITY = 0
 CHECKPOINT_KEY = "serve-controller-checkpoint"
 
 # Feature flag for controller resource checking. If true, controller will
@@ -681,8 +681,11 @@ class ServeController:
             # TODO(edoakes): move this to client side.
             err_prefix = "Cannot create endpoint."
             if route in self.routes:
+
+                # Ensures this method is idempotent
                 if self.routes[route] == (endpoint, methods):
                     return
+
                 else:
                     raise ValueError(
                         "{} Route '{}' is already registered.".format(
@@ -694,8 +697,8 @@ class ServeController:
                         err_prefix, endpoint))
 
             logger.info(
-                "Registering route {} to endpoint {} with methods {}.".format(
-                    route, endpoint, methods))
+                "Registering route '{}' to endpoint '{}' with methods '{}'.".
+                format(route, endpoint, methods))
 
             self.routes[route] = (endpoint, methods)
 
@@ -747,6 +750,13 @@ class ServeController:
                              replica_config):
         """Register a new backend under the specified tag."""
         async with self.write_lock:
+            # Ensures this method is idempotent.
+            if backend_tag in self.backends:
+                backend_info = self.backends[backend_tag]
+                if (backend_info.backend_config == backend_config
+                        and backend_info.replica_config == replica_config):
+                    return
+
             backend_worker = create_backend_worker(
                 replica_config.func_or_class)
 
@@ -759,7 +769,11 @@ class ServeController:
                     backend_tag] = BasicAutoscalingPolicy(
                         backend_tag, backend_config.autoscaling_config)
 
-            self._scale_replicas(backend_tag, backend_config.num_replicas)
+            try:
+                self._scale_replicas(backend_tag, backend_config.num_replicas)
+            except RayServeException as e:
+                del self.backends[backend_tag]
+                raise e
 
             # NOTE(edoakes): we must write a checkpoint before starting new
             # or pushing the updated config to avoid inconsistent state if we

python/ray/serve/tests/test_api.py

@@ -76,6 +76,18 @@ def test_no_route(serve_instance):
     assert result == 1
 
 
+def test_reject_duplicate_backend(serve_instance):
+    def f():
+        pass
+
+    def g():
+        pass
+
+    serve.create_backend("backend", f)
+    with pytest.raises(ValueError):
+        serve.create_backend("backend", g)
+
+
 def test_reject_duplicate_route(serve_instance):
     def f():
         pass
@@ -101,6 +113,22 @@ def test_reject_duplicate_endpoint(serve_instance):
             endpoint_name, backend="backend", route="/different")
 
 
+def test_reject_duplicate_endpoint_and_route(serve_instance):
+    class SimpleBackend(object):
+        def __init__(self, message):
+            self.message = message
+
+        def __call__(self, *args, **kwargs):
+            return {"message": self.message}
+
+    serve.create_backend("backend1", SimpleBackend, "First")
+    serve.create_backend("backend2", SimpleBackend, "Second")
+
+    serve.create_endpoint("test", backend="backend1", route="/test")
+    with pytest.raises(ValueError):
+        serve.create_endpoint("test", backend="backend2", route="/test")
+
+
 def test_set_traffic_missing_data(serve_instance):
     endpoint_name = "foobar"
     backend_name = "foo_backend"

python/ray/serve/tests/test_failure.py

@@ -5,6 +5,7 @@ import time
 
 import ray
 from ray import serve
+from ray.serve.config import BackendConfig, ReplicaConfig
 
 
 def request_with_retries(endpoint, timeout=30):
@@ -209,6 +210,50 @@ def test_worker_replica_failure(serve_instance):
                 time.sleep(0.1)
 
 
+def test_create_backend_idempotent(serve_instance):
+    serve.init()
+
+    def f():
+        return "hello"
+
+    controller = serve.api._get_controller()
+
+    replica_config = ReplicaConfig(f)
+    backend_config = BackendConfig({"num_replicas": 1})
+
+    for i in range(10):
+        ray.get(
+            controller.create_backend.remote("my_backend", backend_config,
+                                             replica_config))
+
+    assert len(ray.get(controller.get_all_backends.remote())) == 1
+    serve.create_endpoint(
+        "my_endpoint", backend="my_backend", route="/my_route")
+
+    assert requests.get("http://127.0.0.1:8000/my_route").text == "hello"
+
+
+def test_create_endpoint_idempotent(serve_instance):
+    serve.init()
+
+    def f():
+        return "hello"
+
+    serve.create_backend("my_backend", f)
+
+    controller = serve.api._get_controller()
+
+    for i in range(10):
+        ray.get(
+            controller.create_endpoint.remote(
+                "my_endpoint", {"my_backend": 1.0}, "/my_route", ["GET"]))
+
+    assert len(ray.get(controller.get_all_endpoints.remote())) == 1
+    assert requests.get("http://127.0.0.1:8000/my_route").text == "hello"
+    resp = requests.get("http://127.0.0.1:8000/-/routes", timeout=0.5).json()
+    assert resp == {"/my_route": ["my_endpoint", ["GET"]]}
+
+
 if __name__ == "__main__":
     import sys
     import pytest