xds: filter EDS localities with clarified specifications (#6874) (#6875)

Fix logic of filtering localites in EDS responses:
   - Each LocalityLbEndpoints message is allowed to contain 0 LbEndpoints. 
   - LocalityLbEndpoints without or with 0 weight are ignored. 
   - NACK responses with sparse locality priorities.

xds/src/main/java/io/grpc/xds/LocalityStore.java

@@ -641,8 +641,7 @@ public void run() {
               && !localityLbInfo.childBalancer.canHandleEmptyAddressListFromNameResolution()) {
             localityLbInfo.childBalancer.handleNameResolutionError(
                 Status.UNAVAILABLE.withDescription(
-                    "No healthy address available from EDS update '" + localityLbEndpoints
-                        + "' for locality '" + locality + "'"));
+                    "Locality " + locality + " has no healthy endpoint"));
           } else {
             localityLbInfo.childBalancer
                 .handleResolvedAddresses(ResolvedAddresses.newBuilder()

xds/src/main/java/io/grpc/xds/XdsClientImpl.java

@@ -946,18 +946,23 @@ private void handleEdsResponse(DiscoveryResponse edsResponse) {
         errorMessage = "ClusterLoadAssignment " + clusterName + " : no locality endpoints.";
         break;
       }
-
-      // The policy.disable_overprovisioning field must be set to true.
-      // TODO(chengyuanzhang): temporarily not requiring this field to be set, should push
-      //  server implementors to do this or TBD with design.
-
+      Set<Integer> priorities = new HashSet<>();
+      int maxPriority = -1;
       for (io.envoyproxy.envoy.api.v2.endpoint.LocalityLbEndpoints localityLbEndpoints
           : assignment.getEndpointsList()) {
-        // The lb_endpoints field for LbEndpoint must contain at least one entry.
-        if (localityLbEndpoints.getLbEndpointsCount() == 0) {
-          errorMessage = "ClusterLoadAssignment " + clusterName + " : locality with no endpoint.";
+        // Filter out localities without or with 0 weight.
+        if (!localityLbEndpoints.hasLoadBalancingWeight()
+            || localityLbEndpoints.getLoadBalancingWeight().getValue() < 1) {
+          continue;
+        }
+        int localityPriority = localityLbEndpoints.getPriority();
+        if (localityPriority < 0) {
+          errorMessage =
+              "ClusterLoadAssignment " + clusterName + " : locality with negative priority.";
           break;
         }
+        maxPriority = Math.max(maxPriority, localityPriority);
+        priorities.add(localityPriority);
         // The endpoint field of each lb_endpoints must be set.
         // Inside of it: the address field must be set.
         for (io.envoyproxy.envoy.api.v2.endpoint.LbEndpoint lbEndpoint
@@ -980,6 +985,10 @@ private void handleEdsResponse(DiscoveryResponse edsResponse) {
       if (errorMessage != null) {
         break;
       }
+      if (priorities.size() != maxPriority + 1) {
+        errorMessage = "ClusterLoadAssignment " + clusterName + " : sparse priorities.";
+        break;
+      }
       for (io.envoyproxy.envoy.api.v2.ClusterLoadAssignment.Policy.DropOverload dropOverload
           : assignment.getPolicy().getDropOverloadsList()) {
         updateBuilder.addDropPolicy(DropOverload.fromEnvoyProtoDropOverload(dropOverload));

xds/src/test/java/io/grpc/xds/XdsClientImplTest.java

@@ -2001,7 +2001,7 @@ public void multipleEndpointWatchers() {
                 buildLocalityLbEndpoints("region2", "zone2", "subzone2",
                     ImmutableList.of(
                         buildLbEndpoint("192.168.234.52", 8888, HealthStatus.UNKNOWN, 5)),
-                    6, 1)),
+                    6, 0)),
             ImmutableList.<ClusterLoadAssignment.Policy.DropOverload>of())));
 
     response = buildDiscoveryResponse("1", clusterLoadAssignments,
@@ -2027,7 +2027,7 @@ public void multipleEndpointWatchers() {
             new LocalityLbEndpoints(
                 ImmutableList.of(
                     new LbEndpoint("192.168.234.52", 8888,
-                        5, true)), 6, 1));
+                        5, true)), 6, 0));
   }
 
   /**
@@ -2180,7 +2180,7 @@ public void addRemoveEndpointWatchers() {
                 buildLocalityLbEndpoints("region2", "zone2", "subzone2",
                     ImmutableList.of(
                         buildLbEndpoint("192.168.312.6", 443, HealthStatus.HEALTHY, 1)),
-                    6, 1)),
+                    6, 0)),
             ImmutableList.<Policy.DropOverload>of())));
 
     response = buildDiscoveryResponse("1", clusterLoadAssignments,
@@ -2205,7 +2205,7 @@ public void addRemoveEndpointWatchers() {
             new LocalityLbEndpoints(
                 ImmutableList.of(
                     new LbEndpoint("192.168.312.6", 443, 1, true)),
-                6, 1));
+                6, 0));
 
     // Cancel one of the watcher.
     xdsClient.cancelEndpointDataWatch("cluster-foo.googleapis.com", watcher1);