wehub-resource-sync
1124 lines
46 KiB
Python
1124 lines
46 KiB
Python
import inspect
|
|
import json
|
|
from typing import Any, Callable, Dict, List, Optional, Set, Tuple, Union
|
|
|
|
from google.protobuf.descriptor import FieldDescriptor
|
|
from google.protobuf.message import Message
|
|
from pydantic import (
|
|
BaseModel,
|
|
ConfigDict,
|
|
Field,
|
|
NonNegativeFloat,
|
|
NonNegativeInt,
|
|
PositiveFloat,
|
|
PositiveInt,
|
|
field_validator,
|
|
model_validator,
|
|
)
|
|
|
|
from ray import cloudpickle
|
|
from ray._common import ray_option_utils
|
|
from ray._common.serialization import pickle_dumps
|
|
from ray._common.utils import resources_from_ray_options
|
|
from ray.serve._private.constants import (
|
|
DEFAULT_CONSTRUCTOR_RETRY_COUNT,
|
|
DEFAULT_GRACEFUL_SHUTDOWN_TIMEOUT_S,
|
|
DEFAULT_GRACEFUL_SHUTDOWN_WAIT_LOOP_S,
|
|
DEFAULT_HEALTH_CHECK_PERIOD_S,
|
|
DEFAULT_HEALTH_CHECK_TIMEOUT_S,
|
|
DEFAULT_MAX_ONGOING_REQUESTS,
|
|
DEFAULT_ROLLING_UPDATE_PERCENTAGE,
|
|
MAX_REPLICAS_PER_NODE_MAX_VALUE,
|
|
)
|
|
from ray.serve._private.utils import DEFAULT, DeploymentOptionUpdateType
|
|
from ray.serve.config import (
|
|
AggregationFunction,
|
|
AutoscalingConfig,
|
|
DeploymentActorConfig,
|
|
GangPlacementStrategy,
|
|
GangRuntimeFailurePolicy,
|
|
GangSchedulingConfig,
|
|
RequestRouterConfig,
|
|
)
|
|
from ray.serve.generated.serve_pb2 import (
|
|
AutoscalingConfig as AutoscalingConfigProto,
|
|
DeploymentActorConfig as DeploymentActorConfigProto,
|
|
DeploymentConfig as DeploymentConfigProto,
|
|
DeploymentLanguage,
|
|
EncodingType as EncodingTypeProto,
|
|
GangPlacementStrategy as GangPlacementStrategyProto,
|
|
GangRuntimeFailurePolicy as GangRuntimeFailurePolicyProto,
|
|
GangSchedulingConfig as GangSchedulingConfigProto,
|
|
LoggingConfig as LoggingConfigProto,
|
|
ReplicaConfig as ReplicaConfigProto,
|
|
RequestRouterConfig as RequestRouterConfigProto,
|
|
)
|
|
from ray.util.placement_group import validate_placement_group
|
|
|
|
|
|
def _needs_pickle(deployment_language: DeploymentLanguage, is_cross_language: bool):
|
|
"""From Serve client API's perspective, decide whether pickling is needed."""
|
|
if deployment_language == DeploymentLanguage.PYTHON and not is_cross_language:
|
|
# Python client deploying Python replicas.
|
|
return True
|
|
elif deployment_language == DeploymentLanguage.JAVA and is_cross_language:
|
|
# Python client deploying Java replicas,
|
|
# using xlang serialization via cloudpickle.
|
|
return True
|
|
else:
|
|
return False
|
|
|
|
|
|
# protobuf>=7 removed the deprecated FieldDescriptor.label in favor of the
|
|
# is_repeated property; detect once at import and bind the right check.
|
|
if hasattr(FieldDescriptor, "is_repeated"):
|
|
|
|
def _field_is_repeated(field: FieldDescriptor) -> bool:
|
|
return bool(field.is_repeated)
|
|
|
|
else:
|
|
|
|
def _field_is_repeated(field: FieldDescriptor) -> bool:
|
|
return field.label == FieldDescriptor.LABEL_REPEATED
|
|
|
|
|
|
def _proto_to_dict(proto: Message) -> Dict:
|
|
"""Recursively convert a protobuf into a Python dictionary.
|
|
|
|
This is an alternative to protobuf's `MessageToDict`. Unlike
|
|
`MessageToDict`, this function doesn't add an extra base64
|
|
encoding to bytes when constructing a json response.
|
|
"""
|
|
data = {}
|
|
# Fill data with non-empty fields.
|
|
for field, value in proto.ListFields():
|
|
# Handle repeated fields
|
|
if _field_is_repeated(field):
|
|
# if we dont do this block the repeated field will be a list of
|
|
# `google.protobuf.internal.containers.RepeatedScalarFieldContainer
|
|
# Explicitly convert to list
|
|
if field.type == FieldDescriptor.TYPE_MESSAGE:
|
|
data[field.name] = [
|
|
_proto_to_dict(v) for v in value
|
|
] # Convert each item
|
|
else:
|
|
data[field.name] = list(value) # Convert to list directly
|
|
# Recursively call if the field is another protobuf.
|
|
elif field.type == FieldDescriptor.TYPE_MESSAGE:
|
|
data[field.name] = _proto_to_dict(value)
|
|
else:
|
|
data[field.name] = value
|
|
|
|
# Fill data default values.
|
|
for field in proto.DESCRIPTOR.fields:
|
|
if (
|
|
field.name not in data # skip the fields that are already set
|
|
and field.type != FieldDescriptor.TYPE_MESSAGE # skip nested messages
|
|
and not field.containing_oneof # skip optional fields
|
|
):
|
|
data[field.name] = field.default_value
|
|
return data
|
|
|
|
|
|
class DeploymentConfig(BaseModel):
|
|
"""Internal datastructure wrapping config options for a deployment.
|
|
|
|
Args:
|
|
num_replicas: The number of processes to start up that
|
|
handles requests to this deployment. Defaults to 1.
|
|
max_ongoing_requests: The maximum number of queries
|
|
that is sent to a replica of this deployment without receiving
|
|
a response. Defaults to 5.
|
|
max_queued_requests: Maximum number of requests to this deployment that will be
|
|
queued at each *caller* (proxy or DeploymentHandle). Once this limit is
|
|
reached, subsequent requests will raise a BackPressureError (for handles) or
|
|
return an HTTP 503 status code (for HTTP requests). Defaults to -1 (no
|
|
limit).
|
|
user_config: Arguments to pass to the reconfigure
|
|
method of the deployment. The reconfigure method is called if
|
|
user_config is not None. Must be JSON-serializable.
|
|
graceful_shutdown_wait_loop_s: Duration
|
|
that deployment replicas wait until there is no more work to
|
|
be done before shutting down.
|
|
graceful_shutdown_timeout_s: Controller waits for this duration
|
|
to forcefully kill the replica for shutdown.
|
|
health_check_period_s: Frequency at which the controller health
|
|
checks replicas.
|
|
health_check_timeout_s: Timeout that the controller waits for a
|
|
response from the replica's health check before marking it
|
|
unhealthy.
|
|
autoscaling_config: Autoscaling configuration.
|
|
logging_config: Configuration for deployment logs.
|
|
user_configured_option_names: The names of options manually
|
|
configured by the user.
|
|
request_router_config: Configuration for deployment request router.
|
|
max_constructor_retry_count: Maximum number of times to retry the
|
|
deployment constructor. Defaults to 20.
|
|
rolling_update_percentage: The fraction of replicas (of
|
|
``target_num_replicas``) to update at a time during a rolling
|
|
update. Must be in ``(0.0, 1.0]``. Defaults to 0.2 (20%).
|
|
"""
|
|
|
|
num_replicas: Optional[NonNegativeInt] = Field(
|
|
default=1, update_type=DeploymentOptionUpdateType.LightWeight
|
|
)
|
|
max_ongoing_requests: PositiveInt = Field(
|
|
default=DEFAULT_MAX_ONGOING_REQUESTS,
|
|
update_type=DeploymentOptionUpdateType.NeedsActorReconfigure,
|
|
)
|
|
max_queued_requests: int = Field(
|
|
default=-1,
|
|
update_type=DeploymentOptionUpdateType.LightWeight,
|
|
)
|
|
user_config: Any = Field(
|
|
default=None, update_type=DeploymentOptionUpdateType.NeedsActorReconfigure
|
|
)
|
|
|
|
graceful_shutdown_timeout_s: NonNegativeFloat = Field(
|
|
default=DEFAULT_GRACEFUL_SHUTDOWN_TIMEOUT_S,
|
|
update_type=DeploymentOptionUpdateType.NeedsReconfigure,
|
|
)
|
|
graceful_shutdown_wait_loop_s: NonNegativeFloat = Field(
|
|
default=DEFAULT_GRACEFUL_SHUTDOWN_WAIT_LOOP_S,
|
|
update_type=DeploymentOptionUpdateType.NeedsActorReconfigure,
|
|
)
|
|
|
|
health_check_period_s: PositiveFloat = Field(
|
|
default=DEFAULT_HEALTH_CHECK_PERIOD_S,
|
|
update_type=DeploymentOptionUpdateType.NeedsReconfigure,
|
|
)
|
|
health_check_timeout_s: PositiveFloat = Field(
|
|
default=DEFAULT_HEALTH_CHECK_TIMEOUT_S,
|
|
update_type=DeploymentOptionUpdateType.NeedsReconfigure,
|
|
)
|
|
|
|
autoscaling_config: Optional[AutoscalingConfig] = Field(
|
|
default=None, update_type=DeploymentOptionUpdateType.NeedsActorReconfigure
|
|
)
|
|
|
|
request_router_config: RequestRouterConfig = Field(
|
|
default_factory=RequestRouterConfig,
|
|
update_type=DeploymentOptionUpdateType.NeedsActorReconfigure,
|
|
)
|
|
|
|
# This flag is used to let replica know they are deployed from
|
|
# a different language.
|
|
is_cross_language: bool = False
|
|
|
|
# This flag is used to let controller know which language does
|
|
# the deployment use.
|
|
deployment_language: Any = DeploymentLanguage.PYTHON
|
|
|
|
version: Optional[str] = Field(
|
|
default=None,
|
|
update_type=DeploymentOptionUpdateType.HeavyWeight,
|
|
)
|
|
|
|
logging_config: Optional[dict] = Field(
|
|
default=None,
|
|
update_type=DeploymentOptionUpdateType.NeedsActorReconfigure,
|
|
)
|
|
|
|
max_constructor_retry_count: PositiveInt = Field(
|
|
default=DEFAULT_CONSTRUCTOR_RETRY_COUNT,
|
|
update_type=DeploymentOptionUpdateType.NeedsReconfigure,
|
|
)
|
|
gang_scheduling_config: Optional[GangSchedulingConfig] = Field(
|
|
default=None,
|
|
update_type=DeploymentOptionUpdateType.HeavyWeight,
|
|
)
|
|
|
|
deployment_actors: Optional[List[DeploymentActorConfig]] = Field(
|
|
default=None,
|
|
update_type=DeploymentOptionUpdateType.HeavyWeight,
|
|
)
|
|
|
|
rolling_update_percentage: float = Field(
|
|
default=DEFAULT_ROLLING_UPDATE_PERCENTAGE,
|
|
gt=0.0,
|
|
le=1.0,
|
|
update_type=DeploymentOptionUpdateType.LightWeight,
|
|
)
|
|
|
|
# Contains the names of deployment options manually set by the user
|
|
user_configured_option_names: Set[str] = set()
|
|
|
|
model_config = ConfigDict(validate_assignment=True, arbitrary_types_allowed=True)
|
|
|
|
@field_validator("user_config")
|
|
@classmethod
|
|
def user_config_json_serializable(cls, v):
|
|
if isinstance(v, bytes):
|
|
return v
|
|
if v is not None:
|
|
try:
|
|
json.dumps(v)
|
|
except TypeError as e:
|
|
raise ValueError(f"user_config is not JSON-serializable: {str(e)}.")
|
|
|
|
return v
|
|
|
|
@field_validator("logging_config")
|
|
@classmethod
|
|
def logging_config_valid(cls, v):
|
|
if v is None:
|
|
return v
|
|
if not isinstance(v, dict):
|
|
raise TypeError(
|
|
f"Got invalid type '{type(v)}' for logging_config. "
|
|
"Expected a dictionary."
|
|
)
|
|
# Handle default value
|
|
from ray.serve.schema import LoggingConfig
|
|
|
|
v = LoggingConfig(**v).model_dump()
|
|
return v
|
|
|
|
@field_validator("max_queued_requests")
|
|
@classmethod
|
|
def validate_max_queued_requests(cls, v):
|
|
if not isinstance(v, int):
|
|
raise TypeError("max_queued_requests must be an integer.")
|
|
|
|
if v < 1 and v != -1:
|
|
raise ValueError(
|
|
"max_queued_requests must be -1 (no limit) or a positive integer."
|
|
)
|
|
|
|
return v
|
|
|
|
@model_validator(mode="after")
|
|
def validate_gang_scheduling_config(self):
|
|
if self.gang_scheduling_config is None:
|
|
return self
|
|
if (
|
|
self.autoscaling_config is not None
|
|
and self.autoscaling_config.min_replicas == 0
|
|
):
|
|
raise ValueError(
|
|
"Scale to zero isn't supported for gang-scheduled deployments."
|
|
)
|
|
# Skip the num_replicas alignment check when autoscaling is enabled
|
|
if (
|
|
self.autoscaling_config is None
|
|
and self.num_replicas is not None
|
|
and self.num_replicas % self.gang_scheduling_config.gang_size != 0
|
|
):
|
|
raise ValueError(
|
|
f"num_replicas ({self.num_replicas}) must be a multiple of "
|
|
f"gang_size ({self.gang_scheduling_config.gang_size})."
|
|
)
|
|
|
|
return self
|
|
|
|
@model_validator(mode="after")
|
|
def validate_deployment_actors_unique_names(self):
|
|
if self.deployment_actors is None:
|
|
return self
|
|
seen = set()
|
|
duplicates = set()
|
|
for cfg in self.deployment_actors:
|
|
if cfg.name in seen:
|
|
duplicates.add(cfg.name)
|
|
seen.add(cfg.name)
|
|
if duplicates:
|
|
raise ValueError(
|
|
f"deployment_actors must have unique names. "
|
|
f"Duplicate name(s): {sorted(duplicates)}"
|
|
)
|
|
return self
|
|
|
|
def needs_pickle(self):
|
|
return _needs_pickle(self.deployment_language, self.is_cross_language)
|
|
|
|
def to_proto(self):
|
|
data = self.model_dump()
|
|
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"):
|
|
# By setting the serialized policy def, on the protobuf level, AutoscalingConfig constructor will not
|
|
# try to import the policy from the string import path when the protobuf is deserialized on the controller side
|
|
data["autoscaling_config"]["policy"][
|
|
"_serialized_policy_def"
|
|
] = self.autoscaling_config.policy._serialized_policy_def
|
|
# Serialize policy_kwargs dict to bytes for the proto
|
|
policy_kwargs = data["autoscaling_config"]["policy"].get("policy_kwargs")
|
|
if policy_kwargs is not None:
|
|
if not policy_kwargs:
|
|
data["autoscaling_config"]["policy"]["policy_kwargs"] = b""
|
|
else:
|
|
data["autoscaling_config"]["policy"][
|
|
"policy_kwargs"
|
|
] = cloudpickle.dumps(policy_kwargs)
|
|
data["autoscaling_config"] = AutoscalingConfigProto(
|
|
**data["autoscaling_config"]
|
|
)
|
|
if data.get("request_router_config"):
|
|
router_kwargs = data["request_router_config"].get("request_router_kwargs")
|
|
if router_kwargs is not None:
|
|
if not router_kwargs:
|
|
data["request_router_config"]["request_router_kwargs"] = b""
|
|
elif self.needs_pickle():
|
|
# Protobuf requires bytes, so we need to pickle
|
|
data["request_router_config"][
|
|
"request_router_kwargs"
|
|
] = cloudpickle.dumps(router_kwargs)
|
|
else:
|
|
raise ValueError(
|
|
"Non-empty request_router_kwargs not supported"
|
|
f"for cross-language deployments. Got: {router_kwargs}"
|
|
)
|
|
# By setting the serialized request router cls, on the protobuf level, RequestRouterConfig constructor will not
|
|
# try to import the request router cls from the string import path when the protobuf is deserialized on the controller side
|
|
data["request_router_config"][
|
|
"_serialized_request_router_cls"
|
|
] = self.request_router_config._serialized_request_router_cls
|
|
data["request_router_config"] = RequestRouterConfigProto(
|
|
**data["request_router_config"]
|
|
)
|
|
if data.get("logging_config"):
|
|
if "encoding" in data["logging_config"]:
|
|
data["logging_config"]["encoding"] = EncodingTypeProto.Value(
|
|
data["logging_config"]["encoding"]
|
|
)
|
|
data["logging_config"] = LoggingConfigProto(**data["logging_config"])
|
|
data["user_configured_option_names"] = list(
|
|
data["user_configured_option_names"]
|
|
)
|
|
if data.get("gang_scheduling_config"):
|
|
gang_config = data["gang_scheduling_config"]
|
|
placement_strategy = GangPlacementStrategyProto.Value(
|
|
gang_config["gang_placement_strategy"]
|
|
)
|
|
failure_policy = GangRuntimeFailurePolicyProto.Value(
|
|
gang_config["runtime_failure_policy"]
|
|
)
|
|
data["gang_scheduling_config"] = GangSchedulingConfigProto(
|
|
gang_size=gang_config["gang_size"],
|
|
gang_placement_strategy=placement_strategy,
|
|
runtime_failure_policy=failure_policy,
|
|
)
|
|
if self.deployment_actors:
|
|
deployment_actors_proto = []
|
|
for cfg in self.deployment_actors:
|
|
if not cfg._serialized_actor_class:
|
|
cfg._serialize_actor_class()
|
|
deployment_actors_proto.append(
|
|
DeploymentActorConfigProto(
|
|
name=cfg.name,
|
|
actor_class_name=cfg.actor_class,
|
|
_serialized_actor_class=cfg._serialized_actor_class,
|
|
serialized_init_args=cloudpickle.dumps(cfg.init_args or ()),
|
|
serialized_init_kwargs=cloudpickle.dumps(cfg.init_kwargs or {}),
|
|
serialized_actor_options=cloudpickle.dumps(
|
|
cfg.actor_options or {}
|
|
),
|
|
)
|
|
)
|
|
data["deployment_actors"] = deployment_actors_proto
|
|
else:
|
|
data.pop("deployment_actors", None)
|
|
return DeploymentConfigProto(**data)
|
|
|
|
def to_proto_bytes(self):
|
|
return self.to_proto().SerializeToString()
|
|
|
|
def to_dict(self):
|
|
# only use for logging purposes
|
|
return self.model_dump()
|
|
|
|
@classmethod
|
|
def from_proto(cls, proto: DeploymentConfigProto):
|
|
data = _proto_to_dict(proto)
|
|
deployment_language = (
|
|
data["deployment_language"]
|
|
if "deployment_language" in data
|
|
else DeploymentLanguage.PYTHON
|
|
)
|
|
is_cross_language = (
|
|
data["is_cross_language"] if "is_cross_language" in data else False
|
|
)
|
|
needs_pickle = _needs_pickle(deployment_language, is_cross_language)
|
|
if "user_config" in data:
|
|
if data["user_config"] != b"":
|
|
if needs_pickle:
|
|
data["user_config"] = cloudpickle.loads(proto.user_config)
|
|
else:
|
|
data["user_config"] = proto.user_config
|
|
else:
|
|
data["user_config"] = None
|
|
if "request_router_config" in data:
|
|
if "request_router_kwargs" in data["request_router_config"]:
|
|
request_router_kwargs = data["request_router_config"][
|
|
"request_router_kwargs"
|
|
]
|
|
if request_router_kwargs != b"":
|
|
if needs_pickle:
|
|
data["request_router_config"][
|
|
"request_router_kwargs"
|
|
] = cloudpickle.loads(
|
|
proto.request_router_config.request_router_kwargs
|
|
)
|
|
else:
|
|
data["request_router_config"][
|
|
"request_router_kwargs"
|
|
] = proto.request_router_config.request_router_kwargs
|
|
else:
|
|
data["request_router_config"]["request_router_kwargs"] = {}
|
|
|
|
# Remove falsy proto defaults so Pydantic uses its Field defaults.
|
|
# This is important during rolling upgrades when older controllers
|
|
# send configs without these fields (proto3 defaults to 0.0).
|
|
if not data["request_router_config"].get("initial_backoff_s"):
|
|
data["request_router_config"].pop("initial_backoff_s", None)
|
|
if not data["request_router_config"].get("backoff_multiplier"):
|
|
data["request_router_config"].pop("backoff_multiplier", None)
|
|
if not data["request_router_config"].get("max_backoff_s"):
|
|
data["request_router_config"].pop("max_backoff_s", None)
|
|
|
|
data["request_router_config"] = RequestRouterConfig(
|
|
**data["request_router_config"]
|
|
)
|
|
if "autoscaling_config" in data:
|
|
if not data["autoscaling_config"].get("upscale_smoothing_factor"):
|
|
data["autoscaling_config"]["upscale_smoothing_factor"] = None
|
|
if not data["autoscaling_config"].get("downscale_smoothing_factor"):
|
|
data["autoscaling_config"]["downscale_smoothing_factor"] = None
|
|
if not data["autoscaling_config"].get("upscaling_factor"):
|
|
data["autoscaling_config"]["upscaling_factor"] = None
|
|
if not data["autoscaling_config"].get("downscaling_factor"):
|
|
data["autoscaling_config"]["downscaling_factor"] = None
|
|
if not data["autoscaling_config"].get("target_ongoing_requests"):
|
|
data["autoscaling_config"]["target_ongoing_requests"] = None
|
|
if not data["autoscaling_config"].get("aggregation_function"):
|
|
data["autoscaling_config"][
|
|
"aggregation_function"
|
|
] = AggregationFunction.MEAN
|
|
# Deserialize policy_kwargs bytes back to a dict
|
|
if "policy" in data["autoscaling_config"]:
|
|
policy_data = data["autoscaling_config"]["policy"]
|
|
if "policy_kwargs" in policy_data:
|
|
raw = policy_data["policy_kwargs"]
|
|
if raw and raw != b"":
|
|
policy_data["policy_kwargs"] = cloudpickle.loads(
|
|
proto.autoscaling_config.policy.policy_kwargs
|
|
)
|
|
else:
|
|
policy_data["policy_kwargs"] = {}
|
|
data["autoscaling_config"] = AutoscalingConfig(**data["autoscaling_config"])
|
|
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"]
|
|
)
|
|
if "logging_config" in data:
|
|
if "encoding" in data["logging_config"]:
|
|
data["logging_config"]["encoding"] = EncodingTypeProto.Name(
|
|
data["logging_config"]["encoding"]
|
|
)
|
|
if "gang_scheduling_config" in data and data["gang_scheduling_config"]:
|
|
gang_config = data["gang_scheduling_config"]
|
|
gang_config["gang_placement_strategy"] = GangPlacementStrategy(
|
|
GangPlacementStrategyProto.Name(gang_config["gang_placement_strategy"])
|
|
)
|
|
gang_config["runtime_failure_policy"] = GangRuntimeFailurePolicy(
|
|
GangRuntimeFailurePolicyProto.Name(
|
|
gang_config["runtime_failure_policy"]
|
|
)
|
|
)
|
|
data["gang_scheduling_config"] = GangSchedulingConfig(**gang_config)
|
|
else:
|
|
data.pop("gang_scheduling_config", None)
|
|
if "deployment_actors" in data and data["deployment_actors"]:
|
|
deployment_actors = []
|
|
|
|
def _loads(b):
|
|
return cloudpickle.loads(b) if b else None
|
|
|
|
for proto_dict in data["deployment_actors"]:
|
|
serialized_cls = proto_dict.get("_serialized_actor_class")
|
|
serialized_args = proto_dict.get("serialized_init_args")
|
|
serialized_kwargs = proto_dict.get("serialized_init_kwargs")
|
|
serialized_opts = proto_dict.get("serialized_actor_options")
|
|
actor_class_name = proto_dict.get("actor_class_name", "")
|
|
deployment_actors.append(
|
|
DeploymentActorConfig(
|
|
name=proto_dict.get("name"),
|
|
actor_class=actor_class_name,
|
|
_serialized_actor_class=serialized_cls,
|
|
init_args=_loads(serialized_args) or (),
|
|
init_kwargs=_loads(serialized_kwargs) or {},
|
|
actor_options=_loads(serialized_opts) or {},
|
|
)
|
|
)
|
|
data["deployment_actors"] = deployment_actors
|
|
else:
|
|
data.pop("deployment_actors", None)
|
|
|
|
return cls(**data)
|
|
|
|
@classmethod
|
|
def from_proto_bytes(cls, proto_bytes: bytes):
|
|
proto = DeploymentConfigProto.FromString(proto_bytes)
|
|
return cls.from_proto(proto)
|
|
|
|
@classmethod
|
|
def from_default(cls, **kwargs: Any) -> "DeploymentConfig":
|
|
"""Creates a default DeploymentConfig and overrides it with kwargs.
|
|
|
|
Ignores any kwargs set to DEFAULT.VALUE.
|
|
|
|
Args:
|
|
**kwargs: Field overrides for ``DeploymentConfig``. Keys must match
|
|
the class's field names; values equal to ``DEFAULT.VALUE`` are
|
|
skipped (the default is kept).
|
|
|
|
Returns:
|
|
A ``DeploymentConfig`` initialized from defaults and updated with
|
|
the supplied (non-``DEFAULT.VALUE``) kwargs.
|
|
|
|
Raises:
|
|
TypeError: when a keyword that's not an argument to the class is
|
|
passed in.
|
|
"""
|
|
|
|
config = cls()
|
|
valid_config_options = set(cls.model_fields.keys())
|
|
|
|
# Friendly error if a non-DeploymentConfig kwarg was passed in
|
|
for key, val in kwargs.items():
|
|
if key not in valid_config_options:
|
|
raise TypeError(
|
|
f'Got invalid Deployment config option "{key}" '
|
|
f"(with value {val}) as keyword argument. All Deployment "
|
|
"config options must come from this list: "
|
|
f"{list(valid_config_options)}."
|
|
)
|
|
|
|
kwargs = {key: val for key, val in kwargs.items() if val != DEFAULT.VALUE}
|
|
|
|
for key, val in kwargs.items():
|
|
config.__setattr__(key, val)
|
|
|
|
return config
|
|
|
|
|
|
def handle_num_replicas_auto(
|
|
max_ongoing_requests: Union[int, DEFAULT],
|
|
autoscaling_config: Optional[Union[Dict, AutoscalingConfig, DEFAULT]],
|
|
):
|
|
"""Return modified `max_ongoing_requests` and `autoscaling_config`
|
|
for when num_replicas="auto".
|
|
|
|
If `autoscaling_config` is unspecified, returns the modified value
|
|
AutoscalingConfig.default().
|
|
If it is specified, the specified fields in `autoscaling_config`
|
|
override that of AutoscalingConfig.default().
|
|
"""
|
|
|
|
if autoscaling_config in [DEFAULT.VALUE, None]:
|
|
# If autoscaling config wasn't specified, use default
|
|
# configuration
|
|
autoscaling_config = AutoscalingConfig.default()
|
|
else:
|
|
# If autoscaling config was specified, values specified in
|
|
# autoscaling config overrides the default configuration
|
|
default_config = AutoscalingConfig.default().model_dump(exclude_unset=True)
|
|
autoscaling_config = (
|
|
autoscaling_config
|
|
if isinstance(autoscaling_config, dict)
|
|
else autoscaling_config.model_dump(exclude_unset=True)
|
|
)
|
|
default_config.update(autoscaling_config)
|
|
autoscaling_config = AutoscalingConfig(**default_config)
|
|
|
|
return max_ongoing_requests, autoscaling_config
|
|
|
|
|
|
class ReplicaConfig:
|
|
"""Internal datastructure wrapping config options for a deployment's replicas.
|
|
|
|
Provides five main properties (see property docstrings for more info):
|
|
deployment_def: the code, or a reference to the code, that this
|
|
replica should run.
|
|
init_args: the deployment_def's init_args.
|
|
init_kwargs: the deployment_def's init_kwargs.
|
|
ray_actor_options: the Ray actor options to pass into the replica's
|
|
actor.
|
|
resource_dict: contains info on this replica's actor's resource needs.
|
|
|
|
Offers a serialized equivalent (e.g. serialized_deployment_def) for
|
|
deployment_def, init_args, and init_kwargs. Deserializes these properties
|
|
when they're first accessed, if they were not passed in directly through
|
|
create().
|
|
|
|
Use the classmethod create() to make a ReplicaConfig with the deserialized
|
|
properties.
|
|
|
|
Note: overwriting or setting any property after the ReplicaConfig has been
|
|
constructed is currently undefined behavior. The config's fields should not
|
|
be modified externally after it is created.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
deployment_def_name: str,
|
|
serialized_deployment_def: bytes,
|
|
serialized_init_args: bytes,
|
|
serialized_init_kwargs: bytes,
|
|
ray_actor_options: Dict,
|
|
placement_group_bundles: Optional[List[Dict[str, float]]] = None,
|
|
placement_group_strategy: Optional[str] = None,
|
|
placement_group_bundle_label_selector: Optional[List[Dict[str, str]]] = None,
|
|
placement_group_fallback_strategy: Optional[List[Dict[str, Any]]] = None,
|
|
max_replicas_per_node: Optional[int] = None,
|
|
needs_pickle: bool = True,
|
|
):
|
|
"""Construct a ReplicaConfig with serialized properties.
|
|
|
|
All parameters are required. See classmethod create() for defaults.
|
|
"""
|
|
self.deployment_def_name = deployment_def_name
|
|
|
|
# Store serialized versions of code properties.
|
|
self.serialized_deployment_def = serialized_deployment_def
|
|
self.serialized_init_args = serialized_init_args
|
|
self.serialized_init_kwargs = serialized_init_kwargs
|
|
|
|
# Deserialize properties when first accessed. See @property methods.
|
|
self._deployment_def = None
|
|
self._init_args = None
|
|
self._init_kwargs = None
|
|
|
|
# Configure ray_actor_options. These are the Ray options ultimately
|
|
# passed into the replica's actor when it's created.
|
|
self.ray_actor_options = ray_actor_options
|
|
|
|
self.placement_group_bundles = placement_group_bundles
|
|
self.placement_group_strategy = placement_group_strategy
|
|
self.placement_group_bundle_label_selector = (
|
|
placement_group_bundle_label_selector
|
|
)
|
|
self.placement_group_fallback_strategy = placement_group_fallback_strategy
|
|
|
|
self.max_replicas_per_node = max_replicas_per_node
|
|
|
|
self._normalize_bundle_label_selector()
|
|
self._validate()
|
|
|
|
# Create resource_dict. This contains info about the replica's resource
|
|
# needs. It does NOT set the replica's resource usage. That's done by
|
|
# the ray_actor_options.
|
|
self.resource_dict = resources_from_ray_options(self.ray_actor_options)
|
|
self.needs_pickle = needs_pickle
|
|
|
|
def _normalize_bundle_label_selector(self):
|
|
"""If a single selector is provided for multiple bundles, it is broadcasted
|
|
uniformly to all bundles.
|
|
"""
|
|
if (
|
|
self.placement_group_bundles
|
|
and self.placement_group_bundle_label_selector
|
|
and len(self.placement_group_bundle_label_selector) == 1
|
|
and len(self.placement_group_bundles) > 1
|
|
):
|
|
single_selector = self.placement_group_bundle_label_selector[0]
|
|
self.placement_group_bundle_label_selector = [
|
|
single_selector.copy() for _ in range(len(self.placement_group_bundles))
|
|
]
|
|
|
|
def _validate(self):
|
|
self._validate_ray_actor_options()
|
|
self._validate_placement_group_options()
|
|
self._validate_max_replicas_per_node()
|
|
|
|
if (
|
|
self.max_replicas_per_node is not None
|
|
and self.placement_group_bundles is not None
|
|
):
|
|
raise ValueError(
|
|
"Setting max_replicas_per_node is not allowed when "
|
|
"placement_group_bundles is provided."
|
|
)
|
|
|
|
def update(
|
|
self,
|
|
ray_actor_options: dict,
|
|
placement_group_bundles: Optional[List[Dict[str, float]]] = None,
|
|
placement_group_strategy: Optional[str] = None,
|
|
placement_group_bundle_label_selector: Optional[List[Dict[str, str]]] = None,
|
|
placement_group_fallback_strategy: Optional[List[Dict[str, Any]]] = None,
|
|
max_replicas_per_node: Optional[int] = None,
|
|
):
|
|
self.ray_actor_options = ray_actor_options
|
|
|
|
self.placement_group_bundles = placement_group_bundles
|
|
self.placement_group_strategy = placement_group_strategy
|
|
self.placement_group_bundle_label_selector = (
|
|
placement_group_bundle_label_selector
|
|
)
|
|
self.placement_group_fallback_strategy = placement_group_fallback_strategy
|
|
|
|
self.max_replicas_per_node = max_replicas_per_node
|
|
|
|
self._normalize_bundle_label_selector()
|
|
self._validate()
|
|
|
|
self.resource_dict = resources_from_ray_options(self.ray_actor_options)
|
|
|
|
@classmethod
|
|
def create(
|
|
cls,
|
|
deployment_def: Union[Callable, str],
|
|
init_args: Optional[Tuple[Any]] = None,
|
|
init_kwargs: Optional[Dict[Any, Any]] = None,
|
|
ray_actor_options: Optional[Dict] = None,
|
|
placement_group_bundles: Optional[List[Dict[str, float]]] = None,
|
|
placement_group_strategy: Optional[str] = None,
|
|
placement_group_bundle_label_selector: Optional[List[Dict[str, str]]] = None,
|
|
placement_group_fallback_strategy: Optional[List[Dict[str, Any]]] = None,
|
|
max_replicas_per_node: Optional[int] = None,
|
|
deployment_def_name: Optional[str] = None,
|
|
):
|
|
"""Create a ReplicaConfig from deserialized parameters."""
|
|
|
|
if not callable(deployment_def) and not isinstance(deployment_def, str):
|
|
raise TypeError("@serve.deployment must be called on a class or function.")
|
|
|
|
if not (init_args is None or isinstance(init_args, (tuple, list))):
|
|
raise TypeError("init_args must be a tuple.")
|
|
|
|
if not (init_kwargs is None or isinstance(init_kwargs, dict)):
|
|
raise TypeError("init_kwargs must be a dict.")
|
|
|
|
if inspect.isfunction(deployment_def):
|
|
if init_args:
|
|
raise ValueError("init_args not supported for function deployments.")
|
|
elif init_kwargs:
|
|
raise ValueError("init_kwargs not supported for function deployments.")
|
|
|
|
if not isinstance(deployment_def, (Callable, str)):
|
|
raise TypeError(
|
|
f'Got invalid type "{type(deployment_def)}" for '
|
|
"deployment_def. Expected deployment_def to be a "
|
|
"class, function, or string."
|
|
)
|
|
# Set defaults
|
|
if init_args is None:
|
|
init_args = ()
|
|
if init_kwargs is None:
|
|
init_kwargs = {}
|
|
if ray_actor_options is None:
|
|
ray_actor_options = {}
|
|
if deployment_def_name is None:
|
|
if isinstance(deployment_def, str):
|
|
deployment_def_name = deployment_def
|
|
else:
|
|
deployment_def_name = deployment_def.__name__
|
|
|
|
config = cls(
|
|
deployment_def_name=deployment_def_name,
|
|
serialized_deployment_def=pickle_dumps(
|
|
deployment_def,
|
|
f"Could not serialize the deployment {repr(deployment_def)}",
|
|
),
|
|
serialized_init_args=pickle_dumps(
|
|
init_args, "Could not serialize the deployment init args"
|
|
),
|
|
serialized_init_kwargs=pickle_dumps(
|
|
init_kwargs, "Could not serialize the deployment 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,
|
|
)
|
|
|
|
config._deployment_def = deployment_def
|
|
config._init_args = init_args
|
|
config._init_kwargs = init_kwargs
|
|
|
|
return config
|
|
|
|
def _validate_ray_actor_options(self):
|
|
if not isinstance(self.ray_actor_options, dict):
|
|
raise TypeError(
|
|
f'Got invalid type "{type(self.ray_actor_options)}" for '
|
|
"ray_actor_options. Expected a dictionary."
|
|
)
|
|
# Please keep this in sync with the docstring for the ray_actor_options
|
|
# kwarg in api.py.
|
|
allowed_ray_actor_options = {
|
|
# Resource options
|
|
"accelerator_type",
|
|
"memory",
|
|
"num_cpus",
|
|
"num_gpus",
|
|
"resources",
|
|
# Other options
|
|
"runtime_env",
|
|
"label_selector",
|
|
"fallback_strategy",
|
|
}
|
|
|
|
for option in self.ray_actor_options:
|
|
if option not in allowed_ray_actor_options:
|
|
raise ValueError(
|
|
f"Specifying '{option}' in ray_actor_options is not allowed. "
|
|
f"Allowed options: {allowed_ray_actor_options}"
|
|
)
|
|
ray_option_utils.validate_actor_options(self.ray_actor_options, in_options=True)
|
|
|
|
# Set Serve replica defaults
|
|
if self.ray_actor_options.get("num_cpus") is None:
|
|
self.ray_actor_options["num_cpus"] = 1
|
|
|
|
def _validate_max_replicas_per_node(self) -> None:
|
|
if self.max_replicas_per_node is None:
|
|
return
|
|
if not isinstance(self.max_replicas_per_node, int):
|
|
raise TypeError(
|
|
f"Get invalid type '{type(self.max_replicas_per_node)}' for "
|
|
"max_replicas_per_node. Expected None or an integer "
|
|
f"in the range of [1, {MAX_REPLICAS_PER_NODE_MAX_VALUE}]."
|
|
)
|
|
if (
|
|
self.max_replicas_per_node < 1
|
|
or self.max_replicas_per_node > MAX_REPLICAS_PER_NODE_MAX_VALUE
|
|
):
|
|
raise ValueError(
|
|
f"Invalid max_replicas_per_node {self.max_replicas_per_node}. "
|
|
"Valid values are None or an integer "
|
|
f"in the range of [1, {MAX_REPLICAS_PER_NODE_MAX_VALUE}]."
|
|
)
|
|
|
|
def _validate_placement_group_options(self) -> None:
|
|
if self.placement_group_strategy is not None:
|
|
if self.placement_group_bundles is None:
|
|
raise ValueError(
|
|
"If `placement_group_strategy` is provided, "
|
|
"`placement_group_bundles` must also be provided."
|
|
)
|
|
|
|
if self.placement_group_fallback_strategy is not None:
|
|
if self.placement_group_bundles is None:
|
|
raise ValueError(
|
|
"If `placement_group_fallback_strategy` is provided, "
|
|
"`placement_group_bundles` must also be provided."
|
|
)
|
|
if not isinstance(self.placement_group_fallback_strategy, list):
|
|
raise TypeError(
|
|
"placement_group_fallback_strategy must be a list of dictionaries. "
|
|
f"Got: {type(self.placement_group_fallback_strategy)}."
|
|
)
|
|
for i, strategy in enumerate(self.placement_group_fallback_strategy):
|
|
if not isinstance(strategy, dict):
|
|
raise TypeError(
|
|
f"placement_group_fallback_strategy entry at index {i} must be a dictionary. "
|
|
f"Got: {type(strategy)}."
|
|
)
|
|
|
|
if self.placement_group_bundle_label_selector is not None:
|
|
if self.placement_group_bundles is None:
|
|
raise ValueError(
|
|
"If `placement_group_bundle_label_selector` is provided, "
|
|
"`placement_group_bundles` must also be provided."
|
|
)
|
|
|
|
if self.placement_group_bundles is not None:
|
|
validate_placement_group(
|
|
bundles=self.placement_group_bundles,
|
|
strategy=self.placement_group_strategy or "PACK",
|
|
lifetime="detached",
|
|
bundle_label_selector=self.placement_group_bundle_label_selector,
|
|
)
|
|
|
|
resource_error_prefix = (
|
|
"When using `placement_group_bundles`, the replica actor "
|
|
"will be placed in the first bundle, so the resource "
|
|
"requirements for the actor must be a subset of the first "
|
|
"bundle."
|
|
)
|
|
|
|
first_bundle = self.placement_group_bundles[0]
|
|
|
|
# Validate that the replica actor fits in the first bundle.
|
|
# Downstream code depends on this validation. The scheduler pins the
|
|
# actor to bundle 0 in deployment_scheduler._schedule_replica, and
|
|
# DeploymentSchedulingInfo.required_resources reads bundle 0 as the
|
|
# replica's demand.
|
|
bundle_cpu = first_bundle.get("CPU", 0)
|
|
replica_actor_num_cpus = self.ray_actor_options.get("num_cpus", 0)
|
|
if bundle_cpu < replica_actor_num_cpus:
|
|
raise ValueError(
|
|
f"{resource_error_prefix} `num_cpus` for the actor is "
|
|
f"{replica_actor_num_cpus}, but the bundle only has "
|
|
f"{bundle_cpu} `CPU` specified."
|
|
)
|
|
|
|
bundle_gpu = first_bundle.get("GPU", 0)
|
|
replica_actor_num_gpus = self.ray_actor_options.get("num_gpus", 0)
|
|
if bundle_gpu < replica_actor_num_gpus:
|
|
raise ValueError(
|
|
f"{resource_error_prefix} `num_gpus` for the actor is "
|
|
f"{replica_actor_num_gpus}, but the bundle only has "
|
|
f"{bundle_gpu} `GPU` specified."
|
|
)
|
|
|
|
replica_actor_resources = self.ray_actor_options.get("resources", {})
|
|
for actor_resource, actor_value in replica_actor_resources.items():
|
|
bundle_value = first_bundle.get(actor_resource, 0)
|
|
if bundle_value < actor_value:
|
|
raise ValueError(
|
|
f"{resource_error_prefix} `{actor_resource}` requirement "
|
|
f"for the actor is {actor_value}, but the bundle only "
|
|
f"has {bundle_value} `{actor_resource}` specified."
|
|
)
|
|
|
|
@property
|
|
def deployment_def(self) -> Union[Callable, str]:
|
|
"""The code, or a reference to the code, that this replica runs.
|
|
|
|
For Python replicas, this can be one of the following:
|
|
- Function (Callable)
|
|
- Class (Callable)
|
|
- Import path (str)
|
|
|
|
For Java replicas, this can be one of the following:
|
|
- Class path (str)
|
|
"""
|
|
if self._deployment_def is None:
|
|
if self.needs_pickle:
|
|
self._deployment_def = cloudpickle.loads(self.serialized_deployment_def)
|
|
else:
|
|
self._deployment_def = self.serialized_deployment_def.decode(
|
|
encoding="utf-8"
|
|
)
|
|
|
|
return self._deployment_def
|
|
|
|
@property
|
|
def init_args(self) -> Optional[Union[Tuple[Any], bytes]]:
|
|
"""The init_args for a Python class.
|
|
|
|
This property is only meaningful if deployment_def is a Python class.
|
|
Otherwise, it is None.
|
|
"""
|
|
if self._init_args is None:
|
|
if self.needs_pickle:
|
|
self._init_args = cloudpickle.loads(self.serialized_init_args)
|
|
else:
|
|
self._init_args = self.serialized_init_args
|
|
|
|
return self._init_args
|
|
|
|
@property
|
|
def init_kwargs(self) -> Optional[Tuple[Any]]:
|
|
"""The init_kwargs for a Python class.
|
|
|
|
This property is only meaningful if deployment_def is a Python class.
|
|
Otherwise, it is None.
|
|
"""
|
|
|
|
if self._init_kwargs is None:
|
|
self._init_kwargs = cloudpickle.loads(self.serialized_init_kwargs)
|
|
|
|
return self._init_kwargs
|
|
|
|
@classmethod
|
|
def from_proto(cls, proto: ReplicaConfigProto, needs_pickle: bool = True):
|
|
return ReplicaConfig(
|
|
deployment_def_name=proto.deployment_def_name,
|
|
serialized_deployment_def=proto.deployment_def,
|
|
serialized_init_args=(proto.init_args if proto.init_args != b"" else None),
|
|
serialized_init_kwargs=(
|
|
proto.init_kwargs if proto.init_kwargs != b"" else None
|
|
),
|
|
ray_actor_options=json.loads(proto.ray_actor_options),
|
|
placement_group_bundles=(
|
|
json.loads(proto.placement_group_bundles)
|
|
if proto.placement_group_bundles
|
|
else None
|
|
),
|
|
placement_group_strategy=(
|
|
proto.placement_group_strategy
|
|
if proto.placement_group_strategy != ""
|
|
else None
|
|
),
|
|
placement_group_bundle_label_selector=(
|
|
json.loads(proto.placement_group_bundle_label_selector)
|
|
if proto.placement_group_bundle_label_selector
|
|
else None
|
|
),
|
|
placement_group_fallback_strategy=(
|
|
json.loads(proto.placement_group_fallback_strategy)
|
|
if proto.placement_group_fallback_strategy
|
|
else None
|
|
),
|
|
max_replicas_per_node=(
|
|
proto.max_replicas_per_node if proto.max_replicas_per_node else None
|
|
),
|
|
needs_pickle=needs_pickle,
|
|
)
|
|
|
|
@classmethod
|
|
def from_proto_bytes(cls, proto_bytes: bytes, needs_pickle: bool = True):
|
|
proto = ReplicaConfigProto.FromString(proto_bytes)
|
|
return cls.from_proto(proto, needs_pickle)
|
|
|
|
def to_proto(self):
|
|
placement_group_bundles = (
|
|
json.dumps(self.placement_group_bundles)
|
|
if self.placement_group_bundles is not None
|
|
else ""
|
|
)
|
|
|
|
bundle_label_selector = (
|
|
json.dumps(self.placement_group_bundle_label_selector)
|
|
if self.placement_group_bundle_label_selector is not None
|
|
else ""
|
|
)
|
|
|
|
fallback_strategy = (
|
|
json.dumps(self.placement_group_fallback_strategy)
|
|
if self.placement_group_fallback_strategy is not None
|
|
else ""
|
|
)
|
|
|
|
max_replicas_per_node = (
|
|
self.max_replicas_per_node if self.max_replicas_per_node is not None else 0
|
|
)
|
|
|
|
return ReplicaConfigProto(
|
|
deployment_def_name=self.deployment_def_name,
|
|
deployment_def=self.serialized_deployment_def,
|
|
init_args=self.serialized_init_args,
|
|
init_kwargs=self.serialized_init_kwargs,
|
|
ray_actor_options=json.dumps(self.ray_actor_options),
|
|
placement_group_bundles=placement_group_bundles,
|
|
placement_group_strategy=self.placement_group_strategy,
|
|
placement_group_bundle_label_selector=bundle_label_selector,
|
|
placement_group_fallback_strategy=fallback_strategy,
|
|
max_replicas_per_node=max_replicas_per_node,
|
|
)
|
|
|
|
def to_proto_bytes(self):
|
|
return self.to_proto().SerializeToString()
|
|
|
|
def to_dict(self):
|
|
# only use for logging purposes
|
|
return {
|
|
"deployment_def_name": self.deployment_def_name,
|
|
"ray_actor_options": self.ray_actor_options,
|
|
"placement_group_bundles": self.placement_group_bundles,
|
|
"placement_group_strategy": self.placement_group_strategy,
|
|
"placement_group_bundle_label_selector": self.placement_group_bundle_label_selector,
|
|
"placement_group_fallback_strategy": self.placement_group_fallback_strategy,
|
|
"max_replicas_per_node": self.max_replicas_per_node,
|
|
}
|