[serve] Implement serve.run() and Application

dashboard/modules/serve/serve_head.py

@@ -1,16 +1,17 @@
 from aiohttp.web import Request, Response
+import json
 import logging
 
 import ray.dashboard.utils as dashboard_utils
 import ray.dashboard.optional_utils as optional_utils
 
 from ray import serve
-from ray.serve.application import Application
-from ray.serve.schema import (
-    serve_application_to_schema,
+from ray.serve.api import (
+    Application,
+    get_deployment_statuses,
+    internal_get_global_client,
     serve_application_status_to_schema,
 )
-from ray.serve.api import get_deployment_statuses
 
 logger = logging.getLogger(__name__)
 logger.setLevel(logging.INFO)
@@ -25,10 +26,9 @@ def __init__(self, dashboard_head):
     @routes.get("/api/serve/deployments/")
     @optional_utils.init_ray_and_catch_exceptions(connect_to_serve=True)
     async def get_all_deployments(self, req: Request) -> Response:
-        deployments = list(serve.list_deployments().values())
-        serve_application_schema = serve_application_to_schema(deployments=deployments)
+        app = Application(list(serve.list_deployments().values()))
         return Response(
-            text=serve_application_schema.json(),
+            text=json.dumps(app.to_dict()),
             content_type="application/json",
         )
 
@@ -53,17 +53,16 @@ async def delete_serve_application(self, req: Request) -> Response:
     @optional_utils.init_ray_and_catch_exceptions(connect_to_serve=True)
     async def put_all_deployments(self, req: Request) -> Response:
         app = Application.from_dict(await req.json())
-        app.deploy(blocking=False)
+        serve.run(app, _blocking=False)
 
         new_names = set()
-        for deployment in app:
+        for deployment in app.deployments.values():
             new_names.add(deployment.name)
 
         all_deployments = serve.list_deployments()
         all_names = set(all_deployments.keys())
         names_to_delete = all_names.difference(new_names)
-        for name in names_to_delete:
-            all_deployments[name].delete()
+        internal_get_global_client().delete_deployments(names_to_delete)
 
         return Response()
 

python/ray/serve/api.py

@@ -7,6 +7,7 @@
 import random
 import re
 import time
+import yaml
 import json
 from dataclasses import dataclass
 from functools import wraps
@@ -64,12 +65,20 @@
     format_actor_name,
     get_current_node_resource_key,
     get_random_letters,
+    get_deployment_import_path,
     logger,
     DEFAULT,
 )
 from ray.util.annotations import PublicAPI
 import ray
 from ray import cloudpickle
+from ray.serve.schema import (
+    RayActorOptionsSchema,
+    DeploymentSchema,
+    DeploymentStatusSchema,
+    ServeApplicationSchema,
+    ServeApplicationStatusSchema,
+)
 
 
 _INTERNAL_REPLICA_CONTEXT = None
@@ -1673,8 +1682,9 @@ def get_deployment_statuses() -> Dict[str, DeploymentStatusInfo]:
 
 
 class ImmutableDeploymentDict(dict):
-    def __init__(self, deployments: List[Deployment]):
-        raise NotImplementedError()
+    def __init__(self, deployments: Dict[str, Deployment]):
+        super().__init__()
+        self.update(deployments)
 
     def __setitem__(self, *args):
         """Not allowed. Modify deployment options using set_options instead."""
@@ -1702,16 +1712,41 @@ class Application:
     to production using the Serve CLI or REST API.
     """
 
-    def __init__(self, ingress: Deployment, deployments: List[Deployment]):
-        raise NotImplementedError()
+    def __init__(self, deployments: List[Deployment]):
+        deployment_dict = {}
+        for d in deployments:
+            if not isinstance(d, Deployment):
+                raise TypeError(f"Got {type(d)}. Expected deployment.")
+            elif d.name in deployment_dict:
+                raise ValueError(f"App got multiple deployments named '{d.name}'.")
 
-    @property
-    def ingress(self) -> Deployment:
-        raise NotImplementedError()
+            deployment_dict[d.name] = d
+
+        self._deployments = ImmutableDeploymentDict(deployment_dict)
 
     @property
     def deployments(self) -> ImmutableDeploymentDict:
-        raise NotImplementedError()
+        return self._deployments
+
+    @property
+    def ingress(self) -> Optional[Deployment]:
+        """Gets the app's ingress, if one exists.
+
+        The ingress is the single deployment with a non-None route prefix. If more
+        or less than one deployment has a route prefix, no single ingress exists,
+        so returns None.
+        """
+
+        ingress = None
+
+        for deployment in self._deployments.values():
+            if deployment.route_prefix is not None:
+                if ingress is None:
+                    ingress = deployment
+                else:
+                    return None
+
+        return ingress
 
     def to_dict(self) -> Dict:
         """Returns this Application's deployments as a dictionary.
