from copy import deepcopy
from typing import Any, Callable, Dict, List, Optional, Tuple, Union
from ray.serve._private.config import (
DeploymentConfig,
ReplicaConfig,
RequestRouterConfig,
handle_num_replicas_auto,
)
from ray.serve._private.usage import ServeUsageTag
from ray.serve._private.utils import DEFAULT, Default
from ray.serve.config import (
AutoscalingConfig,
BackpressureConfig,
DeploymentActorConfig,
GangSchedulingConfig,
)
from ray.serve.schema import DeploymentSchema, LoggingConfig, RayActorOptionsSchema
from ray.util.annotations import PublicAPI
[docs]
@PublicAPI(stability="stable")
class Application:
"""One or more deployments bound with arguments that can be deployed together.
Can be passed into another `Deployment.bind()` to compose multiple deployments in a
single application, passed to `serve.run`, or deployed via a Serve config file.
For example, to define an Application and run it in Python:
.. code-block:: python
from ray import serve
from ray.serve import Application
@serve.deployment
class MyDeployment:
pass
app: Application = MyDeployment.bind(OtherDeployment.bind())
serve.run(app)
To run the same app using the command line interface (CLI):
.. code-block:: bash
serve run python_file:app
To deploy the same app via a config file:
.. code-block:: yaml
applications:
my_app:
import_path: python_file:app
"""
def __init__(self, bound_deployment: "Deployment"):
# This is used by `build_app`, but made private so users don't use it.
self._bound_deployment = bound_deployment
# Optional peer ingress request router for ingress bypass mode.
self._ingress_request_router: Optional["Application"] = None
def _with_ingress_request_router(
self, ingress_request_router: "Application"
) -> "Application":
# Internal-only, unstable hook for the Serve LLM direct-ingress stack.
# This is not a stable public Serve API.
self._ingress_request_router = ingress_request_router
return self
[docs]
@PublicAPI(stability="stable")
class Deployment:
"""Class (or function) decorated with the `@serve.deployment` decorator.
This is run on a number of replica actors. Requests to those replicas call
this class.
One or more deployments can be composed together into an `Application` which is
then run via `serve.run` or a config file.
Example:
.. code-block:: python
@serve.deployment
class MyDeployment:
def __init__(self, name: str):
self._name = name
def __call__(self, request):
return "Hello world!"
app = MyDeployment.bind()
# Run via `serve.run` or the `serve run` CLI command.
serve.run(app)
"""
def __init__(
self,
name: str,
deployment_config: DeploymentConfig,
replica_config: ReplicaConfig,
version: Optional[str] = None,
_internal: bool = False,
) -> None:
"""Construct a Deployment. Should only be called by Serve internals.
Args:
name: Unique name of this deployment.
deployment_config: Serve-level configuration (number of replicas,
user config, autoscaling, etc.).
replica_config: Replica-level configuration (actor options, init
args/kwargs, etc.).
version: Optional opaque deployment version used to determine
whether replicas need to be restarted on update.
_internal: Internal flag; ``Deployment`` instances must be created
via the ``@serve.deployment`` decorator, which sets this to
``True``.
"""
if not _internal:
raise RuntimeError(
"The Deployment constructor should not be called "
"directly. Use `@serve.deployment` instead."
)
self._validate_name(name)
if not (version is None or isinstance(version, str)):
raise TypeError("version must be a string.")
self._name = name
self._version = version
self._deployment_config = deployment_config
self._replica_config = replica_config
def _validate_name(self, name: str):
if not isinstance(name, str):
raise TypeError("name must be a string.")
# '#' is the replica ID delimiter, so it cannot appear in a name.
if "#" in name:
raise ValueError(
f"Deployment names must not contain the '#' character. "
f"Current name: {name}."
)
@property
def name(self) -> str:
"""Unique name of this deployment."""
return self._name
@property
def version(self) -> Optional[str]:
return self._version
@property
def func_or_class(self) -> Union[Callable, str]:
"""Underlying class or function that this deployment wraps."""
return self._replica_config.deployment_def
@property
def num_replicas(self) -> int:
"""Target number of replicas."""
return self._deployment_config.num_replicas
@property
def user_config(self) -> Any:
"""Dynamic user-provided config options."""
return self._deployment_config.user_config
@property
def max_ongoing_requests(self) -> int:
"""Max number of requests a replica can handle at once."""
return self._deployment_config.max_ongoing_requests
@property
def max_queued_requests(self) -> int:
"""Max number of requests that can be queued in each deployment handle."""
return self._deployment_config.max_queued_requests
@property
def backpressure_config(self) -> BackpressureConfig:
"""Config of the HTTP response for requests rejected due to backpressure."""
return self._deployment_config.backpressure_config
@property
def ray_actor_options(self) -> Optional[Dict]:
"""Actor options such as resources required for each replica."""
return self._replica_config.ray_actor_options
@property
def init_args(self) -> Tuple[Any]:
return self._replica_config.init_args
@property
def init_kwargs(self) -> Tuple[Any]:
return self._replica_config.init_kwargs
@property
def logging_config(self) -> Dict:
return self._deployment_config.logging_config
def set_logging_config(self, logging_config: Dict):
self._deployment_config.logging_config = logging_config
def __call__(self):
raise RuntimeError(
"Deployments cannot be constructed directly. "
"Use `deployment.deploy() instead.`"
)
[docs]
def bind(self, *args, **kwargs) -> Application:
"""Bind the arguments to the deployment and return an Application.
The returned Application can be deployed using `serve.run` (or via
config file) or bound to another deployment for composition.
"""
return Application(self.options(_init_args=args, _init_kwargs=kwargs))
[docs]
def options(
self,
func_or_class: Optional[Callable] = None,
name: Default[str] = DEFAULT.VALUE,
version: Default[str] = DEFAULT.VALUE,
num_replicas: Default[Optional[Union[int, str]]] = DEFAULT.VALUE,
ray_actor_options: Default[Optional[Dict]] = DEFAULT.VALUE,
placement_group_bundles: Default[List[Dict[str, float]]] = DEFAULT.VALUE,
placement_group_strategy: Default[str] = DEFAULT.VALUE,
placement_group_bundle_label_selector: Default[
List[Dict[str, str]]
] = DEFAULT.VALUE,
max_replicas_per_node: Default[int] = DEFAULT.VALUE,
user_config: Default[Optional[Any]] = DEFAULT.VALUE,
max_ongoing_requests: Default[int] = DEFAULT.VALUE,
max_queued_requests: Default[int] = DEFAULT.VALUE,
backpressure_config: Default[
Union[Dict, BackpressureConfig, None]
] = 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,
logging_config: Default[Union[Dict, LoggingConfig, None]] = DEFAULT.VALUE,
request_router_config: Default[
Union[Dict, RequestRouterConfig, None]
] = DEFAULT.VALUE,
_init_args: Default[Tuple[Any]] = DEFAULT.VALUE,
_init_kwargs: Default[Dict[Any, Any]] = DEFAULT.VALUE,
_internal: bool = False,
max_constructor_retry_count: Default[int] = DEFAULT.VALUE,
gang_scheduling_config: Default[
Union[Dict, GangSchedulingConfig, None]
] = DEFAULT.VALUE,
deployment_actors: Default[
Optional[List[Union[Dict, DeploymentActorConfig]]]
] = DEFAULT.VALUE,
max_surge_percent: Default[int] = DEFAULT.VALUE,
) -> "Deployment":
"""Return a copy of this deployment with updated options.
Only those options passed in will be updated, all others will remain
unchanged from the existing deployment.
See the `@serve.deployment` decorator docs for details on each option.
Args:
func_or_class: The class or function to run as this deployment.
name: Unique name for this deployment within the application.
version: Removed. Specifying this argument raises a ValueError.
num_replicas: Number of replicas handling requests, or "auto" to
autoscale.
ray_actor_options: Options for the replica actor, such as resources and
runtime environment.
placement_group_bundles: Placement group bundles scheduled for each
replica. Cannot be combined with `max_replicas_per_node`.
placement_group_strategy: Scheduling strategy for those bundles.
placement_group_bundle_label_selector: Label selectors applied to the
placement group, either one for all bundles or one per bundle.
max_replicas_per_node: Maximum replicas of this deployment on one node.
Cannot be combined with `placement_group_bundles`.
user_config: Config passed to the replica's `reconfigure` method without
restarting replicas.
max_ongoing_requests: Maximum number of requests a replica can handle
at once.
max_queued_requests: Maximum number of requests queued at each caller.
Additional requests are rejected when the queue is full.
backpressure_config: HTTP response settings for requests rejected
because the queue is full.
autoscaling_config: Autoscaling parameters. Set `num_replicas` to "auto"
or leave it unset when using this.
graceful_shutdown_wait_loop_s: Seconds between checks for unfinished
requests during shutdown.
graceful_shutdown_timeout_s: Seconds allowed for a graceful shutdown
before the replica is forcibly stopped.
health_check_period_s: Seconds between health checks on a replica.
health_check_timeout_s: Seconds allowed for a health check to finish
before it is considered failed.
logging_config: Logging options for this deployment's replicas.
request_router_config: Request routing options for this deployment.
_init_args: Internal. Positional arguments for the deployment
constructor; use `.bind()` instead.
_init_kwargs: Internal. Keyword arguments for the deployment
constructor; use `.bind()` instead.
_internal: Internal. Set by Serve when it calls this method.
max_constructor_retry_count: Maximum retries of the deployment
constructor before the deployment is marked failed.
gang_scheduling_config: Options for scheduling replicas together in
groups called gangs.
deployment_actors: Actors shared by all replicas of this deployment.
max_surge_percent: Extra replicas allowed during a rolling restart,
as a percentage of the target count. Must be in ``[0, 100]``.
The allowance rounds up to whole replicas or gangs. When positive,
replacements start before old replicas stop. If the cluster lacks
capacity, old replicas keep serving while replacements wait.
Stopping replicas do not count toward the limit, so live actors
can exceed the target plus surge while old replicas drain.
Lightweight configuration changes to replicas on the new code
wait until restarts finish. Setting this to ``0`` stops old
replicas before starting replacements.
Returns:
A copy of this deployment with the given options applied.
"""
if not _internal and version is not DEFAULT.VALUE:
raise ValueError(
"`version` in `Deployment.options()` has been removed. "
"Serve manages deployment versions internally."
)
# Modify max_ongoing_requests and autoscaling_config if
# `num_replicas="auto"`
if max_ongoing_requests is None:
raise ValueError("`max_ongoing_requests` must be non-null, got None.")
if num_replicas == "auto":
max_ongoing_requests, autoscaling_config = handle_num_replicas_auto(
max_ongoing_requests, autoscaling_config
)
ServeUsageTag.AUTO_NUM_REPLICAS_USED.record("1")
# NOTE: The user_configured_option_names should be the first thing that's
# defined in this method. It depends on the locals() dictionary storing
# only the function args/kwargs.
# Create list of all user-configured options from keyword args
user_configured_option_names = [
option
for option, value in locals().items()
if option not in {"self", "func_or_class", "_internal"}
and value is not DEFAULT.VALUE
]
new_deployment_config = deepcopy(self._deployment_config)
if not _internal:
new_deployment_config.user_configured_option_names.update(
user_configured_option_names
)
if num_replicas not in [
DEFAULT.VALUE,
None,
"auto",
] and autoscaling_config not in [
DEFAULT.VALUE,
None,
]:
raise ValueError(
"Manually setting num_replicas is not allowed when "
"autoscaling_config is provided."
)
if num_replicas == 0:
raise ValueError("num_replicas is expected to larger than 0")
if num_replicas not in [DEFAULT.VALUE, None, "auto"]:
new_deployment_config.num_replicas = num_replicas
if user_config is not DEFAULT.VALUE:
new_deployment_config.user_config = user_config
if max_ongoing_requests is not DEFAULT.VALUE:
new_deployment_config.max_ongoing_requests = max_ongoing_requests
if max_queued_requests is not DEFAULT.VALUE:
new_deployment_config.max_queued_requests = max_queued_requests
if backpressure_config is not DEFAULT.VALUE:
new_deployment_config.backpressure_config = backpressure_config
if max_constructor_retry_count is not DEFAULT.VALUE:
new_deployment_config.max_constructor_retry_count = (
max_constructor_retry_count
)
if max_surge_percent is not DEFAULT.VALUE:
new_deployment_config.max_surge_percent = max_surge_percent
if func_or_class is None:
func_or_class = self._replica_config.deployment_def
if name is DEFAULT.VALUE:
name = self._name
if version is DEFAULT.VALUE:
version = self._version
if _init_args is DEFAULT.VALUE:
_init_args = self._replica_config.init_args
if _init_kwargs is DEFAULT.VALUE:
_init_kwargs = self._replica_config.init_kwargs
if ray_actor_options is DEFAULT.VALUE:
ray_actor_options = self._replica_config.ray_actor_options
if placement_group_bundles is DEFAULT.VALUE:
placement_group_bundles = self._replica_config.placement_group_bundles
if placement_group_strategy is DEFAULT.VALUE:
placement_group_strategy = self._replica_config.placement_group_strategy
if placement_group_bundle_label_selector is DEFAULT.VALUE:
placement_group_bundle_label_selector = (
self._replica_config.placement_group_bundle_label_selector
)
# TODO(ryanaoleary@): Add conditional check once fallback_strategy is
# added to placement group options.
placement_group_fallback_strategy = (
self._replica_config.placement_group_fallback_strategy
)
if max_replicas_per_node is DEFAULT.VALUE:
max_replicas_per_node = self._replica_config.max_replicas_per_node
if autoscaling_config is not DEFAULT.VALUE:
new_deployment_config.autoscaling_config = autoscaling_config
if request_router_config is not DEFAULT.VALUE:
new_deployment_config.request_router_config = request_router_config
if graceful_shutdown_wait_loop_s is not DEFAULT.VALUE:
new_deployment_config.graceful_shutdown_wait_loop_s = (
graceful_shutdown_wait_loop_s
)
if graceful_shutdown_timeout_s is not DEFAULT.VALUE:
new_deployment_config.graceful_shutdown_timeout_s = (
graceful_shutdown_timeout_s
)
if health_check_period_s is not DEFAULT.VALUE:
new_deployment_config.health_check_period_s = health_check_period_s
if health_check_timeout_s is not DEFAULT.VALUE:
new_deployment_config.health_check_timeout_s = health_check_timeout_s
if logging_config is not DEFAULT.VALUE:
if isinstance(logging_config, LoggingConfig):
logging_config = logging_config.model_dump()
new_deployment_config.logging_config = logging_config
if gang_scheduling_config is not DEFAULT.VALUE:
new_deployment_config.gang_scheduling_config = gang_scheduling_config
if deployment_actors is not DEFAULT.VALUE:
new_deployment_config.deployment_actors = deployment_actors
gc = new_deployment_config.gang_scheduling_config
if (
gc is not None
and isinstance(new_deployment_config.num_replicas, int)
and new_deployment_config.autoscaling_config is None
):
# When autoscaling is enabled, num_replicas defaults to 1
if new_deployment_config.num_replicas % gc.gang_size != 0:
raise ValueError(
f"num_replicas ({new_deployment_config.num_replicas}) must "
f"be a multiple of gang_size ({gc.gang_size})."
)
if gc is not None and max_replicas_per_node is not None:
raise ValueError(
"Setting max_replicas_per_node is not allowed when "
"gang_scheduling_config is provided."
)
if gc is not None and placement_group_strategy is not None:
raise ValueError(
"Setting placement_group_strategy is not allowed when "
"gang_scheduling_config is provided. Use "
"gang_scheduling_config.gang_placement_strategy instead."
)
new_replica_config = ReplicaConfig.create(
func_or_class,
init_args=_init_args,
init_kwargs=_init_kwargs,
ray_actor_options=ray_actor_options,
placement_group_bundles=placement_group_bundles,
placement_group_strategy=placement_group_strategy,
placement_group_bundle_label_selector=placement_group_bundle_label_selector,
placement_group_fallback_strategy=placement_group_fallback_strategy,
max_replicas_per_node=max_replicas_per_node,
)
return Deployment(
name,
new_deployment_config,
new_replica_config,
version=version,
_internal=True,
)
def __eq__(self, other):
return all(
[
self._name == other._name,
self._version == other._version,
self._deployment_config == other._deployment_config,
self._replica_config.init_args == other._replica_config.init_args,
self._replica_config.init_kwargs == other._replica_config.init_kwargs,
self._replica_config.ray_actor_options
== other._replica_config.ray_actor_options,
]
)
def __str__(self):
return f"Deployment(name={self._name})"
def __repr__(self):
return str(self)
def deployment_to_schema(d: Deployment) -> DeploymentSchema:
"""Converts a live deployment object to a corresponding structured schema.
Args:
d: Deployment object to convert
Returns:
The structured ``DeploymentSchema`` representing ``d``.
"""
if d.ray_actor_options is not None:
ray_actor_options_schema = RayActorOptionsSchema.model_validate(
d.ray_actor_options
)
else:
ray_actor_options_schema = None
deployment_options = {
"name": d.name,
"num_replicas": None
if d._deployment_config.autoscaling_config
else d.num_replicas,
"max_ongoing_requests": d.max_ongoing_requests,
"max_queued_requests": d.max_queued_requests,
"backpressure_config": d.backpressure_config,
"user_config": d.user_config,
"autoscaling_config": d._deployment_config.autoscaling_config,
"graceful_shutdown_wait_loop_s": d._deployment_config.graceful_shutdown_wait_loop_s, # noqa: E501
"graceful_shutdown_timeout_s": d._deployment_config.graceful_shutdown_timeout_s,
"health_check_period_s": d._deployment_config.health_check_period_s,
"health_check_timeout_s": d._deployment_config.health_check_timeout_s,
"ray_actor_options": ray_actor_options_schema,
"placement_group_strategy": d._replica_config.placement_group_strategy,
"placement_group_bundles": d._replica_config.placement_group_bundles,
"max_replicas_per_node": d._replica_config.max_replicas_per_node,
"logging_config": d._deployment_config.logging_config,
"request_router_config": d._deployment_config.request_router_config,
"gang_scheduling_config": d._deployment_config.gang_scheduling_config,
"deployment_actors": d._deployment_config.deployment_actors,
"rolling_update_percentage": d._deployment_config.rolling_update_percentage,
"max_surge_percent": d._deployment_config.max_surge_percent,
}
# Let non-user-configured options be set to defaults. If the schema
# is converted back to a deployment, this lets Serve continue tracking
# which options were set by the user. Name is a required field in the
# schema, so it should be passed in explicitly.
for option in list(deployment_options.keys()):
if (
option != "name"
and option not in d._deployment_config.user_configured_option_names
):
del deployment_options[option]
# TODO(Sihan) DeploymentConfig num_replicas and auto_config can be set together
# because internally we use these two field for autoscale and deploy.
# We can improve the code after we separate the user faced deployment config and
# internal deployment config.
return DeploymentSchema(**deployment_options)
def schema_to_deployment(s: DeploymentSchema) -> Deployment:
"""Creates a deployment with parameters specified in schema.
The returned deployment CANNOT be deployed immediately. It's func_or_class
value is an empty string (""), which is not a valid import path. The
func_or_class value must be overwritten with a valid function or class
before the deployment can be deployed.
"""
if s.ray_actor_options is DEFAULT.VALUE:
ray_actor_options = None
else:
ray_actor_options = s.ray_actor_options.model_dump(exclude_unset=True)
if s.placement_group_bundles is DEFAULT.VALUE:
placement_group_bundles = None
else:
placement_group_bundles = s.placement_group_bundles
if s.placement_group_strategy is DEFAULT.VALUE:
placement_group_strategy = None
else:
placement_group_strategy = s.placement_group_strategy
if s.max_replicas_per_node is DEFAULT.VALUE:
max_replicas_per_node = None
else:
max_replicas_per_node = s.max_replicas_per_node
deployment_config = DeploymentConfig.from_default(
num_replicas=s.num_replicas,
user_config=s.user_config,
max_ongoing_requests=s.max_ongoing_requests,
max_queued_requests=s.max_queued_requests,
backpressure_config=s.backpressure_config,
autoscaling_config=s.autoscaling_config,
graceful_shutdown_wait_loop_s=s.graceful_shutdown_wait_loop_s,
graceful_shutdown_timeout_s=s.graceful_shutdown_timeout_s,
health_check_period_s=s.health_check_period_s,
health_check_timeout_s=s.health_check_timeout_s,
logging_config=s.logging_config,
request_router_config=s.request_router_config,
gang_scheduling_config=s.gang_scheduling_config,
deployment_actors=s.deployment_actors,
rolling_update_percentage=s.rolling_update_percentage,
max_surge_percent=s.max_surge_percent,
)
deployment_config.user_configured_option_names = (
s._get_user_configured_option_names()
)
replica_config = ReplicaConfig.create(
deployment_def="",
init_args=(),
init_kwargs={},
ray_actor_options=ray_actor_options,
placement_group_bundles=placement_group_bundles,
placement_group_strategy=placement_group_strategy,
max_replicas_per_node=max_replicas_per_node,
)
return Deployment(
name=s.name,
deployment_config=deployment_config,
replica_config=replica_config,
_internal=True,
)