wehub-resource-sync
1782 lines
57 KiB
Python
1782 lines
57 KiB
Python
import concurrent.futures
|
||
import os
|
||
import sys
|
||
import time
|
||
|
||
import httpx
|
||
import pytest
|
||
import yaml
|
||
|
||
import ray
|
||
from ray import serve
|
||
from ray._common.test_utils import SignalActor, wait_for_condition
|
||
from ray.serve._private.common import DeploymentID, DeploymentStatus, ReplicaState
|
||
from ray.serve._private.constants import (
|
||
REPLICA_HEALTH_CHECK_UNHEALTHY_THRESHOLD,
|
||
SERVE_DEFAULT_APP_NAME,
|
||
SERVE_DEPLOYMENT_ACTOR_PREFIX,
|
||
SERVE_NAMESPACE,
|
||
)
|
||
from ray.serve._private.test_utils import (
|
||
check_replica_counts,
|
||
check_running,
|
||
get_application_url,
|
||
)
|
||
from ray.serve.config import (
|
||
AutoscalingConfig,
|
||
DeploymentActorConfig,
|
||
GangSchedulingConfig,
|
||
)
|
||
from ray.serve.exceptions import RayServeException
|
||
from ray.serve.schema import ApplicationStatus, ServeDeploySchema
|
||
from ray.util.state import list_actors
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Test actor classes – must be at module level for importability via string
|
||
# paths in the config-based (declarative) tests.
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@ray.remote
|
||
class SharedCounter:
|
||
"""A trivial deployment-scoped actor used across several tests."""
|
||
|
||
def __init__(self, start: int = 0):
|
||
self.count = start
|
||
|
||
def increment(self):
|
||
self.count += 1
|
||
return self.count
|
||
|
||
def get(self):
|
||
return self.count
|
||
|
||
def ray_actor_id(self) -> str:
|
||
"""Stable Ray actor id for tests that must detect recreation vs same process."""
|
||
return ray.get_runtime_context().get_actor_id()
|
||
|
||
|
||
@ray.remote
|
||
class ConstructorGate:
|
||
"""Coordination actor: deployment-scoped tests can block ``__init__`` until opened."""
|
||
|
||
def __init__(self):
|
||
self._open = True
|
||
|
||
def close(self):
|
||
self._open = False
|
||
|
||
def open(self):
|
||
self._open = True
|
||
|
||
def wait_until_open(self):
|
||
|
||
while not self._open:
|
||
time.sleep(0.01)
|
||
|
||
|
||
@ray.remote
|
||
class GatedSharedCounter:
|
||
"""Like ``SharedCounter`` but blocks in ``__init__`` until ``ConstructorGate`` opens."""
|
||
|
||
def __init__(self, gate, start: int = 0):
|
||
ray.get(gate.wait_until_open.remote())
|
||
self.count = start
|
||
|
||
def get(self):
|
||
return self.count
|
||
|
||
def ray_actor_id(self) -> str:
|
||
return ray.get_runtime_context().get_actor_id()
|
||
|
||
|
||
@ray.remote
|
||
class SharedCache:
|
||
def __init__(self):
|
||
self.data = {}
|
||
|
||
def put(self, key, value):
|
||
self.data[key] = value
|
||
|
||
def get(self, key):
|
||
return self.data.get(key)
|
||
|
||
|
||
@ray.remote
|
||
class EnvReportingActor:
|
||
"""Reports env vars for testing runtime_env inheritance."""
|
||
|
||
def get_env(self, key: str) -> str:
|
||
import os
|
||
|
||
return os.environ.get(key, "NOT_SET")
|
||
|
||
|
||
@ray.remote
|
||
class AltCounter:
|
||
"""Alternative counter with different behavior, for testing actor class swap."""
|
||
|
||
def __init__(self, start: int = 0):
|
||
self.count = start * 10
|
||
|
||
def get(self):
|
||
return self.count
|
||
|
||
|
||
@ray.remote
|
||
class InitArgsActor:
|
||
"""Accepts init_args for testing positional constructor args."""
|
||
|
||
def __init__(self, prefix: str, suffix: str = "default"):
|
||
self.prefix = prefix
|
||
self.suffix = suffix
|
||
|
||
def get_value(self) -> str:
|
||
return f"{self.prefix}:{self.suffix}"
|
||
|
||
|
||
@ray.remote
|
||
class FailingDeploymentActor:
|
||
"""Raises in __init__ for testing deployment actor constructor failure."""
|
||
|
||
def __init__(self, should_fail: bool = True):
|
||
if should_fail:
|
||
raise RuntimeError("Deployment actor init failed")
|
||
|
||
|
||
@ray.remote
|
||
class DeploymentActorContextActor:
|
||
"""Reads deployment actor runtime context for testing."""
|
||
|
||
def get_context(self):
|
||
ctx = serve.get_deployment_actor_context()
|
||
return {
|
||
"deployment": ctx.deployment,
|
||
"app_name": ctx.app_name,
|
||
"actor_name": ctx.actor_name,
|
||
"code_version": ctx.code_version,
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Helpers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _get_deployment_actor_names() -> list:
|
||
"""Return Ray actor names matching the deployment-actor naming prefix."""
|
||
actors = list_actors(
|
||
filters=[
|
||
("state", "=", "ALIVE"),
|
||
]
|
||
)
|
||
return [
|
||
a["name"]
|
||
for a in actors
|
||
if a["name"] and a["name"].startswith(SERVE_DEPLOYMENT_ACTOR_PREFIX)
|
||
]
|
||
|
||
|
||
def _check_no_deployment_actors():
|
||
return len(_get_deployment_actor_names()) == 0
|
||
|
||
|
||
def _check_deployment_actor_count(expected: int):
|
||
return len(_get_deployment_actor_names()) == expected
|
||
|
||
|
||
def _get_deployment_actor_names_for_app(app_name: str, deployment_name: str) -> list:
|
||
"""Return deployment actor names for a specific app and deployment."""
|
||
prefix = f"{SERVE_DEPLOYMENT_ACTOR_PREFIX}{app_name}::{deployment_name}::"
|
||
return [
|
||
a["name"]
|
||
for a in list_actors(filters=[("state", "=", "ALIVE")])
|
||
if a.get("name", "").startswith(prefix)
|
||
]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests – Imperative (code-based) API
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_imperative_deploy_with_actor_class(serve_instance):
|
||
"""Imperative API: deploy with @ray.remote actor class object.
|
||
|
||
Verifies the actor is created, accessible from the replica via
|
||
serve.get_deployment_actor, and that state is shared across requests.
|
||
"""
|
||
|
||
@serve.deployment(
|
||
deployment_actors=[
|
||
DeploymentActorConfig(
|
||
name="counter",
|
||
actor_class=SharedCounter,
|
||
init_kwargs={"start": 10},
|
||
),
|
||
],
|
||
)
|
||
class MyDeployment:
|
||
def __call__(self):
|
||
counter = serve.get_deployment_actor("counter")
|
||
val = ray.get(counter.increment.remote())
|
||
return str(val)
|
||
|
||
serve.run(MyDeployment.bind())
|
||
url = f"{get_application_url()}/"
|
||
|
||
resp1 = httpx.get(url)
|
||
assert resp1.status_code == 200
|
||
assert resp1.text == "11"
|
||
|
||
resp2 = httpx.get(url)
|
||
assert resp2.status_code == 200
|
||
assert resp2.text == "12"
|
||
|
||
|
||
def test_imperative_deploy_with_string_import_path(serve_instance):
|
||
"""Imperative API: deploy with actor_class as a string import path.
|
||
|
||
The build process should import and serialize the class by value so
|
||
it is usable on the controller without the original module.
|
||
"""
|
||
|
||
@serve.deployment(
|
||
deployment_actors=[
|
||
DeploymentActorConfig(
|
||
name="counter",
|
||
actor_class=("ray.serve.tests.test_deployment_actors:SharedCounter"),
|
||
init_kwargs={"start": 0},
|
||
),
|
||
],
|
||
)
|
||
class MyDeployment:
|
||
def __call__(self):
|
||
counter = serve.get_deployment_actor("counter")
|
||
return str(ray.get(counter.get.remote()))
|
||
|
||
serve.run(MyDeployment.bind())
|
||
url = f"{get_application_url()}/"
|
||
resp = httpx.get(url)
|
||
assert resp.status_code == 200
|
||
assert resp.text == "0"
|
||
|
||
|
||
def test_deployment_actor_inherits_runtime_env(serve_instance):
|
||
"""Deployment actors inherit runtime_env from deployment ray_actor_options."""
|
||
|
||
@serve.deployment(
|
||
ray_actor_options={
|
||
"runtime_env": {"env_vars": {"DEPLOYMENT_RUNTIME_ENV": "inherited"}}
|
||
},
|
||
deployment_actors=[
|
||
DeploymentActorConfig(
|
||
name="env_actor",
|
||
actor_class=EnvReportingActor,
|
||
init_kwargs={},
|
||
),
|
||
],
|
||
)
|
||
class MyDeployment:
|
||
def __call__(self):
|
||
actor = serve.get_deployment_actor("env_actor")
|
||
return ray.get(actor.get_env.remote("DEPLOYMENT_RUNTIME_ENV"))
|
||
|
||
serve.run(MyDeployment.bind())
|
||
url = f"{get_application_url()}/"
|
||
wait_for_condition(lambda: httpx.get(url).text == "inherited")
|
||
|
||
|
||
def test_imperative_deploy_multiple_actors(serve_instance):
|
||
"""Imperative API: deploy with multiple deployment-scoped actors.
|
||
|
||
Verifies each actor is independently accessible from the replica.
|
||
"""
|
||
|
||
@serve.deployment(
|
||
deployment_actors=[
|
||
DeploymentActorConfig(
|
||
name="counter",
|
||
actor_class=SharedCounter,
|
||
init_kwargs={"start": 100},
|
||
),
|
||
DeploymentActorConfig(
|
||
name="cache",
|
||
actor_class=SharedCache,
|
||
),
|
||
],
|
||
)
|
||
class MyDeployment:
|
||
def __call__(self, request):
|
||
counter = serve.get_deployment_actor("counter")
|
||
cache = serve.get_deployment_actor("cache")
|
||
val = ray.get(counter.get.remote())
|
||
ray.get(cache.put.remote("count", val))
|
||
cached = ray.get(cache.get.remote("count"))
|
||
return str(cached)
|
||
|
||
serve.run(MyDeployment.bind())
|
||
url = f"{get_application_url()}/"
|
||
resp = httpx.get(url)
|
||
assert resp.status_code == 200
|
||
assert resp.text == "100"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests – Declarative (config-based) API
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_declarative_deploy_with_deployment_actors(serve_instance):
|
||
"""Declarative API: deploy via config dict with string actor_class paths.
|
||
|
||
This tests the full build-task serialization path where actor classes
|
||
are serialized by value inside build_serve_application.
|
||
"""
|
||
client = serve_instance
|
||
|
||
config = {
|
||
"applications": [
|
||
{
|
||
"name": "default",
|
||
"route_prefix": "/",
|
||
"import_path": ("ray.serve.tests.test_config_files.world.DagNode"),
|
||
"deployments": [
|
||
{
|
||
"name": "BasicDriver",
|
||
"deployment_actors": [
|
||
{
|
||
"name": "counter",
|
||
"actor_class": (
|
||
"ray.serve.tests.test_deployment_actors"
|
||
":SharedCounter"
|
||
),
|
||
"init_kwargs": {"start": 0},
|
||
},
|
||
],
|
||
},
|
||
],
|
||
},
|
||
],
|
||
}
|
||
|
||
client.deploy_apps(ServeDeploySchema.model_validate(config))
|
||
wait_for_condition(check_running, timeout=30)
|
||
|
||
wait_for_condition(lambda: _check_deployment_actor_count(1))
|
||
|
||
actor_names = [
|
||
n
|
||
for n in _get_deployment_actor_names()
|
||
if "BasicDriver" in n and "counter" in n
|
||
]
|
||
assert len(actor_names) == 1
|
||
handle = ray.get_actor(actor_names[0], namespace=SERVE_NAMESPACE)
|
||
assert ray.get(handle.get.remote()) == 0
|
||
assert ray.get(handle.increment.remote()) == 1
|
||
|
||
|
||
def test_declarative_config_only_actors(serve_instance):
|
||
"""Config-only deployment actors: actors defined only in config YAML,
|
||
not in the code.
|
||
|
||
The actor class string is extracted on the controller, sent to the build
|
||
task for serialization, and the bytes are injected during
|
||
override_deployment_info.
|
||
"""
|
||
client = serve_instance
|
||
|
||
config = {
|
||
"applications": [
|
||
{
|
||
"name": "default",
|
||
"route_prefix": "/",
|
||
"import_path": ("ray.serve.tests.test_config_files.world.DagNode"),
|
||
"deployments": [
|
||
{
|
||
"name": "BasicDriver",
|
||
"deployment_actors": [
|
||
{
|
||
"name": "shared_counter",
|
||
"actor_class": (
|
||
"ray.serve.tests.test_deployment_actors"
|
||
":SharedCounter"
|
||
),
|
||
"init_kwargs": {"start": 42},
|
||
},
|
||
],
|
||
},
|
||
],
|
||
},
|
||
],
|
||
}
|
||
|
||
client.deploy_apps(ServeDeploySchema.model_validate(config))
|
||
wait_for_condition(check_running, timeout=30)
|
||
|
||
wait_for_condition(lambda: _check_deployment_actor_count(1))
|
||
|
||
actor_names = [
|
||
n
|
||
for n in _get_deployment_actor_names()
|
||
if "BasicDriver" in n and "shared_counter" in n
|
||
]
|
||
assert len(actor_names) == 1
|
||
handle = ray.get_actor(actor_names[0], namespace=SERVE_NAMESPACE)
|
||
assert ray.get(handle.get.remote()) == 42
|
||
assert ray.get(handle.increment.remote()) == 43
|
||
|
||
|
||
def test_declarative_config_only_actors_no_version_change(serve_instance):
|
||
"""Config-only deployment actors: redeploy with same config (no version change).
|
||
|
||
Same config redeploy preserves deployment actors; no orphan cleanup.
|
||
"""
|
||
client = serve_instance
|
||
|
||
config = {
|
||
"applications": [
|
||
{
|
||
"name": "default",
|
||
"route_prefix": "/",
|
||
"import_path": (
|
||
"ray.serve.tests.test_config_files.config_only_deployment_actor:app"
|
||
),
|
||
"deployments": [
|
||
{
|
||
"name": "ConfigOnlyDriver",
|
||
"deployment_actors": [
|
||
{
|
||
"name": "counter",
|
||
"actor_class": (
|
||
"ray.serve.tests.test_deployment_actors"
|
||
":SharedCounter"
|
||
),
|
||
"init_kwargs": {"start": 10},
|
||
},
|
||
],
|
||
},
|
||
],
|
||
},
|
||
],
|
||
}
|
||
|
||
client.deploy_apps(ServeDeploySchema.model_validate(config))
|
||
wait_for_condition(check_running, timeout=30)
|
||
url = f"{get_application_url()}/"
|
||
wait_for_condition(lambda: httpx.get(url).text == "10")
|
||
|
||
actor_names_before = _get_deployment_actor_names()
|
||
assert len(actor_names_before) == 1
|
||
|
||
client.deploy_apps(ServeDeploySchema.model_validate(config))
|
||
wait_for_condition(check_running, timeout=30)
|
||
wait_for_condition(lambda: httpx.get(url).text == "10")
|
||
|
||
actor_names_after = _get_deployment_actor_names()
|
||
assert actor_names_after == actor_names_before
|
||
|
||
|
||
def test_declarative_config_only_actors_version_change(serve_instance):
|
||
"""Config-only deployment actors: redeploy with different config (version change).
|
||
|
||
Changing init_kwargs triggers new code version; old actors cleaned up,
|
||
new actors created.
|
||
"""
|
||
client = serve_instance
|
||
|
||
config_v1 = {
|
||
"applications": [
|
||
{
|
||
"name": "default",
|
||
"route_prefix": "/",
|
||
"import_path": (
|
||
"ray.serve.tests.test_config_files.config_only_deployment_actor:app"
|
||
),
|
||
"deployments": [
|
||
{
|
||
"name": "ConfigOnlyDriver",
|
||
"deployment_actors": [
|
||
{
|
||
"name": "counter",
|
||
"actor_class": (
|
||
"ray.serve.tests.test_deployment_actors"
|
||
":SharedCounter"
|
||
),
|
||
"init_kwargs": {"start": 100},
|
||
},
|
||
],
|
||
},
|
||
],
|
||
},
|
||
],
|
||
}
|
||
|
||
client.deploy_apps(ServeDeploySchema.model_validate(config_v1))
|
||
wait_for_condition(check_running, timeout=30)
|
||
url = f"{get_application_url()}/"
|
||
wait_for_condition(lambda: httpx.get(url).text == "100")
|
||
|
||
actor_names_v1 = _get_deployment_actor_names()
|
||
assert len(actor_names_v1) == 1
|
||
|
||
config_v2 = {
|
||
"applications": [
|
||
{
|
||
"name": "default",
|
||
"route_prefix": "/",
|
||
"import_path": (
|
||
"ray.serve.tests.test_config_files.config_only_deployment_actor:app"
|
||
),
|
||
"deployments": [
|
||
{
|
||
"name": "ConfigOnlyDriver",
|
||
"deployment_actors": [
|
||
{
|
||
"name": "counter",
|
||
"actor_class": (
|
||
"ray.serve.tests.test_deployment_actors"
|
||
":SharedCounter"
|
||
),
|
||
"init_kwargs": {"start": 200},
|
||
},
|
||
],
|
||
},
|
||
],
|
||
},
|
||
],
|
||
}
|
||
|
||
client.deploy_apps(ServeDeploySchema.model_validate(config_v2))
|
||
wait_for_condition(check_running, timeout=30)
|
||
wait_for_condition(lambda: httpx.get(url).text == "200")
|
||
|
||
actor_names_v2 = _get_deployment_actor_names()
|
||
assert len(actor_names_v2) == 1
|
||
assert actor_names_v1[0] != actor_names_v2[0]
|
||
|
||
|
||
def test_incremental_rollout_version_isolation(serve_instance):
|
||
"""During incremental rollout, v1 replicas use v1 deployment actors and
|
||
v2 replicas use v2 deployment actors.
|
||
|
||
Uses SignalActor to synchronize: block 1 v1 replica so it stays during
|
||
rollout. New requests go to v2; blocked request proves v1 used v1 actor.
|
||
get_deployment_actor resolves by code_version.
|
||
"""
|
||
|
||
signal = SignalActor.remote()
|
||
|
||
@serve.deployment(
|
||
name="test",
|
||
num_replicas=2,
|
||
deployment_actors=[
|
||
DeploymentActorConfig(
|
||
name="counter",
|
||
actor_class=SharedCounter,
|
||
init_kwargs={"start": 1},
|
||
),
|
||
],
|
||
)
|
||
class V1:
|
||
async def __call__(self):
|
||
await signal.wait.remote()
|
||
counter = serve.get_deployment_actor("counter")
|
||
return "1", ray.get(counter.get.remote()), os.getpid()
|
||
|
||
@serve.deployment(
|
||
name="test",
|
||
num_replicas=2,
|
||
deployment_actors=[
|
||
DeploymentActorConfig(
|
||
name="counter",
|
||
actor_class=SharedCounter,
|
||
init_kwargs={"start": 2},
|
||
),
|
||
],
|
||
)
|
||
class V2:
|
||
def __call__(self):
|
||
counter = serve.get_deployment_actor("counter")
|
||
return "2", ray.get(counter.get.remote()), os.getpid()
|
||
|
||
h = serve.run(V1.bind(), name="app", route_prefix="/incremental_rollout_da")
|
||
|
||
signal.send.remote()
|
||
refs = [h.remote() for _ in range(10)]
|
||
results = [ref.result() for ref in refs]
|
||
versions, counter_vals, _ = zip(*results)
|
||
assert versions.count("1") == 10
|
||
assert counter_vals.count(1) == 10
|
||
initial_pids = {r[2] for r in results}
|
||
assert len(initial_pids) == 2
|
||
|
||
# Block 1 v1 replica so it stays during rollout
|
||
signal.send.remote(clear=True)
|
||
blocked_ref = h.remote()
|
||
|
||
serve._run(
|
||
V2.bind(),
|
||
_blocking=False,
|
||
name="app",
|
||
route_prefix="/incremental_rollout_da",
|
||
)
|
||
|
||
# 1 v1 STOPPING (blocked) + 2 v2 RUNNING
|
||
wait_for_condition(
|
||
check_replica_counts,
|
||
controller=serve_instance._controller,
|
||
deployment_id=DeploymentID(name="test", app_name="app"),
|
||
total=3,
|
||
by_state=[
|
||
(ReplicaState.STOPPING, 1, lambda r: r._actor.pid in initial_pids),
|
||
(ReplicaState.RUNNING, 2, lambda r: r._actor.pid not in initial_pids),
|
||
],
|
||
)
|
||
wait_for_condition(lambda: _check_deployment_actor_count(2))
|
||
|
||
# New requests go to v2; verify v2 uses v2 actor
|
||
refs = [h.remote() for _ in range(10)]
|
||
versions, counter_vals, _ = zip(*[ref.result(timeout_s=5) for ref in refs])
|
||
assert versions.count("2") == 10
|
||
assert counter_vals.count(2) == 10
|
||
|
||
# Release blocked v1; its response proves v1 used v1 actor
|
||
ray.get(signal.send.remote())
|
||
version, counter_val, pid = blocked_ref.result(timeout_s=5)
|
||
assert version == "1" and counter_val == 1 and pid in initial_pids
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests – Actor cleanup and redeployment
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_actors_cleaned_up_on_app_delete(serve_instance):
|
||
"""Deployment actors are killed when the application is deleted."""
|
||
|
||
@serve.deployment(
|
||
deployment_actors=[
|
||
DeploymentActorConfig(
|
||
name="counter",
|
||
actor_class=SharedCounter,
|
||
),
|
||
],
|
||
)
|
||
class MyDeployment:
|
||
def __call__(self):
|
||
return "ok"
|
||
|
||
serve.run(MyDeployment.bind())
|
||
url = f"{get_application_url()}/"
|
||
wait_for_condition(lambda: httpx.get(url).status_code == 200)
|
||
|
||
wait_for_condition(lambda: _check_deployment_actor_count(1))
|
||
|
||
serve.delete(name="default")
|
||
|
||
wait_for_condition(_check_no_deployment_actors, timeout=15)
|
||
|
||
|
||
def test_redeployment_replaces_actors(serve_instance):
|
||
"""Redeploying with updated deployment_actors creates new actors and
|
||
cleans up old ones (orphan cleanup).
|
||
"""
|
||
|
||
@serve.deployment(
|
||
deployment_actors=[
|
||
DeploymentActorConfig(
|
||
name="counter_v1",
|
||
actor_class=SharedCounter,
|
||
init_kwargs={"start": 1},
|
||
),
|
||
],
|
||
)
|
||
class MyDeployment:
|
||
def __call__(self):
|
||
counter = serve.get_deployment_actor("counter_v1")
|
||
return str(ray.get(counter.get.remote()))
|
||
|
||
MyDeployment = MyDeployment.options(_internal=True, version="v1")
|
||
serve.run(MyDeployment.bind())
|
||
url = f"{get_application_url()}/"
|
||
wait_for_condition(lambda: httpx.get(url).text == "1")
|
||
|
||
actor_names_v1 = _get_deployment_actor_names()
|
||
assert len(actor_names_v1) == 1
|
||
|
||
@serve.deployment(
|
||
deployment_actors=[
|
||
DeploymentActorConfig(
|
||
name="counter_v2",
|
||
actor_class=SharedCounter,
|
||
init_kwargs={"start": 2},
|
||
),
|
||
],
|
||
)
|
||
class MyDeployment: # noqa: F811
|
||
def __call__(self):
|
||
counter = serve.get_deployment_actor("counter_v2")
|
||
return str(ray.get(counter.get.remote()))
|
||
|
||
MyDeployment = MyDeployment.options(_internal=True, version="v2")
|
||
serve.run(MyDeployment.bind())
|
||
wait_for_condition(lambda: httpx.get(url).text == "2")
|
||
|
||
actor_names_v2 = _get_deployment_actor_names()
|
||
assert len(actor_names_v2) == 1
|
||
assert actor_names_v1[0] != actor_names_v2[0]
|
||
|
||
|
||
def test_replica_context_includes_code_version(serve_instance):
|
||
"""Replica context exposes deployment code version."""
|
||
|
||
@serve.deployment(
|
||
num_replicas=1,
|
||
deployment_actors=[
|
||
DeploymentActorConfig(
|
||
name="counter",
|
||
actor_class=SharedCounter,
|
||
init_kwargs={"start": 1},
|
||
),
|
||
],
|
||
)
|
||
class MyDeployment:
|
||
def __call__(self):
|
||
ctx = serve.get_replica_context()
|
||
counter = serve.get_deployment_actor("counter")
|
||
return f"{ctx.code_version}|{ray.get(counter.get.remote())}"
|
||
|
||
MyDeployment = MyDeployment.options(_internal=True, version="v1")
|
||
serve.run(MyDeployment.bind())
|
||
url = f"{get_application_url()}/"
|
||
wait_for_condition(lambda: httpx.get(url).text == "v1|1")
|
||
|
||
@serve.deployment(
|
||
num_replicas=1,
|
||
deployment_actors=[
|
||
DeploymentActorConfig(
|
||
name="counter",
|
||
actor_class=SharedCounter,
|
||
init_kwargs={"start": 2},
|
||
),
|
||
],
|
||
)
|
||
class MyDeployment: # noqa: F811
|
||
def __call__(self):
|
||
ctx = serve.get_replica_context()
|
||
counter = serve.get_deployment_actor("counter")
|
||
return f"{ctx.code_version}|{ray.get(counter.get.remote())}"
|
||
|
||
MyDeployment = MyDeployment.options(_internal=True, version="v2")
|
||
serve.run(MyDeployment.bind())
|
||
wait_for_condition(lambda: httpx.get(url).text == "v2|2")
|
||
|
||
|
||
def test_redeployment_with_no_actors_cleans_up_old(serve_instance):
|
||
"""Redeploying with NO deployment actors cleans up old actors.
|
||
|
||
Regression test: previously the orphan cleanup early-returned if the
|
||
new version had no deployment actors.
|
||
"""
|
||
|
||
@serve.deployment(
|
||
deployment_actors=[
|
||
DeploymentActorConfig(
|
||
name="counter",
|
||
actor_class=SharedCounter,
|
||
init_kwargs={"start": 0},
|
||
),
|
||
],
|
||
)
|
||
class MyDeployment:
|
||
def __call__(self):
|
||
return "v1"
|
||
|
||
MyDeployment = MyDeployment.options(_internal=True, version="v1")
|
||
serve.run(MyDeployment.bind())
|
||
url = f"{get_application_url()}/"
|
||
wait_for_condition(lambda: httpx.get(url).text == "v1")
|
||
wait_for_condition(lambda: _check_deployment_actor_count(1))
|
||
|
||
@serve.deployment
|
||
class MyDeployment: # noqa: F811
|
||
def __call__(self):
|
||
return "v2"
|
||
|
||
MyDeployment = MyDeployment.options(_internal=True, version="v2")
|
||
serve.run(MyDeployment.bind())
|
||
wait_for_condition(lambda: httpx.get(url).text == "v2")
|
||
|
||
wait_for_condition(_check_no_deployment_actors, timeout=15)
|
||
|
||
|
||
def test_deployment_actor_constructor_failure_app_status(serve_instance):
|
||
"""When a deployment actor's constructor fails, app deploy status is DEPLOY_FAILED."""
|
||
|
||
@serve.deployment(
|
||
max_constructor_retry_count=3,
|
||
deployment_actors=[
|
||
DeploymentActorConfig(
|
||
name="bad_actor",
|
||
actor_class=FailingDeploymentActor,
|
||
init_kwargs={"should_fail": True},
|
||
),
|
||
],
|
||
)
|
||
class MyDeployment:
|
||
def __call__(self):
|
||
return "ok"
|
||
|
||
# Non-default app: avoid leaving `default` in DEPLOY_FAILED (teardown).
|
||
serve._run(
|
||
MyDeployment.bind(),
|
||
name="da_ctor_fail_app",
|
||
route_prefix="/da_ctor_fail",
|
||
_blocking=False,
|
||
)
|
||
|
||
wait_for_condition(
|
||
lambda: any(
|
||
"ctor_fail" in n and a.status == ApplicationStatus.DEPLOY_FAILED
|
||
for n, a in serve.status().applications.items()
|
||
),
|
||
timeout=60,
|
||
)
|
||
|
||
status = serve.status()
|
||
app_status = next(a for n, a in status.applications.items() if "ctor_fail" in n)
|
||
assert app_status.status == ApplicationStatus.DEPLOY_FAILED
|
||
assert "MyDeployment" in app_status.deployments
|
||
assert (
|
||
app_status.deployments["MyDeployment"].status == DeploymentStatus.DEPLOY_FAILED
|
||
)
|
||
|
||
serve.delete(
|
||
next(n for n in status.applications if "ctor_fail" in n),
|
||
_blocking=True,
|
||
)
|
||
|
||
|
||
def test_actor_remote_call_error_causes_replica_fail_and_restart(serve_instance):
|
||
"""When a replica's health check fails due to an error in an actor remote call
|
||
(e.g. deployment actor raises), the replica is marked unhealthy and restarted.
|
||
"""
|
||
|
||
@ray.remote
|
||
class SignalGatedHealthCheckActor:
|
||
"""Deployment actor whose ping() fails after a signal is sent.
|
||
|
||
Starts healthy so the test can capture the original replica PID.
|
||
After the signal fires, fails exactly `fail_count` times then recovers,
|
||
so the replacement replica's health checks succeed.
|
||
"""
|
||
|
||
def __init__(self, fail_count: int = 3):
|
||
self._fail_count = fail_count
|
||
self._triggered = False
|
||
self._fail_calls = 0
|
||
|
||
def trigger(self):
|
||
self._triggered = True
|
||
|
||
def ping(self):
|
||
if not self._triggered:
|
||
return "ok"
|
||
self._fail_calls += 1
|
||
if self._fail_calls <= self._fail_count:
|
||
raise RuntimeError("Deployment actor remote call failed")
|
||
return "ok"
|
||
|
||
@serve.deployment(
|
||
name="HealthCheckFailDeployment",
|
||
deployment_actors=[
|
||
DeploymentActorConfig(
|
||
name="health_actor",
|
||
actor_class=SignalGatedHealthCheckActor,
|
||
init_kwargs={
|
||
"fail_count": REPLICA_HEALTH_CHECK_UNHEALTHY_THRESHOLD,
|
||
},
|
||
),
|
||
],
|
||
health_check_period_s=0.5,
|
||
health_check_timeout_s=1,
|
||
)
|
||
class HealthCheckFailDeployment:
|
||
def __call__(self):
|
||
return str(__import__("os").getpid())
|
||
|
||
def check_health(self):
|
||
actor = serve.get_deployment_actor("health_actor")
|
||
ray.get(actor.ping.remote())
|
||
|
||
serve.run(
|
||
HealthCheckFailDeployment.bind(),
|
||
name=SERVE_DEFAULT_APP_NAME,
|
||
route_prefix="/health_check_fail_da",
|
||
)
|
||
url = f"{get_application_url(app_name=SERVE_DEFAULT_APP_NAME)}/"
|
||
wait_for_condition(lambda: httpx.get(url).status_code == 200)
|
||
old_pid = httpx.get(url).text
|
||
|
||
# Trigger the deployment actor to start failing health checks. After
|
||
# REPLICA_HEALTH_CHECK_UNHEALTHY_THRESHOLD consecutive failures the
|
||
# replica is marked unhealthy and restarted. The replacement replica's
|
||
# health checks succeed (actor stops raising after fail_count).
|
||
actor_names = [
|
||
n
|
||
for n in _get_deployment_actor_names()
|
||
if "HealthCheckFailDeployment" in n and "health_actor" in n
|
||
]
|
||
assert len(actor_names) == 1
|
||
da_handle = ray.get_actor(actor_names[0], namespace=SERVE_NAMESPACE)
|
||
ray.get(da_handle.trigger.remote())
|
||
|
||
wait_for_condition(
|
||
lambda: httpx.get(url, timeout=5).text != old_pid,
|
||
timeout=60,
|
||
)
|
||
for _ in range(5):
|
||
assert httpx.get(url).status_code == 200
|
||
|
||
|
||
def test_deployment_actor_killed_no_restart_documents_behavior(serve_instance):
|
||
"""Deployment actor killed with no_restart=True: requests fail.
|
||
|
||
Controller does not detect or recreate. Documents current behavior.
|
||
"""
|
||
|
||
@serve.deployment(
|
||
name="NoRestartDeployment",
|
||
deployment_actors=[
|
||
DeploymentActorConfig(
|
||
name="counter",
|
||
actor_class=SharedCounter,
|
||
init_kwargs={"start": 200},
|
||
),
|
||
],
|
||
)
|
||
class MyDeployment:
|
||
def __call__(self):
|
||
counter = serve.get_deployment_actor("counter")
|
||
return str(ray.get(counter.get.remote()))
|
||
|
||
serve.run(MyDeployment.bind())
|
||
url = f"{get_application_url()}/"
|
||
wait_for_condition(lambda: httpx.get(url).text == "200")
|
||
|
||
actor_names = [
|
||
n
|
||
for n in _get_deployment_actor_names()
|
||
if "NoRestartDeployment" in n and "counter" in n
|
||
]
|
||
assert len(actor_names) == 1
|
||
handle = ray.get_actor(actor_names[0], namespace=SERVE_NAMESPACE)
|
||
ray.kill(handle, no_restart=True)
|
||
|
||
# Requests fail (500 or RayActorError) when calling the dead actor
|
||
resp = httpx.get(url, timeout=10)
|
||
assert resp.status_code == 500
|
||
|
||
|
||
def test_app_delete_cleans_up_deployment_actors(serve_instance):
|
||
"""Application delete cleans up deployment actors."""
|
||
|
||
@serve.deployment(
|
||
deployment_actors=[
|
||
DeploymentActorConfig(
|
||
name="counter",
|
||
actor_class=SharedCounter,
|
||
init_kwargs={"start": 0},
|
||
actor_options={"num_cpus": 0.01},
|
||
),
|
||
],
|
||
)
|
||
class MyDeployment:
|
||
def __call__(self):
|
||
return "ok"
|
||
|
||
serve.run(MyDeployment.bind(), name="delete_test_app", route_prefix="/delete_test")
|
||
wait_for_condition(
|
||
lambda: len(
|
||
_get_deployment_actor_names_for_app("delete_test_app", "MyDeployment")
|
||
)
|
||
== 1
|
||
)
|
||
serve.delete(name="delete_test_app")
|
||
wait_for_condition(
|
||
lambda: len(
|
||
_get_deployment_actor_names_for_app("delete_test_app", "MyDeployment")
|
||
)
|
||
== 0,
|
||
timeout=15,
|
||
)
|
||
|
||
|
||
def test_deployment_actor_actor_options_override_runtime_env(serve_instance):
|
||
"""Deployment actor actor_options override deployment runtime_env."""
|
||
|
||
@serve.deployment(
|
||
ray_actor_options={"runtime_env": {"env_vars": {"ENV_KEY": "from_deployment"}}},
|
||
deployment_actors=[
|
||
DeploymentActorConfig(
|
||
name="env_actor",
|
||
actor_class=EnvReportingActor,
|
||
actor_options={"runtime_env": {"env_vars": {"ENV_KEY": "from_actor"}}},
|
||
),
|
||
],
|
||
)
|
||
class MyDeployment:
|
||
def __call__(self):
|
||
actor = serve.get_deployment_actor("env_actor")
|
||
return ray.get(actor.get_env.remote("ENV_KEY"))
|
||
|
||
serve.run(MyDeployment.bind())
|
||
url = f"{get_application_url()}/"
|
||
wait_for_condition(lambda: httpx.get(url).text == "from_actor")
|
||
|
||
|
||
def test_deployment_actor_with_init_args(serve_instance):
|
||
"""Deployment actor with init_args (not just init_kwargs)."""
|
||
|
||
@serve.deployment(
|
||
deployment_actors=[
|
||
DeploymentActorConfig(
|
||
name="init_args_actor",
|
||
actor_class=InitArgsActor,
|
||
init_args=("my_prefix",),
|
||
init_kwargs={"suffix": "my_suffix"},
|
||
),
|
||
],
|
||
)
|
||
class MyDeployment:
|
||
def __call__(self):
|
||
actor = serve.get_deployment_actor("init_args_actor")
|
||
return ray.get(actor.get_value.remote())
|
||
|
||
serve.run(MyDeployment.bind())
|
||
url = f"{get_application_url()}/"
|
||
wait_for_condition(lambda: httpx.get(url).text == "my_prefix:my_suffix")
|
||
|
||
|
||
def test_deployment_actor_with_resource_constraints(serve_instance):
|
||
"""Deployment actor with actor_options num_cpus schedules correctly."""
|
||
|
||
@serve.deployment(
|
||
deployment_actors=[
|
||
DeploymentActorConfig(
|
||
name="counter",
|
||
actor_class=SharedCounter,
|
||
init_kwargs={"start": 0},
|
||
actor_options={"num_cpus": 0.01},
|
||
),
|
||
],
|
||
)
|
||
class MyDeployment:
|
||
def __call__(self):
|
||
counter = serve.get_deployment_actor("counter")
|
||
return str(ray.get(counter.get.remote()))
|
||
|
||
serve.run(MyDeployment.bind())
|
||
url = f"{get_application_url()}/"
|
||
wait_for_condition(lambda: httpx.get(url).text == "0")
|
||
|
||
|
||
def test_multiple_apps_same_named_deployment_actors(serve_instance):
|
||
"""Multiple apps with same-named deployment actors are isolated."""
|
||
|
||
@serve.deployment(
|
||
deployment_actors=[
|
||
DeploymentActorConfig(
|
||
name="counter",
|
||
actor_class=SharedCounter,
|
||
init_kwargs={"start": 1},
|
||
),
|
||
],
|
||
)
|
||
class App1Deployment:
|
||
def __call__(self):
|
||
counter = serve.get_deployment_actor("counter")
|
||
return f"app1:{ray.get(counter.get.remote())}"
|
||
|
||
@serve.deployment(
|
||
deployment_actors=[
|
||
DeploymentActorConfig(
|
||
name="counter",
|
||
actor_class=SharedCounter,
|
||
init_kwargs={"start": 2},
|
||
),
|
||
],
|
||
)
|
||
class App2Deployment:
|
||
def __call__(self):
|
||
counter = serve.get_deployment_actor("counter")
|
||
return f"app2:{ray.get(counter.get.remote())}"
|
||
|
||
serve.run(App1Deployment.bind(), name="app1", route_prefix="/multi_app1")
|
||
serve.run(App2Deployment.bind(), name="app2", route_prefix="/multi_app2")
|
||
|
||
url1 = get_application_url("HTTP", "app1", use_localhost=True)
|
||
url2 = get_application_url("HTTP", "app2", use_localhost=True)
|
||
wait_for_condition(lambda: httpx.get(url1).text == "app1:1")
|
||
wait_for_condition(lambda: httpx.get(url2).text == "app2:2")
|
||
|
||
# Each app has its own deployment actor (isolated by app name)
|
||
app1_actors = _get_deployment_actor_names_for_app("app1", "App1Deployment")
|
||
app2_actors = _get_deployment_actor_names_for_app("app2", "App2Deployment")
|
||
assert len(app1_actors) == 1
|
||
assert len(app2_actors) == 1
|
||
assert app1_actors[0] != app2_actors[0]
|
||
|
||
|
||
def test_multi_app_deployment_actor_cleanup(serve_instance):
|
||
"""Deleting one app cleans only that app's deployment actors."""
|
||
|
||
@serve.deployment(
|
||
deployment_actors=[
|
||
DeploymentActorConfig(
|
||
name="counter",
|
||
actor_class=SharedCounter,
|
||
init_kwargs={"start": 1},
|
||
),
|
||
],
|
||
)
|
||
class App1Deployment:
|
||
def __call__(self):
|
||
counter = serve.get_deployment_actor("counter")
|
||
return str(ray.get(counter.get.remote()))
|
||
|
||
@serve.deployment(
|
||
deployment_actors=[
|
||
DeploymentActorConfig(
|
||
name="counter",
|
||
actor_class=SharedCounter,
|
||
init_kwargs={"start": 2},
|
||
),
|
||
],
|
||
)
|
||
class App2Deployment:
|
||
def __call__(self):
|
||
counter = serve.get_deployment_actor("counter")
|
||
return str(ray.get(counter.get.remote()))
|
||
|
||
serve.run(App1Deployment.bind(), name="app1", route_prefix="/cleanup1")
|
||
serve.run(App2Deployment.bind(), name="app2", route_prefix="/cleanup2")
|
||
|
||
wait_for_condition(
|
||
lambda: (
|
||
len(_get_deployment_actor_names_for_app("app1", "App1Deployment")) == 1
|
||
and len(_get_deployment_actor_names_for_app("app2", "App2Deployment")) == 1
|
||
)
|
||
)
|
||
serve.delete(name="app1")
|
||
wait_for_condition(
|
||
lambda: (
|
||
len(_get_deployment_actor_names_for_app("app1", "App1Deployment")) == 0
|
||
and len(_get_deployment_actor_names_for_app("app2", "App2Deployment")) == 1
|
||
),
|
||
timeout=15,
|
||
)
|
||
url2 = get_application_url("HTTP", "app2", use_localhost=True)
|
||
assert httpx.get(url2).text == "2"
|
||
|
||
|
||
def test_autoscaling_deployment_with_deployment_actors(serve_instance):
|
||
"""Autoscaling deployment with deployment actors: replicas share actors."""
|
||
|
||
@serve.deployment(
|
||
autoscaling_config=AutoscalingConfig(
|
||
min_replicas=1,
|
||
max_replicas=4,
|
||
target_ongoing_requests=1,
|
||
),
|
||
deployment_actors=[
|
||
DeploymentActorConfig(
|
||
name="counter",
|
||
actor_class=SharedCounter,
|
||
init_kwargs={"start": 0},
|
||
),
|
||
],
|
||
)
|
||
class MyDeployment:
|
||
def __call__(self):
|
||
counter = serve.get_deployment_actor("counter")
|
||
return str(ray.get(counter.increment.remote()))
|
||
|
||
serve.run(MyDeployment.bind())
|
||
url = f"{get_application_url()}/"
|
||
# Send enough requests to trigger scale-up
|
||
for _ in range(20):
|
||
httpx.get(url)
|
||
# All replicas share the same counter - values should be sequential
|
||
values = [int(httpx.get(url).text) for _ in range(5)]
|
||
assert values == sorted(values)
|
||
assert len(set(values)) == len(values)
|
||
|
||
|
||
def test_multiple_replicas_share_deployment_actor(serve_instance):
|
||
"""Multiple replicas share one deployment actor; state is consistent."""
|
||
|
||
@serve.deployment(
|
||
num_replicas=3,
|
||
deployment_actors=[
|
||
DeploymentActorConfig(
|
||
name="counter",
|
||
actor_class=SharedCounter,
|
||
init_kwargs={"start": 0},
|
||
),
|
||
],
|
||
)
|
||
class MyDeployment:
|
||
def __call__(self):
|
||
counter = serve.get_deployment_actor("counter")
|
||
return str(ray.get(counter.increment.remote()))
|
||
|
||
serve.run(MyDeployment.bind())
|
||
url = f"{get_application_url()}/"
|
||
values = set()
|
||
for _ in range(15):
|
||
values.add(httpx.get(url).text)
|
||
# All replicas share same counter; we should see sequential values
|
||
assert len(values) >= 2
|
||
sorted_vals = sorted(int(v) for v in values)
|
||
assert sorted_vals == list(range(min(sorted_vals), max(sorted_vals) + 1))
|
||
|
||
|
||
def test_concurrent_requests_multiple_replicas_to_deployment_actor(serve_instance):
|
||
"""Concurrent requests from multiple replicas to shared actor: no races."""
|
||
|
||
@serve.deployment(
|
||
num_replicas=2,
|
||
deployment_actors=[
|
||
DeploymentActorConfig(
|
||
name="counter",
|
||
actor_class=SharedCounter,
|
||
init_kwargs={"start": 0},
|
||
),
|
||
],
|
||
)
|
||
class MyDeployment:
|
||
def __call__(self):
|
||
counter = serve.get_deployment_actor("counter")
|
||
return str(ray.get(counter.increment.remote()))
|
||
|
||
serve.run(MyDeployment.bind())
|
||
url = f"{get_application_url()}/"
|
||
|
||
with concurrent.futures.ThreadPoolExecutor(max_workers=8) as ex:
|
||
futures = [ex.submit(lambda: httpx.get(url).text) for _ in range(20)]
|
||
results = [f.result() for f in futures]
|
||
# All increments should be unique (no lost updates)
|
||
assert len(results) == len(set(results))
|
||
|
||
|
||
def test_get_deployment_actor_invalid_name(serve_instance):
|
||
"""get_deployment_actor with nonexistent name causes request to fail."""
|
||
|
||
@serve.deployment(
|
||
deployment_actors=[
|
||
DeploymentActorConfig(
|
||
name="counter",
|
||
actor_class=SharedCounter,
|
||
init_kwargs={"start": 0},
|
||
),
|
||
],
|
||
)
|
||
class MyDeployment:
|
||
def __call__(self):
|
||
serve.get_deployment_actor("nonexistent")
|
||
return "ok"
|
||
|
||
serve.run(MyDeployment.bind())
|
||
url = f"{get_application_url()}/"
|
||
resp = httpx.get(url)
|
||
assert resp.status_code == 500
|
||
|
||
|
||
def test_get_deployment_actor_outside_replica_raises(serve_instance):
|
||
"""get_deployment_actor called outside replica raises RayServeException."""
|
||
with pytest.raises(RayServeException, match="may only be called from within"):
|
||
serve.get_deployment_actor("counter")
|
||
|
||
|
||
def test_get_deployment_actor_context_outside_actor_raises(serve_instance):
|
||
"""get_deployment_actor_context called outside actor raises RayServeException."""
|
||
with pytest.raises(RayServeException, match="may only be called from within"):
|
||
serve.get_deployment_actor_context()
|
||
|
||
|
||
def test_get_deployment_actor_context_returns_runtime_metadata(serve_instance):
|
||
"""Deployment actors can read deployment metadata via context API."""
|
||
|
||
@serve.deployment(
|
||
deployment_actors=[
|
||
DeploymentActorConfig(
|
||
name="ctx_actor",
|
||
actor_class=DeploymentActorContextActor,
|
||
),
|
||
],
|
||
)
|
||
class MyDeployment:
|
||
def __call__(self):
|
||
actor = serve.get_deployment_actor("ctx_actor")
|
||
return ray.get(actor.get_context.remote())
|
||
|
||
handle = serve.run(MyDeployment.bind(), name="metadata-app")
|
||
result = handle.remote().result()
|
||
|
||
assert result["deployment"] == "MyDeployment"
|
||
assert result["app_name"] == "metadata-app"
|
||
assert result["actor_name"] == "ctx_actor"
|
||
assert isinstance(result["code_version"], str)
|
||
|
||
|
||
def test_deploy_from_yaml_config_file_with_deployment_actors(serve_instance):
|
||
"""Deploy from YAML config file with deployment_actors."""
|
||
config_path = os.path.join(
|
||
os.path.dirname(__file__),
|
||
"test_config_files",
|
||
"deployment_actors.yaml",
|
||
)
|
||
with open(config_path) as f:
|
||
config = yaml.safe_load(f)
|
||
serve_instance.deploy_apps(ServeDeploySchema.model_validate(config))
|
||
wait_for_condition(check_running, timeout=30)
|
||
url = f"{get_application_url()}/"
|
||
wait_for_condition(lambda: httpx.get(url).text == "88", timeout=30)
|
||
wait_for_condition(
|
||
lambda: len(_get_deployment_actor_names_for_app("default", "ConfigOnlyDriver"))
|
||
== 1,
|
||
timeout=15,
|
||
)
|
||
|
||
|
||
def test_config_only_deployment_actors_with_actor_options(serve_instance):
|
||
"""Config-only deployment actors with actor_options in declarative config."""
|
||
config = {
|
||
"applications": [
|
||
{
|
||
"name": "default",
|
||
"route_prefix": "/",
|
||
"import_path": (
|
||
"ray.serve.tests.test_config_files.config_only_deployment_actor:app"
|
||
),
|
||
"deployments": [
|
||
{
|
||
"name": "ConfigOnlyDriver",
|
||
"deployment_actors": [
|
||
{
|
||
"name": "counter",
|
||
"actor_class": (
|
||
"ray.serve.tests.test_deployment_actors"
|
||
":SharedCounter"
|
||
),
|
||
"init_kwargs": {"start": 99},
|
||
"actor_options": {"num_cpus": 0.01},
|
||
},
|
||
],
|
||
},
|
||
],
|
||
},
|
||
],
|
||
}
|
||
serve_instance.deploy_apps(ServeDeploySchema.model_validate(config))
|
||
wait_for_condition(check_running, timeout=30)
|
||
url = f"{get_application_url()}/"
|
||
wait_for_condition(lambda: httpx.get(url).text == "99")
|
||
|
||
|
||
def test_gang_scheduling_with_deployment_actors(serve_instance):
|
||
"""Gang scheduling and deployment actors work together."""
|
||
|
||
@serve.deployment(
|
||
num_replicas=2,
|
||
gang_scheduling_config=GangSchedulingConfig(gang_size=2),
|
||
deployment_actors=[
|
||
DeploymentActorConfig(
|
||
name="counter",
|
||
actor_class=SharedCounter,
|
||
init_kwargs={"start": 0},
|
||
),
|
||
],
|
||
)
|
||
class MyDeployment:
|
||
def __call__(self):
|
||
counter = serve.get_deployment_actor("counter")
|
||
return str(ray.get(counter.increment.remote()))
|
||
|
||
serve.run(MyDeployment.bind())
|
||
url = f"{get_application_url()}/"
|
||
wait_for_condition(lambda: httpx.get(url).status_code == 200)
|
||
vals = [httpx.get(url).text for _ in range(5)]
|
||
assert all(v.isdigit() for v in vals)
|
||
|
||
|
||
def test_user_config_update_with_deployment_actors(serve_instance):
|
||
"""user_config update does not affect deployment actors."""
|
||
|
||
@serve.deployment(
|
||
deployment_actors=[
|
||
DeploymentActorConfig(
|
||
name="counter",
|
||
actor_class=SharedCounter,
|
||
init_kwargs={"start": 50},
|
||
),
|
||
],
|
||
)
|
||
class MyDeployment:
|
||
def __init__(self):
|
||
self.config_val = None
|
||
|
||
def reconfigure(self, config):
|
||
self.config_val = config
|
||
|
||
def __call__(self):
|
||
counter = serve.get_deployment_actor("counter")
|
||
return f"{self.config_val}:{ray.get(counter.get.remote())}"
|
||
|
||
serve.run(MyDeployment.options(user_config=None).bind())
|
||
url = f"{get_application_url()}/"
|
||
wait_for_condition(lambda: "None:50" in httpx.get(url).text)
|
||
|
||
serve.run(MyDeployment.options(user_config="updated").bind(), blocking=False)
|
||
wait_for_condition(lambda: "updated:50" in httpx.get(url).text, timeout=30)
|
||
|
||
|
||
def test_redeployment_adds_actors_to_existing_deployment(serve_instance):
|
||
"""Redeploying adds deployment actors to a deployment that previously had none.
|
||
|
||
Uses the declarative path (deploy_apps) so the controller diffs old vs.
|
||
new config and exercises the actor-creation branch, unlike serve.run()
|
||
which always creates a fresh deployment.
|
||
"""
|
||
client = serve_instance
|
||
|
||
config_v1 = {
|
||
"applications": [
|
||
{
|
||
"name": "default",
|
||
"route_prefix": "/",
|
||
"import_path": (
|
||
"ray.serve.tests.test_config_files.flex_deployment_actor:app"
|
||
),
|
||
"deployments": [{"name": "FlexDriver"}],
|
||
},
|
||
],
|
||
}
|
||
|
||
client.deploy_apps(ServeDeploySchema.model_validate(config_v1))
|
||
wait_for_condition(check_running, timeout=30)
|
||
url = f"{get_application_url()}/"
|
||
wait_for_condition(lambda: httpx.get(url).text == "no_actor")
|
||
assert _check_deployment_actor_count(0)
|
||
|
||
config_v2 = {
|
||
"applications": [
|
||
{
|
||
"name": "default",
|
||
"route_prefix": "/",
|
||
"import_path": (
|
||
"ray.serve.tests.test_config_files.flex_deployment_actor:app"
|
||
),
|
||
"deployments": [
|
||
{
|
||
"name": "FlexDriver",
|
||
"deployment_actors": [
|
||
{
|
||
"name": "counter",
|
||
"actor_class": (
|
||
"ray.serve.tests.test_deployment_actors"
|
||
":SharedCounter"
|
||
),
|
||
"init_kwargs": {"start": 42},
|
||
},
|
||
],
|
||
},
|
||
],
|
||
},
|
||
],
|
||
}
|
||
|
||
client.deploy_apps(ServeDeploySchema.model_validate(config_v2))
|
||
wait_for_condition(check_running, timeout=30)
|
||
wait_for_condition(lambda: httpx.get(url).text == "42")
|
||
wait_for_condition(lambda: _check_deployment_actor_count(1))
|
||
|
||
|
||
def test_redeployment_changes_actor_class(serve_instance):
|
||
"""Redeploying with a different actor class (same logical name) replaces the actor.
|
||
|
||
'Actor class changed → Kill old, create new'.
|
||
Uses the declarative path so the controller diffs old vs. new actor class.
|
||
SharedCounter.get() returns start; AltCounter.get() returns start * 10.
|
||
"""
|
||
client = serve_instance
|
||
|
||
config_v1 = {
|
||
"applications": [
|
||
{
|
||
"name": "default",
|
||
"route_prefix": "/",
|
||
"import_path": (
|
||
"ray.serve.tests.test_config_files.flex_deployment_actor:app"
|
||
),
|
||
"deployments": [
|
||
{
|
||
"name": "FlexDriver",
|
||
"deployment_actors": [
|
||
{
|
||
"name": "counter",
|
||
"actor_class": (
|
||
"ray.serve.tests.test_deployment_actors"
|
||
":SharedCounter"
|
||
),
|
||
"init_kwargs": {"start": 10},
|
||
},
|
||
],
|
||
},
|
||
],
|
||
},
|
||
],
|
||
}
|
||
|
||
client.deploy_apps(ServeDeploySchema.model_validate(config_v1))
|
||
wait_for_condition(check_running, timeout=30)
|
||
url = f"{get_application_url()}/"
|
||
wait_for_condition(lambda: httpx.get(url).text == "10")
|
||
v1_actor_names = _get_deployment_actor_names()
|
||
assert len(v1_actor_names) == 1
|
||
|
||
config_v2 = {
|
||
"applications": [
|
||
{
|
||
"name": "default",
|
||
"route_prefix": "/",
|
||
"import_path": (
|
||
"ray.serve.tests.test_config_files.flex_deployment_actor:app"
|
||
),
|
||
"deployments": [
|
||
{
|
||
"name": "FlexDriver",
|
||
"deployment_actors": [
|
||
{
|
||
"name": "counter",
|
||
"actor_class": (
|
||
"ray.serve.tests.test_deployment_actors"
|
||
":AltCounter"
|
||
),
|
||
"init_kwargs": {"start": 10},
|
||
},
|
||
],
|
||
},
|
||
],
|
||
},
|
||
],
|
||
}
|
||
|
||
client.deploy_apps(ServeDeploySchema.model_validate(config_v2))
|
||
wait_for_condition(check_running, timeout=30)
|
||
# AltCounter.get() returns start * 10 = 100, proving the class changed
|
||
wait_for_condition(lambda: httpx.get(url).text == "100", timeout=10)
|
||
v2_actor_names = _get_deployment_actor_names()
|
||
assert len(v2_actor_names) == 1
|
||
assert v1_actor_names[0] != v2_actor_names[0]
|
||
|
||
|
||
def test_redeployment_same_name_changed_init_kwargs_resets_state(serve_instance):
|
||
"""Changed init_kwargs with the same actor name kills and recreates the actor.
|
||
|
||
'init_args/kwargs changed → Kill and recreate
|
||
(state loss expected)'. Uses the declarative path so the controller diffs
|
||
old vs. new init_kwargs. Mutates actor state before redeploy and verifies
|
||
the accumulated state is discarded.
|
||
"""
|
||
client = serve_instance
|
||
|
||
config_v1 = {
|
||
"applications": [
|
||
{
|
||
"name": "default",
|
||
"route_prefix": "/",
|
||
"import_path": (
|
||
"ray.serve.tests.test_config_files"
|
||
".config_only_deployment_actor:app"
|
||
),
|
||
"deployments": [
|
||
{
|
||
"name": "ConfigOnlyDriver",
|
||
"deployment_actors": [
|
||
{
|
||
"name": "counter",
|
||
"actor_class": (
|
||
"ray.serve.tests.test_deployment_actors"
|
||
":SharedCounter"
|
||
),
|
||
"init_kwargs": {"start": 0},
|
||
},
|
||
],
|
||
},
|
||
],
|
||
},
|
||
],
|
||
}
|
||
|
||
client.deploy_apps(ServeDeploySchema.model_validate(config_v1))
|
||
wait_for_condition(check_running, timeout=30)
|
||
url = f"{get_application_url()}/"
|
||
wait_for_condition(lambda: httpx.get(url).text == "0")
|
||
|
||
# Mutate actor state directly so it diverges from init value
|
||
actor_names = _get_deployment_actor_names()
|
||
assert len(actor_names) == 1
|
||
handle = ray.get_actor(actor_names[0], namespace=SERVE_NAMESPACE)
|
||
for _ in range(5):
|
||
ray.get(handle.increment.remote())
|
||
wait_for_condition(lambda: httpx.get(url).text == "5")
|
||
|
||
config_v2 = {
|
||
"applications": [
|
||
{
|
||
"name": "default",
|
||
"route_prefix": "/",
|
||
"import_path": (
|
||
"ray.serve.tests.test_config_files"
|
||
".config_only_deployment_actor:app"
|
||
),
|
||
"deployments": [
|
||
{
|
||
"name": "ConfigOnlyDriver",
|
||
"deployment_actors": [
|
||
{
|
||
"name": "counter",
|
||
"actor_class": (
|
||
"ray.serve.tests.test_deployment_actors"
|
||
":SharedCounter"
|
||
),
|
||
"init_kwargs": {"start": 100},
|
||
},
|
||
],
|
||
},
|
||
],
|
||
},
|
||
],
|
||
}
|
||
|
||
client.deploy_apps(ServeDeploySchema.model_validate(config_v2))
|
||
wait_for_condition(check_running, timeout=30)
|
||
# State reset: returns 100 (new init value), not 5 (old mutated state)
|
||
wait_for_condition(lambda: httpx.get(url).text == "100", timeout=30)
|
||
|
||
|
||
def test_partial_actor_creation_failure(serve_instance):
|
||
"""When one of multiple deployment actors fails, the deployment fails.
|
||
|
||
A deployment with two actors — one healthy, one whose constructor raises —
|
||
should transition to DEPLOY_FAILED, not partially succeed.
|
||
"""
|
||
|
||
@serve.deployment(
|
||
max_constructor_retry_count=3,
|
||
deployment_actors=[
|
||
DeploymentActorConfig(
|
||
name="good_counter",
|
||
actor_class=SharedCounter,
|
||
init_kwargs={"start": 0},
|
||
),
|
||
DeploymentActorConfig(
|
||
name="bad_actor",
|
||
actor_class=FailingDeploymentActor,
|
||
init_kwargs={"should_fail": True},
|
||
),
|
||
],
|
||
)
|
||
class MyDeployment:
|
||
def __call__(self):
|
||
return "ok"
|
||
|
||
serve._run(
|
||
MyDeployment.bind(),
|
||
name="da_partial_fail_app",
|
||
route_prefix="/da_partial_fail",
|
||
_blocking=False,
|
||
)
|
||
|
||
wait_for_condition(
|
||
lambda: any(
|
||
"partial_fail" in n and a.status == ApplicationStatus.DEPLOY_FAILED
|
||
for n, a in serve.status().applications.items()
|
||
),
|
||
timeout=60,
|
||
)
|
||
|
||
status = serve.status()
|
||
app_status = next(a for n, a in status.applications.items() if "partial_fail" in n)
|
||
assert app_status.status == ApplicationStatus.DEPLOY_FAILED
|
||
|
||
serve.delete(
|
||
next(n for n in status.applications if "partial_fail" in n),
|
||
_blocking=True,
|
||
)
|
||
|
||
|
||
def test_deployment_actors_outlive_replicas_during_deletion(serve_instance):
|
||
"""Deployment actors remain alive while replicas are still stopping.
|
||
|
||
Deployment actors are killed only after all replicas have stopped. This ensures replicas can still reach their deployment
|
||
actors during graceful shutdown / request draining.
|
||
"""
|
||
signal = SignalActor.remote()
|
||
|
||
@serve.deployment(
|
||
name="OutliveTest",
|
||
graceful_shutdown_timeout_s=120,
|
||
deployment_actors=[
|
||
DeploymentActorConfig(
|
||
name="counter",
|
||
actor_class=SharedCounter,
|
||
init_kwargs={"start": 0},
|
||
),
|
||
],
|
||
)
|
||
class SlowDrainDeployment:
|
||
async def __call__(self):
|
||
await signal.wait.remote()
|
||
counter = serve.get_deployment_actor("counter")
|
||
return str(ray.get(counter.get.remote()))
|
||
|
||
h = serve.run(
|
||
SlowDrainDeployment.bind(),
|
||
name="app",
|
||
route_prefix="/outlive_replicas_da",
|
||
)
|
||
wait_for_condition(lambda: _check_deployment_actor_count(1))
|
||
|
||
# Send a request that blocks — keeps replica busy during deletion
|
||
_ = h.remote()
|
||
|
||
serve.delete(name="app", _blocking=False)
|
||
|
||
# Replica should enter STOPPING while draining the in-flight request
|
||
wait_for_condition(
|
||
lambda: check_replica_counts(
|
||
controller=serve_instance._controller,
|
||
deployment_id=DeploymentID(name="OutliveTest", app_name="app"),
|
||
total=1,
|
||
by_state=[(ReplicaState.STOPPING, 1, None)],
|
||
),
|
||
timeout=15,
|
||
)
|
||
|
||
# Deployment actor must still be alive while replica is stopping
|
||
assert _check_deployment_actor_count(1)
|
||
|
||
# Release the blocked request so replica can finish draining
|
||
ray.get(signal.send.remote())
|
||
|
||
# After replica stops, deployment actor should be cleaned up
|
||
wait_for_condition(_check_no_deployment_actors, timeout=30)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(pytest.main(["-v", "-s", __file__]))
|