Repository: ray-project/ray
Pull request: [core][autoscaler][v1] fix crashes by infeasible strict spread placement groups
## Why are these changes needed?

As the issue https://github.com/ray-project/ray/issues/39691 describes, the autoscaler v1 will **crash intentionally** on any infeasible strict spread placement group. This PR changes the behavior to **skipping** such groups.

<!-- Please give a short summary of the change and the problem this solves. -->

## Related issue number

Closes https://github.com/ray-project/ray/issues/39691
Related to https://github.com/ray-project/kuberay/issues/2385

## Manual Test

Use the following script locally:

```python
import ray
from ray.cluster_utils import AutoscalingCluster
from ray.util.placement_group import placement_group

cluster = AutoscalingCluster(
    head_resources={"CPU": 1},
    worker_node_types={
        "type-1": {
            "resources": {"CPU": 2},
            "node_config": {},
            "min_workers": 0,
            "max_workers": 2,
        },
    },
    max_workers=2,
)

cluster.start()
ray.init("auto")

pg1 = placement_group([{"CPU": 2}] * 3, strategy="STRICT_SPREAD")
pg2 = placement_group([{"CPU": 2}] * 2, strategy="STRICT_SPREAD")

ray.get(pg2.ready())
```

Without this fix, the `pg2` never gets ready because the autoscaler keeps crashing on `assert not unfulfilled` caused by `pg1`.
However, with this fix, `pg1` keeps pending and the `pg2` will get ready.

## 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/.
    - [ ] 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.
- [ ] 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:
[core][autoscaler][v1] Infeasible strict spread Placement group would cause autoscaler stops
### What happened + What you expected to happen

When a strict spread PG cannot be fulfilled, it currently crashed the autoscaler v1's processing loop. 

```
2023-09-15 18:46:02,079 CRITICAL autoscaler.py:377 -- StandardAutoscaler: Too many errors, abort.
2023-09-15 18:46:02,079 ERROR monitor.py:434 -- Monitor: Execution exception. Trying again...
Traceback (most recent call last):
  File "/data/home/rickyx/ray-2/python/ray/autoscaler/_private/monitor.py", line 391, in _run
    self.autoscaler.update()
  File "/data/home/rickyx/ray-2/python/ray/autoscaler/_private/autoscaler.py", line 378, in update
    raise e
  File "/data/home/rickyx/ray-2/python/ray/autoscaler/_private/autoscaler.py", line 371, in update
    self._update()
  File "/data/home/rickyx/ray-2/python/ray/autoscaler/_private/autoscaler.py", line 442, in _update
    to_launch, unfulfilled = self.resource_demand_scheduler.get_nodes_to_launch(
  File "/data/home/rickyx/ray-2/python/ray/autoscaler/_private/resource_demand_scheduler.py", line 270, in get_nodes_to_launch
    ) = self.reserve_and_allocate_spread(
  File "/data/home/rickyx/ray-2/python/ray/autoscaler/_private/resource_demand_scheduler.py", line 595, in reserve_and_allocate_spread
    assert not unfulfilled
```


This will crash the loop and not have any scaling effect. 

### Versions / Dependencies

master

### Reproduction script

```
import ray
from ray.cluster_utils import AutoscalingCluster
from ray.util.placement_group import placement_group


cluster = AutoscalingCluster(
    head_resources={"CPU": 1},
    worker_node_types={
        "type-1": {
            "resources": {"CPU": 2},
            "node_config": {},
            "min_workers": 0,
            "max_workers": 2,
        },
    },
    max_workers=2,
)

cluster.start()
ray.init("auto")

from ray.util.placement_group import (
    placement_group,
    placement_group_table,
    remove_placement_group,
)
from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy


pg = placement_group([{"CPU": 2}] * 3, strategy="STRICT_SPREAD")

from ray.autoscaler.sdk import request_resources
request_resources(num_cpus=4)
```

### Issue Severity

None

Hints from issue discussion:
1. Is it a regression?
2. When is the plan to migrate to V2? 
> 1. Is it a regression?
> 2. When is the plan to migrate to V2?