@@ -1722,7 +1757,9 @@ def to_dict(self) -> Dict:
         Returns:
             Dict: The Application's deployments formatted in a dictionary.
         """
-        raise NotImplementedError()
+        return ServeApplicationSchema(
+            deployments=[deployment_to_schema(d) for d in self._deployments.values()]
+        ).dict()
 
     @classmethod
     def from_dict(cls, d: Dict) -> "Application":
@@ -1738,7 +1775,9 @@ def from_dict(cls, d: Dict) -> "Application":
         Returns:
             Application: a new application object containing the deployments.
         """
-        raise NotImplementedError()
+
+        schema = ServeApplicationSchema.parse_obj(d)
+        return cls([schema_to_deployment(s) for s in schema.deployments])
 
     def to_yaml(self, f: Optional[TextIO] = None) -> Optional[str]:
         """Returns this application's deployments as a YAML string.
@@ -1760,7 +1799,9 @@ def to_yaml(self, f: Optional[TextIO] = None) -> Optional[str]:
             Optional[String]: The deployments' YAML string. The output is from
                 yaml.safe_dump(). Returned only if no file pointer is passed in.
         """
-        raise NotImplementedError()
+        return yaml.safe_dump(
+            self.to_dict(), stream=f, default_flow_style=False, sort_keys=False
+        )
 
     @classmethod
     def from_yaml(cls, str_or_file: Union[str, TextIO]) -> "Application":
@@ -1784,46 +1825,54 @@ def from_yaml(cls, str_or_file: Union[str, TextIO]) -> "Application":
                 Serve YAML config files.
 
         Returns:
-            Application: a new application object containing the deployments.
+            Application: a new Application object containing the deployments.
         """
-        raise NotImplementedError()
+        return cls.from_dict(yaml.safe_load(str_or_file))
 
 
 @PublicAPI(stability="alpha")
 def run(
-    target: Union[DeploymentNode, Application],
+    target: Union[DeploymentNode, DeploymentFunctionNode, Application],
+    _blocking: bool = True,
     *,
     host: str = DEFAULT_HTTP_HOST,
     port: int = DEFAULT_HTTP_PORT,
-    driver: Optional[Deployment] = None,
     **kwargs,
 ) -> RayServeHandle:
     """Run a Serve application and return a ServeHandle to the ingress.
 
-    Either a DeploymentNode or a pre-built application can be passed in.
-    If a DeploymentNode is passed in, all of the deployments it depends on
-    will be deployed.
+    Either a DeploymentNode, DeploymentFunctionNode, or a pre-built application
+    can be passed in. If a node is passed in, all of the deployments it depends
+    on will be deployed. If there is an ingress, its handle will be returned.
 
     Args:
-        target: User built serve Application or DeploymentNode that acts as
-            the root node of DAG. By default DeploymentNode is the Driver
-            deployment unless user provided customized one.
+        target (Union[DeploymentNode, DeploymentFunctionNode, Application]):
+            A user-built Serve Application or a DeploymentNode that acts as the
+            root node of DAG. By default DeploymentNode is the Driver
+            deployment unless user provides a customized one.
+        host (str): The host passed into serve.start().
+        port (int): The port passed into serve.start().
 
     Returns:
