Do proper loop detection for agg clusters

Author: larry-safran

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

@@ -251,8 +251,10 @@ private void releaseSubscription(ClusterSubscription subscription) {
     checkNotNull(subscription, "subscription");
     String clusterName = subscription.getClusterName();
     syncContext.execute(() -> {
-      XdsWatcherBase<?> cdsWatcher =
-          resourceWatchers.get(CLUSTER_RESOURCE).watchers.get(clusterName);
+      XdsWatcherBase<?> cdsWatcher = null;
+      if (resourceWatchers.containsKey(CLUSTER_RESOURCE)) {
+        cdsWatcher = resourceWatchers.get(CLUSTER_RESOURCE).watchers.get(clusterName);
+      }
       if (cdsWatcher == null) {
         return; // already released while waiting for the syncContext
       }
@@ -324,6 +326,13 @@ private void maybePublishConfig() {
       return;
     }
 
+    List<String> namesInLoop = detectLoops(rawClusterWatchers);
+    if (namesInLoop != null) {
+      String error = "Detected loop in cluster dependencies: " + namesInLoop;
+      xdsConfigWatcher.onError("xDS node ID: " + dataPlaneAuthority,
+          Status.UNAVAILABLE.withDescription(error));
+      return;
+    }
     XdsConfig newConfig = buildConfig();
     if (Objects.equals(newConfig, lastXdsConfig)) {
       return;
@@ -332,6 +341,51 @@ private void maybePublishConfig() {
     xdsConfigWatcher.onUpdate(lastXdsConfig);
   }
 
+  private List<String> detectLoops(TypeWatchers<?> rawClusterWatchers) {
+    for (XdsWatcherBase<?> watcher : rawClusterWatchers.watchers.values()) {
+      if (!watcher.hasDataValue()) {
+        continue;
+      }
+      CdsWatcher cdsWatcher = (CdsWatcher) watcher;
+
+      XdsClusterResource.CdsUpdate cdsUpdate = cdsWatcher.getData().getValue();
+      if (cdsUpdate.clusterType() != ClusterType.AGGREGATE) {
+        continue;
+      }
+      List<String> namesInLoop =
+          detectLoops(Arrays.asList(watcher.resourceName), cdsUpdate.prioritizedClusterNames());
+      if (namesInLoop != null) {
+        return namesInLoop;
+      }
+    }
+
+    return null;
+  }
+
+  private List<String> detectLoops(List<String> parents, ImmutableList<String> children) {
+    if (!Collections.disjoint(parents, children)) {
+      String problemChild = children.stream().filter(c -> parents.contains(c)).findFirst().get();
+      return new ImmutableList.Builder<String>().addAll(parents).add(problemChild).build();
+    }
+
+    for (String child : children) {
+      CdsWatcher childWatcher = getCluster(child);
+      if (childWatcher == null || !childWatcher.getData().hasValue()
+          || childWatcher.getData().getValue().clusterType() != ClusterType.AGGREGATE) {
+        continue;
+      }
+      ImmutableList newParents =
+          new ImmutableList.Builder().addAll(parents).add(childWatcher.resourceName()).build();
+      List<String> childLoop =
+          detectLoops(newParents, childWatcher.getData().getValue().prioritizedClusterNames());
+      if (childLoop != null) {
+        return childLoop;
+      }
+    }
+
+    return null;
+  }
+
   @VisibleForTesting
   XdsConfig buildConfig() {
     XdsConfig.XdsConfigBuilder builder = new XdsConfig.XdsConfigBuilder();
@@ -988,10 +1042,7 @@ public void onChanged(XdsClusterResource.CdsUpdate update) {
               setData(update);
               Set<String> addedClusters = Sets.difference(newNames, oldNames);
               addedClusters.forEach((cluster) -> addClusterWatcher(cluster, parentContext, depth));
-
-              if (addedClusters.isEmpty()) {
-                maybePublishConfig();
-              }
+              maybePublishConfig();
             } else { // data was set to error status above
               maybePublishConfig();
             }

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

@@ -734,11 +734,13 @@ public void aggregateCluster_withLoops_afterEds() {
     xdsClient.deliverCdsUpdate(cluster3, update3);
 
     // cluster2 (aggr.) -> [cluster3 (EDS)]
+    reset(helper);
     CdsUpdate update2a =
         CdsUpdate.forAggregate(cluster2, Arrays.asList(cluster3, cluster1, cluster2, cluster3))
             .roundRobinLbPolicy().build();
     xdsClient.deliverCdsUpdate(cluster2, update2a);
     assertThat(xdsClient.watchers.keySet()).containsExactly(CLUSTER, cluster1, cluster2, cluster3);
+
     verify(helper).updateBalancingState(
         eq(ConnectivityState.TRANSIENT_FAILURE), pickerCaptor.capture());
     Status unavailable = Status.UNAVAILABLE.withDescription(