wehub-resource-sync
853 lines
32 KiB
Python
853 lines
32 KiB
Python
import sys
|
|
|
|
import pytest
|
|
|
|
import ray
|
|
from ray import serve
|
|
from ray._common.test_utils import wait_for_condition
|
|
from ray._raylet import GcsClient
|
|
from ray.serve._private import default_impl
|
|
from ray.serve._private.common import DeploymentID, ReplicaID
|
|
from ray.serve._private.constants import RAY_SERVE_USE_PACK_SCHEDULING_STRATEGY
|
|
from ray.serve._private.deployment_scheduler import (
|
|
ReplicaSchedulingRequest,
|
|
SpreadDeploymentSchedulingPolicy,
|
|
)
|
|
from ray.serve._private.test_utils import check_apps_running, get_node_id
|
|
from ray.serve._private.utils import get_head_node_id
|
|
from ray.tests.conftest import * # noqa
|
|
|
|
|
|
@ray.remote(num_cpus=1)
|
|
class Replica:
|
|
def get_node_id(self):
|
|
return ray.get_runtime_context().get_node_id()
|
|
|
|
def get_placement_group(self):
|
|
return ray.util.get_current_placement_group()
|
|
|
|
|
|
@pytest.mark.skipif(
|
|
RAY_SERVE_USE_PACK_SCHEDULING_STRATEGY, reason="Need to use spread strategy"
|
|
)
|
|
class TestSpreadScheduling:
|
|
@pytest.mark.parametrize(
|
|
"placement_group_config",
|
|
[
|
|
{},
|
|
{"bundles": [{"CPU": 3}]},
|
|
{
|
|
"bundles": [{"CPU": 1}, {"CPU": 1}, {"CPU": 1}],
|
|
"strategy": "STRICT_PACK",
|
|
},
|
|
],
|
|
)
|
|
def test_spread_deployment_scheduling_policy_upscale(
|
|
self, ray_start_cluster, placement_group_config
|
|
):
|
|
"""Test to make sure replicas are spreaded."""
|
|
cluster = ray_start_cluster
|
|
cluster.add_node(num_cpus=3)
|
|
cluster.add_node(num_cpus=3)
|
|
cluster.wait_for_nodes()
|
|
ray.init(address=cluster.address)
|
|
|
|
cluster_node_info_cache = default_impl.create_cluster_node_info_cache(
|
|
GcsClient(address=ray.get_runtime_context().gcs_address)
|
|
)
|
|
cluster_node_info_cache.update()
|
|
|
|
scheduler = default_impl.create_deployment_scheduler(
|
|
cluster_node_info_cache,
|
|
get_head_node_id(),
|
|
)
|
|
dep_id = DeploymentID(name="deployment1")
|
|
r1_id = ReplicaID(unique_id="replica1", deployment_id=dep_id)
|
|
r2_id = ReplicaID(unique_id="replica2", deployment_id=dep_id)
|
|
scheduler.on_deployment_created(dep_id, SpreadDeploymentSchedulingPolicy())
|
|
replica_actor_handles = []
|
|
replica_placement_groups = []
|
|
|
|
def on_scheduled(actor_handle, placement_group):
|
|
replica_actor_handles.append(actor_handle)
|
|
replica_placement_groups.append(placement_group)
|
|
|
|
deployment_to_replicas_to_stop = scheduler.schedule(
|
|
upscales={
|
|
dep_id: [
|
|
ReplicaSchedulingRequest(
|
|
replica_id=r1_id,
|
|
actor_def=Replica,
|
|
actor_resources={"CPU": 1},
|
|
actor_options={"name": "deployment1_replica1"},
|
|
actor_init_args=(),
|
|
on_scheduled=on_scheduled,
|
|
placement_group_bundles=placement_group_config.get(
|
|
"bundles", None
|
|
),
|
|
placement_group_strategy=placement_group_config.get(
|
|
"strategy", None
|
|
),
|
|
),
|
|
ReplicaSchedulingRequest(
|
|
replica_id=r2_id,
|
|
actor_def=Replica,
|
|
actor_resources={"CPU": 1},
|
|
actor_options={"name": "deployment1_replica2"},
|
|
actor_init_args=(),
|
|
on_scheduled=on_scheduled,
|
|
placement_group_bundles=placement_group_config.get(
|
|
"bundles", None
|
|
),
|
|
placement_group_strategy=placement_group_config.get(
|
|
"strategy", None
|
|
),
|
|
),
|
|
]
|
|
},
|
|
downscales={},
|
|
)
|
|
assert not deployment_to_replicas_to_stop
|
|
assert len(replica_actor_handles) == 2
|
|
assert len(replica_placement_groups) == 2
|
|
assert not scheduler._pending_replicas[dep_id]
|
|
assert len(scheduler._launching_replicas[dep_id]) == 2
|
|
assert (
|
|
len(
|
|
{
|
|
ray.get(replica_actor_handles[0].get_node_id.remote()),
|
|
ray.get(replica_actor_handles[1].get_node_id.remote()),
|
|
}
|
|
)
|
|
== 2
|
|
)
|
|
if "bundles" in placement_group_config:
|
|
assert (
|
|
len(
|
|
{
|
|
ray.get(replica_actor_handles[0].get_placement_group.remote()),
|
|
ray.get(replica_actor_handles[1].get_placement_group.remote()),
|
|
}
|
|
)
|
|
== 2
|
|
)
|
|
scheduler.on_replica_stopping(r1_id)
|
|
scheduler.on_replica_stopping(r2_id)
|
|
scheduler.on_deployment_deleted(dep_id)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_spread_serve_strict_spread_pg(self, ray_cluster):
|
|
"""
|
|
Verifies STRICT_SPREAD PG strategy runs successfully in the Spread Scheduler
|
|
and spreads bundles across distinct nodes.
|
|
"""
|
|
cluster = ray_cluster
|
|
cluster.add_node(num_cpus=3)
|
|
cluster.add_node(num_cpus=3)
|
|
cluster.wait_for_nodes()
|
|
ray.init(address=cluster.address)
|
|
serve.start()
|
|
|
|
@ray.remote(num_cpus=0)
|
|
def get_task_node_id():
|
|
return ray.get_runtime_context().get_node_id()
|
|
|
|
@serve.deployment(
|
|
placement_group_bundles=[{"CPU": 1}, {"CPU": 1}],
|
|
placement_group_strategy="STRICT_SPREAD",
|
|
)
|
|
class StrictSpread:
|
|
async def get_bundle_node_id(self, bundle_index: int):
|
|
pg = ray.util.get_current_placement_group()
|
|
return await get_task_node_id.options(
|
|
scheduling_strategy=ray.util.scheduling_strategies.PlacementGroupSchedulingStrategy(
|
|
placement_group=pg,
|
|
placement_group_bundle_index=bundle_index,
|
|
)
|
|
).remote()
|
|
|
|
handle = serve.run(StrictSpread.bind(), name="strict_spread_app")
|
|
|
|
node_0 = await handle.get_bundle_node_id.remote(0)
|
|
node_1 = await handle.get_bundle_node_id.remote(1)
|
|
|
|
assert node_0 != node_1
|
|
|
|
serve.delete("strict_spread_app")
|
|
serve.shutdown()
|
|
|
|
|
|
@serve.deployment
|
|
def A():
|
|
return ray.get_runtime_context().get_node_id()
|
|
|
|
|
|
app_A = A.bind()
|
|
|
|
|
|
@pytest.mark.skipif(
|
|
not RAY_SERVE_USE_PACK_SCHEDULING_STRATEGY, reason="Needs pack strategy."
|
|
)
|
|
class TestPackScheduling:
|
|
@pytest.mark.parametrize("use_pg", [True, False])
|
|
def test_e2e_basic(self, ray_cluster, use_pg: bool):
|
|
cluster = ray_cluster
|
|
cluster.add_node(num_cpus=2, resources={"head": 1})
|
|
cluster.add_node(num_cpus=3, resources={"worker1": 1})
|
|
cluster.add_node(num_cpus=4, resources={"worker2": 1})
|
|
cluster.wait_for_nodes()
|
|
ray.init(address=cluster.address)
|
|
|
|
head_node_id = ray.get(get_node_id.options(resources={"head": 1}).remote())
|
|
worker1_node_id = ray.get(
|
|
get_node_id.options(resources={"worker1": 1}).remote()
|
|
)
|
|
worker2_node_id = ray.get(
|
|
get_node_id.options(resources={"worker2": 1}).remote()
|
|
)
|
|
print("head", head_node_id)
|
|
print("worker1", worker1_node_id)
|
|
print("worker2", worker2_node_id)
|
|
|
|
# Both f replicas should be scheduled on head node to minimize
|
|
# fragmentation
|
|
if use_pg:
|
|
app1 = A.options(
|
|
num_replicas=2,
|
|
ray_actor_options={"num_cpus": 0.1},
|
|
placement_group_bundles=[{"CPU": 0.5}, {"CPU": 0.5}],
|
|
placement_group_strategy="STRICT_PACK",
|
|
).bind()
|
|
else:
|
|
app1 = A.options(num_replicas=2, ray_actor_options={"num_cpus": 1}).bind()
|
|
|
|
# Both app1 replicas should have been scheduled on head node
|
|
f_handle = serve.run(app1, name="app1", route_prefix="/app1")
|
|
refs = [f_handle.remote() for _ in range(20)]
|
|
assert {ref.result() for ref in refs} == {head_node_id}
|
|
|
|
if use_pg:
|
|
app2 = A.options(
|
|
num_replicas=1,
|
|
ray_actor_options={"num_cpus": 0.1},
|
|
placement_group_bundles=[{"CPU": 1}, {"CPU": 2}],
|
|
placement_group_strategy="STRICT_PACK",
|
|
).bind()
|
|
else:
|
|
app2 = A.options(num_replicas=1, ray_actor_options={"num_cpus": 3}).bind()
|
|
|
|
# Then there should be enough space for the g replica
|
|
# The g replica should be scheduled on worker1, not worker2, to
|
|
# minimize fragmentation
|
|
g_handle = serve.run(app2, name="app2", route_prefix="/app2")
|
|
assert g_handle.remote().result() == worker1_node_id
|
|
|
|
serve.shutdown()
|
|
|
|
@pytest.mark.parametrize("use_pg", [True, False])
|
|
@pytest.mark.parametrize(
|
|
"app_resources,expected_worker_nodes",
|
|
[
|
|
# [2, 5, 3, 3, 7, 6, 4] -> 3 nodes
|
|
({5: 1, 3: 2, 7: 1, 2: 1, 6: 1, 4: 1}, 3),
|
|
# [1, 7, 7, 3, 2] -> 2 nodes
|
|
({1: 1, 7: 2, 3: 1, 2: 1}, 2),
|
|
# [7, 3, 2, 7, 7, 2] -> 3 nodes
|
|
({7: 3, 3: 1, 2: 2}, 3),
|
|
],
|
|
)
|
|
def test_e2e_fit_replicas(
|
|
self, ray_cluster, use_pg, app_resources, expected_worker_nodes
|
|
):
|
|
for _ in range(expected_worker_nodes):
|
|
ray_cluster.add_node(num_cpus=1)
|
|
ray_cluster.wait_for_nodes()
|
|
ray.init(address=ray_cluster.address)
|
|
|
|
serve.start()
|
|
|
|
@serve.deployment
|
|
def A():
|
|
return ray.get_runtime_context().get_node_id()
|
|
|
|
@serve.deployment(ray_actor_options={"num_cpus": 0})
|
|
class Ingress:
|
|
def __init__(self, *handles):
|
|
self.handles = handles
|
|
|
|
def __call__(self):
|
|
pass
|
|
|
|
deployments = []
|
|
for n, count in app_resources.items():
|
|
num_cpus = 0.1 * n
|
|
deployments.append(
|
|
A.options(
|
|
name=f"A{n}",
|
|
num_replicas=count,
|
|
ray_actor_options={"num_cpus": 0 if use_pg else num_cpus},
|
|
placement_group_bundles=[{"CPU": num_cpus}] if use_pg else None,
|
|
placement_group_strategy="STRICT_PACK" if use_pg else None,
|
|
).bind()
|
|
)
|
|
|
|
serve.run(Ingress.bind(*deployments))
|
|
wait_for_condition(check_apps_running, apps=["default"])
|
|
print("Test passed!")
|
|
|
|
@pytest.mark.parametrize("use_pg", [True, False])
|
|
def test_e2e_custom_resources(self, ray_cluster, use_pg):
|
|
cluster = ray_cluster
|
|
cluster.add_node(num_cpus=1, resources={"head": 1})
|
|
cluster.add_node(num_cpus=3, resources={"worker1": 1, "customabcd": 1})
|
|
cluster.wait_for_nodes()
|
|
ray.init(address=cluster.address)
|
|
|
|
worker1_node_id = ray.get(
|
|
get_node_id.options(resources={"worker1": 1}).remote()
|
|
)
|
|
|
|
if use_pg:
|
|
app = A.options(
|
|
num_replicas=1,
|
|
ray_actor_options={"num_cpus": 0},
|
|
placement_group_bundles=[{"CPU": 0.5}, {"CPU": 0.5, "customabcd": 0.1}],
|
|
placement_group_strategy="STRICT_PACK",
|
|
).bind()
|
|
else:
|
|
app = A.options(
|
|
num_replicas=1,
|
|
ray_actor_options={"num_cpus": 1, "resources": {"customabcd": 0.1}},
|
|
).bind()
|
|
|
|
handle1 = serve.run(app, name="app1", route_prefix="/app1")
|
|
refs = [handle1.remote() for _ in range(20)]
|
|
assert all(ref.result() == worker1_node_id for ref in refs)
|
|
|
|
serve.shutdown()
|
|
|
|
def test_high_priority_memory_schedules_before_cpu_hogs(
|
|
self, ray_cluster, monkeypatch
|
|
):
|
|
"""Memory in RAY_SERVE_HIGH_PRIORITY_CUSTOM_RESOURCES overrides CPU priority.
|
|
|
|
Pack scheduling sorts pending replicas by ``Resources.__lt__``. By default
|
|
CPU is compared before memory, so a deployment with higher ``num_cpus``
|
|
is scheduled first when only one replica fits on the node.
|
|
|
|
Here each ``cpu_i`` requests ``replica_cpus`` CPUs while ``memory_hog``
|
|
requests ``replica_cpus - 1`` CPUs plus most of the node memory. Without
|
|
the env var, a ``cpu_i`` deployment would win; with
|
|
``RAY_SERVE_HIGH_PRIORITY_CUSTOM_RESOURCES=memory``, ``memory_hog`` must
|
|
be the only deployment that reaches RUNNING.
|
|
"""
|
|
monkeypatch.setenv("RAY_SERVE_HIGH_PRIORITY_CUSTOM_RESOURCES", "memory")
|
|
|
|
cluster = ray_cluster
|
|
cluster.add_node(num_cpus=4)
|
|
cluster.wait_for_nodes()
|
|
ray.init(address=cluster.address)
|
|
|
|
total_memory = int(ray.cluster_resources()["memory"])
|
|
high_memory = max(int(total_memory * 0.9), 1)
|
|
|
|
@serve.deployment
|
|
def replica_fn():
|
|
return "ok"
|
|
|
|
@serve.deployment(ray_actor_options={"num_cpus": 0})
|
|
class Ingress:
|
|
def __init__(self, *handles):
|
|
self.handles = handles
|
|
|
|
def __call__(self):
|
|
pass
|
|
|
|
# Reserve head/proxy CPU by measuring what Serve leaves available.
|
|
serve.start()
|
|
replica_cpus = int(ray.available_resources()["CPU"])
|
|
assert replica_cpus >= 2, "Need at least 2 CPUs for cpu vs memory_hog split"
|
|
|
|
memory_hog_cpus = replica_cpus - 1
|
|
|
|
deployments = []
|
|
for i in range(9):
|
|
deployments.append(
|
|
replica_fn.options(
|
|
name=f"cpu_{i}",
|
|
# Higher CPU than memory_hog: would sort first by default.
|
|
ray_actor_options={"num_cpus": replica_cpus},
|
|
).bind()
|
|
)
|
|
deployments.append(
|
|
replica_fn.options(
|
|
name="memory_hog",
|
|
ray_actor_options={
|
|
"num_cpus": memory_hog_cpus,
|
|
"memory": high_memory,
|
|
},
|
|
).bind()
|
|
)
|
|
|
|
serve._run(Ingress.bind(*deployments), _blocking=False)
|
|
|
|
def check_only_memory_hog_running():
|
|
app_status = serve.status().applications["default"]
|
|
if app_status.status == "DEPLOY_FAILED":
|
|
raise AssertionError(f"App failed: {app_status.message}")
|
|
|
|
deployments_status = app_status.deployments
|
|
memory_hog = deployments_status["memory_hog"]
|
|
assert memory_hog.replica_states.get("RUNNING", 0) == 1, memory_hog
|
|
|
|
for i in range(9):
|
|
cpu_dep = deployments_status[f"cpu_{i}"]
|
|
running = cpu_dep.replica_states.get("RUNNING", 0)
|
|
assert running == 0, (f"cpu_{i}", cpu_dep.replica_states)
|
|
|
|
return True
|
|
|
|
wait_for_condition(check_only_memory_hog_running, timeout=60)
|
|
|
|
app_status = serve.status().applications["default"]
|
|
assert app_status.status == "DEPLOYING"
|
|
|
|
# Add a second node: pack logs should show a new schedule-order batch
|
|
# and one cpu_* replica placed on the new node while memory_hog stays up.
|
|
cluster.add_node(num_cpus=4)
|
|
cluster.wait_for_nodes()
|
|
|
|
def check_one_cpu_replica_after_scale_out():
|
|
app_status = serve.status().applications["default"]
|
|
if app_status.status == "DEPLOY_FAILED":
|
|
raise AssertionError(f"App failed: {app_status.message}")
|
|
|
|
deployments_status = app_status.deployments
|
|
assert (
|
|
deployments_status["memory_hog"].replica_states.get("RUNNING", 0) == 1
|
|
)
|
|
|
|
cpu_running = sum(
|
|
deployments_status[f"cpu_{i}"].replica_states.get("RUNNING", 0)
|
|
for i in range(9)
|
|
)
|
|
assert cpu_running == 1, {
|
|
f"cpu_{i}": deployments_status[f"cpu_{i}"].replica_states
|
|
for i in range(9)
|
|
}
|
|
return True
|
|
|
|
wait_for_condition(check_one_cpu_replica_after_scale_out, timeout=60)
|
|
|
|
app_status = serve.status().applications["default"]
|
|
assert app_status.status == "DEPLOYING"
|
|
|
|
serve.shutdown()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_e2e_serve_strict_pack_pg_label_selector(
|
|
self, serve_instance_with_labeled_nodes
|
|
):
|
|
"""
|
|
Verifies STRICT_PACK strategy with placement_group_bundle_label_selector in Pack Scheduling Mode.
|
|
|
|
Since the strategy is STRICT_PACK, both bundles must be scheduled on the same node,
|
|
and that node must satisfy the label constraints in each selector.
|
|
"""
|
|
_, _, us_east_node_id, _ = serve_instance_with_labeled_nodes
|
|
|
|
@ray.remote(num_cpus=0)
|
|
def get_task_node_id():
|
|
return ray.get_runtime_context().get_node_id()
|
|
|
|
@serve.deployment(
|
|
placement_group_bundles=[{"CPU": 1}, {"CPU": 1}],
|
|
placement_group_strategy="STRICT_PACK",
|
|
placement_group_bundle_label_selector=[
|
|
{"gpu-type": "H100", "region": "us-east"}
|
|
],
|
|
)
|
|
class StrictPackSelector:
|
|
async def get_bundle_node_id(self, bundle_index: int):
|
|
pg = ray.util.get_current_placement_group()
|
|
return await get_task_node_id.options(
|
|
scheduling_strategy=ray.util.scheduling_strategies.PlacementGroupSchedulingStrategy(
|
|
placement_group=pg,
|
|
placement_group_bundle_index=bundle_index,
|
|
)
|
|
).remote()
|
|
|
|
handle = serve.run(StrictPackSelector.bind(), name="strict_pack_app")
|
|
|
|
# Both bundles are scheduled to the same node which matches the label constraints.
|
|
assert await handle.get_bundle_node_id.remote(0) == us_east_node_id
|
|
assert await handle.get_bundle_node_id.remote(1) == us_east_node_id
|
|
|
|
serve.delete("strict_pack_app")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_e2e_serve_pack_pg_forces_spread(
|
|
self, serve_instance_with_labeled_nodes
|
|
):
|
|
"""
|
|
Verifies that using non-strict PACK PG strategy with label selectors works.
|
|
|
|
STRICT_PACK throws NotImplementedError for selectors. However, 'PACK' is considered a
|
|
'Non-Strict' strategy which forces the scheduler to fall back to 'Spread Mode'.
|
|
"""
|
|
_, _, us_east_node_id, _ = serve_instance_with_labeled_nodes
|
|
|
|
@serve.deployment(
|
|
placement_group_bundles=[{"CPU": 1}],
|
|
placement_group_strategy="PACK",
|
|
placement_group_bundle_label_selector=[{"gpu-type": "H100"}],
|
|
)
|
|
class PackSelector:
|
|
def get_node_id(self):
|
|
return ray.get_runtime_context().get_node_id()
|
|
|
|
# If this stayed in the Pack Scheduler, it would raise NotImplementedError.
|
|
# Because it forces Spread Mode, it succeeds.
|
|
handle = serve.run(PackSelector.bind(), name="pack_selector_app")
|
|
assert await handle.get_node_id.remote() == us_east_node_id
|
|
serve.delete("pack_selector_app")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_e2e_serve_multiple_bundles_selector(
|
|
self, serve_instance_with_labeled_nodes
|
|
):
|
|
"""Verifies multiple bundles with bundle_label_selector are applied correctly."""
|
|
_, us_west_node_id, us_east_node_id, _ = serve_instance_with_labeled_nodes
|
|
|
|
# Helper task to return the node ID it's running on
|
|
@ray.remote(num_cpus=0)
|
|
def get_task_node_id():
|
|
return ray.get_runtime_context().get_node_id()
|
|
|
|
@serve.deployment(
|
|
placement_group_bundles=[{"CPU": 1}, {"CPU": 1}],
|
|
placement_group_strategy="SPREAD",
|
|
placement_group_bundle_label_selector=[
|
|
{"gpu-type": "H100"}, # matches us-east node
|
|
{"gpu-type": "A100"}, # matches us-west node
|
|
],
|
|
)
|
|
class MultiBundleSelector:
|
|
async def get_bundle_node_id(self, bundle_index: int):
|
|
pg = ray.util.get_current_placement_group()
|
|
return await get_task_node_id.options(
|
|
scheduling_strategy=ray.util.scheduling_strategies.PlacementGroupSchedulingStrategy(
|
|
placement_group=pg,
|
|
placement_group_bundle_index=bundle_index,
|
|
)
|
|
).remote()
|
|
|
|
handle = serve.run(MultiBundleSelector.bind(), name="multi_bundle_app")
|
|
|
|
# Verify bundles are scheduled to expected nodes based on label selectors.
|
|
assert await handle.get_bundle_node_id.remote(0) == us_east_node_id
|
|
assert await handle.get_bundle_node_id.remote(1) == us_west_node_id
|
|
serve.delete("multi_bundle_app")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_e2e_serve_multiple_bundles_single_bundle_label_selector(
|
|
self, serve_instance_with_labeled_nodes
|
|
):
|
|
"""
|
|
Verifies that when only one bundle_label_selector is provided for multiple bundles,
|
|
the label_selector is applied to each bundle uniformly.
|
|
"""
|
|
_, _, us_east_node_id, _ = serve_instance_with_labeled_nodes
|
|
|
|
@ray.remote(num_cpus=0)
|
|
def get_task_node_id():
|
|
return ray.get_runtime_context().get_node_id()
|
|
|
|
@serve.deployment(
|
|
placement_group_bundles=[{"CPU": 1}, {"CPU": 1}],
|
|
# Use SPREAD to verify the label constraint forces them to same node.
|
|
placement_group_strategy="SPREAD",
|
|
placement_group_bundle_label_selector=[
|
|
{"gpu-type": "H100"},
|
|
],
|
|
)
|
|
class MultiBundleSelector:
|
|
async def get_bundle_node_id(self, bundle_index: int):
|
|
pg = ray.util.get_current_placement_group()
|
|
return await get_task_node_id.options(
|
|
scheduling_strategy=ray.util.scheduling_strategies.PlacementGroupSchedulingStrategy(
|
|
placement_group=pg,
|
|
placement_group_bundle_index=bundle_index,
|
|
)
|
|
).remote()
|
|
|
|
handle = serve.run(MultiBundleSelector.bind(), name="multi_bundle_app")
|
|
assert await handle.get_bundle_node_id.remote(0) == us_east_node_id
|
|
assert await handle.get_bundle_node_id.remote(1) == us_east_node_id
|
|
serve.delete("multi_bundle_app")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_e2e_serve_actor_multiple_fallbacks(
|
|
self, serve_instance_with_labeled_nodes
|
|
):
|
|
"""
|
|
Verifies that the scheduler can iterate through a label selector and multiple fallback options.
|
|
"""
|
|
_, us_west_node_id, _, _ = serve_instance_with_labeled_nodes
|
|
|
|
@serve.deployment(
|
|
ray_actor_options={
|
|
"label_selector": {"region": "invalid-label-1"},
|
|
"fallback_strategy": [
|
|
{"label_selector": {"region": "invalid-label-2"}},
|
|
{"label_selector": {"region": "us-west"}}, # Should match
|
|
],
|
|
}
|
|
)
|
|
class MultiFallbackActor:
|
|
def get_node_id(self):
|
|
return ray.get_runtime_context().get_node_id()
|
|
|
|
handle = serve.run(MultiFallbackActor.bind(), name="multi_fallback_app")
|
|
assert await handle.get_node_id.remote() == us_west_node_id
|
|
serve.delete("multi_fallback_app")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_e2e_serve_label_selector(serve_instance_with_labeled_nodes):
|
|
"""
|
|
Verifies that label selectors work correctly for both Actors and Placement Groups.
|
|
|
|
This test also verifies that label selectors are respected when scheduling with a
|
|
preferred node ID for resource compaction. This test verifies both the Pack and
|
|
Spread scheduler paths.
|
|
"""
|
|
_, us_west_node_id, us_east_node_id, _ = serve_instance_with_labeled_nodes
|
|
|
|
# Validate a Serve deplyoment utilizes a label_selector when passed to the Ray Actor options.
|
|
@serve.deployment(ray_actor_options={"label_selector": {"region": "us-west"}})
|
|
class DeploymentActor:
|
|
def get_node_id(self):
|
|
return ray.get_runtime_context().get_node_id()
|
|
|
|
handle = serve.run(DeploymentActor.bind(), name="actor_app")
|
|
assert await handle.get_node_id.remote() == us_west_node_id
|
|
serve.delete("actor_app")
|
|
|
|
# Validate placement_group scheduling strategy with placement_group_bundle_label_selector
|
|
# and PACK strategy.
|
|
@serve.deployment(
|
|
placement_group_bundles=[{"CPU": 1}],
|
|
placement_group_strategy="PACK",
|
|
placement_group_bundle_label_selector=[{"gpu-type": "H100"}],
|
|
)
|
|
class DeploymentPGPack:
|
|
def get_node_id(self):
|
|
return ray.get_runtime_context().get_node_id()
|
|
|
|
handle_pack = serve.run(DeploymentPGPack.bind(), name="pg_pack_app")
|
|
assert await handle_pack.get_node_id.remote() == us_east_node_id
|
|
serve.delete("pg_pack_app")
|
|
|
|
# Validate placement_group scheduling strategy with placement_group_bundle_label_selector
|
|
# and SPREAD strategy.
|
|
@serve.deployment(
|
|
placement_group_bundles=[{"CPU": 1}],
|
|
placement_group_strategy="SPREAD",
|
|
placement_group_bundle_label_selector=[{"gpu-type": "H100"}],
|
|
)
|
|
class DeploymentPGSpread:
|
|
def get_node_id(self):
|
|
return ray.get_runtime_context().get_node_id()
|
|
|
|
handle_spread = serve.run(DeploymentPGSpread.bind(), name="pg_spread_app")
|
|
assert await handle_spread.get_node_id.remote() == us_east_node_id
|
|
serve.delete("pg_spread_app")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_e2e_serve_fallback_strategy(serve_instance_with_labeled_nodes):
|
|
"""
|
|
Verifies that fallback strategies allow scheduling on alternative nodes when
|
|
primary constraints fail.
|
|
"""
|
|
_, _, h100_node_id, _ = serve_instance_with_labeled_nodes
|
|
|
|
# Fallback strategy specified for Ray Actor in Serve deployment.
|
|
@serve.deployment(
|
|
ray_actor_options={
|
|
"label_selector": {"region": "unavailable"},
|
|
"fallback_strategy": [{"label_selector": {"gpu-type": "H100"}}],
|
|
}
|
|
)
|
|
class FallbackDeployment:
|
|
def get_node_id(self):
|
|
return ray.get_runtime_context().get_node_id()
|
|
|
|
# TODO (ryanaoleary@): Add a test for fallback_strategy in placement group options
|
|
# when support is added.
|
|
|
|
handle = serve.run(FallbackDeployment.bind(), name="fallback_app")
|
|
assert await handle.get_node_id.remote() == h100_node_id
|
|
serve.delete("fallback_app")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"use_pg,strategy",
|
|
[
|
|
(False, None), # Actor-level label_selector.
|
|
(True, "PACK"), # PG bundle_label_selector with PACK strategy.
|
|
(True, "STRICT_PACK"), # PG bundle_label_selector with STRICT_PACK strategy.
|
|
(True, "SPREAD"), # PG bundle_label_selector with SPREAD strategy.
|
|
(
|
|
True,
|
|
"STRICT_SPREAD",
|
|
), # PG bundle_label_selector with STRICT_SPREAD strategy.
|
|
],
|
|
)
|
|
async def test_e2e_serve_label_selector_unschedulable(
|
|
serve_instance_with_labeled_nodes, use_pg, strategy
|
|
):
|
|
"""
|
|
Verifies the interaction between unschedulable a placement_group_bundle_label_selector
|
|
and different scheduling strategies in the Pack and Spread Serve scheduler.
|
|
"""
|
|
_, _, _, cluster = serve_instance_with_labeled_nodes
|
|
|
|
@serve.deployment
|
|
def A():
|
|
return ray.get_runtime_context().get_node_id()
|
|
|
|
# Cluster in fixture only contains us-west and us-east.
|
|
target_label = {"region": "eu-central"}
|
|
|
|
if use_pg:
|
|
app = A.options(
|
|
num_replicas=1,
|
|
placement_group_bundles=[{"CPU": 1}],
|
|
placement_group_strategy=strategy,
|
|
placement_group_bundle_label_selector=[target_label],
|
|
).bind()
|
|
else:
|
|
app = A.options(
|
|
num_replicas=1,
|
|
ray_actor_options={"label_selector": target_label},
|
|
).bind()
|
|
|
|
handle = serve._run(app, name="unschedulable_label_app", _blocking=False)
|
|
|
|
def check_status(expected_status):
|
|
try:
|
|
status_info = serve.status().applications["unschedulable_label_app"]
|
|
return status_info.status == expected_status
|
|
except KeyError:
|
|
return False
|
|
|
|
def verify_resource_request_stuck():
|
|
"""Verifies that the underlying resource request is pending."""
|
|
# Serve deployment should be stuck DEPLOYING.
|
|
if not check_status("DEPLOYING"):
|
|
return False
|
|
|
|
# Check PG/Actor is actually pending.
|
|
if use_pg:
|
|
pgs = ray.util.state.list_placement_groups()
|
|
return any(pg["state"] == "PENDING" for pg in pgs)
|
|
else:
|
|
actors = ray.util.state.list_actors()
|
|
return any(a["state"] == "PENDING_CREATION" for a in actors)
|
|
|
|
# Serve deployment should remain stuck in deploying because Actor/PG can't be scheduled.
|
|
wait_for_condition(verify_resource_request_stuck, timeout=30)
|
|
assert not check_status("RUNNING"), (
|
|
"Test setup failed: The deployment became RUNNING before the required "
|
|
"node was added. The label selector constraint was ignored."
|
|
)
|
|
|
|
# Add a suitable node to the cluster.
|
|
new_node = cluster.add_node(
|
|
num_cpus=2, labels=target_label, resources={"target_node": 1}
|
|
)
|
|
cluster.wait_for_nodes()
|
|
expected_node_id = ray.get(
|
|
get_node_id.options(resources={"target_node": 1}).remote()
|
|
)
|
|
|
|
# Validate deployment can now be scheduled since label selector is satisfied.
|
|
wait_for_condition(lambda: check_status("RUNNING"), timeout=30)
|
|
assert await handle.remote() == expected_node_id
|
|
|
|
serve.delete("unschedulable_label_app")
|
|
cluster.remove_node(new_node)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_e2e_serve_fallback_strategy_unschedulable(
|
|
serve_instance_with_labeled_nodes,
|
|
):
|
|
"""
|
|
Verifies that an unschedulable fallback_strategy causes the Serve deployment to wait
|
|
until a suitable node is added to the cluster.
|
|
"""
|
|
_, _, _, cluster = serve_instance_with_labeled_nodes
|
|
|
|
@serve.deployment
|
|
def A():
|
|
return ray.get_runtime_context().get_node_id()
|
|
|
|
fallback_label = {"region": "me-central2"}
|
|
|
|
app = A.options(
|
|
num_replicas=1,
|
|
ray_actor_options={
|
|
"label_selector": {"region": "non-existant"},
|
|
"fallback_strategy": [{"label_selector": fallback_label}],
|
|
},
|
|
).bind()
|
|
|
|
handle = serve._run(app, name="unschedulable_fallback_app", _blocking=False)
|
|
|
|
def check_status(expected_status):
|
|
try:
|
|
status_info = serve.status().applications["unschedulable_fallback_app"]
|
|
return status_info.status == expected_status
|
|
except KeyError:
|
|
return False
|
|
|
|
def verify_resource_request_stuck():
|
|
"""Verifies that the underlying resource request is pending."""
|
|
# Serve deployment should be stuck DEPLOYING.
|
|
if not check_status("DEPLOYING"):
|
|
return False
|
|
|
|
actors = ray.util.state.list_actors()
|
|
return any(a["state"] == "PENDING_CREATION" for a in actors)
|
|
|
|
# Serve deployment should remain stuck in deploying because Actor/PG can't be scheduled.
|
|
wait_for_condition(verify_resource_request_stuck, timeout=30)
|
|
assert not check_status("RUNNING"), (
|
|
"Test setup failed: The deployment became RUNNING before the required "
|
|
"node was added. The label selector constraint was ignored."
|
|
)
|
|
|
|
# Add a node that matches the fallback.
|
|
new_node = cluster.add_node(
|
|
num_cpus=2, labels=fallback_label, resources={"fallback_node": 1}
|
|
)
|
|
cluster.wait_for_nodes()
|
|
expected_node_id = ray.get(
|
|
get_node_id.options(resources={"fallback_node": 1}).remote()
|
|
)
|
|
|
|
# The serve deployment should recover and start running on the fallback node.
|
|
wait_for_condition(lambda: check_status("RUNNING"), timeout=30)
|
|
assert await handle.remote() == expected_node_id
|
|
|
|
serve.delete("unschedulable_fallback_app")
|
|
cluster.remove_node(new_node)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(pytest.main(["-v", "-s", __file__]))
|