-        handle: A regular ray serve handle that can be called by user to exeucte
-            the serve DAG.
+        RayServeHandle: A regular ray serve handle that can be called by user
+            to execute the serve DAG.
     """
     # TODO (jiaodong): Resolve circular reference in pipeline codebase and serve
     from ray.serve.pipeline.api import build as pipeline_build
 
     client = start(detached=True, http_options={"host": host, "port": port})
 
+    if isinstance(target, Application):
+        deployments = list(target.deployments.values())
+        ingress = target.ingress
     # Each DAG should always provide a valid Driver DeploymentNode
-    if isinstance(target, (DeploymentNode)):
+    elif isinstance(target, DeploymentNode):
         deployments = pipeline_build(target)
+        ingress = deployments[-1]
     # Special case where user is doing single function serve.run(func.bind())
-    elif isinstance(target, (DeploymentFunctionNode)):
+    elif isinstance(target, DeploymentFunctionNode):
         deployments = pipeline_build(target)
+        ingress = deployments[-1]
         if len(deployments) != 1:
             raise ValueError(
                 "We only support single function node in serve.run, ex: "
@@ -1839,10 +1888,16 @@ def run(
             "as entrypoint to your Serve DAG."
         )
     else:
-        raise NotImplementedError()
+        raise TypeError(
+            "Expected a DeploymentNode, DeploymentFunctionNode, or "
+            "Application as target. Got unexpected type "
+            f'"{type(target)}" instead.'
+        )
+
+    parameter_group = []
 
-    parameter_group = [
-        {
+    for deployment in deployments:
+        deployment_parameters = {
             "name": deployment._name,
             "func_or_class": deployment._func_or_class,
             "init_args": deployment.init_args,
@@ -1854,11 +1909,13 @@ def run(
             "route_prefix": deployment.route_prefix,
             "url": deployment.url,
         }
-        for deployment in deployments
-    ]
 
-    client.deploy_group(parameter_group, _blocking=True)
-    return deployments[-1].get_handle()
+        parameter_group.append(deployment_parameters)
+
+    client.deploy_group(parameter_group, _blocking=_blocking)
+
+    if ingress is not None:
+        return ingress.get_handle()
 
 
 @PublicAPI(stability="alpha")
@@ -1880,3 +1937,103 @@ def build(target: DeploymentNode) -> Application:
     # TODO(edoakes): this should accept host and port, but we don't
     # currently support them in the REST API.
     raise NotImplementedError()
+
+
+def deployment_to_schema(d: Deployment) -> DeploymentSchema:
+    """Converts a live deployment object to a corresponding structured schema.
+
+    If the deployment has a class or function, it will be attemptetd to be
+    converted to a valid corresponding import path.
+
+    init_args and init_kwargs must also be JSON-serializable or this call will
+    fail.
+    """
+
+    if d.ray_actor_options is not None:
+        ray_actor_options_schema = RayActorOptionsSchema.parse_obj(d.ray_actor_options)
+    else:
+        ray_actor_options_schema = None
+
+    return DeploymentSchema(
+        name=d.name,
+        import_path=get_deployment_import_path(d),
+        init_args=d.init_args,
+        init_kwargs=d.init_kwargs,
+        num_replicas=d.num_replicas,
+        route_prefix=d.route_prefix,
+        max_concurrent_queries=d.max_concurrent_queries,
+        user_config=d.user_config,
+        autoscaling_config=d._config.autoscaling_config,
+        graceful_shutdown_wait_loop_s=d._config.graceful_shutdown_wait_loop_s,
+        graceful_shutdown_timeout_s=d._config.graceful_shutdown_timeout_s,
+        health_check_period_s=d._config.health_check_period_s,
+        health_check_timeout_s=d._config.health_check_timeout_s,
+        ray_actor_options=ray_actor_options_schema,
+    )
+
+
+def schema_to_deployment(s: DeploymentSchema) -> Deployment:
+    from ray.serve.pipeline.json_serde import convert_from_json_safe_obj
+
+    if s.ray_actor_options is None:
+        ray_actor_options = None
+    else:
+        ray_actor_options = s.ray_actor_options.dict(exclude_unset=True)
+
+    return deployment(
+        name=s.name,
+        init_args=convert_from_json_safe_obj(s.init_args),
+        init_kwargs=convert_from_json_safe_obj(s.init_kwargs),
+        num_replicas=s.num_replicas,
+        route_prefix=s.route_prefix,
+        max_concurrent_queries=s.max_concurrent_queries,
+        user_config=convert_from_json_safe_obj(s.user_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,
+        ray_actor_options=ray_actor_options,
+    )(s.import_path)
+
+
+def serve_application_to_schema(
+    deployments: List[Deployment],
+) -> ServeApplicationSchema:
+    schemas = [deployment_to_schema(d) for d in deployments]
+    return ServeApplicationSchema(deployments=schemas)
+
+
+def schema_to_serve_application(schema: ServeApplicationSchema) -> List[Deployment]:
+    return [schema_to_deployment(s) for s in schema.deployments]
+
+
+def status_info_to_schema(
+    deployment_name: str, status_info: Union[DeploymentStatusInfo, Dict]
+) -> DeploymentStatusSchema:
+    if isinstance(status_info, DeploymentStatusInfo):
+        return DeploymentStatusSchema(
+            name=deployment_name, status=status_info.status, message=status_info.message
+        )
+    elif isinstance(status_info, dict):
+        return DeploymentStatusSchema(
+            name=deployment_name,
+            status=status_info["status"],
+            message=status_info["message"],
+        )
+    else:
+        raise TypeError(
+            f"Got {type(status_info)} as status_info's "
+            "type. Expected status_info to be either a "
+            "DeploymentStatusInfo or a dictionary."
+        )
+
+
+def serve_application_status_to_schema(
+    status_infos: Dict[str, Union[DeploymentStatusInfo, Dict]]
+) -> ServeApplicationStatusSchema:
+    schemas = [
+        status_info_to_schema(deployment_name, status_info)
+        for deployment_name, status_info in status_infos.items()
+    ]
+    return ServeApplicationStatusSchema(statuses=schemas)