import copy import json import logging import os import re import shutil import sys import tempfile import time import unittest from collections import defaultdict from enum import Enum from unittest.mock import ( Mock, patch, ) import pytest import yaml from jsonschema.exceptions import ValidationError import ray from ray.autoscaler._private import commands from ray.autoscaler._private.autoscaler import NonTerminatedNodes, StandardAutoscaler from ray.autoscaler._private.commands import get_or_create_head_node from ray.autoscaler._private.constants import ( AUTOSCALER_HEARTBEAT_TIMEOUT_S, DISABLE_LAUNCH_CONFIG_CHECK_KEY, DISABLE_NODE_UPDATERS_KEY, FOREGROUND_NODE_LAUNCH_KEY, WORKER_LIVENESS_CHECK_KEY, ) from ray.autoscaler._private.load_metrics import LoadMetrics from ray.autoscaler._private.monitor import Monitor from ray.autoscaler._private.prom_metrics import ( AutoscalerPrometheusMetrics, NullMetric, ) from ray.autoscaler._private.providers import ( _DEFAULT_CONFIGS, _NODE_PROVIDERS, _clear_provider_cache, ) from ray.autoscaler._private.readonly.node_provider import ReadOnlyNodeProvider from ray.autoscaler._private.util import prepare_config, validate_config from ray.autoscaler.node_launch_exception import NodeLaunchException from ray.autoscaler.node_provider import NodeProvider from ray.autoscaler.sdk import get_docker_host_mount_location from ray.autoscaler.tags import ( NODE_KIND_HEAD, NODE_KIND_WORKER, STATUS_UNINITIALIZED, STATUS_UP_TO_DATE, STATUS_UPDATE_FAILED, TAG_RAY_CLUSTER_NAME, TAG_RAY_NODE_KIND, TAG_RAY_NODE_STATUS, TAG_RAY_USER_NODE_TYPE, ) from ray.core.generated import common_pb2, gcs_pb2 from ray.exceptions import RpcError from ray.tests.autoscaler_test_utils import ( MockNode, MockProcessRunner, MockProvider, ) from ray.tests.test_batch_node_provider_unit import ( MockBatchingNodeProvider, ) WORKER_FILTER = {TAG_RAY_NODE_KIND: NODE_KIND_WORKER} class DrainNodeOutcome(str, Enum): """Potential outcomes of DrainNode calls, each of which is handled differently by the autoscaler. """ # Return a reponse indicating all nodes were successfully drained. Succeeded = "Succeeded" # Return response indicating at least one node failed to be drained. NotAllDrained = "NotAllDrained" # Return an unimplemented gRPC error, indicating an old GCS. Unimplemented = "Unimplemented" # Raise a generic unexpected RPC error. GenericRpcError = "GenericRpcError" # Raise a generic unexpected exception. GenericException = "GenericException" # Tell the autoscaler to fail finding ips during drain FailedToFindIp = "FailedToFindIp" class MockGcsClient: """Mock for GCSClient used by autoscaler to drain Ray nodes. Can simulate DrainNode failures via the `drain_node_outcome` parameter. Comments in DrainNodeOutcome enum class indicate the behavior for each outcome. """ def __init__(self, drain_node_outcome=DrainNodeOutcome.Succeeded): self.drain_node_outcome = drain_node_outcome # Tracks how many times we've called DrainNode. self.drain_node_call_count = 0 # Tracks how many times DrainNode returned a successful RPC response. self.drain_node_reply_success = 0 def drain_nodes(self, node_ids_to_drain, timeout: int): """Simulate NodeInfo stub's DrainNode call. Outcome determined by self.drain_outcome. """ self.drain_node_call_count += 1 if self.drain_node_outcome == DrainNodeOutcome.Unimplemented: raise RpcError( "not implemented!", rpc_code=ray._raylet.GRPC_STATUS_CODE_UNIMPLEMENTED ) elif self.drain_node_outcome == DrainNodeOutcome.GenericRpcError: # Any StatusCode besides UNIMPLEMENTED will do here. raise RpcError( "server is not feeling well", rpc_code=ray._raylet.GRPC_STATUS_CODE_UNAVAILABLE, ) elif self.drain_node_outcome == DrainNodeOutcome.GenericException: raise Exception("DrainNode failed in some unexpected way.") self.drain_node_reply_success += 1 if self.drain_node_outcome in [ DrainNodeOutcome.Succeeded, DrainNodeOutcome.FailedToFindIp, ]: return node_ids_to_drain elif self.drain_node_outcome == DrainNodeOutcome.NotAllDrained: # All but the last. return node_ids_to_drain[:-1] else: # Shouldn't land here. assert False, "Possible drain node outcomes exhausted." def mock_node_id() -> bytes: """Random node id to pass to load_metrics.update.""" return os.urandom(10) def fill_in_node_ids(provider, load_metrics) -> None: """Node ids for each ip are usually obtained by polling the GCS in monitor.py. For test purposes, we sometimes need to manually fill these fields with mocks. """ for node in provider.non_terminated_nodes({}): ip = provider.internal_ip(node) load_metrics.node_id_by_ip[ip] = mock_node_id() class MockAutoscaler(StandardAutoscaler): """Test autoscaler constructed to verify the property that each autoscaler update issues at most one provider.non_terminated_nodes call. """ def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.fail_to_find_ip_during_drain = False def _update(self): # Only works with MockProvider or MockBatchingNodeProvider. assert isinstance(self.provider, MockProvider) or isinstance( self.provider, MockBatchingNodeProvider ) start_calls = self.provider.num_non_terminated_nodes_calls super()._update() end_calls = self.provider.num_non_terminated_nodes_calls # Strict inequality if update is called twice within the throttling # interval `self.update_interval_s` assert end_calls <= start_calls + 1 def drain_nodes_via_gcs(self, provider_node_ids_to_drain): if self.fail_to_find_ip_during_drain: self.provider.fail_to_fetch_ip = True super().drain_nodes_via_gcs(provider_node_ids_to_drain) self.provider.fail_to_fetch_ip = False class NoUpdaterMockAutoscaler(MockAutoscaler): def update_nodes(self): raise AssertionError( "Node updaters are disabled. This method should not be accessed!" ) SMALL_CLUSTER = { "cluster_name": "default", "idle_timeout_minutes": 5, "max_workers": 2, "provider": { "type": "mock", "region": "us-east-1", "availability_zone": "us-east-1a", }, "docker": { "image": "example", "container_name": "mock", }, "auth": { "ssh_user": "ubuntu", "ssh_private_key": os.devnull, }, "available_node_types": { "head": { "node_config": { "TestProp": 1, }, "resources": {"CPU": 1}, "max_workers": 0, }, "worker": { "node_config": { "TestProp": 2, }, "resources": {"CPU": 1}, "min_workers": 0, "max_workers": 2, }, }, "head_node_type": "head", "file_mounts": {}, "cluster_synced_files": [], "initialization_commands": ["init_cmd"], "setup_commands": ["setup_cmd"], "head_setup_commands": ["head_setup_cmd"], "worker_setup_commands": ["worker_setup_cmd"], "head_start_ray_commands": ["start_ray_head"], "worker_start_ray_commands": ["start_ray_worker"], } MOCK_DEFAULT_CONFIG = { "cluster_name": "default", "max_workers": 2, "upscaling_speed": 1.0, "idle_timeout_minutes": 5, "provider": { "type": "mock", "region": "us-east-1", "availability_zone": "us-east-1a", }, "docker": { "image": "example", "container_name": "mock", }, "auth": { "ssh_user": "ubuntu", "ssh_private_key": os.devnull, }, "available_node_types": { "ray.head.default": {"resources": {}, "node_config": {"head_default_prop": 4}}, "ray.worker.default": { "min_workers": 0, "max_workers": 2, "resources": {}, "node_config": {"worker_default_prop": 7}, }, }, "head_node_type": "ray.head.default", "head_node": {}, "worker_nodes": {}, "file_mounts": {}, "cluster_synced_files": [], "initialization_commands": [], "setup_commands": [], "head_setup_commands": [], "worker_setup_commands": [], "head_start_ray_commands": [], "worker_start_ray_commands": [], } TYPES_A = { "empty_node": { "node_config": { "FooProperty": 42, "TestProp": 1, }, "resources": {}, "max_workers": 0, }, "m4.large": { "node_config": {}, "resources": {"CPU": 2}, "max_workers": 10, }, "m4.4xlarge": { "node_config": {}, "resources": {"CPU": 16}, "max_workers": 8, }, "m4.16xlarge": { "node_config": {}, "resources": {"CPU": 64}, "max_workers": 4, }, "p2.xlarge": { "node_config": {}, "resources": {"CPU": 16, "GPU": 1}, "max_workers": 10, }, "p2.8xlarge": { "node_config": {}, "resources": {"CPU": 32, "GPU": 8}, "max_workers": 4, }, } MULTI_WORKER_CLUSTER = dict( SMALL_CLUSTER, **{"available_node_types": TYPES_A, "head_node_type": "empty_node"} ) # `DUMMY_IDLE_DURATION_S` is used as a dummy value for # `node_idle_duration_s` in load_metrics.update() # when we want to simulate worker node's idle duration # (`total_resources` == `available_resources`), # but don't want to cause autoscaler downscaling. # (`DUMMY_IDLE_DURATION_S` < `idle_timeout_minutes`). DUMMY_IDLE_DURATION_S = 3 exc_info = None try: raise Exception("Test exception.") except Exception: exc_info = sys.exc_info() assert exc_info is not None class LoadMetricsTest(unittest.TestCase): def testHeartbeat(self): lm = LoadMetrics() lm.update("1.1.1.1", mock_node_id(), {"CPU": 2}, {"CPU": 1}, 0) lm.mark_active("2.2.2.2") assert "1.1.1.1" in lm.last_heartbeat_time_by_ip assert "2.2.2.2" in lm.last_heartbeat_time_by_ip assert "3.3.3.3" not in lm.last_heartbeat_time_by_ip def testDebugString(self): lm = LoadMetrics() lm.update("1.1.1.1", mock_node_id(), {"CPU": 2}, {"CPU": 0}, 0) lm.update( "2.2.2.2", mock_node_id(), {"CPU": 2, "GPU": 16}, {"CPU": 2, "GPU": 2}, 0 ) lm.update( "3.3.3.3", mock_node_id(), { "memory": 1.05 * 1024 * 1024 * 1024, "object_store_memory": 2.1 * 1024 * 1024 * 1024, }, { "memory": 0, "object_store_memory": 1.05 * 1024 * 1024 * 1024, }, 0, ) debug = lm.info_string() assert ( "ResourceUsage: 2.0/4.0 CPU, 14.0/16.0 GPU, " "1.05 GiB/1.05 GiB memory, " "1.05 GiB/2.1 GiB object_store_memory" ) in debug class AutoscalingTest(unittest.TestCase): def setUp(self): _NODE_PROVIDERS["mock"] = lambda config: self.create_provider _DEFAULT_CONFIGS["mock"] = _DEFAULT_CONFIGS["aws"] self.provider = None self.tmpdir = tempfile.mkdtemp() def tearDown(self): self.provider = None del _NODE_PROVIDERS["mock"] _clear_provider_cache() shutil.rmtree(self.tmpdir) ray.shutdown() def waitFor(self, condition, num_retries=50, fail_msg=None): for _ in range(num_retries): if condition(): return time.sleep(0.1) fail_msg = fail_msg or "Timed out waiting for {}".format(condition) raise TimeoutError(fail_msg) def waitForUpdatersToFinish(self, autoscaler): self.waitFor( lambda: all( not updater.is_alive() for updater in autoscaler.updaters.values() ), num_retries=500, fail_msg="Last round of updaters didn't complete on time.", ) def num_nodes(self, tag_filters=None): if tag_filters is None: tag_filters = {} return len(self.provider.non_terminated_nodes(tag_filters)) def waitForNodes(self, expected, comparison=None, tag_filters=None): if comparison is None: comparison = self.assertEqual MAX_ITER = 50 for i in range(MAX_ITER): n = self.num_nodes(tag_filters) try: comparison(n, expected, msg="Unexpected node quantity.") return except Exception: if i == MAX_ITER - 1: print(self.provider.non_terminated_nodes(tag_filters)) raise time.sleep(0.1) def create_provider(self, config, cluster_name): assert self.provider return self.provider def write_config(self, config, call_prepare_config=True): new_config = copy.deepcopy(config) if call_prepare_config: new_config = prepare_config(new_config) path = os.path.join(self.tmpdir, "simple.yaml") with open(path, "w") as f: f.write(yaml.dump(new_config)) return path def worker_node_thread_check(self, foreground_node_launcher: bool): """Confirms that worker nodes were launched in the main thread if foreground node launch is enabled, in a subthread otherwise. Args: foreground_node_launcher: Whether workers nodes are expected to be launched in the foreground. """ worker_ids = self.provider.non_terminated_nodes(tag_filters=WORKER_FILTER) worker_nodes = [self.provider.mock_nodes[worker_id] for worker_id in worker_ids] if foreground_node_launcher: # All workers were created in the main thread. assert all( worker_node.created_in_main_thread for worker_node in worker_nodes ) else: # All workers were created in a background thread. assert not any( worker_node.created_in_main_thread for worker_node in worker_nodes ) def testAutoscalerConfigValidationFailNotFatal(self): invalid_config = {**SMALL_CLUSTER, "invalid_property_12345": "test"} # First check that this config is actually invalid with pytest.raises(ValidationError): validate_config(invalid_config) config_path = self.write_config(invalid_config) self.provider = MockProvider() runner = MockProcessRunner() autoscaler = MockAutoscaler( config_path, LoadMetrics(), MockGcsClient(), max_failures=0, process_runner=runner, update_interval_s=0, ) assert len(self.provider.non_terminated_nodes({})) == 0 autoscaler.update() self.waitForNodes(1) autoscaler.update() self.waitForNodes(1) def testValidation(self): """Ensures that schema validation is working.""" config = copy.deepcopy(SMALL_CLUSTER) try: validate_config(config) except Exception: self.fail("Test config did not pass validation test!") config["blah"] = "blah" with pytest.raises(ValidationError): validate_config(config) del config["blah"] del config["provider"] with pytest.raises(ValidationError): validate_config(config) def testValidateDefaultConfig(self): config = {} config["provider"] = { "type": "aws", "region": "us-east-1", "availability_zone": "us-east-1a", } config = prepare_config(config) try: validate_config(config) except ValidationError: self.fail("Default config did not pass validation test!") def testGetOrCreateHeadNode(self): config = copy.deepcopy(SMALL_CLUSTER) head_run_option = "--kernel-memory=10g" standard_run_option = "--memory-swap=5g" config["docker"]["head_run_options"] = [head_run_option] config["docker"]["run_options"] = [standard_run_option] config_path = self.write_config(config) self.provider = MockProvider() runner = MockProcessRunner() runner.respond_to_call("json .Mounts", ["[]"]) # Two initial calls to rsync, + 2 more calls during run_init runner.respond_to_call(".State.Running", ["false", "false", "false", "false"]) runner.respond_to_call("json .Config.Env", ["[]"]) def _create_node(node_config, tags, count, _skip_wait=False): assert tags[TAG_RAY_NODE_STATUS] == STATUS_UNINITIALIZED if not _skip_wait: self.provider.ready_to_create.wait() if self.provider.fail_creates: return with self.provider.lock: if self.provider.cache_stopped: for node in self.provider.mock_nodes.values(): if node.state == "stopped" and count > 0: count -= 1 node.state = "pending" node.tags.update(tags) for _ in range(count): self.provider.mock_nodes[str(self.provider.next_id)] = MockNode( str(self.provider.next_id), tags.copy(), node_config, tags.get(TAG_RAY_USER_NODE_TYPE), unique_ips=self.provider.unique_ips, ) self.provider.next_id += 1 self.provider.create_node = _create_node commands.get_or_create_head_node( config, printable_config_file=config_path, no_restart=False, restart_only=False, yes=True, override_cluster_name=None, _provider=self.provider, _runner=runner, ) self.waitForNodes(1) runner.assert_has_call("1.2.3.4", "init_cmd") runner.assert_has_call("1.2.3.4", "head_setup_cmd") runner.assert_has_call("1.2.3.4", "start_ray_head") self.assertEqual(self.provider.mock_nodes["0"].node_type, "head") runner.assert_has_call("1.2.3.4", pattern="docker run") runner.assert_has_call("1.2.3.4", pattern=head_run_option) runner.assert_has_call("1.2.3.4", pattern=standard_run_option) docker_mount_prefix = get_docker_host_mount_location( SMALL_CLUSTER["cluster_name"] ) runner.assert_not_has_call( "1.2.3.4", pattern=f"-v {docker_mount_prefix}/~/ray_bootstrap_config" ) common_container_copy = f"rsync -e.*docker exec -i.*{docker_mount_prefix}/~/" runner.assert_has_call( "1.2.3.4", pattern=common_container_copy + "ray_bootstrap_key.pem" ) runner.assert_has_call( "1.2.3.4", pattern=common_container_copy + "ray_bootstrap_config.yaml" ) return config def testNodeTypeNameChange(self): """ Tests that cluster launcher and autoscaler have correct behavior under changes and deletions of node type keys. Specifically if we change the key from "old-type" to "new-type", nodes of type "old-type" are deleted and (if required by the config) replaced by nodes of type "new-type". Strategy: 1. launch a test cluster with a head and one `min_worker` 2. change node type keys for both head and worker in cluster yaml 3. update cluster with new yaml 4. verify graceful replacement of the two nodes with old node types with two nodes with new node types. """ # Default config with renamed node types, min_worker 1, docker off. config = copy.deepcopy(MOCK_DEFAULT_CONFIG) config["docker"] = {} node_types = config["available_node_types"] node_types["ray.head.old"] = node_types.pop("ray.head.default") node_types["ray.worker.old"] = node_types.pop("ray.worker.default") config["head_node_type"] = "ray.head.old" node_types["ray.worker.old"]["min_workers"] = 1 # Create head and launch autoscaler runner = MockProcessRunner() self.provider = MockProvider() config_path = self.write_config(config) commands.get_or_create_head_node( config, printable_config_file=config_path, no_restart=False, restart_only=False, yes=True, override_cluster_name=None, _provider=self.provider, _runner=runner, ) self.waitForNodes(1) lm = LoadMetrics() autoscaler = MockAutoscaler( config_path, lm, MockGcsClient(), max_failures=0, process_runner=runner, update_interval_s=0, ) autoscaler.update() self.waitForNodes(2) head_list = self.provider.non_terminated_nodes( {TAG_RAY_NODE_KIND: NODE_KIND_HEAD} ) worker_list = self.provider.non_terminated_nodes( {TAG_RAY_NODE_KIND: NODE_KIND_WORKER} ) # One head (as always) # One worker (min_workers 1 with no resource demands) assert len(head_list) == 1 and len(worker_list) == 1 worker, head = worker_list.pop(), head_list.pop() # Confirm node type tags assert ( self.provider.node_tags(head).get(TAG_RAY_USER_NODE_TYPE) == "ray.head.old" ) assert ( self.provider.node_tags(worker).get(TAG_RAY_USER_NODE_TYPE) == "ray.worker.old" ) # Rename head and worker types new_config = copy.deepcopy(config) node_types = new_config["available_node_types"] node_types["ray.head.new"] = node_types.pop("ray.head.old") node_types["ray.worker.new"] = node_types.pop("ray.worker.old") new_config["head_node_type"] = "ray.head.new" config_path = self.write_config(new_config) # Expect this to delete "ray.head.old" head and create "ray.head.new" # head. commands.get_or_create_head_node( new_config, printable_config_file=config_path, no_restart=False, restart_only=False, yes=True, override_cluster_name=None, _provider=self.provider, _runner=runner, ) self.waitForNodes(2) head_list = self.provider.non_terminated_nodes( {TAG_RAY_NODE_KIND: NODE_KIND_HEAD} ) worker_list = self.provider.non_terminated_nodes( {TAG_RAY_NODE_KIND: NODE_KIND_WORKER} ) # One head (as always) # One worker (maintained from previous autoscaler update) assert len(head_list) == 1 and len(worker_list) == 1 worker, head = worker_list.pop(), head_list.pop() # Confirm new head assert ( self.provider.node_tags(head).get(TAG_RAY_USER_NODE_TYPE) == "ray.head.new" ) # Still old worker, as we haven't made an autoscaler update yet. assert ( self.provider.node_tags(worker).get(TAG_RAY_USER_NODE_TYPE) == "ray.worker.old" ) fill_in_node_ids(self.provider, lm) autoscaler.update() self.waitForNodes(2) events = autoscaler.event_summarizer.summary() # Just one node (node_id 1) terminated in the last update. # Validates that we didn't try to double-terminate node 0. assert sorted(events) == [ "Adding 1 node(s) of type ray.worker.new.", "Adding 1 node(s) of type ray.worker.old.", "Removing 1 nodes of type ray.worker.old (not " "in available_node_types: ['ray.head.new', 'ray.worker.new']).", ] head_list = self.provider.non_terminated_nodes( {TAG_RAY_NODE_KIND: NODE_KIND_HEAD} ) worker_list = self.provider.non_terminated_nodes( {TAG_RAY_NODE_KIND: NODE_KIND_WORKER} ) # One head (as always) # One worker (min_workers 1 with no resource demands) assert len(head_list) == 1 and len(worker_list) == 1 worker, head = worker_list.pop(), head_list.pop() # After the autoscaler update, new head and new worker. assert ( self.provider.node_tags(head).get(TAG_RAY_USER_NODE_TYPE) == "ray.head.new" ) assert ( self.provider.node_tags(worker).get(TAG_RAY_USER_NODE_TYPE) == "ray.worker.new" ) def testGetOrCreateHeadNodePodman(self): config = copy.deepcopy(SMALL_CLUSTER) config["docker"]["use_podman"] = True config_path = self.write_config(config) self.provider = MockProvider() runner = MockProcessRunner() runner.respond_to_call("json .Mounts", ["[]"]) # Two initial calls to rsync, + 2 more calls during run_init runner.respond_to_call(".State.Running", ["false", "false", "false", "false"]) runner.respond_to_call("json .Config.Env", ["[]"]) commands.get_or_create_head_node( config, printable_config_file=config_path, no_restart=False, restart_only=False, yes=True, override_cluster_name=None, _provider=self.provider, _runner=runner, ) self.waitForNodes(1) runner.assert_has_call("1.2.3.4", "init_cmd") runner.assert_has_call("1.2.3.4", "podman exec .*head_setup_cmd.*") runner.assert_has_call("1.2.3.4", "podman exec .*start_ray_head.*") self.assertEqual(self.provider.mock_nodes["0"].node_type, "head") runner.assert_has_call("1.2.3.4", pattern="podman run") docker_mount_prefix = get_docker_host_mount_location( SMALL_CLUSTER["cluster_name"] ) runner.assert_not_has_call( "1.2.3.4", pattern=f"-v {docker_mount_prefix}/~/ray_bootstrap_config" ) common_container_copy = f"rsync -e.*podman exec -i.*{docker_mount_prefix}/~/" runner.assert_has_call( "1.2.3.4", pattern=common_container_copy + "ray_bootstrap_key.pem" ) runner.assert_has_call( "1.2.3.4", pattern=common_container_copy + "ray_bootstrap_config.yaml" ) for cmd in runner.command_history(): assert "docker" not in cmd, "Docker (not podman) found in call: " f"{cmd}" runner.assert_has_call("1.2.3.4", "podman inspect") runner.assert_has_call("1.2.3.4", "podman exec") def testGetOrCreateHeadNodeFromStopped(self): config = self.testGetOrCreateHeadNode() self.provider.cache_stopped = True existing_nodes = self.provider.non_terminated_nodes({}) assert len(existing_nodes) == 1 self.provider.terminate_node(existing_nodes[0]) config_path = self.write_config(config) runner = MockProcessRunner() runner.respond_to_call("json .Mounts", ["[]"]) # Two initial calls to rsync, + 2 more calls during run_init runner.respond_to_call(".State.Running", ["false", "false", "false", "false"]) runner.respond_to_call("json .Config.Env", ["[]"]) commands.get_or_create_head_node( config, printable_config_file=config_path, no_restart=False, restart_only=False, yes=True, override_cluster_name=None, _provider=self.provider, _runner=runner, ) self.waitForNodes(1) # Init & Setup commands must be run for Docker! runner.assert_has_call("1.2.3.4", "init_cmd") runner.assert_has_call("1.2.3.4", "head_setup_cmd") runner.assert_has_call("1.2.3.4", "start_ray_head") self.assertEqual(self.provider.mock_nodes["0"].node_type, "head") runner.assert_has_call("1.2.3.4", pattern="docker run") docker_mount_prefix = get_docker_host_mount_location( SMALL_CLUSTER["cluster_name"] ) runner.assert_not_has_call( "1.2.3.4", pattern=f"-v {docker_mount_prefix}/~/ray_bootstrap_config" ) common_container_copy = f"rsync -e.*docker exec -i.*{docker_mount_prefix}/~/" runner.assert_has_call( "1.2.3.4", pattern=common_container_copy + "ray_bootstrap_key.pem" ) runner.assert_has_call( "1.2.3.4", pattern=common_container_copy + "ray_bootstrap_config.yaml" ) # This next section of code ensures that the following order of # commands are executed: # 1. mkdir -p {docker_mount_prefix} # 2. rsync bootstrap files (over ssh) # 3. rsync bootstrap files into container commands_with_mount = [ (i, cmd) for i, cmd in enumerate(runner.command_history()) if docker_mount_prefix in cmd ] rsync_commands = [x for x in commands_with_mount if "rsync --rsh" in x[1]] copy_into_container = [ x for x in commands_with_mount if re.search("rsync -e.*docker exec -i", x[1]) ] first_mkdir = min(x[0] for x in commands_with_mount if "mkdir" in x[1]) docker_run_cmd_indx = [ i for i, cmd in enumerate(runner.command_history()) if "docker run" in cmd ][0] for file_to_check in ["ray_bootstrap_config.yaml", "ray_bootstrap_key.pem"]: first_rsync = min( x[0] for x in rsync_commands if "ray_bootstrap_config.yaml" in x[1] ) first_cp = min(x[0] for x in copy_into_container if file_to_check in x[1]) # Ensures that `mkdir -p` precedes `docker run` because Docker # will auto-create the folder with wrong permissions. assert first_mkdir < docker_run_cmd_indx # Ensures that the folder is created before running rsync. assert first_mkdir < first_rsync # Checks that the file is present before copying into the container assert first_rsync < first_cp def testGetOrCreateHeadNodeFromStoppedRestartOnly(self): config = self.testGetOrCreateHeadNode() self.provider.cache_stopped = True existing_nodes = self.provider.non_terminated_nodes({}) assert len(existing_nodes) == 1 self.provider.terminate_node(existing_nodes[0]) config_path = self.write_config(config) runner = MockProcessRunner() runner.respond_to_call("json .Mounts", ["[]"]) # Two initial calls to rsync, + 2 more calls during run_init runner.respond_to_call(".State.Running", ["false", "false", "false", "false"]) runner.respond_to_call("json .Config.Env", ["[]"]) commands.get_or_create_head_node( config, printable_config_file=config_path, no_restart=False, restart_only=True, yes=True, override_cluster_name=None, _provider=self.provider, _runner=runner, ) self.waitForNodes(1) # Init & Setup commands must be run for Docker! runner.assert_has_call("1.2.3.4", "init_cmd") runner.assert_has_call("1.2.3.4", "head_setup_cmd") runner.assert_has_call("1.2.3.4", "start_ray_head") def testDockerFileMountsAdded(self): config = copy.deepcopy(SMALL_CLUSTER) config["file_mounts"] = {"source": "/dev/null"} config_path = self.write_config(config) self.provider = MockProvider() runner = MockProcessRunner() mounts = [ { "Type": "bind", "Source": "/sys", "Destination": "/sys", "Mode": "ro", "RW": False, "Propagation": "rprivate", } ] runner.respond_to_call("json .Mounts", [json.dumps(mounts)]) # Two initial calls to rsync, +1 more call during run_init runner.respond_to_call(".State.Running", ["false", "false", "true", "true"]) runner.respond_to_call("json .Config.Env", ["[]"]) commands.get_or_create_head_node( config, printable_config_file=config_path, no_restart=False, restart_only=False, yes=True, override_cluster_name=None, _provider=self.provider, _runner=runner, ) self.waitForNodes(1) runner.assert_has_call("1.2.3.4", "init_cmd") runner.assert_has_call("1.2.3.4", "head_setup_cmd") runner.assert_has_call("1.2.3.4", "start_ray_head") self.assertEqual(self.provider.mock_nodes["0"].node_type, "head") runner.assert_has_call("1.2.3.4", pattern="docker stop") runner.assert_has_call("1.2.3.4", pattern="docker run") docker_mount_prefix = get_docker_host_mount_location( SMALL_CLUSTER["cluster_name"] ) runner.assert_not_has_call( "1.2.3.4", pattern=f"-v {docker_mount_prefix}/~/ray_bootstrap_config" ) common_container_copy = f"rsync -e.*docker exec -i.*{docker_mount_prefix}/~/" runner.assert_has_call( "1.2.3.4", pattern=common_container_copy + "ray_bootstrap_key.pem" ) runner.assert_has_call( "1.2.3.4", pattern=common_container_copy + "ray_bootstrap_config.yaml" ) def testDockerFileMountsRemoved(self): config = copy.deepcopy(SMALL_CLUSTER) config["file_mounts"] = {} config_path = self.write_config(config) self.provider = MockProvider() runner = MockProcessRunner() mounts = [ { "Type": "bind", "Source": "/sys", "Destination": "/sys", "Mode": "ro", "RW": False, "Propagation": "rprivate", } ] runner.respond_to_call("json .Mounts", [json.dumps(mounts)]) # Two initial calls to rsync, +1 more call during run_init runner.respond_to_call(".State.Running", ["false", "false", "true", "true"]) runner.respond_to_call("json .Config.Env", ["[]"]) commands.get_or_create_head_node( config, printable_config_file=config_path, no_restart=False, restart_only=False, yes=True, override_cluster_name=None, _provider=self.provider, _runner=runner, ) self.waitForNodes(1) runner.assert_has_call("1.2.3.4", "init_cmd") runner.assert_has_call("1.2.3.4", "head_setup_cmd") runner.assert_has_call("1.2.3.4", "start_ray_head") self.assertEqual(self.provider.mock_nodes["0"].node_type, "head") # We only removed amount from the YAML, no changes should happen. runner.assert_not_has_call("1.2.3.4", pattern="docker stop") runner.assert_not_has_call("1.2.3.4", pattern="docker run") docker_mount_prefix = get_docker_host_mount_location( SMALL_CLUSTER["cluster_name"] ) runner.assert_not_has_call( "1.2.3.4", pattern=f"-v {docker_mount_prefix}/~/ray_bootstrap_config" ) common_container_copy = f"rsync -e.*docker exec -i.*{docker_mount_prefix}/~/" runner.assert_has_call( "1.2.3.4", pattern=common_container_copy + "ray_bootstrap_key.pem" ) runner.assert_has_call( "1.2.3.4", pattern=common_container_copy + "ray_bootstrap_config.yaml" ) def testRsyncCommandWithDocker(self): assert SMALL_CLUSTER["docker"]["container_name"] config_path = self.write_config(SMALL_CLUSTER) self.provider = MockProvider(unique_ips=True) self.provider.create_node( {}, {TAG_RAY_NODE_KIND: "head", TAG_RAY_NODE_STATUS: "up-to-date"}, 1 ) self.provider.create_node( {}, {TAG_RAY_NODE_KIND: "worker", TAG_RAY_NODE_STATUS: "up-to-date"}, 10 ) self.provider.finish_starting_nodes() ray.autoscaler.node_provider._get_node_provider = Mock( return_value=self.provider ) ray.autoscaler._private.commands._bootstrap_config = Mock( return_value=SMALL_CLUSTER ) runner = MockProcessRunner() commands.rsync( config_path, source=config_path, target="/tmp/test_path", override_cluster_name=None, down=True, _runner=runner, ) runner.assert_has_call("1.2.3.0", pattern="rsync -e.*docker exec -i") runner.assert_has_call("1.2.3.0", pattern="rsync --rsh") runner.clear_history() commands.rsync( config_path, source=config_path, target="/tmp/test_path", override_cluster_name=None, down=True, ip_address="1.2.3.5", _runner=runner, ) runner.assert_has_call("1.2.3.5", pattern="rsync -e.*docker exec -i") runner.assert_has_call("1.2.3.5", pattern="rsync --rsh") runner.clear_history() commands.rsync( config_path, source=config_path, target="/tmp/test_path", ip_address="172.0.0.4", override_cluster_name=None, down=True, use_internal_ip=True, _runner=runner, ) runner.assert_has_call("172.0.0.4", pattern="rsync -e.*docker exec -i") runner.assert_has_call("172.0.0.4", pattern="rsync --rsh") def testRsyncCommandWithoutDocker(self): cluster_cfg = copy.deepcopy(SMALL_CLUSTER) cluster_cfg["docker"] = {} config_path = self.write_config(cluster_cfg) self.provider = MockProvider(unique_ips=True) self.provider.create_node( {}, {TAG_RAY_NODE_KIND: "head", TAG_RAY_NODE_STATUS: "up-to-date"}, 1 ) self.provider.create_node( {}, {TAG_RAY_NODE_KIND: "worker", TAG_RAY_NODE_STATUS: "up-to-date"}, 10 ) self.provider.finish_starting_nodes() runner = MockProcessRunner() ray.autoscaler.node_provider._get_node_provider = Mock( return_value=self.provider ) ray.autoscaler._private.commands._bootstrap_config = Mock( return_value=cluster_cfg ) commands.rsync( config_path, source=config_path, target="/tmp/test_path", override_cluster_name=None, down=True, _runner=runner, ) runner.assert_has_call("1.2.3.0", pattern="rsync") commands.rsync( config_path, source=config_path, target="/tmp/test_path", override_cluster_name=None, down=True, ip_address="1.2.3.5", _runner=runner, ) runner.assert_has_call("1.2.3.5", pattern="rsync") runner.clear_history() commands.rsync( config_path, source=config_path, target="/tmp/test_path", override_cluster_name=None, down=True, ip_address="172.0.0.4", use_internal_ip=True, _runner=runner, ) runner.assert_has_call("172.0.0.4", pattern="rsync") runner.clear_history() def testSummarizerFailedCreate(self): """Checks that event summarizer reports failed node creation.""" config = copy.deepcopy(SMALL_CLUSTER) config["available_node_types"]["worker"]["min_workers"] = 2 config_path = self.write_config(config) self.provider = MockProvider() runner = MockProcessRunner() mock_metrics = Mock(spec=AutoscalerPrometheusMetrics()) self.provider.create_node( {}, { TAG_RAY_NODE_KIND: NODE_KIND_HEAD, TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, TAG_RAY_USER_NODE_TYPE: "head", }, 1, ) self.provider.creation_error = Exception(":(") autoscaler = MockAutoscaler( config_path, LoadMetrics(), MockGcsClient(), max_failures=0, process_runner=runner, update_interval_s=0, prom_metrics=mock_metrics, ) assert len(self.provider.non_terminated_nodes(WORKER_FILTER)) == 0 autoscaler.update() # Expect the next two messages in the logs. msg = "Failed to launch 2 node(s) of type worker." def expected_message_logged(): return msg in autoscaler.event_summarizer.summary() self.waitFor(expected_message_logged) def testSummarizerFailedCreateStructuredError(self): """Checks that event summarizer reports failed node creation with additional details when the node provider thorws a NodeLaunchException.""" config = copy.deepcopy(SMALL_CLUSTER) config["available_node_types"]["worker"]["min_workers"] = 2 config_path = self.write_config(config) self.provider = MockProvider() runner = MockProcessRunner() mock_metrics = Mock(spec=AutoscalerPrometheusMetrics()) self.provider.create_node( {}, { TAG_RAY_NODE_KIND: NODE_KIND_HEAD, TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, TAG_RAY_USER_NODE_TYPE: "head", }, 1, ) self.provider.creation_error = NodeLaunchException( "didn't work", "never did", exc_info ) autoscaler = MockAutoscaler( config_path, LoadMetrics(), MockGcsClient(), max_failures=0, process_runner=runner, update_interval_s=0, prom_metrics=mock_metrics, ) assert len(self.provider.non_terminated_nodes(WORKER_FILTER)) == 0 autoscaler.update() # Expect the next message in the logs. msg = "Failed to launch 2 node(s) of type worker. (didn't work): never did." def expected_message_logged(): print(autoscaler.event_summarizer.summary()) return msg in autoscaler.event_summarizer.summary() self.waitFor(expected_message_logged) def testSummarizerFailedCreateStructuredErrorNoUnderlyingException(self): """Checks that event summarizer reports failed node creation with additional details when the node provider thorws a NodeLaunchException.""" config = copy.deepcopy(SMALL_CLUSTER) config["available_node_types"]["worker"]["min_workers"] = 2 config_path = self.write_config(config) self.provider = MockProvider() runner = MockProcessRunner() mock_metrics = Mock(spec=AutoscalerPrometheusMetrics()) self.provider.create_node( {}, { TAG_RAY_NODE_KIND: NODE_KIND_HEAD, TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, TAG_RAY_USER_NODE_TYPE: "head", }, 1, ) self.provider.creation_error = NodeLaunchException( "didn't work", "never did", src_exc_info=None ) autoscaler = MockAutoscaler( config_path, LoadMetrics(), MockGcsClient(), max_failures=0, process_runner=runner, update_interval_s=0, prom_metrics=mock_metrics, ) assert len(self.provider.non_terminated_nodes(WORKER_FILTER)) == 0 autoscaler.update() # Expect the next message in the logs. msg = "Failed to launch 2 node(s) of type worker. (didn't work): never did." def expected_message_logged(): print(autoscaler.event_summarizer.summary()) return msg in autoscaler.event_summarizer.summary() self.waitFor(expected_message_logged) def testReadonlyNodeProvider(self): config = copy.deepcopy(SMALL_CLUSTER) config["available_node_types"]["worker"]["min_workers"] = 2 config_path = self.write_config(config) self.provider = ReadOnlyNodeProvider(config_path, "readonly") runner = MockProcessRunner() mock_metrics = Mock(spec=AutoscalerPrometheusMetrics()) autoscaler = StandardAutoscaler( config_path, LoadMetrics(), MockGcsClient(), max_failures=0, process_runner=runner, update_interval_s=0, prom_metrics=mock_metrics, ) assert len(self.provider.non_terminated_nodes({})) == 0 # No updates in read-only mode. autoscaler.update() self.waitForNodes(0) assert mock_metrics.started_nodes.inc.call_count == 0 assert len(runner.calls) == 0 # Reflect updates to the readonly provider. self.provider._set_nodes( [ ("foo1", "1.1.1.1"), ("foo2", "1.1.1.1"), ("foo3", "1.1.1.1"), ] ) # No updates in read-only mode. autoscaler.update() self.waitForNodes(3) assert mock_metrics.started_nodes.inc.call_count == 0 assert mock_metrics.stopped_nodes.inc.call_count == 0 assert mock_metrics.drain_node_exceptions.inc.call_count == 0 assert len(runner.calls) == 0 events = autoscaler.event_summarizer.summary() assert not events, events def ScaleUpHelper(self, disable_node_updaters): config = copy.deepcopy(SMALL_CLUSTER) config["available_node_types"]["worker"]["min_workers"] = 2 config_path = self.write_config(config) config["provider"]["disable_node_updaters"] = disable_node_updaters config_path = self.write_config(config) self.provider = MockProvider() runner = MockProcessRunner() mock_metrics = Mock(spec=AutoscalerPrometheusMetrics()) self.provider.create_node( {}, { TAG_RAY_NODE_KIND: NODE_KIND_HEAD, TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, TAG_RAY_USER_NODE_TYPE: "head", }, 1, ) if disable_node_updaters: # This class raises an assertion error if we try to create # a node updater thread. autoscaler_class = NoUpdaterMockAutoscaler else: autoscaler_class = MockAutoscaler autoscaler = autoscaler_class( config_path, LoadMetrics(), MockGcsClient(), max_failures=0, process_runner=runner, update_interval_s=0, prom_metrics=mock_metrics, ) assert len(self.provider.non_terminated_nodes(WORKER_FILTER)) == 0 autoscaler.update() self.waitForNodes(2, tag_filters=WORKER_FILTER) # started_nodes metric should have been incremented by 2 assert mock_metrics.started_nodes.inc.call_count == 1 mock_metrics.started_nodes.inc.assert_called_with(2) assert mock_metrics.worker_create_node_time.observe.call_count == 2 autoscaler.update() # The two autoscaler update iterations in this test led to two # observations of the update time. assert mock_metrics.update_time.observe.call_count == 2 self.waitForNodes(2, tag_filters=WORKER_FILTER) # running_workers metric should be set to 2 mock_metrics.running_workers.set.assert_called_with(2) if disable_node_updaters: # Node Updaters have NOT been invoked because they were explicitly # disabled. assert len(runner.calls) == 0 # Nodes were create in uninitialized and not updated. self.waitForNodes( 2, tag_filters={ TAG_RAY_NODE_STATUS: STATUS_UNINITIALIZED, **WORKER_FILTER, }, ) else: # Node Updaters have been invoked. self.waitFor(lambda: len(runner.calls) > 0) # The updates failed. Key thing is that the updates completed. self.waitForNodes( 2, tag_filters={ TAG_RAY_NODE_STATUS: STATUS_UPDATE_FAILED, **WORKER_FILTER, }, ) assert mock_metrics.drain_node_exceptions.inc.call_count == 0 def testScaleUp(self): self.ScaleUpHelper(disable_node_updaters=False) def testScaleUpNoUpdaters(self): self.ScaleUpHelper(disable_node_updaters=True) def testTerminateOutdatedNodesGracefully(self): config = copy.deepcopy(SMALL_CLUSTER) config["available_node_types"]["worker"]["min_workers"] = 5 config["max_workers"] = 5 config["available_node_types"]["worker"]["max_workers"] = 5 config_path = self.write_config(config) self.provider = MockProvider() self.provider.create_node( {}, { TAG_RAY_NODE_KIND: NODE_KIND_HEAD, TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, TAG_RAY_USER_NODE_TYPE: "head", }, 1, ) self.provider.create_node( {}, { TAG_RAY_NODE_KIND: "worker", TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, TAG_RAY_USER_NODE_TYPE: "worker", }, 10, ) runner = MockProcessRunner() runner.respond_to_call("json .Config.Env", ["[]" for i in range(10)]) mock_metrics = Mock(spec=AutoscalerPrometheusMetrics()) lm = LoadMetrics() autoscaler = MockAutoscaler( config_path, lm, MockGcsClient(), max_failures=0, process_runner=runner, update_interval_s=0, prom_metrics=mock_metrics, ) self.waitForNodes(10, tag_filters=WORKER_FILTER) fill_in_node_ids(self.provider, lm) # Gradually scales down to meet target size, never going too low for _ in range(10): autoscaler.update() self.waitForNodes( 5, comparison=self.assertLessEqual, tag_filters=WORKER_FILTER ) self.waitForNodes( 4, comparison=self.assertGreaterEqual, tag_filters=WORKER_FILTER ) # Eventually reaches steady state self.waitForNodes(5, tag_filters=WORKER_FILTER) # Check the outdated node removal event is generated. autoscaler.update() events = autoscaler.event_summarizer.summary() assert "Removing 10 nodes of type worker (outdated)." in events, events assert mock_metrics.stopped_nodes.inc.call_count == 10 mock_metrics.started_nodes.inc.assert_called_with(5) assert mock_metrics.worker_create_node_time.observe.call_count == 5 assert mock_metrics.drain_node_exceptions.inc.call_count == 0 # Parameterization functionality in the unittest module is not great. # To test scale-down behavior, we parameterize the DynamicScaling test # manually over outcomes for the DrainNode RPC call. def testDynamicScaling1(self): self.helperDynamicScaling(DrainNodeOutcome.Succeeded) def testDynamicScaling2(self): self.helperDynamicScaling(DrainNodeOutcome.NotAllDrained) def testDynamicScaling3(self): self.helperDynamicScaling(DrainNodeOutcome.Unimplemented) def testDynamicScaling4(self): self.helperDynamicScaling(DrainNodeOutcome.GenericRpcError) def testDynamicScaling5(self): self.helperDynamicScaling(DrainNodeOutcome.GenericException) def testDynamicScaling6(self): self.helperDynamicScaling(DrainNodeOutcome.FailedToFindIp) def testDynamicScalingForegroundLauncher(self): """Test autoscaling with node launcher in the foreground.""" self.helperDynamicScaling(foreground_node_launcher=True) def testDynamicScalingBatchingNodeProvider(self): """Test autoscaling with BatchingNodeProvider""" self.helperDynamicScaling( foreground_node_launcher=True, batching_node_provider=True ) def helperDynamicScaling( self, drain_node_outcome: DrainNodeOutcome = DrainNodeOutcome.Succeeded, foreground_node_launcher: bool = False, batching_node_provider: bool = False, ): mock_metrics = Mock(spec=AutoscalerPrometheusMetrics()) mock_gcs_client = MockGcsClient(drain_node_outcome) # Run the core of the test logic. self._helperDynamicScaling( mock_metrics, mock_gcs_client, foreground_node_launcher=foreground_node_launcher, batching_node_provider=batching_node_provider, ) # Make assertions about DrainNode error handling during scale-down. if drain_node_outcome == DrainNodeOutcome.Succeeded: # DrainNode call was made. assert mock_gcs_client.drain_node_call_count > 0 # No drain node exceptions. assert mock_metrics.drain_node_exceptions.inc.call_count == 0 # Each drain node call succeeded. assert ( mock_gcs_client.drain_node_reply_success == mock_gcs_client.drain_node_call_count ) elif drain_node_outcome == DrainNodeOutcome.Unimplemented: # DrainNode call was made. assert mock_gcs_client.drain_node_call_count > 0 # All errors were supressed. assert mock_metrics.drain_node_exceptions.inc.call_count == 0 # Every call failed. assert mock_gcs_client.drain_node_reply_success == 0 elif drain_node_outcome in ( DrainNodeOutcome.GenericRpcError, DrainNodeOutcome.GenericException, ): # DrainNode call was made. assert mock_gcs_client.drain_node_call_count > 0 # We encountered an exception. assert mock_metrics.drain_node_exceptions.inc.call_count > 0 # Every call failed. assert ( mock_metrics.drain_node_exceptions.inc.call_count == mock_gcs_client.drain_node_call_count ) assert mock_gcs_client.drain_node_reply_success == 0 elif drain_node_outcome == DrainNodeOutcome.FailedToFindIp: # We never called the drain node api because we were unable to # fetch ips assert mock_gcs_client.drain_node_call_count == 0 # We encountered an exception fetching ip. assert mock_metrics.drain_node_exceptions.inc.call_count > 0 def _helperDynamicScaling( self, mock_metrics, mock_gcs_client, foreground_node_launcher=False, batching_node_provider=False, ): if batching_node_provider: assert ( foreground_node_launcher ), "BatchingNodeProvider requires foreground node launch." config = copy.deepcopy(SMALL_CLUSTER) config["available_node_types"]["worker"]["min_workers"] = 2 if foreground_node_launcher: config["provider"][FOREGROUND_NODE_LAUNCH_KEY] = True if batching_node_provider: config["provider"][FOREGROUND_NODE_LAUNCH_KEY] = True config["provider"][DISABLE_LAUNCH_CONFIG_CHECK_KEY] = True config["provider"][DISABLE_NODE_UPDATERS_KEY] = True config_path = self.write_config(config) if batching_node_provider: self.provider = MockBatchingNodeProvider( provider_config={ DISABLE_LAUNCH_CONFIG_CHECK_KEY: True, DISABLE_NODE_UPDATERS_KEY: True, FOREGROUND_NODE_LAUNCH_KEY: True, }, cluster_name="test-cluster", ) else: self.provider = MockProvider() runner = MockProcessRunner() runner.respond_to_call("json .Config.Env", ["[]" for i in range(12)]) lm = LoadMetrics() # As part of setup for this test, ensure there is a head node. if batching_node_provider: # MockBatchingNodeProvider creates a head node in the __init__ method. pass else: # MockProvider needs to create a head node with create_node. self.provider.create_node( {}, { TAG_RAY_NODE_KIND: NODE_KIND_HEAD, TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, TAG_RAY_USER_NODE_TYPE: "head", }, 1, ) lm.update("172.0.0.0", mock_node_id(), {"CPU": 1}, {"CPU": 0}, 0) autoscaler = MockAutoscaler( config_path, lm, mock_gcs_client, max_launch_batch=5, max_concurrent_launches=5, max_failures=0, process_runner=runner, update_interval_s=0, prom_metrics=mock_metrics, ) if mock_gcs_client.drain_node_outcome == DrainNodeOutcome.FailedToFindIp: autoscaler.fail_to_find_ip_during_drain = True self.waitForNodes(0, tag_filters=WORKER_FILTER) # Test aborting an autoscaler update with the batching NodeProvider. if batching_node_provider: self.provider.safe_to_scale_flag = False autoscaler.update() # The autoscaler update was aborted, so there's no change in worker count. assert self.num_nodes(tag_filters=WORKER_FILTER) == 0 self.provider.safe_to_scale_flag = True autoscaler.update() if foreground_node_launcher: # If we launched in the foreground, shouldn't need to wait for nodes # to be available. (Node creation should block.) assert self.num_nodes(tag_filters=WORKER_FILTER) == 2, ( self.provider.non_terminated_nodes(tag_filters=WORKER_FILTER), self.provider.non_terminated_nodes(tag_filters={}), ) else: self.waitForNodes(2, tag_filters=WORKER_FILTER) # Update the config to reduce the cluster size new_config = copy.deepcopy(SMALL_CLUSTER) new_config["max_workers"] = 1 new_config["available_node_types"]["worker"]["max_workers"] = 1 new_config["available_node_types"]["worker"]["min_workers"] = 1 self.write_config(new_config) fill_in_node_ids(self.provider, lm) autoscaler.update() self.waitForNodes(1, tag_filters={TAG_RAY_NODE_KIND: NODE_KIND_WORKER}) # Check the scale-down event is generated. events = autoscaler.event_summarizer.summary() assert "Removing 1 nodes of type worker (max_workers_per_type)." in events assert mock_metrics.stopped_nodes.inc.call_count == 1 # Update the config to increase the cluster size new_config["available_node_types"]["worker"]["min_workers"] = 10 new_config["available_node_types"]["worker"]["max_workers"] = 10 new_config["max_workers"] = 10 self.write_config(new_config) autoscaler.update() # TODO(rickyx): This is a hack to avoid running into race conditions # within v1 autoscaler. These should no longer be relevant in v2. time.sleep(3) # Because one worker already started, the scheduler waits for its # resources to be updated before it launches the remaining min_workers. worker_ip = self.provider.non_terminated_node_ips( tag_filters={TAG_RAY_NODE_KIND: NODE_KIND_WORKER}, )[0] lm.update( worker_ip, mock_node_id(), {"CPU": 1}, {"CPU": 1}, DUMMY_IDLE_DURATION_S ) autoscaler.update() # TODO(rickyx): This is a hack to avoid running into race conditions # within v1 autoscaler. These should no longer be relevant in v2. time.sleep(3) if foreground_node_launcher: # If we launched in the foreground, shouldn't need to wait for nodes # to be available. (Node creation should block.) assert self.num_nodes(tag_filters=WORKER_FILTER) == 10 else: self.waitForNodes(10, tag_filters=WORKER_FILTER) # Awkward and unecessary to repeat the following check for BatchingNodeProvider. if not batching_node_provider: # Verify that worker nodes were launched in the main thread if foreground # node launch is enabled, in a subthread otherwise. self.worker_node_thread_check(foreground_node_launcher) autoscaler.update() assert mock_metrics.running_workers.set.call_args_list[-1][0][0] >= 10 # def testAggressiveAutoscaling(self): # self._aggressiveAutoscalingHelper() # def testAggressiveAutoscalingWithForegroundLauncher(self): # self._aggressiveAutoscalingHelper(foreground_node_launcher=True) # def _aggressiveAutoscalingHelper(self, foreground_node_launcher: bool = False): # config = copy.deepcopy(SMALL_CLUSTER) # config["available_node_types"]["worker"]["min_workers"] = 0 # config["available_node_types"]["worker"]["max_workers"] = 10 # config["max_workers"] = 10 # config["idle_timeout_minutes"] = 0 # config["upscaling_speed"] = config["available_node_types"]["worker"][ # "max_workers" # ] # if foreground_node_launcher: # config["provider"][FOREGROUND_NODE_LAUNCH_KEY] = True # config_path = self.write_config(config) # self.provider = MockProvider() # self.provider.create_node( # {}, # { # TAG_RAY_NODE_KIND: NODE_KIND_HEAD, # TAG_RAY_USER_NODE_TYPE: "head", # }, # 1, # ) # head_ip = self.provider.non_terminated_node_ips( # tag_filters={TAG_RAY_NODE_KIND: NODE_KIND_HEAD}, # )[0] # runner = MockProcessRunner() # runner.respond_to_call("json .Config.Env", ["[]" for i in range(11)]) # lm = LoadMetrics() # autoscaler = MockAutoscaler( # config_path, # lm, # MockGcsClient(), # max_launch_batch=5, # max_concurrent_launches=5, # max_failures=0, # process_runner=runner, # update_interval_s=0, # ) # self.waitForNodes(1) # lm.update( # head_ip, # mock_node_id(), # {"CPU": 1}, # {"CPU": 0}, # waiting_bundles=[{"CPU": 1}] * 7, # infeasible_bundles=[{"CPU": 1}] * 3, # ) # autoscaler.update() # if foreground_node_launcher: # # No wait if node launch is blocking and happens in the foreground. # assert self.num_nodes() == 11 # else: # self.waitForNodes(11) # self.worker_node_thread_check(foreground_node_launcher) # worker_ips = self.provider.non_terminated_node_ips( # tag_filters={TAG_RAY_NODE_KIND: NODE_KIND_WORKER}, # ) # for ip in worker_ips: # # Mark workers inactive. # lm.last_used_time_by_ip[ip] = 0 # # Clear the resource demands. # # Otherwise in "foreground launcher" mode, workers would be deleted # # for being idle and instantly re-created due to resource demand! # lm.update( # head_ip, # mock_node_id(), # {}, # {}, # waiting_bundles=[], # infeasible_bundles=[], # ) # autoscaler.update() # self.waitForNodes(1) # only the head node # # Make sure they don't get overwritten. # assert autoscaler.resource_demand_scheduler.node_types["head"]["resources"] # == { # "CPU": 1 # } # assert autoscaler.resource_demand_scheduler.node_types["worker"][ # "resources" # ] == {"CPU": 1} def testUnmanagedNodes(self): config = copy.deepcopy(SMALL_CLUSTER) config["available_node_types"]["worker"]["min_workers"] = 0 config["available_node_types"]["worker"]["max_workers"] = 20 config["max_workers"] = 20 config["idle_timeout_minutes"] = 0 config["upscaling_speed"] = 9999 config_path = self.write_config(config) self.provider = MockProvider() self.provider.create_node( {}, { TAG_RAY_NODE_KIND: "head", TAG_RAY_USER_NODE_TYPE: "head", TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, }, 1, ) head_ip = self.provider.non_terminated_node_ips( tag_filters={TAG_RAY_NODE_KIND: "head"}, )[0] self.provider.create_node({}, {TAG_RAY_NODE_KIND: "unmanaged"}, 1) unmanaged_ip = self.provider.non_terminated_node_ips( tag_filters={TAG_RAY_NODE_KIND: "unmanaged"}, )[0] runner = MockProcessRunner() lm = LoadMetrics() lm.local_ip = head_ip autoscaler = MockAutoscaler( config_path, lm, MockGcsClient(), max_launch_batch=5, max_concurrent_launches=5, max_failures=0, process_runner=runner, update_interval_s=0, ) autoscaler.update() self.waitForNodes(2) # This node has num_cpus=0 lm.update(head_ip, mock_node_id(), {"CPU": 1}, {"CPU": 0}, 0) lm.update( unmanaged_ip, mock_node_id(), {"CPU": 0}, {"CPU": 0}, DUMMY_IDLE_DURATION_S, ) autoscaler.update() self.waitForNodes(2) # 1 CPU task cannot be scheduled. lm.update( unmanaged_ip, mock_node_id(), {"CPU": 0}, {"CPU": 0}, DUMMY_IDLE_DURATION_S, waiting_bundles=[{"CPU": 1}], ) autoscaler.update() self.waitForNodes(3) def testUnmanagedNodes2(self): config = copy.deepcopy(SMALL_CLUSTER) config["available_node_types"]["worker"]["min_workers"] = 0 config["available_node_types"]["worker"]["max_workers"] = 20 config["max_workers"] = 20 config["idle_timeout_minutes"] = 0 config["upscaling_speed"] = 9999 config_path = self.write_config(config) self.provider = MockProvider() self.provider.create_node( {}, { TAG_RAY_NODE_KIND: "head", TAG_RAY_USER_NODE_TYPE: "head", TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, }, 1, ) head_ip = self.provider.non_terminated_node_ips( tag_filters={TAG_RAY_NODE_KIND: "head"}, )[0] self.provider.create_node({}, {TAG_RAY_NODE_KIND: "unmanaged"}, 1) unmanaged_ip = self.provider.non_terminated_node_ips( tag_filters={TAG_RAY_NODE_KIND: "unmanaged"}, )[0] runner = MockProcessRunner() lm = LoadMetrics() lm.local_ip = head_ip autoscaler = MockAutoscaler( config_path, lm, MockGcsClient(), max_launch_batch=5, max_concurrent_launches=5, max_failures=0, process_runner=runner, update_interval_s=0, ) lm.update(head_ip, mock_node_id(), {"CPU": 1}, {"CPU": 0}, 0) lm.update( unmanaged_ip, mock_node_id(), {"CPU": 0}, {"CPU": 0}, DUMMY_IDLE_DURATION_S, ) # Note that we shouldn't autoscale here because the resource demand # vector is not set and target utilization fraction = 1. autoscaler.update() # If the autoscaler was behaving incorrectly, it needs time to start # the new node, otherwise it could scale up after this check. time.sleep(0.2) self.waitForNodes(2) def testDelayedLaunch(self): config_path = self.write_config(SMALL_CLUSTER) self.provider = MockProvider() runner = MockProcessRunner() runner.respond_to_call("json .Config.Env", ["[]" for i in range(2)]) lm = LoadMetrics() self.provider.create_node( {}, { TAG_RAY_NODE_KIND: NODE_KIND_HEAD, TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, TAG_RAY_USER_NODE_TYPE: "head", }, 1, ) head_ip = self.provider.non_terminated_node_ips( tag_filters={TAG_RAY_NODE_KIND: NODE_KIND_HEAD}, )[0] autoscaler = MockAutoscaler( config_path, lm, MockGcsClient(), max_launch_batch=5, max_concurrent_launches=5, max_failures=0, process_runner=runner, update_interval_s=0, ) assert ( len( self.provider.non_terminated_nodes( {TAG_RAY_NODE_KIND: NODE_KIND_WORKER} ) ) == 0 ) # Update will try to create, but will block until we set the flag self.provider.ready_to_create.clear() lm.update( head_ip, mock_node_id(), {"CPU": 1}, {"CPU": 0}, 0, waiting_bundles=[{"CPU": 1}] * 2, ) autoscaler.update() assert ( len( self.provider.non_terminated_nodes( {TAG_RAY_NODE_KIND: NODE_KIND_WORKER} ) ) == 0 ) assert autoscaler.pending_launches.value == 2 # Set the flag, check it updates self.provider.ready_to_create.set() self.waitForNodes(2, tag_filters={TAG_RAY_NODE_KIND: NODE_KIND_WORKER}) assert autoscaler.pending_launches.value == 0 # Update the config to reduce the cluster size new_config = copy.deepcopy(SMALL_CLUSTER) new_config["available_node_types"]["worker"]["max_workers"] = 1 self.write_config(new_config) fill_in_node_ids(self.provider, lm) autoscaler.update() assert ( len( self.provider.non_terminated_nodes( {TAG_RAY_NODE_KIND: NODE_KIND_WORKER} ) ) == 1 ) def testDelayedLaunchWithMinWorkers(self): config = copy.deepcopy(SMALL_CLUSTER) config["available_node_types"]["worker"]["min_workers"] = 10 config["available_node_types"]["worker"]["max_workers"] = 10 config["max_workers"] = 10 config_path = self.write_config(config) self.provider = MockProvider() runner = MockProcessRunner() runner.respond_to_call("json .Config.Env", ["[]" for i in range(10)]) mock_metrics = Mock(spec=AutoscalerPrometheusMetrics()) self.provider.create_node( {}, { TAG_RAY_NODE_KIND: NODE_KIND_HEAD, TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, TAG_RAY_USER_NODE_TYPE: "head", }, 1, ) autoscaler = MockAutoscaler( config_path, LoadMetrics(), MockGcsClient(), max_launch_batch=5, max_concurrent_launches=8, max_failures=0, process_runner=runner, update_interval_s=0, prom_metrics=mock_metrics, ) assert ( len( self.provider.non_terminated_nodes( {TAG_RAY_NODE_KIND: NODE_KIND_WORKER} ) ) == 0 ) # update() should launch a wave of 5 nodes (max_launch_batch) # Force this first wave to block. rtc1 = self.provider.ready_to_create rtc1.clear() autoscaler.update() # Synchronization: wait for launchy thread to be blocked on rtc1 waiters = rtc1._cond._waiters self.waitFor(lambda: len(waiters) == 2) assert autoscaler.pending_launches.value == 10 assert ( len( self.provider.non_terminated_nodes( {TAG_RAY_NODE_KIND: NODE_KIND_WORKER} ) ) == 0 ) autoscaler.update() self.waitForNodes( 0, tag_filters={TAG_RAY_NODE_KIND: NODE_KIND_WORKER} ) # Nodes are not added on top of pending. rtc1.set() self.waitFor(lambda: autoscaler.pending_launches.value == 0) assert ( len( self.provider.non_terminated_nodes( {TAG_RAY_NODE_KIND: NODE_KIND_WORKER} ) ) == 10 ) self.waitForNodes(10, tag_filters={TAG_RAY_NODE_KIND: NODE_KIND_WORKER}) assert autoscaler.pending_launches.value == 0 autoscaler.update() self.waitForNodes(10, tag_filters={TAG_RAY_NODE_KIND: NODE_KIND_WORKER}) assert autoscaler.pending_launches.value == 0 assert mock_metrics.drain_node_exceptions.inc.call_count == 0 def testUpdateThrottling(self): config = copy.deepcopy(SMALL_CLUSTER) config["available_node_types"]["worker"]["min_workers"] = 2 config_path = self.write_config(config) self.provider = MockProvider() runner = MockProcessRunner() self.provider.create_node( {}, { TAG_RAY_NODE_KIND: NODE_KIND_HEAD, TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, TAG_RAY_USER_NODE_TYPE: "head", }, 1, ) autoscaler = MockAutoscaler( config_path, LoadMetrics(), MockGcsClient(), max_launch_batch=5, max_concurrent_launches=5, max_failures=0, process_runner=runner, update_interval_s=10, ) autoscaler.update() self.waitForNodes(2, tag_filters=WORKER_FILTER) assert autoscaler.pending_launches.value == 0 new_config = copy.deepcopy(SMALL_CLUSTER) new_config["max_workers"] = 1 self.write_config(new_config) autoscaler.update() # not updated yet # note that node termination happens in the main thread, so # we do not need to add any delay here before checking assert len(self.provider.non_terminated_nodes(WORKER_FILTER)) == 2 assert autoscaler.pending_launches.value == 0 def testLaunchConfigChange(self): config = copy.deepcopy(SMALL_CLUSTER) config["available_node_types"]["worker"]["min_workers"] = 2 config_path = self.write_config(config) self.provider = MockProvider() lm = LoadMetrics() self.provider.create_node( {}, { TAG_RAY_NODE_KIND: NODE_KIND_HEAD, TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, TAG_RAY_USER_NODE_TYPE: "head", }, 1, ) autoscaler = MockAutoscaler( config_path, lm, MockGcsClient(), max_failures=0, update_interval_s=0 ) autoscaler.update() self.waitForNodes(2, tag_filters=WORKER_FILTER) # Update the config to change the node type new_config = copy.deepcopy(config) new_config["available_node_types"]["worker"]["node_config"][ "InstanceType" ] = "updated" self.write_config(new_config) self.provider.ready_to_create.clear() fill_in_node_ids(self.provider, lm) for _ in range(5): autoscaler.update() self.waitForNodes(0, tag_filters=WORKER_FILTER) self.provider.ready_to_create.set() self.waitForNodes(2, tag_filters=WORKER_FILTER) def testIgnoresCorruptedConfig(self): config = copy.deepcopy(SMALL_CLUSTER) config["available_node_types"]["worker"]["min_workers"] = 2 config_path = self.write_config(config) self.provider = MockProvider() runner = MockProcessRunner() runner.respond_to_call("json .Config.Env", ["[]" for i in range(11)]) self.provider.create_node( {}, { TAG_RAY_NODE_KIND: NODE_KIND_HEAD, TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, TAG_RAY_USER_NODE_TYPE: "head", }, 1, ) lm = LoadMetrics() lm.update("172.0.0.0", mock_node_id(), {"CPU": 1}, {"CPU": 0}, 0) mock_metrics = Mock(spec=AutoscalerPrometheusMetrics()) autoscaler = MockAutoscaler( config_path, lm, MockGcsClient(), max_launch_batch=10, max_concurrent_launches=10, process_runner=runner, max_failures=0, update_interval_s=0, prom_metrics=mock_metrics, ) autoscaler.update() assert mock_metrics.config_validation_exceptions.inc.call_count == 0 self.waitForNodes(2, tag_filters=WORKER_FILTER) # Write a corrupted config self.write_config("asdf", call_prepare_config=False) for _ in range(10): autoscaler.update() # config validation exceptions metrics should be incremented 10 times assert mock_metrics.config_validation_exceptions.inc.call_count == 10 time.sleep(0.1) assert autoscaler.pending_launches.value == 0 assert ( len( self.provider.non_terminated_nodes( {TAG_RAY_NODE_KIND: NODE_KIND_WORKER} ) ) == 2 ) # New a good config again new_config = copy.deepcopy(SMALL_CLUSTER) new_config["available_node_types"]["worker"]["min_workers"] = 10 new_config["max_workers"] = 10 new_config["available_node_types"]["worker"]["max_workers"] = 10 self.write_config(new_config) worker_ip = self.provider.non_terminated_node_ips( tag_filters={TAG_RAY_NODE_KIND: NODE_KIND_WORKER}, )[0] # Because one worker already started, the scheduler waits for its # resources to be updated before it launches the remaining min_workers. lm.update( worker_ip, mock_node_id(), {"CPU": 1}, {"CPU": 1}, DUMMY_IDLE_DURATION_S ) autoscaler.update() self.waitForNodes(10, tag_filters={TAG_RAY_NODE_KIND: NODE_KIND_WORKER}) assert mock_metrics.drain_node_exceptions.inc.call_count == 0 def testMaxFailures(self): config_path = self.write_config(SMALL_CLUSTER) self.provider = MockProvider() self.provider.throw = True runner = MockProcessRunner() mock_metrics = Mock(spec=AutoscalerPrometheusMetrics()) autoscaler = MockAutoscaler( config_path, LoadMetrics(), MockGcsClient(), max_failures=2, process_runner=runner, update_interval_s=0, prom_metrics=mock_metrics, ) autoscaler.update() assert autoscaler.summary() is None assert mock_metrics.update_loop_exceptions.inc.call_count == 1 autoscaler.update() assert mock_metrics.update_loop_exceptions.inc.call_count == 2 with pytest.raises(Exception): autoscaler.update() assert mock_metrics.drain_node_exceptions.inc.call_count == 0 def testLaunchNewNodeOnOutOfBandTerminate(self): config = copy.deepcopy(SMALL_CLUSTER) config["available_node_types"]["worker"]["min_workers"] = 2 config_path = self.write_config(config) self.provider = MockProvider() runner = MockProcessRunner() runner.respond_to_call("json .Config.Env", ["[]" for i in range(4)]) self.provider.create_node( {}, { TAG_RAY_NODE_KIND: NODE_KIND_HEAD, TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, TAG_RAY_USER_NODE_TYPE: "head", }, 1, ) head_ip = self.provider.non_terminated_node_ips( tag_filters={TAG_RAY_NODE_KIND: NODE_KIND_HEAD}, )[0] autoscaler = MockAutoscaler( config_path, LoadMetrics(), MockGcsClient(), max_failures=0, process_runner=runner, update_interval_s=0, ) autoscaler.update() autoscaler.update() self.waitForNodes(2, tag_filters=WORKER_FILTER) for node in self.provider.mock_nodes.values(): if node.internal_ip == head_ip: continue node.state = "terminated" assert len(self.provider.non_terminated_nodes(WORKER_FILTER)) == 0 autoscaler.update() self.waitForNodes(2, tag_filters=WORKER_FILTER) def testConfiguresNewNodes(self): config = copy.deepcopy(SMALL_CLUSTER) config["available_node_types"]["worker"]["min_workers"] = 1 config_path = self.write_config(config) self.provider = MockProvider() runner = MockProcessRunner() runner.respond_to_call("json .Config.Env", ["[]" for i in range(2)]) self.provider.create_node( {}, { TAG_RAY_NODE_KIND: NODE_KIND_HEAD, TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, TAG_RAY_USER_NODE_TYPE: "head", }, 1, ) autoscaler = MockAutoscaler( config_path, LoadMetrics(), MockGcsClient(), max_failures=0, process_runner=runner, update_interval_s=0, ) autoscaler.update() # TODO(rueian): This is a hack to avoid running into race conditions # within v1 autoscaler. These should no longer be relevant in v2. self.waitForNodes(2) autoscaler.update() self.waitForNodes(2) self.provider.finish_starting_nodes() autoscaler.update() self.waitForNodes(2, tag_filters={TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE}) def testReportsConfigFailures(self): config = copy.deepcopy(SMALL_CLUSTER) config["available_node_types"]["worker"]["min_workers"] = 2 config_path = self.write_config(config) config["provider"]["type"] = "mock" config_path = self.write_config(config) self.provider = MockProvider() runner = MockProcessRunner(fail_cmds=["setup_cmd"]) runner.respond_to_call("json .Config.Env", ["[]" for i in range(2)]) lm = LoadMetrics() self.provider.create_node( {}, { TAG_RAY_NODE_KIND: NODE_KIND_HEAD, TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, TAG_RAY_USER_NODE_TYPE: "head", }, 1, ) autoscaler = MockAutoscaler( config_path, lm, MockGcsClient(), max_failures=0, process_runner=runner, update_interval_s=0, ) autoscaler.update() self.waitForNodes(2, tag_filters=WORKER_FILTER) self.provider.finish_starting_nodes() fill_in_node_ids(self.provider, lm) autoscaler.update() try: self.waitForNodes( 2, tag_filters={ TAG_RAY_NODE_STATUS: STATUS_UPDATE_FAILED, **WORKER_FILTER, }, ) except AssertionError: # The failed nodes might have been already terminated by autoscaler assert len(self.provider.non_terminated_nodes({})) < 2 # Check the launch failure event is generated. autoscaler.update() events = autoscaler.event_summarizer.summary() assert "Removing 2 nodes of type worker (launch failed)." in events, events def testConfiguresOutdatedNodes(self): from ray.autoscaler._private.cli_logger import cli_logger def do_nothing(*args, **kwargs): pass cli_logger._print = type(cli_logger._print)(do_nothing, type(cli_logger)) config = copy.deepcopy(SMALL_CLUSTER) config["available_node_types"]["worker"]["min_workers"] = 1 config_path = self.write_config(config) self.provider = MockProvider() runner = MockProcessRunner() runner.respond_to_call("json .Config.Env", ["[]" for i in range(4)]) self.provider.create_node( {}, { TAG_RAY_NODE_KIND: NODE_KIND_HEAD, TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, TAG_RAY_USER_NODE_TYPE: "head", }, 1, ) autoscaler = MockAutoscaler( config_path, LoadMetrics(), MockGcsClient(), max_failures=0, process_runner=runner, update_interval_s=0, ) autoscaler.update() # TODO(rickyx): This is a hack to avoid running into race conditions # within v1 autoscaler. These should no longer be relevant in v2. time.sleep(3) autoscaler.update() self.waitForNodes(2) self.provider.finish_starting_nodes() time.sleep(3) autoscaler.update() time.sleep(3) self.waitForNodes(2, tag_filters={TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE}) runner.calls = [] new_config = copy.deepcopy(SMALL_CLUSTER) new_config["worker_setup_commands"] = ["cmdX", "cmdY"] self.write_config(new_config) autoscaler.update() autoscaler.update() self.waitFor(lambda: len(runner.calls) > 0) def testScaleDownMaxWorkers(self): """Tests terminating nodes due to max_nodes per type.""" config = copy.deepcopy(MULTI_WORKER_CLUSTER) config["available_node_types"]["m4.large"]["min_workers"] = 3 config["available_node_types"]["m4.large"]["max_workers"] = 3 config["available_node_types"]["m4.large"]["resources"] = {} config["available_node_types"]["m4.16xlarge"]["resources"] = {} config["available_node_types"]["p2.xlarge"]["min_workers"] = 5 config["available_node_types"]["p2.xlarge"]["max_workers"] = 8 config["available_node_types"]["p2.xlarge"]["resources"] = {} config["available_node_types"]["p2.8xlarge"]["min_workers"] = 2 config["available_node_types"]["p2.8xlarge"]["max_workers"] = 4 config["available_node_types"]["p2.8xlarge"]["resources"] = {} config["max_workers"] = 13 config_path = self.write_config(config) self.provider = MockProvider() runner = MockProcessRunner() runner.respond_to_call("json .Config.Env", ["[]" for i in range(15)]) lm = LoadMetrics() get_or_create_head_node( config, printable_config_file=config_path, no_restart=False, restart_only=False, yes=True, override_cluster_name=None, _provider=self.provider, _runner=runner, ) self.waitForNodes(1) autoscaler = MockAutoscaler( config_path, lm, MockGcsClient(), max_failures=0, max_concurrent_launches=13, max_launch_batch=13, process_runner=runner, update_interval_s=0, ) autoscaler.update() self.waitForNodes(11) assert autoscaler.pending_launches.value == 0 assert ( len( self.provider.non_terminated_nodes( {TAG_RAY_NODE_KIND: NODE_KIND_WORKER} ) ) == 10 ) # Terminate some nodes config["available_node_types"]["m4.large"]["min_workers"] = 2 # 3 config["available_node_types"]["m4.large"]["max_workers"] = 2 config["available_node_types"]["p2.8xlarge"]["min_workers"] = 0 # 2 config["available_node_types"]["p2.8xlarge"]["max_workers"] = 0 # And spawn one. config["available_node_types"]["p2.xlarge"]["min_workers"] = 6 # 5 config["available_node_types"]["p2.xlarge"]["max_workers"] = 6 self.write_config(config) fill_in_node_ids(self.provider, lm) autoscaler.update() events = autoscaler.event_summarizer.summary() self.waitFor(lambda: autoscaler.pending_launches.value == 0) self.waitForNodes(8, tag_filters={TAG_RAY_NODE_KIND: NODE_KIND_WORKER}) assert autoscaler.pending_launches.value == 0 events = autoscaler.event_summarizer.summary() assert "Removing 1 nodes of type m4.large (max_workers_per_type)." in events assert "Removing 2 nodes of type p2.8xlarge (max_workers_per_type)." in events # We should not be starting/stopping empty_node at all. for event in events: assert "empty_node" not in event node_type_counts = defaultdict(int) for node_id in NonTerminatedNodes(self.provider).worker_ids: tags = self.provider.node_tags(node_id) if TAG_RAY_USER_NODE_TYPE in tags: node_type = tags[TAG_RAY_USER_NODE_TYPE] node_type_counts[node_type] += 1 assert node_type_counts == {"m4.large": 2, "p2.xlarge": 6} def testFalseyLoadMetrics(self): lm = LoadMetrics() assert not lm lm.update("172.0.0.0", mock_node_id(), {"CPU": 1}, {"CPU": 0}, 0) assert lm def testRecoverUnhealthyWorkers(self): config = copy.deepcopy(SMALL_CLUSTER) config["available_node_types"]["worker"]["min_workers"] = 2 config_path = self.write_config(config) self.provider = MockProvider() runner = MockProcessRunner() runner.respond_to_call("json .Config.Env", ["[]" for i in range(3)]) lm = LoadMetrics() mock_metrics = Mock(spec=AutoscalerPrometheusMetrics()) self.provider.create_node( {}, { TAG_RAY_NODE_KIND: NODE_KIND_HEAD, TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, TAG_RAY_USER_NODE_TYPE: "head", }, 1, ) autoscaler = MockAutoscaler( config_path, lm, MockGcsClient(), max_failures=0, process_runner=runner, update_interval_s=0, prom_metrics=mock_metrics, ) autoscaler.update() self.waitForNodes(2, tag_filters=WORKER_FILTER) self.provider.finish_starting_nodes() autoscaler.update() self.waitForNodes( 2, tag_filters={TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, **WORKER_FILTER} ) # Mark a node as unhealthy for _ in range(5): if autoscaler.updaters: time.sleep(0.05) autoscaler.update() assert not autoscaler.updaters mock_metrics.recovering_nodes.set.assert_called_with(0) num_calls = len(runner.calls) lm.last_heartbeat_time_by_ip["172.0.0.1"] = 0 autoscaler.update() mock_metrics.recovering_nodes.set.assert_called_with(1) self.waitFor(lambda: len(runner.calls) > num_calls, num_retries=150) # Check the node removal event is generated. autoscaler.update() events = autoscaler.event_summarizer.summary() assert ( "Restarting 1 nodes of type worker (lost contact with raylet)." in events ), events assert mock_metrics.drain_node_exceptions.inc.call_count == 0 def testTerminateUnhealthyWorkers(self): """Test termination of unhealthy workers, when autoscaler.disable_node_updaters == True. Similar to testRecoverUnhealthyWorkers. """ self.unhealthyWorkerHelper(disable_liveness_check=False) def testDontTerminateUnhealthyWorkers(self): """Test that the autoscaler leaves unhealthy workers alone when the worker liveness check is disabled. """ self.unhealthyWorkerHelper(disable_liveness_check=True) def unhealthyWorkerHelper(self, disable_liveness_check: bool): """Helper used to test the autoscaler's handling of unhealthy worker nodes. If disable liveness check is False, the default code path is tested and we expect to see workers terminated. If disable liveness check is True, we expect the autoscaler not to take action on unhealthy nodes, instead delegating node management to another component. """ config = copy.deepcopy(SMALL_CLUSTER) config["available_node_types"]["worker"]["min_workers"] = 2 # Make it clear we're not timing out idle nodes here. config["idle_timeout_minutes"] = 1000000000 if disable_liveness_check: config["provider"][WORKER_LIVENESS_CHECK_KEY] = False config_path = self.write_config(config) self.provider = MockProvider() runner = MockProcessRunner() runner.respond_to_call("json .Config.Env", ["[]" for i in range(3)]) lm = LoadMetrics() mock_metrics = Mock(spec=AutoscalerPrometheusMetrics()) self.provider.create_node( {}, { TAG_RAY_NODE_KIND: NODE_KIND_HEAD, TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, TAG_RAY_USER_NODE_TYPE: "head", }, 1, ) autoscaler = MockAutoscaler( config_path, lm, MockGcsClient(), max_failures=0, process_runner=runner, update_interval_s=0, prom_metrics=mock_metrics, ) autoscaler.update() self.waitForNodes(2, tag_filters=WORKER_FILTER) self.provider.finish_starting_nodes() autoscaler.update() self.waitForNodes( 2, tag_filters={TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, **WORKER_FILTER} ) # Clear out updaters. for _ in range(5): if autoscaler.updaters: time.sleep(0.05) autoscaler.update() assert not autoscaler.updaters num_calls = len(runner.calls) # Mark a node as unhealthy lm.last_heartbeat_time_by_ip["172.0.0.1"] = 0 # Turn off updaters. autoscaler.disable_node_updaters = True # Reduce min_workers to 1 autoscaler.config["available_node_types"]["worker"]["min_workers"] = 1 fill_in_node_ids(self.provider, lm) if disable_liveness_check: # We've disabled the liveness check, so the unhealthy node should stick # around until someone else takes care of it. # Do several autoscaler updates, to reinforce the fact that the # autoscaler will never take down the unhealthy nodes. for _ in range(10): autoscaler.update() # The nodes are still there. assert self.num_nodes(tag_filters=WORKER_FILTER) == 2 # There's no synchronization required to make the last assertion valid: # The autoscaler's node termination is synchronous and blocking, as is # the terminate_node method of the mock node provider used in this test. # No events generated indicating that we are removing nodes. for event in autoscaler.event_summarizer.summary(): assert "Removing" not in event else: # We expect the unhealthy node to be cleared out with a single # autoscaler update. autoscaler.update() # Stopped node metric incremented. mock_metrics.stopped_nodes.inc.assert_called_once_with() # One node left. self.waitForNodes(1, tag_filters=WORKER_FILTER) # Check the node removal event is generated. autoscaler.update() events = autoscaler.event_summarizer.summary() assert ( "Removing 1 nodes of type " "worker (lost contact with raylet)." in events ), events # No additional runner calls, since updaters were disabled. assert len(runner.calls) == num_calls assert mock_metrics.drain_node_exceptions.inc.call_count == 0 def testTerminateUnhealthyWorkers2(self): """Tests finer details of termination of unhealthy workers when node updaters are disabled. Specifically, test that newly up-to-date nodes which haven't sent a heartbeat are marked active. """ config = copy.deepcopy(SMALL_CLUSTER) config["provider"]["disable_node_updaters"] = True config["available_node_types"]["worker"]["min_workers"] = 2 config_path = self.write_config(config) self.provider = MockProvider() runner = MockProcessRunner() mock_metrics = Mock(spec=AutoscalerPrometheusMetrics()) lm = LoadMetrics() self.provider.create_node( {}, { TAG_RAY_NODE_KIND: NODE_KIND_HEAD, TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, TAG_RAY_USER_NODE_TYPE: "head", }, 1, ) autoscaler = MockAutoscaler( config_path, lm, MockGcsClient(), max_failures=0, process_runner=runner, update_interval_s=0, prom_metrics=mock_metrics, ) assert len(self.provider.non_terminated_nodes(WORKER_FILTER)) == 0 for _ in range(10): autoscaler.update() # Nodes stay in uninitialized state because no one has finished # updating them. self.waitForNodes( 2, tag_filters={ TAG_RAY_NODE_STATUS: STATUS_UNINITIALIZED, **WORKER_FILTER, }, ) nodes = self.provider.non_terminated_nodes(WORKER_FILTER) ips = [self.provider.internal_ip(node) for node in nodes] # No heartbeats recorded yet. assert not any(ip in lm.last_heartbeat_time_by_ip for ip in ips) for node in nodes: self.provider.set_node_tags( node, {TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, **WORKER_FILTER} ) autoscaler.update() # Nodes marked active after up-to-date status detected. assert all(ip in lm.last_heartbeat_time_by_ip for ip in ips) # Nodes are kept. self.waitForNodes( 2, tag_filters={TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, **WORKER_FILTER} ) # Mark nodes unhealthy. for ip in ips: lm.last_heartbeat_time_by_ip[ip] = 0 fill_in_node_ids(self.provider, lm) autoscaler.update() # Unhealthy nodes are gone. self.waitForNodes(0, tag_filters=WORKER_FILTER) autoscaler.update() # IPs pruned assert lm.last_heartbeat_time_by_ip == {} assert mock_metrics.drain_node_exceptions.inc.call_count == 0 def testExternalNodeScaler(self): config = copy.deepcopy(SMALL_CLUSTER) config["provider"] = { "type": "external", "module": "ray.autoscaler.node_provider.NodeProvider", } config_path = self.write_config(config) autoscaler = MockAutoscaler( config_path, LoadMetrics(), MockGcsClient(), max_failures=0, update_interval_s=0, ) assert isinstance(autoscaler.provider, NodeProvider) def testExternalNodeScalerWrongImport(self): config = copy.deepcopy(SMALL_CLUSTER) config["provider"] = { "type": "external", "module": "mymodule.provider_class", } invalid_provider = self.write_config(config) with pytest.raises(ImportError): MockAutoscaler( invalid_provider, LoadMetrics(), MockGcsClient(), update_interval_s=0 ) def testExternalNodeScalerWrongModuleFormat(self): config = copy.deepcopy(SMALL_CLUSTER) config["provider"] = { "type": "external", "module": "does-not-exist", } invalid_provider = self.write_config(config, call_prepare_config=False) with pytest.raises(ValueError): MockAutoscaler( invalid_provider, LoadMetrics(), MockGcsClient(), update_interval_s=0 ) def testSetupCommandsWithNoNodeCaching(self): config = copy.deepcopy(SMALL_CLUSTER) config["available_node_types"]["worker"]["min_workers"] = 1 config["available_node_types"]["worker"]["max_workers"] = 1 config["max_workers"] = 1 config_path = self.write_config(config) self.provider = MockProvider(cache_stopped=False) runner = MockProcessRunner() runner.respond_to_call("json .Config.Env", ["[]" for i in range(2)]) lm = LoadMetrics() self.provider.create_node( {}, { TAG_RAY_NODE_KIND: NODE_KIND_HEAD, TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, TAG_RAY_USER_NODE_TYPE: "head", }, 1, ) autoscaler = MockAutoscaler( config_path, lm, MockGcsClient(), max_failures=0, process_runner=runner, update_interval_s=0, ) autoscaler.update() self.waitForNodes(1, tag_filters=WORKER_FILTER) self.provider.finish_starting_nodes() autoscaler.update() self.waitForNodes( 1, tag_filters={TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, **WORKER_FILTER} ) worker_ip = self.provider.non_terminated_node_ips(WORKER_FILTER)[0] runner.assert_has_call(worker_ip, "init_cmd") runner.assert_has_call(worker_ip, "setup_cmd") runner.assert_has_call(worker_ip, "worker_setup_cmd") runner.assert_has_call(worker_ip, "start_ray_worker") # Check the node was not reused self.provider.terminate_node("1") autoscaler.update() runner.clear_history() self.waitForNodes(1, tag_filters=WORKER_FILTER) self.provider.finish_starting_nodes() autoscaler.update() self.waitForNodes( 1, tag_filters={TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, **WORKER_FILTER} ) new_worker_ip = self.provider.non_terminated_node_ips(WORKER_FILTER)[0] runner.assert_has_call(new_worker_ip, "init_cmd") runner.assert_has_call(new_worker_ip, "setup_cmd") runner.assert_has_call(new_worker_ip, "worker_setup_cmd") runner.assert_has_call(new_worker_ip, "start_ray_worker") assert worker_ip != new_worker_ip def testSetupCommandsWithStoppedNodeCachingNoDocker(self): file_mount_dir = tempfile.mkdtemp() config = copy.deepcopy(SMALL_CLUSTER) del config["docker"] config["file_mounts"] = {"/root/test-folder": file_mount_dir} config["file_mounts_sync_continuously"] = True config["available_node_types"]["worker"]["min_workers"] = 1 config["available_node_types"]["worker"]["max_workers"] = 1 config["max_workers"] = 1 config_path = self.write_config(config) self.provider = MockProvider(cache_stopped=True) runner = MockProcessRunner() runner.respond_to_call("json .Config.Env", ["[]" for i in range(3)]) lm = LoadMetrics() self.provider.create_node( {}, { TAG_RAY_NODE_KIND: NODE_KIND_HEAD, TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, TAG_RAY_USER_NODE_TYPE: "head", }, 1, ) autoscaler = MockAutoscaler( config_path, lm, MockGcsClient(), max_failures=0, process_runner=runner, update_interval_s=0, ) autoscaler.update() self.waitForNodes(1, tag_filters=WORKER_FILTER) self.provider.finish_starting_nodes() autoscaler.update() self.waitForNodes( 1, tag_filters={TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, **WORKER_FILTER} ) worker_ip = self.provider.non_terminated_node_ips(WORKER_FILTER)[0] runner.assert_has_call(worker_ip, "init_cmd") runner.assert_has_call(worker_ip, "setup_cmd") runner.assert_has_call(worker_ip, "worker_setup_cmd") runner.assert_has_call(worker_ip, "start_ray_worker") # Check the node was indeed reused self.provider.terminate_node("1") runner.clear_history() autoscaler.update() self.waitForNodes(1, tag_filters=WORKER_FILTER) self.provider.finish_starting_nodes() autoscaler.update() self.waitForNodes( 1, tag_filters={TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, **WORKER_FILTER} ) runner.assert_not_has_call(worker_ip, "init_cmd") runner.assert_not_has_call(worker_ip, "setup_cmd") runner.assert_not_has_call(worker_ip, "worker_setup_cmd") runner.assert_has_call(worker_ip, "start_ray_worker") with open(f"{file_mount_dir}/new_file", "w") as f: f.write("abcdefgh") # Check that run_init happens when file_mounts have updated self.provider.terminate_node("1") autoscaler.update() runner.clear_history() self.waitForNodes(1, tag_filters=WORKER_FILTER) self.provider.finish_starting_nodes() autoscaler.update() self.waitForNodes( 1, tag_filters={TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, **WORKER_FILTER} ) runner.assert_not_has_call(worker_ip, "init_cmd") runner.assert_not_has_call(worker_ip, "setup_cmd") runner.assert_not_has_call(worker_ip, "worker_setup_cmd") runner.assert_has_call(worker_ip, "start_ray_worker") autoscaler.update() runner.assert_not_has_call(worker_ip, "setup_cmd") # We did not start any other nodes self.waitForNodes( 1, tag_filters={TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, **WORKER_FILTER} ) def testSetupCommandsWithStoppedNodeCachingDocker(self): # NOTE(ilr) Setup & Init commands **should** run with stopped nodes # when Docker is in use. file_mount_dir = tempfile.mkdtemp() config = copy.deepcopy(SMALL_CLUSTER) config["file_mounts"] = {"/root/test-folder": file_mount_dir} config["file_mounts_sync_continuously"] = True config["available_node_types"]["worker"]["min_workers"] = 1 config["available_node_types"]["worker"]["max_workers"] = 1 config["max_workers"] = 1 config_path = self.write_config(config) self.provider = MockProvider(cache_stopped=True) runner = MockProcessRunner() runner.respond_to_call("json .Config.Env", ["[]" for i in range(3)]) lm = LoadMetrics() self.provider.create_node( {}, { TAG_RAY_NODE_KIND: NODE_KIND_HEAD, TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, TAG_RAY_USER_NODE_TYPE: "head", }, 1, ) autoscaler = MockAutoscaler( config_path, lm, MockGcsClient(), max_failures=0, process_runner=runner, update_interval_s=0, ) autoscaler.update() self.waitForNodes(1, tag_filters=WORKER_FILTER) self.provider.finish_starting_nodes() autoscaler.update() self.waitForNodes( 1, tag_filters={TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, **WORKER_FILTER} ) worker_ip = self.provider.non_terminated_node_ips(WORKER_FILTER)[0] runner.assert_has_call(worker_ip, "init_cmd") runner.assert_has_call(worker_ip, "setup_cmd") runner.assert_has_call(worker_ip, "worker_setup_cmd") runner.assert_has_call(worker_ip, "start_ray_worker") runner.assert_has_call(worker_ip, "docker run") # Check the node was indeed reused self.provider.terminate_node("1") runner.clear_history() autoscaler.update() self.waitForNodes(1, tag_filters=WORKER_FILTER) self.provider.finish_starting_nodes() autoscaler.update() self.waitForNodes( 1, tag_filters={TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, **WORKER_FILTER} ) print(runner.command_history()) # These all must happen when the node is stopped and resued runner.assert_has_call(worker_ip, "init_cmd") runner.assert_has_call(worker_ip, "setup_cmd") runner.assert_has_call(worker_ip, "worker_setup_cmd") runner.assert_has_call(worker_ip, "start_ray_worker") runner.assert_has_call(worker_ip, "docker run") with open(f"{file_mount_dir}/new_file", "w") as f: f.write("abcdefgh") # Check that run_init happens when file_mounts have updated self.provider.terminate_node("0") runner.clear_history() autoscaler.update() self.waitForNodes(1, tag_filters=WORKER_FILTER) self.provider.finish_starting_nodes() autoscaler.update() self.waitForNodes( 1, tag_filters={TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, **WORKER_FILTER} ) runner.assert_has_call(worker_ip, "init_cmd") runner.assert_has_call(worker_ip, "setup_cmd") runner.assert_has_call(worker_ip, "worker_setup_cmd") runner.assert_has_call(worker_ip, "start_ray_worker") runner.assert_has_call(worker_ip, "docker run") docker_run_cmd_indx = [ i for i, cmd in enumerate(runner.command_history()) if "docker run" in cmd ][0] mkdir_cmd_indx = [ i for i, cmd in enumerate(runner.command_history()) if "mkdir -p" in cmd ][0] assert mkdir_cmd_indx < docker_run_cmd_indx runner.clear_history() autoscaler.update() runner.assert_not_has_call(worker_ip, "setup_cmd") # We did not start any other nodes runner.assert_not_has_call("172.0.0.2", " ") def testMultiNodeReuse(self): config = copy.deepcopy(SMALL_CLUSTER) # Docker re-runs setup commands when nodes are reused. del config["docker"] config["available_node_types"]["worker"]["min_workers"] = 3 config["available_node_types"]["worker"]["max_workers"] = 3 config["max_workers"] = 3 config_path = self.write_config(config) self.provider = MockProvider(cache_stopped=True) runner = MockProcessRunner() lm = LoadMetrics() self.provider.create_node( {}, { TAG_RAY_NODE_KIND: NODE_KIND_HEAD, TAG_RAY_USER_NODE_TYPE: "head", }, 1, ) autoscaler = MockAutoscaler( config_path, lm, MockGcsClient(), max_failures=0, process_runner=runner, update_interval_s=0, ) autoscaler.update() self.waitForNodes(3, tag_filters=WORKER_FILTER) self.provider.finish_starting_nodes() autoscaler.update() self.waitForNodes( 3, tag_filters={TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, **WORKER_FILTER} ) self.provider.terminate_node("1") self.provider.terminate_node("2") self.provider.terminate_node("3") runner.clear_history() # Scale up to 10 nodes, check we reuse the first 3 and add 5 more. config["available_node_types"]["worker"]["min_workers"] = 8 config["available_node_types"]["worker"]["max_workers"] = 8 config["max_workers"] = 8 self.write_config(config) autoscaler.update() autoscaler.update() self.waitForNodes(8, tag_filters=WORKER_FILTER) self.provider.finish_starting_nodes() autoscaler.update() self.waitForNodes( 8, tag_filters={TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, **WORKER_FILTER} ) autoscaler.update() for i in [1, 2, 3]: runner.assert_not_has_call("172.0.0.{}".format(i), "setup_cmd") runner.assert_has_call("172.0.0.{}".format(i), "start_ray_worker") for i in range(4, 9): runner.assert_has_call("172.0.0.{}".format(i), "setup_cmd") runner.assert_has_call("172.0.0.{}".format(i), "start_ray_worker") def testContinuousFileMounts(self): file_mount_dir = tempfile.mkdtemp() self.provider = MockProvider() config = copy.deepcopy(SMALL_CLUSTER) config["file_mounts"] = {"/home/test-folder": file_mount_dir} config["file_mounts_sync_continuously"] = True config["available_node_types"]["worker"]["min_workers"] = 2 config_path = self.write_config(config) runner = MockProcessRunner() runner.respond_to_call("json .Config.Env", ["[]" for i in range(4)]) runner.respond_to_call("command -v docker", ["docker" for _ in range(4)]) lm = LoadMetrics() self.provider.create_node( {}, { TAG_RAY_NODE_KIND: NODE_KIND_HEAD, TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, TAG_RAY_USER_NODE_TYPE: "head", }, 1, ) autoscaler = MockAutoscaler( config_path, lm, MockGcsClient(), max_failures=0, process_runner=runner, update_interval_s=0, ) autoscaler.update() self.waitForNodes(3) self.provider.finish_starting_nodes() autoscaler.update() self.waitForNodes(3, tag_filters={TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE}) autoscaler.update() docker_mount_prefix = get_docker_host_mount_location(config["cluster_name"]) for i in self.provider.non_terminated_nodes( tag_filters={TAG_RAY_NODE_KIND: NODE_KIND_WORKER} ): runner.assert_has_call(f"172.0.0.{i}", "setup_cmd") runner.assert_has_call( f"172.0.0.{i}", f"{file_mount_dir}/ ubuntu@172.0.0.{i}:" f"{docker_mount_prefix}/home/test-folder/", ) runner.clear_history() with open(os.path.join(file_mount_dir, "test.txt"), "wb") as temp_file: temp_file.write("hello".encode()) runner.respond_to_call(".Config.Image", ["example" for _ in range(4)]) runner.respond_to_call(".State.Running", ["true" for _ in range(4)]) autoscaler.update() self.waitForNodes(3) self.provider.finish_starting_nodes() autoscaler.update() self.waitForNodes(3, tag_filters={TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE}) autoscaler.update() for i in self.provider.non_terminated_nodes( tag_filters={TAG_RAY_NODE_KIND: NODE_KIND_WORKER} ): runner.assert_not_has_call(f"172.0.0.{i}", "setup_cmd") runner.assert_has_call( f"172.0.0.{i}", f"{file_mount_dir}/ ubuntu@172.0.0.{i}:" f"{docker_mount_prefix}/home/test-folder/", ) def testFileMountsNonContinuous(self): file_mount_dir = tempfile.mkdtemp() self.provider = MockProvider() config = copy.deepcopy(SMALL_CLUSTER) config["file_mounts"] = {"/home/test-folder": file_mount_dir} config["available_node_types"]["worker"]["min_workers"] = 2 config_path = self.write_config(config) runner = MockProcessRunner() runner.respond_to_call("json .Config.Env", ["[]" for i in range(2)]) lm = LoadMetrics() self.provider.create_node( {}, { TAG_RAY_NODE_KIND: NODE_KIND_HEAD, TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, TAG_RAY_USER_NODE_TYPE: "head", }, 1, ) autoscaler = MockAutoscaler( config_path, lm, MockGcsClient(), max_failures=0, process_runner=runner, update_interval_s=0, ) autoscaler.update() self.waitForNodes(2, tag_filters=WORKER_FILTER) self.provider.finish_starting_nodes() autoscaler.update() self.waitForNodes( 2, tag_filters={TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, **WORKER_FILTER} ) autoscaler.update() docker_mount_prefix = get_docker_host_mount_location(config["cluster_name"]) for i in self.provider.non_terminated_nodes(WORKER_FILTER): runner.assert_has_call(f"172.0.0.{i}", "setup_cmd") runner.assert_has_call( f"172.0.0.{i}", f"{file_mount_dir}/ ubuntu@172.0.0.{i}:" f"{docker_mount_prefix}/home/test-folder/", ) runner.clear_history() with open(os.path.join(file_mount_dir, "test.txt"), "wb") as temp_file: temp_file.write("hello".encode()) autoscaler.update() self.waitForNodes(2, tag_filters=WORKER_FILTER) self.provider.finish_starting_nodes() self.waitForNodes( 2, tag_filters={TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, **WORKER_FILTER} ) for i in self.provider.non_terminated_nodes(WORKER_FILTER): runner.assert_not_has_call(f"172.0.0.{i}", "setup_cmd") runner.assert_not_has_call( f"172.0.0.{i}", f"{file_mount_dir}/ ubuntu@172.0.0.{i}:" f"{docker_mount_prefix}/home/test-folder/", ) # Simulate a second `ray up` call from ray.autoscaler._private import util util._hash_cache = {} runner = MockProcessRunner() runner.respond_to_call("json .Config.Env", ["[]" for i in range(2)]) lm = LoadMetrics() autoscaler = MockAutoscaler( config_path, lm, MockGcsClient(), max_failures=0, process_runner=runner, update_interval_s=0, ) autoscaler.update() self.waitForNodes(2, tag_filters=WORKER_FILTER) self.provider.finish_starting_nodes() self.waitForNodes( 2, tag_filters={TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, **WORKER_FILTER} ) autoscaler.update() for i in self.provider.non_terminated_nodes(WORKER_FILTER): runner.assert_has_call(f"172.0.0.{i}", "setup_cmd") runner.assert_has_call( f"172.0.0.{i}", f"{file_mount_dir}/ ubuntu@172.0.0.{i}:" f"{docker_mount_prefix}/home/test-folder/", ) def testDockerImageExistsBeforeInspect(self): config = copy.deepcopy(SMALL_CLUSTER) config["available_node_types"]["worker"]["min_workers"] = 1 config["available_node_types"]["worker"]["max_workers"] = 1 config["max_workers"] = 1 config["docker"]["pull_before_run"] = False config_path = self.write_config(config) self.provider = MockProvider() runner = MockProcessRunner() runner.respond_to_call("json .Config.Env", ["[]" for i in range(1)]) self.provider.create_node( {}, { TAG_RAY_NODE_KIND: NODE_KIND_HEAD, TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, TAG_RAY_USER_NODE_TYPE: "head", }, 1, ) autoscaler = MockAutoscaler( config_path, LoadMetrics(), MockGcsClient(), max_failures=0, process_runner=runner, update_interval_s=0, ) autoscaler.update() autoscaler.update() self.waitForNodes(1, tag_filters={TAG_RAY_NODE_KIND: NODE_KIND_WORKER}) self.provider.finish_starting_nodes() autoscaler.update() self.waitForNodes( 1, tag_filters={ TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, TAG_RAY_NODE_KIND: NODE_KIND_WORKER, }, ) first_pull = [ (i, cmd) for i, cmd in enumerate(runner.command_history()) if "docker pull" in cmd ] first_targeted_inspect = [ (i, cmd) for i, cmd in enumerate(runner.command_history()) if "docker inspect -f" in cmd ] # This checks for the bug mentioned #13128 where the image is inspected # before the image is present. assert min(x[0] for x in first_pull) < min(x[0] for x in first_targeted_inspect) def testGetRunningHeadNode(self): config = copy.deepcopy(SMALL_CLUSTER) self.provider = MockProvider() # Node 0 is failed. self.provider.create_node( {}, { TAG_RAY_CLUSTER_NAME: "default", TAG_RAY_NODE_KIND: "head", TAG_RAY_NODE_STATUS: "update-failed", }, 1, ) # `_allow_uninitialized_state` should return the head node # in the `update-failed` state. allow_failed = commands._get_running_head_node( config, "/fake/path", override_cluster_name=None, create_if_needed=False, _provider=self.provider, _allow_uninitialized_state=True, ) assert allow_failed == "0" # Node 1 is okay. self.provider.create_node( {}, { TAG_RAY_CLUSTER_NAME: "default", TAG_RAY_NODE_KIND: "head", TAG_RAY_NODE_STATUS: "up-to-date", }, 1, ) node = commands._get_running_head_node( config, "/fake/path", override_cluster_name=None, create_if_needed=False, _provider=self.provider, ) assert node == "1" # `_allow_uninitialized_state` should return the up-to-date head node # if it is present. optionally_failed = commands._get_running_head_node( config, "/fake/path", override_cluster_name=None, create_if_needed=False, _provider=self.provider, _allow_uninitialized_state=True, ) assert optionally_failed == "1" def testNodeTerminatedDuringUpdate(self): """ Tests autoscaler handling a node getting terminated during an update triggered by the node missing a heartbeat. Extension of testRecoverUnhealthyWorkers. In this test, two nodes miss a heartbeat. One of them (node 0) is terminated during its recovery update. The other (node 1) just fails its update. When processing completed updates, the autoscaler terminates node 1 but does not try to terminate node 0 again. """ cluster_config = copy.deepcopy(MOCK_DEFAULT_CONFIG) cluster_config["available_node_types"]["ray.worker.default"]["min_workers"] = 2 cluster_config["worker_start_ray_commands"] = ["ray_start_cmd"] # Don't need the extra node type or a docker config. cluster_config["head_node_type"] = ["ray.worker.default"] del cluster_config["available_node_types"]["ray.head.default"] del cluster_config["docker"] config_path = self.write_config(cluster_config) self.provider = MockProvider() runner = MockProcessRunner() lm = LoadMetrics() mock_metrics = Mock(spec=AutoscalerPrometheusMetrics()) autoscaler = MockAutoscaler( config_path, lm, MockGcsClient(), max_failures=0, process_runner=runner, update_interval_s=0, prom_metrics=mock_metrics, ) # Scale up to two up-to-date workers autoscaler.update() self.waitForNodes(2) self.provider.finish_starting_nodes() autoscaler.update() self.waitForNodes(2, tag_filters={TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE}) # Mark both nodes as unhealthy for _ in range(5): if autoscaler.updaters: time.sleep(0.05) autoscaler.update() lm.last_heartbeat_time_by_ip["172.0.0.0"] = 0 lm.last_heartbeat_time_by_ip["172.0.0.1"] = 0 # Expect both updates to be successful, no nodes in updating state assert mock_metrics.successful_updates.inc.call_count == 2 assert mock_metrics.worker_update_time.observe.call_count == 2 mock_metrics.updating_nodes.set.assert_called_with(0) assert not autoscaler.updaters # Set up process runner to terminate worker 0 during missed heartbeat # recovery and also cause the updater to fail. def terminate_worker_zero(): self.provider.terminate_node("0") autoscaler.process_runner = MockProcessRunner( fail_cmds=["ray_start_cmd"], cmd_to_callback={"ray_start_cmd": terminate_worker_zero}, ) # ensures that no updates are completed until after the next call # to update() autoscaler.process_runner.ready_to_run.clear() num_calls = len(autoscaler.process_runner.calls) autoscaler.update() mock_metrics.updating_nodes.set.assert_called_with(2) mock_metrics.recovering_nodes.set.assert_called_with(2) autoscaler.process_runner.ready_to_run.set() # Wait for updaters spawned by last autoscaler update to finish. self.waitForUpdatersToFinish(autoscaler) # Check that updaters processed some commands in the last autoscaler # update. assert ( len(autoscaler.process_runner.calls) > num_calls ), "Did not get additional process runner calls on last autoscaler update." # Missed heartbeat triggered recovery for both nodes. events = autoscaler.event_summarizer.summary() assert ( "Restarting 2 nodes of type " "ray.worker.default (lost contact with raylet)." in events ), events # Node 0 was terminated during the last update. # Node 1's updater failed, but node 1 won't be terminated until the # next autoscaler update. assert ( "0" not in NonTerminatedNodes(self.provider).worker_ids ), "Node zero still non-terminated." assert not self.provider.is_terminated("1"), "Node one terminated prematurely." fill_in_node_ids(self.provider, lm) autoscaler.update() # Failed updates processed are now processed. assert ( autoscaler.num_failed_updates["0"] == 1 ), "Node zero update failure not registered" assert ( autoscaler.num_failed_updates["1"] == 1 ), "Node one update failure not registered" assert mock_metrics.failed_updates.inc.call_count == 2 assert mock_metrics.failed_recoveries.inc.call_count == 2 assert mock_metrics.successful_recoveries.inc.call_count == 0 # Completed-update-processing logic should have terminated node 1. assert self.provider.is_terminated("1"), "Node 1 not terminated on time." events = autoscaler.event_summarizer.summary() # Just one node (node_id 1) terminated in the last update. # Validates that we didn't try to double-terminate node 0. assert ( "Removing 1 nodes of type ray.worker.default (launch failed)." in events ), events # To be more explicit, assert ( "Removing 2 nodes of type " "ray.worker.default (launch failed)." not in events ), events # Should get two new nodes after the next update. fill_in_node_ids(self.provider, lm) autoscaler.update() self.waitForNodes(2) assert set(NonTerminatedNodes(self.provider).worker_ids) == { "2", "3", }, "Unexpected node_ids" assert mock_metrics.stopped_nodes.inc.call_count == 1 assert mock_metrics.drain_node_exceptions.inc.call_count == 0 def testProviderException(self): config = copy.deepcopy(SMALL_CLUSTER) config["available_node_types"]["worker"]["min_workers"] = 2 config_path = self.write_config(config) self.provider = MockProvider() runner = MockProcessRunner() mock_metrics = Mock(spec=AutoscalerPrometheusMetrics()) self.provider.create_node( {}, { TAG_RAY_NODE_KIND: NODE_KIND_HEAD, TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, TAG_RAY_USER_NODE_TYPE: "head", }, 1, ) self.provider.creation_error = Exception(":(") autoscaler = MockAutoscaler( config_path, LoadMetrics(), MockGcsClient(), max_failures=0, process_runner=runner, update_interval_s=0, prom_metrics=mock_metrics, ) autoscaler.update() def metrics_incremented(): exceptions = mock_metrics.node_launch_exceptions.inc.call_count == 1 create_failures = mock_metrics.failed_create_nodes.inc.call_count == 1 create_arg = False if create_failures: # number of failed creations should be incremented by 2 create_arg = mock_metrics.failed_create_nodes.inc.call_args[0] == (2,) return exceptions and create_failures and create_arg self.waitFor(metrics_incremented, fail_msg="Expected metrics to update") assert mock_metrics.drain_node_exceptions.inc.call_count == 0 def testDefaultMinMaxWorkers(self): config = copy.deepcopy(MOCK_DEFAULT_CONFIG) config = prepare_config(config) node_types = config["available_node_types"] head_node_config = node_types["ray.head.default"] assert head_node_config["min_workers"] == 0 assert head_node_config["max_workers"] == 0 def testAutoscalerInitFailure(self): """Validates error handling for failed autoscaler initialization in the Monitor. """ class AutoscalerInitFailException(Exception): pass class FaultyAutoscaler: def __init__(self, *args, **kwargs): raise AutoscalerInitFailException prev_port = os.environ.get("RAY_GCS_SERVER_PORT") os.environ["RAY_GCS_SERVER_PORT"] = "12345" ray.init() with patch("ray._private.utils.publish_error_to_driver") as mock_publish: with patch.multiple( "ray.autoscaler._private.monitor", StandardAutoscaler=FaultyAutoscaler, _internal_kv_initialized=Mock(return_value=False), ): monitor = Monitor( address="localhost:12345", autoscaling_config="", log_dir=self.tmpdir, ) with pytest.raises(AutoscalerInitFailException): monitor.run() mock_publish.assert_called_once() if prev_port is not None: os.environ["RAY_GCS_SERVER_PORT"] = prev_port else: del os.environ["RAY_GCS_SERVER_PORT"] def testInitializeSDKArguments(self): # https://github.com/ray-project/ray/issues/23166 from ray.autoscaler.sdk import request_resources with self.assertRaises(TypeError): request_resources(num_cpus="bar") with self.assertRaises(TypeError): request_resources(bundles="bar") with self.assertRaises(TypeError): request_resources(bundles=["foo"]) # Non-numerical bundle values are rejected. with self.assertRaises(TypeError): request_resources(bundles=[{"foo": "bar"}]) with self.assertRaises(TypeError): request_resources(bundles=[{"foo": 1}, {"bar": "baz"}]) # bool is a subclass of int but is not a meaningful resource value. with self.assertRaises(TypeError): request_resources(bundles=[{"CPU": True}]) with patch("ray.autoscaler.sdk.sdk.commands.request_resources") as mock_cmd: request_resources(bundles=[{"CPU": 0.1}]) request_resources(bundles=[{"CPU": 1, "GPU": 0.5}]) assert mock_cmd.call_count == 2 def test_autoscaler_status_log(self): self._test_autoscaler_status_log(status_log_enabled_env=1) self._test_autoscaler_status_log(status_log_enabled_env=0) def _test_autoscaler_status_log(self, status_log_enabled_env: int): mock_logger = Mock(spec=logging.Logger("")) with patch.multiple( "ray.autoscaler._private.autoscaler", logger=mock_logger, AUTOSCALER_STATUS_LOG=status_log_enabled_env, ): config = copy.deepcopy(SMALL_CLUSTER) config_path = self.write_config(config) runner = MockProcessRunner() mock_metrics = Mock(spec=AutoscalerPrometheusMetrics()) self.provider = MockProvider() autoscaler = MockAutoscaler( config_path, LoadMetrics(), MockGcsClient(), max_failures=0, process_runner=runner, update_interval_s=0, prom_metrics=mock_metrics, ) autoscaler.update() status_log_found = False for call in mock_logger.info.call_args_list: args, _ = call arg = args[0] if " Autoscaler status: " in arg: status_log_found = True break assert status_log_found is bool(status_log_enabled_env) def testScaleDownIdleTimeOut(self): config = copy.deepcopy(SMALL_CLUSTER) config["available_node_types"]["worker"]["min_workers"] = 1 config_path = self.write_config(config) self.provider = MockProvider() self.provider.create_node( {}, { TAG_RAY_NODE_KIND: NODE_KIND_HEAD, TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, TAG_RAY_USER_NODE_TYPE: "head", }, 1, ) runner = MockProcessRunner() lm = LoadMetrics() mock_gcs_client = MockGcsClient() autoscaler = MockAutoscaler( config_path, lm, mock_gcs_client, max_failures=0, process_runner=runner, update_interval_s=0, ) autoscaler.update() self.waitForNodes(1, tag_filters=WORKER_FILTER) # Reduce cluster size to 1 new_config = copy.deepcopy(SMALL_CLUSTER) new_config["available_node_types"]["worker"]["min_workers"] = 0 new_config["idle_timeout_minutes"] = 0.1 self.write_config(new_config) autoscaler.update() worker_ip = self.provider.non_terminated_node_ips(WORKER_FILTER)[0] # Mark the node as idle lm.update(worker_ip, mock_node_id(), {"CPU": 1}, {"CPU": 1}, 20) autoscaler.update() assert self.provider.internal_ip("1") == worker_ip events = autoscaler.event_summarizer.summary() assert "Removing 1 nodes of type worker (idle)." in events, events autoscaler.update() assert mock_gcs_client.drain_node_call_count == 1 def testDontScaleDownIdleTimeOutForPlacementGroups(self): config = copy.deepcopy(SMALL_CLUSTER) config["available_node_types"]["head"]["resources"][ "CPU" ] = 0 # make the head node not consume any resources. config["available_node_types"]["worker"][ "min_workers" ] = 1 # prepare 1 worker upfront. config["idle_timeout_minutes"] = 0.1 config_path = self.write_config(config) self.provider = MockProvider() self.provider.create_node( {}, { TAG_RAY_NODE_KIND: NODE_KIND_HEAD, TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, TAG_RAY_USER_NODE_TYPE: "head", }, 1, ) runner = MockProcessRunner() # Avoid the "Unable to deserialize `image_env` to Python object" error in the DockerCommandRunner. runner.respond_to_call("json .Config.Env", ["[]" for i in range(2)]) lm = LoadMetrics() mock_gcs_client = MockGcsClient() autoscaler = MockAutoscaler( config_path, lm, mock_gcs_client, max_failures=0, process_runner=runner, update_interval_s=0, ) autoscaler.update() # 1 worker is ready upfront. self.waitForNodes(1, tag_filters=WORKER_FILTER) # clear the summary for later check. autoscaler.event_summarizer.clear() # Restore min_workers to allow scaling down to 0. config["available_node_types"]["worker"]["min_workers"] = 0 self.write_config(config) # Create a placement group with 2 bundles that require 2 workers. placement_group_table_data = gcs_pb2.PlacementGroupTableData( placement_group_id=b"\000", strategy=common_pb2.PlacementStrategy.SPREAD, ) for i in range(2): bundle = common_pb2.Bundle() bundle.bundle_id.placement_group_id = ( placement_group_table_data.placement_group_id ) bundle.bundle_id.bundle_index = i bundle.unit_resources["CPU"] = 1 placement_group_table_data.bundles.append(bundle) # Mark the first worker as idle, but it should not be scaled down by the autoscaler because it will be used by the placement group. worker_ip = self.provider.non_terminated_node_ips(WORKER_FILTER)[0] lm.update( worker_ip, mock_node_id(), {"CPU": 1}, {"CPU": 1}, 20, # idle for 20 seconds, which is longer than the idle_timeout_minutes. None, None, [placement_group_table_data], ) autoscaler.update() # TODO(rueian): This is a hack to avoid running into race conditions # within v1 autoscaler. These should no longer be relevant in v2. self.waitForNodes(2, tag_filters=WORKER_FILTER) events = autoscaler.event_summarizer.summary() assert "Removing 1 nodes of type worker (idle)." not in events, events assert "Adding 1 node(s) of type worker." in events, events autoscaler.update() self.waitForNodes(2, tag_filters=WORKER_FILTER) def testRecoverUnhealthyWorkersWithNodeSpecificDocker(self): """Test that recovery uses node-specific docker configuration. This test verifies that when a worker node becomes unhealthy and needs recovery, the autoscaler uses the node-specific docker configuration rather than the global docker configuration. """ config = copy.deepcopy(SMALL_CLUSTER) # Top-level global docker config (should be overridden by node-specific config) config["docker"]["image"] = "global-image:latest" config["docker"]["worker_image"] = "global-worker-image:latest" # Add node-specific docker configuration config["available_node_types"]["worker"]["docker"] = { "worker_image": "node-specific-worker-image:latest", "worker_run_options": ["--gpus=all"], } config["available_node_types"]["worker"]["min_workers"] = 1 config_path = self.write_config(config) self.provider = MockProvider() runner = MockProcessRunner() runner.respond_to_call("json .Config.Env", ["[]" for i in range(2)]) lm = LoadMetrics() mock_metrics = Mock() # Create head node self.provider.create_node( {}, { TAG_RAY_NODE_KIND: NODE_KIND_HEAD, TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, TAG_RAY_USER_NODE_TYPE: "head", }, 1, ) autoscaler = MockAutoscaler( config_path, lm, MockGcsClient(), max_failures=0, process_runner=runner, update_interval_s=0, prom_metrics=mock_metrics, ) autoscaler.update() self.waitForNodes(1, tag_filters=WORKER_FILTER) self.provider.finish_starting_nodes() autoscaler.update() self.waitForNodes( 1, tag_filters={TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, **WORKER_FILTER} ) # Wait for initial updaters to finish self.waitForUpdatersToFinish(autoscaler) autoscaler.update() # Ensure initial updaters are cleared after they finish assert not autoscaler.updaters # Clear command history before triggering recovery to ensure we only check # commands from the recovery process, not the initial node creation runner.clear_history() # Trigger node recovery by setting the last heartbeat time to be before the timeout worker_ip = "172.0.0.1" # Expected IP of the first worker node lm.last_heartbeat_time_by_ip[worker_ip] = ( time.time() - AUTOSCALER_HEARTBEAT_TIMEOUT_S - 1 ) autoscaler.update() # Wait for recovery to start and finish self.waitFor(lambda: len(autoscaler.updaters) > 0, num_retries=150) self.waitForUpdatersToFinish(autoscaler) # Verify that recovery has started by checking multiple indicators: # 1. Check that an updater was created for recovery assert len(autoscaler.updaters) == 1 node_id = list(autoscaler.updaters.keys())[0] updater = autoscaler.updaters[node_id] # 2. Verify the updater is marked as a recovery updater assert updater.for_recovery is True # 3. Verify the recovery event was logged events = autoscaler.event_summarizer.summary() assert any( "Restarting" in event and "lost contact with raylet" in event for event in events ) # 4. Verify that the recovery process uses the node-specific docker image # instead of the global docker image runner.assert_has_call(worker_ip, pattern="node-specific-worker-image:latest") # 5. Verify that the recovery process uses the node-specific run options runner.assert_has_call(worker_ip, pattern="--gpus=all") # 6. Verify that the recovery updater has the correct docker config # by checking that it uses the node-specific docker configuration assert ( updater.docker_config.get("worker_image") == "node-specific-worker-image:latest" ) assert "--gpus=all" in updater.docker_config.get("worker_run_options") def test_node_becomes_inactive_after_heartbeat_timeout(self): cluster_config = copy.deepcopy(MOCK_DEFAULT_CONFIG) cluster_config["available_node_types"]["ray.worker.default"]["min_workers"] = 1 cluster_config["worker_start_ray_commands"] = ["ray_start_cmd"] cluster_config["head_node_type"] = ["ray.worker.default"] del cluster_config["available_node_types"]["ray.head.default"] del cluster_config["docker"] config_path = self.write_config(cluster_config) self.provider = MockProvider() runner = MockProcessRunner() lm = LoadMetrics() mock_gcs_client = MockGcsClient() autoscaler = MockAutoscaler( config_path, lm, mock_gcs_client, max_failures=0, process_runner=runner, update_interval_s=0, ) autoscaler.update() self.waitForNodes(1, tag_filters=WORKER_FILTER) self.provider.finish_starting_nodes() autoscaler.update() self.waitForNodes( 1, tag_filters={TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, **WORKER_FILTER} ) self.waitForUpdatersToFinish(autoscaler) autoscaler.update() assert not autoscaler.updaters worker_ip = self.provider.non_terminated_node_ips(WORKER_FILTER)[0] now = time.time() past_heartbeat = now - AUTOSCALER_HEARTBEAT_TIMEOUT_S - 1 lm.last_heartbeat_time_by_ip[worker_ip] = past_heartbeat autoscaler.update() self.waitForNodes( 1, tag_filters={TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE, **WORKER_FILTER} ) events = autoscaler.summary() assert events.failed_nodes == [("172.0.0.0", "ray.worker.default")] def test_import(): """This test ensures that all the autoscaler imports work as expected to prevent errors such as #19840. """ import ray # noqa ray.autoscaler.sdk.request_resources # noqa import ray.autoscaler # noqa import ray.autoscaler.sdk # noqa from ray.autoscaler.sdk import request_resources # noqa def test_prom_null_metric_inc_fix(): """Verify the bug fix https://github.com/ray-project/ray/pull/27532 for NullMetric's signature. Check that NullMetric can be called with or without an argument. """ NullMetric().inc() NullMetric().inc(5) if __name__ == "__main__": sys.exit(pytest.main(["-sv", __file__]))