[autoscaler] Do not count head node with min_workers constraint. (#12980)

python/ray/autoscaler/_private/autoscaler.py

@@ -149,7 +149,6 @@ class StandardAutoscaler:
 
     def _update(self):
         now = time.time()
-
         # Throttle autoscaling updates to this interval to avoid exceeding
         # rate limits on API calls.
         if now - self.last_update_time < self.update_interval_s:
@@ -333,7 +332,7 @@ class StandardAutoscaler:
                     NodeIP,
                     ResourceDict] = \
                     self.load_metrics.get_static_node_resources_by_ip()
-                head_node_resources = static_nodes[head_ip]
+                head_node_resources = static_nodes.get(head_ip, {})
             else:
                 head_node_resources = {}
 
@@ -482,11 +481,13 @@ class StandardAutoscaler:
                 # for legacy yamls.
                 self.resource_demand_scheduler.reset_config(
                     self.provider, self.available_node_types,
-                    self.config["max_workers"], upscaling_speed)
+                    self.config["max_workers"], self.config["head_node_type"],
+                    upscaling_speed)
             else:
                 self.resource_demand_scheduler = ResourceDemandScheduler(
                     self.provider, self.available_node_types,
-                    self.config["max_workers"], upscaling_speed)
+                    self.config["max_workers"], self.config["head_node_type"],
+                    upscaling_speed)
 
         except Exception as e:
             if errors_fatal:

python/ray/autoscaler/_private/resource_demand_scheduler.py

@@ -47,16 +47,19 @@ class ResourceDemandScheduler:
                  provider: NodeProvider,
                  node_types: Dict[NodeType, NodeTypeConfigDict],
                  max_workers: int,
+                 head_node_type: NodeType,
                  upscaling_speed: float = 1) -> None:
         self.provider = provider
         self.node_types = copy.deepcopy(node_types)
         self.max_workers = max_workers
+        self.head_node_type = head_node_type
         self.upscaling_speed = upscaling_speed
 
     def reset_config(self,
                      provider: NodeProvider,
                      node_types: Dict[NodeType, NodeTypeConfigDict],
                      max_workers: int,
+                     head_node_type: NodeType,
                      upscaling_speed: float = 1) -> None:
         """Updates the class state variables.
 
@@ -89,6 +92,7 @@ class ResourceDemandScheduler:
         self.provider = provider
         self.node_types = copy.deepcopy(final_node_types)
         self.max_workers = max_workers
+        self.head_node_type = head_node_type
         self.upscaling_speed = upscaling_speed
 
     def is_legacy_yaml(self,
@@ -153,7 +157,7 @@ class ResourceDemandScheduler:
          adjusted_min_workers) = \
             _add_min_workers_nodes(
                 node_resources, node_type_counts, self.node_types,
-                self.max_workers, ensure_min_cluster_size)
+                self.max_workers, self.head_node_type, ensure_min_cluster_size)
 
         # Step 3: add nodes for strict spread groups
         logger.info(f"Placement group demands: {pending_placement_groups}")
@@ -490,7 +494,7 @@ def _add_min_workers_nodes(
         node_resources: List[ResourceDict],
         node_type_counts: Dict[NodeType, int],
         node_types: Dict[NodeType, NodeTypeConfigDict], max_workers: int,
-        ensure_min_cluster_size: List[ResourceDict]
+        head_node_type: NodeType, ensure_min_cluster_size: List[ResourceDict]
 ) -> (List[ResourceDict], Dict[NodeType, int], Dict[NodeType, int]):
     """Updates resource demands to respect the min_workers and
     request_resources() constraints.
@@ -515,6 +519,9 @@ def _add_min_workers_nodes(
         existing = node_type_counts.get(node_type, 0)
         target = min(
             config.get("min_workers", 0), config.get("max_workers", 0))
+        if node_type == head_node_type:
+            # Add 1 to account for head node.
+            target = target + 1
         if existing < target:
             total_nodes_to_add_dict[node_type] = target - existing
             node_type_counts[node_type] = target