Xds: Aggregate cluster fixes (A75) (#12186)

Instead of representing an aggregate cluster as a single cluster whose
priorities come from different underlying clusters, represent an aggregate cluster as an instance of a priority LB policy where each child is a cds LB policy for the underlying
cluster.

Author: kannanjgithub

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

@@ -18,10 +18,10 @@
 
 import static com.google.common.base.Preconditions.checkNotNull;
 import static io.grpc.ConnectivityState.TRANSIENT_FAILURE;
+import static io.grpc.xds.XdsLbPolicies.CDS_POLICY_NAME;
 import static io.grpc.xds.XdsLbPolicies.CLUSTER_RESOLVER_POLICY_NAME;
+import static io.grpc.xds.XdsLbPolicies.PRIORITY_POLICY_NAME;
 
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.collect.ImmutableList;
 import com.google.errorprone.annotations.CheckReturnValue;
 import io.grpc.InternalLogId;
 import io.grpc.LoadBalancer;
@@ -33,6 +33,7 @@
 import io.grpc.xds.CdsLoadBalancerProvider.CdsConfig;
 import io.grpc.xds.ClusterResolverLoadBalancerProvider.ClusterResolverConfig;
 import io.grpc.xds.ClusterResolverLoadBalancerProvider.ClusterResolverConfig.DiscoveryMechanism;
+import io.grpc.xds.PriorityLoadBalancerProvider.PriorityLbConfig.PriorityChildConfig;
 import io.grpc.xds.XdsClusterResource.CdsUpdate;
 import io.grpc.xds.XdsClusterResource.CdsUpdate.ClusterType;
 import io.grpc.xds.XdsConfig.Subscription;
@@ -41,10 +42,11 @@
 import io.grpc.xds.XdsConfig.XdsClusterConfig.EndpointConfig;
 import io.grpc.xds.client.XdsLogger;
 import io.grpc.xds.client.XdsLogger.XdsLogLevel;
-import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 
 /**
  * Load balancer for cds_experimental LB policy. One instance per top-level cluster.
@@ -55,19 +57,15 @@ final class CdsLoadBalancer2 extends LoadBalancer {
   private final XdsLogger logger;
   private final Helper helper;
   private final LoadBalancerRegistry lbRegistry;
+  private GracefulSwitchLoadBalancer delegate;
   // Following fields are effectively final.
   private String clusterName;
   private Subscription clusterSubscription;
-  private LoadBalancer childLb;
 
-  CdsLoadBalancer2(Helper helper) {
-    this(helper, LoadBalancerRegistry.getDefaultRegistry());
-  }
-
-  @VisibleForTesting
   CdsLoadBalancer2(Helper helper, LoadBalancerRegistry lbRegistry) {
     this.helper = checkNotNull(helper, "helper");
     this.lbRegistry = checkNotNull(lbRegistry, "lbRegistry");
+    this.delegate = new GracefulSwitchLoadBalancer(helper);
     logger = XdsLogger.withLogId(InternalLogId.allocate("cds-lb", helper.getAuthority()));
     logger.log(XdsLogLevel.INFO, "Created");
   }
@@ -91,7 +89,7 @@ public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {
       if (clusterSubscription == null) {
         // Should be impossible, because XdsDependencyManager wouldn't have generated this
         return fail(Status.INTERNAL.withDescription(
-            errorPrefix() + "Unable to find non-dynamic root cluster"));
+            errorPrefix() + "Unable to find non-dynamic cluster"));
       }
       // The dynamic cluster must not have loaded yet
       return Status.OK;
@@ -100,42 +98,25 @@ public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {
       return fail(clusterConfigOr.getStatus());
     }
     XdsClusterConfig clusterConfig = clusterConfigOr.getValue();
-    List<String> leafNames;
-    if (clusterConfig.getChildren() instanceof AggregateConfig) {
-      leafNames = ((AggregateConfig) clusterConfig.getChildren()).getLeafNames();
-    } else if (clusterConfig.getChildren() instanceof EndpointConfig) {
-      leafNames = ImmutableList.of(clusterName);
-    } else {
-      return fail(Status.INTERNAL.withDescription(
-          errorPrefix() + "Unexpected cluster children type: "
-          + clusterConfig.getChildren().getClass()));
-    }
-    if (leafNames.isEmpty()) {
-      // Should be impossible, because XdsClusterResource validated this
-      return fail(Status.UNAVAILABLE.withDescription(
-          errorPrefix() + "Zero leaf clusters for root cluster " + clusterName));
-    }
 
-    Status noneFoundError = Status.INTERNAL
-        .withDescription(errorPrefix() + "No leaves and no error; this is a bug");
-    List<DiscoveryMechanism> instances = new ArrayList<>();
-    for (String leafName : leafNames) {
-      StatusOr<XdsClusterConfig> leafConfigOr = xdsConfig.getClusters().get(leafName);
-      if (!leafConfigOr.hasValue()) {
-        noneFoundError = leafConfigOr.getStatus();
-        continue;
-      }
-      if (!(leafConfigOr.getValue().getChildren() instanceof EndpointConfig)) {
-        noneFoundError = Status.INTERNAL.withDescription(
-            errorPrefix() + "Unexpected child " + leafName + " cluster children type: "
-            + leafConfigOr.getValue().getChildren().getClass());
-        continue;
+    NameResolver.ConfigOrError configOrError;
+    Object gracefulConfig;
+    if (clusterConfig.getChildren() instanceof EndpointConfig) {
+      // The LB policy config is provided in service_config.proto/JSON format.
+      configOrError =
+              GracefulSwitchLoadBalancer.parseLoadBalancingPolicyConfig(
+                      Arrays.asList(clusterConfig.getClusterResource().lbPolicyConfig()),
+                      lbRegistry);
+      if (configOrError.getError() != null) {
+        // Should be impossible, because XdsClusterResource validated this
+        return fail(Status.INTERNAL.withDescription(
+                errorPrefix() + "Unable to parse the LB config: " + configOrError.getError()));
       }
-      CdsUpdate result = leafConfigOr.getValue().getClusterResource();
+      CdsUpdate result = clusterConfig.getClusterResource();
       DiscoveryMechanism instance;
       if (result.clusterType() == ClusterType.EDS) {
         instance = DiscoveryMechanism.forEds(
-            leafName,
+            clusterName,
             result.edsServiceName(),
             result.lrsServerInfo(),
             result.maxConcurrentRequests(),
@@ -144,45 +125,49 @@ public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {
             result.outlierDetection());
       } else {
         instance = DiscoveryMechanism.forLogicalDns(
-            leafName,
+            clusterName,
             result.dnsHostName(),
             result.lrsServerInfo(),
             result.maxConcurrentRequests(),
             result.upstreamTlsContext(),
             result.filterMetadata());
       }
-      instances.add(instance);
-    }
-    if (instances.isEmpty()) {
-      return fail(noneFoundError);
-    }
-
-    // The LB policy config is provided in service_config.proto/JSON format.
-    NameResolver.ConfigOrError configOrError =
-        GracefulSwitchLoadBalancer.parseLoadBalancingPolicyConfig(
-            Arrays.asList(clusterConfig.getClusterResource().lbPolicyConfig()), lbRegistry);
-    if (configOrError.getError() != null) {
-      // Should be impossible, because XdsClusterResource validated this
+      gracefulConfig = GracefulSwitchLoadBalancer.createLoadBalancingPolicyConfig(
+          lbRegistry.getProvider(CLUSTER_RESOLVER_POLICY_NAME),
+          new ClusterResolverConfig(
+              instance,
+              configOrError.getConfig(),
+              clusterConfig.getClusterResource().isHttp11ProxyAvailable()));
+    } else if (clusterConfig.getChildren() instanceof AggregateConfig) {
+      Map<String, PriorityChildConfig> priorityChildConfigs = new HashMap<>();
+      List<String> leafClusters = ((AggregateConfig) clusterConfig.getChildren()).getLeafNames();
+      for (String childCluster: leafClusters) {
+        priorityChildConfigs.put(childCluster,
+                new PriorityChildConfig(
+                        GracefulSwitchLoadBalancer.createLoadBalancingPolicyConfig(
+                                lbRegistry.getProvider(CDS_POLICY_NAME),
+                                new CdsConfig(childCluster)),
+                        false));
+      }
+      gracefulConfig = GracefulSwitchLoadBalancer.createLoadBalancingPolicyConfig(
+          lbRegistry.getProvider(PRIORITY_POLICY_NAME),
+          new PriorityLoadBalancerProvider.PriorityLbConfig(
+              Collections.unmodifiableMap(priorityChildConfigs), leafClusters));
+    } else {
       return fail(Status.INTERNAL.withDescription(
-          errorPrefix() + "Unable to parse the LB config: " + configOrError.getError()));
+              errorPrefix() + "Unexpected cluster children type: "
+                      + clusterConfig.getChildren().getClass()));
     }
 
-    ClusterResolverConfig config = new ClusterResolverConfig(
-        Collections.unmodifiableList(instances),
-        configOrError.getConfig(),
-        clusterConfig.getClusterResource().isHttp11ProxyAvailable());
-    if (childLb == null) {
-      childLb = lbRegistry.getProvider(CLUSTER_RESOLVER_POLICY_NAME).newLoadBalancer(helper);
-    }
-    return childLb.acceptResolvedAddresses(
-        resolvedAddresses.toBuilder().setLoadBalancingPolicyConfig(config).build());
+    return delegate.acceptResolvedAddresses(
+        resolvedAddresses.toBuilder().setLoadBalancingPolicyConfig(gracefulConfig).build());
   }
 
   @Override
   public void handleNameResolutionError(Status error) {
     logger.log(XdsLogLevel.WARNING, "Received name resolution error: {0}", error);
-    if (childLb != null) {
-      childLb.handleNameResolutionError(error);
+    if (delegate != null) {
+      delegate.handleNameResolutionError(error);
     } else {
       helper.updateBalancingState(
           TRANSIENT_FAILURE, new FixedResultPicker(PickResult.withError(error)));
@@ -192,10 +177,8 @@ public void handleNameResolutionError(Status error) {
   @Override
   public void shutdown() {
     logger.log(XdsLogLevel.INFO, "Shutdown");
-    if (childLb != null) {
-      childLb.shutdown();
-      childLb = null;
-    }
+    delegate.shutdown();
+    delegate = new GracefulSwitchLoadBalancer(helper);
     if (clusterSubscription != null) {
       clusterSubscription.close();
       clusterSubscription = null;
@@ -204,10 +187,7 @@ public void shutdown() {
 
   @CheckReturnValue // don't forget to return up the stack after the fail call
   private Status fail(Status error) {
-    if (childLb != null) {
-      childLb.shutdown();
-      childLb = null;
-    }
+    delegate.shutdown();
     helper.updateBalancingState(
         TRANSIENT_FAILURE, new FixedResultPicker(PickResult.withError(error)));
     return Status.OK; // XdsNameResolver isn't a polling NR, so this value doesn't matter

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

@@ -23,6 +23,7 @@
 import io.grpc.LoadBalancer;
 import io.grpc.LoadBalancer.Helper;
 import io.grpc.LoadBalancerProvider;
+import io.grpc.LoadBalancerRegistry;
 import io.grpc.NameResolver.ConfigOrError;
 import io.grpc.Status;
 import io.grpc.internal.JsonUtil;
@@ -51,9 +52,24 @@ public String getPolicyName() {
     return XdsLbPolicies.CDS_POLICY_NAME;
   }
 
+  private final LoadBalancerRegistry loadBalancerRegistry;
+
+  public CdsLoadBalancerProvider() {
+    this.loadBalancerRegistry = null;
+  }
+
+  public CdsLoadBalancerProvider(LoadBalancerRegistry loadBalancerRegistry) {
+    this.loadBalancerRegistry = loadBalancerRegistry;
+  }
+
   @Override
   public LoadBalancer newLoadBalancer(Helper helper) {
-    return new CdsLoadBalancer2(helper);
+    LoadBalancerRegistry loadBalancerRegistry = this.loadBalancerRegistry;
+    if (loadBalancerRegistry == null) {
+      loadBalancerRegistry = LoadBalancerRegistry.getDefaultRegistry();
+    }
+
+    return new CdsLoadBalancer2(helper, loadBalancerRegistry);
   }
 
   @Override

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

@@ -168,8 +168,8 @@ public LoadBalancer newLoadBalancer(Helper helper) {
    */
   private final class ClusterResolverLbState extends LoadBalancer {
     private final Helper helper;
-    private final List<String> clusters = new ArrayList<>();
-    private final Map<String, ClusterState> clusterStates = new HashMap<>();
+    private ClusterState clusterState;
+    private String cluster;
     private Object endpointLbConfig;
     private ResolvedAddresses resolvedAddresses;
     private LoadBalancer childLb;
@@ -185,21 +185,18 @@ public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {
       ClusterResolverConfig config =
           (ClusterResolverConfig) resolvedAddresses.getLoadBalancingPolicyConfig();
       endpointLbConfig = config.lbConfig;
-      for (DiscoveryMechanism instance : config.discoveryMechanisms) {
-        clusters.add(instance.cluster);
-        ClusterState state;
-        if (instance.type == DiscoveryMechanism.Type.EDS) {
-          state = new EdsClusterState(instance.cluster, instance.edsServiceName,
-              instance.lrsServerInfo, instance.maxConcurrentRequests, instance.tlsContext,
-              instance.filterMetadata, instance.outlierDetection);
-        } else {  // logical DNS
-          state = new LogicalDnsClusterState(instance.cluster, instance.dnsHostName,
-              instance.lrsServerInfo, instance.maxConcurrentRequests, instance.tlsContext,
-              instance.filterMetadata);
-        }
-        clusterStates.put(instance.cluster, state);
-        state.start();
-      }
+      DiscoveryMechanism instance = config.discoveryMechanism;
+      cluster = instance.cluster;
+      if (instance.type == DiscoveryMechanism.Type.EDS) {
+        clusterState = new EdsClusterState(instance.cluster, instance.edsServiceName,
+            instance.lrsServerInfo, instance.maxConcurrentRequests, instance.tlsContext,
+            instance.filterMetadata, instance.outlierDetection);
+      } else {  // logical DNS
+        clusterState = new LogicalDnsClusterState(instance.cluster, instance.dnsHostName,
+            instance.lrsServerInfo, instance.maxConcurrentRequests, instance.tlsContext,
+            instance.filterMetadata);
+      }
+      clusterState.start();
       return Status.OK;
     }
 
@@ -215,9 +212,7 @@ public void handleNameResolutionError(Status error) {
 
     @Override
     public void shutdown() {
-      for (ClusterState state : clusterStates.values()) {
-        state.shutdown();
-      }
+      clusterState.shutdown();
       if (childLb != null) {
         childLb.shutdown();
       }
@@ -229,24 +224,21 @@ private void handleEndpointResourceUpdate() {
       List<String> priorities = new ArrayList<>();  // totally ordered priority list
 
       Status endpointNotFound = Status.OK;
-      for (String cluster : clusters) {
-        ClusterState state = clusterStates.get(cluster);
-        // Propagate endpoints to the child LB policy only after all clusters have been resolved.
-        if (!state.resolved && state.status.isOk()) {
-          return;
-        }
-        if (state.result != null) {
-          addresses.addAll(state.result.addresses);
-          priorityChildConfigs.putAll(state.result.priorityChildConfigs);
-          priorities.addAll(state.result.priorities);
-        } else {
-          endpointNotFound = state.status;
-        }
+      // Propagate endpoints to the child LB policy only after all clusters have been resolved.
+      if (!clusterState.resolved && clusterState.status.isOk()) {
+        return;
+      }
+      if (clusterState.result != null) {
+        addresses.addAll(clusterState.result.addresses);
+        priorityChildConfigs.putAll(clusterState.result.priorityChildConfigs);
+        priorities.addAll(clusterState.result.priorities);
+      } else {
+        endpointNotFound = clusterState.status;
       }
       if (addresses.isEmpty()) {
         if (endpointNotFound.isOk()) {
           endpointNotFound = Status.UNAVAILABLE.withDescription(
-              "No usable endpoint from cluster(s): " + clusters);
+              "No usable endpoint from cluster: " + cluster);
         } else {
           endpointNotFound =
               Status.UNAVAILABLE.withCause(endpointNotFound.getCause())
@@ -274,22 +266,12 @@ private void handleEndpointResourceUpdate() {
     }
 
     private void handleEndpointResolutionError() {
-      boolean allInError = true;
-      Status error = null;
-      for (String cluster : clusters) {
-        ClusterState state = clusterStates.get(cluster);
-        if (state.status.isOk()) {
-          allInError = false;
-        } else {
-          error = state.status;
-        }
-      }
-      if (allInError) {
+      if (!clusterState.status.isOk()) {
         if (childLb != null) {
-          childLb.handleNameResolutionError(error);
+          childLb.handleNameResolutionError(clusterState.status);
         } else {
           helper.updateBalancingState(
-              TRANSIENT_FAILURE, new FixedResultPicker(PickResult.withError(error)));
+              TRANSIENT_FAILURE, new FixedResultPicker(PickResult.withError(clusterState.status)));
         }
       }
     }
@@ -306,10 +288,8 @@ private RefreshableHelper(Helper delegate) {
 
       @Override
       public void refreshNameResolution() {
-        for (ClusterState state : clusterStates.values()) {
-          if (state instanceof LogicalDnsClusterState) {
-            ((LogicalDnsClusterState) state).refresh();
-          }
+        if (clusterState instanceof LogicalDnsClusterState) {
+          ((LogicalDnsClusterState) clusterState).refresh();
         }
       }