wehub-resource-sync
1166 lines
41 KiB
Python
1166 lines
41 KiB
Python
import asyncio
|
|
import sys
|
|
from copy import deepcopy
|
|
from typing import Dict, Optional
|
|
|
|
import pytest
|
|
from pydantic import BaseModel
|
|
|
|
import ray
|
|
from ray import serve
|
|
from ray._common.test_utils import SignalActor, wait_for_condition
|
|
from ray.exceptions import RayActorError
|
|
from ray.serve import Application
|
|
from ray.serve._private.client import ServeControllerClient
|
|
from ray.serve._private.common import (
|
|
DeploymentID,
|
|
DeploymentStatus,
|
|
DeploymentStatusTrigger,
|
|
ReplicaState,
|
|
TargetCapacityDirection,
|
|
)
|
|
from ray.serve._private.constants import SERVE_DEFAULT_APP_NAME, SERVE_NAMESPACE
|
|
from ray.serve.config import AutoscalingConfig
|
|
from ray.serve.context import _get_global_client
|
|
from ray.serve.schema import (
|
|
ApplicationStatus,
|
|
ServeApplicationSchema,
|
|
ServeDeploySchema,
|
|
)
|
|
|
|
INGRESS_DEPLOYMENT_NAME = "ingress"
|
|
INGRESS_DEPLOYMENT_NUM_REPLICAS = 6
|
|
DOWNSTREAM_DEPLOYMENT_NAME = "downstream"
|
|
DOWNSTREAM_DEPLOYMENT_NUM_REPLICAS = 2
|
|
SCALE_TO_ZERO_DEPLOYMENT_NAME = "zero"
|
|
SCALE_TO_ZERO_DEPLOYMENT_MAX_REPLICAS = 4
|
|
START_AT_10_DEPLOYMENT_NAME = "start_at_ten"
|
|
START_AT_10_DEPLOYMENT_INITIAL_REPLICAS = 10
|
|
START_AT_10_DEPLOYMENT_MIN_REPLICAS = 0
|
|
START_AT_10_DEPLOYMENT_MAX_REPLICAS = 20
|
|
|
|
|
|
@pytest.fixture
|
|
def shutdown_ray_and_serve():
|
|
serve.shutdown()
|
|
if ray.is_initialized():
|
|
# wait_for_processes=True blocks until the raylet/GCS/etc. subprocesses
|
|
# have fully exited, so the next test's serve.start() (which calls
|
|
# ray.init()) doesn't race a still-terminating raylet.
|
|
ray.shutdown(wait_for_processes=True)
|
|
yield
|
|
serve.shutdown()
|
|
if ray.is_initialized():
|
|
ray.shutdown(wait_for_processes=True)
|
|
|
|
|
|
@serve.deployment(ray_actor_options={"num_cpus": 0})
|
|
class DummyDeployment:
|
|
def __init__(self, *args):
|
|
pass
|
|
|
|
def __call__(self, *args):
|
|
return "hi"
|
|
|
|
|
|
test_app = DummyDeployment.options(
|
|
name=INGRESS_DEPLOYMENT_NAME,
|
|
num_replicas=INGRESS_DEPLOYMENT_NUM_REPLICAS,
|
|
).bind(
|
|
DummyDeployment.options(
|
|
name=DOWNSTREAM_DEPLOYMENT_NAME,
|
|
num_replicas=DOWNSTREAM_DEPLOYMENT_NUM_REPLICAS,
|
|
).bind()
|
|
)
|
|
|
|
|
|
def check_expected_num_replicas(
|
|
deployment_to_num_replicas: Dict[str, int], app_name: str = SERVE_DEFAULT_APP_NAME
|
|
) -> bool:
|
|
"""Checks that the expected number of replicas for each deployment is running.
|
|
|
|
Looks for the deployments in the app named app_name.
|
|
"""
|
|
|
|
status = serve.status()
|
|
assert app_name in status.applications
|
|
|
|
application = status.applications[app_name]
|
|
assert application.status == ApplicationStatus.RUNNING
|
|
|
|
for name, num_replicas in deployment_to_num_replicas.items():
|
|
assert name in application.deployments
|
|
assert (
|
|
sum(application.deployments[name].replica_states.values()) == num_replicas
|
|
)
|
|
|
|
return True
|
|
|
|
|
|
@pytest.fixture
|
|
def client() -> ServeControllerClient:
|
|
serve.start()
|
|
yield _get_global_client()
|
|
|
|
|
|
def test_incremental_scale_up(shutdown_ray_and_serve, client: ServeControllerClient):
|
|
config = ServeDeploySchema(
|
|
applications=[
|
|
ServeApplicationSchema(
|
|
import_path="ray.serve.tests.test_target_capacity:test_app"
|
|
)
|
|
]
|
|
)
|
|
|
|
# Initially deploy at target_capacity 0, should have no replicas.
|
|
config.target_capacity = 0.0
|
|
client.deploy_apps(config)
|
|
wait_for_condition(lambda: serve.status().target_capacity == 0.0)
|
|
wait_for_condition(
|
|
check_expected_num_replicas,
|
|
deployment_to_num_replicas={
|
|
INGRESS_DEPLOYMENT_NAME: 0,
|
|
DOWNSTREAM_DEPLOYMENT_NAME: 0,
|
|
},
|
|
timeout=30,
|
|
)
|
|
|
|
# Initially deploy at target_capacity 1, should have 1 replica of each.
|
|
config.target_capacity = 1.0
|
|
client.deploy_apps(config)
|
|
wait_for_condition(lambda: serve.status().target_capacity == 1.0)
|
|
wait_for_condition(
|
|
check_expected_num_replicas,
|
|
deployment_to_num_replicas={
|
|
INGRESS_DEPLOYMENT_NAME: 1,
|
|
DOWNSTREAM_DEPLOYMENT_NAME: 1,
|
|
},
|
|
timeout=30,
|
|
)
|
|
|
|
# Increase target_capacity to 50, ingress deployment should scale up.
|
|
config.target_capacity = 50.0
|
|
client.deploy_apps(config)
|
|
wait_for_condition(lambda: serve.status().target_capacity == 50.0)
|
|
wait_for_condition(
|
|
check_expected_num_replicas,
|
|
deployment_to_num_replicas={
|
|
INGRESS_DEPLOYMENT_NAME: INGRESS_DEPLOYMENT_NUM_REPLICAS / 2,
|
|
DOWNSTREAM_DEPLOYMENT_NAME: DOWNSTREAM_DEPLOYMENT_NUM_REPLICAS / 2,
|
|
},
|
|
timeout=30,
|
|
)
|
|
|
|
# Increase target_capacity to 100, both should fully scale up.
|
|
config.target_capacity = 100.0
|
|
client.deploy_apps(config)
|
|
wait_for_condition(lambda: serve.status().target_capacity == 100.0)
|
|
wait_for_condition(
|
|
check_expected_num_replicas,
|
|
deployment_to_num_replicas={
|
|
INGRESS_DEPLOYMENT_NAME: INGRESS_DEPLOYMENT_NUM_REPLICAS,
|
|
DOWNSTREAM_DEPLOYMENT_NAME: DOWNSTREAM_DEPLOYMENT_NUM_REPLICAS,
|
|
},
|
|
timeout=30,
|
|
)
|
|
|
|
# Finish rollout (remove target_capacity), should have no effect.
|
|
config.target_capacity = None
|
|
client.deploy_apps(config)
|
|
wait_for_condition(lambda: serve.status().target_capacity is None)
|
|
wait_for_condition(
|
|
check_expected_num_replicas,
|
|
deployment_to_num_replicas={
|
|
INGRESS_DEPLOYMENT_NAME: INGRESS_DEPLOYMENT_NUM_REPLICAS,
|
|
DOWNSTREAM_DEPLOYMENT_NAME: DOWNSTREAM_DEPLOYMENT_NUM_REPLICAS,
|
|
},
|
|
timeout=30,
|
|
)
|
|
|
|
|
|
def test_incremental_scale_down(shutdown_ray_and_serve, client: ServeControllerClient):
|
|
config = ServeDeploySchema(
|
|
applications=[
|
|
ServeApplicationSchema(
|
|
import_path="ray.serve.tests.test_target_capacity:test_app"
|
|
)
|
|
]
|
|
)
|
|
|
|
# Initially deploy with no target_capacity (full scale).
|
|
client.deploy_apps(config)
|
|
wait_for_condition(lambda: serve.status().target_capacity is None)
|
|
wait_for_condition(
|
|
check_expected_num_replicas,
|
|
deployment_to_num_replicas={
|
|
INGRESS_DEPLOYMENT_NAME: INGRESS_DEPLOYMENT_NUM_REPLICAS,
|
|
DOWNSTREAM_DEPLOYMENT_NAME: DOWNSTREAM_DEPLOYMENT_NUM_REPLICAS,
|
|
},
|
|
timeout=30,
|
|
)
|
|
|
|
# Decrease target_capacity to 50, both deployments should scale down.
|
|
config.target_capacity = 50.0
|
|
client.deploy_apps(config)
|
|
wait_for_condition(lambda: serve.status().target_capacity == 50.0)
|
|
wait_for_condition(
|
|
check_expected_num_replicas,
|
|
deployment_to_num_replicas={
|
|
INGRESS_DEPLOYMENT_NAME: INGRESS_DEPLOYMENT_NUM_REPLICAS / 2,
|
|
DOWNSTREAM_DEPLOYMENT_NAME: DOWNSTREAM_DEPLOYMENT_NUM_REPLICAS / 2,
|
|
},
|
|
timeout=30,
|
|
)
|
|
|
|
# Decrease target_capacity to 1, both should fully scale down.
|
|
config.target_capacity = 1.0
|
|
client.deploy_apps(config)
|
|
wait_for_condition(lambda: serve.status().target_capacity == 1.0)
|
|
wait_for_condition(
|
|
check_expected_num_replicas,
|
|
deployment_to_num_replicas={
|
|
INGRESS_DEPLOYMENT_NAME: 1,
|
|
DOWNSTREAM_DEPLOYMENT_NAME: 1,
|
|
},
|
|
timeout=30,
|
|
)
|
|
|
|
# Decrease target_capacity to 0, both should fully scale down to zero.
|
|
config.target_capacity = 0.0
|
|
client.deploy_apps(config)
|
|
wait_for_condition(lambda: serve.status().target_capacity == 0.0)
|
|
wait_for_condition(
|
|
check_expected_num_replicas,
|
|
deployment_to_num_replicas={
|
|
INGRESS_DEPLOYMENT_NAME: 0,
|
|
DOWNSTREAM_DEPLOYMENT_NAME: 0,
|
|
},
|
|
timeout=30,
|
|
)
|
|
|
|
|
|
def test_controller_recover_target_capacity(
|
|
shutdown_ray_and_serve, client: ServeControllerClient
|
|
):
|
|
config = ServeDeploySchema(
|
|
target_capacity=100,
|
|
applications=[
|
|
ServeApplicationSchema(
|
|
import_path="ray.serve.tests.test_target_capacity:test_app"
|
|
)
|
|
],
|
|
)
|
|
|
|
client.deploy_apps(config)
|
|
wait_for_condition(lambda: serve.status().target_capacity == 100.0)
|
|
assert (
|
|
ray.get(client._controller._get_target_capacity_direction.remote())
|
|
== TargetCapacityDirection.UP
|
|
)
|
|
|
|
# Scale down to target_capacity 50, both deployments should be at half scale.
|
|
config.target_capacity = 50.0
|
|
client.deploy_apps(config)
|
|
wait_for_condition(lambda: serve.status().target_capacity == 50.0)
|
|
wait_for_condition(
|
|
check_expected_num_replicas,
|
|
deployment_to_num_replicas={
|
|
INGRESS_DEPLOYMENT_NAME: INGRESS_DEPLOYMENT_NUM_REPLICAS / 2,
|
|
DOWNSTREAM_DEPLOYMENT_NAME: DOWNSTREAM_DEPLOYMENT_NUM_REPLICAS / 2,
|
|
},
|
|
timeout=30,
|
|
)
|
|
assert (
|
|
ray.get(client._controller._get_target_capacity_direction.remote())
|
|
== TargetCapacityDirection.DOWN
|
|
)
|
|
|
|
ray.kill(client._controller, no_restart=False)
|
|
|
|
# Verify that the target_capacity is recovered after the controller comes back.
|
|
wait_for_condition(lambda: serve.status().target_capacity == 50.0)
|
|
wait_for_condition(
|
|
check_expected_num_replicas,
|
|
deployment_to_num_replicas={
|
|
INGRESS_DEPLOYMENT_NAME: INGRESS_DEPLOYMENT_NUM_REPLICAS / 2,
|
|
DOWNSTREAM_DEPLOYMENT_NAME: DOWNSTREAM_DEPLOYMENT_NUM_REPLICAS / 2,
|
|
},
|
|
timeout=30,
|
|
)
|
|
assert (
|
|
ray.get(client._controller._get_target_capacity_direction.remote())
|
|
== TargetCapacityDirection.DOWN
|
|
)
|
|
|
|
|
|
@serve.deployment(
|
|
ray_actor_options={"num_cpus": 0},
|
|
autoscaling_config={
|
|
"min_replicas": 0,
|
|
"max_replicas": SCALE_TO_ZERO_DEPLOYMENT_MAX_REPLICAS,
|
|
"target_ongoing_requests": 1,
|
|
"upscale_delay_s": 1,
|
|
"downscale_delay_s": 1,
|
|
"upscaling_factor": 4,
|
|
"downscaling_factor": 4,
|
|
"metrics_interval_s": 1,
|
|
# The default look_back_period_s is 30, which means the test assertions will be
|
|
# slow to respond to changes in metrics. Setting it to 2 makes the test assertions
|
|
# more responsive to changes in metrics, hence reducing flakiness.
|
|
"look_back_period_s": 2,
|
|
},
|
|
max_ongoing_requests=2,
|
|
graceful_shutdown_timeout_s=0,
|
|
)
|
|
class ScaleToZeroDeployment:
|
|
async def __call__(self, *args):
|
|
await asyncio.sleep(10000)
|
|
|
|
|
|
scale_to_zero_app = ScaleToZeroDeployment.options(
|
|
name=SCALE_TO_ZERO_DEPLOYMENT_NAME,
|
|
).bind()
|
|
|
|
|
|
def test_autoscaling_scale_to_zero(
|
|
shutdown_ray_and_serve, client: ServeControllerClient
|
|
):
|
|
config = ServeDeploySchema(
|
|
applications=[
|
|
ServeApplicationSchema(
|
|
import_path="ray.serve.tests.test_target_capacity:scale_to_zero_app"
|
|
)
|
|
]
|
|
)
|
|
|
|
# Initially deploy at target_capacity 0, should have 0 replicas.
|
|
config.target_capacity = 0.0
|
|
client.deploy_apps(config)
|
|
wait_for_condition(lambda: serve.status().target_capacity == 0.0)
|
|
wait_for_condition(
|
|
check_expected_num_replicas,
|
|
deployment_to_num_replicas={
|
|
SCALE_TO_ZERO_DEPLOYMENT_NAME: 0,
|
|
},
|
|
)
|
|
|
|
# Increase to target_capacity 1, should remain at 0 replicas initially.
|
|
config.target_capacity = 1.0
|
|
client.deploy_apps(config)
|
|
wait_for_condition(lambda: serve.status().target_capacity == 1.0)
|
|
wait_for_condition(
|
|
check_expected_num_replicas,
|
|
deployment_to_num_replicas={
|
|
SCALE_TO_ZERO_DEPLOYMENT_NAME: 0,
|
|
},
|
|
)
|
|
|
|
# Send a bunch of requests. Autoscaler will want to scale it up to max replicas,
|
|
# but it should stay at one due to the target_capacity.
|
|
handle = serve.get_app_handle(SERVE_DEFAULT_APP_NAME)
|
|
responses = [
|
|
handle.remote() for _ in range(5 * SCALE_TO_ZERO_DEPLOYMENT_MAX_REPLICAS)
|
|
]
|
|
wait_for_condition(
|
|
check_expected_num_replicas,
|
|
deployment_to_num_replicas={
|
|
SCALE_TO_ZERO_DEPLOYMENT_NAME: 1,
|
|
},
|
|
timeout=30,
|
|
)
|
|
|
|
# Increase to target_capacity 100, should scale all the way up.
|
|
config.target_capacity = 100.0
|
|
client.deploy_apps(config)
|
|
wait_for_condition(lambda: serve.status().target_capacity == 100.0)
|
|
wait_for_condition(
|
|
check_expected_num_replicas,
|
|
deployment_to_num_replicas={
|
|
SCALE_TO_ZERO_DEPLOYMENT_NAME: SCALE_TO_ZERO_DEPLOYMENT_MAX_REPLICAS,
|
|
},
|
|
timeout=30,
|
|
)
|
|
|
|
# Decrease to target_capacity 50, should scale down.
|
|
config.target_capacity = 50.0
|
|
client.deploy_apps(config)
|
|
wait_for_condition(lambda: serve.status().target_capacity == 50.0)
|
|
wait_for_condition(
|
|
check_expected_num_replicas,
|
|
deployment_to_num_replicas={
|
|
SCALE_TO_ZERO_DEPLOYMENT_NAME: SCALE_TO_ZERO_DEPLOYMENT_MAX_REPLICAS / 2,
|
|
},
|
|
timeout=30,
|
|
)
|
|
|
|
# Cancel all of the requests, should scale down to zero.
|
|
[r.cancel() for r in responses]
|
|
wait_for_condition(
|
|
check_expected_num_replicas,
|
|
deployment_to_num_replicas={
|
|
SCALE_TO_ZERO_DEPLOYMENT_NAME: 0,
|
|
},
|
|
timeout=30,
|
|
)
|
|
|
|
|
|
@serve.deployment(ray_actor_options={"num_cpus": 0})
|
|
class ControlledLifecycleDeployment:
|
|
async def __init__(self):
|
|
self.lifecycle_signal = ray.get_actor(
|
|
name="lifecycle_signal", namespace=SERVE_NAMESPACE
|
|
)
|
|
await self.lifecycle_signal.wait.remote()
|
|
|
|
try:
|
|
self.request_signal = ray.get_actor(
|
|
name="request_signal", namespace=SERVE_NAMESPACE
|
|
)
|
|
except ValueError:
|
|
# No request signal actor was launched.
|
|
self.request_signal = None
|
|
|
|
async def __call__(self) -> str:
|
|
if self.request_signal is not None:
|
|
try:
|
|
await self.request_signal.wait.remote()
|
|
except RayActorError:
|
|
print("Request signal actor already dead. Running request.")
|
|
return "Hello world!"
|
|
|
|
async def __del__(self):
|
|
# Use try-except here to avoid cluttering logs if the SignalActor
|
|
# gets torn down before the replicas during clean up.
|
|
try:
|
|
await self.lifecycle_signal.wait.remote()
|
|
except RayActorError:
|
|
print("Lifecycle signal actor already dead. Replica exiting.")
|
|
|
|
|
|
class ControllerAppConfig(BaseModel):
|
|
num_replicas: int
|
|
|
|
|
|
def create_controlled_app(config: ControllerAppConfig) -> Application:
|
|
num_replicas = config.num_replicas
|
|
return ControlledLifecycleDeployment.options(
|
|
name="controlled",
|
|
num_replicas=num_replicas,
|
|
).bind()
|
|
|
|
|
|
class AutoscalingControllerAppConfig(BaseModel):
|
|
min_replicas: int
|
|
initial_replicas: Optional[int] = None
|
|
max_replicas: int
|
|
|
|
|
|
def create_autoscaling_controlled_app(
|
|
config: AutoscalingControllerAppConfig,
|
|
) -> Application:
|
|
min_replicas = config.min_replicas
|
|
initial_replicas = config.initial_replicas
|
|
max_replicas = config.max_replicas
|
|
return ControlledLifecycleDeployment.options(
|
|
name="controlled",
|
|
autoscaling_config=AutoscalingConfig(
|
|
min_replicas=min_replicas,
|
|
initial_replicas=initial_replicas,
|
|
max_replicas=max_replicas,
|
|
target_ongoing_requests=1,
|
|
metrics_interval_s=0.1,
|
|
look_back_period_s=0.2,
|
|
upscale_delay_s=0.1,
|
|
downscale_delay_s=0.1,
|
|
),
|
|
graceful_shutdown_timeout_s=0,
|
|
).bind()
|
|
|
|
|
|
class TestTargetCapacityUpdateAndServeStatus:
|
|
def check_num_replicas(
|
|
self,
|
|
expected_num_replicas: int,
|
|
app_name: str,
|
|
deployment_name: str,
|
|
replica_state: ReplicaState = ReplicaState.RUNNING,
|
|
controller_handle: Optional[ray.actor.ActorHandle] = None,
|
|
) -> bool:
|
|
"""Checks that the number of replicas are as expected.
|
|
|
|
Args:
|
|
expected_num_replicas: the expected number of replicas.
|
|
app_name: the deployment's application name.
|
|
deployment_name: the deployment's name.
|
|
replica_state: only replicas in this state are counted.
|
|
controller_handle: this is an optional argument. If provided, the
|
|
controller handle is used to get the current autoscaling
|
|
metrics and print them if the assertion fails.
|
|
|
|
Returns:
|
|
True when the replica count matches (raises ``AssertionError`` otherwise).
|
|
"""
|
|
deployment = serve.status().applications[app_name].deployments[deployment_name]
|
|
num_running_replicas = deployment.replica_states.get(replica_state, 0)
|
|
if controller_handle is None:
|
|
assert num_running_replicas == expected_num_replicas, f"{deployment}"
|
|
else:
|
|
deployment_id = DeploymentID(name=deployment_name, app_name=app_name)
|
|
autoscaling_metrics = ray.get(
|
|
controller_handle._get_metrics_for_deployment_for_testing.remote(
|
|
deployment_id
|
|
)
|
|
)
|
|
assert num_running_replicas == expected_num_replicas, (
|
|
f"Status: {deployment}" f"\nAutoscaling metrics: {autoscaling_metrics}"
|
|
)
|
|
return True
|
|
|
|
def apply_config_and_check_status(
|
|
self,
|
|
client: ServeControllerClient,
|
|
target_capacity: Optional[float],
|
|
config: ServeDeploySchema,
|
|
app_name: str,
|
|
deployment_name: str,
|
|
expected_app_status: Optional[ApplicationStatus] = None,
|
|
expected_deployment_status: Optional[DeploymentStatus] = None,
|
|
expected_deployment_status_trigger: Optional[DeploymentStatusTrigger] = None,
|
|
timeout=10,
|
|
):
|
|
"""Applies config with specified target_capacity."""
|
|
|
|
config = deepcopy(config)
|
|
config.target_capacity = target_capacity
|
|
client.deploy_apps(config)
|
|
|
|
def check():
|
|
status = serve.status()
|
|
assert status.target_capacity == target_capacity
|
|
|
|
if expected_app_status is not None:
|
|
assert status.applications[app_name].status == expected_app_status
|
|
|
|
dep_status = status.applications[app_name].deployments[deployment_name]
|
|
if expected_deployment_status is not None:
|
|
assert dep_status.status == expected_deployment_status
|
|
|
|
if expected_deployment_status_trigger is not None:
|
|
assert dep_status.status_trigger == expected_deployment_status_trigger
|
|
|
|
return True
|
|
|
|
wait_for_condition(check, timeout=timeout)
|
|
|
|
def unblock_replica_creation_and_deletion(self, lifecycle_signal, app_name: str):
|
|
"""Unblocks creating and deleting ControlledLifecycleDeployment replicas.
|
|
|
|
These replicas can't initialize or be deleted until the
|
|
"lifecycle_signal" actor runs send, so this method runs send, waits
|
|
until the replicas start or stop running, and then resets the signal.
|
|
"""
|
|
|
|
def check_running():
|
|
app_status_data = serve.status().applications[app_name]
|
|
app_status = app_status_data.status
|
|
assert app_status == ApplicationStatus.RUNNING, f"{app_status_data}"
|
|
return True
|
|
|
|
ray.get(lifecycle_signal.send.remote())
|
|
wait_for_condition(check_running, timeout=30, retry_interval_ms=500)
|
|
ray.get(lifecycle_signal.send.remote(clear=True))
|
|
|
|
def test_static_num_replicas_target_capacity_update(
|
|
self, shutdown_ray_and_serve, client: ServeControllerClient
|
|
):
|
|
"""Check how Serve's status updates when target_capacity changes."""
|
|
|
|
app_name = "controlled_app"
|
|
deployment_name = "controlled"
|
|
num_replicas = 20
|
|
|
|
signal = SignalActor.options(
|
|
name="lifecycle_signal", namespace=SERVE_NAMESPACE
|
|
).remote()
|
|
|
|
config = ServeDeploySchema(
|
|
applications=[
|
|
ServeApplicationSchema(
|
|
name=app_name,
|
|
import_path=(
|
|
"ray.serve.tests.test_target_capacity:create_controlled_app"
|
|
),
|
|
args={"num_replicas": num_replicas},
|
|
)
|
|
]
|
|
)
|
|
|
|
# Initially deploy at target_capacity 0, and check status.
|
|
self.apply_config_and_check_status(
|
|
client,
|
|
target_capacity=0.0,
|
|
config=config,
|
|
app_name=app_name,
|
|
deployment_name=deployment_name,
|
|
expected_app_status=ApplicationStatus.RUNNING,
|
|
expected_deployment_status=DeploymentStatus.HEALTHY,
|
|
expected_deployment_status_trigger=(
|
|
DeploymentStatusTrigger.CONFIG_UPDATE_COMPLETED
|
|
),
|
|
)
|
|
self.check_num_replicas(0, app_name, deployment_name)
|
|
|
|
# Increase the target_capacity, and check again.
|
|
self.apply_config_and_check_status(
|
|
client,
|
|
target_capacity=50.0,
|
|
config=config,
|
|
app_name=app_name,
|
|
deployment_name=deployment_name,
|
|
expected_app_status=ApplicationStatus.DEPLOYING,
|
|
expected_deployment_status=DeploymentStatus.UPSCALING,
|
|
expected_deployment_status_trigger=(
|
|
DeploymentStatusTrigger.CONFIG_UPDATE_STARTED
|
|
),
|
|
)
|
|
self.check_num_replicas(0, app_name, deployment_name)
|
|
self.unblock_replica_creation_and_deletion(signal, app_name)
|
|
self.check_num_replicas(int(0.5 * num_replicas), app_name, deployment_name)
|
|
|
|
# Decrease the target_capacity, and check again.
|
|
self.apply_config_and_check_status(
|
|
client,
|
|
target_capacity=10.0,
|
|
config=config,
|
|
app_name=app_name,
|
|
deployment_name=deployment_name,
|
|
expected_app_status=ApplicationStatus.DEPLOYING,
|
|
expected_deployment_status=DeploymentStatus.DOWNSCALING,
|
|
expected_deployment_status_trigger=(
|
|
DeploymentStatusTrigger.CONFIG_UPDATE_STARTED
|
|
),
|
|
)
|
|
# DeploymentStateManager marks replicas as STOPPING once the deployment
|
|
# starts downscaling.
|
|
wait_for_condition(
|
|
self.check_num_replicas,
|
|
expected_num_replicas=int(0.1 * num_replicas),
|
|
app_name=app_name,
|
|
deployment_name=deployment_name,
|
|
replica_state=ReplicaState.RUNNING,
|
|
timeout=20,
|
|
)
|
|
self.check_num_replicas(
|
|
int(0.4 * num_replicas),
|
|
app_name,
|
|
deployment_name,
|
|
replica_state=ReplicaState.STOPPING,
|
|
)
|
|
self.unblock_replica_creation_and_deletion(signal, app_name)
|
|
self.check_num_replicas(int(0.1 * num_replicas), app_name, deployment_name)
|
|
self.check_num_replicas(
|
|
0, app_name, deployment_name, replica_state=ReplicaState.STOPPING
|
|
)
|
|
|
|
# Send a signal so all replicas shut down quickly when the test finishes.
|
|
ray.get(signal.send.remote())
|
|
|
|
@pytest.mark.skipif(sys.platform == "win32", reason="Autoscaling flaky on Windows.")
|
|
def test_autoscaling_target_capacity_update(
|
|
self, shutdown_ray_and_serve, client: ServeControllerClient
|
|
):
|
|
"""Check Serve's status when target_capacity changes while autoscaling."""
|
|
# TODO(landscapepainter): This test fails locally due to the stall for replica initialization
|
|
# during upscaling and delayed response from serve.status(). It does not fail from
|
|
# buildkite, but need to investigate why it fails locally.
|
|
app_name = "controlled_app"
|
|
deployment_name = "controlled"
|
|
min_replicas = 10
|
|
initial_replicas = 12
|
|
max_replicas = 20
|
|
|
|
lifecycle_signal = SignalActor.options(
|
|
name="lifecycle_signal", namespace=SERVE_NAMESPACE
|
|
).remote()
|
|
request_signal = SignalActor.options(
|
|
name="request_signal", namespace=SERVE_NAMESPACE
|
|
).remote()
|
|
|
|
config = ServeDeploySchema(
|
|
applications=[
|
|
ServeApplicationSchema(
|
|
name=app_name,
|
|
import_path=(
|
|
"ray.serve.tests.test_target_capacity:"
|
|
"create_autoscaling_controlled_app"
|
|
),
|
|
args=dict(
|
|
min_replicas=min_replicas,
|
|
initial_replicas=initial_replicas,
|
|
max_replicas=max_replicas,
|
|
),
|
|
)
|
|
]
|
|
)
|
|
|
|
# Initially deploy at target_capacity 0, and check status.
|
|
self.apply_config_and_check_status(
|
|
client,
|
|
target_capacity=0.0,
|
|
config=config,
|
|
app_name=app_name,
|
|
deployment_name=deployment_name,
|
|
timeout=20,
|
|
expected_app_status=ApplicationStatus.RUNNING,
|
|
expected_deployment_status=DeploymentStatus.HEALTHY,
|
|
expected_deployment_status_trigger=(
|
|
DeploymentStatusTrigger.CONFIG_UPDATE_COMPLETED
|
|
),
|
|
)
|
|
self.check_num_replicas(0, app_name, deployment_name)
|
|
|
|
# Increase the target_capacity, and check again.
|
|
self.apply_config_and_check_status(
|
|
client,
|
|
target_capacity=50.0,
|
|
config=config,
|
|
app_name=app_name,
|
|
deployment_name=deployment_name,
|
|
expected_app_status=ApplicationStatus.DEPLOYING,
|
|
expected_deployment_status=DeploymentStatus.UPSCALING,
|
|
expected_deployment_status_trigger=(
|
|
DeploymentStatusTrigger.CONFIG_UPDATE_STARTED
|
|
),
|
|
)
|
|
self.check_num_replicas(0, app_name, deployment_name)
|
|
self.unblock_replica_creation_and_deletion(lifecycle_signal, app_name)
|
|
self.check_num_replicas(int(0.5 * initial_replicas), app_name, deployment_name)
|
|
|
|
# Send requests and check that the application scales up.
|
|
requests = []
|
|
handle = serve.get_app_handle(app_name)
|
|
for _ in range(max_replicas):
|
|
requests.append(handle.remote())
|
|
ray.get(lifecycle_signal.send.remote())
|
|
wait_for_condition(
|
|
self.check_num_replicas,
|
|
expected_num_replicas=int(0.5 * max_replicas),
|
|
app_name=app_name,
|
|
deployment_name=deployment_name,
|
|
timeout=30,
|
|
)
|
|
|
|
# Clear requests and check that application scales down.
|
|
ray.get(request_signal.send.remote())
|
|
results = [request.result() for request in requests]
|
|
assert results == ["Hello world!"] * (max_replicas)
|
|
wait_for_condition(
|
|
self.check_num_replicas,
|
|
expected_num_replicas=int(0.5 * initial_replicas),
|
|
app_name=app_name,
|
|
deployment_name=deployment_name,
|
|
replica_state=ReplicaState.RUNNING,
|
|
timeout=20,
|
|
retry_interval_ms=1000,
|
|
)
|
|
wait_for_condition(
|
|
self.check_num_replicas,
|
|
expected_num_replicas=0,
|
|
app_name=app_name,
|
|
deployment_name=deployment_name,
|
|
replica_state=ReplicaState.STOPPING,
|
|
timeout=20,
|
|
retry_interval_ms=1000,
|
|
)
|
|
ray.get(lifecycle_signal.send.remote(clear=True))
|
|
ray.get(request_signal.send.remote(clear=True))
|
|
|
|
# Decrease the target_capacity, and check that min_replicas is used
|
|
# to create the lower bound.
|
|
self.apply_config_and_check_status(
|
|
client,
|
|
target_capacity=10.0,
|
|
config=config,
|
|
app_name=app_name,
|
|
deployment_name=deployment_name,
|
|
expected_app_status=ApplicationStatus.DEPLOYING,
|
|
expected_deployment_status=DeploymentStatus.DOWNSCALING,
|
|
expected_deployment_status_trigger=(
|
|
DeploymentStatusTrigger.CONFIG_UPDATE_STARTED
|
|
),
|
|
)
|
|
self.unblock_replica_creation_and_deletion(lifecycle_signal, app_name)
|
|
self.check_num_replicas(
|
|
int(0.1 * min_replicas),
|
|
app_name,
|
|
deployment_name,
|
|
replica_state=ReplicaState.RUNNING,
|
|
)
|
|
self.check_num_replicas(
|
|
0, app_name, deployment_name, replica_state=ReplicaState.STOPPING
|
|
)
|
|
|
|
# Check that target_capacity * max_replicas is still the upper bound.
|
|
requests = []
|
|
handle = serve.get_app_handle(app_name)
|
|
for _ in range(max_replicas):
|
|
requests.append(handle.remote())
|
|
ray.get(lifecycle_signal.send.remote())
|
|
wait_for_condition(
|
|
self.check_num_replicas,
|
|
expected_num_replicas=int(0.1 * max_replicas),
|
|
app_name=app_name,
|
|
deployment_name=deployment_name,
|
|
controller_handle=client._controller,
|
|
timeout=25,
|
|
retry_interval_ms=2000,
|
|
)
|
|
|
|
# Clear requests and check that application scales down to
|
|
# target_capacity * min_replicas.
|
|
ray.get(request_signal.send.remote())
|
|
results = [request.result() for request in requests]
|
|
assert results == ["Hello world!"] * (max_replicas)
|
|
|
|
wait_for_condition(
|
|
self.check_num_replicas,
|
|
expected_num_replicas=int(0.1 * min_replicas),
|
|
app_name=app_name,
|
|
deployment_name=deployment_name,
|
|
controller_handle=client._controller,
|
|
timeout=25,
|
|
retry_interval_ms=2000,
|
|
)
|
|
ray.get(lifecycle_signal.send.remote(clear=True))
|
|
ray.get(request_signal.send.remote(clear=True))
|
|
|
|
# Scaling up to 100% target_capacity should make Serve use
|
|
# initial_replicas as lower bound.
|
|
self.apply_config_and_check_status(
|
|
client,
|
|
target_capacity=100.0,
|
|
config=config,
|
|
app_name=app_name,
|
|
deployment_name=deployment_name,
|
|
expected_app_status=ApplicationStatus.DEPLOYING,
|
|
expected_deployment_status=DeploymentStatus.UPSCALING,
|
|
expected_deployment_status_trigger=(
|
|
DeploymentStatusTrigger.CONFIG_UPDATE_STARTED
|
|
),
|
|
)
|
|
self.check_num_replicas(int(0.1 * min_replicas), app_name, deployment_name)
|
|
self.unblock_replica_creation_and_deletion(lifecycle_signal, app_name)
|
|
self.check_num_replicas(initial_replicas, app_name, deployment_name)
|
|
|
|
# Unsetting target_capacity should make Serve use
|
|
# min_replicas as lower bound. The current number of
|
|
# replicas is already a valid number, so the application stays
|
|
# RUNNING.
|
|
self.apply_config_and_check_status(
|
|
client,
|
|
target_capacity=None,
|
|
config=config,
|
|
app_name=app_name,
|
|
deployment_name=deployment_name,
|
|
expected_app_status=ApplicationStatus.RUNNING,
|
|
expected_deployment_status=DeploymentStatus.DOWNSCALING,
|
|
expected_deployment_status_trigger=(DeploymentStatusTrigger.AUTOSCALING),
|
|
)
|
|
self.unblock_replica_creation_and_deletion(lifecycle_signal, app_name)
|
|
wait_for_condition(
|
|
self.check_num_replicas,
|
|
expected_num_replicas=min_replicas,
|
|
app_name=app_name,
|
|
deployment_name=deployment_name,
|
|
)
|
|
|
|
# Send a signal so all replicas shut down quickly when the test finishes.
|
|
ray.get(request_signal.send.remote())
|
|
ray.get(lifecycle_signal.send.remote())
|
|
|
|
|
|
@serve.deployment(
|
|
ray_actor_options={"num_cpus": 0},
|
|
max_ongoing_requests=2,
|
|
graceful_shutdown_timeout_s=0,
|
|
)
|
|
class HangDeployment:
|
|
async def __call__(self, *args):
|
|
await asyncio.sleep(10000)
|
|
|
|
|
|
def create_hang_app(config: Dict) -> Application:
|
|
name: str = config["name"]
|
|
min_replicas: int = config["min_replicas"]
|
|
initial_replicas: int = config["initial_replicas"]
|
|
max_replicas: int = config["max_replicas"]
|
|
|
|
return HangDeployment.options(
|
|
name=name,
|
|
autoscaling_config={
|
|
"min_replicas": min_replicas,
|
|
"initial_replicas": initial_replicas,
|
|
"max_replicas": max_replicas,
|
|
"target_ongoing_requests": 1,
|
|
"metrics_interval_s": 0.01,
|
|
"downscale_delay_s": 0.01,
|
|
},
|
|
graceful_shutdown_timeout_s=0.01,
|
|
).bind()
|
|
|
|
|
|
class TestInitialReplicasHandling:
|
|
def deploy_config_and_wait_for_target_capacity(
|
|
self,
|
|
client: ServeControllerClient,
|
|
config: ServeDeploySchema,
|
|
target_capacity: float,
|
|
):
|
|
config.target_capacity = target_capacity
|
|
client.deploy_apps(config)
|
|
wait_for_condition(lambda: serve.status().target_capacity == target_capacity)
|
|
|
|
def test_initial_replicas_scales_down(
|
|
self, shutdown_ray_and_serve, client: ServeControllerClient
|
|
):
|
|
# TODO(landscapepainter): This test fails locally due to the stall for replica initialization
|
|
# during upscaling and delayed response from serve.status(). It does not fail from
|
|
# buildkite, but need to investigate why it fails locally.
|
|
deployment_name = "start_at_ten"
|
|
min_replicas = 5
|
|
initial_replicas = 10
|
|
|
|
config = ServeDeploySchema(
|
|
applications=[
|
|
ServeApplicationSchema(
|
|
import_path="ray.serve.tests.test_target_capacity:create_hang_app",
|
|
args={
|
|
"name": deployment_name,
|
|
"min_replicas": min_replicas,
|
|
"initial_replicas": initial_replicas,
|
|
"max_replicas": 15,
|
|
},
|
|
),
|
|
],
|
|
)
|
|
|
|
client.deploy_apps(config)
|
|
wait_for_condition(lambda: len(serve.status().applications) == 1)
|
|
assert serve.status().target_capacity is None
|
|
|
|
# Wait for initial deployment to reach RUNNING before starting the
|
|
# target_capacity loop. Without this, the first loop iteration must
|
|
# handle both initial replica startup AND target_capacity convergence
|
|
# within its timeout, which is marginal under CI resource contention.
|
|
wait_for_condition(
|
|
lambda: (
|
|
SERVE_DEFAULT_APP_NAME in serve.status().applications
|
|
and serve.status().applications[SERVE_DEFAULT_APP_NAME].status
|
|
== ApplicationStatus.RUNNING
|
|
),
|
|
timeout=30,
|
|
)
|
|
|
|
# Kick off downscaling pattern.
|
|
test_target_capacities = [100, 60, 40, 0]
|
|
expected_num_replicas = [
|
|
int(min_replicas * capacity / 100) for capacity in test_target_capacities
|
|
]
|
|
|
|
for target_capacity, num_replicas in zip(
|
|
test_target_capacities, expected_num_replicas
|
|
):
|
|
print(f"target_capacity: {target_capacity}, num_replicas: {num_replicas}")
|
|
config.target_capacity = target_capacity
|
|
client.deploy_apps(config)
|
|
wait_for_condition(
|
|
lambda: serve.status().target_capacity == target_capacity
|
|
)
|
|
wait_for_condition(
|
|
check_expected_num_replicas,
|
|
deployment_to_num_replicas={deployment_name: num_replicas},
|
|
timeout=30,
|
|
)
|
|
|
|
def test_initial_replicas_scales_up_and_down(
|
|
self, shutdown_ray_and_serve, client: ServeControllerClient
|
|
):
|
|
# TODO(landscapepainter): This test fails locally due to the stall for replica initialization
|
|
# during upscaling and delayed response from serve.status(). It does not fail from
|
|
# buildkite, but need to investigate why it fails locally.
|
|
deployment_name = "start_at_five"
|
|
min_replicas = 0
|
|
initial_replicas = 5
|
|
|
|
config = ServeDeploySchema(
|
|
target_capacity=0,
|
|
applications=[
|
|
ServeApplicationSchema(
|
|
import_path="ray.serve.tests.test_target_capacity:create_hang_app",
|
|
args={
|
|
"name": deployment_name,
|
|
"min_replicas": min_replicas,
|
|
"initial_replicas": initial_replicas,
|
|
"max_replicas": 2 * initial_replicas,
|
|
},
|
|
),
|
|
],
|
|
)
|
|
|
|
test_target_capacities = [0, 1, 20, 60, 40, 30, 100, None]
|
|
expected_num_replicas = [
|
|
0,
|
|
1,
|
|
1,
|
|
3,
|
|
min_replicas,
|
|
min_replicas,
|
|
initial_replicas,
|
|
min_replicas,
|
|
]
|
|
|
|
for target_capacity, num_replicas in zip(
|
|
test_target_capacities, expected_num_replicas
|
|
):
|
|
print(f"target_capacity: {target_capacity}, num_replicas: {num_replicas}")
|
|
config.target_capacity = target_capacity
|
|
client.deploy_apps(config)
|
|
wait_for_condition(
|
|
lambda: serve.status().target_capacity == target_capacity
|
|
)
|
|
wait_for_condition(
|
|
check_expected_num_replicas,
|
|
deployment_to_num_replicas={deployment_name: num_replicas},
|
|
timeout=30,
|
|
)
|
|
|
|
def test_initial_replicas_zero(
|
|
self, shutdown_ray_and_serve, client: ServeControllerClient
|
|
):
|
|
deployment_name = "start_at_zero"
|
|
min_replicas = 0
|
|
initial_replicas = 0
|
|
|
|
config = ServeDeploySchema(
|
|
target_capacity=0,
|
|
applications=[
|
|
ServeApplicationSchema(
|
|
import_path="ray.serve.tests.test_target_capacity:create_hang_app",
|
|
args={
|
|
"name": deployment_name,
|
|
"min_replicas": min_replicas,
|
|
"initial_replicas": initial_replicas,
|
|
"max_replicas": 20,
|
|
},
|
|
),
|
|
],
|
|
)
|
|
|
|
test_target_capacities = [0, 1, 20, 60, 40, 30, 100, None]
|
|
expected_num_replicas = [0] * len(test_target_capacities)
|
|
|
|
for target_capacity, num_replicas in zip(
|
|
test_target_capacities, expected_num_replicas
|
|
):
|
|
config.target_capacity = target_capacity
|
|
client.deploy_apps(config)
|
|
wait_for_condition(
|
|
lambda: serve.status().target_capacity == target_capacity
|
|
)
|
|
wait_for_condition(
|
|
check_expected_num_replicas,
|
|
deployment_to_num_replicas={deployment_name: num_replicas},
|
|
)
|
|
|
|
def test_initial_replicas_new_configs(
|
|
self, shutdown_ray_and_serve, client: ServeControllerClient
|
|
):
|
|
# TODO(landscapepainter): This test fails locally due to the stall for replica initialization
|
|
# during upscaling and delayed response from serve.status(). It does not fail from
|
|
# buildkite, but need to investigate why it fails locally.
|
|
deployment_name = "start_at_ten"
|
|
min_replicas = 0
|
|
initial_replicas = 10
|
|
config_target_capacity = 40
|
|
|
|
config = ServeDeploySchema(
|
|
target_capacity=config_target_capacity,
|
|
applications=[
|
|
ServeApplicationSchema(
|
|
name="app1",
|
|
import_path="ray.serve.tests.test_target_capacity:create_hang_app",
|
|
args={
|
|
"name": deployment_name,
|
|
"min_replicas": min_replicas,
|
|
"initial_replicas": initial_replicas,
|
|
"max_replicas": initial_replicas * 2,
|
|
},
|
|
),
|
|
],
|
|
)
|
|
|
|
# When deploying first config, initial_replicas * target_capacity
|
|
# should be treated as floor
|
|
client.deploy_apps(config)
|
|
wait_for_condition(
|
|
lambda: serve.status().target_capacity == config_target_capacity
|
|
)
|
|
wait_for_condition(
|
|
check_expected_num_replicas,
|
|
deployment_to_num_replicas={
|
|
deployment_name: int(initial_replicas * config_target_capacity / 100)
|
|
},
|
|
app_name="app1",
|
|
timeout=30,
|
|
)
|
|
|
|
# When deploying a new config, initial_replicas * target_capacity
|
|
# should be treated as floor for all deployments, even if
|
|
# target_capacity is lower
|
|
new_deployment_name = f"new_{deployment_name}"
|
|
new_config_target_capacity = config_target_capacity / 2
|
|
|
|
new_config = deepcopy(config)
|
|
new_config.applications.append(
|
|
ServeApplicationSchema(
|
|
name="app2",
|
|
import_path="ray.serve.tests.test_target_capacity:create_hang_app",
|
|
args={
|
|
"name": new_deployment_name,
|
|
"min_replicas": min_replicas,
|
|
"initial_replicas": initial_replicas,
|
|
"max_replicas": initial_replicas * 2,
|
|
},
|
|
),
|
|
)
|
|
new_config.target_capacity = new_config_target_capacity
|
|
|
|
client.deploy_apps(new_config)
|
|
wait_for_condition(
|
|
lambda: serve.status().target_capacity == new_config_target_capacity
|
|
)
|
|
wait_for_condition(
|
|
check_expected_num_replicas,
|
|
deployment_to_num_replicas={
|
|
deployment_name: int(
|
|
initial_replicas * new_config_target_capacity / 100
|
|
)
|
|
},
|
|
app_name="app1",
|
|
timeout=30,
|
|
)
|
|
wait_for_condition(
|
|
check_expected_num_replicas,
|
|
deployment_to_num_replicas={
|
|
new_deployment_name: int(
|
|
initial_replicas * new_config_target_capacity / 100
|
|
)
|
|
},
|
|
app_name="app2",
|
|
timeout=30,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(pytest.main(["-v", "-s", __file__]))
|