wehub-resource-sync
857 lines
30 KiB
Python
857 lines
30 KiB
Python
import asyncio
|
|
import sys
|
|
|
|
import pytest
|
|
import requests
|
|
from fastapi import FastAPI
|
|
|
|
import ray
|
|
from ray import serve
|
|
from ray._common.test_utils import SignalActor, wait_for_condition
|
|
from ray.serve.api import get_deployment_handle
|
|
from ray.serve.context import _get_global_client
|
|
from ray.serve.handle import DeploymentHandle
|
|
from ray.serve.schema import ApplicationStatus
|
|
|
|
|
|
class TestDeploymentTopology:
|
|
def test_simple_chain_topology(self, serve_instance):
|
|
"""Test topology with a simple chain: ingress -> downstream."""
|
|
|
|
# Define a downstream deployment
|
|
@serve.deployment
|
|
class Downstream:
|
|
def __call__(self):
|
|
return "downstream response"
|
|
|
|
# Define an ingress deployment that calls downstream
|
|
@serve.deployment
|
|
class Ingress:
|
|
def __init__(self, downstream: DeploymentHandle):
|
|
self.downstream = downstream
|
|
|
|
async def __call__(self):
|
|
result = await self.downstream.remote()
|
|
return f"ingress -> {result}"
|
|
|
|
# Deploy the application
|
|
downstream = Downstream.bind()
|
|
ingress = Ingress.bind(downstream)
|
|
serve.run(ingress, name="test_app", route_prefix="/test")
|
|
|
|
# Make a request to ensure deployments are fully initialized
|
|
response = requests.get("http://localhost:8000/test")
|
|
assert response.status_code == 200
|
|
assert "downstream response" in response.text
|
|
|
|
# Define expected topology
|
|
expected_topology = {
|
|
"app_name": "test_app",
|
|
"ingress_deployment": "Ingress",
|
|
"nodes": {
|
|
"Ingress": {
|
|
"name": "Ingress",
|
|
"app_name": "test_app",
|
|
"is_ingress": True,
|
|
"outbound_deployments": [
|
|
{"name": "Downstream", "app_name": "test_app"}
|
|
],
|
|
},
|
|
"Downstream": {
|
|
"name": "Downstream",
|
|
"app_name": "test_app",
|
|
"is_ingress": False,
|
|
"outbound_deployments": [],
|
|
},
|
|
},
|
|
}
|
|
|
|
# Final verification - get the actual topology
|
|
client = _get_global_client()
|
|
status = client.get_serve_details()
|
|
actual_topology = status["applications"]["test_app"]["deployment_topology"]
|
|
|
|
# Direct comparison with expected topology
|
|
assert actual_topology == expected_topology
|
|
|
|
def test_multi_level_chain_topology(self, serve_instance):
|
|
"""Test topology with deep dependencies: Ingress -> A -> B -> C."""
|
|
|
|
@serve.deployment
|
|
class ServiceC:
|
|
def __call__(self):
|
|
return "C"
|
|
|
|
@serve.deployment
|
|
class ServiceB:
|
|
def __init__(self, service_c: DeploymentHandle):
|
|
self.service_c = service_c
|
|
|
|
async def __call__(self):
|
|
result = await self.service_c.remote()
|
|
return f"B->{result}"
|
|
|
|
@serve.deployment
|
|
class ServiceA:
|
|
def __init__(self, service_b: DeploymentHandle):
|
|
self.service_b = service_b
|
|
|
|
async def __call__(self):
|
|
result = await self.service_b.remote()
|
|
return f"A->{result}"
|
|
|
|
@serve.deployment
|
|
class Ingress:
|
|
def __init__(self, service_a: DeploymentHandle):
|
|
self.service_a = service_a
|
|
|
|
async def __call__(self):
|
|
result = await self.service_a.remote()
|
|
return f"Ingress->{result}"
|
|
|
|
# Deploy the application
|
|
service_c = ServiceC.bind()
|
|
service_b = ServiceB.bind(service_c)
|
|
service_a = ServiceA.bind(service_b)
|
|
ingress = Ingress.bind(service_a)
|
|
serve.run(ingress, name="test_app", route_prefix="/test")
|
|
|
|
# Make a request
|
|
response = requests.get("http://localhost:8000/test")
|
|
assert response.status_code == 200
|
|
assert "C" in response.text
|
|
|
|
# Expected topology
|
|
expected_topology = {
|
|
"app_name": "test_app",
|
|
"ingress_deployment": "Ingress",
|
|
"nodes": {
|
|
"Ingress": {
|
|
"name": "Ingress",
|
|
"app_name": "test_app",
|
|
"is_ingress": True,
|
|
"outbound_deployments": [
|
|
{"name": "ServiceA", "app_name": "test_app"}
|
|
],
|
|
},
|
|
"ServiceA": {
|
|
"name": "ServiceA",
|
|
"app_name": "test_app",
|
|
"is_ingress": False,
|
|
"outbound_deployments": [
|
|
{"name": "ServiceB", "app_name": "test_app"}
|
|
],
|
|
},
|
|
"ServiceB": {
|
|
"name": "ServiceB",
|
|
"app_name": "test_app",
|
|
"is_ingress": False,
|
|
"outbound_deployments": [
|
|
{"name": "ServiceC", "app_name": "test_app"}
|
|
],
|
|
},
|
|
"ServiceC": {
|
|
"name": "ServiceC",
|
|
"app_name": "test_app",
|
|
"is_ingress": False,
|
|
"outbound_deployments": [],
|
|
},
|
|
},
|
|
}
|
|
|
|
# Final verification
|
|
client = _get_global_client()
|
|
status = client.get_serve_details()
|
|
actual_topology = status["applications"]["test_app"]["deployment_topology"]
|
|
assert actual_topology == expected_topology
|
|
|
|
def test_fan_out_topology(self, serve_instance):
|
|
"""Test topology with fan-out: Ingress -> [Service1, Service2, Service3]."""
|
|
|
|
@serve.deployment
|
|
class Service1:
|
|
def __call__(self):
|
|
return "service1"
|
|
|
|
@serve.deployment
|
|
class Service2:
|
|
def __call__(self):
|
|
return "service2"
|
|
|
|
@serve.deployment
|
|
class Service3:
|
|
def __call__(self):
|
|
return "service3"
|
|
|
|
@serve.deployment
|
|
class Ingress:
|
|
def __init__(
|
|
self,
|
|
service1: DeploymentHandle,
|
|
service2: DeploymentHandle,
|
|
service3: DeploymentHandle,
|
|
):
|
|
self.service1 = service1
|
|
self.service2 = service2
|
|
self.service3 = service3
|
|
|
|
async def __call__(self):
|
|
results = await asyncio.gather(
|
|
self.service1.remote(),
|
|
self.service2.remote(),
|
|
self.service3.remote(),
|
|
)
|
|
return f"Results: {','.join(results)}"
|
|
|
|
# Deploy the application
|
|
service1 = Service1.bind()
|
|
service2 = Service2.bind()
|
|
service3 = Service3.bind()
|
|
ingress = Ingress.bind(service1, service2, service3)
|
|
serve.run(ingress, name="test_app", route_prefix="/test")
|
|
|
|
# Make a request
|
|
response = requests.get("http://localhost:8000/test")
|
|
assert response.status_code == 200
|
|
|
|
# Expected topology - outbound_deployments should contain all three services
|
|
expected_topology = {
|
|
"app_name": "test_app",
|
|
"ingress_deployment": "Ingress",
|
|
"nodes": {
|
|
"Ingress": {
|
|
"name": "Ingress",
|
|
"app_name": "test_app",
|
|
"is_ingress": True,
|
|
"outbound_deployments": [
|
|
{"name": "Service1", "app_name": "test_app"},
|
|
{"name": "Service2", "app_name": "test_app"},
|
|
{"name": "Service3", "app_name": "test_app"},
|
|
],
|
|
},
|
|
"Service1": {
|
|
"name": "Service1",
|
|
"app_name": "test_app",
|
|
"is_ingress": False,
|
|
"outbound_deployments": [],
|
|
},
|
|
"Service2": {
|
|
"name": "Service2",
|
|
"app_name": "test_app",
|
|
"is_ingress": False,
|
|
"outbound_deployments": [],
|
|
},
|
|
"Service3": {
|
|
"name": "Service3",
|
|
"app_name": "test_app",
|
|
"is_ingress": False,
|
|
"outbound_deployments": [],
|
|
},
|
|
},
|
|
}
|
|
|
|
# Final verification
|
|
client = _get_global_client()
|
|
status = client.get_serve_details()
|
|
actual_topology = status["applications"]["test_app"]["deployment_topology"]
|
|
# Sort for comparison
|
|
actual_topology["nodes"]["Ingress"]["outbound_deployments"] = sorted(
|
|
actual_topology["nodes"]["Ingress"]["outbound_deployments"],
|
|
key=lambda x: x["name"],
|
|
)
|
|
expected_topology["nodes"]["Ingress"]["outbound_deployments"] = sorted(
|
|
expected_topology["nodes"]["Ingress"]["outbound_deployments"],
|
|
key=lambda x: x["name"],
|
|
)
|
|
assert actual_topology == expected_topology
|
|
|
|
def test_diamond_topology(self, serve_instance):
|
|
"""Test diamond pattern: Ingress -> [ServiceA, ServiceB] -> Database."""
|
|
|
|
@serve.deployment
|
|
class Database:
|
|
def __call__(self):
|
|
return "db_data"
|
|
|
|
@serve.deployment
|
|
class ServiceA:
|
|
def __init__(self, database: DeploymentHandle):
|
|
self.database = database
|
|
|
|
async def __call__(self):
|
|
data = await self.database.remote()
|
|
return f"A:{data}"
|
|
|
|
@serve.deployment
|
|
class ServiceB:
|
|
def __init__(self, database: DeploymentHandle):
|
|
self.database = database
|
|
|
|
async def __call__(self):
|
|
data = await self.database.remote()
|
|
return f"B:{data}"
|
|
|
|
@serve.deployment
|
|
class Ingress:
|
|
def __init__(
|
|
self, service_a: DeploymentHandle, service_b: DeploymentHandle
|
|
):
|
|
self.service_a = service_a
|
|
self.service_b = service_b
|
|
|
|
async def __call__(self):
|
|
results = await asyncio.gather(
|
|
self.service_a.remote(),
|
|
self.service_b.remote(),
|
|
)
|
|
return f"Results: {','.join(results)}"
|
|
|
|
# Deploy the application
|
|
database = Database.bind()
|
|
service_a = ServiceA.bind(database)
|
|
service_b = ServiceB.bind(database)
|
|
ingress = Ingress.bind(service_a, service_b)
|
|
serve.run(ingress, name="test_app", route_prefix="/test")
|
|
|
|
# Make a request
|
|
response = requests.get("http://localhost:8000/test")
|
|
assert response.status_code == 200
|
|
|
|
# Expected topology
|
|
expected_topology = {
|
|
"app_name": "test_app",
|
|
"ingress_deployment": "Ingress",
|
|
"nodes": {
|
|
"Ingress": {
|
|
"name": "Ingress",
|
|
"app_name": "test_app",
|
|
"is_ingress": True,
|
|
"outbound_deployments": [
|
|
{"name": "ServiceA", "app_name": "test_app"},
|
|
{"name": "ServiceB", "app_name": "test_app"},
|
|
],
|
|
},
|
|
"ServiceA": {
|
|
"name": "ServiceA",
|
|
"app_name": "test_app",
|
|
"is_ingress": False,
|
|
"outbound_deployments": [
|
|
{"name": "Database", "app_name": "test_app"}
|
|
],
|
|
},
|
|
"ServiceB": {
|
|
"name": "ServiceB",
|
|
"app_name": "test_app",
|
|
"is_ingress": False,
|
|
"outbound_deployments": [
|
|
{"name": "Database", "app_name": "test_app"}
|
|
],
|
|
},
|
|
"Database": {
|
|
"name": "Database",
|
|
"app_name": "test_app",
|
|
"is_ingress": False,
|
|
"outbound_deployments": [],
|
|
},
|
|
},
|
|
}
|
|
|
|
# Final verification
|
|
client = _get_global_client()
|
|
status = client.get_serve_details()
|
|
actual_topology = status["applications"]["test_app"]["deployment_topology"]
|
|
# Sort for comparison
|
|
actual_topology["nodes"]["Ingress"]["outbound_deployments"] = sorted(
|
|
actual_topology["nodes"]["Ingress"]["outbound_deployments"],
|
|
key=lambda x: x["name"],
|
|
)
|
|
expected_topology["nodes"]["Ingress"]["outbound_deployments"] = sorted(
|
|
expected_topology["nodes"]["Ingress"]["outbound_deployments"],
|
|
key=lambda x: x["name"],
|
|
)
|
|
assert actual_topology == expected_topology
|
|
|
|
def test_cross_app_topology(self, serve_instance):
|
|
"""Test cross-application dependencies."""
|
|
|
|
# Deploy App2 first (the dependency)
|
|
@serve.deployment
|
|
class App2Service:
|
|
def __call__(self):
|
|
return "app2_response"
|
|
|
|
app2_service = App2Service.bind()
|
|
serve.run(app2_service, name="app2", route_prefix="/app2")
|
|
|
|
# Deploy App1 that depends on App2
|
|
@serve.deployment
|
|
class App1Service:
|
|
def __init__(self):
|
|
self.app2_handle = get_deployment_handle("App2Service", "app2")
|
|
|
|
async def __call__(self):
|
|
result = await self.app2_handle.remote()
|
|
return f"app1->{result}"
|
|
|
|
app1_service = App1Service.bind()
|
|
serve.run(app1_service, name="app1", route_prefix="/app1")
|
|
|
|
# Make requests to both apps to ensure they're initialized
|
|
response1 = requests.get("http://localhost:8000/app1")
|
|
assert response1.status_code == 200
|
|
response2 = requests.get("http://localhost:8000/app2")
|
|
assert response2.status_code == 200
|
|
|
|
# Expected topology for app1
|
|
expected_topology_app1 = {
|
|
"app_name": "app1",
|
|
"ingress_deployment": "App1Service",
|
|
"nodes": {
|
|
"App1Service": {
|
|
"name": "App1Service",
|
|
"app_name": "app1",
|
|
"is_ingress": True,
|
|
"outbound_deployments": [
|
|
{"name": "App2Service", "app_name": "app2"}
|
|
],
|
|
},
|
|
},
|
|
}
|
|
|
|
# Expected topology for app2
|
|
expected_topology_app2 = {
|
|
"app_name": "app2",
|
|
"ingress_deployment": "App2Service",
|
|
"nodes": {
|
|
"App2Service": {
|
|
"name": "App2Service",
|
|
"app_name": "app2",
|
|
"is_ingress": True,
|
|
"outbound_deployments": [],
|
|
},
|
|
},
|
|
}
|
|
|
|
# Wait for topologies to be built
|
|
def check_topology():
|
|
client = _get_global_client()
|
|
status = client.get_serve_details()
|
|
topology_app1 = (
|
|
status.get("applications", {})
|
|
.get("app1", {})
|
|
.get("deployment_topology")
|
|
)
|
|
topology_app2 = (
|
|
status.get("applications", {})
|
|
.get("app2", {})
|
|
.get("deployment_topology")
|
|
)
|
|
return (
|
|
topology_app1 == expected_topology_app1
|
|
and topology_app2 == expected_topology_app2
|
|
)
|
|
|
|
# NOTE(abrar): using wait_for_condition because outbound deployments are added asynchronously
|
|
# when get_deployment_handle is used.
|
|
wait_for_condition(check_topology)
|
|
|
|
# Final verification
|
|
client = _get_global_client()
|
|
status = client.get_serve_details()
|
|
actual_topology_app1 = status["applications"]["app1"]["deployment_topology"]
|
|
actual_topology_app2 = status["applications"]["app2"]["deployment_topology"]
|
|
assert actual_topology_app1 == expected_topology_app1
|
|
assert actual_topology_app2 == expected_topology_app2
|
|
|
|
def test_single_deployment_no_dependencies(self, serve_instance):
|
|
"""Test single deployment with no outbound dependencies."""
|
|
|
|
@serve.deployment
|
|
class Standalone:
|
|
def __call__(self):
|
|
return "standalone"
|
|
|
|
standalone = Standalone.bind()
|
|
serve.run(standalone, name="test_app", route_prefix="/test")
|
|
|
|
# Make a request
|
|
response = requests.get("http://localhost:8000/test")
|
|
assert response.status_code == 200
|
|
assert response.text == "standalone"
|
|
|
|
# Expected topology
|
|
expected_topology = {
|
|
"app_name": "test_app",
|
|
"ingress_deployment": "Standalone",
|
|
"nodes": {
|
|
"Standalone": {
|
|
"name": "Standalone",
|
|
"app_name": "test_app",
|
|
"is_ingress": True,
|
|
"outbound_deployments": [],
|
|
},
|
|
},
|
|
}
|
|
|
|
# Final verification
|
|
client = _get_global_client()
|
|
status = client.get_serve_details()
|
|
actual_topology = status["applications"]["test_app"]["deployment_topology"]
|
|
assert actual_topology == expected_topology
|
|
|
|
def test_topology_update_on_redeploy(self, serve_instance):
|
|
"""Test that topology updates when application is redeployed with different structure."""
|
|
|
|
# Initial deployment: Ingress -> ServiceA
|
|
@serve.deployment
|
|
class ServiceA:
|
|
def __call__(self):
|
|
return "A"
|
|
|
|
@serve.deployment
|
|
class Ingress:
|
|
def __init__(self, service_a: DeploymentHandle):
|
|
self.service_a = service_a
|
|
|
|
async def __call__(self):
|
|
result = await self.service_a.remote()
|
|
return f"Ingress->{result}"
|
|
|
|
service_a = ServiceA.bind()
|
|
ingress = Ingress.bind(service_a)
|
|
serve.run(ingress, name="test_app", route_prefix="/test")
|
|
|
|
# Make a request
|
|
response = requests.get("http://localhost:8000/test")
|
|
assert response.status_code == 200
|
|
|
|
# Expected initial topology
|
|
expected_topology_v1 = {
|
|
"app_name": "test_app",
|
|
"ingress_deployment": "Ingress",
|
|
"nodes": {
|
|
"Ingress": {
|
|
"name": "Ingress",
|
|
"app_name": "test_app",
|
|
"is_ingress": True,
|
|
"outbound_deployments": [
|
|
{"name": "ServiceA", "app_name": "test_app"}
|
|
],
|
|
},
|
|
"ServiceA": {
|
|
"name": "ServiceA",
|
|
"app_name": "test_app",
|
|
"is_ingress": False,
|
|
"outbound_deployments": [],
|
|
},
|
|
},
|
|
}
|
|
client = _get_global_client()
|
|
status = client.get_serve_details()
|
|
actual_topology = status["applications"]["test_app"]["deployment_topology"]
|
|
assert expected_topology_v1 == actual_topology
|
|
|
|
# Redeploy with different structure: Ingress -> ServiceB
|
|
@serve.deployment
|
|
class ServiceB:
|
|
def __call__(self):
|
|
return "B"
|
|
|
|
@serve.deployment
|
|
class IngressV2:
|
|
def __init__(self, service_b: DeploymentHandle):
|
|
self.service_b = service_b
|
|
|
|
async def __call__(self):
|
|
result = await self.service_b.remote()
|
|
return f"IngressV2->{result}"
|
|
|
|
service_b = ServiceB.bind()
|
|
ingress_v2 = IngressV2.bind(service_b)
|
|
serve.run(ingress_v2, name="test_app", route_prefix="/test")
|
|
|
|
# Make a request to new deployment
|
|
response = requests.get("http://localhost:8000/test")
|
|
assert response.status_code == 200
|
|
assert "B" in response.text
|
|
|
|
# Expected updated topology
|
|
expected_topology_v2 = {
|
|
"app_name": "test_app",
|
|
"ingress_deployment": "IngressV2",
|
|
"nodes": {
|
|
"IngressV2": {
|
|
"name": "IngressV2",
|
|
"app_name": "test_app",
|
|
"is_ingress": True,
|
|
"outbound_deployments": [
|
|
{"name": "ServiceB", "app_name": "test_app"}
|
|
],
|
|
},
|
|
"ServiceB": {
|
|
"name": "ServiceB",
|
|
"app_name": "test_app",
|
|
"is_ingress": False,
|
|
"outbound_deployments": [],
|
|
},
|
|
},
|
|
}
|
|
|
|
# Final verification
|
|
client = _get_global_client()
|
|
status = client.get_serve_details()
|
|
actual_topology = status["applications"]["test_app"]["deployment_topology"]
|
|
assert actual_topology == expected_topology_v2
|
|
|
|
def test_fastapi_factory_with_get_deployment_handle(self, serve_instance):
|
|
"""Test topology with FastAPI factory pattern using get_deployment_handle.
|
|
|
|
This test demonstrates the pattern where get_deployment_handle is used
|
|
inside a FastAPI request handler (lazy lookup), which is common in
|
|
factory patterns where dependencies aren't injected at construction time.
|
|
"""
|
|
|
|
# Backend service that will be called via get_deployment_handle
|
|
@serve.deployment(name="BackendService")
|
|
class BackendService:
|
|
def __call__(self):
|
|
return "backend_data"
|
|
|
|
# FastAPI ingress using factory pattern
|
|
def create_fastapi_app():
|
|
app = FastAPI()
|
|
|
|
@app.get("/")
|
|
async def root():
|
|
# Use get_deployment_handle inside the request handler (lazy lookup)
|
|
# This is the factory pattern - handle is obtained at request time
|
|
backend_handle = get_deployment_handle("BackendService", "test_app")
|
|
result = await backend_handle.remote()
|
|
return {"data": result}
|
|
|
|
@app.get("/health")
|
|
async def health():
|
|
return {"status": "ok"}
|
|
|
|
return app
|
|
|
|
@serve.deployment(name="FastAPIIngress")
|
|
@serve.ingress(create_fastapi_app())
|
|
class FastAPIIngress:
|
|
def __init__(self, backend_service: DeploymentHandle):
|
|
self.backend_service = backend_service
|
|
|
|
# Deploy both deployments
|
|
ingress_deployment = FastAPIIngress.bind(BackendService.bind())
|
|
|
|
serve.run(ingress_deployment, name="test_app", route_prefix="/test")
|
|
|
|
# Make requests to ensure deployments are initialized
|
|
response = requests.get("http://localhost:8000/test/health")
|
|
assert response.status_code == 200
|
|
|
|
# Make a request that triggers get_deployment_handle
|
|
response = requests.get("http://localhost:8000/test/")
|
|
assert response.status_code == 200
|
|
assert "backend_data" in response.text
|
|
|
|
# Expected topology
|
|
expected_topology = {
|
|
"app_name": "test_app",
|
|
"ingress_deployment": "FastAPIIngress",
|
|
"nodes": {
|
|
"FastAPIIngress": {
|
|
"name": "FastAPIIngress",
|
|
"app_name": "test_app",
|
|
"is_ingress": True,
|
|
"outbound_deployments": [
|
|
{"name": "BackendService", "app_name": "test_app"}
|
|
],
|
|
},
|
|
"BackendService": {
|
|
"name": "BackendService",
|
|
"app_name": "test_app",
|
|
"is_ingress": False,
|
|
"outbound_deployments": [],
|
|
},
|
|
},
|
|
}
|
|
|
|
# Wait for topology to be built
|
|
# NOTE(abrar): using wait_for_condition because outbound deployments are added asynchronously
|
|
# when get_deployment_handle is used.
|
|
def check_topology():
|
|
client = _get_global_client()
|
|
status = client.get_serve_details()
|
|
topology = (
|
|
status.get("applications", {})
|
|
.get("test_app", {})
|
|
.get("deployment_topology")
|
|
)
|
|
return topology == expected_topology
|
|
|
|
wait_for_condition(check_topology)
|
|
|
|
# Final verification
|
|
client = _get_global_client()
|
|
status = client.get_serve_details()
|
|
actual_topology = status["applications"]["test_app"]["deployment_topology"]
|
|
assert actual_topology == expected_topology
|
|
|
|
def test_topology_with_updating_deployment(self, serve_instance):
|
|
"""Test topology when a deployment is updating (not fully rolled out)."""
|
|
|
|
# Create signal actor for synchronization
|
|
signal = SignalActor.remote()
|
|
|
|
# Initial deployment
|
|
@serve.deployment(name="ServiceA")
|
|
class ServiceA:
|
|
def __call__(self):
|
|
return "v1"
|
|
|
|
@serve.deployment(name="Ingress")
|
|
class Ingress:
|
|
def __init__(self, service_a: DeploymentHandle):
|
|
# Block until signal is sent to keep deployment in UPDATING state
|
|
ray.get(signal.wait.remote())
|
|
self.service_a = service_a
|
|
|
|
async def __call__(self):
|
|
result = await self.service_a.remote()
|
|
return f"Ingress->{result}"
|
|
|
|
# Deploy initial version
|
|
service_a_v1 = ServiceA.bind()
|
|
ingress = Ingress.bind(service_a_v1)
|
|
serve._run(ingress, name="test_app", route_prefix="/test", _blocking=False)
|
|
|
|
wait_for_condition(lambda: ray.get(signal.cur_num_waiters.remote()) == 1)
|
|
|
|
# Verify initial topology
|
|
initial_topology = {
|
|
"app_name": "test_app",
|
|
"ingress_deployment": "Ingress",
|
|
"nodes": {
|
|
"Ingress": {
|
|
"name": "Ingress",
|
|
"app_name": "test_app",
|
|
"is_ingress": True,
|
|
"outbound_deployments": [],
|
|
},
|
|
"ServiceA": {
|
|
"name": "ServiceA",
|
|
"app_name": "test_app",
|
|
"is_ingress": False,
|
|
"outbound_deployments": [],
|
|
},
|
|
},
|
|
}
|
|
|
|
def check_initial_topology():
|
|
client = _get_global_client()
|
|
status = client.get_serve_details()
|
|
if "test_app" not in status["applications"]:
|
|
return False
|
|
app_status = status["applications"]["test_app"]["status"]
|
|
return app_status == ApplicationStatus.DEPLOYING.value
|
|
|
|
wait_for_condition(check_initial_topology)
|
|
|
|
client = _get_global_client()
|
|
status = client.get_serve_details()
|
|
actual_topology = status["applications"]["test_app"]["deployment_topology"]
|
|
assert actual_topology == initial_topology
|
|
|
|
# Clean up: send signal to unblock replicas
|
|
ray.get(signal.send.remote())
|
|
|
|
def test_topology_with_failed_deployment(self, serve_instance):
|
|
"""Test topology when a deployment fails to start."""
|
|
|
|
# A deployment that will fail to start
|
|
@serve.deployment
|
|
class ServiceB:
|
|
def __call__(self):
|
|
return "b"
|
|
|
|
@serve.deployment(name="FailingService")
|
|
class FailingService:
|
|
def __init__(self, service_b: DeploymentHandle):
|
|
self.service_b = service_b
|
|
raise RuntimeError("Intentional failure for testing")
|
|
|
|
def __call__(self):
|
|
return "should never reach here"
|
|
|
|
@serve.deployment(name="Ingress")
|
|
class Ingress:
|
|
def __init__(self, failing_service: DeploymentHandle):
|
|
self.failing_service = failing_service
|
|
|
|
async def __call__(self):
|
|
result = await self.failing_service.remote()
|
|
return f"Ingress->{result}"
|
|
|
|
# Deploy with failing service
|
|
service_b = ServiceB.bind()
|
|
failing_service = FailingService.bind(service_b)
|
|
ingress = Ingress.bind(failing_service)
|
|
|
|
# Deploy (this won't fail immediately, but replicas will fail to start)
|
|
serve._run(ingress, name="test_app", route_prefix="/test", _blocking=False)
|
|
|
|
# Wait for the deployment to be processed and replicas to fail
|
|
def deployment_has_failures():
|
|
client = _get_global_client()
|
|
status = client.get_serve_details()
|
|
if "test_app" not in status["applications"]:
|
|
return False
|
|
app_status = status["applications"]["test_app"]["status"]
|
|
return app_status == ApplicationStatus.DEPLOY_FAILED.value
|
|
|
|
wait_for_condition(deployment_has_failures, timeout=10)
|
|
|
|
# Get the serve details
|
|
client = _get_global_client()
|
|
status = client.get_serve_details()
|
|
|
|
# Verify the application exists (even with failed deployment)
|
|
assert "test_app" in status["applications"]
|
|
|
|
# Verify deployment_topology still exists
|
|
# The topology should be built based on the deployment structure,
|
|
# regardless of whether the replicas are healthy
|
|
topology = status["applications"]["test_app"]["deployment_topology"]
|
|
expected_topology = {
|
|
"app_name": "test_app",
|
|
"ingress_deployment": "Ingress",
|
|
"nodes": {
|
|
"Ingress": {
|
|
"name": "Ingress",
|
|
"app_name": "test_app",
|
|
"is_ingress": True,
|
|
"outbound_deployments": [
|
|
{"name": "FailingService", "app_name": "test_app"}
|
|
],
|
|
},
|
|
"FailingService": {
|
|
"name": "FailingService",
|
|
"app_name": "test_app",
|
|
"is_ingress": False,
|
|
"outbound_deployments": [],
|
|
},
|
|
"ServiceB": {
|
|
"name": "ServiceB",
|
|
"app_name": "test_app",
|
|
"is_ingress": False,
|
|
"outbound_deployments": [],
|
|
},
|
|
},
|
|
}
|
|
assert topology == expected_topology
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(pytest.main(["-v", "-s", __file__]))
|