wehub/ray-project--ray
1
0
Fork 0
Code Issues Pull Requests Actions Deployments Models Agent Notes Packages Projects Releases Wiki Activity
Files
main
ray-project--ray/python/ray/autoscaler/v2/tests/test_scheduler.py
T
wehub-resource-sync f1825c8ceb chore: import upstream snapshot with attribution
2026-07-13 13:17:40 +08:00

3647 lines
118 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
import os
import sys
import time
# coding: utf-8
from typing import Dict, List, Optional, Tuple
from unittest.mock import patch
import pytest
import ray
from ray.autoscaler.v2.event_logger import AutoscalerEventLogger
from ray.autoscaler.v2.scheduler import (
NodeTypeConfig,
ResourceDemandScheduler,
ResourceRequestSource,
SchedulingNode,
SchedulingNodeStatus,
SchedulingReply,
SchedulingRequest,
logger,
)
from ray.autoscaler.v2.schema import (
AutoscalerInstance,
IPPRGroupSpec,
IPPRSpecs,
IPPRStatus,
NodeType,
)
from ray.autoscaler.v2.tests.util import MockEventLogger, make_autoscaler_instance
from ray.autoscaler.v2.utils import ResourceRequestUtil
from ray.core.generated.autoscaler_pb2 import (
ClusterResourceConstraint,
GangResourceRequest,
NodeState,
NodeStatus,
ResourceRequest,
)
from ray.core.generated.common_pb2 import LabelSelectorOperator
from ray.core.generated.instance_manager_pb2 import (
Instance,
NodeKind,
TerminationRequest,
)
ResourceMap = Dict[str, float]
logger.setLevel("DEBUG")
event_logger = AutoscalerEventLogger(MockEventLogger(logger))
def sched_request(
node_type_configs: Dict[NodeType, NodeTypeConfig],
max_num_nodes: Optional[int] = None,
resource_requests: Optional[List[ResourceRequest]] = None,
gang_resource_requests: Optional[List[List[ResourceRequest]]] = None,
cluster_resource_constraints: Optional[List[ResourceRequest]] = None,
instances: Optional[List[AutoscalerInstance]] = None,
idle_timeout_s: Optional[float] = None,
disable_launch_config_check: Optional[bool] = False,
ippr_specs: Optional[IPPRSpecs] = None,
ippr_statuses: Optional[Dict[str, IPPRStatus]] = None,
cloud_resource_availabilities: Optional[Dict[NodeType, float]] = None,
) -> SchedulingRequest:
if resource_requests is None:
resource_requests = []
if gang_resource_requests is None:
gang_resource_requests = []
if cluster_resource_constraints is None:
cluster_resource_constraints = []
if instances is None:
instances = []
if ippr_statuses is None:
ippr_statuses = {}
if cloud_resource_availabilities is None:
cloud_resource_availabilities = {}
return SchedulingRequest(
resource_requests=ResourceRequestUtil.group_by_count(resource_requests),
gang_resource_requests=[
GangResourceRequest(requests=reqs) for reqs in gang_resource_requests
],
cluster_resource_constraints=(
[
ClusterResourceConstraint(
resource_requests=ResourceRequestUtil.group_by_count(
cluster_resource_constraints
)
)
]
if cluster_resource_constraints
else []
),
current_instances=instances,
node_type_configs=node_type_configs,
max_num_nodes=max_num_nodes,
idle_timeout_s=idle_timeout_s,
disable_launch_config_check=disable_launch_config_check,
ippr_specs=ippr_specs,
ippr_statuses=ippr_statuses,
cloud_resource_availabilities=cloud_resource_availabilities,
)
def _launch_and_terminate(
reply: SchedulingReply,
) -> Tuple[Dict[NodeType, int], List[str]]:
actual_to_launch = {req.instance_type: req.count for req in reply.to_launch}
actual_to_terminate = [
(req.instance_id, req.ray_node_id, req.cause) for req in reply.to_terminate
]
return actual_to_launch, actual_to_terminate
def schedule(
node_type_configs: Dict[NodeType, NodeTypeConfig],
current_nodes_available_count: Dict,
resource_requests: List[Dict],
anti_affinity: bool = False,
max_nodes: Optional[int] = None,
cloud_resource_availabilities: Optional[Dict[NodeType, float]] = None,
) -> SchedulingReply:
ANTI_AFFINITY = ResourceRequestUtil.PlacementConstraintType.ANTI_AFFINITY
instances: List[AutoscalerInstance] = []
for node_type, count in current_nodes_available_count.items():
for i in range(count):
instances.append(
make_autoscaler_instance(
im_instance=Instance(
instance_type=node_type,
status=Instance.RAY_RUNNING,
instance_id=f"{node_type}-{i}",
node_id=f"r{i}{node_type}",
),
ray_node=NodeState(
node_id=f"r{i}{node_type}".encode("utf-8"),
ray_node_type_name=node_type,
available_resources=node_type_configs[node_type].resources,
total_resources=node_type_configs[node_type].resources,
idle_duration_ms=0,
status=NodeStatus.RUNNING,
),
cloud_instance_id=f"c-{node_type}-{i}",
)
)
if anti_affinity:
gang_requests = [
[
ResourceRequestUtil.make(r, [(ANTI_AFFINITY, "af", "af")])
for r in resource_requests
]
]
request = sched_request(
node_type_configs=node_type_configs,
gang_resource_requests=gang_requests,
max_num_nodes=max_nodes,
instances=instances,
cloud_resource_availabilities=cloud_resource_availabilities,
)
else:
request = sched_request(
node_type_configs=node_type_configs,
resource_requests=[ResourceRequestUtil.make(r) for r in resource_requests],
instances=instances,
max_num_nodes=max_nodes,
cloud_resource_availabilities=cloud_resource_availabilities,
)
return ResourceDemandScheduler(event_logger).schedule(request)
class TestSchedulingNode:
@staticmethod
def test_is_schedulable():
instance = make_autoscaler_instance(im_instance=None)
assert SchedulingNode.is_schedulable(instance) is False
all_im_status = set(Instance.InstanceStatus.values())
positive_statuses = {
Instance.QUEUED,
Instance.REQUESTED,
Instance.ALLOCATED,
Instance.RAY_INSTALLING,
Instance.RAY_RUNNING,
Instance.RAY_STOP_REQUESTED,
}
negative_statues = {
Instance.UNKNOWN,
Instance.RAY_STOPPING,
Instance.RAY_STOPPED,
Instance.TERMINATING,
Instance.TERMINATED,
Instance.ALLOCATION_FAILED,
Instance.RAY_INSTALL_FAILED,
Instance.TERMINATION_FAILED,
Instance.ALLOCATION_TIMEOUT,
}
for status in all_im_status:
instance = make_autoscaler_instance(
im_instance=Instance(instance_type="type_1", status=status)
)
if status in positive_statuses:
assert SchedulingNode.is_schedulable(instance) is True
elif status in negative_statues:
assert SchedulingNode.is_schedulable(instance) is False
else:
assert False, f"Unknown status {status}"
@staticmethod
@pytest.mark.parametrize(
"disable_launch_config_check", [True, False], ids=["disabled", "enabled"]
)
def test_new_node(disable_launch_config_check):
# Assert none IM instance.
node_type_configs = {
"type_1": NodeTypeConfig(
name="type_1",
resources={"CPU": 1},
min_worker_nodes=0,
max_worker_nodes=10,
labels={"foo": "foo"},
),
}
instance = make_autoscaler_instance(im_instance=None)
assert (
SchedulingNode.new(instance, node_type_configs, disable_launch_config_check)
is None
)
# A running ray node
instance = make_autoscaler_instance(
ray_node=NodeState(
ray_node_type_name="type_1",
available_resources={"CPU": 0},
total_resources={"CPU": 1},
node_id=b"r1",
dynamic_labels={"foo": "bar"},
),
im_instance=Instance(
instance_type="type_1",
status=Instance.RAY_RUNNING,
instance_id="1",
node_id="r1",
),
)
node = SchedulingNode.new(
instance, node_type_configs, disable_launch_config_check
)
assert node is not None
assert node.node_type == "type_1"
assert node.status == SchedulingNodeStatus.SCHEDULABLE
assert node.ray_node_id == "r1"
assert node.im_instance_id == "1"
assert node.available_resources_for_sched == {
ResourceRequestSource.PENDING_DEMAND: {"CPU": 0},
ResourceRequestSource.CLUSTER_RESOURCE_CONSTRAINT: {"CPU": 1},
}
assert node.total_resources == {"CPU": 1}
assert node.labels == {"foo": "bar"}
# A outdated node.
instance = make_autoscaler_instance(
im_instance=Instance(
instance_type="type_no_longer_exists",
status=Instance.REQUESTED,
instance_id="1",
),
)
node = SchedulingNode.new(
instance, node_type_configs, disable_launch_config_check
)
if not disable_launch_config_check:
assert node is not None
assert node.node_type == "type_no_longer_exists"
assert node.status == SchedulingNodeStatus.TO_TERMINATE
assert node.termination_request is not None
assert node.termination_request.cause == TerminationRequest.Cause.OUTDATED
else:
assert node is None
# A pending ray node
instance = make_autoscaler_instance(
im_instance=Instance(
instance_type="type_1",
status=Instance.REQUESTED,
instance_id="1",
)
)
node = SchedulingNode.new(
instance, node_type_configs, disable_launch_config_check
)
assert node is not None
assert node.node_type == "type_1"
assert node.status == SchedulingNodeStatus.SCHEDULABLE
assert node.available_resources_for_sched == {
ResourceRequestSource.PENDING_DEMAND: {"CPU": 1},
ResourceRequestSource.CLUSTER_RESOURCE_CONSTRAINT: {"CPU": 1},
}
assert node.total_resources == {"CPU": 1}
assert node.labels == {"foo": "foo"}
@staticmethod
def test_new_running_instance_without_ray_node_returns_none():
# Regression test: a RAY_RUNNING instance whose ray_node is missing in
# GCS used to crash SchedulingNode.new with an AssertionError. The
# defensive guard returns None instead.
node_type_configs = {
"type_1": NodeTypeConfig(
name="type_1",
resources={"CPU": 1},
min_worker_nodes=0,
max_worker_nodes=10,
),
}
instance = make_autoscaler_instance(
ray_node=None,
im_instance=Instance(
instance_type="type_1",
status=Instance.RAY_RUNNING,
instance_id="i-stale",
node_id="r-gone",
),
)
node = SchedulingNode.new(
instance,
node_type_configs,
disable_launch_config_check=False,
)
assert node is None
@staticmethod
def test_new_head_node():
# An allocated head node.
node_type_configs = {
"head": NodeTypeConfig(
name="head",
resources={"CPU": 1},
min_worker_nodes=0,
max_worker_nodes=1,
),
}
instance = make_autoscaler_instance(
im_instance=Instance(
instance_type="head",
status=Instance.ALLOCATED,
instance_id="1",
node_kind=NodeKind.HEAD,
)
)
node = SchedulingNode.new(
instance, node_type_configs, disable_launch_config_check=False
)
assert node is not None
# It's important to check if the node is a head node
assert node.node_kind == NodeKind.HEAD
assert node.status == SchedulingNodeStatus.SCHEDULABLE
# An running head node.
instance = make_autoscaler_instance(
ray_node=NodeState(
ray_node_type_name="head",
available_resources={"CPU": 0},
total_resources={"CPU": 1},
node_id=b"r1",
),
im_instance=Instance(
instance_type="head",
status=Instance.RAY_RUNNING,
instance_id="1",
node_id="r1",
node_kind=NodeKind.HEAD,
),
)
node = SchedulingNode.new(
instance, node_type_configs, disable_launch_config_check=False
)
assert node is not None
assert node.node_kind == NodeKind.HEAD
assert node.status == SchedulingNodeStatus.SCHEDULABLE
def test_min_worker_nodes():
scheduler = ResourceDemandScheduler(event_logger)
node_type_configs = {
"type_1": NodeTypeConfig(
name="type_1",
resources={"CPU": 1},
min_worker_nodes=1,
max_worker_nodes=10,
),
"type_2": NodeTypeConfig(
name="type_2",
resources={"CPU": 1},
min_worker_nodes=0,
max_worker_nodes=10,
),
"type_3": NodeTypeConfig(
name="type_3",
resources={"CPU": 1},
min_worker_nodes=2,
max_worker_nodes=10,
),
}
# With empty cluster
request = sched_request(
node_type_configs=node_type_configs,
)
reply = scheduler.schedule(request)
expected_to_launch = {"type_1": 1, "type_3": 2}
reply = scheduler.schedule(request)
actual_to_launch, _ = _launch_and_terminate(reply)
assert sorted(actual_to_launch) == sorted(expected_to_launch)
# With existing ray nodes
request = sched_request(
node_type_configs=node_type_configs,
instances=[
make_autoscaler_instance(
im_instance=Instance(
instance_type="type_1", status=Instance.RAY_RUNNING
),
ray_node=NodeState(ray_node_type_name="type_1"),
),
make_autoscaler_instance(
im_instance=Instance(
instance_type="type_1", status=Instance.RAY_RUNNING
),
ray_node=NodeState(ray_node_type_name="type_1"),
),
],
)
expected_to_launch = {"type_3": 2}
reply = scheduler.schedule(request)
actual_to_launch, _ = _launch_and_terminate(reply)
assert actual_to_launch == expected_to_launch
# With existing instances pending.
request = sched_request(
node_type_configs=node_type_configs,
instances=[
make_autoscaler_instance(
im_instance=Instance(instance_type="type_1", status=Instance.REQUESTED)
),
make_autoscaler_instance(
im_instance=Instance(instance_type="type_1", status=Instance.ALLOCATED)
),
make_autoscaler_instance(
im_instance=Instance(
instance_type="type_no_longer_exists",
status=Instance.REQUESTED,
instance_id="0",
)
),
],
)
expected_to_launch = {"type_3": 2}
reply = scheduler.schedule(request)
actual_to_launch, _ = _launch_and_terminate(reply)
assert actual_to_launch == expected_to_launch
def test_max_workers_head_node_type():
scheduler = ResourceDemandScheduler()
node_type_configs = {
"head_type": NodeTypeConfig(
name="head_type",
resources={},
min_worker_nodes=0,
max_worker_nodes=2,
)
}
instances = [
# A head node
make_autoscaler_instance(
im_instance=Instance(
instance_type="head_type",
status=Instance.ALLOCATED,
instance_id="0",
node_kind=NodeKind.HEAD,
),
),
# A worker node
make_autoscaler_instance(
im_instance=Instance(
instance_type="head_type",
status=Instance.ALLOCATED,
instance_id="1",
node_kind=NodeKind.WORKER,
),
),
# A worker node
make_autoscaler_instance(
im_instance=Instance(
instance_type="head_type",
status=Instance.ALLOCATED,
instance_id="2",
node_kind=NodeKind.WORKER,
),
),
]
request = sched_request(node_type_configs=node_type_configs, instances=instances)
reply = scheduler.schedule(request)
_, actual_to_terminate = _launch_and_terminate(reply)
assert len(actual_to_terminate) == 1
assert actual_to_terminate[0][0] in ["1", "2"]
assert actual_to_terminate[0][2] == TerminationRequest.Cause.MAX_NUM_NODE_PER_TYPE
def test_max_workers_per_type():
scheduler = ResourceDemandScheduler(event_logger)
node_type_configs = {
"type_1": NodeTypeConfig(
name="type_1",
resources={"CPU": 1},
min_worker_nodes=2,
max_worker_nodes=2,
),
}
request = sched_request(
node_type_configs=node_type_configs,
)
reply = scheduler.schedule(request)
expected_to_terminate = []
_, actual_to_terminate = _launch_and_terminate(reply)
assert sorted(actual_to_terminate) == sorted(expected_to_terminate)
instances = [
make_autoscaler_instance(
im_instance=Instance(
instance_type="type_1", status=Instance.ALLOCATED, instance_id="0"
),
),
make_autoscaler_instance(
ray_node=NodeState(
ray_node_type_name="type_1",
available_resources={"CPU": 1},
total_resources={"CPU": 1},
node_id=b"r1",
),
im_instance=Instance(
instance_type="type_1",
status=Instance.RAY_RUNNING,
instance_id="1",
node_id="r1",
),
),
make_autoscaler_instance(
ray_node=NodeState(
ray_node_type_name="type_1",
available_resources={"CPU": 0.5},
total_resources={"CPU": 1},
node_id=b"r2",
),
im_instance=Instance(
instance_type="type_1",
status=Instance.RAY_RUNNING,
instance_id="2",
node_id="r2",
),
),
]
# 3 running instances with max of 2 allowed for type 1.
request = sched_request(
node_type_configs=node_type_configs,
instances=instances,
)
reply = scheduler.schedule(request)
_, actual_to_terminate = _launch_and_terminate(reply)
assert actual_to_terminate == [
("0", "", TerminationRequest.Cause.MAX_NUM_NODE_PER_TYPE)
]
# 3 running instances with max of 1 allowed for type 1.
node_type_configs = {
"type_1": NodeTypeConfig(
name="type_1",
resources={"CPU": 1},
min_worker_nodes=0,
max_worker_nodes=1,
),
}
request = sched_request(
node_type_configs=node_type_configs,
instances=instances,
)
reply = scheduler.schedule(request)
_, actual_to_terminate = _launch_and_terminate(reply)
assert sorted(actual_to_terminate) == sorted(
[
("0", "", TerminationRequest.Cause.MAX_NUM_NODE_PER_TYPE),
# Lower resource util.
(
"1",
"r1",
TerminationRequest.Cause.MAX_NUM_NODE_PER_TYPE,
),
]
)
def test_terminate_max_allocated_workers_per_type():
scheduler = ResourceDemandScheduler(event_logger)
node_type_configs = {
"type_1": NodeTypeConfig(
name="type_1",
resources={"CPU": 1},
min_worker_nodes=0,
max_worker_nodes=2,
),
}
request = sched_request(
node_type_configs=node_type_configs,
)
reply = scheduler.schedule(request)
# No instances created, no-op.
expected_to_terminate = []
_, actual_to_terminate = _launch_and_terminate(reply)
assert sorted(actual_to_terminate) == sorted(expected_to_terminate)
instances = [
make_autoscaler_instance(
ray_node=NodeState(
ray_node_type_name="type_0",
available_resources={"CPU": 1},
total_resources={"CPU": 1},
node_id=b"r0",
),
im_instance=Instance(
instance_type="type_1",
status=Instance.ALLOCATED,
instance_id="0",
node_id="r0",
),
),
make_autoscaler_instance(
ray_node=NodeState(
ray_node_type_name="type_1",
available_resources={"CPU": 1},
total_resources={"CPU": 1},
node_id=b"r1",
),
im_instance=Instance(
instance_type="type_1",
status=Instance.ALLOCATED,
instance_id="1",
node_id="r1",
),
),
]
# 2 nodes in allocated state with max of 2 allowed for type 1.
# Scheduler should leave all of the allocated instances.
request = sched_request(
node_type_configs=node_type_configs,
instances=instances,
)
reply = scheduler.schedule(request)
_, actual_to_terminate = _launch_and_terminate(reply)
assert actual_to_terminate == []
# Max nodes is now 0 for type 1, scheduler should terminate
# both allocated instances to conform with max num nodes per type.
node_type_configs = {
"type_1": NodeTypeConfig(
name="type_1",
resources={"CPU": 1},
min_worker_nodes=0,
max_worker_nodes=0,
),
}
request = sched_request(
node_type_configs=node_type_configs,
instances=instances,
)
reply = scheduler.schedule(request)
_, actual_to_terminate = _launch_and_terminate(reply)
assert sorted(actual_to_terminate) == sorted(
[
(
"0",
"",
TerminationRequest.Cause.MAX_NUM_NODE_PER_TYPE,
), # allocated instance
("1", "", TerminationRequest.Cause.MAX_NUM_NODE_PER_TYPE),
]
)
def test_max_num_nodes():
scheduler = ResourceDemandScheduler(event_logger)
node_type_configs = {
"type_1": NodeTypeConfig(
name="type_1",
resources={"CPU": 1},
min_worker_nodes=0,
max_worker_nodes=2,
),
"type_2": NodeTypeConfig(
name="type_2",
resources={"CPU": 1},
min_worker_nodes=0,
max_worker_nodes=2,
),
}
request = sched_request(
node_type_configs=node_type_configs,
max_num_nodes=1,
)
reply = scheduler.schedule(request)
expected_to_terminate = []
_, actual_to_terminate = _launch_and_terminate(reply)
assert sorted(actual_to_terminate) == sorted(expected_to_terminate)
instances = [
make_autoscaler_instance(
im_instance=Instance(
instance_type="type_1", status=Instance.ALLOCATED, instance_id="0"
),
),
make_autoscaler_instance(
ray_node=NodeState(
ray_node_type_name="type_1",
available_resources={"CPU": 1},
total_resources={"CPU": 1},
node_id=b"r1",
idle_duration_ms=10,
),
im_instance=Instance(
instance_type="type_1",
status=Instance.RAY_RUNNING,
instance_id="1",
node_id="r1",
),
),
make_autoscaler_instance(
ray_node=NodeState(
ray_node_type_name="type_2",
available_resources={"CPU": 0.5},
total_resources={"CPU": 1},
node_id=b"r2",
),
im_instance=Instance(
instance_type="type_2",
status=Instance.RAY_RUNNING,
instance_id="2",
node_id="r2",
),
),
make_autoscaler_instance(
ray_node=NodeState(
ray_node_type_name="type_2",
available_resources={"CPU": 0.0},
total_resources={"CPU": 1},
node_id=b"r3",
),
im_instance=Instance(
instance_type="type_2",
status=Instance.RAY_RUNNING,
instance_id="3",
node_id="r3",
),
),
]
# 4 running with 4 max => no termination
request = sched_request(
node_type_configs=node_type_configs,
instances=instances,
max_num_nodes=4,
)
reply = scheduler.schedule(request)
_, actual_to_terminate = _launch_and_terminate(reply)
assert actual_to_terminate == []
# 4 running with 3 max => terminate 1
request = sched_request(
node_type_configs=node_type_configs,
instances=instances,
max_num_nodes=3,
)
reply = scheduler.schedule(request)
_, actual_to_terminate = _launch_and_terminate(reply)
# Terminate one non-ray running first.
assert actual_to_terminate == [("0", "", TerminationRequest.Cause.MAX_NUM_NODES)]
# 4 running with 2 max => terminate 2
request = sched_request(
node_type_configs=node_type_configs,
instances=instances,
max_num_nodes=2,
)
reply = scheduler.schedule(request)
_, actual_to_terminate = _launch_and_terminate(reply)
# Terminate one non-ray running first.
assert sorted(actual_to_terminate) == sorted(
[
("0", "", TerminationRequest.Cause.MAX_NUM_NODES), # non-ray running
("1", "r1", TerminationRequest.Cause.MAX_NUM_NODES), # idle
]
)
# 4 running with 1 max => terminate 3
request = sched_request(
node_type_configs=node_type_configs,
instances=instances,
max_num_nodes=1,
)
reply = scheduler.schedule(request)
_, actual_to_terminate = _launch_and_terminate(reply)
assert sorted(actual_to_terminate) == sorted(
[
("0", "", TerminationRequest.Cause.MAX_NUM_NODES), # non-ray running
("1", "r1", TerminationRequest.Cause.MAX_NUM_NODES), # idle
("2", "r2", TerminationRequest.Cause.MAX_NUM_NODES), # less resource util
]
)
# Combine max_num_nodes with max_num_nodes_per_type
node_type_configs = {
"type_1": NodeTypeConfig(
name="type_1",
resources={"CPU": 1},
min_worker_nodes=0,
max_worker_nodes=2,
),
"type_2": NodeTypeConfig(
name="type_2",
resources={"CPU": 1},
min_worker_nodes=0,
max_worker_nodes=0,
),
}
request = sched_request(
node_type_configs=node_type_configs,
instances=instances,
max_num_nodes=1,
)
reply = scheduler.schedule(request)
_, actual_to_terminate = _launch_and_terminate(reply)
assert sorted(actual_to_terminate) == sorted(
[
("0", "", TerminationRequest.Cause.MAX_NUM_NODES), # non-ray running
("2", "r2", TerminationRequest.Cause.MAX_NUM_NODE_PER_TYPE), # type-2
("3", "r3", TerminationRequest.Cause.MAX_NUM_NODE_PER_TYPE), # type-2
]
)
def test_single_resources():
scheduler = ResourceDemandScheduler(event_logger)
node_type_configs = {
"type_1": NodeTypeConfig(
name="type_1",
resources={"CPU": 1},
min_worker_nodes=0,
max_worker_nodes=10,
),
}
# Request 1 CPU should start a node.
request = sched_request(
node_type_configs=node_type_configs,
resource_requests=[ResourceRequestUtil.make({"CPU": 1})],
)
reply = scheduler.schedule(request)
to_lauch, _ = _launch_and_terminate(reply)
assert sorted(to_lauch) == sorted({"type_1": 1})
# Request multiple CPUs should start multiple nodes
request = sched_request(
node_type_configs=node_type_configs,
resource_requests=[ResourceRequestUtil.make({"CPU": 1})] * 3,
)
reply = scheduler.schedule(request)
to_lauch, _ = _launch_and_terminate(reply)
assert sorted(to_lauch) == sorted({"type_1": 3})
# Request resources with already existing nodes should not launch new nodes.
request = sched_request(
node_type_configs=node_type_configs,
resource_requests=[ResourceRequestUtil.make({"CPU": 1})],
instances=[
make_autoscaler_instance(
ray_node=NodeState(
ray_node_type_name="type_1",
available_resources={"CPU": 1},
total_resources={"CPU": 1},
node_id=b"r1",
),
im_instance=Instance(
instance_type="type_1",
status=Instance.RAY_RUNNING,
instance_id="1",
node_id="r1",
),
),
],
)
reply = scheduler.schedule(request)
to_lauch, _ = _launch_and_terminate(reply)
assert sorted(to_lauch) == sorted({})
# Request resources with already existing nodes not sufficient should launch
# new nodes.
request = sched_request(
node_type_configs=node_type_configs,
resource_requests=[ResourceRequestUtil.make({"CPU": 1})],
instances=[
make_autoscaler_instance(
ray_node=NodeState(
ray_node_type_name="type_1",
available_resources={"CPU": 0.9},
total_resources={"CPU": 1},
node_id=b"r1",
),
im_instance=Instance(
instance_type="type_1",
status=Instance.RAY_RUNNING,
instance_id="1",
node_id="r1",
),
),
],
)
reply = scheduler.schedule(request)
to_lauch, _ = _launch_and_terminate(reply)
assert sorted(to_lauch) == sorted({"type_1": 1})
# Request resources with already pending nodes should NOT launch new nodes
request = sched_request(
node_type_configs=node_type_configs,
resource_requests=[ResourceRequestUtil.make({"CPU": 1})],
instances=[
make_autoscaler_instance(
im_instance=Instance(
instance_type="type_1", status=Instance.REQUESTED, instance_id="0"
),
),
],
)
reply = scheduler.schedule(request)
to_lauch, _ = _launch_and_terminate(reply)
assert sorted(to_lauch) == sorted({})
def test_implicit_resources():
scheduler = ResourceDemandScheduler(event_logger)
node_type_configs = {
"type_1": NodeTypeConfig(
name="type_1",
resources={"CPU": 1},
min_worker_nodes=0,
max_worker_nodes=10,
),
}
implicit_resource = ray._raylet.IMPLICIT_RESOURCE_PREFIX + "a"
# implicit resources should scale up clusters.
request = sched_request(
node_type_configs=node_type_configs,
resource_requests=[ResourceRequestUtil.make({implicit_resource: 1})],
)
reply = scheduler.schedule(request)
to_launch, _ = _launch_and_terminate(reply)
assert sorted(to_launch) == sorted({"type_1": 1})
# implicit resources should be satisfied by existing node.
request = sched_request(
node_type_configs=node_type_configs,
resource_requests=[
ResourceRequestUtil.make({implicit_resource: 1}),
ResourceRequestUtil.make({"CPU": 1}),
],
instances=[
make_autoscaler_instance(
ray_node=NodeState(
ray_node_type_name="type_1",
available_resources={"CPU": 1},
total_resources={"CPU": 1},
node_id=b"r1",
),
im_instance=Instance(
instance_type="type_1",
status=Instance.RAY_RUNNING,
instance_id="1",
node_id="r1",
),
),
],
)
reply = scheduler.schedule(request)
to_launch, _ = _launch_and_terminate(reply)
assert to_launch == {}
def test_max_worker_num_enforce_with_resource_requests():
scheduler = ResourceDemandScheduler(event_logger)
node_type_configs = {
"type_1": NodeTypeConfig(
name="type_1",
resources={"CPU": 1},
min_worker_nodes=0,
max_worker_nodes=10,
),
}
max_num_nodes = 2
# Request 10 CPUs should start at most 2 nodes.
request = sched_request(
node_type_configs=node_type_configs,
max_num_nodes=max_num_nodes,
resource_requests=[ResourceRequestUtil.make({"CPU": 1})] * 3,
instances=[
make_autoscaler_instance(
ray_node=NodeState(
ray_node_type_name="type_1",
available_resources={"CPU": 1},
total_resources={"CPU": 1},
node_id=b"r1",
),
im_instance=Instance(
instance_type="type_1",
status=Instance.RAY_RUNNING,
instance_id="1",
node_id="r1",
),
),
],
)
reply = scheduler.schedule(request)
to_lauch, _ = _launch_and_terminate(reply)
assert sorted(to_lauch) == sorted({"type_1": 1})
def test_multi_requests_fittable():
"""
Test multiple requests can be fit into a single node.
"""
scheduler = ResourceDemandScheduler(event_logger)
node_type_configs = {
"type_1": NodeTypeConfig(
name="type_1",
resources={"CPU": 1, "GPU": 1},
min_worker_nodes=0,
max_worker_nodes=1,
),
"type_2": NodeTypeConfig(
name="type_2",
resources={"CPU": 3},
min_worker_nodes=0,
max_worker_nodes=1,
),
}
request = sched_request(
node_type_configs=node_type_configs,
resource_requests=[
ResourceRequestUtil.make({"CPU": 1}),
ResourceRequestUtil.make({"CPU": 1}),
ResourceRequestUtil.make({"CPU": 1}),
ResourceRequestUtil.make({"CPU": 1, "GPU": 1}),
],
)
reply = scheduler.schedule(request)
to_launch, _ = _launch_and_terminate(reply)
assert sorted(to_launch) == sorted({"type_1": 1, "type_2": 1})
assert reply.infeasible_resource_requests == []
# Change the ordering of requests should not affect the result.
request = sched_request(
node_type_configs=node_type_configs,
resource_requests=[
ResourceRequestUtil.make({"CPU": 1, "GPU": 1}),
ResourceRequestUtil.make({"CPU": 1}),
ResourceRequestUtil.make({"CPU": 1}),
ResourceRequestUtil.make({"CPU": 1}),
],
)
reply = scheduler.schedule(request)
to_launch, _ = _launch_and_terminate(reply)
assert sorted(to_launch) == sorted({"type_1": 1, "type_2": 1})
assert reply.infeasible_resource_requests == []
request = sched_request(
node_type_configs=node_type_configs,
resource_requests=[
ResourceRequestUtil.make({"CPU": 2}),
ResourceRequestUtil.make({"CPU": 1}),
ResourceRequestUtil.make({"CPU": 0.5, "GPU": 0.5}),
ResourceRequestUtil.make({"CPU": 0.5, "GPU": 0.5}),
],
)
reply = scheduler.schedule(request)
to_launch, _ = _launch_and_terminate(reply)
assert sorted(to_launch) == sorted({"type_1": 1, "type_2": 1})
assert reply.infeasible_resource_requests == []
# However, if we already have fragmentation. We should not be able
# to fit more requests.
request = sched_request(
node_type_configs=node_type_configs,
resource_requests=[
ResourceRequestUtil.make({"CPU": 1}),
ResourceRequestUtil.make({"CPU": 1}),
ResourceRequestUtil.make({"CPU": 1, "GPU": 1}),
],
instances=[
make_autoscaler_instance(
ray_node=NodeState(
ray_node_type_name="type_1",
available_resources={"CPU": 0, "GPU": 1},
total_resources={"CPU": 1, "GPU": 1},
node_id=b"r1",
),
im_instance=Instance(
instance_type="type_1",
status=Instance.RAY_RUNNING,
instance_id="1",
node_id="r1",
),
),
],
)
reply = scheduler.schedule(request)
to_launch, _ = _launch_and_terminate(reply)
assert sorted(to_launch) == sorted({"type_2": 1})
assert len(reply.infeasible_resource_requests) == 1
def test_multi_node_types_score():
"""
Test that when multiple node types are possible, choose the best scoring ones:
1. The number of resources utilized.
2. The amount of utilization.
"""
scheduler = ResourceDemandScheduler(event_logger)
node_type_configs = {
"type_large": NodeTypeConfig(
name="type_large",
resources={"CPU": 10}, # Large machines
min_worker_nodes=0,
max_worker_nodes=1,
),
"type_small": NodeTypeConfig(
name="type_small",
resources={"CPU": 5},
min_worker_nodes=0,
max_worker_nodes=1,
),
"type_gpu": NodeTypeConfig(
name="type_gpu",
resources={"CPU": 2, "GPU": 2},
min_worker_nodes=0,
max_worker_nodes=1,
),
}
# Request 1 CPU should just start the small machine and not the GPU machine
# since it has more types of resources.
request = sched_request(
node_type_configs=node_type_configs,
resource_requests=[ResourceRequestUtil.make({"CPU": 1})],
)
reply = scheduler.schedule(request)
to_launch, _ = _launch_and_terminate(reply)
assert sorted(to_launch) == sorted({"type_small": 1})
# type_small should be preferred over type_large.
request = sched_request(
node_type_configs=node_type_configs,
resource_requests=[ResourceRequestUtil.make({"CPU": 2})],
)
reply = scheduler.schedule(request)
to_launch, _ = _launch_and_terminate(reply)
assert sorted(to_launch) == sorted({"type_small": 1})
def test_multi_node_types_score_with_gpu(monkeypatch):
"""
Test that when multiple node types are possible, choose the best scoring ones:
- The GPU scoring.
"""
scheduler = ResourceDemandScheduler(event_logger)
node_type_configs = {
"type_gpu": NodeTypeConfig(
name="type_gpu",
resources={"CPU": 1, "GPU": 2},
min_worker_nodes=0,
max_worker_nodes=1,
),
"type_multi": NodeTypeConfig(
name="type_multi",
resources={"CPU": 2, "XXX": 2}, # Some random resource.
min_worker_nodes=0,
max_worker_nodes=1,
),
}
request = sched_request(
node_type_configs=node_type_configs,
resource_requests=[ResourceRequestUtil.make({"CPU": 1})],
)
reply = scheduler.schedule(request)
to_launch, _ = _launch_and_terminate(reply)
assert sorted(to_launch) == sorted({"type_multi": 1})
with monkeypatch.context() as m:
m.setattr(ray.autoscaler.v2.scheduler, "AUTOSCALER_CONSERVE_GPU_NODES", 0)
# type_multi should now be preferred over type_gpu.
reply = scheduler.schedule(request)
to_launch, _ = _launch_and_terminate(reply)
assert sorted(to_launch) == sorted({"type_gpu": 1})
def test_resource_constrains():
scheduler = ResourceDemandScheduler(event_logger)
node_type_configs = {
"type_cpu": NodeTypeConfig(
name="type_cpu",
resources={"CPU": 1},
min_worker_nodes=1,
max_worker_nodes=5,
),
"type_gpu": NodeTypeConfig(
name="type_gpu",
resources={"CPU": 1, "GPU": 2},
min_worker_nodes=0,
max_worker_nodes=1,
),
}
# Resource constraints should not launch extra with min_nodes
request = sched_request(
node_type_configs=node_type_configs,
cluster_resource_constraints=[
ResourceRequestUtil.make({"CPU": 1}),
],
)
reply = scheduler.schedule(request)
to_launch, _ = _launch_and_terminate(reply)
assert sorted(to_launch) == sorted({"type_cpu": 1})
# Constraints should launch extra nodes.
request = sched_request(
node_type_configs=node_type_configs,
cluster_resource_constraints=[
ResourceRequestUtil.make({"CPU": 1}),
ResourceRequestUtil.make({"CPU": 1}),
ResourceRequestUtil.make({"GPU": 1}),
],
)
reply = scheduler.schedule(request)
to_launch, _ = _launch_and_terminate(reply)
assert sorted(to_launch) == sorted({"type_cpu": 1, "type_gpu": 1})
# Resource constraints should not launch extra with max_nodes
# fails to atomically ensure constraints.
request = sched_request(
node_type_configs=node_type_configs,
cluster_resource_constraints=[
ResourceRequestUtil.make({"CPU": 1}),
ResourceRequestUtil.make({"CPU": 1}),
ResourceRequestUtil.make({"GPU": 2}),
ResourceRequestUtil.make({"GPU": 2}),
],
)
reply = scheduler.schedule(request)
to_launch, _ = _launch_and_terminate(reply)
assert sorted(to_launch) == sorted({"type_cpu": 1})
assert len(reply.infeasible_cluster_resource_constraints) == 1
@pytest.mark.parametrize(
"disable_launch_config_check", [True, False], ids=["disabled", "enabled"]
)
def test_outdated_nodes(disable_launch_config_check):
"""
Test that nodes with outdated node configs are terminated.
"""
scheduler = ResourceDemandScheduler(event_logger)
node_type_configs = {
"type_cpu": NodeTypeConfig(
name="type_cpu",
resources={"CPU": 1},
min_worker_nodes=2,
max_worker_nodes=5,
launch_config_hash="hash1",
),
"head_node": NodeTypeConfig(
name="head_node",
resources={"CPU": 0},
launch_config_hash="hash2",
min_worker_nodes=0,
max_worker_nodes=1,
),
}
request = sched_request(
node_type_configs=node_type_configs,
disable_launch_config_check=disable_launch_config_check,
instances=[
make_autoscaler_instance(
im_instance=Instance(
instance_type="type_cpu",
status=Instance.RAY_RUNNING,
launch_config_hash="hash2",
instance_id="i-1",
node_id="r-1",
),
ray_node=NodeState(
ray_node_type_name="type_cpu",
available_resources={"CPU": 1},
total_resources={"CPU": 1},
node_id=b"r-1",
),
cloud_instance_id="c-1",
),
make_autoscaler_instance(
im_instance=Instance(
instance_type="type_cpu",
status=Instance.RAY_RUNNING,
launch_config_hash="hash1", # matched
instance_id="i-2",
node_id="r-2",
),
ray_node=NodeState(
ray_node_type_name="type_cpu",
available_resources={"CPU": 1},
total_resources={"CPU": 1},
node_id=b"r-2",
),
cloud_instance_id="c-2",
),
make_autoscaler_instance(
im_instance=Instance(
instance_type="head_node",
status=Instance.RAY_RUNNING,
launch_config_hash="hash1", # mismatched -> but don't terminate
instance_id="i-3",
node_kind=NodeKind.HEAD,
node_id="r-3",
),
ray_node=NodeState(
ray_node_type_name="head_node",
available_resources={"CPU": 0},
total_resources={"CPU": 0},
node_id=b"r-3",
),
cloud_instance_id="c-3",
),
],
)
reply = scheduler.schedule(request)
to_launch, to_terminate = _launch_and_terminate(reply)
if not disable_launch_config_check:
assert to_terminate == [("i-1", "r-1", TerminationRequest.Cause.OUTDATED)]
assert to_launch == {"type_cpu": 1} # Launch 1 to replace the outdated node.
else:
assert to_terminate == []
assert to_launch == {}
@pytest.mark.parametrize("idle_timeout_s", [1, 2, 10])
@pytest.mark.parametrize("has_resource_constraints", [True, False])
@pytest.mark.parametrize("has_resource_requests", [True, False])
@pytest.mark.parametrize("has_gang_resource_requests", [True, False])
def test_idle_termination(
idle_timeout_s,
has_resource_constraints,
has_resource_requests,
has_gang_resource_requests,
):
"""
Test that idle nodes are terminated.
"""
scheduler = ResourceDemandScheduler(event_logger)
node_type_configs = {
"type_cpu": NodeTypeConfig(
name="type_cpu",
resources={"CPU": 1},
min_worker_nodes=0,
max_worker_nodes=5,
launch_config_hash="hash1",
),
"head_node": NodeTypeConfig(
name="head_node",
resources={"CPU": 0},
launch_config_hash="hash2",
min_worker_nodes=0,
max_worker_nodes=1,
),
}
idle_time_s = 5
constraints = []
if has_resource_constraints:
constraints = [ResourceRequestUtil.make({"CPU": 1})] * 2
resource_requests = []
if has_resource_requests:
resource_requests = [ResourceRequestUtil.make({"CPU": 1})] * 2
ANTI_AFFINITY = ResourceRequestUtil.PlacementConstraintType.ANTI_AFFINITY
gang_resource_requests = []
if has_gang_resource_requests:
gang_resource_requests = [
[ # This is a strict spread placement group that requires 2 nodes.
ResourceRequestUtil.make({"CPU": 1}, [(ANTI_AFFINITY, "pg", "")]),
ResourceRequestUtil.make({"CPU": 1}, [(ANTI_AFFINITY, "pg", "")]),
]
]
request = sched_request(
node_type_configs=node_type_configs,
instances=[
make_autoscaler_instance(
im_instance=Instance(
instance_type="type_cpu",
status=Instance.RAY_RUNNING,
launch_config_hash="hash1",
instance_id="i-1",
node_id="r-1",
),
ray_node=NodeState(
node_id=b"r-1",
ray_node_type_name="type_cpu",
available_resources={"CPU": 0},
total_resources={"CPU": 1},
idle_duration_ms=0, # Non idle
status=NodeStatus.RUNNING,
),
cloud_instance_id="c-1",
),
make_autoscaler_instance(
im_instance=Instance(
instance_id="i-2",
instance_type="type_cpu",
status=Instance.RAY_RUNNING,
launch_config_hash="hash1",
node_id="r-2",
),
ray_node=NodeState(
ray_node_type_name="type_cpu",
node_id=b"r-2",
available_resources={"CPU": 1},
total_resources={"CPU": 1},
idle_duration_ms=idle_time_s * 1000,
status=NodeStatus.IDLE,
),
cloud_instance_id="c-2",
),
make_autoscaler_instance(
im_instance=Instance(
instance_id="i-3",
instance_type="head_node",
status=Instance.RAY_RUNNING,
launch_config_hash="hash2",
node_kind=NodeKind.HEAD,
node_id="r-3",
),
ray_node=NodeState(
ray_node_type_name="head_node",
node_id=b"r-3",
available_resources={"CPU": 0},
total_resources={"CPU": 0},
idle_duration_ms=999 * 1000, # idle
status=NodeStatus.IDLE,
),
cloud_instance_id="c-3",
),
],
idle_timeout_s=idle_timeout_s,
cluster_resource_constraints=constraints,
resource_requests=resource_requests,
gang_resource_requests=gang_resource_requests,
)
reply = scheduler.schedule(request)
_, to_terminate = _launch_and_terminate(reply)
if (
idle_timeout_s <= idle_time_s
and not has_resource_constraints
and not has_resource_requests
and not has_gang_resource_requests
):
assert len(to_terminate) == 1
assert to_terminate == [("i-2", "r-2", TerminationRequest.Cause.IDLE)]
else:
assert len(to_terminate) == 0
@pytest.mark.parametrize("min_workers", [0, 1])
def test_idle_termination_with_min_worker(min_workers):
"""
Test that idle nodes are terminated.
"""
idle_timeout_s = 1
scheduler = ResourceDemandScheduler(event_logger)
node_type_configs = {
"type_cpu": NodeTypeConfig(
name="type_cpu",
resources={"CPU": 1},
min_worker_nodes=min_workers,
max_worker_nodes=5,
launch_config_hash="hash1",
),
"head_node": NodeTypeConfig(
name="head_node",
resources={"CPU": 0},
launch_config_hash="hash2",
min_worker_nodes=0,
max_worker_nodes=1,
),
}
idle_time_s = 5
constraints = []
request = sched_request(
node_type_configs=node_type_configs,
instances=[
make_autoscaler_instance(
im_instance=Instance(
instance_id="i-1",
instance_type="type_cpu",
status=Instance.RAY_RUNNING,
launch_config_hash="hash1",
node_id="r-1",
),
ray_node=NodeState(
ray_node_type_name="type_cpu",
node_id=b"r-1",
available_resources={"CPU": 1},
total_resources={"CPU": 1},
idle_duration_ms=idle_time_s * 1000,
status=NodeStatus.IDLE,
),
cloud_instance_id="c-1",
),
make_autoscaler_instance(
im_instance=Instance(
instance_id="i-2",
instance_type="head_node",
status=Instance.RAY_RUNNING,
launch_config_hash="hash2",
node_kind=NodeKind.HEAD,
node_id="r-2",
),
ray_node=NodeState(
ray_node_type_name="head_node",
node_id=b"r-2",
available_resources={"CPU": 0},
total_resources={"CPU": 0},
idle_duration_ms=999 * 1000, # idle
status=NodeStatus.IDLE,
),
cloud_instance_id="c-2",
),
],
idle_timeout_s=idle_timeout_s,
cluster_resource_constraints=constraints,
)
reply = scheduler.schedule(request)
_, to_terminate = _launch_and_terminate(reply)
assert idle_timeout_s <= idle_time_s
if min_workers == 0:
assert len(to_terminate) == 1
assert to_terminate == [("i-1", "r-1", TerminationRequest.Cause.IDLE)]
else:
assert min_workers > 0
assert len(to_terminate) == 0
@pytest.mark.parametrize("node_type_idle_timeout_s", [1, 2, 10])
def test_idle_termination_with_node_type_idle_timeout(node_type_idle_timeout_s):
"""
Test that idle nodes are terminated when idle_timeout_s is set for node type.
"""
scheduler = ResourceDemandScheduler(event_logger)
node_type_configs = {
"type_cpu_with_idle_timeout": NodeTypeConfig(
name="type_cpu",
resources={"CPU": 1},
min_worker_nodes=0,
max_worker_nodes=5,
idle_timeout_s=node_type_idle_timeout_s,
launch_config_hash="hash1",
),
}
idle_time_s = 5
constraints = []
request = sched_request(
node_type_configs=node_type_configs,
instances=[
make_autoscaler_instance(
im_instance=Instance(
instance_type="type_cpu_with_idle_timeout",
status=Instance.RAY_RUNNING,
launch_config_hash="hash1",
instance_id="i-1",
node_id="r-1",
),
ray_node=NodeState(
node_id=b"r-1",
ray_node_type_name="type_cpu_with_idle_timeout",
available_resources={"CPU": 0},
total_resources={"CPU": 1},
idle_duration_ms=0, # Non idle
status=NodeStatus.RUNNING,
),
cloud_instance_id="c-1",
),
make_autoscaler_instance(
im_instance=Instance(
instance_id="i-2",
instance_type="type_cpu_with_idle_timeout",
status=Instance.RAY_RUNNING,
launch_config_hash="hash1",
node_id="r-2",
),
ray_node=NodeState(
ray_node_type_name="type_cpu_with_idle_timeout",
node_id=b"r-2",
available_resources={"CPU": 1},
total_resources={"CPU": 1},
idle_duration_ms=idle_time_s * 1000,
status=NodeStatus.IDLE,
),
cloud_instance_id="c-2",
),
],
# Set autoscaler idle_timeout_s to a value greater than
# node_type_idle_timeout_s and idle_time_s.
idle_timeout_s=idle_time_s * 1000,
cluster_resource_constraints=constraints,
)
reply = scheduler.schedule(request)
_, to_terminate = _launch_and_terminate(reply)
if node_type_idle_timeout_s <= idle_time_s:
assert len(to_terminate) == 1
assert to_terminate == [("i-2", "r-2", TerminationRequest.Cause.IDLE)]
else:
assert len(to_terminate) == 0
def test_gang_scheduling():
"""
Test that gang scheduling works.
"""
scheduler = ResourceDemandScheduler(event_logger)
AFFINITY = ResourceRequestUtil.PlacementConstraintType.AFFINITY
ANTI_AFFINITY = ResourceRequestUtil.PlacementConstraintType.ANTI_AFFINITY
node_type_configs = {
"type_cpu": NodeTypeConfig(
name="type_cpu",
resources={"CPU": 2},
min_worker_nodes=0,
max_worker_nodes=5,
launch_config_hash="hash1",
)
}
request = sched_request(
node_type_configs=node_type_configs,
gang_resource_requests=[
[
ResourceRequestUtil.make({"CPU": 1}, [(AFFINITY, "pg", "")]),
ResourceRequestUtil.make({"CPU": 1}, [(AFFINITY, "pg", "")]),
]
],
)
reply = scheduler.schedule(request)
to_launch, _ = _launch_and_terminate(reply)
# Should be grouped on the same node.
assert sorted(to_launch) == sorted({"type_cpu": 1})
request = sched_request(
node_type_configs=node_type_configs,
gang_resource_requests=[
[
ResourceRequestUtil.make({"CPU": 1}, [(ANTI_AFFINITY, "pg", "")]),
ResourceRequestUtil.make({"CPU": 1}, [(ANTI_AFFINITY, "pg", "")]),
]
],
)
reply = scheduler.schedule(request)
to_launch, _ = _launch_and_terminate(reply)
# Should be placed on different nodes.
assert sorted(to_launch) == sorted({"type_cpu": 2})
# Atomic gang scheduling
request = sched_request(
node_type_configs=node_type_configs,
gang_resource_requests=[
[
# Couldn't fit on a node.
ResourceRequestUtil.make({"CPU": 3}, [(AFFINITY, "pg", "")]),
ResourceRequestUtil.make({"CPU": 3}, [(AFFINITY, "pg", "")]),
]
],
)
reply = scheduler.schedule(request)
to_launch, _ = _launch_and_terminate(reply)
assert to_launch == {}
assert len(reply.infeasible_gang_resource_requests) == 1
def test_gang_scheduling_with_others():
"""
Test that a mix of the various demands:
- resource requests from tasks/actors
- gang requests from placement groups
- cluster resource constraints
- min/max worker counts
- existing nodes.
"""
scheduler = ResourceDemandScheduler(event_logger)
node_type_configs = {
"type_1": NodeTypeConfig(
name="type_1",
resources={"CPU": 4},
min_worker_nodes=2,
max_worker_nodes=4,
launch_config_hash="hash1",
),
"type_2": NodeTypeConfig(
name="type_2",
resources={"CPU": 1, "GPU": 1},
min_worker_nodes=0,
max_worker_nodes=10,
launch_config_hash="hash2",
),
}
# Placement constraints
AFFINITY = ResourceRequestUtil.PlacementConstraintType.AFFINITY
ANTI_AFFINITY = ResourceRequestUtil.PlacementConstraintType.ANTI_AFFINITY
gang_requests = [
[
ResourceRequestUtil.make({"CPU": 2}, [(ANTI_AFFINITY, "ak", "av")]),
ResourceRequestUtil.make({"CPU": 2}, [(ANTI_AFFINITY, "ak", "av")]),
ResourceRequestUtil.make({"CPU": 2}, [(ANTI_AFFINITY, "ak", "av")]),
ResourceRequestUtil.make({"CPU": 2}, [(ANTI_AFFINITY, "ak", "av")]),
],
[
ResourceRequestUtil.make({"CPU": 3}, [(AFFINITY, "c", "c1")]),
ResourceRequestUtil.make({"CPU": 3}, [(AFFINITY, "c", "c1")]),
],
[
ResourceRequestUtil.make({"CPU": 1}),
ResourceRequestUtil.make({"CPU": 1}),
ResourceRequestUtil.make({"CPU": 1}),
],
]
# Resource requests
resource_requests = [
ResourceRequestUtil.make({"CPU": 2}),
ResourceRequestUtil.make({"GPU": 1, "CPU": 1}),
ResourceRequestUtil.make({"GPU": 1}),
]
# Cluster constraints
cluster_constraints = [ResourceRequestUtil.make({"CPU": 1})] * 10
instances = [
make_autoscaler_instance(
im_instance=Instance(
instance_type="type_1",
status=Instance.RAY_RUNNING,
launch_config_hash="hash1",
instance_id="i-1",
),
ray_node=NodeState(
node_id=b"r-1",
ray_node_type_name="type_1",
available_resources={"CPU": 2},
total_resources={"CPU": 4},
idle_duration_ms=0,
status=NodeStatus.RUNNING,
),
cloud_instance_id="c-1",
),
make_autoscaler_instance(
im_instance=Instance(
instance_type="type_2",
status=Instance.RAY_RUNNING,
launch_config_hash="hash2",
instance_id="i-2",
),
ray_node=NodeState(
node_id=b"r-2",
ray_node_type_name="type_2",
available_resources={"CPU": 1, "GPU": 1},
total_resources={"CPU": 1, "GPU": 1},
idle_duration_ms=0,
status=NodeStatus.RUNNING,
),
cloud_instance_id="c-2",
),
]
request = sched_request(
node_type_configs=node_type_configs,
gang_resource_requests=gang_requests,
resource_requests=resource_requests,
cluster_resource_constraints=cluster_constraints,
instances=instances,
idle_timeout_s=999,
)
# Calculate the expected number of nodes to launch:
# - 1 type_1, 1 type_2 to start with => CPU: 2/5, GPU: 1/1
# - added 1 type_1 for minimal request -> +1 type_1
# ==> 2 type_1, 1 type_2 (CPU: 6/9, GPU: 1/1)
# - enforce cluster constraint (10 CPU) -> +1 type_1, CPU: 10/13, GPU: 1/1
# ==> 3 type_1, 1 type_2 (CPU: 10/13, GPU: 1/1)
# - sched gang requests:
# - anti affinity (8CPU) => +1 type_1, CPU: 6/17, GPU: 1/1
# - no constraint (3CPU) => CPU: 3/17, GPU: 1/1
# - affinity (not feasible)
# ==> 4 type_1, 1 type_2 (CPU: 3/17, GPU: 1/1)
# - sched resource requests:
# - 2CPU => CPU: 1/17, GPU: 1/1
# - 1GPU, 1CPU => CPU: 0/17, GPU: 0/1
# - 1GPU => adding a new type_2
# ==> 4 type_1, 2 type_2 (CPU: 0/17, GPU: 0/2)
# Therefore:
# - added nodes: 3 type_1, 1 type_2
# - infeasible: 1 gang request, 1 resource request
expected_to_launch = {"type_1": 3, "type_2": 1}
reply = scheduler.schedule(request)
to_launch, _ = _launch_and_terminate(reply)
assert sorted(to_launch) == sorted(expected_to_launch)
assert len(reply.infeasible_gang_resource_requests) == 1
assert len(reply.infeasible_resource_requests) == 0
def test_bin_pack():
def bin_pack_residual(
node_resources: Dict[NodeType, Dict],
resource_requests: List[Dict],
anti_affinity: bool = False,
) -> List[Dict]:
node_type_configs = {}
for type_name, node_resource_dict in node_resources.items():
node_type_configs[type_name] = NodeTypeConfig(
name=type_name,
resources=node_resource_dict,
min_worker_nodes=0,
max_worker_nodes=1, # we only care about bin packing. Just allow 1
)
reply = schedule(node_type_configs, {}, resource_requests, anti_affinity)
if anti_affinity:
infeasible = []
for r in reply.infeasible_gang_resource_requests:
infeasible.append(ResourceRequestUtil.to_resource_maps(r.requests))
return infeasible
else:
return ResourceRequestUtil.to_resource_maps(
reply.infeasible_resource_requests
)
assert bin_pack_residual({"type_1": {"CPU": 0}}, [{"GPU": 2}, {"GPU": 2}]) == [
{"GPU": 2},
{"GPU": 2},
]
assert bin_pack_residual({"type_1": {"GPU": 2}}, [{"GPU": 2}, {"GPU": 2}]) == [
{"GPU": 2}
]
assert bin_pack_residual({"type_1": {"GPU": 4}}, [{"GPU": 2}, {"GPU": 2}]) == []
assert (
bin_pack_residual(
{"type_1": {"GPU": 2}, "type_2": {"GPU": 2, "CPU": 2}},
[{"GPU": 2}, {"GPU": 2}],
)
== []
)
assert bin_pack_residual(
{"type_1": {"GPU": 2}, "type_2": {"CPU": 2}},
[{"GPU": 2}, {"GPU": 2}],
) == [{"GPU": 2}]
assert bin_pack_residual(
{"type_1": {"GPU": 2}, "type_2": {"CPU": 2}},
[{"GPU": 2}, {"GPU": 2}],
) == [{"GPU": 2}]
assert bin_pack_residual(
{"type_1": {"GPU": 3}},
[{"GPU": 1}, {"GPU": 1}],
anti_affinity=True,
) == [[{"GPU": 1.0}, {"GPU": 1.0}]]
assert (
bin_pack_residual(
{"type_1": {"GPU": 3}},
[{"GPU": 1}, {"GPU": 1}],
anti_affinity=False,
)
== []
)
implicit_resource = ray._raylet.IMPLICIT_RESOURCE_PREFIX + "a"
assert (
bin_pack_residual(
{"type_1": {"CPU": 1}},
[{implicit_resource: 0.5}, {implicit_resource: 0.5}],
)
== []
)
assert bin_pack_residual(
{"type_1": {"CPU": 1}},
[{implicit_resource: 1}, {implicit_resource: 0.5}],
) == [{implicit_resource: 0.5}]
@pytest.mark.parametrize(
"source",
[
ResourceRequestSource.PENDING_DEMAND,
ResourceRequestSource.CLUSTER_RESOURCE_CONSTRAINT,
],
ids=["demand", "cluster_resource_constraint"],
)
def test_node_schedule_score(source):
def try_schedule(node_resources: Dict, requests: List[Dict]) -> Tuple:
node_type_config = NodeTypeConfig(
name="type_1",
resources=node_resources,
min_worker_nodes=0,
max_worker_nodes=1,
)
node = SchedulingNode.from_node_config(
node_config=node_type_config,
status=SchedulingNodeStatus.SCHEDULABLE,
node_kind=NodeKind.WORKER,
)
requests = [ResourceRequestUtil.make(r) for r in requests]
infeasible, score = node.try_schedule(requests, source)
return ResourceRequestUtil.to_resource_maps(infeasible), score
assert try_schedule({"CPU": 1}, [{"CPU": 1}]) == ([], (0, True, 1, 1.0, 1.0))
assert try_schedule({"GPU": 4}, [{"GPU": 2}]) == ([], (0, True, 1, 0.5, 0.5))
assert try_schedule({"GPU": 4}, [{"GPU": 1}, {"GPU": 1}]) == (
[],
(0, True, 1, 0.5, 0.5),
)
assert try_schedule({"GPU": 2}, [{"GPU": 2}]) == ([], (0, True, 1, 2, 2))
assert try_schedule({"GPU": 2}, [{"GPU": 1}, {"GPU": 1}]) == (
[],
(0, True, 1, 2, 2),
)
assert try_schedule({"GPU": 1}, [{"GPU": 1, "CPU": 1}, {"GPU": 1}]) == (
[{"GPU": 1, "CPU": 1}],
(0, True, 1, 1, 1),
)
assert try_schedule({"GPU": 1, "CPU": 1}, [{"GPU": 1, "CPU": 1}, {"GPU": 1}]) == (
[{"GPU": 1}],
(0, True, 2, 1, 1),
)
assert try_schedule({"GPU": 2, "TPU": 1}, [{"GPU": 2}]) == ([], (0, True, 1, 0, 1))
assert try_schedule({"CPU": 64}, [{"CPU": 64}]) == ([], (0, True, 1, 64, 64))
assert try_schedule({"CPU": 64}, [{"CPU": 32}]) == ([], (0, True, 1, 8, 8))
assert try_schedule({"CPU": 64}, [{"CPU": 16}, {"CPU": 16}]) == (
[],
(0, True, 1, 8, 8),
)
# GPU Scores
assert try_schedule({"GPU": 1, "CPU": 1}, [{"CPU": 1}]) == (
[],
(0, False, 1, 0.0, 0.5),
)
assert try_schedule({"GPU": 1, "CPU": 1}, [{"CPU": 1, "GPU": 1}]) == (
[],
(0, True, 2, 1.0, 1.0),
)
assert try_schedule({"GPU": 1, "CPU": 1}, [{"GPU": 1}]) == (
[],
(0, True, 1, 0.0, 0.5),
)
# Zero resources
assert try_schedule({"CPU": 0, "custom": 1}, [{"custom": 1}]) == (
[],
(0, True, 1, 1, 1),
)
assert try_schedule({"CPU": 0, "custom": 1}, [{"CPU": 1}]) == (
[{"CPU": 1}],
(0, True, 0, 0.0, 0.0),
)
# Implicit resources
implicit_resource = ray._raylet.IMPLICIT_RESOURCE_PREFIX + "a"
assert try_schedule({"CPU": 1}, [{implicit_resource: 1}]) == (
[],
(0, True, 0, 0.0, 0.0),
)
assert try_schedule({"CPU": 1}, [{implicit_resource: 1}] * 2) == (
[{implicit_resource: 1}],
(0, True, 0, 0.0, 0.0),
)
@pytest.mark.parametrize(
"source",
[
ResourceRequestSource.PENDING_DEMAND,
ResourceRequestSource.CLUSTER_RESOURCE_CONSTRAINT,
],
ids=["demand", "cluster_resource_constraint"],
)
def test_node_schedule_label_selector_score(source):
def try_schedule_ls(
node_resources: Dict,
node_labels: Dict[str, str],
selectors,
) -> Tuple:
cfg = NodeTypeConfig(
name="type_1",
resources=node_resources,
min_worker_nodes=0,
max_worker_nodes=1,
labels=node_labels,
)
node = SchedulingNode.from_node_config(
node_config=cfg,
status=SchedulingNodeStatus.SCHEDULABLE,
node_kind=NodeKind.WORKER,
)
req = ResourceRequestUtil.make({"CPU": 1}, label_selectors=selectors)
infeasible, score = node.try_schedule([req], source)
return ResourceRequestUtil.to_resource_maps(infeasible), score
labels = {"ray.io/accelerator-type": "A100"}
# 1) A matching label selector should be schedulable on node type_1
label_selector_1 = [
[
(
"ray.io/accelerator-type",
LabelSelectorOperator.LABEL_OPERATOR_IN,
["TPU-v6e"],
)
],
[
(
"ray.io/accelerator-type",
LabelSelectorOperator.LABEL_OPERATOR_IN,
["B200"],
)
],
[
(
"ray.io/accelerator-type",
LabelSelectorOperator.LABEL_OPERATOR_IN,
["A100"],
)
],
]
assert try_schedule_ls({"CPU": 1}, labels, label_selector_1) == (
[],
(1, True, 1, 1.0, 1.0),
)
# 2) A non‑matching label selector should be infeasible
label_selector_2 = [
[("ray.io/accelerator-type", LabelSelectorOperator.LABEL_OPERATOR_IN, ["B200"])]
]
assert try_schedule_ls({"CPU": 1}, labels, label_selector_2) == (
[{"CPU": 1.0}],
(0, True, 0, 0.0, 0.0),
)
def test_get_nodes_packing_heuristic():
node_type_configs = {
"m4.large": NodeTypeConfig(
name="m4.large",
resources={"CPU": 2},
min_worker_nodes=0,
max_worker_nodes=10,
),
"m4.4xlarge": NodeTypeConfig(
name="m4.4xlarge",
resources={"CPU": 16},
min_worker_nodes=0,
max_worker_nodes=8,
),
"m4.16xlarge": NodeTypeConfig(
name="m4.16xlarge",
resources={"CPU": 64},
min_worker_nodes=0,
max_worker_nodes=4,
),
"p2.xlarge": NodeTypeConfig(
name="p2.xlarge",
resources={"CPU": 16, "GPU": 1},
min_worker_nodes=0,
max_worker_nodes=10,
),
"p2.8xlarge": NodeTypeConfig(
name="p2.8xlarge",
resources={"CPU": 32, "GPU": 8},
min_worker_nodes=0,
max_worker_nodes=4,
),
}
def get_nodes_for(
resource_requests,
anti_affinity=False,
max_nodes: Optional[int] = None,
current_nodes: Optional[Dict] = None,
):
reply = schedule(
node_type_configs,
current_nodes or {},
resource_requests,
anti_affinity=anti_affinity,
max_nodes=max_nodes,
)
to_launch, _ = _launch_and_terminate(reply)
return to_launch
assert get_nodes_for([{"GPU": 8}]) == {"p2.8xlarge": 1}
assert get_nodes_for([{"GPU": 1}] * 6) == {"p2.8xlarge": 1}
assert get_nodes_for([{"GPU": 1}] * 4) == {"p2.xlarge": 4}
assert get_nodes_for([{"CPU": 32, "GPU": 1}] * 3) == {"p2.8xlarge": 3}
assert get_nodes_for([{"CPU": 64, "GPU": 1}] * 3) == {}
assert get_nodes_for([{"CPU": 64}] * 3) == {"m4.16xlarge": 3}
assert get_nodes_for([{"CPU": 64}, {"CPU": 1}]) == {
"m4.16xlarge": 1,
"m4.large": 1,
}
assert get_nodes_for([{"CPU": 64}, {"CPU": 9}, {"CPU": 9}]) == {
"m4.16xlarge": 1,
"m4.4xlarge": 2,
}
assert get_nodes_for([{"CPU": 16}] * 5) == {
"m4.16xlarge": 1,
"m4.4xlarge": 1,
}
assert get_nodes_for([{"CPU": 8}] * 10) == {
"m4.16xlarge": 1,
"m4.4xlarge": 1,
}
assert get_nodes_for([{"CPU": 1}] * 100) == {
"m4.16xlarge": 1,
"m4.4xlarge": 2,
"m4.large": 2,
}
assert get_nodes_for([{"GPU": 1}] + ([{"CPU": 1}] * 64)) == {
"m4.16xlarge": 1,
"p2.xlarge": 1,
}
assert get_nodes_for(([{"GPU": 1}] * 8) + ([{"CPU": 1}] * 64)) == {
"m4.4xlarge": 2,
"p2.8xlarge": 1,
}
assert get_nodes_for([{"GPU": 1}] * 8, anti_affinity=False) == {"p2.8xlarge": 1}
assert get_nodes_for([{"GPU": 1}] * 8, anti_affinity=True) == {"p2.xlarge": 8}
# GPU/CPU scheduling
node_type_configs = {
"cpu": NodeTypeConfig(
name="cpu",
resources={"CPU": 16},
min_worker_nodes=0,
max_worker_nodes=10,
),
"gpu": NodeTypeConfig(
name="gpu",
resources={"CPU": 16, "GPU": 1},
min_worker_nodes=0,
max_worker_nodes=10,
),
}
assert get_nodes_for([{"CPU": 16}]) == {"cpu": 1}
assert get_nodes_for([{"CPU": 1}] * 30 + [{"GPU": 1, "CPU": 1}]) == {
"cpu": 1,
"gpu": 1,
}
assert get_nodes_for([{"GPU": 1, "CPU": 1}] + [{"CPU": 1}] * 30) == {
"cpu": 1,
"gpu": 1,
}
assert get_nodes_for([{"GPU": 1, "CPU": 1}] + [{"CPU": 1}] * 15) == {
"gpu": 1,
}
# GPU should be avoided
node_type_configs = {
"cpu": NodeTypeConfig(
name="cpu",
resources={"CPU": 1},
min_worker_nodes=0,
max_worker_nodes=10,
),
"gpu": NodeTypeConfig(
name="gpu",
resources={"CPU": 100, "GPU": 1},
min_worker_nodes=0,
max_worker_nodes=10,
),
}
assert get_nodes_for([{"CPU": 1}] * 100, max_nodes=10) == {"cpu": 10}
# max_to_add eleven nodes allowed. First ten chosen to be "cpu",
# last chosen to be "gpu" due max_workers constraint on "cpu".
assert get_nodes_for([{"CPU": 1}] * 100, max_nodes=11) == {"cpu": 10, "gpu": 1}
assert get_nodes_for([{"CPU": 1}] * 100 + [{"GPU": 1}], max_nodes=100) == {"gpu": 1}
assert get_nodes_for([{"GPU": 1}] * 4 + [{"CPU": 1}] * 404, max_nodes=100) == {
"gpu": 4,
"cpu": 4,
}
# Max limit should be respected
node_type_configs = {
"m4.large": NodeTypeConfig(
name="m4.large",
resources={"CPU": 2},
min_worker_nodes=0,
max_worker_nodes=10,
),
}
assert get_nodes_for([{"CPU": 1}] * 10, max_nodes=2) == {"m4.large": 2}
assert (
get_nodes_for([{"CPU": 1}] * 10, max_nodes=10, current_nodes={"m4.large": 10})
== {}
)
assert get_nodes_for([{"CPU": 1}] * 10, max_nodes=10) == {"m4.large": 5}
assert get_nodes_for([{"CPU": 1}] * 40) == {"m4.large": 10}
# Min workers should be respected
node_type_configs = {
"m2.large": NodeTypeConfig(
name="m2.large",
resources={"CPU": 1},
min_worker_nodes=5,
max_worker_nodes=10,
),
"m4.large": NodeTypeConfig(
name="m4.large",
resources={"CPU": 2},
min_worker_nodes=0,
max_worker_nodes=10,
),
"gpu": NodeTypeConfig(
name="gpu",
resources={"GPU": 2},
min_worker_nodes=2,
max_worker_nodes=2,
),
"gpubla": NodeTypeConfig(
name="gpubla", resources={"GPU": 1}, min_worker_nodes=0, max_worker_nodes=0
),
}
assert get_nodes_for([{"CPU": 2}] * 5) == {"m2.large": 5, "m4.large": 5, "gpu": 2}
assert get_nodes_for(
[{"CPU": 2}] * 5, current_nodes={"m2.large": 1, "m4.large": 1}
) == {"m2.large": 4, "m4.large": 4, "gpu": 2}
assert get_nodes_for([{"GPU": 1}] * 5) == {"m2.large": 5, "gpu": 2}
def test_min_workers_and_others():
node_type_configs = {
"p2.8xlarge": NodeTypeConfig(
name="p2.8xlarge",
resources={"CPU": 32, "GPU": 8},
min_worker_nodes=2,
max_worker_nodes=4,
),
"p2.xlarge": NodeTypeConfig(
name="p2.xlarge",
resources={"CPU": 16, "GPU": 1},
min_worker_nodes=0,
max_worker_nodes=10,
),
}
def get_nodes_for(resource_requests, current_nodes=None, max_nodes=None):
reply = schedule(
node_type_configs,
current_nodes or {},
resource_requests,
max_nodes=max_nodes,
)
to_launch, _ = _launch_and_terminate(reply)
infeasible = ResourceRequestUtil.to_resource_maps(
reply.infeasible_resource_requests
)
return to_launch, infeasible
assert get_nodes_for([{"GPU": 8}]) == ({"p2.8xlarge": 2}, [])
assert get_nodes_for([{"GPU": 8}] * 2) == ({"p2.8xlarge": 2}, [])
assert get_nodes_for([{"GPU": 8}] * 4) == ({"p2.8xlarge": 4}, [])
assert get_nodes_for([{"GPU": 8}] * 8) == ({"p2.8xlarge": 4}, [{"GPU": 8}] * 4)
assert get_nodes_for(
[{"GPU": 8}] * 3 + [{"GPU": 1}], current_nodes={"p2.8xlarge": 1}
) == ({"p2.xlarge": 1, "p2.8xlarge": 2}, [])
assert get_nodes_for(
[{"GPU": 8}] * 3 + [{"GPU": 1}], current_nodes={"p2.8xlarge": 2}
) == ({"p2.xlarge": 1, "p2.8xlarge": 1}, [])
assert get_nodes_for(
[{"GPU": 8}] * 3 + [{"GPU": 1}], current_nodes={"p2.8xlarge": 3}
) == ({"p2.xlarge": 1}, [])
assert get_nodes_for(
[{"GPU": 8}] * 5 + [{"GPU": 1}], current_nodes={"p2.8xlarge": 3}
) == ({"p2.xlarge": 1, "p2.8xlarge": 1}, [{"GPU": 8}])
node_type_configs = {
"p2.xlarge": NodeTypeConfig(
name="p2.xlarge",
resources={"CPU": 16, "GPU": 1},
min_worker_nodes=0,
max_worker_nodes=10,
),
}
assert get_nodes_for([{"GPU": 1}] * 4, max_nodes=1) == (
{"p2.xlarge": 1},
[{"GPU": 1}] * 3,
)
assert get_nodes_for([{"GPU": 1}] * 4, max_nodes=2) == (
{"p2.xlarge": 2},
[{"GPU": 1}] * 2,
)
assert get_nodes_for([{"GPU": 1}] * 4, max_nodes=4) == ({"p2.xlarge": 4}, [])
def test_gang_scheduling_complex():
node_type_configs = {
"m4.large": NodeTypeConfig(
name="m4.large",
resources={"CPU": 2},
min_worker_nodes=0,
max_worker_nodes=10,
),
"p2.8xlarge": NodeTypeConfig(
name="p2.8xlarge",
resources={"CPU": 32, "GPU": 8},
min_worker_nodes=0,
max_worker_nodes=4,
),
}
ANTI_AFFINITY = ResourceRequestUtil.PlacementConstraintType.ANTI_AFFINITY
AFFINITY = ResourceRequestUtil.PlacementConstraintType.AFFINITY
def get_nodes_for(gang_resource_requests) -> Tuple[Dict, List[List[Dict]]]:
scheduler = ResourceDemandScheduler(event_logger)
gang_requests = []
for resource_requests, placement_constraint in gang_resource_requests:
key = f"PG_{str(time.time())}"
gang_requests.append(
[
ResourceRequestUtil.make(
r,
[(placement_constraint, key, key)]
if placement_constraint
else [],
)
for r in resource_requests
]
)
request = sched_request(
node_type_configs=node_type_configs,
gang_resource_requests=gang_requests,
)
reply = scheduler.schedule(request)
to_launch, _ = _launch_and_terminate(reply)
infeasible = []
for r in reply.infeasible_gang_resource_requests:
infeasible.append(ResourceRequestUtil.to_resource_maps(r.requests))
return to_launch, infeasible
# Test various constraints.
get_nodes_for([([{"CPU": 16}, {"CPU": 16}], ANTI_AFFINITY)]) == (
{"p2.8xlarge": 2},
[],
)
get_nodes_for([([{"CPU": 16}, {"CPU": 16}], None)]) == (
{"p2.8xlarge": 1},
[],
)
get_nodes_for([([{"CPU": 2}, {"CPU": 2}], AFFINITY)]) == (
{"p2.8xlarge": 1},
[],
)
get_nodes_for([([{"CPU": 2}, {"CPU": 2}], None)]) == (
{"m4.large": 2},
[],
)
get_nodes_for([([{"CPU": 32}, {"CPU": 32}], AFFINITY)]) == (
{},
[[{"CPU": 32}, {"CPU": 32}]],
)
# Test many anti-affinity
get_nodes_for(
[
([{"CPU": 4}, {"CPU": 4}], ANTI_AFFINITY),
([{"CPU": 4}, {"CPU": 4}], ANTI_AFFINITY),
([{"CPU": 4}, {"CPU": 4}], ANTI_AFFINITY),
([{"CPU": 4}, {"CPU": 4}], ANTI_AFFINITY),
]
) == ({"p2.8xlarge": 2}, [])
# Test multiple affinity
get_nodes_for(
[
([{"CPU": 16}, {"CPU": 16}], AFFINITY),
([{"GPU": 4}, {"GPU": 4}], AFFINITY),
]
) == ({"p2.8xlarge": 1}, [])
def test_schedule_node_with_matching_labels():
"""
Test that a node with matching labels is considered schedulable and used to satisfy a request
with a label_selector.
"""
scheduler = ResourceDemandScheduler(event_logger)
node_type_configs = {
"labelled_node": NodeTypeConfig(
name="labelled_node",
resources={"CPU": 1},
min_worker_nodes=0,
max_worker_nodes=10,
labels={"accelerator": "A100"},
),
}
# The existing instance has matching dynamic label.
instance = make_autoscaler_instance(
im_instance=Instance(
instance_type="labelled_node",
status=Instance.RAY_RUNNING,
instance_id="1",
node_id=b"r-1",
),
ray_node=NodeState(
node_id=b"r-1",
ray_node_type_name="labelled_node",
available_resources={"CPU": 1},
total_resources={"CPU": 1},
labels={"accelerator": "A100"},
status=NodeStatus.RUNNING,
),
cloud_instance_id="c-1",
)
# No new nodes should be launched if the existing node satisfies the request.
resource_request = ResourceRequestUtil.make(
{"CPU": 1},
label_selectors=[
[("accelerator", LabelSelectorOperator.LABEL_OPERATOR_IN, ["A100"])]
],
)
request = sched_request(
node_type_configs=node_type_configs,
resource_requests=[resource_request],
instances=[instance],
)
reply = scheduler.schedule(request)
to_launch, _ = _launch_and_terminate(reply)
assert to_launch == {}
def test_scale_up_node_to_satisfy_labels():
"""
Test that a resource request with a label selector scales up a new node with
labels to satisfy the constraint.
"""
scheduler = ResourceDemandScheduler(event_logger)
node_type_configs = {
"tpu_node": NodeTypeConfig(
name="tpu_node",
resources={"CPU": 1},
labels={"accelerator": "TPU"},
min_worker_nodes=0,
max_worker_nodes=10,
),
"gpu_node": NodeTypeConfig(
name="gpu_node",
resources={"CPU": 1},
labels={"accelerator": "A100"},
min_worker_nodes=0,
max_worker_nodes=10,
),
}
# Request: want a node with label "accelerator: A100"
resource_request = ResourceRequestUtil.make(
{"CPU": 1},
label_selectors=[
[("accelerator", LabelSelectorOperator.LABEL_OPERATOR_IN, ["A100"])]
],
)
request = sched_request(
node_type_configs=node_type_configs,
resource_requests=[resource_request],
)
reply = scheduler.schedule(request)
to_launch, _ = _launch_and_terminate(reply)
assert to_launch == {"gpu_node": 1}
def test_label_selector_fallback_priority():
"""
Test that a resource request with multiple label selectors scales up
the expected node given its fallback priority (i.e. earlier selectors are
satisfied first).
"""
scheduler = ResourceDemandScheduler(event_logger)
node_type_configs = {
"tpu_node": NodeTypeConfig(
name="tpu_node",
resources={"CPU": 1},
labels={"accelerator-type": "TPU"},
min_worker_nodes=0,
max_worker_nodes=10,
),
"gpu_node": NodeTypeConfig(
name="gpu_node",
resources={"CPU": 1},
labels={"accelerator-type": "A100"},
min_worker_nodes=0,
max_worker_nodes=10,
),
}
# 1). TPU node is scaled up to satisfy first label selector.
req1 = ResourceRequestUtil.make(
{"CPU": 1},
label_selectors=[
[("accelerator-type", LabelSelectorOperator.LABEL_OPERATOR_IN, ["TPU"])],
[("accelerator-type", LabelSelectorOperator.LABEL_OPERATOR_IN, ["A100"])],
],
)
reply1 = scheduler.schedule(
sched_request(node_type_configs=node_type_configs, resource_requests=[req1])
)
to_launch1, _ = _launch_and_terminate(reply1)
assert to_launch1 == {"tpu_node": 1}
# 1). Label selector falls back to second priority and scales up A100 node.
req2 = ResourceRequestUtil.make(
{"CPU": 1},
label_selectors=[
# infeasible
[("accelerator-type", LabelSelectorOperator.LABEL_OPERATOR_IN, ["B200"])],
[("accelerator-type", LabelSelectorOperator.LABEL_OPERATOR_IN, ["A100"])],
],
)
reply2 = scheduler.schedule(
sched_request(node_type_configs=node_type_configs, resource_requests=[req2])
)
to_launch2, _ = _launch_and_terminate(reply2)
assert to_launch2 == {"gpu_node": 1}
def test_pg_with_bundle_infeasible_label_selectors():
"""
Test that placement group scheduling honors bundle_label_selectors.
"""
scheduler = ResourceDemandScheduler(event_logger)
AFFINITY = ResourceRequestUtil.PlacementConstraintType.AFFINITY
node_type_configs = {
"gpu_node": NodeTypeConfig(
name="gpu_node",
resources={"CPU": 4, "GPU": 1},
min_worker_nodes=0,
max_worker_nodes=5,
labels={"accelerator": "A100"},
),
"tpu_node": NodeTypeConfig(
name="tpu_node",
resources={"CPU": 4},
min_worker_nodes=0,
max_worker_nodes=5,
labels={"accelerator": "TPU"},
),
}
# Create ResourceRequests for a placement group where each bundle has different label selectors
gpu_request = ResourceRequestUtil.make(
{"CPU": 2, "GPU": 1},
constraints=[(AFFINITY, "pg-1", "")],
label_selectors=[
[("accelerator", LabelSelectorOperator.LABEL_OPERATOR_IN, ["A100"])]
],
)
tpu_request = ResourceRequestUtil.make(
{"CPU": 2},
constraints=[(AFFINITY, "pg-1", "")],
label_selectors=[
[("accelerator", LabelSelectorOperator.LABEL_OPERATOR_IN, ["TPU"])]
],
)
request = sched_request(
node_type_configs=node_type_configs,
gang_resource_requests=[[gpu_request, tpu_request]],
)
reply = scheduler.schedule(request)
to_launch, _ = _launch_and_terminate(reply)
assert sorted(to_launch) == sorted({"gpu_node": 1, "tpu_node": 1})
# Both bundles require A100, but no node has enough resources -> infeasible
infeasbile_gpu_request = ResourceRequestUtil.make(
{"CPU": 3, "GPU": 1},
constraints=[(AFFINITY, "pg-2", "")],
label_selectors=[
[("accelerator", LabelSelectorOperator.LABEL_OPERATOR_IN, ["A100"])]
],
)
request = sched_request(
node_type_configs=node_type_configs,
gang_resource_requests=[[infeasbile_gpu_request, infeasbile_gpu_request]],
)
reply = scheduler.schedule(request)
to_launch, _ = _launch_and_terminate(reply)
assert to_launch == {}
assert len(reply.infeasible_gang_resource_requests) == 1
def test_get_nodes_with_resource_availabilities():
node_type_configs = {
"type_gpu1": NodeTypeConfig(
name="type_gpu1",
resources={"CPU": 8, "GPU": 1, "gpu1": 1},
min_worker_nodes=0,
max_worker_nodes=10,
),
"type_gpu2": NodeTypeConfig(
name="type_gpu2",
resources={"CPU": 8, "GPU": 1, "gpu2": 1},
min_worker_nodes=0,
max_worker_nodes=10,
),
"type_gpu3": NodeTypeConfig(
name="type_gpu3",
resources={"CPU": 8, "GPU": 1, "gpu3": 1},
min_worker_nodes=0,
max_worker_nodes=10,
),
"type_gpu4": NodeTypeConfig(
name="type_gpu4",
resources={"CPU": 1, "GPU": 1, "gpu4": 1},
min_worker_nodes=0,
max_worker_nodes=10,
),
}
def get_nodes_for(
resource_requests,
anti_affinity=False,
max_nodes: Optional[int] = None,
current_nodes: Optional[Dict] = None,
cloud_resource_availabilities=None,
):
reply = schedule(
node_type_configs,
current_nodes or {},
resource_requests,
anti_affinity=anti_affinity,
max_nodes=max_nodes,
cloud_resource_availabilities=cloud_resource_availabilities,
)
to_launch, _ = _launch_and_terminate(reply)
infeasible = ResourceRequestUtil.to_resource_maps(
reply.infeasible_resource_requests
)
return to_launch, infeasible
# Pick the node type with the highest availability score when utilization scores are equal.
assert get_nodes_for(
[{"CPU": 8, "GPU": 1}],
cloud_resource_availabilities={
"type_gpu1": 0.1,
"type_gpu2": 1,
"type_gpu3": 0.2,
},
) == ({"type_gpu2": 1}, [])
# The availability score is set to 1 by default.
assert get_nodes_for(
[{"CPU": 8, "GPU": 1}],
cloud_resource_availabilities={"type_gpu2": 0.1, "type_gpu3": 0.2},
) == ({"type_gpu1": 1}, [])
assert get_nodes_for(
[{"CPU": 8, "GPU": 1}] * 2,
cloud_resource_availabilities={
"type_gpu1": 0.1,
"type_gpu2": 0.1,
"type_gpu3": 1,
},
) == ({"type_gpu3": 2}, [])
# The utilization score is the first factor to be considered.
assert get_nodes_for([{"CPU": 1, "GPU": 1}], cloud_resource_availabilities={}) == (
{"type_gpu4": 1},
[],
)
# The utilization score is the first factor to be considered.
assert get_nodes_for(
[{"CPU": 1, "GPU": 1}],
cloud_resource_availabilities={
"type_gpu1": 0.1,
"type_gpu2": 0.1,
"type_gpu3": 1,
"type_gpu4": 0.1,
},
) == ({"type_gpu4": 1}, [])
def test_infeasible_shape_caching():
"""
Test that identical requests failing to schedule on a node are cached,
drastically reducing calls to _try_schedule_one to prevent O(N^2 * M) hangs.
"""
scheduler = ResourceDemandScheduler(event_logger)
node_type_configs = {
"type_1": NodeTypeConfig(
name="type_1",
resources={"CPU": 2},
min_worker_nodes=0,
max_worker_nodes=1, # Cluster can fit max one node
),
}
# Start with 1 existing node that has 2 CPUs available.
instances = [
make_autoscaler_instance(
ray_node=NodeState(
ray_node_type_name="type_1",
available_resources={"CPU": 2},
total_resources={"CPU": 2},
node_id=b"r1",
),
im_instance=Instance(
instance_type="type_1",
status=Instance.RAY_RUNNING,
instance_id="1",
node_id="r1",
),
cloud_instance_id="c-1",
),
]
# Submit 1,000 identical tasks that all request 2 CPUs.
# Every request after the initial one should be cached and fail early.
resource_requests = [ResourceRequestUtil.make({"CPU": 2}) for _ in range(1000)]
request = sched_request(
node_type_configs=node_type_configs,
resource_requests=resource_requests,
instances=instances,
max_num_nodes=1,
)
# Validate _try_schedule_one is only called twice by the scheduler.
orig_try_schedule_one = SchedulingNode._try_schedule_one
with patch.object(
SchedulingNode,
"_try_schedule_one",
autospec=True,
side_effect=orig_try_schedule_one,
) as mock_try_schedule:
reply = scheduler.schedule(request)
# 1 task should be scheduled on the existing node. The other 999 fail.
assert len(reply.infeasible_resource_requests) == 999
# Call 1: Fits the first 2-CPU request (Node is now full).
# Call 2: Evaluates the second 2-CPU request, fails, and adds to infeasible_shapes.
# Calls 3-1000: Bypassed entirely by the cache.
assert mock_try_schedule.call_count == 2
def test_infeasible_shape_caching_with_label_mutation():
"""
Test that dynamically adding labels clears the unavailable_shapes cache
so interleaved valid requests aren't skipped.
"""
scheduler = ResourceDemandScheduler(event_logger)
ANTI_AFFINITY = ResourceRequestUtil.PlacementConstraintType.ANTI_AFFINITY
node_type_configs = {
"type_1": NodeTypeConfig(
name="type_1",
resources={"CPU": 4},
min_worker_nodes=0,
max_worker_nodes=1,
),
}
instances = [
make_autoscaler_instance(
ray_node=NodeState(
ray_node_type_name="type_1",
available_resources={"CPU": 4},
total_resources={"CPU": 4},
node_id=b"r1",
labels={"required-label": "true"},
),
im_instance=Instance(
instance_type="type_1",
status=Instance.RAY_RUNNING,
instance_id="1",
node_id="r1",
),
cloud_instance_id="c-1",
),
]
# Req A: needs "pg: 1" which is missing on node, fails initially.
req_a = ResourceRequestUtil.make(
{"CPU": 1},
constraints=[(ANTI_AFFINITY, "dummy-anti", "1")],
label_selectors=[[("pg", LabelSelectorOperator.LABEL_OPERATOR_IN, ["1"])]],
)
# Req B: Needs "required-label: true", adds "pg=1" to the node upon scheduling.
req_b = ResourceRequestUtil.make(
{"CPU": 1},
constraints=[(ANTI_AFFINITY, "pg", "1")],
label_selectors=[
[("required-label", LabelSelectorOperator.LABEL_OPERATOR_IN, ["true"])]
],
)
# Manually group the requests to force an interleaved [A, B, A] ordering.
req_a_grouped = ResourceRequestUtil.group_by_count([req_a])[0]
req_b_grouped = ResourceRequestUtil.group_by_count([req_b])[0]
resource_requests = [req_a_grouped, req_b_grouped, req_a_grouped]
request = SchedulingRequest(
disable_launch_config_check=False,
node_type_configs=node_type_configs,
resource_requests=resource_requests,
current_instances=instances,
max_num_nodes=1,
)
reply = scheduler.schedule(request)
# Expected Sequence of Events:
# 1. Req A1 evaluates -> fails (no pg=1 on node). Caches shape A.
# 2. Req B evaluates -> succeeds. Mutates node state to add pg=1. Clears cache.
# 3. Req A2 evaluates -> Cache miss. Succeeds because node now has pg=1.
# Only the first Request A should have been marked infeasible.
assert len(reply.infeasible_resource_requests) == 1
# The scheduled node should have launched 0 new nodes (everything fit on the existing node)
to_launch, _ = _launch_and_terminate(reply)
assert to_launch == {}
def test_identical_node_state_caching():
"""
Test that the scheduler avoids redundant deepcopies and simulations
for nodes with identical states.
"""
scheduler = ResourceDemandScheduler(event_logger)
node_type_configs = {
"type_1": NodeTypeConfig(
name="type_1",
resources={"CPU": 4},
min_worker_nodes=0,
max_worker_nodes=100,
),
}
# Create 100 identical pending nodes
instances = []
for i in range(100):
instances.append(
make_autoscaler_instance(
im_instance=Instance(
instance_type="type_1",
status=Instance.REQUESTED,
instance_id=f"pending-{i}",
)
)
)
# Submit a single request that requires 1 CPU
resource_requests = [ResourceRequestUtil.make({"CPU": 1})]
request = sched_request(
node_type_configs=node_type_configs,
resource_requests=resource_requests,
instances=instances,
max_num_nodes=100,
)
# Track how many times try_schedule is actually called on a node
orig_try_schedule = SchedulingNode.try_schedule
with patch.object(
SchedulingNode,
"try_schedule",
autospec=True,
side_effect=orig_try_schedule,
) as mock_try_schedule:
reply = scheduler.schedule(request)
# The scheduler should evaluate exactly one of the 100 identical pending nodes
assert mock_try_schedule.call_count == 1
# It should successfully schedule the task without needing to launch any new nodes,
# because it used the first pending node.
to_launch, _ = _launch_and_terminate(reply)
assert to_launch == {}
assert len(reply.infeasible_resource_requests) == 0
def test_ippr_resize_to_maximum_capacity():
scheduler = ResourceDemandScheduler(event_logger)
node_type_configs = {
"type_1": NodeTypeConfig(
name="type_1",
resources={"CPU": 1},
min_worker_nodes=0,
max_worker_nodes=10,
),
}
# Existing running node
instance = make_autoscaler_instance(
ray_node=NodeState(
ray_node_type_name="type_1",
available_resources={"CPU": 1},
total_resources={"CPU": 1},
node_id=b"r1",
),
im_instance=Instance(
instance_type="type_1",
status=Instance.RAY_RUNNING,
instance_id="i-1",
node_id="r1",
),
cloud_instance_id="pod-1",
)
# IPPR limits/specs and provider suggestion to upsize CPU to 2
ippr_specs = IPPRSpecs(
groups={
"type_1": IPPRGroupSpec(
min_cpu=1,
max_cpu=4,
min_memory=1 * 1024 * 1024 * 1024,
max_memory=8 * 1024 * 1024 * 1024,
resize_timeout=60,
)
}
)
ippr_status = IPPRStatus(
cloud_instance_id="pod-1",
spec=ippr_specs.groups["type_1"],
current_cpu=1,
current_memory=1 * 1024 * 1024 * 1024,
desired_cpu=1,
desired_memory=1 * 1024 * 1024 * 1024,
)
request = sched_request(
node_type_configs=node_type_configs,
resource_requests=[ResourceRequestUtil.make({"CPU": 2})],
instances=[instance],
ippr_specs=ippr_specs,
ippr_statuses={"pod-1": ippr_status},
)
reply = scheduler.schedule(request)
# Scheduler should issue one IPPR action with desired set to suggested values
assert len(reply.to_ippr) == 1
assert reply.to_ippr[0].cloud_instance_id == "pod-1"
assert reply.to_ippr[0].desired_cpu == 4.0
assert reply.to_ippr[0].desired_memory == 8 * 1024 * 1024 * 1024
assert reply.to_launch == []
def test_ippr_resize_scale_out_if_one_ippr_is_new():
scheduler = ResourceDemandScheduler(event_logger)
node_type_configs = {
"type_1": NodeTypeConfig(
name="type_1",
resources={"CPU": 1},
min_worker_nodes=0,
max_worker_nodes=10,
),
}
# Existing running node
instance = make_autoscaler_instance(
ray_node=NodeState(
ray_node_type_name="type_1",
available_resources={"CPU": 1},
total_resources={"CPU": 1},
node_id=b"r1",
),
im_instance=Instance(
instance_type="type_1",
status=Instance.RAY_RUNNING,
instance_id="i-1",
node_id="r1",
),
cloud_instance_id="pod-1",
)
# IPPR limits/specs and provider suggestion to upsize CPU to 2
ippr_specs = IPPRSpecs(
groups={
"type_1": IPPRGroupSpec(
min_cpu=1,
max_cpu=4,
min_memory=1 * 1024 * 1024 * 1024,
max_memory=8 * 1024 * 1024 * 1024,
resize_timeout=60,
)
}
)
ippr_status = IPPRStatus(
cloud_instance_id="pod-1",
spec=ippr_specs.groups["type_1"],
current_cpu=1,
current_memory=1 * 1024 * 1024 * 1024,
desired_cpu=1,
desired_memory=1 * 1024 * 1024 * 1024,
k8s_resize_status="new", # error or timeout will be rollback with a new IPPR action
raylet_id="r1",
)
request = sched_request(
node_type_configs=node_type_configs,
resource_requests=[ResourceRequestUtil.make({"CPU": 2})],
instances=[instance],
ippr_specs=ippr_specs,
ippr_statuses={"pod-1": ippr_status},
)
reply = scheduler.schedule(request)
# Scheduler should issue a new IPPR for the rollback.
assert len(reply.to_ippr) == 1
assert reply.to_ippr[0].cloud_instance_id == "pod-1"
assert reply.to_ippr[0].desired_cpu == 1
assert reply.to_ippr[0].desired_memory == 1 * 1024 * 1024 * 1024
# Scheduler should scale out a new node
to_launch, _ = _launch_and_terminate(reply)
assert to_launch == {"type_1": 1}
def test_ippr_resize_scale_out_if_one_ippr_is_inprogress():
scheduler = ResourceDemandScheduler(event_logger)
node_type_configs = {
"type_1": NodeTypeConfig(
name="type_1",
resources={"CPU": 1},
min_worker_nodes=0,
max_worker_nodes=10,
),
}
# Existing running node
instance = make_autoscaler_instance(
ray_node=NodeState(
ray_node_type_name="type_1",
available_resources={"CPU": 1},
total_resources={"CPU": 1},
node_id=b"r1",
),
im_instance=Instance(
instance_type="type_1",
status=Instance.RAY_RUNNING,
instance_id="i-1",
node_id="r1",
),
cloud_instance_id="pod-1",
)
# IPPR limits/specs and provider suggestion to upsize CPU to 2
ippr_specs = IPPRSpecs(
groups={
"type_1": IPPRGroupSpec(
min_cpu=1,
max_cpu=4,
min_memory=1 * 1024 * 1024 * 1024,
max_memory=8 * 1024 * 1024 * 1024,
resize_timeout=60,
)
}
)
ippr_status = IPPRStatus(
cloud_instance_id="pod-1",
spec=ippr_specs.groups["type_1"],
current_cpu=1,
current_memory=1 * 1024 * 1024 * 1024,
desired_cpu=2,
desired_memory=2 * 1024 * 1024 * 1024,
k8s_resize_status="inprogress",
raylet_id="r1",
)
request = sched_request(
node_type_configs=node_type_configs,
resource_requests=[ResourceRequestUtil.make({"CPU": 4})],
instances=[instance],
ippr_specs=ippr_specs,
ippr_statuses={"pod-1": ippr_status},
)
reply = scheduler.schedule(request)
# Scheduler should not issue new IPPR action because the IPPR is in progress
assert len(reply.to_ippr) == 0
# Scheduler should scale out a new node
to_launch, _ = _launch_and_terminate(reply)
assert to_launch == {"type_1": 1}
def test_ippr_in_progress_exposes_desired_capacity_avoids_launch():
scheduler = ResourceDemandScheduler(event_logger)
node_type_configs = {
"type_1": NodeTypeConfig(
name="type_1",
resources={"CPU": 1},
min_worker_nodes=0,
max_worker_nodes=10,
),
}
# Existing running node with an in-progress resize to CPU=4
instance = make_autoscaler_instance(
ray_node=NodeState(
ray_node_type_name="type_1",
available_resources={"CPU": 1},
total_resources={"CPU": 1},
node_id=b"r1",
),
im_instance=Instance(
instance_type="type_1",
status=Instance.RAY_RUNNING,
instance_id="i-1",
node_id="r1",
),
cloud_instance_id="pod-1",
)
ippr_specs = IPPRSpecs(
groups={
"type_1": IPPRGroupSpec(
min_cpu=1,
max_cpu=4,
min_memory=1 * 1024 * 1024 * 1024,
max_memory=8 * 1024 * 1024 * 1024,
resize_timeout=60,
)
}
)
ippr_status = IPPRStatus(
cloud_instance_id="pod-1",
spec=ippr_specs.groups["type_1"],
current_cpu=1,
current_memory=1 * 1024 * 1024 * 1024,
desired_cpu=4,
desired_memory=8 * 1024 * 1024 * 1024,
resizing_at=int(time.time()),
k8s_resize_status="inprogress",
)
request = sched_request(
node_type_configs=node_type_configs,
resource_requests=[ResourceRequestUtil.make({"CPU": 2})],
instances=[instance],
ippr_specs=ippr_specs,
ippr_statuses={"pod-1": ippr_status},
)
reply = scheduler.schedule(request)
# The scheduler should fit the 2-CPU request on the existing node (using desired capacity)
assert reply.to_launch == []
assert reply.to_ippr == [] # already in progress, no new IPPR action
def test_ippr_does_not_resize_pending_node_without_ray_node_id():
scheduler = ResourceDemandScheduler(event_logger)
node_type_configs = {
"type_1": NodeTypeConfig(
name="type_1",
resources={"CPU": 1},
min_worker_nodes=0,
max_worker_nodes=10,
),
}
# Existing pending node has no ray_node_id yet.
instance = make_autoscaler_instance(
im_instance=Instance(
instance_type="type_1",
status=Instance.ALLOCATED,
instance_id="i-1",
),
cloud_instance_id="pod-1",
)
ippr_specs = IPPRSpecs(
groups={
"type_1": IPPRGroupSpec(
min_cpu=1,
max_cpu=4,
min_memory=1 * 1024 * 1024 * 1024,
max_memory=8 * 1024 * 1024 * 1024,
resize_timeout=60,
)
}
)
ippr_status = IPPRStatus(
cloud_instance_id="pod-1",
spec=ippr_specs.groups["type_1"],
current_cpu=1,
current_memory=1 * 1024 * 1024 * 1024,
desired_cpu=1,
desired_memory=1 * 1024 * 1024 * 1024,
)
request = sched_request(
node_type_configs=node_type_configs,
resource_requests=[ResourceRequestUtil.make({"CPU": 2})],
instances=[instance],
ippr_specs=ippr_specs,
ippr_statuses={"pod-1": ippr_status},
)
reply = scheduler.schedule(request)
# Pending nodes without a ray_node_id should not be selected for IPPR.
assert reply.to_ippr == []
# Scheduler should also not launch a new node since the pending node could fulfill the request after IPPR.
to_launch, _ = _launch_and_terminate(reply)
assert to_launch == {}
def test_ippr_capacity_of_unselected_candidates_not_modified():
scheduler = ResourceDemandScheduler(event_logger)
node_type_configs = {
"type_1": NodeTypeConfig(
name="type_1",
resources={"CPU": 1},
min_worker_nodes=0,
max_worker_nodes=10,
),
}
instances = [
make_autoscaler_instance(
ray_node=NodeState(
ray_node_type_name="type_1",
available_resources={"CPU": 1},
total_resources={"CPU": 1},
node_id=b"r1",
),
im_instance=Instance(
instance_type="type_1",
status=Instance.RAY_RUNNING,
instance_id="i-1",
node_id="r1",
),
cloud_instance_id="pod-1",
),
make_autoscaler_instance(
ray_node=NodeState(
ray_node_type_name="type_1",
available_resources={"CPU": 1},
total_resources={"CPU": 1},
node_id=b"r2",
),
im_instance=Instance(
instance_type="type_1",
status=Instance.RAY_RUNNING,
instance_id="i-2",
node_id="r2",
),
cloud_instance_id="pod-2",
),
make_autoscaler_instance(
ray_node=NodeState(
ray_node_type_name="type_1",
available_resources={"CPU": 1},
total_resources={"CPU": 1},
node_id=b"r3",
idle_duration_ms=10_000,
),
im_instance=Instance(
instance_type="type_1",
status=Instance.RAY_RUNNING,
instance_id="i-3",
node_id="r3",
),
cloud_instance_id="pod-3",
),
]
ippr_specs = IPPRSpecs(
groups={
"type_1": IPPRGroupSpec(
min_cpu=1,
max_cpu=4,
min_memory=1 * 1024 * 1024 * 1024,
max_memory=8 * 1024 * 1024 * 1024,
resize_timeout=60,
)
}
)
ippr_statuses = {
"pod-1": IPPRStatus(
cloud_instance_id="pod-1",
spec=ippr_specs.groups["type_1"],
current_cpu=1,
current_memory=1 * 1024 * 1024 * 1024,
desired_cpu=1,
desired_memory=1 * 1024 * 1024 * 1024,
),
"pod-2": IPPRStatus(
cloud_instance_id="pod-2",
spec=ippr_specs.groups["type_1"],
current_cpu=1,
current_memory=1 * 1024 * 1024 * 1024,
desired_cpu=1,
desired_memory=1 * 1024 * 1024 * 1024,
),
}
request = sched_request(
node_type_configs=node_type_configs,
gang_resource_requests=[[ResourceRequestUtil.make({"CPU": 2})]],
instances=instances,
idle_timeout_s=0,
ippr_specs=ippr_specs,
ippr_statuses=ippr_statuses,
)
reply = scheduler.schedule(request)
assert reply.to_launch == []
# Only one IPPR candidate is selected for this gang request.
assert len(reply.to_ippr) == 1
assert {status.cloud_instance_id for status in reply.to_ippr} == {"pod-1"}
assert {status.desired_cpu for status in reply.to_ippr} == {4.0}
_, to_terminate = _launch_and_terminate(reply)
assert [instance_id for instance_id, _, _ in to_terminate] == ["i-3"]
# if pod-2 is accidentally selected for IPPR (it should not be),
# the cluster resources should be bigger than 5.0
assert reply.cluster_resources["CPU"] == 5.0
def test_ippr_max_limits_affect_new_node_capacity():
scheduler = ResourceDemandScheduler(event_logger)
node_type_configs = {
"type_1": NodeTypeConfig(
name="type_1",
resources={"CPU": 1},
min_worker_nodes=0,
max_worker_nodes=10,
),
}
# No existing instances; IPPR max allows new nodes to expose larger capacity
ippr_specs = IPPRSpecs(
groups={
"type_1": IPPRGroupSpec(
min_cpu=1,
max_cpu=4,
min_memory=1 * 1024 * 1024 * 1024,
max_memory=4 * 1024 * 1024 * 1024,
resize_timeout=60,
)
}
)
request = sched_request(
node_type_configs=node_type_configs,
resource_requests=[ResourceRequestUtil.make({"CPU": 1})] * 4,
ippr_specs=ippr_specs,
)
reply = scheduler.schedule(request)
to_launch, _ = _launch_and_terminate(reply)
# With IPPR max=4, all four 1-CPU bundles should fit on a single launched node
assert to_launch == {"type_1": 1}
assert reply.to_ippr == []
def test_ippr_max_limits_affect_new_node_capacity_2():
scheduler = ResourceDemandScheduler(event_logger)
node_type_configs = {
"type_1": NodeTypeConfig(
name="type_1",
resources={"CPU": 1},
min_worker_nodes=0,
max_worker_nodes=10,
),
}
ippr_specs = IPPRSpecs(
groups={
"type_1": IPPRGroupSpec(
min_cpu=1,
max_cpu=4,
min_memory=1 * 1024 * 1024 * 1024,
max_memory=4 * 1024 * 1024 * 1024,
resize_timeout=60,
)
}
)
request = sched_request(
node_type_configs=node_type_configs,
resource_requests=[ResourceRequestUtil.make({"CPU": 1})] * 6,
ippr_specs=ippr_specs,
)
reply = scheduler.schedule(request)
to_launch, _ = _launch_and_terminate(reply)
# Each launched node should be evaluated with IPPR max=4 CPU capacity:
# six 1-CPU bundles should require exactly two new nodes, not three.
assert to_launch == {"type_1": 2}
assert reply.to_ippr == []
class TestSchedulerPerformanceOptimizations:
"""Tests for large-cluster performance optimizations."""
def test_quick_reject_skips_exhausted_nodes(self):
"""Nodes with no available resources should be skipped without deepcopy."""
node_type_configs = {
"type_1": NodeTypeConfig(
name="type_1",
resources={"CPU": 10, "memory": 100},
min_worker_nodes=0,
max_worker_nodes=100,
),
}
# Create instances where all resources are allocated (available = 0).
instances = []
for i in range(50):
instances.append(
make_autoscaler_instance(
im_instance=Instance(
instance_type="type_1",
status=Instance.RAY_RUNNING,
instance_id=f"type_1-{i}",
node_id=f"r{i}type_1",
),
ray_node=NodeState(
node_id=f"r{i}type_1".encode("utf-8"),
ray_node_type_name="type_1",
available_resources={}, # All resources used up
total_resources={"CPU": 10, "memory": 100},
idle_duration_ms=0,
status=NodeStatus.RUNNING,
),
cloud_instance_id=f"c-type_1-{i}",
)
)
request = sched_request(
node_type_configs=node_type_configs,
resource_requests=[ResourceRequestUtil.make({"CPU": 2})] * 10,
instances=instances,
)
reply = ResourceDemandScheduler(event_logger).schedule(request)
to_launch, _ = _launch_and_terminate(reply)
# Should launch new nodes since existing are exhausted.
assert to_launch == {"type_1": 2}
def test_quick_reject_partial_resources(self):
"""Nodes with some resources but below minimum demand are skipped."""
node_type_configs = {
"type_1": NodeTypeConfig(
name="type_1",
resources={"CPU": 10, "memory": 100},
min_worker_nodes=0,
max_worker_nodes=100,
),
}
# Node has 1 CPU available but all requests need 4 CPU.
instances = []
for i in range(10):
instances.append(
make_autoscaler_instance(
im_instance=Instance(
instance_type="type_1",
status=Instance.RAY_RUNNING,
instance_id=f"type_1-{i}",
node_id=f"r{i}type_1",
),
ray_node=NodeState(
node_id=f"r{i}type_1".encode("utf-8"),
ray_node_type_name="type_1",
available_resources={"CPU": 1, "memory": 10},
total_resources={"CPU": 10, "memory": 100},
idle_duration_ms=0,
status=NodeStatus.RUNNING,
),
cloud_instance_id=f"c-type_1-{i}",
)
)
request = sched_request(
node_type_configs=node_type_configs,
resource_requests=[ResourceRequestUtil.make({"CPU": 4})] * 5,
instances=instances,
)
reply = ResourceDemandScheduler(event_logger).schedule(request)
to_launch, _ = _launch_and_terminate(reply)
# All existing nodes have < 4 CPU available, must launch new.
# 5 requests × 4 CPU each, new nodes have 10 CPU → fits 2 per node → need 3.
assert to_launch == {"type_1": 3}
def test_quick_reject_does_not_skip_feasible_nodes(self):
"""Nodes with sufficient resources should still be scheduled on."""
node_type_configs = {
"type_1": NodeTypeConfig(
name="type_1",
resources={"CPU": 10, "memory": 100},
min_worker_nodes=0,
max_worker_nodes=100,
),
}
# Nodes have plenty of resources.
instances = []
for i in range(5):
instances.append(
make_autoscaler_instance(
im_instance=Instance(
instance_type="type_1",
status=Instance.RAY_RUNNING,
instance_id=f"type_1-{i}",
node_id=f"r{i}type_1",
),
ray_node=NodeState(
node_id=f"r{i}type_1".encode("utf-8"),
ray_node_type_name="type_1",
available_resources={"CPU": 10, "memory": 100},
total_resources={"CPU": 10, "memory": 100},
idle_duration_ms=0,
status=NodeStatus.RUNNING,
),
cloud_instance_id=f"c-type_1-{i}",
)
)
request = sched_request(
node_type_configs=node_type_configs,
resource_requests=[ResourceRequestUtil.make({"CPU": 2})] * 10,
instances=instances,
)
reply = ResourceDemandScheduler(event_logger).schedule(request)
to_launch, _ = _launch_and_terminate(reply)
# Existing nodes can handle all requests (5 nodes × 10 CPU ÷ 2 CPU = 25 slots).
assert to_launch == {}
if __name__ == "__main__":
if os.environ.get("PARALLEL_CI"):
sys.exit(pytest.main(["-n", "auto", "--boxed", "-vs", __file__]))
else:
sys.exit(pytest.main(["-sv", __file__]))
Reference in New Issue View Git Blame Copy Permalink