Repository: ray-project/ray
Pull request: [core][autoscaler][v1] do not removing nodes for upcoming placement groups
## Why are these changes needed?

As described in the issue https://github.com/ray-project/ray/issues/50783, the autoscaler may remove existing nodes that is going to satisfy a placement group due to being idle while new nodes are coming up. This can cause a placement group to be never scheduled if the time for setting up a new node closes to the idle_timeout_minutes config.

For example, say we have a placement group that requires 2 nodes and we have one node A already in the cluster, then the autoscaler will add another node B to fulfill the 2 nodes requirement. But if the new B node takes a long time to be ready, the A node will be removed by the current autoscaler due to being idle. And this loops continue.

This PR fixes the issue by making the statement:
```python
nodes_not_allowed_to_terminate = self._get_nodes_needed_for_request_resources
```
to take placement groups into account when removing idle nodes.

## Related issue number

Closes https://github.com/ray-project/ray/issues/50783

## Manual Test

To test the situation, we can patch the `FakeMultiNodeProvider` to delay a node startup time for `1 minute` to match the `idle_timeout_minutes` autoscaler config in the later testing application:

```diff
diff --git a/python/ray/autoscaler/_private/fake_multi_node/node_provider.py b/python/ray/autoscaler/_private/fake_multi_node/node_provider.py
index ccbb7c1fae..122f32c01b 100644
--- a/python/ray/autoscaler/_private/fake_multi_node/node_provider.py
+++ b/python/ray/autoscaler/_private/fake_multi_node/node_provider.py
@@ -4,6 +4,7 @@ import logging
 import os
 import subprocess
 import sys
+import threading
 import time
 from threading import RLock
 from types import ModuleType
@@ -303,7 +304,7 @@ class FakeMultiNodeProvider(NodeProvider):
 
     def is_running(self, node_id):
         with self.lock:
-            return node_id in self._nodes
+            return node_id in self._nodes and self._nodes[node_id]["node"] is not None
 
     def is_terminated(self, node_id):
         with self.lock:
@@ -377,9 +378,6 @@ class FakeMultiNodeProvider(NodeProvider):
                     ray_constants.LABELS_ENVIRONMENT_VARIABLE: json.dumps(labels),
                 },
             )
-            node = ray._private.node.Node(
-                ray_params, head=False, shutdown_at_exit=False, spawn_reaper=False
-            )
             all_tags = {
                 TAG_RAY_NODE_KIND: NODE_KIND_WORKER,
                 TAG_RAY_USER_NODE_TYPE: node_type,
@@ -389,9 +387,20 @@ class FakeMultiNodeProvider(NodeProvider):
             all_tags.update(tags)
             self._nodes[next_id] = {
                 "tags": all_tags,
-                "node": node,
+                "node": None,
             }
 
+            def _newnode():
+                time.sleep(60)
+                node = ray._private.node.Node(
+                    ray_params, head=False, shutdown_at_exit=False, spawn_reaper=False
+                )
+                with self.lock:
+                    self._nodes[next_id]["node"] = node
+
+            thread = threading.Thread(target=_newnode)
+            thread.start()
+
     def terminate_node(self, node_id):
         with self.lock:
             if self._termination_errors:
@@ -405,7 +414,8 @@ class FakeMultiNodeProvider(NodeProvider):
             self._terminate_node(node)
 
     def _terminate_node(self, node):
-        node["node"].kill_all_processes(check_alive=False, allow_graceful=True)
+        if node["node"]:
+            node["node"].kill_all_processes(check_alive=False, allow_graceful=True)
 
     @staticmethod
     def bootstrap_config(cluster_config):
```

And run this test application locally:

```python
import ray
from ray.cluster_utils import AutoscalingCluster
from ray.util.placement_group import placement_group, remove_placement_group

cluster = AutoscalingCluster(
    head_resources={"CPU": 0},
    worker_node_types={
        "type-1": {
            "resources": {"CPU": 2},
            "node_config": {},
            "min_workers": 0,
            "max_workers": 10,
        },
    },
    idle_timeout_minutes=1,
    max_workers=10
)

cluster.start()
ray.init("auto")

pg1 = placement_group([{"CPU": 1}] * 4, strategy="STRICT_SPREAD")
ray.get(pg1.ready())
remove_placement_group(pg1)
print('pg1 done')

pg2 = placement_group([{"CPU": 1}] * 10, strategy="STRICT_SPREAD")
ray.get(pg2.ready())
remove_placement_group(pg2)
print('pg2 done')
```

