xds: EDS resource is not NACKed when locality weights at one priority sum to more than uint32 max

从 xds/src/main/java/io/grpc/xds/XdsEndpointResource.java 中的 processClusterLoadAssignment 开始，那里已经检查了 locality 是否重复以及 priority 是否稀疏，但从未按 priority 对其权重求和。GrpcXdsClientImplDataTest.java 中的回归测试 processClusterLoadAssignment_localityWeightSumExceedsUint32Max 展示了预期的 NACK 消息。当该测试通过且现有的 grpc-xds 测试仍然通过时，即为完成。

What version of gRPC-Java are you using?

master at 15d64a4 (1.86.0-SNAPSHOT). Current master has since added a duplicate-endpoint check to processClusterLoadAssignment, but there is still no weight-sum check.

What is your environment?

Linux x86_64, OpenJDK 21.0.2, Gradle 8.14.5 (wrapper).

What did you expect to see?

An EDS ClusterLoadAssignment whose locality weights at the same priority add up to more than 4294967295 is NACKed (ResourceInvalidException). The Envoy API documents LocalityLbEndpoints.load_balancing_weight with:

The sum of the weights of all localities at the same priority level must not exceed uint32_t maximal value (4294967295).

What did you see instead?

The resource is accepted.

Location (grpc/grpc-java @ 15d64a4):

• xds/src/main/java/io/grpc/xds/XdsEndpointResource.java:115-157: processClusterLoadAssignment checks duplicate localities and sparse priorities, but never sums the weights per priority.
• xds/src/main/java/io/grpc/xds/WeightedRandomPicker.java:98-107: WeightedRandomPicker (built by weighted_target, xds/src/main/java/io/grpc/xds/WeightedTargetLoadBalancer.java:146-178) rejects a total above uint32 max.

      LocalityLbEndpoints localityLbEndpoints = structOrError.getStruct();
      int priority = localityLbEndpoints.priority();
      maxPriority = Math.max(maxPriority, priority);
      ...
      Locality locality =  parseLocality(localityLbEndpointsProto.getLocality());
      localityLbEndpointsMap.put(locality, localityLbEndpoints);

    this.totalWeight = totalWeight;
    checkArgument(totalWeight <= UnsignedInteger.MAX_VALUE.longValue(),
        "total weight greater than unsigned int can hold");

Problem

Because the invalid resource is accepted, the bad weights reach the LB tree: wrr_locality passes the locality weights to weighted_target, and as the localities become READY weighted_target builds a WeightedRandomPicker from them. Once the READY localities' weights add up to more than uint32 max, the constructor throws. I reproduced this by driving WeightedTargetLoadBalancer directly with three targets (weights 2147483647, 2147483647, 2): the first two READY updates succeed, the third throws java.lang.IllegalArgumentException: total weight greater than unsigned int can hold out of Helper.updateBalancingState. In a real channel that call runs in the channel's SynchronizationContext, whose uncaught-exception handler panics the channel (core/src/main/java/io/grpc/internal/ManagedChannelImpl.java:195-198). I have not run that end-to-end. A NACK keeps the previous good resource and reports the error to the control plane instead.

Note: every locality weight must be below 2^31 to reach this, because parseLocalityLbEndpoints reads the uint32 weight into an int and skips localities with < 1 (xds/src/main/java/io/grpc/xds/XdsEndpointResource.java:197-200). So the overflow needs several large localities at one priority.

Steps to reproduce the bug

Reproduction

Regression test added to GrpcXdsClientImplDataTest:

diff --git a/xds/src/test/java/io/grpc/xds/GrpcXdsClientImplDataTest.java b/xds/src/test/java/io/grpc/xds/GrpcXdsClientImplDataTest.java
--- a/xds/src/test/java/io/grpc/xds/GrpcXdsClientImplDataTest.java
+++ b/xds/src/test/java/io/grpc/xds/GrpcXdsClientImplDataTest.java
@@ -1080,6 +1080,39 @@ public class GrpcXdsClientImplDataTest {
             100, 1, ImmutableMap.of()));
   }
 
