[core][autoscaler] Fix incorrectly terminating nodes misclassified as idle in autoscaler v1

python/ray/autoscaler/_private/autoscaler.py

@@ -490,8 +490,11 @@ def terminate_nodes_to_enforce_config_constraints(self, now: float):
         assert self.non_terminated_nodes
         assert self.provider
 
-        last_used = self.load_metrics.last_used_time_by_ip
-        horizon = now - (60 * self.config["idle_timeout_minutes"])
+        last_used = self.load_metrics.ray_nodes_last_used_time_by_ip
+
+        idle_timeout_s = 60 * self.config["idle_timeout_minutes"]
+
+        last_used_cutoff = now - idle_timeout_s
 
         # Sort based on last used to make sure to keep min_workers that
         # were most recently used. Otherwise, _keep_min_workers_of_node_type
@@ -539,7 +542,8 @@ def keep_node(node_id: NodeID) -> None:
                 continue
 
             node_ip = self.provider.internal_ip(node_id)
-            if node_ip in last_used and last_used[node_ip] < horizon:
+
+            if node_ip in last_used and last_used[node_ip] < last_used_cutoff:
                 self.schedule_node_termination(node_id, "idle", logger.info)
                 # Get the local time of the node's last use as a string.
                 formatted_last_used_time = time.asctime(

python/ray/autoscaler/_private/load_metrics.py

@@ -70,7 +70,6 @@ class LoadMetrics:
     """
 
     def __init__(self):
-        self.last_used_time_by_ip = {}
         self.last_heartbeat_time_by_ip = {}
         self.static_resources_by_ip = {}
         self.dynamic_resources_by_ip = {}
@@ -80,6 +79,7 @@ def __init__(self):
         self.pending_placement_groups = []
         self.resource_requests = []
         self.cluster_full_of_actors_detected = False
+        self.ray_nodes_last_used_time_by_ip = {}
 
     def __bool__(self):
         """A load metrics instance is Falsey iff the autoscaler process
@@ -93,6 +93,7 @@ def update(
         raylet_id: bytes,
         static_resources: Dict[str, Dict],
         dynamic_resources: Dict[str, Dict],
+        node_idle_duration_s: float,
         waiting_bundles: List[Dict[str, float]] = None,
         infeasible_bundles: List[Dict[str, float]] = None,
         pending_placement_groups: List[PlacementGroupTableData] = None,
@@ -120,11 +121,7 @@ def update(
         self.dynamic_resources_by_ip[ip] = dynamic_resources_update
 
         now = time.time()
-        if (
-            ip not in self.last_used_time_by_ip
-            or self.static_resources_by_ip[ip] != self.dynamic_resources_by_ip[ip]
-        ):
-            self.last_used_time_by_ip[ip] = now
+        self.ray_nodes_last_used_time_by_ip[ip] = now - node_idle_duration_s
         self.last_heartbeat_time_by_ip[ip] = now
         self.waiting_bundles = waiting_bundles
         self.infeasible_bundles = infeasible_bundles
@@ -167,7 +164,7 @@ def prune(mapping, should_log):
                 )
             assert not (unwanted_ips & set(mapping))
 
-        prune(self.last_used_time_by_ip, should_log=True)
+        prune(self.ray_nodes_last_used_time_by_ip, should_log=True)
         prune(self.static_resources_by_ip, should_log=False)
         prune(self.raylet_id_by_ip, should_log=False)
         prune(self.dynamic_resources_by_ip, should_log=False)
@@ -337,7 +334,7 @@ def _info(self):
         resources_used, resources_total = self._get_resource_usage()
 
         now = time.time()
-        idle_times = [now - t for t in self.last_used_time_by_ip.values()]
+        idle_times = [now - t for t in self.ray_nodes_last_used_time_by_ip.values()]
         heartbeat_times = [now - t for t in self.last_heartbeat_time_by_ip.values()]
         most_delayed_heartbeats = sorted(
             self.last_heartbeat_time_by_ip.items(), key=lambda pair: pair[1]

python/ray/autoscaler/_private/monitor.py

@@ -30,6 +30,7 @@
 from ray.autoscaler._private.load_metrics import LoadMetrics
 from ray.autoscaler._private.prom_metrics import AutoscalerPrometheusMetrics
 from ray.autoscaler._private.util import format_readonly_node_type
+from ray.autoscaler.v2.sdk import get_cluster_resource_state
 from ray.core.generated import gcs_pb2
 from ray.core.generated.event_pb2 import Event as RayEvent
 from ray.experimental.internal_kv import (
@@ -245,6 +246,15 @@ def update_load_metrics(self):
         resources_batch_data = response.resource_usage_data
         log_resource_batch_data_if_desired(resources_batch_data)
 
+        # This is a workaround to get correct idle_duration_ms
+        # from "get_cluster_resource_state"
+        # ref: https://github.com/ray-project/ray/pull/48519#issuecomment-2481659346
+        cluster_resource_state = get_cluster_resource_state(self.gcs_client)
+        ray_node_states = cluster_resource_state.node_states
+        ray_nodes_idle_duration_ms_by_id = {
+            node.node_id: node.idle_duration_ms for node in ray_node_states
+        }
+
         # Tell the readonly node provider what nodes to report.
         if self.readonly_config:
             new_nodes = []
@@ -309,11 +319,21 @@ def update_load_metrics(self):
                     ip = node_id.hex()
             else:
                 ip = resource_message.node_manager_address
+
+            idle_duration_s = 0.0
+            if node_id in ray_nodes_idle_duration_ms_by_id:
+                idle_duration_s = ray_nodes_idle_duration_ms_by_id[node_id] / 1000
+            else:
+                logger.warning(
+                    f"node_id {node_id} not found in ray_nodes_idle_duration_ms_by_id"
+                )
+
             self.load_metrics.update(
                 ip,
                 node_id,
                 total_resources,
                 available_resources,
+                idle_duration_s,
                 waiting_bundles,
                 infeasible_bundles,
                 pending_placement_groups,