1. Not a regression, has been around in v1.
2. Will be fixed in v2. 
Fixing in v2 is P0 to explicitly scope the problem here.
is this fixed in v2? 
I will downgrade all v1 autoscaling issue to p2. please make surer all relevant bugs are fixed in v2!
@rickyyx confirming - has this been addressed in the v2 autoscaler codepath?
yes
We're running into this Exception when we use STRICT_SPREAD placement groups on our ray cluster. 

```
 File "/venv/lib/python3.10/site-packages/ray/autoscaler/_private/monitor.py", line 584, in run
self._run()
File "/venv/lib/python3.10/site-packages/ray/autoscaler/_private/monitor.py", line 389, in _run
self.autoscaler.update()
File "/venv/lib/python3.10/site-packages/ray/autoscaler/_private/autoscaler.py", line 384, in update
raise e
File "/venv/lib/python3.10/site-packages/ray/autoscaler/_private/autoscaler.py", line 377, in update
self._update()
File "/venv/lib/python3.10/site-packages/ray/autoscaler/_private/autoscaler.py", line 448, in _update
to_launch, unfulfilled = self.resource_demand_scheduler.get_nodes_to_launch(
File "/venv/lib/python3.10/site-packages/ray/autoscaler/_private/resource_demand_scheduler.py", line 270, in get_nodes_to_launch
) = self.reserve_and_allocate_spread(
File "/venv/lib/python3.10/site-packages/ray/autoscaler/_private/resource_demand_scheduler.py", line 595, in reserve_and_allocate_spread
assert not unfulfilled
AssertionError
```

Does that mean we use v1 of autoscaler? if so, is there a way to switch on v2 ? We deploy our cluster using kuberay.

I think I found my answer here: https://docs.ray.io/en/latest/cluster/kubernetes/user-guides/configuring-autoscaling.html#autoscaler-v2-with-kuberay

Commit message: Merge branch 'master' into fix-crash-unfulfilled-strict-spread

Patch to review:
diff --git a/python/ray/autoscaler/_private/resource_demand_scheduler.py b/python/ray/autoscaler/_private/resource_demand_scheduler.py
index 1db20be69c62..a185749ba3fa 100644
--- a/python/ray/autoscaler/_private/resource_demand_scheduler.py
+++ b/python/ray/autoscaler/_private/resource_demand_scheduler.py
@@ -568,7 +568,7 @@ def reserve_and_allocate_spread(
         for bundles in strict_spreads:
             # Try to pack as many bundles of this group as possible on existing
             # nodes. The remaining will be allocated on new nodes.
-            unfulfilled, node_resources = get_bin_pack_residual(
+            unfulfilled, updated_node_resources = get_bin_pack_residual(
                 node_resources, bundles, strict_spread=True
             )
             max_to_add = self.max_workers + 1 - sum(node_type_counts.values())
@@ -582,8 +582,6 @@ def reserve_and_allocate_spread(
                 utilization_scorer=utilization_scorer,
                 strict_spread=True,
             )
-            _inplace_add(node_type_counts, to_launch)
-            _inplace_add(to_add, to_launch)
             new_node_resources = _node_type_counts_to_node_resources(
                 self.node_types, to_launch
             )
@@ -592,8 +590,12 @@ def reserve_and_allocate_spread(
             unfulfilled, including_reserved = get_bin_pack_residual(
                 new_node_resources, unfulfilled, strict_spread=True
             )
-            assert not unfulfilled
-            node_resources += including_reserved
+            if unfulfilled:
+                logger.debug("Unfulfilled strict placement group: {}".format(bundles))
+                continue
+            _inplace_add(node_type_counts, to_launch)
+            _inplace_add(to_add, to_launch)
+            node_resources = updated_node_resources + including_reserved
         return to_add, node_resources, node_type_counts
 
     def debug_string(


Code review comments:
diff_hunk: @@ -592,8 +590,12 @@ def reserve_and_allocate_spread(
             unfulfilled, including_reserved = get_bin_pack_residual(
                 new_node_resources, unfulfilled, strict_spread=True
             )
-            assert not unfulfilled
-            node_resources += including_reserved
+            if unfulfilled:
+                logger.debug("Unfulfilled strict placement group: {}".format(bundles))
path: python/ray/autoscaler/_private/resource_demand_scheduler.py
text: @user1:
```suggestion
                logger.debug("Unfulfilled strict spread placement group: {}".format(bundles))
```