[3/n] [Serve] expose deployment topology in serve instance details (ray-project#58355)

### Summary

This PR exposes deployment topology information in Ray Serve instance
details, allowing users to visualize and understand the dependency graph
of deployments within their applications.

### What's Changed

#### New Data Structures

Added two new schema classes to represent deployment topology:

- **`DeploymentNode`** - Represents a node in the deployment DAG

- **`DeploymentTopology`** - Represents the full dependency graph

#### Implementation

**Controller Integration**
- Updated `ServeController` to include `deployment_topology` in
`ApplicationDetails` when serving instance details
- Topology is now accessible via the `get_serve_details()` API

---

**Example Output:**

```python
{
    "app_name": "my_app",
    "ingress_deployment": "Ingress",
    "nodes": {
        "Ingress": {
            "name": "Ingress",
            "is_ingress": True,
            "outbound_deployments": [
                {"name": "ServiceA", "app_name": "my_app"}
            ]
        },
        "ServiceA": {
            "name": "ServiceA",
            "is_ingress": False,
            "outbound_deployments": [
                {"name": "Database", "app_name": "my_app"}
            ]
        },
        "Database": {
            "name": "Database",
            "is_ingress": False,
            "outbound_deployments": []
        }
    }
}
```

---------

Signed-off-by: abrar <abrar@anyscale.com>
Co-authored-by: Lonnie Liu <95255098+aslonnie@users.noreply.github.com>

Author: 2 people

doc/source/serve/api/index.md

@@ -416,6 +416,9 @@ Content-Type: application/json
    schema.ProxyStatus
    schema.TargetGroup
    schema.Target
+   schema.DeploymentNode
+   schema.DeploymentTopology
+
 ```
 
 ## Observability

python/ray/serve/_private/application_state.py

@@ -61,6 +61,8 @@
     APIType,
     ApplicationStatus,
     DeploymentDetails,
+    DeploymentNode,
+    DeploymentTopology,
     LoggingConfig,
     ServeApplicationSchema,
 )
@@ -935,6 +937,46 @@ def _reconcile_target_deployments(self) -> None:
 
         return target_state_changed
 
+    def get_deployment_topology(self) -> Optional[DeploymentTopology]:
+        """Get the deployment topology for this application.
+
+        Returns:
+            The deployment topology, or None if not yet built.
+        """
+        if not self.target_deployments:
+            return None
+
+        nodes = {}
+
+        # Using target deployments because we wish to build best effort topology based on current state.
+        for deployment_name in self.target_deployments:
+            deployment_id = DeploymentID(name=deployment_name, app_name=self._name)
+
+            # Get outbound deployment names from deployment state
+            outbound_deployment = (
+                self._deployment_state_manager.get_deployment_outbound_deployments(
+                    deployment_id
+                )
+            ) or []
+
+            # Create node for this deployment
+            node = DeploymentNode(
+                name=deployment_name,
+                app_name=self._name,
+                outbound_deployments=[
+                    {"name": dep.name, "app_name": dep.app_name}
+                    for dep in outbound_deployment
+                ],
+                is_ingress=(deployment_name == self._ingress_deployment_name),
+            )
+            nodes[deployment_name] = node
+
+        return DeploymentTopology(
+            app_name=self._name,
+            ingress_deployment=self._ingress_deployment_name,
+            nodes=nodes,
+        )
+
     def update(self) -> Tuple[bool, bool]:
         """Attempts to reconcile this application to match its target state.
 
@@ -1280,6 +1322,20 @@ def list_deployment_details(self, name: str) -> Dict[str, DeploymentDetails]:
             return {}
         return self._application_states[name].list_deployment_details()
 
+    def get_deployment_topology(self, app_name: str) -> Optional[DeploymentTopology]:
+        """Get the deployment topology for an application.
+
+        Args:
+            app_name: Name of the application.
+
+        Returns:
+            The deployment topology for the application, or None if the application
+            doesn't exist or the topology hasn't been built yet.
+        """
+        if app_name not in self._application_states:
+            return None
+        return self._application_states[app_name].get_deployment_topology()
+
     def update(self):
         """Update each application state."""
         apps_to_be_deleted = []

python/ray/serve/_private/controller.py

@@ -988,6 +988,9 @@ def get_serve_instance_details(self, source: Optional[APIType] = None) -> Dict:
                 deployments=self.application_state_manager.list_deployment_details(
                     app_name
                 ),
+                deployment_topology=self.application_state_manager.get_deployment_topology(
+                    app_name
+                ),
             )
 
         # NOTE(zcin): We use exclude_unset here because we explicitly and intentionally

python/ray/serve/schema.py

@@ -1109,6 +1109,46 @@ def get_valid_user_values(cls):
         return [cls.IMPERATIVE.value, cls.DECLARATIVE.value]
 
 
+@PublicAPI(stability="alpha")
+class DeploymentNode(BaseModel):
+    """Represents a node in the deployment topology.
+
+    Each node represents a deployment and tracks which other deployments it calls.
+    """
+
+    name: str = Field(description="The name of the deployment.")
+    app_name: str = Field(
+        description="The name of the application this deployment belongs to."
+    )
+    # using name and app_name instead of just deployment name because outbound dependencies can be in different apps
+    outbound_deployments: List[dict] = Field(
+        default_factory=list,
+        description="The deployment IDs that this deployment calls (outbound dependencies).",
+    )
+    is_ingress: bool = Field(
+        default=False, description="Whether this is the ingress deployment."
+    )
+
+
+@PublicAPI(stability="alpha")
+class DeploymentTopology(BaseModel):
+    """Represents the dependency graph of deployments in an application.
+
+    The topology shows which deployments call which other deployments,
+    with the ingress deployment as the entry point.
+    """
+
+    app_name: str = Field(
+        description="The name of the application this topology belongs to."
+    )
+    nodes: Dict[str, DeploymentNode] = Field(
+        description="The adjacency list of deployment nodes."
+    )
+    ingress_deployment: Optional[str] = Field(
+        default=None, description="The name of the ingress deployment (entry point)."
+    )
+
+
 @PublicAPI(stability="stable")
 class ApplicationDetails(BaseModel, extra=Extra.forbid, frozen=True):
     """Detailed info about a Serve application."""
@@ -1172,6 +1212,11 @@ class ApplicationDetails(BaseModel, extra=Extra.forbid, frozen=True):
         "route_prefix", allow_reuse=True
     )(_route_prefix_format)
 
+    deployment_topology: Optional[DeploymentTopology] = Field(
+        default=None,
+        description="The deployment topology showing how deployments in this application call each other.",
+    )
+
 
 @PublicAPI(stability="stable")
 class ProxyDetails(ServeActorDetails, frozen=True):

python/ray/serve/tests/BUILD.bazel

@@ -110,6 +110,7 @@ py_test_module_list(
         "test_controller_recovery.py",
         "test_deploy_2.py",
         "test_deployment_scheduler.py",
+        "test_deployment_topology.py",
         "test_failure.py",
         "test_handle_1.py",
         "test_handle_2.py",

python/ray/serve/tests/test_controller.py

@@ -216,6 +216,18 @@ def autoscaling_app():
                             ],
                         }
                     },
+                    "deployment_topology": {
+                        "app_name": "default",
+                        "nodes": {
+                            "autoscaling_app": {
+                                "name": "autoscaling_app",
+                                "app_name": "default",
+                                "outbound_deployments": [],
+                                "is_ingress": True,
+                            }
+                        },
+                        "ingress_deployment": "autoscaling_app",
+                    },
                 }
             },
             "target_capacity": None,