[Serve] Check multiple FastAPI ingress deployments in a single application (#53647)

## Why are these changes needed?

- Currently, Serve can not catch multiple FastAPI deployments in a
single application if user sets the docs path to None in their FastAPI
app.
- We can check multiple ASGIAppReplicaWrapper in a single application to
avoid this issue.

## Related issue number

Closes #53024

## Checks

- [x] I've signed off every commit(by using the -s flag, i.e., `git
commit -s`) in this PR.
- [x] I've run `scripts/format.sh` to lint the changes in this PR.
- [x] I've included any doc changes needed for
https://docs.ray.io/en/master/.
- [ ] I've added any new APIs to the API Reference. For example, if I
added a
method in Tune, I've added it in `doc/source/tune/api/` under the
           corresponding `.rst` file.
- [x] I've made sure the tests are passing. Note that there might be a
few flaky tests, see the recent failures at https://flakey-tests.ray.io/
- Testing Strategy
   - [ ] Unit tests
   - [ ] Release tests
   - [ ] This PR is not tested :(

---------

Signed-off-by: Ziy1-Tan <ajb459684460@gmail.com>

Author: Ziy1-Tan

python/ray/serve/_private/application_state.py

@@ -1,3 +1,4 @@
+import inspect
 import json
 import logging
 import os
@@ -40,6 +41,7 @@
     override_runtime_envs_except_env_vars,
     validate_route_prefix,
 )
+from ray.serve.api import ASGIAppReplicaWrapper
 from ray.serve.config import AutoscalingConfig
 from ray.serve.exceptions import RayServeException
 from ray.serve.generated.serve_pb2 import (
@@ -468,6 +470,7 @@ def deploy_app(self, deployment_infos: Dict[str, DeploymentInfo]):
             or docs path.
         """
 
+        self._check_ingress_deployments(deployment_infos)
         # Check routes are unique in deployment infos
         self._route_prefix, self._docs_path = self._check_routes(deployment_infos)
 
@@ -704,6 +707,28 @@ def _reconcile_build_app_task(self) -> Tuple[Optional[Dict], BuildAppStatus, str
             )
             return None, BuildAppStatus.FAILED, error_msg
 
+    def _check_ingress_deployments(
+        self, deployment_infos: Dict[str, DeploymentInfo]
+    ) -> None:
+        """Check @serve.ingress of deployments in app.
+
+        Raises: RayServeException if more than one @serve.ingress
+            is found among deployments.
+        """
+        num_ingress_deployments = 0
+        for info in deployment_infos.values():
+            if inspect.isclass(info.replica_config.deployment_def) and issubclass(
+                info.replica_config.deployment_def, ASGIAppReplicaWrapper
+            ):
+                num_ingress_deployments += 1
+
+        if num_ingress_deployments > 1:
+            raise RayServeException(
+                f'Found multiple FastAPI deployments in application "{self._name}".'
+                "Please only include one deployment with @serve.ingress"
+                "in your application to avoid this issue."
+            )
+
     def _check_routes(
         self, deployment_infos: Dict[str, DeploymentInfo]
     ) -> Tuple[str, str]:
@@ -721,6 +746,9 @@ def _check_routes(
         num_route_prefixes = 0
         num_docs_paths = 0
         route_prefix = None
+        # TODO(Ziy1-Tan): `docs_path` will be removed when
+        # https://github.com/ray-project/ray/issues/53023 is resolved.
+        # We can get it from DeploymentStateManager directly.
         docs_path = None
         for info in deployment_infos.values():
             # Update route prefix of application, which may be updated
@@ -1177,7 +1205,12 @@ def build_serve_application(
             name=name,
             default_runtime_env=ray.get_runtime_context().runtime_env,
         )
+        num_ingress_deployments = 0
         for deployment in built_app.deployments:
+            if inspect.isclass(deployment.func_or_class) and issubclass(
+                deployment.func_or_class, ASGIAppReplicaWrapper
+            ):
+                num_ingress_deployments += 1
             is_ingress = deployment.name == built_app.ingress_deployment_name
             deploy_args_list.append(
                 get_deploy_args(
@@ -1190,6 +1223,12 @@ def build_serve_application(
                     docs_path=deployment._docs_path,
                 )
             )
+        if num_ingress_deployments > 1:
+            return None, (
+                f'Found multiple FastAPI deployments in application "{built_app.name}". '
+                "Please only include one deployment with @serve.ingress "
+                "in your application to avoid this issue."
+            )
         return deploy_args_list, None
     except KeyboardInterrupt:
         # Error is raised when this task is canceled with ray.cancel(), which

python/ray/serve/tests/test_config_files/multi_fastapi.py

@@ -0,0 +1,28 @@
+from fastapi import FastAPI
+
+from ray import serve
+from ray.serve.handle import DeploymentHandle
+
+app1 = FastAPI()
+app2 = FastAPI()
+
+
+@serve.deployment
+@serve.ingress(app2)
+class SubModel:
+    def add(self, a: int):
+        return a + 1
+
+
+@serve.deployment
+@serve.ingress(app1)
+class Model:
+    def __init__(self, submodel: DeploymentHandle):
+        self.submodel = submodel
+
+    @app1.get("/{a}")
+    async def func(self, a: int):
+        return await self.submodel.add.remote(a)
+
+
+invalid_model = Model.bind(SubModel.bind())

python/ray/serve/tests/test_deploy_app.py

@@ -51,6 +51,13 @@ def check_deployments_dead(deployment_ids: List[DeploymentID]):
     return all(f"ServeReplica::{p}" not in actor_names for p in prefixes)
 
 
+def check_deploy_failed(app_name: str, message: str):
+    status = serve.status().applications[app_name]
+    assert status.status == "DEPLOY_FAILED"
+    assert message in status.message
+    return True
+
+
 def get_test_config() -> Dict:
     return {"import_path": "ray.serve.tests.test_config_files.pizza.serve_dag"}
 
@@ -127,9 +134,23 @@ def test_deploy_multi_app_basic(serve_instance):
     check_multi_app()
 
 
-def test_deploy_multi_app_update_config(serve_instance):
+def test_two_fastapi_in_one_application(serve_instance):
     client = serve_instance
+    config = {
+        "applications": [
+            {
+                "name": "app1",
+                "route_prefix": "/app1",
+                "import_path": "ray.serve.tests.test_config_files.multi_fastapi.invalid_model",
+            }
+        ],
+    }
+    client.deploy_apps(ServeDeploySchema.parse_obj(config))
+    wait_for_condition(check_deploy_failed, app_name="app1", message="FastAPI")
 
+
+def test_deploy_multi_app_update_config(serve_instance):
+    client = serve_instance
     config = get_test_deploy_config()
     client.deploy_apps(ServeDeploySchema.parse_obj(config))
     check_multi_app()

python/ray/serve/tests/test_fastapi.py

@@ -727,15 +727,16 @@ def decr2(self):
 
 
 @pytest.mark.parametrize("two_fastapi", [True, False])
+@pytest.mark.parametrize("docs_url", ["/docs", None])
 def test_two_fastapi_in_one_application(
-    serve_instance: ServeControllerClient, two_fastapi
+    serve_instance: ServeControllerClient, two_fastapi, docs_url
 ):
     """
     Check that a deployment graph that would normally work, will not deploy
     successfully if there are two FastAPI deployments.
     """
-    app1 = FastAPI()
-    app2 = FastAPI()
+    app1 = FastAPI(docs_url=docs_url)
+    app2 = FastAPI(docs_url=docs_url)
 
     class SubModel:
         def add(self, a: int):