From fe5fcb6b9c5fc06424b977624c14281366506dc8 Mon Sep 17 00:00:00 2001 From: architkulkarni Date: Thu, 13 Aug 2020 09:56:50 -0700 Subject: [PATCH] [serve] backend and endpoint validation (#9954) --- python/ray/serve/api.py | 16 +++++++++ python/ray/serve/controller.py | 22 ++++++++++--- python/ray/serve/tests/test_api.py | 28 ++++++++++++++++ python/ray/serve/tests/test_failure.py | 45 ++++++++++++++++++++++++++ 4 files changed, 107 insertions(+), 4 deletions(-) diff --git a/python/ray/serve/api.py b/python/ray/serve/api.py index 8a70d1954..2420c9abe 100644 --- a/python/ray/serve/api.py +++ b/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): diff --git a/python/ray/serve/controller.py b/python/ray/serve/controller.py index 53d2144cd..2e2c7adaf 100644 --- a/python/ray/serve/controller.py +++ b/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 diff --git a/python/ray/serve/tests/test_api.py b/python/ray/serve/tests/test_api.py index 44618f472..f023fd65a 100644 --- a/python/ray/serve/tests/test_api.py +++ b/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" diff --git a/python/ray/serve/tests/test_failure.py b/python/ray/serve/tests/test_failure.py index 4919013a6..3917a4050 100644 --- a/python/ray/serve/tests/test_failure.py +++ b/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