Revert "xds: Convert CdsLb to XdsDepManager"

This reverts commit 297ab05.

b/430347751 shows multiple concerning behaviors in the xDS stack with
the new A74 config update model. XdsDepManager and CdsLB2 still seem to
be working correctly, but the change is exacerbated issues in other
parts of the stack, like RingHashConfig not having equals fixed in
a8de9f0.

Revert only for the v1.74.x release, leaving it on master.

Author: ejona86

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

@@ -36,6 +36,8 @@
 @Internal
 public class CdsLoadBalancerProvider extends LoadBalancerProvider {
 
+  private static final String CLUSTER_KEY = "cluster";
+
   @Override
   public boolean isAvailable() {
     return true;
@@ -68,12 +70,9 @@ public ConfigOrError parseLoadBalancingPolicyConfig(
    */
   static ConfigOrError parseLoadBalancingConfigPolicy(Map<String, ?> rawLoadBalancingPolicyConfig) {
     try {
-      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));
+      String cluster =
+          JsonUtil.getString(rawLoadBalancingPolicyConfig, CLUSTER_KEY);
+      return ConfigOrError.fromConfig(new CdsConfig(cluster));
     } catch (RuntimeException e) {
       return ConfigOrError.fromError(
           Status.UNAVAILABLE.withCause(e).withDescription(
@@ -90,28 +89,15 @@ 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)
-          .add("isDynamic", isDynamic)
-          .toString();
+      return MoreObjects.toStringHelper(this).add("name", name).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 'minRingSize'/'maxRingSize'"));
+          "Invalid 'mingRingSize'/'maxRingSize'"));
     }
     return ConfigOrError.fromConfig(
         new RingHashConfig(minRingSize, maxRingSize, requestHashHeader));

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

@@ -172,9 +172,7 @@ static CdsUpdate processCluster(Cluster cluster,
         lbConfig.getPolicyName()).parseLoadBalancingPolicyConfig(
         lbConfig.getRawConfigValue());
     if (configOrError.getError() != null) {
-      throw new ResourceInvalidException(
-          "Failed to parse lb config for cluster '" + cluster.getName() + "': "
-          + configOrError.getError());
+      throw new ResourceInvalidException(structOrError.getErrorDetail());
     }
 
     updateBuilder.lbPolicyConfig(lbPolicyConfig);
@@ -211,10 +209,6 @@ 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,12 +254,6 @@ XdsConfig build() {
   }
 
   public interface XdsClusterSubscriptionRegistry {
-    Subscription subscribeToCluster(String clusterName);
-  }
-
-  public interface Subscription extends Closeable {
-    /** Release resources without throwing exceptions. */
-    @Override
-    void close();
+    Closeable subscribeToCluster(String clusterName);
   }
 }

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

@@ -18,7 +18,6 @@
 
 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;
@@ -35,6 +34,8 @@
 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;
@@ -55,43 +56,39 @@
 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; // Specified by gRFC A37
+  private static final int MAX_CLUSTER_RECURSION_DEPTH = 16; // Matches C++
   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,
+  XdsDependencyManager(XdsClient xdsClient, XdsConfigWatcher xdsConfigWatcher,
                        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 XdsConfig.Subscription subscribeToCluster(String clusterName) {
-    checkState(this.xdsConfigWatcher != null, "dep manager must first be started");
+  public Closeable subscribeToCluster(String clusterName) {
     checkNotNull(clusterName, "clusterName");
     ClusterSubscription subscription = new ClusterSubscription(clusterName);
 
@@ -294,17 +291,10 @@ private static void addConfigForCluster(
           addConfigForCluster(clusters, childCluster, ancestors, tracer);
           StatusOr<XdsConfig.XdsClusterConfig> config = clusters.get(childCluster);
           if (!config.hasValue()) {
-            // 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;
+            clusters.put(clusterName, StatusOr.fromStatus(Status.INTERNAL.withDescription(
+                "Unable to get leaves for " + clusterName + ": "
+                + config.getStatus().getDescription())));
+            return;
           }
           XdsConfig.XdsClusterConfig.ClusterChild children = config.getValue().getChildren();
           if (children instanceof AggregateConfig) {
@@ -336,11 +326,6 @@ private static void addConfigForCluster(
         child = new EndpointConfig(StatusOr.fromStatus(Status.UNAVAILABLE.withDescription(
               "Unknown type in cluster " + clusterName + " " + 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)));
   }
@@ -422,7 +407,7 @@ public interface XdsConfigWatcher {
     void onUpdate(StatusOr<XdsConfig> config);
   }
 
-  private final class ClusterSubscription implements XdsConfig.Subscription {
+  private final class ClusterSubscription implements Closeable {
     private final String clusterName;
     boolean closed; // Accessed from syncContext
 
@@ -435,7 +420,7 @@ String getClusterName() {
     }
 
     @Override
-    public void close() {
+    public void close() throws IOException {
       releaseSubscription(this);
     }
   }

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

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