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

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

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

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

@@ -1173,18 +1173,23 @@ final class XdsClientImpl extends XdsClient {
         errorMessage = "ClusterLoadAssignment " + clusterName + " : no locality endpoints.";
         break;
       }
-      
+      Set<Integer> priorities = new HashSet<>();
-      // The policy.disable_overprovisioning field must be set to true.
+      int maxPriority = -1;
-      // TODO(chengyuanzhang): temporarily not requiring this field to be set, should push
-      //  server implementors to do this or TBD with design.
-
       for (io.envoyproxy.envoy.api.v2.endpoint.LocalityLbEndpoints localityLbEndpoints
           : assignment.getEndpointsList()) {
-        // The lb_endpoints field for LbEndpoint must contain at least one entry.
+        // Filter out localities without or with 0 weight.
-        if (localityLbEndpoints.getLbEndpointsCount() == 0) {
+        if (!localityLbEndpoints.hasLoadBalancingWeight()
-          errorMessage = "ClusterLoadAssignment " + clusterName + " : locality with no endpoint.";
+            || 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
@@ -1207,6 +1212,10 @@ final class XdsClientImpl extends XdsClient {
       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

@@ -2132,7 +2132,7 @@ public class XdsClientImplTest {
                 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,
@@ -2158,7 +2158,7 @@ public class XdsClientImplTest {
             new LocalityLbEndpoints(
                 ImmutableList.of(
                     new LbEndpoint("192.168.234.52", 8888,
-                        5, true)), 6, 1));
+                        5, true)), 6, 0));
   }
 
   /**
@@ -2311,7 +2311,7 @@ public class XdsClientImplTest {
                 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,
@@ -2336,7 +2336,7 @@ public class XdsClientImplTest {
             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);