xds: Convert CdsLb to XdsDepManager

I noticed we deviated from gRFC A37 in some ways. It turned out those
were added to the gRFC later in grpc/proposal#344:
- NACKing empty aggregate clusters
- Failing aggregate cluster when children could not be loaded
- Recusion limit of 16. We had this behavior already, but it was
  ascribed to matching C++

There's disagreement on whether we should actually fail the aggregate
cluster for bad children, so I'm preserving the pre-existing behavior
for now.

The code is now doing a depth-first leaf traversal, not breadth-first.
This was odd to see, but the code was also pretty old, so the reasoning
seems lost to history. Since we haven't seen more than a single level of
aggregate clusters in practice, this wouldn't have been noticed by
users.

XdsDependencyManager.start() was created to guarantee that the callback
could not be called before returning from the constructor. Otherwise
XDS_CLUSTER_SUBSCRIPT_REGISTRY could potentially be null.

Author: ejona86

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

@@ -36,8 +36,6 @@
 @Internal
 public class CdsLoadBalancerProvider extends LoadBalancerProvider {
 
-  private static final String CLUSTER_KEY = "cluster";
-
   @Override
   public boolean isAvailable() {
     return true;
@@ -70,9 +68,12 @@ public ConfigOrError parseLoadBalancingPolicyConfig(
    */
   static ConfigOrError parseLoadBalancingConfigPolicy(Map<String, ?> rawLoadBalancingPolicyConfig) {
     try {
-      String cluster =
-          JsonUtil.getString(rawLoadBalancingPolicyConfig, CLUSTER_KEY);
-      return ConfigOrError.fromConfig(new CdsConfig(cluster));
+      String cluster = JsonUtil.getString(rawLoadBalancingPolicyConfig, "cluster");
+      Boolean isDynamic = JsonUtil.getBoolean(rawLoadBalancingPolicyConfig, "is_dynamic");
+      if (isDynamic == null) {
+        isDynamic = Boolean.FALSE;
+      }
+      return ConfigOrError.fromConfig(new CdsConfig(cluster, isDynamic));
     } catch (RuntimeException e) {
       return ConfigOrError.fromError(
           Status.UNAVAILABLE.withCause(e).withDescription(
@@ -89,15 +90,28 @@ static final class CdsConfig {
      * Name of cluster to query CDS for.
      */
     final String name;
+    /**
+     * Whether this cluster was dynamically chosen, so the XdsDependencyManager may be unaware of
+     * it without an explicit cluster subscription.
+     */
+    final boolean isDynamic;
 
     CdsConfig(String name) {
+      this(name, false);
+    }
+
+    CdsConfig(String name, boolean isDynamic) {
       checkArgument(name != null && !name.isEmpty(), "name is null or empty");
       this.name = name;
+      this.isDynamic = isDynamic;
     }
 
     @Override
     public String toString() {
-      return MoreObjects.toStringHelper(this).add("name", name).toString();
+      return MoreObjects.toStringHelper(this)
+          .add("name", name)
+          .add("isDynamic", isDynamic)
+          .toString();
     }
   }
 }

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

@@ -104,7 +104,7 @@ private ConfigOrError parseLoadBalancingPolicyConfigInternal(
     }
     if (minRingSize <= 0 || maxRingSize <= 0 || minRingSize > maxRingSize) {
       return ConfigOrError.fromError(Status.UNAVAILABLE.withDescription(
-          "Invalid 'mingRingSize'/'maxRingSize'"));
+          "Invalid 'minRingSize'/'maxRingSize'"));
     }
     return ConfigOrError.fromConfig(
         new RingHashConfig(minRingSize, maxRingSize, requestHashHeader));

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

@@ -172,7 +172,9 @@ static CdsUpdate processCluster(Cluster cluster,
         lbConfig.getPolicyName()).parseLoadBalancingPolicyConfig(
         lbConfig.getRawConfigValue());
     if (configOrError.getError() != null) {
-      throw new ResourceInvalidException(structOrError.getErrorDetail());
+      throw new ResourceInvalidException(
+          "Failed to parse lb config for cluster '" + cluster.getName() + "': "
+          + configOrError.getError());
     }
 
     updateBuilder.lbPolicyConfig(lbPolicyConfig);
@@ -209,6 +211,10 @@ private static StructOrError<CdsUpdate.Builder> parseAggregateCluster(Cluster cl
     } catch (InvalidProtocolBufferException e) {
       return StructOrError.fromError("Cluster " + clusterName + ": malformed ClusterConfig: " + e);
     }
+    if (clusterConfig.getClustersList().isEmpty()) {
+      return StructOrError.fromError("Cluster " + clusterName
+          + ": aggregate ClusterConfig.clusters must not be empty");
+    }
     return StructOrError.fromStruct(CdsUpdate.forAggregate(
         clusterName, clusterConfig.getClustersList()));
   }

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

@@ -254,6 +254,12 @@ XdsConfig build() {
   }
 
   public interface XdsClusterSubscriptionRegistry {
-    Closeable subscribeToCluster(String clusterName);
+    Subscription subscribeToCluster(String clusterName);
+  }
+
+  public interface Subscription extends Closeable {
+    /** Release resources without throwing exceptions. */
+    @Override
+    void close();
   }
 }

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

@@ -18,6 +18,7 @@
 
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkNotNull;
+import static com.google.common.base.Preconditions.checkState;
 import static io.grpc.xds.client.XdsClient.ResourceUpdate;
 
 import com.google.common.annotations.VisibleForTesting;
@@ -34,8 +35,6 @@
 import io.grpc.xds.client.XdsClient;
 import io.grpc.xds.client.XdsClient.ResourceWatcher;
 import io.grpc.xds.client.XdsResourceType;
-import java.io.Closeable;
-import java.io.IOException;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -56,39 +55,43 @@
 final class XdsDependencyManager implements XdsConfig.XdsClusterSubscriptionRegistry {
   public static final XdsClusterResource CLUSTER_RESOURCE = XdsClusterResource.getInstance();
   public static final XdsEndpointResource ENDPOINT_RESOURCE = XdsEndpointResource.getInstance();
-  private static final int MAX_CLUSTER_RECURSION_DEPTH = 16; // Matches C++
+  private static final int MAX_CLUSTER_RECURSION_DEPTH = 16; // Specified by gRFC A37
   private final String listenerName;
   private final XdsClient xdsClient;
-  private final XdsConfigWatcher xdsConfigWatcher;
   private final SynchronizationContext syncContext;
   private final String dataPlaneAuthority;
+  private XdsConfigWatcher xdsConfigWatcher;
 
   private StatusOr<XdsConfig> lastUpdate = null;
   private final Map<XdsResourceType<?>, TypeWatchers<?>> resourceWatchers = new HashMap<>();
   private final Set<ClusterSubscription> subscriptions = new HashSet<>();
 
-  XdsDependencyManager(XdsClient xdsClient, XdsConfigWatcher xdsConfigWatcher,
+  XdsDependencyManager(XdsClient xdsClient,
                        SynchronizationContext syncContext, String dataPlaneAuthority,
                        String listenerName, NameResolver.Args nameResolverArgs,
                        ScheduledExecutorService scheduler) {
     this.listenerName = checkNotNull(listenerName, "listenerName");
     this.xdsClient = checkNotNull(xdsClient, "xdsClient");
-    this.xdsConfigWatcher = checkNotNull(xdsConfigWatcher, "xdsConfigWatcher");
     this.syncContext = checkNotNull(syncContext, "syncContext");
     this.dataPlaneAuthority = checkNotNull(dataPlaneAuthority, "dataPlaneAuthority");
     checkNotNull(nameResolverArgs, "nameResolverArgs");
     checkNotNull(scheduler, "scheduler");
-
-    // start the ball rolling
-    syncContext.execute(() -> addWatcher(new LdsWatcher(listenerName)));
   }
 
   public static String toContextStr(String typeName, String resourceName) {
     return typeName + " resource " + resourceName;
   }
 
+  public void start(XdsConfigWatcher xdsConfigWatcher) {
+    checkState(this.xdsConfigWatcher == null, "dep manager may not be restarted");
+    this.xdsConfigWatcher = checkNotNull(xdsConfigWatcher, "xdsConfigWatcher");
+    // start the ball rolling
+    syncContext.execute(() -> addWatcher(new LdsWatcher(listenerName)));
+  }
+
   @Override
-  public Closeable subscribeToCluster(String clusterName) {
+  public XdsConfig.Subscription subscribeToCluster(String clusterName) {
+    checkState(this.xdsConfigWatcher != null, "dep manager must first be started");
     checkNotNull(clusterName, "clusterName");
     ClusterSubscription subscription = new ClusterSubscription(clusterName);
 
@@ -291,10 +294,17 @@ private static void addConfigForCluster(
           addConfigForCluster(clusters, childCluster, ancestors, tracer);
           StatusOr<XdsConfig.XdsClusterConfig> config = clusters.get(childCluster);
           if (!config.hasValue()) {
-            clusters.put(clusterName, StatusOr.fromStatus(Status.INTERNAL.withDescription(
-                "Unable to get leaves for " + clusterName + ": "
-                + config.getStatus().getDescription())));
-            return;
+            // gRFC A37 says: If any of a CDS policy's watchers reports that the resource does not
+            // exist the policy should report that it is in TRANSIENT_FAILURE. If any of the
+            // watchers reports a transient ADS stream error, the policy should report that it is in
+            // TRANSIENT_FAILURE if it has never passed a config to its child.
+            //
+            // But there's currently disagreement about whether that is actually what we want, and
+            // that was not originally implemented in gRPC Java. So we're keeping Java's old
+            // behavior for now and only failing the "leaves" (which is a bit arbitrary for a
+            // cycle).
+            leafNames.add(childCluster);
+            continue;
           }
           XdsConfig.XdsClusterConfig.ClusterChild children = config.getValue().getChildren();
           if (children instanceof AggregateConfig) {
@@ -325,6 +335,11 @@ private static void addConfigForCluster(
       default:
         throw new IllegalStateException("Unexpected value: " + cdsUpdate.clusterType());
     }
+    if (clusters.containsKey(clusterName)) {
+      // If a cycle is detected, we'll have detected it while recursing, so now there will be a key
+      // present. We don't want to overwrite it with a non-error value.
+      return;
+    }
     clusters.put(clusterName, StatusOr.fromValue(
         new XdsConfig.XdsClusterConfig(clusterName, cdsUpdate, child)));
   }
@@ -406,7 +421,7 @@ public interface XdsConfigWatcher {
     void onUpdate(StatusOr<XdsConfig> config);
   }
 
-  private final class ClusterSubscription implements Closeable {
+  private final class ClusterSubscription implements XdsConfig.Subscription {
     private final String clusterName;
     boolean closed; // Accessed from syncContext
 
@@ -419,7 +434,7 @@ String getClusterName() {
     }
 
     @Override
-    public void close() throws IOException {
+    public void close() {
       releaseSubscription(this);
     }
   }

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

@@ -230,7 +230,8 @@ public void start(Listener2 listener) {
     ldsResourceName = XdsClient.canonifyResourceName(ldsResourceName);
     callCounterProvider = SharedCallCounterMap.getInstance();
 
-    resolveState = new ResolveState(ldsResourceName); // auto starts
+    resolveState = new ResolveState(ldsResourceName);
+    resolveState.start();
   }
 
   private static String expandPercentS(String template, String replacement) {
@@ -653,10 +654,14 @@ class ResolveState implements XdsDependencyManager.XdsConfigWatcher {
     private ResolveState(String ldsResourceName) {
       authority = overrideAuthority != null ? overrideAuthority : encodedServiceAuthority;
       xdsDependencyManager =
-          new XdsDependencyManager(xdsClient, this, syncContext, authority, ldsResourceName,
+          new XdsDependencyManager(xdsClient, syncContext, authority, ldsResourceName,
               nameResolverArgs, scheduler);
     }
 
+    void start() {
+      xdsDependencyManager.start(this);
+    }
+
     private void shutdown() {
       if (stopped) {
         return;