+  @Test
+  public void processClusterLoadAssignment_localityWeightSumExceedsUint32Max() {
+    // Sum of locality weights at priority 1 is 2147483647 + 2147483647 + 2 = 4294967296.
+    io.envoyproxy.envoy.config.endpoint.v3.ClusterLoadAssignment assignment =
+        io.envoyproxy.envoy.config.endpoint.v3.ClusterLoadAssignment.newBuilder()
+            .setClusterName("cluster-foo")
+            .addEndpoints(buildLocalityLbEndpoints("region1", "172.14.14.1", 1, 0))
+            .addEndpoints(buildLocalityLbEndpoints("region2", "172.14.14.2", Integer.MAX_VALUE, 1))
+            .addEndpoints(buildLocalityLbEndpoints("region3", "172.14.14.3", Integer.MAX_VALUE, 1))
+            .addEndpoints(buildLocalityLbEndpoints("region4", "172.14.14.4", 2, 1))
+            .build();
+    ResourceInvalidException ex = assertThrows(
+        ResourceInvalidException.class,
+        () -> XdsEndpointResource.getInstance().doParse(getXdsResourceTypeArgs(true), assignment));
+    assertThat(ex).hasMessageThat().isEqualTo(
+        "Sum of weights of localities at the same priority 1 exceeded maximal value");
+  }
+
+  private static io.envoyproxy.envoy.config.endpoint.v3.LocalityLbEndpoints
+      buildLocalityLbEndpoints(String region, String address, int localityWeight, int priority) {
+    return io.envoyproxy.envoy.config.endpoint.v3.LocalityLbEndpoints.newBuilder()
+        .setLocality(Locality.newBuilder().setRegion(region))
+        .setLoadBalancingWeight(UInt32Value.of(localityWeight))
+        .setPriority(priority)
+        .addLbEndpoints(io.envoyproxy.envoy.config.endpoint.v3.LbEndpoint.newBuilder()
+            .setEndpoint(Endpoint.newBuilder()
+                .setAddress(Address.newBuilder()
+                    .setSocketAddress(
+                        SocketAddress.newBuilder().setAddress(address).setPortValue(8888))))
+            .setHealthStatus(io.envoyproxy.envoy.config.core.v3.HealthStatus.HEALTHY))
+        .build();
+  }
+
   @Test
   public void parseLocalityLbEndpoints_onlyPermitIp() {
     io.envoyproxy.envoy.config.endpoint.v3.LocalityLbEndpoints proto =

./gradlew :grpc-xds:test --tests io.grpc.xds.GrpcXdsClientImplDataTest -PskipCodegen=true -PskipAndroid=true

GrpcXdsClientImplDataTest > processClusterLoadAssignment_localityWeightSumExceedsUint32Max FAILED
    java.lang.AssertionError: expected io.grpc.xds.client.XdsResourceType.ResourceInvalidException to be thrown, but nothing was thrown
        at org.junit.Assert.assertThrows(Assert.java:1028)
        at org.junit.Assert.assertThrows(Assert.java:981)
        at io.grpc.xds.GrpcXdsClientImplDataTest.processClusterLoadAssignment_localityWeightSumExceedsUint32Max(GrpcXdsClientImplDataTest.java:1094)

Environment:

• Commit: 15d64a4678450991aa5bf7af23066d36b3391351
• Toolchain: OpenJDK 21.0.2, Gradle 8.14.5

Already fixed in grpc/grpc-go

PR: https://github.com/grpc/grpc-go/pull/5703 ("xds/xdsclient: add sum of EDS locality weights check", merged 2022-10-10T19:48:02Z, merge commit c6724519506). grpc-go's EDS parser keeps a running per-priority sum of locality weights and returns an error (NACK) once it exceeds math.MaxUint32.

How grpc-go fixed it (relevant hunks, verbatim):

--- a/xds/internal/xdsclient/xdsresource/unmarshal_eds.go
+++ b/xds/internal/xdsclient/xdsresource/unmarshal_eds.go
@@ -118,6 +119,7 @@ func parseEDSRespProto(m *v3endpointpb.ClusterLoadAssignment, logger *grpclog.Pr
 		ret.Drops = append(ret.Drops, parseDropPolicy(dropPolicy))
 	}
 	priorities := make(map[uint32]map[string]bool)
+	sumOfWeights := make(map[uint32]uint64)
 	for _, locality := range m.Endpoints {
 		l := locality.GetLocality()
 		if l == nil {
@@ -128,17 +130,21 @@ func parseEDSRespProto(m *v3endpointpb.ClusterLoadAssignment, logger *grpclog.Pr
...
 		priority := locality.GetPriority()
+		sumOfWeights[priority] += uint64(weight)
+		if sumOfWeights[priority] > math.MaxUint32 {
+			return EndpointsUpdate{}, fmt.Errorf("sum of weights of localities at the same priority %d exceeded maximal value", priority)
+		}
 		localitiesWithPriority := priorities[priority]
...

Suggested fix

Keep a per-priority running sum of the (already filtered) locality weights in processClusterLoadAssignment and NACK once it exceeds uint32 max:

diff --git a/xds/src/main/java/io/grpc/xds/XdsEndpointResource.java b/xds/src/main/java/io/grpc/xds/XdsEndpointResource.java
--- a/xds/src/main/java/io/grpc/xds/XdsEndpointResource.java
+++ b/xds/src/main/java/io/grpc/xds/XdsEndpointResource.java
@@ -22,6 +22,7 @@ import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.MoreObjects;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.net.InetAddresses;
+import com.google.common.primitives.UnsignedInteger;
 import com.google.protobuf.Any;
 import com.google.protobuf.InvalidProtocolBufferException;
 import com.google.protobuf.Message;
@@ -115,6 +116,7 @@ class XdsEndpointResource extends XdsResourceType<EdsUpdate> {
   private static EdsUpdate processClusterLoadAssignment(ClusterLoadAssignment assignment)
       throws ResourceInvalidException {
     Map<Integer, Set<Locality>> priorities = new HashMap<>();
+    Map<Integer, Long> localityWeightSums = new HashMap<>();
     Map<Locality, LocalityLbEndpoints> localityLbEndpointsMap = new LinkedHashMap<>();
     List<Endpoints.DropOverload> dropOverloads = new ArrayList<>();
     int maxPriority = -1;
@@ -132,6 +134,14 @@ class XdsEndpointResource extends XdsResourceType<EdsUpdate> {
       LocalityLbEndpoints localityLbEndpoints = structOrError.getStruct();
       int priority = localityLbEndpoints.priority();
       maxPriority = Math.max(maxPriority, priority);
+      long localityWeightSum = localityWeightSums.getOrDefault(priority, 0L)
+          + localityLbEndpoints.localityWeight();
+      if (localityWeightSum > UnsignedInteger.MAX_VALUE.longValue()) {
+        throw new ResourceInvalidException(
+            "Sum of weights of localities at the same priority " + priority
+                + " exceeded maximal value");
+      }
+      localityWeightSums.put(priority, localityWeightSum);
       // Note endpoints with health status other than HEALTHY and UNKNOWN are still
       // handed over to watching parties. It is watching parties' responsibility to
       // filter out unhealthy endpoints. See EnvoyProtoData.LbEndpoint#isHealthy().

Verified on 15d64a4 (OpenJDK 21.0.2, Gradle 8.14.5): the test above fails without this patch and passes with it. The rest of :grpc-xds:test gives the same results, except FakeControlPlaneXdsIntegrationTest.changeClusterForRoute[enableRfc3986UrisParam=true], which is timing-dependent (it sends up to 10 RPCs while waiting for an RDS switch): it failed once in the unpatched run and passed in the patched run.

I can submit a PR for this.

Checked: searches for locality weight sum / uint32 / overflow, XdsEndpointResource weight and the WeightedRandomPicker error message found no existing issue or PR (only #6731, which introduced WeightedTargetLoadBalancer).