[dancingactor]

Description

Context

This PR migrates autoscaler v1 to read only from GcsAutoscalerStateManager, rather than the legacy path in GcsResourceManager. This enables future removal of redudant cluster resource data in GcsResourceManager (reference), significantly improving GCS performance and reducing the complexity of GCS's cluster resource view management logic. I explain this in more detail below:

GCS stores the same cluster resource availability and demand information multiple times.

• Availability:
  • ClusterResourceManager
  • GcsResourceManager
  • GcsAutoscalerStateManager
• Demand:
  • GcsResourceManager
  • GcsAutoscalerStateManager

GcsResourceManager is the redundant copy in both cases, so we can eventually remove it entirely. But before we can do that, we first need to remove one of its consumers — autoscaler v1 — by making it fetch the same information from GcsAutoscalerStateManager instead, which is what this PR does.

What this PR does

Autoscaler v1 currently reads resource demand and placement-group load from GcsResourceManager.

• Resource demand: GcsAutoscalerStateManager already has an equivalent field, so we use it directly.
• Placement-group load: GcsAutoscalerStateManager only exposed the derived pending_gang_resource_requests shape, not the raw PlacementGroupTableData that v1 needs, so we added a placement_group_load field to autoscaler.proto so GcsAutoscalerStateManager can expose it directly.

Related Issues

Related to #66718

[gemini-code-assist]

Code Review

This pull request migrates the v1 autoscaler's resource demand and placement group load tracking from the legacy get_all_resource_usage GCS RPC to get_cluster_resource_state. It updates parse_resource_demands to parse pending_resource_requests from ClusterResourceState and adds temporary support for raw placement group load in ClusterResourceState to maintain compatibility until the v1 autoscaler is retired. Corresponding tests have been updated to reflect these changes. There are no review comments to address.

[cursor]

Stale Bugbot comment from a previous run.

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

^{Reviewed by Cursor Bugbot for commit 5d2d9aa. Configure here.}

[rueian]

Hi @dancingactor, could you try this test:

def test_infeasible_demand_does_not_crowd_out_feasible_demand(monkeypatch):
    monitor = monitor_module.Monitor.__new__(monitor_module.Monitor)
    monitor.gcs_client = types.SimpleNamespace()
    monitor.load_metrics = LoadMetrics()
    monitor.autoscaler = types.SimpleNamespace(config={"provider": {}})
    monitor.autoscaling_config = None
    monitor.readonly_config = None
    monitor.prom_metrics = None
    monitor.event_summarizer = None

    cluster_state = autoscaler_pb2.ClusterResourceState()
    cluster_state.node_states.add(
        node_id=bytes.fromhex("ab" * 20),
        node_ip_address="1.2.3.4",
        total_resources={"CPU": 4.0},
    )
    # 800 GPU tasks no running node can fit, then 300 CPU tasks.
    for shape, count in (({"GPU": 1.0}, 800), ({"CPU": 1.0}, 300)):
        cluster_state.pending_resource_requests.add(
            request=autoscaler_pb2.ResourceRequest(resources_bundle=shape), count=count
        )
    monkeypatch.setattr(
        monitor_module, "get_cluster_resource_state", lambda gcs_client: cluster_state
    )

    monitor.update_load_metrics()

    demand_vector = monitor.load_metrics.get_resource_demand_vector()
    assert demand_vector.count({"CPU": 1.0}) == 300

[dancingactor]

Thanks @rueian, I ran it, and the test fails with 200 == 300. This is because this PR changes how the autoscaler caps demand: the cap is now 1000 for all demand, regardless of whether it is feasible or infeasible.

Cap before this PR:

• Feasible demand: 1000
• Infeasible demand: 1000

What this PR does:

I removed the split between feasible and infeasible demand for two reasons:

1. Autoscaler v1 doesn't need the infeasible demand list at all.
2. GcsAutoscalerStateManager doesn't provide separate feasible and infeasible demand.

So the cap becomes:

• All demand: 1000
• Per resource shape: 1000

Which fix do you think is better?

• Remove the cap completely, but the demand list might get really long
• In autoscaler v1, split feasible and infeasible demand again just so feasible demand isn't capped by infeasible demand. It will cost some performance, but it gives feasible demand more room under the cap.
• Increase the cap to 2000, the same total as before this PR, so the tolerance range is larger.
• Keep the current logic as is.

[rueian]

How about doing something like this:

shapes = [
    (
        dict(request_by_count.request.resources_bundle),
        min(request_by_count.count, AUTOSCALER_MAX_RESOURCE_DEMAND_VECTOR_SIZE),
    )
    for request_by_count in pending_resource_requests
]
for i in range(max((count for _, count in shapes), default=0)):
    demand_bundles.extend(shape for shape, count in shapes if count > i)

That makes the test pass with no GCS change and the vector still capped at 1000.

[dancingactor]

This solution makes sense, changed in 32d8078.

Verified the test passes as well, thanks!