[Autoscaler][V2] Fix autoscaler terminating more nodes than exist of a type

[ryanaoleary]

Why are these changes needed?

We're seeing an issue where the value of worker_to_delete_set is incorrect, stopping the autoscaler from reconciling correctly due to the following assertion failing:

assert num_workers_dict[to_delete_instance.node_type] >= 0

Currently in _get_workers_delete_info, if there are multiple nodes in the same worker group with a pending deletion, only the first one is added to worker_to_delete_set. This set is then used in _initialize_scale_request here:

if to_delete_instance.cloud_instance_id in worker_to_delete_set:
  # If the instance is already in the workersToDelete field of
  # any worker group, skip it.
  continue

but since all workers aren't being added to the set, the above assertion fails and future calls to intialize_scale_request are unsuccessful.

Related issue number

#52264

Checks

• I've signed off every commit(by using the -s flag, i.e., git commit -s) in this PR.
• 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
  • Unit tests
  • Release tests
  • This PR is not tested :(

[ryanaoleary]

cc: @kevin85421 I tested this with the V2 autoscaler unit tests, but haven't been able to manually replicate the issue to where I could test the fix with my dev image. I added a guard and logging in 503eeea, since it seems like bad behavior for the autoscaler to block all future scale up and terminate calls (since they both rely on initialize_scale_request) if the workers_to_delete set is out of sync.

[kevin85421]

@ryanaoleary, I recall you mentioned that a user can consistently reproduce the issue. Could you build an image for them to verify it? Additionally, if possible, please do your best to reproduce the issue so we can add tests to prevent it from happening again.

[kevin85421]

Chat with @ryanaoleary offline. The customer can't use the dev image. Let's merge this PR and ask the users to verify whether the issue is resolved or not.

[kevin85421]

cc @jjyao @edoakes for merging this PR. Thanks!

[ryanaoleary]

I wrote a unit test that sets replicas and minReplicas to 0 and then attempts to terminate a node on the same reconciliation (I haven't pushed it yet because it's failing), this results in the error occurring even with the fix:

            num_workers_dict[to_delete_instance.node_type] -= 1
>           assert num_workers_dict[to_delete_instance.node_type] >= 0
E           AssertionError

../instance_manager/cloud_providers/kuberay/cloud_provider.py:253: AssertionError

 python/ray/autoscaler/v2/tests/test_node_provider.py::KubeRayProviderIntegrationTest.test_scale_down_pods_with_replica_limit ⨯93% █████████▍2025-05-08 18:36:39,480  INFO cloud_provider.py:121 -- Terminating worker pods: ['raycluster-autoscaler-worker-small-group-dkz2r']
2025-05-08 18:36:39,480 INFO cloud_provider.py:480 -- Fetched pod data at resource version .
2025-05-08 18:36:39,481 INFO cloud_provider.py:331 -- Submitting a scale request: KubeRayProvider.ScaleRequest(desired_num_workers=defaultdict(<class 'int'>, {'small-group': 0, 'gpu-group': 1, 'tpu-group': 1}), workers_to_delete=defaultdict(<class 'list'>, {'small-group': [CloudInstance(cloud_instance_id='raycluster-autoscaler-worker-small-group-dkz2r', node_type='small-group', node_kind=2, is_running=False, request_id=None)]}), worker_groups_without_pending_deletes=set(), worker_groups_with_pending_deletes=set())

since cur_instances still contains the instance to be deleted, but the value of replicas has been updated to a lower value. This is why I think we need the change from b4f32fa that adds a guard and a warning (rather than an assertion) when a value of num_workers_dict goes below 0, since I think the autoscaler shouldn't continuously error on each subsequent iteration that calls _initialize_scale_request if a user updates replicas/minReplicas manually. We can instead log a warning and allow it to reconcile on the next iteration - since the Pod will terminate after the scale request gets submitted, workers_to_delete will clear, and the value of replicas/minReplicas will become accurate. What do you think @kevin85421 @jjyao?

[kevin85421]

Move our slack messages to here:

In my opinion, we shouldn’t allow users to manually update replicas; they should only be able to adjust the minimum and maximum replica counts.

I think we shouldn't have the autoscaler completely crash if a user updates replicas/minReplicas manually

This makes sense, but I’m worried it could bury other issues even deeper. Could you ask the users if they’re trying to manually lower both minReplicas and replicas?

If we can confirm that the issue is caused by either _get_workers_delete_info or manually lowering both minReplicas and replicas, I’m OK with accepting b4f32fa.

[ryanaoleary]

Move our slack messages to here:

In my opinion, we shouldn’t allow users to manually update replicas; they should only be able to adjust the minimum and maximum replica counts.

I think we shouldn't have the autoscaler completely crash if a user updates replicas/minReplicas manually

This makes sense, but I’m worried it could bury other issues even deeper. Could you ask the users if they’re trying to manually lower both minReplicas and replicas?

If we can confirm that the issue is caused by either _get_workers_delete_info or manually lowering both minReplicas and replicas, I’m OK with accepting b4f32fa.

I added a unit test in 9a6b143 that verifies that the provider can handle multiple node deletions of the same type in one autoscaler iteration. This should cover the bug with workers_to_delete not being correctly counted that is fixed by this PR. I don't think they're manually updating replicas or minReplicas, can we merge this PR as-is and then revisit the guard/logging change in b4f32fa if it's still required based on their response?

[kevin85421]

can we merge this PR as-is and then revisit the guard/logging change in b4f32fa if it's still required based on their response?

SG. I will review the new changes.

[kevin85421]

#52760 (comment)

[kevin85421]

Approve to unblock the user. I will open a follow up PR to do some improvements.

[kevin85421]

cc @jjyao would you mind merging this PR?

[jjyao]

We just had a conversation about not testing private method (i.e. implementation detail): #52812 (comment). Can we test the public interface instead?

[kevin85421]

I don’t agree with this philosophy in this case. Unit tests are much more appropriate here. We should

(1) Add more real e2e tests for KubeRay.
(2) Add more unit tests for well-defined functions.

We should avoid tests between these two which we need to mock the behaviors, and it's pretty common that the mock behavior is wrong.

[jjyao]

Yea, is there a way to test it through the public APIs?

[kevin85421]

Let's chat in-person.

[jjyao]

_get_workers_delete_info is a static method so we we don't need to create a KubeRayProvider instance at all.

[ryanaoleary]

Oh oops yeah, I was using the provider to get the list of cloud instances, but since it's KubeRay we can just use the Pod names. This should be fixed in 2b10b1a.

[jjyao]

Merging to unblock the fix. @kevin85421 will follow-up on some of the tests cleanup.

[JKBGIT1]

Hello guys, what's the ETA for release, please?