xds: Reduce per-endpoint memory from CDS

The locality name is long, so we don't want a different copy for each
endpoint. The PathChains can use a similar amount of memory, so we also
want to share them.

The code no longer re-creates the EAG and Attributes, reducing garbage;
that isn't as important as we're more concerned about retained memory in
steady-state. But there was little reason for the waste, thus the
cleanup.

This fixes a bug for PathChains longer than 2 entries, which was never
triggered as our PathChains are always 2 entries long. Previously the
code would discard all but the first and last entry, with the same
pathChain.next being repeatedly overwritten.

Noticed the memory use when investigating b/507652503.

Author: ejona86

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

@@ -24,11 +24,12 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.ListIterator;
 import javax.annotation.Nullable;
 
 final class AddressFilter {
   @ResolutionResultAttr
-  private static final Attributes.Key<PathChain> PATH_CHAIN_KEY =
+  static final Attributes.Key<PathChain> PATH_CHAIN_KEY =
       Attributes.Key.create("io.grpc.xds.AddressFilter.PATH_CHAIN_KEY");
 
   // Prevent instantiation.
@@ -41,19 +42,25 @@ private AddressFilter() {}
   static EquivalentAddressGroup setPathFilter(EquivalentAddressGroup address, List<String> names) {
     checkNotNull(address, "address");
     checkNotNull(names, "names");
-    Attributes.Builder attrBuilder = address.getAttributes().toBuilder().discard(PATH_CHAIN_KEY);
-    PathChain pathChain = null;
-    for (String name : names) {
-      if (pathChain == null) {
-        pathChain = new PathChain(name);
-        attrBuilder.set(PATH_CHAIN_KEY, pathChain);
-      } else {
-        pathChain.next = new PathChain(name);
-      }
-    }
+    Attributes.Builder attrBuilder = address.getAttributes().toBuilder()
+        .set(PATH_CHAIN_KEY, createPathChain(names));
     return new EquivalentAddressGroup(address.getAddresses(), attrBuilder.build());
   }
 
+  /**
+   * Creates a PathChain that can be set in an EquivalentAddressGroup's Attributes as a value of
+   * PATH_CHAIN_KEY.
+   */
+  @Nullable static PathChain createPathChain(List<String> names) {
+    checkNotNull(names, "names");
+    PathChain current = null;
+    ListIterator<String> iter = names.listIterator(names.size());
+    while (iter.hasPrevious()) {
+      current = new PathChain(iter.previous(), current);
+    }
+    return current;
+  }
+
   /**
    * Returns the next level hierarchical addresses derived from the given hierarchical addresses
    * with the given filter name (any non-hierarchical addresses in the input will be ignored).
@@ -75,12 +82,13 @@ static List<EquivalentAddressGroup> filter(List<EquivalentAddressGroup> addresse
     return Collections.unmodifiableList(filteredAddresses);
   }
 
-  private static final class PathChain {
+  static final class PathChain {
     final String name;
-    @Nullable PathChain next;
+    @Nullable final PathChain next;
 
-    PathChain(String name) {
+    PathChain(String name, @Nullable PathChain next) {
       this.name = checkNotNull(name, "name");
+      this.next = next;
     }
 
     @Override

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

@@ -332,6 +332,10 @@ StatusOr<ClusterResolutionResult> edsUpdateToResult(
       for (Locality locality : localityLbEndpoints.keySet()) {
         LocalityLbEndpoints localityLbInfo = localityLbEndpoints.get(locality);
         String priorityName = localityPriorityNames.get(locality);
+        String localityName = localityName(locality);
+        AddressFilter.PathChain pathChain =
+            AddressFilter.createPathChain(Arrays.asList(priorityName, localityName));
+
         boolean discard = true;
         // These sums _should_ fit in uint32, but XdsEndpointResource isn't actually verifying that
         // is true today. Since we are using long to avoid signedness trouble, the math happens to
@@ -367,7 +371,6 @@ StatusOr<ClusterResolutionResult> edsUpdateToResult(
               }
             }
 
-            String localityName = localityName(locality);
             Attributes attr =
                 endpoint.eag().getAttributes().toBuilder()
                     .set(InternalEquivalentAddressGroup.ATTR_BACKEND_SERVICE, clusterName)
@@ -377,6 +380,7 @@ StatusOr<ClusterResolutionResult> edsUpdateToResult(
                         localityLbInfo.localityWeight())
                     .set(io.grpc.xds.XdsAttributes.ATTR_SERVER_WEIGHT, weight)
                     .set(XdsInternalAttributes.ATTR_ADDRESS_NAME, endpoint.hostname())
+                    .set(AddressFilter.PATH_CHAIN_KEY, pathChain)
                     .build();
             EquivalentAddressGroup eag;
             if (discovery.isHttp11ProxyAvailable()) {
@@ -389,7 +393,6 @@ StatusOr<ClusterResolutionResult> edsUpdateToResult(
             } else {
               eag = new EquivalentAddressGroup(endpoint.eag().getAddresses(), attr);
             }
-            eag = AddressFilter.setPathFilter(eag, Arrays.asList(priorityName, localityName));
             addresses.add(eag);
           }
         }

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

@@ -58,4 +58,29 @@ public void filterAddresses() {
     assertThat(filteredAddress0.getAttributes().get(key1)).isEqualTo("value1");
     assertThat(filteredAddress1.getAddresses()).containsExactlyElementsIn(eag3.getAddresses());
   }
+
+  @Test
+  public void longerPathChain() {
+    List<EquivalentAddressGroup> addresses = Arrays.asList(
+        newEag(new InetSocketAddress(8000), Arrays.asList("A", "B", "C")),
+        newEag(new InetSocketAddress(8001), Arrays.asList("Z", "B", "C")),
+        newEag(new InetSocketAddress(8002), Arrays.asList("A", "Z", "C")),
+        newEag(new InetSocketAddress(8003), Arrays.asList("A", "B", "Z")));
+    addresses = AddressFilter.filter(addresses, "A");
+    assertThat(addresses).hasSize(3);
+
+    addresses = AddressFilter.filter(addresses, "B");
+    assertThat(addresses).hasSize(2);
+
+    addresses = AddressFilter.filter(addresses, "C");
+    assertThat(addresses).hasSize(1);
+    assertThat(addresses.get(0).getAddresses()).containsExactly(new InetSocketAddress(8000));
+
+    addresses = AddressFilter.filter(addresses, "D");
+    assertThat(addresses).hasSize(0);
+  }
+
+  private static EquivalentAddressGroup newEag(InetSocketAddress address, List<String> names) {
+    return AddressFilter.setPathFilter(new EquivalentAddressGroup(address), names);
+  }
 }