forked from ray-project/ray
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
pull out shared deploy code into deploy utils (ray-project#34321)
Signed-off-by: Cindy Zhang <[email protected]>
- Loading branch information
Showing
6 changed files
with
172 additions
and
129 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,137 @@ | ||
| from typing import Dict, Tuple, Union, Callable, Type, Optional, Any | ||
| import logging | ||
| import time | ||
|
|
||
| from ray.serve.config import ReplicaConfig, DeploymentConfig | ||
| from ray.serve._private.constants import SERVE_LOGGER_NAME | ||
| from ray.serve._private.autoscaling_policy import BasicAutoscalingPolicy | ||
| from ray.serve._private.common import DeploymentInfo | ||
|
|
||
| import ray | ||
| import ray.util.serialization_addons | ||
|
|
||
| logger = logging.getLogger(SERVE_LOGGER_NAME) | ||
|
|
||
|
|
||
| def get_deploy_args( | ||
| name: str, | ||
| deployment_def: Union[Callable, Type[Callable], str], | ||
| init_args: Tuple[Any], | ||
| init_kwargs: Dict[Any, Any], | ||
| ray_actor_options: Optional[Dict] = None, | ||
| config: Optional[Union[DeploymentConfig, Dict[str, Any]]] = None, | ||
| version: Optional[str] = None, | ||
| route_prefix: Optional[str] = None, | ||
| is_driver_deployment: Optional[str] = None, | ||
| docs_path: Optional[str] = None, | ||
| ) -> Dict: | ||
| """ | ||
| Takes a deployment's configuration, and returns the arguments needed | ||
| for the controller to deploy it. | ||
| """ | ||
|
|
||
| if config is None: | ||
| config = {} | ||
| if ray_actor_options is None: | ||
| ray_actor_options = {} | ||
|
|
||
| curr_job_env = ray.get_runtime_context().runtime_env | ||
| if "runtime_env" in ray_actor_options: | ||
| # It is illegal to set field working_dir to None. | ||
| if curr_job_env.get("working_dir") is not None: | ||
| ray_actor_options["runtime_env"].setdefault( | ||
| "working_dir", curr_job_env.get("working_dir") | ||
| ) | ||
| else: | ||
| ray_actor_options["runtime_env"] = curr_job_env | ||
|
|
||
| replica_config = ReplicaConfig.create( | ||
| deployment_def, | ||
| init_args=init_args, | ||
| init_kwargs=init_kwargs, | ||
| ray_actor_options=ray_actor_options, | ||
| ) | ||
|
|
||
| if isinstance(config, dict): | ||
| deployment_config = DeploymentConfig.parse_obj(config) | ||
| elif isinstance(config, DeploymentConfig): | ||
| deployment_config = config | ||
| else: | ||
| raise TypeError("config must be a DeploymentConfig or a dictionary.") | ||
|
|
||
| deployment_config.version = version | ||
|
|
||
| if ( | ||
| deployment_config.autoscaling_config is not None | ||
| and deployment_config.max_concurrent_queries | ||
| < deployment_config.autoscaling_config.target_num_ongoing_requests_per_replica # noqa: E501 | ||
| ): | ||
| logger.warning( | ||
| "Autoscaling will never happen, " | ||
| "because 'max_concurrent_queries' is less than " | ||
| "'target_num_ongoing_requests_per_replica' now." | ||
| ) | ||
|
|
||
| controller_deploy_args = { | ||
| "name": name, | ||
| "deployment_config_proto_bytes": deployment_config.to_proto_bytes(), | ||
| "replica_config_proto_bytes": replica_config.to_proto_bytes(), | ||
| "route_prefix": route_prefix, | ||
| "deployer_job_id": ray.get_runtime_context().get_job_id(), | ||
| "is_driver_deployment": is_driver_deployment, | ||
| "docs_path": docs_path, | ||
| } | ||
|
|
||
| return controller_deploy_args | ||
|
|
||
|
|
||
| def deploy_args_to_deployment_info( | ||
| deployment_name: str, | ||
| deployment_config_proto_bytes: bytes, | ||
| replica_config_proto_bytes: bytes, | ||
| deployer_job_id: Union[str, bytes], | ||
| previous_deployment: DeploymentInfo, | ||
| is_driver_deployment: Optional[bool] = False, | ||
| ) -> DeploymentInfo: | ||
| """Takes deployment args passed to the controller after building an application and | ||
| constructs a DeploymentInfo object. | ||
| """ | ||
|
|
||
| deployment_config = DeploymentConfig.from_proto_bytes(deployment_config_proto_bytes) | ||
| version = deployment_config.version | ||
| replica_config = ReplicaConfig.from_proto_bytes( | ||
| replica_config_proto_bytes, deployment_config.needs_pickle() | ||
| ) | ||
|
|
||
| autoscaling_config = deployment_config.autoscaling_config | ||
| if autoscaling_config is not None: | ||
| if autoscaling_config.initial_replicas is not None: | ||
| deployment_config.num_replicas = autoscaling_config.initial_replicas | ||
| else: | ||
| if previous_deployment is None: | ||
| deployment_config.num_replicas = autoscaling_config.min_replicas | ||
| else: | ||
| deployment_config.num_replicas = ( | ||
| previous_deployment.deployment_config.num_replicas | ||
| ) | ||
|
|
||
| autoscaling_policy = BasicAutoscalingPolicy(autoscaling_config) | ||
| else: | ||
| autoscaling_policy = None | ||
|
|
||
| # Java API passes in JobID as bytes | ||
| if isinstance(deployer_job_id, bytes): | ||
| deployer_job_id = ray.JobID.from_int( | ||
| int.from_bytes(deployer_job_id, "little") | ||
| ).hex() | ||
|
|
||
| return DeploymentInfo( | ||
| actor_name=deployment_name, | ||
| version=version, | ||
| deployment_config=deployment_config, | ||
| replica_config=replica_config, | ||
| deployer_job_id=deployer_job_id, | ||
| start_time_ms=int(time.time() * 1000), | ||
| autoscaling_policy=autoscaling_policy, | ||
| is_driver_deployment=is_driver_deployment, | ||
| ) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.