Repository: ray-project/ray
Pull request: [core][autoscaler] let AWS node provider retry instance retrieval to handle eventual consistency
## Why are these changes needed?

Fixes https://github.com/ray-project/ray/issues/51861. The issue fails on the assertion:
https://github.com/ray-project/ray/blob/574cf65b3996513d476ba40c48b899d8de0a27a0/python/ray/autoscaler/_private/aws/node_provider.py#L597-L598

The assertion can only fail on `len(matches) == 0` because AWS will never return duplicate instances in the API.

As documented [here](https://docs.aws.amazon.com/ec2/latest/devguide/eventual-consistency.html), the AWS EC2 API follows an eventual consistency model. It is possible that we can't query an instance by using `ec2.instances.filter(InstanceIds=[node_id])` immediately after the instance is created. We need to retry for a few seconds. This PR reuses the `BOTO_MAX_RETRIES` environment variable, which is 12 by default, to retry the operation. That's up to 12 seconds.


## Related issue number

Closes https://github.com/ray-project/ray/issues/51861

## 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.
- [ ] 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
   - [x] Unit tests
   - [ ] Release tests
   - [ ] This PR is not tested :(


Problem statement:
[core] EC2 Autoscaler race condition
### What happened + What you expected to happen

Ray Ec2 autoscaler can get into a state where it 'looses' track of an instance, but that instance is present in AWS, and it causes an exception, which leads to the autoscaler to get stuck:

Here's the exception:
```
2025-03-31 11:50:57.575
AssertionError: Invalid instance id i-04074e6d2eba88b27

2025-03-31 11:50:57.575
           ^^^^^^^^^^^^^^^^^
2025-03-31 11:50:57.575
    assert len(matches) == 1, "Invalid instance id {}".format(node_id)
2025-03-31 11:50:57.575
  File "/opt/ml/lib/python3.12/site-packages/ray/autoscaler/_private/aws/node_provider.py", line 598, in _get_node
2025-03-31 11:50:57.575
           ^^^^^^^^^^^^^^^^^^^^^^^
2025-03-31 11:50:57.575
    return self._get_node(node_id)
2025-03-31 11:50:57.575
  File "/opt/ml/lib/python3.12/site-packages/ray/autoscaler/_private/aws/node_provider.py", line 606, in _get_cached_node
2025-03-31 11:50:57.575
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
2025-03-31 11:50:57.575
    node = self._get_cached_node(node_id)
2025-03-31 11:50:57.575
  File "/opt/ml/lib/python3.12/site-packages/ray/autoscaler/_private/aws/node_provider.py", line 195, in internal_ip
2025-03-31 11:50:57.574
                                  ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
2025-03-31 11:50:57.574
    self.load_metrics.mark_active(self.provider.internal_ip(node_id))
2025-03-31 11:50:57.574
  File "/opt/ml/lib/python3.12/site-packages/ray/autoscaler/_private/autoscaler.py", line 775, in process_completed_updates
2025-03-31 11:50:57.574
    self.process_completed_updates()
2025-03-31 11:50:57.574
  File "/opt/ml/lib/python3.12/site-packages/ray/autoscaler/_private/autoscaler.py", line 439, in _update
2025-03-31 11:50:57.574
    self._update()
2025-03-31 11:50:57.574
  File "/opt/ml/lib/python3.12/site-packages/ray/autoscaler/_private/autoscaler.py", line 377, in update
2025-03-31 11:50:57.574
Traceback (most recent call last):
2025-03-31 11:50:57.574
2025-03-31 11:50:57,505 ERROR autoscaler.py:380 -- StandardAutoscaler: Error during autoscaling.
```


### Versions / Dependencies

Ray version: 2.40.0

### Reproduction script

Run a ray job hundres/thousands of tasks and let cluster scale up quickly to hundreds or thousands of nodes.  We observe this happening with clusters as small as ~150 nodes.


### Issue Severity

High: It blocks me from completing my task.

Hints from issue discussion:
@rueian can you take a look at this issue? cc @kevin85421 
The assertion failure comes from here
https://github.com/ray-project/ray/blob/574cf65b3996513d476ba40c48b899d8de0a27a0/python/ray/autoscaler/_private/aws/node_provider.py#L597-L598

The `ec2.instances.filter(InstanceIds=[node_id])` uses the https://docs.aws.amazon.com/AWSEC2/latest/APIReference/API_DescribeInstances.html API under the hood, which guarantees no duplicate instances in its response. That is, the only case that can fail the assertion `(len(matches) == 0)` is a race before the API reaches eventual consistency, as documented [here](https://docs.aws.amazon.com/ec2/latest/devguide/eventual-consistency.html).

I believe there’s no more effective solution to this issue than attempting to retry the `ec2.instances.filter(InstanceIds=[node_id])` operation for a few seconds. WDYT @akazakov, @kevin85421 and @jjyao ?

Commit message: [core][autoscaler] let AWS node provider retry instance retrieval to handle eventual consistency

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

Patch to review:
diff --git a/python/ray/autoscaler/_private/aws/node_provider.py b/python/ray/autoscaler/_private/aws/node_provider.py
index a4f34de70f2b6..abc706de77b9b 100644
--- a/python/ray/autoscaler/_private/aws/node_provider.py
+++ b/python/ray/autoscaler/_private/aws/node_provider.py
@@ -38,6 +38,7 @@
 logger = logging.getLogger(__name__)
 
 TAG_BATCH_DELAY = 1
+LIST_RETRY_DELAY = 1
 
 
 def to_aws_format(tags):
@@ -594,9 +595,23 @@ def _get_node(self, node_id):
 
         # Node not in {pending, running} -- retry with a point query. This
         # usually means the node was recently preempted or terminated.
-        matches = list(self.ec2.instances.filter(InstanceIds=[node_id]))
-        assert len(matches) == 1, "Invalid instance id {}".format(node_id)
-        return matches[0]
+        # The EC2 API is eventually consistent. This means that an instance
+        # might not be immediately visible. So we need to retry the query a few times.
+        # See: https://docs.aws.amazon.com/ec2/latest/devguide/eventual-consistency.html
+        # and https://github.com/ray-project/ray/issues/51861
+        for attempts in range(max(BOTO_MAX_RETRIES, 1)):  # at least try once.
+            matches = list(self.ec2.instances.filter(InstanceIds=[node_id]))
+            if len(matches) == 1:
+                return matches[0]
+            cli_logger.warning(
+                "get_node({}): Attempt failed with len(matches) == {}, retrying ({}/{}).",
+                node_id,
+                len(matches),
+                attempts + 1,
+                BOTO_MAX_RETRIES,
+            )
+            time.sleep(LIST_RETRY_DELAY)
+        raise AssertionError("Invalid instance id {}".format(node_id))
 
     def _get_cached_node(self, node_id):
         """Return node info from cache if possible, otherwise fetches it."""
diff --git a/python/ray/tests/aws/test_autoscaler_aws.py b/python/ray/tests/aws/test_autoscaler_aws.py
index a57916c5f307a..ffc89c64be0a2 100644
--- a/python/ray/tests/aws/test_autoscaler_aws.py
+++ b/python/ray/tests/aws/test_autoscaler_aws.py
@@ -769,6 +769,36 @@ def mock_get_cached_node(node_id):
         assert nodes_to_include_in_call == nodes_included_in_call
 
 
+def test_retry_get_node():
+    """
+    This tests _get_node() retries `ec2.instances.filter` if the first call returns an empty list.
+    This is important because the EC2 API is eventually consistent, and it may take some time for the
+    instance to be available after it has been launched.
+    """
+
+    with patch("ray.autoscaler._private.aws.node_provider.make_ec2_resource"):
+        provider = AWSNodeProvider(
+            provider_config={"region": "nowhere"},
+            cluster_name="default",
+        )
+
+    attempts = 0
+    fake_instance = {"id": "i-1234567890abcdef0"}
+
+    def mock_filter(*args, **kwargs):
+        nonlocal attempts
+        if kwargs.get("InstanceIds") == [fake_instance["id"]]:
+            attempts += 1
+            if attempts > 1:
+                return [fake_instance]
+        return []
+
+    provider.ec2.instances.filter.side_effect = mock_filter
+
+    assert provider._get_node(fake_instance["id"]) == fake_instance
+    assert attempts == 2
+
+
 def test_use_subnets_ordered_by_az(ec2_client_stub):
     """
     This test validates that when bootstrap_aws populates the SubnetIds field,


Code review comments:
diff_hunk: @@ -38,6 +38,7 @@
 logger = logging.getLogger(__name__)
 
 TAG_BATCH_DELAY = 1
+LIST_RETRY_DELAY = 1
path: python/ray/autoscaler/_private/aws/node_provider.py
text: @user1:
nit (no merge blocker, personal preference): is it better to add units? i.e. `LIST_RETRY_DELAY_SEC`

@author:
👍 👍