import math import os import sys import time # coding: utf-8 from collections import defaultdict from unittest import mock from unittest.mock import MagicMock import pytest from ray._common.utils import binary_to_hex from ray.autoscaler.v2.instance_manager.config import InstanceReconcileConfig, Provider from ray.autoscaler.v2.instance_manager.instance_manager import InstanceManager from ray.autoscaler.v2.instance_manager.instance_storage import InstanceStorage from ray.autoscaler.v2.instance_manager.node_provider import ( # noqa CloudInstance, LaunchNodeError, TerminateNodeError, ) from ray.autoscaler.v2.instance_manager.reconciler import Reconciler, logger from ray.autoscaler.v2.instance_manager.storage import InMemoryStorage from ray.autoscaler.v2.instance_manager.subscribers.cloud_resource_monitor import ( CloudResourceMonitor, ) from ray.autoscaler.v2.instance_manager.subscribers.ray_stopper import RayStopError from ray.autoscaler.v2.instance_manager.subscribers.threaded_ray_installer import ( RayInstallError, ) from ray.autoscaler.v2.scheduler import IResourceScheduler, SchedulingReply from ray.autoscaler.v2.tests.util import MockSubscriber, create_instance from ray.core.generated.autoscaler_pb2 import ( ClusterResourceState, GangResourceRequest, NodeState, NodeStatus, ResourceRequest, ) from ray.core.generated.instance_manager_pb2 import ( Instance, LaunchRequest, NodeKind, TerminationRequest, ) s_to_ns = 1 * 1_000_000_000 logger.setLevel("DEBUG") class MockAutoscalingConfig: def __init__(self, configs=None): if configs is None: configs = {} self._configs = configs def get_node_type_configs(self): return self._configs.get("node_type_configs", {}) def get_max_num_worker_nodes(self): return self._configs.get("max_num_worker_nodes") def get_max_num_nodes(self): n = self._configs.get("max_num_worker_nodes") return n + 1 if n is not None else None def get_upscaling_speed(self): return self._configs.get("upscaling_speed", 0.0) def get_max_concurrent_launches(self): return self._configs.get("max_concurrent_launches", 100) def get_instance_reconcile_config(self): return self._configs.get("instance_reconcile_config", InstanceReconcileConfig()) def disable_node_updaters(self): return self._configs.get("disable_node_updaters", True) def disable_launch_config_check(self): return self._configs.get("disable_launch_config_check", False) def get_idle_timeout_s(self): return self._configs.get("idle_timeout_s", 999) def get_provider_instance_type(self, ray_node_type): return self._configs.get("provider_instance_types", {}).get(ray_node_type, "") @property def provider(self): return self._configs.get("provider", Provider.UNKNOWN) class MockScheduler(IResourceScheduler): def __init__(self, to_launch=None, to_terminate=None): if to_launch is None: to_launch = [] if to_terminate is None: to_terminate = [] self.to_launch = to_launch self.to_terminate = to_terminate def schedule(self, req): return SchedulingReply( to_launch=self.to_launch, to_terminate=self.to_terminate, ) @pytest.fixture() def setup(): instance_storage = InstanceStorage( cluster_id="test_cluster_id", storage=InMemoryStorage(), ) mock_subscriber = MockSubscriber() instance_manager = InstanceManager( instance_storage=instance_storage, instance_status_update_subscribers=[mock_subscriber], ) cloud_resource_monitor = CloudResourceMonitor() yield instance_manager, instance_storage, mock_subscriber, cloud_resource_monitor class TestReconciler: @staticmethod def _add_instances(instance_storage, instances): for instance in instances: ok, _ = instance_storage.upsert_instance(instance) assert ok @staticmethod def test_requested_instance_no_op(setup): instance_manager, instance_storage, _, cloud_resource_monitor = setup # Request no timeout yet. TestReconciler._add_instances( instance_storage, [ create_instance( "i-1", Instance.REQUESTED, launch_request_id="l1", instance_type="type-1", status_times=[(Instance.REQUESTED, time.time_ns())], ), ], ) Reconciler.reconcile( instance_manager, scheduler=MockScheduler(), cloud_provider=MagicMock(), cloud_resource_monitor=cloud_resource_monitor, ray_cluster_resource_state=ClusterResourceState(), non_terminated_cloud_instances={}, cloud_provider_errors=[], ray_install_errors=[], autoscaling_config=MockAutoscalingConfig(), ) instances, _ = instance_storage.get_instances() assert len(instances) == 1 assert instances["i-1"].status == Instance.REQUESTED @staticmethod def test_requested_instance_to_allocated(setup): # When there's a matching cloud instance for the requested instance. # The requested instance should be moved to ALLOCATED. instance_manager, instance_storage, _, cloud_resource_monitor = setup TestReconciler._add_instances( instance_storage, [ create_instance( "i-1", status=Instance.REQUESTED, instance_type="type-1", status_times=[(Instance.REQUESTED, time.time_ns())], launch_request_id="l1", ), create_instance( "i-2", status=Instance.REQUESTED, instance_type="type-2", status_times=[(Instance.REQUESTED, time.time_ns())], launch_request_id="l2", ), ], ) cloud_instances = { "c-1": CloudInstance("c-1", "type-1", True, NodeKind.WORKER), } Reconciler.reconcile( instance_manager, scheduler=MockScheduler(), cloud_provider=MagicMock(), cloud_resource_monitor=cloud_resource_monitor, ray_cluster_resource_state=ClusterResourceState(), non_terminated_cloud_instances=cloud_instances, cloud_provider_errors=[], ray_install_errors=[], autoscaling_config=MockAutoscalingConfig(), ) instances, _ = instance_storage.get_instances() assert len(instances) == 2 assert instances["i-1"].status == Instance.ALLOCATED assert instances["i-1"].cloud_instance_id == "c-1" assert instances["i-2"].status == Instance.REQUESTED @staticmethod def test_requested_instance_to_allocation_failed(setup): """ Test that the instance should be transitioned to ALLOCATION_FAILED when launch error happens. """ instance_manager, instance_storage, _, cloud_resource_monitor = setup instances = [ # Should succeed with instance matched. create_instance( "i-1", status=Instance.REQUESTED, instance_type="type-1", launch_request_id="l1", ), # Should fail due to launch error. create_instance( "i-2", status=Instance.REQUESTED, instance_type="type-2", launch_request_id="l1", ), ] TestReconciler._add_instances(instance_storage, instances) launch_errors = [ LaunchNodeError( request_id="l1", count=1, # The request failed. node_type="type-2", timestamp_ns=1, ) ] cloud_instances = { "c-1": CloudInstance("c-1", "type-1", True, NodeKind.WORKER), } Reconciler.reconcile( instance_manager, scheduler=MockScheduler(), cloud_provider=MagicMock(), cloud_resource_monitor=cloud_resource_monitor, ray_cluster_resource_state=ClusterResourceState(), non_terminated_cloud_instances=cloud_instances, cloud_provider_errors=launch_errors, ray_install_errors=[], autoscaling_config=MockAutoscalingConfig(), ) instances, _ = instance_storage.get_instances() assert len(instances) == 2 assert instances["i-1"].status == Instance.ALLOCATED assert instances["i-1"].cloud_instance_id == "c-1" assert instances["i-2"].status == Instance.ALLOCATION_FAILED @staticmethod def test_reconcile_terminated_cloud_instances(setup): instance_manager, instance_storage, subscriber, cloud_resource_monitor = setup instances = [ create_instance( "i-1", status=Instance.ALLOCATED, instance_type="type-1", cloud_instance_id="c-1", ), create_instance( "i-2", status=Instance.TERMINATING, # pending terminated, no update instance_type="type-2", cloud_instance_id="c-2", ), create_instance( "i-3", status=Instance.TERMINATED, # Already terminated, no update instance_type="type-2", cloud_instance_id="c-3", ), ] TestReconciler._add_instances(instance_storage, instances) cloud_instances = { "c-2": CloudInstance("c-2", "type-2", False, NodeKind.WORKER), } termination_errors = [ TerminateNodeError( cloud_instance_id="c-2", timestamp_ns=1, request_id="t1", ) ] subscriber.clear() Reconciler.reconcile( instance_manager, scheduler=MockScheduler(), cloud_provider=MagicMock(), cloud_resource_monitor=cloud_resource_monitor, ray_cluster_resource_state=ClusterResourceState(), non_terminated_cloud_instances=cloud_instances, cloud_provider_errors=termination_errors, ray_install_errors=[], autoscaling_config=MockAutoscalingConfig(), ) instances, _ = instance_storage.get_instances() assert len(instances) == 3 assert instances["i-1"].status == Instance.TERMINATED assert instances["i-2"].status == Instance.TERMINATING events = subscriber.events assert len(events) == 3 events_i_1 = subscriber.events_by_id("i-1") assert len(events_i_1) == 1 assert events_i_1[0].new_instance_status == Instance.TERMINATED assert events_i_1[0].instance_id == "i-1" assert events_i_1[0].cloud_instance_id == "c-1" events_i_2 = subscriber.events_by_id("i-2") assert len(events_i_2) == 2 assert events_i_2[0].new_instance_status == Instance.TERMINATION_FAILED assert events_i_2[1].new_instance_status == Instance.TERMINATING @staticmethod def test_ray_reconciler_no_op(setup): instance_manager, instance_storage, subscriber, cloud_resource_monitor = setup im_instances = [ create_instance( "i-1", status=Instance.ALLOCATED, launch_request_id="l1", instance_type="type-1", cloud_instance_id="c-1", ), ] cloud_instances = { "c-1": CloudInstance("c-1", "type-1", True, NodeKind.WORKER), } TestReconciler._add_instances(instance_storage, im_instances) subscriber.clear() Reconciler.reconcile( instance_manager, scheduler=MockScheduler(), cloud_provider=MagicMock(), cloud_resource_monitor=cloud_resource_monitor, ray_cluster_resource_state=ClusterResourceState(), non_terminated_cloud_instances=cloud_instances, cloud_provider_errors=[], ray_install_errors=[], autoscaling_config=MockAutoscalingConfig(), ) # Assert no changes. assert subscriber.events == [] # Unknown ray node status - no action. ray_nodes = [ NodeState(node_id=b"r-1", status=NodeStatus.UNSPECIFIED, instance_id="c-1"), ] with pytest.raises(ValueError): Reconciler.reconcile( instance_manager, scheduler=MockScheduler(), cloud_provider=MagicMock(), cloud_resource_monitor=cloud_resource_monitor, ray_cluster_resource_state=ClusterResourceState(node_states=ray_nodes), non_terminated_cloud_instances=cloud_instances, cloud_provider_errors=[], ray_install_errors=[], autoscaling_config=MockAutoscalingConfig(), ) # Assert no changes. assert subscriber.events == [] @staticmethod def test_ray_reconciler_new_ray(setup): instance_manager, instance_storage, _, cloud_resource_monitor = setup # A newly running ray node with matching cloud instance id node_states = [ NodeState(node_id=b"r-1", status=NodeStatus.RUNNING, instance_id="c-1"), ] im_instances = [ create_instance( "i-0", status=Instance.TERMINATED, cloud_instance_id="c-1" ), # this should not be matched. create_instance("i-1", status=Instance.ALLOCATED, cloud_instance_id="c-1"), ] cloud_instances = { "c-1": CloudInstance("c-1", "type-1", True, NodeKind.WORKER), } TestReconciler._add_instances(instance_storage, im_instances) Reconciler.reconcile( instance_manager, scheduler=MockScheduler(), cloud_provider=MagicMock(), cloud_resource_monitor=cloud_resource_monitor, ray_cluster_resource_state=ClusterResourceState(node_states=node_states), non_terminated_cloud_instances=cloud_instances, cloud_provider_errors=[], ray_install_errors=[], autoscaling_config=MockAutoscalingConfig(), ) instances, _ = instance_storage.get_instances() assert len(instances) == 2 assert instances["i-0"].status == Instance.TERMINATED assert instances["i-1"].status == Instance.RAY_RUNNING assert instances["i-1"].node_id == binary_to_hex(b"r-1") @staticmethod def test_ray_reconciler_already_ray_running(setup): instance_manager, instance_storage, subscriber, cloud_resource_monitor = setup # A running ray node already reconciled. TestReconciler._add_instances( instance_storage, [ create_instance( "i-1", status=Instance.RAY_RUNNING, cloud_instance_id="c-1" ), create_instance( "i-2", status=Instance.TERMINATING, cloud_instance_id="c-2" ), # Already reconciled. ], ) ray_nodes = [ NodeState(node_id=b"r-1", status=NodeStatus.IDLE, instance_id="c-1"), NodeState( node_id=b"r-2", status=NodeStatus.IDLE, instance_id="c-2" ), # Already being stopped ] cloud_instances = { "c-1": CloudInstance("c-1", "type-1", True, NodeKind.WORKER), "c-2": CloudInstance("c-2", "type-2", True, NodeKind.WORKER), } subscriber.clear() Reconciler.reconcile( instance_manager, scheduler=MockScheduler(), cloud_provider=MagicMock(), cloud_resource_monitor=cloud_resource_monitor, ray_cluster_resource_state=ClusterResourceState(node_states=ray_nodes), non_terminated_cloud_instances=cloud_instances, cloud_provider_errors=[], ray_install_errors=[], autoscaling_config=MockAutoscalingConfig(), ) assert len(subscriber.events) == 0 @staticmethod def test_ray_reconciler_stopping_ray(setup): instance_manager, instance_storage, _, cloud_resource_monitor = setup # draining ray nodes im_instances = [ create_instance( "i-1", status=Instance.RAY_RUNNING, cloud_instance_id="c-1" ), # To be reconciled. create_instance( "i-2", status=Instance.RAY_STOPPING, cloud_instance_id="c-2" ), # Already reconciled. create_instance( "i-3", status=Instance.TERMINATING, cloud_instance_id="c-3" ), # Already reconciled. ] TestReconciler._add_instances(instance_storage, im_instances) ray_nodes = [ NodeState(node_id=b"r-1", status=NodeStatus.DRAINING, instance_id="c-1"), NodeState(node_id=b"r-2", status=NodeStatus.DRAINING, instance_id="c-2"), NodeState(node_id=b"r-3", status=NodeStatus.DRAINING, instance_id="c-3"), ] cloud_instances = { "c-1": CloudInstance("c-1", "type-1", True, NodeKind.WORKER), "c-2": CloudInstance("c-2", "type-2", True, NodeKind.WORKER), "c-3": CloudInstance("c-3", "type-3", True, NodeKind.WORKER), } Reconciler.reconcile( instance_manager, scheduler=MockScheduler(), cloud_provider=MagicMock(), cloud_resource_monitor=cloud_resource_monitor, ray_cluster_resource_state=ClusterResourceState(node_states=ray_nodes), non_terminated_cloud_instances=cloud_instances, cloud_provider_errors=[], ray_install_errors=[], autoscaling_config=MockAutoscalingConfig(), ) instances, _ = instance_storage.get_instances() assert len(instances) == 3 assert instances["i-1"].status == Instance.RAY_STOPPING assert instances["i-2"].status == Instance.RAY_STOPPING assert instances["i-3"].status == Instance.TERMINATING @staticmethod def test_ray_reconciler_stopped_ray(setup): instance_manager, instance_storage, _, cloud_resource_monitor = setup # dead ray nodes im_instances = [ create_instance( "i-1", status=Instance.ALLOCATED, cloud_instance_id="c-1" ), # To be reconciled. create_instance( "i-2", status=Instance.RAY_STOPPING, cloud_instance_id="c-2" ), # To be reconciled. create_instance( "i-3", status=Instance.TERMINATING, cloud_instance_id="c-3" ), # Already reconciled. ] TestReconciler._add_instances(instance_storage, im_instances) ray_nodes = [ NodeState(node_id=b"r-1", status=NodeStatus.DEAD, instance_id="c-1"), NodeState(node_id=b"r-2", status=NodeStatus.DEAD, instance_id="c-2"), NodeState(node_id=b"r-3", status=NodeStatus.DEAD, instance_id="c-3"), ] cloud_instances = { "c-1": CloudInstance("c-1", "type-1", True, NodeKind.WORKER), "c-2": CloudInstance("c-2", "type-2", True, NodeKind.WORKER), "c-3": CloudInstance("c-3", "type-3", True, NodeKind.WORKER), } Reconciler.reconcile( instance_manager, scheduler=MockScheduler(), cloud_provider=MagicMock(), cloud_resource_monitor=cloud_resource_monitor, ray_cluster_resource_state=ClusterResourceState(node_states=ray_nodes), non_terminated_cloud_instances=cloud_instances, cloud_provider_errors=[], ray_install_errors=[], autoscaling_config=MockAutoscalingConfig(), ) instances, _ = instance_storage.get_instances() assert len(instances) == 3 assert instances["i-1"].status == Instance.TERMINATING assert instances["i-2"].status == Instance.TERMINATING assert instances["i-3"].status == Instance.TERMINATING @staticmethod def test_reconcile_ray_installer_failures(setup): instance_manager, instance_storage, _, cloud_resource_monitor = setup # dead ray nodes im_instances = [ create_instance( "i-1", status=Instance.RAY_INSTALLING, cloud_instance_id="c-1" ), # To be reconciled. ] TestReconciler._add_instances(instance_storage, im_instances) ray_install_errors = [ RayInstallError( im_instance_id="i-1", details="failed to install", ) ] cloud_instances = { "c-1": CloudInstance("c-1", "type-1", True, NodeKind.WORKER), } Reconciler.reconcile( instance_manager, scheduler=MockScheduler(), cloud_provider=MagicMock(), cloud_resource_monitor=cloud_resource_monitor, ray_cluster_resource_state=ClusterResourceState(node_states=[]), non_terminated_cloud_instances=cloud_instances, cloud_provider_errors=[], ray_install_errors=ray_install_errors, autoscaling_config=MockAutoscalingConfig(), ) instances, _ = instance_storage.get_instances() assert len(instances) == 1 assert instances["i-1"].status == Instance.TERMINATING @staticmethod def test_draining_ray_node_also_terminated(setup): """ A draining ray node due to a cloud instance termination should also transition the instance to TERMINATED. """ instance_manager, instance_storage, _, cloud_resource_monitor = setup im_instances = [ create_instance( "i-1", status=Instance.RAY_RUNNING, cloud_instance_id="c-1", ray_node_id=binary_to_hex(b"r-1"), ), # To be reconciled. create_instance( "i-2", status=Instance.RAY_RUNNING, cloud_instance_id="c-2", ray_node_id=binary_to_hex(b"r-2"), ), # To be reconciled. ] TestReconciler._add_instances(instance_storage, im_instances) ray_nodes = [ NodeState(node_id=b"r-1", status=NodeStatus.DEAD, instance_id="c-1"), NodeState(node_id=b"r-2", status=NodeStatus.DRAINING, instance_id="c-2"), ] cloud_instances = { # Terminated cloud instance. } Reconciler.reconcile( instance_manager, scheduler=MockScheduler(), cloud_provider=MagicMock(), cloud_resource_monitor=cloud_resource_monitor, ray_cluster_resource_state=ClusterResourceState(node_states=ray_nodes), non_terminated_cloud_instances=cloud_instances, cloud_provider_errors=[], ray_install_errors=[], autoscaling_config=MockAutoscalingConfig(), ) instances, _ = instance_storage.get_instances() assert len(instances) == 2 assert instances["i-1"].status == Instance.TERMINATED assert instances["i-2"].status == Instance.TERMINATED @staticmethod @pytest.mark.parametrize( "max_concurrent_launches,num_allocated,num_requested,num_running", [ (1, 0, 0, 0), (10, 0, 0, 0), (1, 0, 1, 1), (1, 1, 0, 1), (10, 1, 0, 1), (10, 0, 1, 1), (10, 5, 5, 5), ], ) @pytest.mark.parametrize( "upscaling_speed", [0.0, 0.1, 0.5, 1.0, 100.0], ) def test_max_concurrent_launches( max_concurrent_launches, num_allocated, num_requested, num_running, upscaling_speed, setup, ): instance_manager, instance_storage, subscriber, cloud_resource_monitor = setup next_id = 0 # Add some allocated instances. cloud_instances = {} for _ in range(num_allocated): instance = create_instance( str(next_id), status=Instance.ALLOCATED, instance_type="type-1", cloud_instance_id=f"c-{next_id}", ) next_id += 1 cloud_instances[instance.cloud_instance_id] = CloudInstance( instance.cloud_instance_id, "type-1", True, NodeKind.WORKER ) TestReconciler._add_instances(instance_storage, [instance]) # Add some requested instances. for _ in range(num_requested): instance = create_instance( str(next_id), status=Instance.REQUESTED, instance_type="type-1", launch_request_id="l-1", ) TestReconciler._add_instances(instance_storage, [instance]) next_id += 1 # Add many queued instances. queued_instances = [ create_instance( str(i + next_id), status=Instance.QUEUED, instance_type="type-1" ) for i in range(1000) ] TestReconciler._add_instances(instance_storage, queued_instances) # Add some running instances. for _ in range(num_running): instance = create_instance( str(next_id), status=Instance.RAY_RUNNING, instance_type="type-1", launch_request_id="l-1", ) TestReconciler._add_instances(instance_storage, [instance]) next_id += 1 num_desired_upscale = max(1, math.ceil(upscaling_speed * (max(num_running, 1)))) expected_launch_num = min( num_desired_upscale, max(0, max_concurrent_launches - num_requested), # global limit ) subscriber.clear() Reconciler.reconcile( instance_manager=instance_manager, scheduler=MockScheduler(), cloud_provider=MagicMock(), cloud_resource_monitor=cloud_resource_monitor, ray_cluster_resource_state=ClusterResourceState(), non_terminated_cloud_instances=cloud_instances, cloud_provider_errors=[], ray_install_errors=[], autoscaling_config=MockAutoscalingConfig( configs={ "upscaling_speed": upscaling_speed, "max_concurrent_launches": max_concurrent_launches, } ), ) instances, _ = instance_storage.get_instances() assert len(subscriber.events) == expected_launch_num for event in subscriber.events: assert event.new_instance_status == Instance.REQUESTED assert ( event.launch_request_id == instances[event.instance_id].launch_request_id ) assert event.instance_type == "type-1" @staticmethod @mock.patch("time.time_ns") def test_stuck_instances_requested(mock_time_ns, setup): instance_manager, instance_storage, subscriber, cloud_resource_monitor = setup cur_time_s = 10 mock_time_ns.return_value = cur_time_s * s_to_ns reconcile_config = InstanceReconcileConfig( request_status_timeout_s=5, max_num_retry_request_to_allocate=1, # max 1 retry. ) instances = [ create_instance( "no-update", Instance.REQUESTED, status_times=[(Instance.REQUESTED, 9 * s_to_ns)], launch_request_id="l1", ), create_instance( "retry", Instance.REQUESTED, status_times=[(Instance.REQUESTED, 2 * s_to_ns)], launch_request_id="l2", ), create_instance( "failed", Instance.REQUESTED, status_times=[ (Instance.REQUESTED, 1 * s_to_ns), (Instance.REQUESTED, 2 * s_to_ns), # This is a retry. ], launch_request_id="l3", ), ] TestReconciler._add_instances(instance_storage, instances) Reconciler.reconcile( instance_manager=instance_manager, scheduler=MockScheduler(), cloud_provider=MagicMock(), cloud_resource_monitor=cloud_resource_monitor, ray_cluster_resource_state=ClusterResourceState(), non_terminated_cloud_instances={}, cloud_provider_errors=[], ray_install_errors=[], autoscaling_config=MockAutoscalingConfig( configs={ "instance_reconcile_config": reconcile_config, "max_concurrent_launches": 0, # prevent launches } ), ) instances, _ = instance_storage.get_instances() assert instances["no-update"].status == Instance.REQUESTED assert instances["retry"].status == Instance.QUEUED assert instances["failed"].status == Instance.ALLOCATION_FAILED @staticmethod @mock.patch("time.time_ns") def test_stuck_instances_ray_stop_requested(mock_time_ns, setup): instance_manager, instance_storage, subscriber, cloud_resource_monitor = setup timeout_s = 5 cur_time_s = 20 mock_time_ns.return_value = cur_time_s * s_to_ns config = InstanceReconcileConfig( ray_stop_requested_status_timeout_s=timeout_s, ) cur_status = Instance.RAY_STOP_REQUESTED instances = [ create_instance( "no-update", cur_status, status_times=[(cur_status, (cur_time_s - timeout_s + 1) * s_to_ns)], ray_node_id=binary_to_hex(b"r-1"), ), create_instance( "updated", cur_status, status_times=[(cur_status, (cur_time_s - timeout_s - 1) * s_to_ns)], ray_node_id=binary_to_hex(b"r-2"), ), ] # Both nodes are still RUNNING in GCS, so the drain is treated as having # genuinely failed: the stuck handler must fall back to RAY_RUNNING. ray_nodes = [ NodeState(node_id=b"r-1", status=NodeStatus.RUNNING), NodeState(node_id=b"r-2", status=NodeStatus.RUNNING), ] TestReconciler._add_instances(instance_storage, instances) Reconciler.reconcile( instance_manager=instance_manager, scheduler=MockScheduler(), cloud_provider=MagicMock(), cloud_resource_monitor=cloud_resource_monitor, ray_cluster_resource_state=ClusterResourceState(node_states=ray_nodes), non_terminated_cloud_instances={}, cloud_provider_errors=[], ray_install_errors=[], autoscaling_config=MockAutoscalingConfig( configs={ "instance_reconcile_config": config, "max_concurrent_launches": 0, # prevent launches } ), ) instances, _ = instance_storage.get_instances() assert instances["no-update"].status == cur_status assert instances["updated"].status == Instance.RAY_RUNNING @staticmethod @pytest.mark.parametrize( "cur_status", [ Instance.RAY_RUNNING, Instance.RAY_STOP_REQUESTED, Instance.RAY_STOPPING, ], ids=["ray_running", "ray_stop_requested", "ray_stopping"], ) def test_missing_ray_node_marks_ray_stopped_and_terminates(cur_status, setup): # If an instance that should have a ray node points at a node_id that no # longer appears in GCS, the ray node has either been GC'd from the DEAD # list or is otherwise no longer alive. Reconcile it to RAY_STOPPED so it # can move through the normal termination path instead of lingering with a # stale node_id (especially after RAY_STOP_REQUESTED fell back to # RAY_RUNNING in a previous iteration). instance_manager, instance_storage, subscriber, cloud_resource_monitor = setup instances = [ create_instance( "updated", cur_status, ray_node_id=binary_to_hex(b"r-stale"), cloud_instance_id="c-stale", ), ] TestReconciler._add_instances(instance_storage, instances) Reconciler.reconcile( instance_manager=instance_manager, scheduler=MockScheduler(), cloud_provider=MagicMock(), cloud_resource_monitor=cloud_resource_monitor, ray_cluster_resource_state=ClusterResourceState(node_states=[]), non_terminated_cloud_instances={ "c-stale": CloudInstance( cloud_instance_id="c-stale", node_type="worker_nodes1", node_kind=NodeKind.WORKER, is_running=True, ), }, cloud_provider_errors=[], ray_install_errors=[], autoscaling_config=MockAutoscalingConfig( configs={"max_concurrent_launches": 0} ), ) events = subscriber.events assert len(events) == 2 assert events[0].new_instance_status == Instance.RAY_STOPPED assert events[0].ray_node_id == binary_to_hex(b"r-stale") assert events[1].new_instance_status == Instance.TERMINATING instances, _ = instance_storage.get_instances() assert instances["updated"].status == Instance.TERMINATING @staticmethod def test_ray_stopped_shared_cloud_terminates_stale_record_only(setup): instance_manager, instance_storage, subscriber, _ = setup TestReconciler._add_instances( instance_storage, [ create_instance( "stale", Instance.RAY_STOPPED, ray_node_id=binary_to_hex(b"r-stale"), cloud_instance_id="c-shared", ), create_instance( "healthy", Instance.RAY_RUNNING, ray_node_id=binary_to_hex(b"r-live"), cloud_instance_id="c-shared", ), ], ) Reconciler._terminate_instances(instance_manager) assert [e.new_instance_status for e in subscriber.events_by_id("stale")] == [ Instance.TERMINATED ] assert all( e.new_instance_status != Instance.TERMINATING for e in subscriber.events ) instances, _ = instance_storage.get_instances() assert instances["stale"].status == Instance.TERMINATED assert instances["healthy"].status == Instance.RAY_RUNNING @staticmethod @mock.patch("time.time_ns") def test_ray_stop_requested_fail(mock_time_ns, setup): # Test that the instance should be transitioned to RAY_RUNNING # when the ray stop request fails. instance_manager, instance_storage, _, cloud_resource_monitor = setup mock_time_ns.return_value = 10 * s_to_ns instances = [ create_instance( "i-1", status=Instance.RAY_STOP_REQUESTED, ray_node_id=binary_to_hex(b"r-1"), cloud_instance_id="c-1", status_times=[(Instance.RAY_STOP_REQUESTED, 10 * s_to_ns)], ), create_instance( "i-2", status=Instance.RAY_STOP_REQUESTED, ray_node_id=binary_to_hex(b"r-2"), cloud_instance_id="c-2", status_times=[(Instance.RAY_STOP_REQUESTED, 10 * s_to_ns)], ), ] ray_nodes = [ NodeState(node_id=b"r-1", status=NodeStatus.RUNNING, instance_id="c-1"), NodeState(node_id=b"r-2", status=NodeStatus.RUNNING, instance_id="c-2"), ] ray_stop_errors = [ RayStopError(im_instance_id="i-1"), ] cloud_instances = { "c-1": CloudInstance("c-1", "type-1", True, NodeKind.WORKER), "c-2": CloudInstance("c-2", "type-2", True, NodeKind.WORKER), } TestReconciler._add_instances(instance_storage, instances) Reconciler.reconcile( instance_manager=instance_manager, scheduler=MockScheduler(), cloud_provider=MagicMock(), cloud_resource_monitor=cloud_resource_monitor, ray_cluster_resource_state=ClusterResourceState(node_states=ray_nodes), non_terminated_cloud_instances=cloud_instances, cloud_provider_errors=[], ray_install_errors=[], ray_stop_errors=ray_stop_errors, autoscaling_config=MockAutoscalingConfig(), ) instances, _ = instance_storage.get_instances() assert len(instances) == 2 assert instances["i-1"].status == Instance.RAY_RUNNING assert instances["i-2"].status == Instance.RAY_STOP_REQUESTED @staticmethod @mock.patch("time.time_ns") @pytest.mark.parametrize( "cur_status,expect_status", [ (Instance.ALLOCATED, Instance.TERMINATING), (Instance.RAY_INSTALLING, Instance.TERMINATING), (Instance.TERMINATING, Instance.TERMINATING), ], ) def test_stuck_instances(mock_time_ns, cur_status, expect_status, setup): instance_manager, instance_storage, subscriber, cloud_resource_monitor = setup timeout_s = 5 cur_time_s = 20 mock_time_ns.return_value = cur_time_s * s_to_ns config = InstanceReconcileConfig( allocate_status_timeout_s=timeout_s, terminating_status_timeout_s=timeout_s, ray_install_status_timeout_s=timeout_s, ) instances = [ create_instance( "no-update", cur_status, status_times=[(cur_status, (cur_time_s - timeout_s + 1) * s_to_ns)], instance_type="type-1", cloud_instance_id="c-1", ), create_instance( "updated", cur_status, status_times=[(cur_status, (cur_time_s - timeout_s - 1) * s_to_ns)], instance_type="type-1", cloud_instance_id="c-2", ), ] cloud_instances = { "c-1": CloudInstance("c-1", "type-1", True, NodeKind.WORKER), "c-2": CloudInstance("c-2", "type-1", True, NodeKind.WORKER), } TestReconciler._add_instances(instance_storage, instances) Reconciler.reconcile( instance_manager=instance_manager, scheduler=MockScheduler(), cloud_provider=MagicMock(), cloud_resource_monitor=cloud_resource_monitor, ray_cluster_resource_state=ClusterResourceState(), non_terminated_cloud_instances=cloud_instances, cloud_provider_errors=[], ray_install_errors=[], autoscaling_config=MockAutoscalingConfig( configs={ "instance_reconcile_config": config, "max_concurrent_launches": 0, # prevent launches } ), ) instances, _ = instance_storage.get_instances() assert instances["no-update"].status == cur_status assert instances["updated"].status == expect_status @staticmethod @mock.patch("time.time_ns") @pytest.mark.parametrize( "status", [ Instance.InstanceStatus.Name(Instance.RAY_STOPPING), Instance.InstanceStatus.Name(Instance.QUEUED), ], ) def test_warn_stuck_transient_instances(mock_time_ns, status, setup): instance_manager, instance_storage, subscriber, cloud_resource_monitor = setup cur_time_s = 10 mock_time_ns.return_value = cur_time_s * s_to_ns timeout_s = 5 status = Instance.InstanceStatus.Value(status) config = InstanceReconcileConfig( transient_status_warn_interval_s=timeout_s, ) # TODO: we are only mocking the necessary fields for testing. # Probably better to have a more comprehensive mocker for instance # of various statuses. instances = [ create_instance( "no-warn", status, status_times=[(status, (cur_time_s - timeout_s + 1) * s_to_ns)], cloud_instance_id="c-1" if status != Instance.QUEUED else "", instance_type="type-1", ), create_instance( "warn", status, status_times=[(status, (cur_time_s - timeout_s - 1) * s_to_ns)], cloud_instance_id="c-2" if status != Instance.QUEUED else "", instance_type="type-1", ), ] TestReconciler._add_instances(instance_storage, instances) mock_logger = mock.MagicMock() cloud_instances = { "c-1": CloudInstance("c-1", "type-1", True, NodeKind.WORKER), "c-2": CloudInstance("c-2", "type-1", True, NodeKind.WORKER), } Reconciler.reconcile( instance_manager=instance_manager, scheduler=MockScheduler(), cloud_provider=MagicMock(), cloud_resource_monitor=cloud_resource_monitor, ray_cluster_resource_state=ClusterResourceState(), non_terminated_cloud_instances=cloud_instances if status != Instance.QUEUED else {}, cloud_provider_errors=[], ray_install_errors=[], autoscaling_config=MockAutoscalingConfig( configs={ "instance_reconcile_config": config, "max_concurrent_launches": 0, # prevent launches } ), _logger=mock_logger, ) assert mock_logger.warning.call_count == 1 @staticmethod @mock.patch("time.time_ns") def test_stuck_instances_no_op(mock_time_ns, setup): instance_manager, instance_storage, subscriber, cloud_resource_monitor = setup # Large enough to not trigger any timeouts mock_time_ns.return_value = 999999 * s_to_ns config = InstanceReconcileConfig() all_status = set(Instance.InstanceStatus.values()) reconciled_stuck_statuses = { Instance.REQUESTED, Instance.ALLOCATED, Instance.RAY_INSTALLING, Instance.TERMINATING, Instance.RAY_STOP_REQUESTED, } transient_statuses = { Instance.RAY_STOPPING, Instance.RAY_INSTALL_FAILED, Instance.RAY_STOPPED, Instance.TERMINATION_FAILED, Instance.QUEUED, Instance.ALLOCATION_TIMEOUT, } no_op_statuses = all_status - reconciled_stuck_statuses - transient_statuses for status in no_op_statuses: instances = [ create_instance( f"no-op-{status}", status, status_times=[(status, 1 * s_to_ns)], ), ] TestReconciler._add_instances(instance_storage, instances) subscriber.clear() mock_logger = mock.MagicMock() Reconciler.reconcile( instance_manager=instance_manager, scheduler=MockScheduler(), cloud_provider=MagicMock(), cloud_resource_monitor=cloud_resource_monitor, ray_cluster_resource_state=ClusterResourceState(), non_terminated_cloud_instances={}, cloud_provider_errors=[], ray_install_errors=[], autoscaling_config=MockAutoscalingConfig( configs={ "instance_reconcile_config": config, "max_concurrent_launches": 0, # prevent launches } ), _logger=mock_logger, ) assert subscriber.events == [] assert mock_logger.warning.call_count == 0 @staticmethod @pytest.mark.parametrize( "status,expected_running", [(Instance.RAY_RUNNING, True), (Instance.ALLOCATED, False)], ids=["ray_running", "allocated"], ) def test_is_head_node_running(status, expected_running, setup): instance_manager, instance_storage, subscriber, cloud_resource_monitor = setup instances = [ create_instance( "i-1", status=status, cloud_instance_id="c-1", node_kind=NodeKind.HEAD, ), ] TestReconciler._add_instances(instance_storage, instances) assert Reconciler._is_head_node_running(instance_manager) is expected_running @staticmethod def test_scaling_updates(setup): """ Tests that new instances should be launched due to autoscaling decisions, and existing instances should be terminated if needed. """ instance_manager, instance_storage, _, cloud_resource_monitor = setup im_instances = [ create_instance( "head", status=Instance.RAY_RUNNING, cloud_instance_id="c-0", ray_node_id=binary_to_hex(b"r-0"), node_kind=NodeKind.HEAD, ), create_instance( "i-1", status=Instance.RAY_RUNNING, cloud_instance_id="c-1", ray_node_id=binary_to_hex(b"r-1"), node_kind=NodeKind.WORKER, ), # To be reconciled. ] TestReconciler._add_instances(instance_storage, im_instances) ray_nodes = [ NodeState(node_id=b"r-0", status=NodeStatus.RUNNING, instance_id="c-1"), NodeState(node_id=b"r-1", status=NodeStatus.RUNNING, instance_id="c-1"), ] cloud_instances = { "c-0": CloudInstance("c-0", "head", True, NodeKind.HEAD), "c-1": CloudInstance("c-1", "type-1", True, NodeKind.WORKER), } mock_scheduler = MagicMock() mock_scheduler.schedule.return_value = SchedulingReply( to_launch=[ LaunchRequest(instance_type="type-1", count=2), ], to_terminate=[ TerminationRequest( id="t1", ray_node_id="r-1", instance_id="i-1", cause=TerminationRequest.Cause.IDLE, idle_duration_ms=1000, ) ], infeasible_gang_resource_requests=[ GangResourceRequest( requests=[ResourceRequest(resources_bundle={"CPU": 1})] ) ], ) state = Reconciler.reconcile( instance_manager=instance_manager, scheduler=mock_scheduler, cloud_provider=MagicMock(), cloud_resource_monitor=cloud_resource_monitor, ray_cluster_resource_state=ClusterResourceState(node_states=ray_nodes), non_terminated_cloud_instances=cloud_instances, cloud_provider_errors=[], ray_install_errors=[], autoscaling_config=MockAutoscalingConfig( configs={ "max_concurrent_launches": 0, # don't launch anything. } ), ) instances, _ = instance_storage.get_instances() assert len(instances) == 3 + 1 # for head node for id, instance in instances.items(): if id == "head": assert instance.status == Instance.RAY_RUNNING elif id == "i-1": assert instance.status == Instance.RAY_STOP_REQUESTED else: assert instance.status == Instance.QUEUED assert instance.instance_type == "type-1" assert len(state.infeasible_gang_resource_requests) == 1 @staticmethod def test_terminating_instances(setup): instance_manager, instance_storage, subscriber, cloud_resource_monitor = setup instances = [ create_instance( "i-1", status=Instance.RAY_STOPPED, cloud_instance_id="c-1", ), create_instance( "i-2", status=Instance.RAY_INSTALL_FAILED, cloud_instance_id="c-2", ), create_instance( "i-3", status=Instance.TERMINATION_FAILED, cloud_instance_id="c-3", ), ] cloud_instances = { "c-1": CloudInstance("c-1", "type-1", True, NodeKind.WORKER), "c-2": CloudInstance("c-2", "type-2", True, NodeKind.WORKER), "c-3": CloudInstance("c-3", "type-3", True, NodeKind.WORKER), } TestReconciler._add_instances(instance_storage, instances) Reconciler.reconcile( instance_manager=instance_manager, scheduler=MockScheduler(), cloud_provider=MagicMock(), cloud_resource_monitor=cloud_resource_monitor, ray_cluster_resource_state=ClusterResourceState(), non_terminated_cloud_instances=cloud_instances, cloud_provider_errors=[], ray_install_errors=[], autoscaling_config=MockAutoscalingConfig(), ) instances, _ = instance_storage.get_instances() assert instances["i-1"].status == Instance.TERMINATING assert instances["i-2"].status == Instance.TERMINATING assert instances["i-3"].status == Instance.TERMINATING @staticmethod @pytest.mark.parametrize( "disable_node_updaters", [True, False], ) @pytest.mark.parametrize( "cloud_instance_running", [True, False], ) def test_ray_install(disable_node_updaters, cloud_instance_running, setup): instance_manager, instance_storage, _, cloud_resource_monitor = setup instances = [ create_instance( "i-1", status=Instance.ALLOCATED, instance_type="type-1", launch_request_id="l1", cloud_instance_id="c-1", ), ] cloud_instances = { "c-1": CloudInstance( "c-1", "type-1", "", cloud_instance_running, NodeKind.WORKER ), } TestReconciler._add_instances(instance_storage, instances) Reconciler.reconcile( instance_manager=instance_manager, scheduler=MockScheduler(), cloud_provider=MagicMock(), cloud_resource_monitor=cloud_resource_monitor, ray_cluster_resource_state=ClusterResourceState(), non_terminated_cloud_instances=cloud_instances, cloud_provider_errors=[], ray_install_errors=[], autoscaling_config=MockAutoscalingConfig( configs={ "disable_node_updaters": disable_node_updaters, } ), ) instances, _ = instance_storage.get_instances() if disable_node_updaters or not cloud_instance_running: assert instances["i-1"].status == Instance.ALLOCATED else: assert instances["i-1"].status == Instance.RAY_INSTALLING @staticmethod @mock.patch("time.time_ns") def test_autoscaler_state(mock_time_ns, setup): instance_manager, instance_storage, _, cloud_resource_monitor = setup mock_time_ns.return_value = 5 instances = [ create_instance( "head", status=Instance.RAY_RUNNING, cloud_instance_id="c-0", ray_node_id=binary_to_hex(b"r-0"), node_kind=NodeKind.HEAD, ), create_instance( "i-1", status=Instance.ALLOCATED, instance_type="type-1", cloud_instance_id="c-1", launch_request_id="l1", status_times=[ (Instance.QUEUED, 0), (Instance.REQUESTED, 1), (Instance.ALLOCATED, 2), ], ), # requested instance create_instance( "i-2", status=Instance.REQUESTED, instance_type="type-2", launch_request_id="l2", status_times=[(Instance.QUEUED, 0), (Instance.REQUESTED, 1)], ), # queued instance create_instance( "i-3", status=Instance.QUEUED, instance_type="type-3", launch_request_id="l3", status_times=[ (Instance.QUEUED, 0), ], ), # allocation failed create_instance( "i-4", status=Instance.ALLOCATION_FAILED, instance_type="type-4", launch_request_id="l4", status_times=[ (Instance.QUEUED, 0), (Instance.REQUESTED, 1), (Instance.ALLOCATION_FAILED, 2), ], ), # ray installing create_instance( "i-5", status=Instance.RAY_INSTALLING, instance_type="type-5", launch_request_id="l5", cloud_instance_id="c-5", status_times=[ (Instance.QUEUED, 0), (Instance.REQUESTED, 1), (Instance.ALLOCATED, 2), (Instance.RAY_INSTALLING, 3), ], ), ] cloud_instances = { "c-0": CloudInstance("c-0", "head", True, NodeKind.HEAD), "c-1": CloudInstance("c-1", "type-1", True, NodeKind.WORKER), "c-5": CloudInstance("c-5", "type-5", True, NodeKind.WORKER), } TestReconciler._add_instances(instance_storage, instances) mock_scheduler = MagicMock() mock_scheduler.schedule.return_value = SchedulingReply( to_launch=[], to_terminate=[], infeasible_gang_resource_requests=[ GangResourceRequest( requests=[ResourceRequest(resources_bundle={"CPU": 1})] ) ], infeasible_resource_requests=[ResourceRequest(resources_bundle={"CPU": 1})], ) autoscaling_state = Reconciler.reconcile( instance_manager=instance_manager, scheduler=mock_scheduler, cloud_provider=MagicMock(), cloud_resource_monitor=cloud_resource_monitor, ray_cluster_resource_state=ClusterResourceState( cluster_resource_state_version=1, ), non_terminated_cloud_instances=cloud_instances, cloud_provider_errors=[], ray_install_errors=[], autoscaling_config=MockAutoscalingConfig( configs={ "max_concurrent_launches": 0, # don't launch anything. "provider_instance_types": { "type-1": "m5.large", "type-2": "m5.xlarge", "type-3": "c5.large", "type-4": "g5.xlarge", "type-5": "r5.large", }, } ), ) assert autoscaling_state.last_seen_cluster_resource_state_version == 1 assert len(autoscaling_state.infeasible_gang_resource_requests) == 1 assert len(autoscaling_state.infeasible_resource_requests) == 1 assert len(autoscaling_state.pending_instances) == 2 pending_instances = { (i.instance_id, i.instance_type_name, i.ray_node_type_name) for i in autoscaling_state.pending_instances } assert pending_instances == { ("i-1", "m5.large", "type-1"), ("i-5", "r5.large", "type-5"), } pending_instance_requests = defaultdict(int) for r in autoscaling_state.pending_instance_requests: pending_instance_requests[ (r.instance_type_name, r.ray_node_type_name) ] += r.count failed_instance_requests = defaultdict(int) for r in autoscaling_state.failed_instance_requests: failed_instance_requests[ (r.instance_type_name, r.ray_node_type_name) ] += r.count assert pending_instance_requests == { ("m5.xlarge", "type-2"): 1, ("c5.large", "type-3"): 1, } assert failed_instance_requests == {("g5.xlarge", "type-4"): 1} @staticmethod def test_extra_cloud_instances_cloud_provider(setup): """ Test that extra cloud instances should be terminated. """ instance_manager, instance_storage, subscriber, cloud_resource_monitor = setup im_instances = [ create_instance( "i-1", status=Instance.RAY_RUNNING, cloud_instance_id="c-1" ), # To be reconciled. ] TestReconciler._add_instances(instance_storage, im_instances) ray_nodes = [ NodeState(node_id=b"r-1", status=NodeStatus.RUNNING, instance_id="c-1"), # Out of band ray nodes. NodeState( node_id=b"r-2", status=NodeStatus.RUNNING, instance_id="c-3", ray_node_type_name="type-1", ), ] cloud_instances = { "c-1": CloudInstance("c-1", "type-1", True, NodeKind.WORKER), # Out of band cloud instance. "c-2": CloudInstance("c-2", "type-2", True, NodeKind.WORKER), } subscriber.clear() Reconciler.reconcile( instance_manager, scheduler=MockScheduler(), cloud_provider=MagicMock(), cloud_resource_monitor=cloud_resource_monitor, ray_cluster_resource_state=ClusterResourceState(node_states=ray_nodes), non_terminated_cloud_instances=cloud_instances, cloud_provider_errors=[], ray_install_errors=[], autoscaling_config=MockAutoscalingConfig(), ) events = subscriber.events for e in events: if e.new_instance_status == Instance.ALLOCATED: assert e.cloud_instance_id in {"c-2", "c-3"} else: assert e.new_instance_status == Instance.RAY_RUNNING assert e.ray_node_id == binary_to_hex(b"r-2") instances, _ = instance_storage.get_instances() assert len(instances) == 3 statuses = {instance.status for instance in instances.values()} assert statuses == {Instance.RAY_RUNNING, Instance.ALLOCATED} @staticmethod def test_cloud_instance_reboot(setup): """ Test that the case of booting up a previous stopped cloud instance. """ instance_manager, instance_storage, subscriber, cloud_resource_monitor = setup im_instances = [ create_instance( "i-1", status=Instance.TERMINATED, cloud_instance_id="c-1", ray_node_id=binary_to_hex(b"r-1"), ), ] TestReconciler._add_instances(instance_storage, im_instances) ray_nodes = [ NodeState( node_id=b"r-1", status=NodeStatus.DEAD, instance_id="c-1", ray_node_type_name="type-1", ), ] cloud_instances = { "c-1": CloudInstance("c-1", "type-1", True, NodeKind.WORKER), } subscriber.clear() Reconciler.reconcile( instance_manager, scheduler=MockScheduler(), cloud_provider=MagicMock(), cloud_resource_monitor=cloud_resource_monitor, ray_cluster_resource_state=ClusterResourceState(node_states=ray_nodes), non_terminated_cloud_instances=cloud_instances, cloud_provider_errors=[], ray_install_errors=[], autoscaling_config=MockAutoscalingConfig(), ) events = subscriber.events for e in events: assert e.new_instance_status == Instance.ALLOCATED assert e.cloud_instance_id == "c-1" instances, _ = instance_storage.get_instances() assert len(instances) == 2 statuses = {instance.status for instance in instances.values()} assert statuses == {Instance.ALLOCATED, Instance.TERMINATED} @staticmethod def test_ray_node_restarted_on_the_same_cloud_instance(setup): """ Test that the case of reusing cloud instances. """ instance_manager, instance_storage, subscriber, cloud_resource_monitor = setup im_instances = [ create_instance( "i-1", status=Instance.RAY_RUNNING, cloud_instance_id="c-1", ray_node_id=binary_to_hex(b"r-1"), ), ] TestReconciler._add_instances(instance_storage, im_instances) ray_nodes = [ NodeState( node_id=b"r-2", status=NodeStatus.IDLE, instance_id="c-1", ray_node_type_name="type-1", ), NodeState( node_id=b"r-1", status=NodeStatus.DEAD, instance_id="c-1", ray_node_type_name="type-1", ), ] cloud_instances = { "c-1": CloudInstance("c-1", "type-1", True, NodeKind.WORKER), } subscriber.clear() Reconciler.reconcile( instance_manager, scheduler=MockScheduler(), cloud_provider=MagicMock(), cloud_resource_monitor=cloud_resource_monitor, ray_cluster_resource_state=ClusterResourceState(node_states=ray_nodes), non_terminated_cloud_instances=cloud_instances, cloud_provider_errors=[], ray_install_errors=[], autoscaling_config=MockAutoscalingConfig(), ) events = subscriber.events assert len(events) == 4 assert events[0].new_instance_status == Instance.ALLOCATED assert events[0].cloud_instance_id == "c-1" assert events[0].ray_node_id == binary_to_hex(b"r-2") assert events[1].new_instance_status == Instance.RAY_RUNNING assert events[1].instance_id == events[0].instance_id assert events[1].ray_node_id == binary_to_hex(b"r-2") assert events[2].new_instance_status == Instance.RAY_STOPPED assert events[2].instance_id == "i-1" assert events[2].ray_node_id == binary_to_hex(b"r-1") # c-1 has been reused by the new ray node r-2 (now RAY_RUNNING), so the # stale i-1 row must not terminate c-1. It is marked TERMINATED to clean # up the record without issuing a cloud termination request. assert events[3].new_instance_status == Instance.TERMINATED assert events[3].instance_id == "i-1" instances, _ = instance_storage.get_instances() assert len(instances) == 2 statuses = {instance.status for instance in instances.values()} assert statuses == {Instance.RAY_RUNNING, Instance.TERMINATED} @staticmethod def test_ray_head_restarted_on_the_same_cloud_instance(setup): """ Test that the case of restarting Head node with GCS FT. """ instance_manager, instance_storage, subscriber, cloud_resource_monitor = setup ray_nodes = [ NodeState( node_id=b"r-2", status=NodeStatus.IDLE, instance_id="c-1", ray_node_type_name="type-1", ), NodeState( node_id=b"r-1", status=NodeStatus.DEAD, instance_id="c-1", ray_node_type_name="type-1", ), ] cloud_instances = { "c-1": CloudInstance("c-1", "type-1", True, NodeKind.HEAD), } subscriber.clear() Reconciler.reconcile( instance_manager, scheduler=MockScheduler(), cloud_provider=MagicMock(), cloud_resource_monitor=cloud_resource_monitor, ray_cluster_resource_state=ClusterResourceState(node_states=ray_nodes), non_terminated_cloud_instances=cloud_instances, cloud_provider_errors=[], ray_install_errors=[], autoscaling_config=MockAutoscalingConfig(), ) events = subscriber.events assert len(events) == 5 assert events[0].new_instance_status == Instance.ALLOCATED assert events[0].cloud_instance_id == "c-1" assert events[0].ray_node_id == binary_to_hex(b"r-2") assert events[1].new_instance_status == Instance.ALLOCATED assert events[1].cloud_instance_id == "c-1" assert events[1].ray_node_id == binary_to_hex(b"r-1") assert events[1].instance_id != events[0].instance_id assert events[2].new_instance_status == Instance.RAY_RUNNING assert events[2].instance_id == events[0].instance_id assert events[2].ray_node_id == binary_to_hex(b"r-2") assert events[3].new_instance_status == Instance.RAY_STOPPED assert events[3].instance_id == events[1].instance_id assert events[3].ray_node_id == binary_to_hex(b"r-1") # c-1 has been reused by the new ray node r-2 (now RAY_RUNNING), so the # stale row must not terminate c-1. It is marked TERMINATED to clean up # the record without issuing a cloud termination request. assert events[4].new_instance_status == Instance.TERMINATED assert events[4].instance_id == events[1].instance_id instances, _ = instance_storage.get_instances() assert len(instances) == 2 statuses = {instance.status for instance in instances.values()} assert statuses == {Instance.RAY_RUNNING, Instance.TERMINATED} @staticmethod def test_reconcile_max_worker_nodes_limit_triggers_termination(setup): instance_manager, instance_storage, _, cloud_resource_monitor = setup instances = [ create_instance( "head", status=Instance.RAY_RUNNING, node_kind=NodeKind.HEAD, cloud_instance_id="c-head", ray_node_id=binary_to_hex(b"r-head"), ), create_instance( "i-0", status=Instance.ALLOCATED, instance_type="type-1", cloud_instance_id="c-0", ray_node_id=binary_to_hex(b"r-0"), ), create_instance( "i-1", status=Instance.ALLOCATED, instance_type="type-1", cloud_instance_id="c-1", ray_node_id=binary_to_hex(b"r-1"), ), ] TestReconciler._add_instances(instance_storage, instances) # Empty list of Ray nodes - i.e. when instances are pending but not scheduled ray_nodes = [] # Cloud instances corresponding to the 3 IM instances cloud_instances = { "c-head": CloudInstance("c-head", "head", True, NodeKind.HEAD), "c-0": CloudInstance("c-0", "type-1", True, NodeKind.WORKER), "c-1": CloudInstance("c-1", "type-1", True, NodeKind.WORKER), } # Scheduler should add both workers to to_terminate due to max nodes mock_scheduler = MockScheduler( to_launch=[], to_terminate=[ TerminationRequest( id="t0", ray_node_id="r-0", instance_id="i-0", instance_status=Instance.ALLOCATED, cause=TerminationRequest.Cause.MAX_NUM_NODE_PER_TYPE, ), TerminationRequest( id="t1", ray_node_id="r-1", instance_id="i-1", instance_status=Instance.ALLOCATED, cause=TerminationRequest.Cause.MAX_NUM_NODE_PER_TYPE, ), ], ) Reconciler.reconcile( instance_manager=instance_manager, scheduler=mock_scheduler, cloud_provider=MagicMock(), cloud_resource_monitor=cloud_resource_monitor, ray_cluster_resource_state=ClusterResourceState( node_states=ray_nodes, cluster_resource_state_version=1, ), non_terminated_cloud_instances=cloud_instances, cloud_provider_errors=[], ray_install_errors=[], autoscaling_config=MockAutoscalingConfig( configs={ "node_type_configs": { "type-1": { "name": "type-1", "resources": {"CPU": 1}, "min_worker_nodes": 0, "max_worker_nodes": 0, } }, } ), ) instances, _ = instance_storage.get_instances() assert instances["i-0"].status == Instance.TERMINATING assert instances["i-1"].status == Instance.TERMINATING @staticmethod def test_terminate_queued_instances_max_workers(setup): instance_manager, instance_storage, _, cloud_resource_monitor = setup instances = [ create_instance( "head", status=Instance.RAY_RUNNING, node_kind=NodeKind.HEAD, cloud_instance_id="c-head", ray_node_id=binary_to_hex(b"r-head"), ), create_instance( "i-0", status=Instance.QUEUED, instance_type="type-1", ), create_instance( "i-1", status=Instance.QUEUED, instance_type="type-1", ), ] TestReconciler._add_instances(instance_storage, instances) # Empty list of Ray nodes - QUEUED instances have not launched ray yet ray_nodes = [] # Cloud instance for head node only cloud_instances = { "c-head": CloudInstance("c-head", "head", True, NodeKind.HEAD), } # Scheduler should terminate QUEUED instances due to max_workers limit mock_scheduler = MockScheduler( to_launch=[], to_terminate=[ TerminationRequest( id="t0", instance_id="i-0", instance_status=Instance.QUEUED, cause=TerminationRequest.Cause.MAX_NUM_NODE_PER_TYPE, ), TerminationRequest( id="t1", instance_id="i-1", instance_status=Instance.QUEUED, cause=TerminationRequest.Cause.MAX_NUM_NODE_PER_TYPE, ), ], ) Reconciler.reconcile( instance_manager=instance_manager, scheduler=mock_scheduler, cloud_provider=MagicMock(), cloud_resource_monitor=cloud_resource_monitor, ray_cluster_resource_state=ClusterResourceState( node_states=ray_nodes, cluster_resource_state_version=1, ), non_terminated_cloud_instances=cloud_instances, cloud_provider_errors=[], ray_install_errors=[], autoscaling_config=MockAutoscalingConfig( configs={ "node_type_configs": { "type-1": { "name": "type-1", "resources": {"CPU": 1}, "min_worker_nodes": 0, "max_worker_nodes": 0, } }, } ), ) instances, _ = instance_storage.get_instances() assert instances["i-0"].status == Instance.TERMINATED assert instances["i-1"].status == Instance.TERMINATED @staticmethod def test_terminate_ray_installing_instances_max_workers(setup): instance_manager, instance_storage, _, cloud_resource_monitor = setup instances = [ create_instance( "head", status=Instance.RAY_RUNNING, node_kind=NodeKind.HEAD, cloud_instance_id="c-head", ray_node_id=binary_to_hex(b"r-head"), ), create_instance( "i-0", status=Instance.RAY_INSTALLING, instance_type="type-1", cloud_instance_id="c-0", ), create_instance( "i-1", status=Instance.RAY_INSTALLING, instance_type="type-1", cloud_instance_id="c-1", ), ] TestReconciler._add_instances(instance_storage, instances) # Empty list of Ray nodes - RAY_INSTALLING instances have not started ray yet ray_nodes = [] # Cloud instances for head and installing workers cloud_instances = { "c-head": CloudInstance("c-head", "head", True, NodeKind.HEAD), "c-0": CloudInstance("c-0", "type-1", True, NodeKind.WORKER), "c-1": CloudInstance("c-1", "type-1", True, NodeKind.WORKER), } # Scheduler should terminate RAY_INSTALLING instances due to max_workers limit mock_scheduler = MockScheduler( to_launch=[], to_terminate=[ TerminationRequest( id="t0", instance_id="i-0", instance_status=Instance.RAY_INSTALLING, cause=TerminationRequest.Cause.MAX_NUM_NODE_PER_TYPE, ), TerminationRequest( id="t1", instance_id="i-1", instance_status=Instance.RAY_INSTALLING, cause=TerminationRequest.Cause.MAX_NUM_NODE_PER_TYPE, ), ], ) Reconciler.reconcile( instance_manager=instance_manager, scheduler=mock_scheduler, cloud_provider=MagicMock(), cloud_resource_monitor=cloud_resource_monitor, ray_cluster_resource_state=ClusterResourceState( node_states=ray_nodes, cluster_resource_state_version=1, ), non_terminated_cloud_instances=cloud_instances, cloud_provider_errors=[], ray_install_errors=[], autoscaling_config=MockAutoscalingConfig( configs={ "node_type_configs": { "type-1": { "name": "type-1", "resources": {"CPU": 1}, "min_worker_nodes": 0, "max_worker_nodes": 0, } }, } ), ) instances, _ = instance_storage.get_instances() assert instances["i-0"].status == Instance.TERMINATING assert instances["i-1"].status == Instance.TERMINATING if __name__ == "__main__": if os.environ.get("PARALLEL_CI"): sys.exit(pytest.main(["-n", "auto", "--boxed", "-vs", __file__])) else: sys.exit(pytest.main(["-sv", __file__]))