wehub-resource-sync
282 lines
10 KiB
Python
282 lines
10 KiB
Python
import sys
|
|
|
|
import pytest
|
|
|
|
import ray
|
|
from ray import serve
|
|
from ray._common.test_utils import SignalActor, wait_for_condition
|
|
from ray.serve._private.test_utils import check_apps_running, check_num_replicas_eq
|
|
from ray.tests.conftest import * # noqa
|
|
|
|
|
|
class TestScaleDownReplicaSelection:
|
|
@staticmethod
|
|
def _quick_upscale_config():
|
|
return {
|
|
"target_ongoing_requests": 0.01,
|
|
"upscale_delay_s": 0.05,
|
|
"metrics_interval_s": 0.1,
|
|
"look_back_period_s": 0.5,
|
|
"downscale_delay_s": 2,
|
|
"aggregation_function": "max",
|
|
}
|
|
|
|
@staticmethod
|
|
def _deploy_test_app(
|
|
app_name: str,
|
|
deployment_name: str = "test_deployment",
|
|
*,
|
|
signal,
|
|
ray_actor_options: dict,
|
|
placement_group_bundles: list[dict] = None,
|
|
placement_group_bundle_label_selector: list[dict] = None,
|
|
autoscaling_config: dict = None,
|
|
):
|
|
@serve.deployment(name=deployment_name, max_ongoing_requests=100)
|
|
class TestDeployment:
|
|
async def __call__(self):
|
|
# Load path: block until the test releases the signal so the
|
|
# request stays in flight and the deployment looks busy.
|
|
await signal.wait.remote()
|
|
|
|
async def get_info(self):
|
|
return {
|
|
"node_id": ray.get_runtime_context().get_node_id(),
|
|
"replica_tag": serve.get_replica_context().replica_tag,
|
|
}
|
|
|
|
return serve.run(
|
|
TestDeployment.options(
|
|
ray_actor_options=ray_actor_options,
|
|
placement_group_bundles=placement_group_bundles,
|
|
placement_group_bundle_label_selector=placement_group_bundle_label_selector,
|
|
autoscaling_config=autoscaling_config,
|
|
).bind(),
|
|
name=app_name,
|
|
route_prefix=f"/{app_name}",
|
|
)
|
|
|
|
@staticmethod
|
|
def _wait_until_running(app_name: str, deployment_name: str, count: int):
|
|
wait_for_condition(
|
|
check_num_replicas_eq,
|
|
name=deployment_name,
|
|
target=count,
|
|
app_name=app_name,
|
|
use_controller=True,
|
|
timeout=60,
|
|
)
|
|
|
|
def _scale_up_then_down(
|
|
self, handle, app_name: str, deployment_name: str, max_replicas: int, signal
|
|
):
|
|
"""Scale up to ``max_replicas`` then back down to 1.
|
|
|
|
Holds ``max_replicas`` requests in flight (blocked on ``signal``) so the
|
|
autoscaler reaches and holds the cap while replicas start, then releases
|
|
them so, with no load, it scales back to the min.
|
|
"""
|
|
blocked = [handle.remote() for _ in range(max_replicas)]
|
|
try:
|
|
self._wait_until_running(app_name, deployment_name, max_replicas)
|
|
finally:
|
|
ray.get(signal.send.remote())
|
|
for ref in blocked:
|
|
try:
|
|
ref.result()
|
|
except Exception:
|
|
pass
|
|
self._wait_until_running(app_name, deployment_name, 1)
|
|
|
|
def test_downscale_fallback_node(self, ray_cluster):
|
|
cluster = ray_cluster
|
|
|
|
primary_label = {"type": "primary"}
|
|
fallback_label = {"type": "fallback"}
|
|
|
|
ray_actor_options = {
|
|
"num_cpus": 0.25,
|
|
"label_selector": primary_label,
|
|
"fallback_strategy": [{"label_selector": fallback_label}],
|
|
}
|
|
|
|
# Both nodes get equal capacity (1 CPU each = 4 replicas at 0.25 CPU)
|
|
# so that priority #4 (fewer replicas per node) doesn't confound
|
|
# the test for priority #3 (fallback nodes removed first).
|
|
num_replicas_per_node = 4
|
|
max_replicas = num_replicas_per_node * 2
|
|
|
|
cluster.add_node(num_cpus=0)
|
|
cluster.wait_for_nodes()
|
|
fallback_node = cluster.add_node(
|
|
num_cpus=1,
|
|
labels=fallback_label,
|
|
)
|
|
cluster.wait_for_nodes()
|
|
ray.init(address=cluster.address)
|
|
serve.start()
|
|
app_name = "downscale_fallback_app"
|
|
deployment_name = "test_deployment"
|
|
|
|
fallback_node_id = fallback_node.node_id
|
|
signal = SignalActor.remote()
|
|
|
|
try:
|
|
handle = self._deploy_test_app(
|
|
app_name,
|
|
signal=signal,
|
|
ray_actor_options=ray_actor_options,
|
|
autoscaling_config={
|
|
"min_replicas": 1,
|
|
"max_replicas": max_replicas,
|
|
**self._quick_upscale_config(),
|
|
},
|
|
)
|
|
wait_for_condition(check_apps_running, apps=[app_name])
|
|
|
|
primary_node = cluster.add_node(
|
|
num_cpus=1,
|
|
labels=primary_label,
|
|
)
|
|
cluster.wait_for_nodes()
|
|
primary_node_id = primary_node.node_id
|
|
|
|
# The first replica is always the fallback node.
|
|
assert handle.get_info.remote().result()["node_id"] == fallback_node_id
|
|
|
|
# Scale up to the cap (replicas split across the fallback and primary
|
|
# nodes), then back down to the min replica.
|
|
self._scale_up_then_down(
|
|
handle, app_name, deployment_name, max_replicas, signal
|
|
)
|
|
|
|
# Replicas on the fallback node should be removed first (priority #3),
|
|
# so the remaining replica should be on the primary node.
|
|
assert handle.get_info.remote().result()["node_id"] == primary_node_id
|
|
finally:
|
|
serve.shutdown()
|
|
|
|
# TODO: Add test for downscale placement group fallback_strategy when it's added to deployment options.
|
|
|
|
def test_downscale_prefers_nodes_with_fewer_total_replicas(self, ray_cluster):
|
|
cluster = ray_cluster
|
|
cluster.add_node(num_cpus=0)
|
|
cluster.wait_for_nodes()
|
|
primary_label = {"type": "primary"}
|
|
first_node = cluster.add_node(
|
|
num_cpus=1,
|
|
labels=primary_label,
|
|
)
|
|
cluster.wait_for_nodes()
|
|
ray.init(address=cluster.address)
|
|
serve.start()
|
|
|
|
ray_actor_options = {"num_cpus": 0}
|
|
placement_group_bundles = [{"CPU": 0.25}] * 4
|
|
placement_group_bundle_label_selector = [primary_label]
|
|
app_name = "downscale_fewer_total_replicas_app"
|
|
deployment_name = "test_deployment"
|
|
first_node_id = first_node.node_id
|
|
max_replicas = 3
|
|
signal = SignalActor.remote()
|
|
|
|
try:
|
|
handle = self._deploy_test_app(
|
|
app_name,
|
|
deployment_name=deployment_name,
|
|
signal=signal,
|
|
ray_actor_options=ray_actor_options,
|
|
placement_group_bundles=placement_group_bundles,
|
|
placement_group_bundle_label_selector=placement_group_bundle_label_selector,
|
|
autoscaling_config={
|
|
"min_replicas": 1,
|
|
"max_replicas": max_replicas,
|
|
**self._quick_upscale_config(),
|
|
},
|
|
)
|
|
wait_for_condition(check_apps_running, apps=[app_name])
|
|
|
|
second_node = cluster.add_node(
|
|
num_cpus=2,
|
|
labels=primary_label,
|
|
)
|
|
cluster.wait_for_nodes()
|
|
second_node_id = second_node.node_id
|
|
|
|
# The first replica is always the first node.
|
|
assert handle.get_info.remote().result()["node_id"] == first_node_id
|
|
|
|
# Scale up across both nodes, then back down to the min replica.
|
|
self._scale_up_then_down(
|
|
handle, app_name, deployment_name, max_replicas, signal
|
|
)
|
|
|
|
# First node has fewer total replicas, so it is removed first
|
|
# (priority #4). Remaining replica should be on the 2nd node.
|
|
assert handle.get_info.remote().result()["node_id"] == second_node_id
|
|
finally:
|
|
serve.shutdown()
|
|
|
|
def test_downscale_prefers_not_head_node(self, ray_cluster):
|
|
"""Head node is never relinquished, even when it would otherwise be removed first.
|
|
|
|
The head node has only 1 replica, matches only the fallback label, and is
|
|
older, so priorities #3, #4, and #5 all favor removing it. This test
|
|
verifies that priority #2 (keep head node) overrides all of them.
|
|
"""
|
|
cluster = ray_cluster
|
|
fallback_label = {"type": "fallback"}
|
|
primary_label = {"type": "primary"}
|
|
head_node = cluster.add_node(num_cpus=1, labels=fallback_label)
|
|
cluster.wait_for_nodes()
|
|
ray.init(address=cluster.address, ignore_reinit_error=True)
|
|
serve.start()
|
|
|
|
ray_actor_options = {
|
|
"num_cpus": 1,
|
|
"label_selector": primary_label,
|
|
"fallback_strategy": [{"label_selector": fallback_label}],
|
|
}
|
|
app_name = "downscale_prefers_not_head_app"
|
|
deployment_name = "test_deployment"
|
|
head_node_id = head_node.node_id
|
|
max_replicas = 3
|
|
signal = SignalActor.remote()
|
|
|
|
try:
|
|
handle = self._deploy_test_app(
|
|
app_name,
|
|
deployment_name=deployment_name,
|
|
signal=signal,
|
|
ray_actor_options=ray_actor_options,
|
|
autoscaling_config={
|
|
"min_replicas": 1,
|
|
"max_replicas": max_replicas,
|
|
**self._quick_upscale_config(),
|
|
},
|
|
)
|
|
wait_for_condition(check_apps_running, apps=[app_name])
|
|
|
|
cluster.add_node(num_cpus=2, labels=primary_label)
|
|
cluster.wait_for_nodes()
|
|
|
|
# The first replica lands on the head node (the only node with
|
|
# the fallback label, and no primary node exists yet).
|
|
assert handle.get_info.remote().result()["node_id"] == head_node_id
|
|
|
|
# Scale up to 3 replicas (1 head + 2 worker), then back down to 1.
|
|
self._scale_up_then_down(
|
|
handle, app_name, deployment_name, max_replicas, signal
|
|
)
|
|
|
|
# The head node's replica survives despite being on a fallback
|
|
# node (#3), having fewer replicas (#4), and being the oldest
|
|
# (#5): priority #2 (never relinquish head node) wins.
|
|
assert handle.get_info.remote().result()["node_id"] == head_node_id
|
|
finally:
|
|
serve.shutdown()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(pytest.main(["-v", "-s", __file__]))
|