With the patched `FakeMultiNodeProvider`, the `pg2` in the testing application will never be scheduled and the autoscaler will get stuck in a loop of deleting each node. But after applying this PR, the `pg2` can be successfully scheduled. It also works for the default placement group strategy.

## Checks

- [x] I've signed off every commit(by using the -s flag, i.e., `git commit -s`) in this PR.
- [x] I've run `scripts/format.sh` to lint the changes in this PR.
- [x] I've included any doc changes needed for https://docs.ray.io/en/master/.
    - [x] I've added any new APIs to the API Reference. For example, if I added a
           method in Tune, I've added it in `doc/source/tune/api/` under the
           corresponding `.rst` file.
- [x] I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/
- Testing Strategy
   - [ ] Unit tests
   - [ ] Release tests
   - [ ] This PR is not tested :(


Problem statement:
[Autoscaler] Autoscaler gets into infinite cycle of removing and adding nodes, never satisfies placement group
### What happened + What you expected to happen

Autoscaler often removes existing nodes while it is adding nodes to meet the demands of a placement group. This can cause it to get into an endless cycle of removing and adding nodes, never satisfying the placement group demand.

I have noticed that this issue is more prominent when `idle_timeout_minutes` is smaller. However, I do not want to increase idle timeout or I will waste resources - to avoid this issue I found I need to make the timeout significantly longer than the time it takes to start up a node, which can be many minutes in case of a large Docker image. I also do not think I should increase the timeout: while the autoscaler is trying to satisfy a placement group, it should not consider any existing nodes that will satisfy it as idle.

I have observed the same issue also when using KubeRay.

### Versions / Dependencies

Ray version: 2.42.1

### Reproduction script

Cluster config:

```
cluster_name: 'debug'

max_workers: 10
upscaling_speed: 1.0
idle_timeout_minutes: 1

docker:
    image: 'rayproject/ray:latest'
    container_name: "ray_container"
    pull_before_run: True
    run_options:
        - "--ulimit nofile=65536:65536"

provider:
    type: aws
    region: us-east-1
    cache_stopped_nodes: True
    use_internal_ips: True

auth:
    ssh_user: ubuntu
    ssh_private_key: '~/.ssh/jakobleben-ec2.pem'

available_node_types:
    ray.head.default:
        node_config: &node_config
            InstanceType: m5.large
            ImageId: ami-0dd6adfad4ad37eec
            SecurityGroupIds: ['sg-0c15118bc2b72e433']
            SubnetId: subnet-05324181c6015696c
            IamInstanceProfile:
                Name: 'jakobhawkeyeHawkeyeRayNodeProfile'
            KeyName: 'jakobleben'
            BlockDeviceMappings:
                - DeviceName: /dev/xvda
                  Ebs:
                      VolumeSize: 80
                      VolumeType: gp3
    ray.worker:
        min_workers: 0
        max_workers: 10
        resources: {}
        node_config:
            <<: *node_config

head_node_type: ray.head.default

file_mounts: {}
cluster_synced_files: []
file_mounts_sync_continuously: False

rsync_exclude:
    - "__pycache__"
    - "**/.git"
    - "**/.git/**"

rsync_filter:
    - ".gitignore"

initialization_commands: []
setup_commands: []
head_setup_commands: []
worker_setup_commands: []

head_start_ray_commands:
    - ray stop
    - ray start --head --port=6379 --object-manager-port=8076 --autoscaling-config=~/ray_bootstrap_config.yaml --dashboard-host=0.0.0.0

worker_start_ray_commands:
    - ray stop
    - ray start --address=$RAY_HEAD_IP:6379 --object-manager-port=8076

```
Reproducing application (autoscaling.py):

```
import ray
from ray.util.placement_group import placement_group
import time
ray.init()
print("Creating placement group...")
pg = placement_group([{"CPU": 1}] * 4)  # Use this in first job
# pg = placement_group([{"CPU": 1}] * 10)  # Use this in second job
created = pg.wait(timeout_seconds=60 * 60)  # Allow 60 minutes
print(f"Finished. Created = {created}")
```

Submit first job: 
```
ray job submit --working-dir . -- python3 autoscaling.py
```
It will start 2 worker nodes (since each of the head and worker nodes has 2 CPU) and complete with a log like this:
```
...
Creating placement group...
Finished. Created = True
...
```

Edit the application to comment out the line ending in "Use this in first job" and **uncomment** the line ending in "Use this in second job" - this will increase the size of the placement group.

**After at least 30 seconds since the first job completing, but while the 2 worker nodes are still up, submit second job:**
```
ray job submit --working-dir . -- python3 autoscaling.py
```
It will go into a cycle like this and may never complete:
```
(autoscaler +6s) Adding 2 node(s) of type ray.worker.
(autoscaler +22s) Removing 2 nodes of type ray.worker (idle).
(autoscaler +33s) Resized to 2 CPUs.
(autoscaler +2m31s) Resized to 6 CPUs.
(autoscaler +3m28s) Removing 2 nodes of type ray.worker (idle).
(autoscaler +3m33s) Adding 2 node(s) of type ray.worker.
(autoscaler +3m38s) Resized to 2 CPUs.
(autoscaler +5m20s) Resized to 6 CPUs.
(autoscaler +6m17s) Removing 2 nodes of type ray.worker (idle).
(autoscaler +6m27s) Resized to 2 CPUs.
(autoscaler +6m43s) Adding 2 node(s) of type ray.worker.
(autoscaler +6m53s) Adding 2 node(s) of type ray.worker.
(autoscaler +7m35s) Resized to 6 CPUs.
```



### Issue Severity

Medium

Hints from issue discussion:
Hi @jjyao, @kevin85421 I'd like to help with this issue! Could you assign me on this?
@rueian can you check whether the issue is related to https://github.com/ray-project/ray/pull/48924? In addition, would you mind reviewing #48924?
> [@rueian](https://github.com/rueian) can you check whether the issue is related to [#48924](https://github.com/ray-project/ray/pull/48924)? In addition, would you mind reviewing [#48924](https://github.com/ray-project/ray/pull/48924)?

Sure, #48924 looks good to me and it is not related to this issue.

Commit message: [core][autoscaler] add start_node_delay_s config to FakeMultiNodeProvider for testing edge cases of idle_timeout_minutes

Signed-off-by: Rueian <rueiancsie@gmail.com>

Patch to review:
diff --git a/python/ray/autoscaler/_private/autoscaler.py b/python/ray/autoscaler/_private/autoscaler.py
index 6874d90376b6c..8c754707cba4c 100644
--- a/python/ray/autoscaler/_private/autoscaler.py
+++ b/python/ray/autoscaler/_private/autoscaler.py
@@ -47,6 +47,7 @@
     ResourceDemandScheduler,
     ResourceDict,
     get_bin_pack_residual,
+    placement_groups_to_resource_demands,
 )
 from ray.autoscaler._private.updater import NodeUpdaterThread
 from ray.autoscaler._private.util import (
@@ -495,11 +496,9 @@ def terminate_nodes_to_enforce_config_constraints(self, now: float):
         )
 
         # Don't terminate nodes needed by request_resources()
-        nodes_not_allowed_to_terminate: FrozenSet[NodeID] = {}
-        if self.load_metrics.get_resource_requests():
-            nodes_not_allowed_to_terminate = (
-                self._get_nodes_needed_for_request_resources(sorted_node_ids)
-            )
+        nodes_not_allowed_to_terminate = self._get_nodes_needed_for_request_resources(
+            sorted_node_ids
+        )
 
         # Tracks counts of nodes we intend to keep for each node type.
         node_type_counts = defaultdict(int)
@@ -893,6 +892,20 @@ def _get_nodes_needed_for_request_resources(
         assert self.provider
 
         nodes_not_allowed_to_terminate: Set[NodeID] = set()
+
+        (
+            placement_group_demand_vector,
+            strict_spreads,
+        ) = placement_groups_to_resource_demands(
+            self.load_metrics.get_pending_placement_groups()
+        )
+
+        resource_demands = (
+            placement_group_demand_vector + self.load_metrics.get_resource_requests()
+        )
+        if not resource_demands and not strict_spreads:
+            return frozenset(nodes_not_allowed_to_terminate)
+
         static_node_resources: Dict[
             NodeIP, ResourceDict
         ] = self.load_metrics.get_static_node_resources_by_ip()
@@ -926,9 +939,18 @@ def _get_nodes_needed_for_request_resources(
         # Since it is sorted based on last used, we "keep" nodes that are
         # most recently used when we binpack. We assume get_bin_pack_residual
         # is following the given order here.
-        used_resource_requests: List[ResourceDict]
+        used_resource_requests: List[ResourceDict] = copy.deepcopy(max_node_resources)
+
+        for bundles in strict_spreads:
+            unfulfilled, updated_node_resources = get_bin_pack_residual(
+                used_resource_requests, bundles, strict_spread=True
+            )
+            if unfulfilled:
+                continue
+            used_resource_requests = updated_node_resources
+
         _, used_resource_requests = get_bin_pack_residual(
-            max_node_resources, self.load_metrics.get_resource_requests()
+            used_resource_requests, resource_demands
         )
         # Remove the first entry (the head node).
         max_node_resources.pop(0)
diff --git a/python/ray/autoscaler/_private/fake_multi_node/node_provider.py b/python/ray/autoscaler/_private/fake_multi_node/node_provider.py
index ccbb7c1fae653..e017bf4c76a9b 100644
--- a/python/ray/autoscaler/_private/fake_multi_node/node_provider.py
+++ b/python/ray/autoscaler/_private/fake_multi_node/node_provider.py
@@ -5,7 +5,7 @@
 import subprocess
 import sys
 import time
-from threading import RLock
+from threading import RLock, Thread
 from types import ModuleType
 from typing import Any, Dict, Optional
 
@@ -263,6 +263,8 @@ def __init__(
         # Whether to launch multiple nodes at once, or one by one regardless of
         # the count (default)
         self._launch_multiple = provider_config.get("launch_multiple", False)
+        # Whether to delay the start of a new node (seconds)
+        self._start_node_delay_s = provider_config.get("start_node_delay_s", 0)
 
         # These are injected errors for testing purposes. If not None,
         # these will be raised on `create_node_with_resources_and_labels`` and
@@ -303,7 +305,11 @@ def non_terminated_nodes(self, tag_filters):
 
     def is_running(self, node_id):
         with self.lock:
-            return node_id in self._nodes
+            return node_id in self._nodes and (
+                # head node doesn't have a "node" key
+                "node" not in self._nodes[node_id]
+                or self._nodes[node_id]["node"] is not None
+            )
 
     def is_terminated(self, node_id):
         with self.lock:
@@ -377,9 +383,12 @@ def _create_node_with_resources_and_labels(
                     ray_constants.LABELS_ENVIRONMENT_VARIABLE: json.dumps(labels),
                 },
             )
-            node = ray._private.node.Node(
-                ray_params, head=False, shutdown_at_exit=False, spawn_reaper=False
-            )
+
+            def start_node():
+                return ray._private.node.Node(
+                    ray_params, head=False, shutdown_at_exit=False, spawn_reaper=False
+                )
+
             all_tags = {
                 TAG_RAY_NODE_KIND: NODE_KIND_WORKER,
                 TAG_RAY_USER_NODE_TYPE: node_type,
@@ -389,8 +398,20 @@ def _create_node_with_resources_and_labels(
             all_tags.update(tags)
             self._nodes[next_id] = {
                 "tags": all_tags,
-                "node": node,
+                "node": None,
             }
+            if self._start_node_delay_s <= 0:
+                self._nodes[next_id]["node"] = start_node()
+            else:
+
+                def _start_node_delay():
+                    time.sleep(self._start_node_delay_s)
+                    with self.lock:
+                        if next_id in self._nodes:
+                            self._nodes[next_id]["node"] = start_node()
+
+                thread = Thread(target=_start_node_delay, daemon=True)
+                thread.start()
 
     def terminate_node(self, node_id):
         with self.lock:
@@ -405,7 +426,8 @@ def terminate_node(self, node_id):
             self._terminate_node(node)
 
     def _terminate_node(self, node):
-        node["node"].kill_all_processes(check_alive=False, allow_graceful=True)
+        if "node" in node and node["node"] is not None:
+            node["node"].kill_all_processes(check_alive=False, allow_graceful=True)
 
     @staticmethod
     def bootstrap_config(cluster_config):
diff --git a/python/ray/autoscaler/_private/resource_demand_scheduler.py b/python/ray/autoscaler/_private/resource_demand_scheduler.py
index 3f08f6dd10c9f..ad5187d047ee2 100644
--- a/python/ray/autoscaler/_private/resource_demand_scheduler.py
+++ b/python/ray/autoscaler/_private/resource_demand_scheduler.py
@@ -902,7 +902,7 @@ def get_bin_pack_residual(
     # A most naive bin packing algorithm.
     nodes = copy.deepcopy(node_resources)
     # List of nodes that cannot be used again due to strict spread.
-    used = []
+    used = set()
     # We order the resource demands in the following way:
     # More complex demands first.
     # Break ties: heavier demands first.
@@ -919,20 +919,21 @@ def get_bin_pack_residual(
         found = False
         node = None
         for i in range(len(nodes)):
+            if i in used:
+                continue
             node = nodes[i]
             if _fits(node, demand):
                 found = True
                 # In the strict_spread case, we can't reuse nodes.
                 if strict_spread:
-                    used.append(node)
-                    del nodes[i]
+                    used.add(i)
                 break
         if found and node:
             _inplace_subtract(node, demand)
         else:
             unfulfilled.append(demand)
 
-    return unfulfilled, nodes + used
+    return unfulfilled, nodes
 
 
 def _fits(node: ResourceDict, resources: ResourceDict) -> bool:


Code review comments:
diff_hunk: @@ -893,6 +892,20 @@ def _get_nodes_needed_for_request_resources(
         assert self.provider
 
         nodes_not_allowed_to_terminate: Set[NodeID] = set()
+
+        (
+            placement_group_demand_vector,
path: python/ray/autoscaler/_private/autoscaler.py
text: @user1:
The name `placement_group_demand_vector` is a bit confusing. My initial guess for the type of this variable is `List[${resources required for the whole PG}]`. However, it is `List[${resources required for a bundle}]`.

We can consider:

```python
resource_demands: TYPE_HINT = []
strict_spreads: TYPE_HINT = []
resource_demands, strict_spreads = placement_groups_to_resource_demands(...)
resource_demands.extend(self.load_metrics.get_resource_requests())
if not resource_demands and not strict_spreads:
  return frozenset(nodes_not_allowed_to_terminate)
...
```

diff_hunk: @@ -926,9 +939,18 @@ def _get_nodes_needed_for_request_resources(
         # Since it is sorted based on last used, we "keep" nodes that are
         # most recently used when we binpack. We assume get_bin_pack_residual
         # is following the given order here.
-        used_resource_requests: List[ResourceDict]
+        used_resource_requests: List[ResourceDict] = copy.deepcopy(max_node_resources)
path: python/ray/autoscaler/_private/autoscaler.py
text: @user1:
Is this `used_resource_requests`? It's pretty odd that "used resource requests" equals the total node resources at the beginning.

I also found that `get_bin_pack_residual` uses `_inplace_subtract`. It's odd to subtract "used" resources after scheduling some resources. I guess it should be `node_remaining_resources`.

https://github.com/ray-project/ray/blob/1f5a72981d79c8551a8bc3ae48752695fca11454/python/ray/autoscaler/_private/resource_demand_scheduler.py#L931

diff_hunk: @@ -926,9 +939,18 @@ def _get_nodes_needed_for_request_resources(
         # Since it is sorted based on last used, we "keep" nodes that are
         # most recently used when we binpack. We assume get_bin_pack_residual
         # is following the given order here.
-        used_resource_requests: List[ResourceDict]
+        used_resource_requests: List[ResourceDict] = copy.deepcopy(max_node_resources)
path: python/ray/autoscaler/_private/autoscaler.py
text: @user1:
The naming is inconsistent (`max_node_resources`, `used_resource_requests`, `updated_node_resources`).

diff_hunk: @@ -926,9 +939,18 @@ def _get_nodes_needed_for_request_resources(
         # Since it is sorted based on last used, we "keep" nodes that are
path: python/ray/autoscaler/_private/autoscaler.py
text: @user1:
Would you mind renaming `max_node_resources` to `node_total_resources` and adding a comment? It's not common to use "max" resources. Instead, "total" and "used" resources are more common.

diff_hunk: @@ -389,8 +398,20 @@ def _create_node_with_resources_and_labels(
             all_tags.update(tags)
             self._nodes[next_id] = {
                 "tags": all_tags,
-                "node": node,
+                "node": None,
             }
+            if self._start_node_delay_s <= 0:
+                self._nodes[next_id]["node"] = start_node()
+            else:
+
+                def _start_node_delay():
+                    time.sleep(self._start_node_delay_s)
+                    with self.lock:
+                        if next_id in self._nodes:
+                            self._nodes[next_id]["node"] = start_node()
+
+                thread = Thread(target=_start_node_delay, daemon=True)
+                thread.start()
path: python/ray/autoscaler/_private/fake_multi_node/node_provider.py
text: @author:
Add this "delay" feature to `FakeMultiNodeProvider` so that we can have Python tests for the issue.