[Autoscaler][V2] Check IM instance_status before terminating nodes

python/ray/autoscaler/v2/instance_manager/reconciler.py

@@ -1143,9 +1143,14 @@ def _scale_cluster(
         # Add terminating instances.
         for terminate_request in to_terminate:
             instance_id = terminate_request.instance_id
+            new_instance_status = IMInstance.RAY_STOP_REQUESTED
+            # The instance is not yet running, so we can't request to stop/drain Ray.
+            # Therefore, we can skip the RAY_STOP_REQUESTED state and directly terminate the node.
+            if terminate_request.instance_status == IMInstance.ALLOCATED:
+                new_instance_status = IMInstance.TERMINATING
             updates[terminate_request.instance_id] = IMInstanceUpdateEvent(
                 instance_id=instance_id,
-                new_instance_status=IMInstance.RAY_STOP_REQUESTED,
+                new_instance_status=new_instance_status,
                 termination_request=terminate_request,
                 details=f"draining ray: {terminate_request.details}",
             )

python/ray/autoscaler/v2/scheduler.py

@@ -169,6 +169,9 @@ class SchedulingNode:
     # The instance id of the IM(Instance Manager) instance. None if the node
     # is not yet in IM.
     im_instance_id: Optional[str] = None
+    # The instance status of the IM(Instance Manager) instance. None if the node
+    # is not yet in IM.
+    im_instance_status: Optional[Instance.InstanceStatus.ValueType] = None
     # The ray node id of the ray node. None if the node is not included in
     # ray cluster's GCS report yet (not running ray yet).
     ray_node_id: Optional[str] = None
@@ -187,6 +190,7 @@ def __init__(
         labels: Dict[str, str],
         status: SchedulingNodeStatus,
         im_instance_id: str = "",
+        im_instance_status: Optional[Instance.InstanceStatus.ValueType] = None,
         ray_node_id: str = "",
         idle_duration_ms: int = 0,
         launch_config_hash: str = "",
@@ -206,6 +210,7 @@ def __init__(
         self.labels = labels
         self.status = status
         self.im_instance_id = im_instance_id
+        self.im_instance_status = im_instance_status
         self.ray_node_id = ray_node_id
         self.idle_duration_ms = idle_duration_ms
         self.launch_config_hash = launch_config_hash
@@ -274,6 +279,7 @@ def new(
                 labels=dict(instance.ray_node.dynamic_labels),
                 status=SchedulingNodeStatus.SCHEDULABLE,
                 im_instance_id=instance.im_instance.instance_id,
+                im_instance_status=instance.im_instance.status,
                 ray_node_id=instance.im_instance.node_id,
                 idle_duration_ms=instance.ray_node.idle_duration_ms,
                 launch_config_hash=instance.im_instance.launch_config_hash,
@@ -303,9 +309,11 @@ def new(
                 labels={},
                 status=SchedulingNodeStatus.TO_TERMINATE,
                 im_instance_id=instance.im_instance.instance_id,
+                im_instance_status=instance.im_instance.status,
                 termination_request=TerminationRequest(
                     id=str(uuid.uuid4()),
                     instance_id=instance.im_instance.instance_id,
+                    instance_status=instance.im_instance.status,
                     cause=TerminationRequest.Cause.OUTDATED,
                     instance_type=instance.im_instance.instance_type,
                 ),
@@ -1074,6 +1082,7 @@ def _select_nodes_to_terminate(
                 ray_node_id=node.ray_node_id,
                 cause=cause,
                 instance_type=node.node_type,
+                instance_status=node.im_instance_status,
                 details=(
                     f"Terminating node due to {TerminationRequest.Cause.Name(cause)}: "
                     f"max_num_nodes={max_num_nodes}, "
@@ -1550,6 +1559,7 @@ def _terminate_outdated_nodes(
                     instance_id=node.im_instance_id,
                     ray_node_id=node.ray_node_id,
                     instance_type=node.node_type,
+                    instance_status=node.im_instance_status,
                     cause=TerminationRequest.Cause.OUTDATED,
                     details=f"node from {node.node_type} has outdated config",
                 )

src/ray/protobuf/experimental/instance_manager.proto

@@ -71,7 +71,7 @@ message Instance {
     RAY_STOPPING = 7;
     // Ray stopped - follows from the RAY_STOPPING/RAY_RUNNING state.
     RAY_STOPPED = 8;
-    // The instance is terminating - follows from the RAY_STOPPED state.
+    // The instance is terminating - follows from the RAY_STOPPED or ALLOCATED state.
     TERMINATING = 9;
     // The instance is terminated - follows from TERMINATING state or any other running
     // states when instance was preempted.
@@ -202,6 +202,8 @@ message TerminationRequest {
   optional uint32 max_num_nodes = 8;
   // Details of the termination.
   string details = 9;
+  // The last instance status of the instance that's terminated.
+  optional Instance.InstanceStatus instance_status = 10;
 }
 
 message UpdateInstanceManagerStateRequest {