[Serve] Track user-configured options in Serve deployments

python/ray/serve/BUILD

@@ -49,6 +49,14 @@ py_test(
     deps = [":serve_lib"],
 )
 
+py_test(
+    name = "test_deployment_class",
+    size = "small",
+    srcs = serve_tests_srcs,
+    tags = ["exclusive", "team:serve"],
+    deps = [":serve_lib"],
+)
+
 py_test(
     name = "test_healthcheck",
     size = "medium",

python/ray/serve/_private/deployment_function_node.py

@@ -58,6 +58,7 @@ def __init__(
                 init_args=(),
                 init_kwargs={},
                 route_prefix=route_prefix,
+                _internal=True,
             )
         else:
             self._deployment: Deployment = Deployment(

python/ray/serve/_private/deployment_graph_build.py

@@ -211,6 +211,7 @@ def replace_with_handle(node):
             init_args=replaced_deployment_init_args,
             init_kwargs=replaced_deployment_init_kwargs,
             route_prefix=route_prefix,
+            _internal=True,
         )
 
         return DeploymentNode(
@@ -377,6 +378,7 @@ def replace_with_handle(node):
     return original_driver_deployment.options(
         init_args=replaced_deployment_init_args,
         init_kwargs=replaced_deployment_init_kwargs,
+        _internal=True,
     )
 
 
@@ -422,7 +424,9 @@ def process_ingress_deployment_in_serve_dag(
     if ingress_deployment.route_prefix in [None, f"/{ingress_deployment.name}"]:
         # Override default prefix to "/" on the ingress deployment, if user
         # didn't provide anything in particular.
-        new_ingress_deployment = ingress_deployment.options(route_prefix="/")
+        new_ingress_deployment = ingress_deployment.options(
+            route_prefix="/", _internal=True
+        )
         deployments[-1] = new_ingress_deployment
 
     # Erase all non ingress deployment route prefix
@@ -438,8 +442,8 @@ def process_ingress_deployment_in_serve_dag(
                 "serve DAG. "
             )
         else:
-            # Earse all default prefix to None for non-ingress deployments to
+            # Erase all default prefix to None for non-ingress deployments to
             # disable HTTP
-            deployments[i] = deployment.options(route_prefix=None)
+            deployments[i] = deployment.options(route_prefix=None, _internal=True)
 
     return deployments

python/ray/serve/_private/json_serde.py

@@ -88,7 +88,10 @@ def default(self, obj):
         if isinstance(obj, DeploymentSchema):
             return {
                 DAGNODE_TYPE_KEY: "DeploymentSchema",
-                "schema": obj.dict(),
+                # The schema's default values are Python enums that aren't
+                # JSON-serializable by design. exclude_defaults omits these,
+                # so the return value can be JSON-serialized.
+                "schema": obj.dict(exclude_defaults=True),
             }
         elif isinstance(obj, RayServeHandle):
             return _serve_handle_to_json_dict(obj)

python/ray/serve/_private/utils.py

@@ -9,7 +9,7 @@
 import traceback
 from enum import Enum
 from functools import wraps
-from typing import Dict, Iterable, List, Tuple
+from typing import Dict, Iterable, List, Tuple, TypeVar, Union
 
 import fastapi.encoders
 import numpy as np
@@ -43,6 +43,11 @@ class DEFAULT(Enum):
     VALUE = 1
 
 
+# Type alias: objects that can be DEFAULT.VALUE have type Default[T]
+T = TypeVar("T")
+Default = Union[DEFAULT, T]
+
+
 def parse_request_item(request_item):
     if len(request_item.args) == 1:
         arg = request_item.args[0]

python/ray/serve/api.py

@@ -39,6 +39,7 @@
 from ray.serve._private.logging_utils import LoggingContext
 from ray.serve._private.utils import (
     DEFAULT,
+    Default,
     ensure_serialization_context,
     in_interactive_shell,
     install_serve_encoders_to_fastapi,
@@ -253,80 +254,81 @@ def deployment(func_or_class: Callable) -> Deployment:
 
 @overload
 def deployment(
-    name: Optional[str] = None,
-    version: Optional[str] = None,
-    num_replicas: Optional[int] = None,
-    init_args: Optional[Tuple[Any]] = None,
-    init_kwargs: Optional[Dict[Any, Any]] = None,
-    route_prefix: Union[str, None, DEFAULT] = DEFAULT.VALUE,
-    ray_actor_options: Optional[Dict] = None,
-    user_config: Optional[Any] = None,
-    max_concurrent_queries: Optional[int] = None,
-    autoscaling_config: Optional[Union[Dict, AutoscalingConfig]] = None,
-    graceful_shutdown_wait_loop_s: Optional[float] = None,
-    graceful_shutdown_timeout_s: Optional[float] = None,
-    health_check_period_s: Optional[float] = None,
-    health_check_timeout_s: Optional[float] = None,
+    name: Default[str] = DEFAULT.VALUE,
+    version: Default[str] = DEFAULT.VALUE,
+    num_replicas: Default[int] = DEFAULT.VALUE,
+    init_args: Default[Tuple[Any]] = DEFAULT.VALUE,
+    init_kwargs: Default[Dict[Any, Any]] = DEFAULT.VALUE,
+    route_prefix: Default[Union[str, None]] = DEFAULT.VALUE,
+    ray_actor_options: Default[Dict] = DEFAULT.VALUE,
+    user_config: Default[Any] = DEFAULT.VALUE,
+    max_concurrent_queries: Default[int] = DEFAULT.VALUE,
+    autoscaling_config: Default[Union[Dict, AutoscalingConfig]] = DEFAULT.VALUE,
+    graceful_shutdown_wait_loop_s: Default[float] = DEFAULT.VALUE,
+    graceful_shutdown_timeout_s: Default[float] = DEFAULT.VALUE,
+    health_check_period_s: Default[float] = DEFAULT.VALUE,
+    health_check_timeout_s: Default[float] = DEFAULT.VALUE,
 ) -> Callable[[Callable], Deployment]:
     pass
 
 
 @PublicAPI(stability="beta")
 def deployment(
     _func_or_class: Optional[Callable] = None,
-    name: Optional[str] = None,
-    version: Optional[str] = None,
-    num_replicas: Optional[int] = None,
-    init_args: Optional[Tuple[Any]] = None,
-    init_kwargs: Optional[Dict[Any, Any]] = None,
-    route_prefix: Union[str, None, DEFAULT] = DEFAULT.VALUE,
-    ray_actor_options: Optional[Dict] = None,
-    user_config: Optional[Any] = None,
-    max_concurrent_queries: Optional[int] = None,
-    autoscaling_config: Optional[Union[Dict, AutoscalingConfig]] = None,
-    graceful_shutdown_wait_loop_s: Optional[float] = None,
-    graceful_shutdown_timeout_s: Optional[float] = None,
-    health_check_period_s: Optional[float] = None,
-    health_check_timeout_s: Optional[float] = None,
+    name: Default[str] = DEFAULT.VALUE,
+    version: Default[str] = DEFAULT.VALUE,
+    num_replicas: Default[Optional[int]] = DEFAULT.VALUE,
+    init_args: Default[Tuple[Any]] = DEFAULT.VALUE,
+    init_kwargs: Default[Dict[Any, Any]] = DEFAULT.VALUE,
+    route_prefix: Default[Union[str, None]] = DEFAULT.VALUE,
+    ray_actor_options: Default[Dict] = DEFAULT.VALUE,
+    user_config: Default[Optional[Any]] = DEFAULT.VALUE,
+    max_concurrent_queries: Default[int] = DEFAULT.VALUE,
+    autoscaling_config: Default[Union[Dict, AutoscalingConfig, None]] = DEFAULT.VALUE,
+    graceful_shutdown_wait_loop_s: Default[float] = DEFAULT.VALUE,
+    graceful_shutdown_timeout_s: Default[float] = DEFAULT.VALUE,
+    health_check_period_s: Default[float] = DEFAULT.VALUE,
+    health_check_timeout_s: Default[float] = DEFAULT.VALUE,
 ) -> Callable[[Callable], Deployment]:
     """Define a Serve deployment.
 
     Args:
-        name (Optional[str]): Globally-unique name identifying this deployment.
-            If not provided, the name of the class or function will be used.
-        version [DEPRECATED] (Optional[str]): Version of the deployment. This is used to
-            indicate a code change for the deployment; when it is re-deployed
-            with a version change, a rolling update of the replicas will be
-            performed. If not provided, every deployment will be treated as a
-            new version.
-        num_replicas (Optional[int]): The number of processes to start up that
+        name (Default[str]): Globally-unique name identifying this
+            deployment. If not provided, the name of the class or function will
+            be used.
+        version [DEPRECATED] (Default[str]): Version of the deployment.
+            This is used to indicate a code change for the deployment; when it
+            is re-deployed with a version change, a rolling update of the
+            replicas will be performed. If not provided, every deployment will
+            be treated as a new version.
+        num_replicas (Default[Optional[int]]): The number of processes to start up that
             will handle requests to this deployment. Defaults to 1.
-        init_args (Optional[Tuple]): Positional args to be passed to the class
-            constructor when starting up deployment replicas. These can also be
-            passed when you call `.deploy()` on the returned Deployment.
-        init_kwargs (Optional[Dict]): Keyword args to be passed to the class
-            constructor when starting up deployment replicas. These can also be
-            passed when you call `.deploy()` on the returned Deployment.
-        route_prefix (Optional[str]): Requests to paths under this HTTP path
-            prefix will be routed to this deployment. Defaults to '/{name}'.
-            When set to 'None', no HTTP endpoint will be created.
+        init_args (Default[Tuple[Any]]): Positional args to be passed to the
+            class constructor when starting up deployment replicas. These can
+            also be passed when you call `.deploy()` on the returned Deployment.
+        init_kwargs (Default[Dict[Any, Any]]): Keyword args to be passed to the
+            class constructor when starting up deployment replicas. These can
+            also be passed when you call `.deploy()` on the returned Deployment.
+        route_prefix (Default[Union[str, None]]): Requests to paths under this
+            HTTP path prefix will be routed to this deployment. Defaults to
+            '/{name}'. When set to 'None', no HTTP endpoint will be created.
             Routing is done based on longest-prefix match, so if you have
             deployment A with a prefix of '/a' and deployment B with a prefix
             of '/a/b', requests to '/a', '/a/', and '/a/c' go to A and requests
             to '/a/b', '/a/b/', and '/a/b/c' go to B. Routes must not end with
             a '/' unless they're the root (just '/'), which acts as a
             catch-all.
-        ray_actor_options: Options to be passed to the Ray actor
-            constructor such as resource requirements. Valid options are
+        ray_actor_options (Default[Dict]): Options to be passed to the Ray
+            actor constructor such as resource requirements. Valid options are
             `accelerator_type`, `memory`, `num_cpus`, `num_gpus`,
             `object_store_memory`, `resources`, and `runtime_env`.
-        user_config (Optional[Any]): Config to pass to the
+        user_config (Default[Optional[Any]]): Config to pass to the
             reconfigure method of the deployment. This can be updated
             dynamically without changing the version of the deployment and
             restarting its replicas. The user_config must be json-serializable
             to keep track of updates, so it must only contain json-serializable
             types, or json-serializable types nested in lists and dictionaries.
-        max_concurrent_queries (Optional[int]): The maximum number of queries
+        max_concurrent_queries (Default[int]): The maximum number of queries
             that will be sent to a replica of this deployment without receiving
             a response. Defaults to 100.
 
@@ -344,26 +346,35 @@ def deployment(
         Deployment
     """
 
+    # Create list of all user-configured options from keyword args
+    user_configured_option_names = [
+        option
+        for option, value in locals().items()
+        if option != "_func_or_class" and value is not DEFAULT.VALUE
+    ]
+
     # Num of replicas should not be 0.
     # TODO(Sihan) seperate num_replicas attribute from internal and api
     if num_replicas == 0:
         raise ValueError("num_replicas is expected to larger than 0")
 
-    if num_replicas is not None and autoscaling_config is not None:
+    if num_replicas not in [DEFAULT.VALUE, None] and autoscaling_config not in [
+        DEFAULT.VALUE,
+        None,
+    ]:
         raise ValueError(
             "Manually setting num_replicas is not allowed when "
             "autoscaling_config is provided."
         )
 
-    if version is not None:
+    if version is not DEFAULT.VALUE:
         logger.warning(
             "DeprecationWarning: `version` in `@serve.deployment` has been deprecated. "
             "Explicitly specifying version will raise an error in the future!"
         )
 
     config = DeploymentConfig.from_default(
-        ignore_none=True,
-        num_replicas=num_replicas,
+        num_replicas=num_replicas if num_replicas is not None else 1,
         user_config=user_config,
         max_concurrent_queries=max_concurrent_queries,
         autoscaling_config=autoscaling_config,
@@ -372,17 +383,20 @@ def deployment(
         health_check_period_s=health_check_period_s,
         health_check_timeout_s=health_check_timeout_s,
     )
+    config.user_configured_option_names = set(user_configured_option_names)
 
     def decorator(_func_or_class):
         return Deployment(
             _func_or_class,
-            name if name is not None else _func_or_class.__name__,
+            name if name is not DEFAULT.VALUE else _func_or_class.__name__,
             config,
-            version=version,
-            init_args=init_args,
-            init_kwargs=init_kwargs,
+            version=(version if version is not DEFAULT.VALUE else None),
+            init_args=(init_args if init_args is not DEFAULT.VALUE else None),
+            init_kwargs=(init_kwargs if init_kwargs is not DEFAULT.VALUE else None),
             route_prefix=route_prefix,
-            ray_actor_options=ray_actor_options,
+            ray_actor_options=(
+                ray_actor_options if ray_actor_options is not DEFAULT.VALUE else None
+            ),
             _internal=True,
         )
 

python/ray/serve/config.py

@@ -1,7 +1,7 @@
 import inspect
 import json
 from enum import Enum
-from typing import Any, Callable, Dict, List, Optional, Tuple, Union
+from typing import Any, Callable, Dict, List, Optional, Tuple, Union, Set
 
 import pydantic
 from google.protobuf.json_format import MessageToDict
@@ -23,6 +23,7 @@
     DEFAULT_HTTP_HOST,
     DEFAULT_HTTP_PORT,
 )
+from ray.serve._private.utils import DEFAULT
 from ray.serve.generated.serve_pb2 import (
     DeploymentConfig as DeploymentConfigProto,
     DeploymentLanguage,
@@ -122,6 +123,8 @@ class DeploymentConfig(BaseModel):
         health_check_timeout_s (Optional[float]):
             Timeout that the controller will wait for a response from the
             replica's health check before marking it unhealthy.
+        user_configured_option_names (Set[str]):
+            The names of options manually configured by the user.
     """
 
     num_replicas: NonNegativeInt = 1
@@ -150,6 +153,9 @@ class DeploymentConfig(BaseModel):
 
     version: Optional[str] = None
 
+    # Contains the names of deployment options manually set by the user
+    user_configured_option_names: Set[str] = set()
+
     class Config:
         validate_assignment = True
         extra = "forbid"
@@ -182,13 +188,16 @@ def needs_pickle(self):
 
     def to_proto(self):
         data = self.dict()
-        if data.get("user_config"):
+        if data.get("user_config") is not None:
             if self.needs_pickle():
                 data["user_config"] = cloudpickle.dumps(data["user_config"])
         if data.get("autoscaling_config"):
             data["autoscaling_config"] = AutoscalingConfigProto(
                 **data["autoscaling_config"]
             )
+        data["user_configured_option_names"] = list(
+            data["user_configured_option_names"]
+        )
         return DeploymentConfigProto(**data)
 
     def to_proto_bytes(self):
@@ -225,6 +234,10 @@ def from_proto(cls, proto: DeploymentConfigProto):
         if "version" in data:
             if data["version"] == "":
                 data["version"] = None
+        if "user_configured_option_names" in data:
+            data["user_configured_option_names"] = set(
+                data["user_configured_option_names"]
+            )
         return cls(**data)
 
     @classmethod
@@ -233,16 +246,10 @@ def from_proto_bytes(cls, proto_bytes: bytes):
         return cls.from_proto(proto)
 
     @classmethod
-    def from_default(cls, ignore_none: bool = False, **kwargs):
+    def from_default(cls, **kwargs):
         """Creates a default DeploymentConfig and overrides it with kwargs.
 
-        Only accepts the same keywords as the class. Passing in any other
-        keyword raises a ValueError.
-
-        Args:
-            ignore_none: When True, any valid keywords with value None
-                are ignored, and their values stay default. Invalid keywords
-                still raise a TypeError.
+        Ignores any kwargs set to DEFAULT.VALUE.
 
         Raises:
             TypeError: when a keyword that's not an argument to the class is
@@ -262,8 +269,7 @@ def from_default(cls, ignore_none: bool = False, **kwargs):
                     f"{list(valid_config_options)}."
                 )
 
-        if ignore_none:
-            kwargs = {key: val for key, val in kwargs.items() if val is not None}
+        kwargs = {key: val for key, val in kwargs.items() if val != DEFAULT.VALUE}
 
         for key, val in kwargs.items():
             config.__setattr__(key, val)