A75 Aggregate cluster fixes

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

@@ -18,13 +18,16 @@
 
 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 static java.util.Objects.requireNonNull;
 
 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;
+import io.grpc.LoadBalancerProvider;
 import io.grpc.LoadBalancerRegistry;
 import io.grpc.NameResolver;
 import io.grpc.Status;
@@ -33,6 +36,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 +45,10 @@
 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.List;
+import java.util.HashMap;
+import java.util.Map;
 
 /**
  * Load balancer for cds_experimental LB policy. One instance per top-level cluster.
@@ -91,7 +95,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,82 +104,74 @@ 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 childConfig;
+    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,
-            result.edsServiceName(),
-            result.lrsServerInfo(),
-            result.maxConcurrentRequests(),
-            result.upstreamTlsContext(),
-            result.filterMetadata(),
-            result.outlierDetection());
+                clusterName,
+                result.edsServiceName(),
+                result.lrsServerInfo(),
+                result.maxConcurrentRequests(),
+                result.upstreamTlsContext(),
+                result.filterMetadata(),
+                result.outlierDetection());
       } else {
         instance = DiscoveryMechanism.forLogicalDns(
-            leafName,
-            result.dnsHostName(),
-            result.lrsServerInfo(),
-            result.maxConcurrentRequests(),
-            result.upstreamTlsContext(),
-            result.filterMetadata());
+                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
+      childConfig = new ClusterResolverConfig(
+              instance,
+              configOrError.getConfig(),
+              clusterConfig.getClusterResource().isHttp11ProxyAvailable());
+      if (childLb == null) {
+        childLb = lbRegistry.getProvider(CLUSTER_RESOLVER_POLICY_NAME).newLoadBalancer(helper);
+      }
+    } else if (clusterConfig.getChildren() instanceof AggregateConfig) {
+      LoadBalancerProvider priorityLbProvider = lbRegistry.getProvider(PRIORITY_POLICY_NAME);
+      if (childLb == null) {
+        childLb = priorityLbProvider.newLoadBalancer(helper);
+      }
+      Map<String, PriorityChildConfig> priorityChildConfigs = new HashMap<>();
+      for (String childCluster: requireNonNull(
+              clusterConfig.getClusterResource().prioritizedClusterNames())) {
+        priorityChildConfigs.put(childCluster,
+                new PriorityChildConfig(
+                        GracefulSwitchLoadBalancer.createLoadBalancingPolicyConfig(
+                                lbRegistry.getProvider(CDS_POLICY_NAME),
+                                new CdsConfig(childCluster)),
+                        false));
+      }
+      childConfig = new PriorityLoadBalancerProvider.PriorityLbConfig(
+              Collections.unmodifiableMap(priorityChildConfigs),
+              Collections.unmodifiableList(requireNonNull(
+                      clusterConfig.getClusterResource().prioritizedClusterNames())));
+    } 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());
+        resolvedAddresses.toBuilder().setLoadBalancingPolicyConfig(childConfig).build());
   }
 
   @Override

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

@@ -185,21 +185,20 @@ 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;
+      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();
       return Status.OK;
     }
 

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

@@ -30,7 +30,6 @@
 import io.grpc.xds.EnvoyServerProtoData.OutlierDetection;
 import io.grpc.xds.EnvoyServerProtoData.UpstreamTlsContext;
 import io.grpc.xds.client.Bootstrapper.ServerInfo;
-import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import javax.annotation.Nullable;
@@ -70,15 +69,15 @@ public LoadBalancer newLoadBalancer(Helper helper) {
   }
 
   static final class ClusterResolverConfig {
-    // Ordered list of clusters to be resolved.
-    final List<DiscoveryMechanism> discoveryMechanisms;
+    // Clusters to be resolved.
+    final DiscoveryMechanism discoveryMechanism;
     // GracefulSwitch configuration
     final Object lbConfig;
     private final boolean isHttp11ProxyAvailable;
 
-    ClusterResolverConfig(List<DiscoveryMechanism> discoveryMechanisms, Object lbConfig,
-        boolean isHttp11ProxyAvailable) {
-      this.discoveryMechanisms = checkNotNull(discoveryMechanisms, "discoveryMechanisms");
+    ClusterResolverConfig(DiscoveryMechanism discoveryMechanism, Object lbConfig,
+                          boolean isHttp11ProxyAvailable) {
+      this.discoveryMechanism = checkNotNull(discoveryMechanism, "discoveryMechanism");
       this.lbConfig = checkNotNull(lbConfig, "lbConfig");
       this.isHttp11ProxyAvailable = isHttp11ProxyAvailable;
     }
@@ -89,7 +88,7 @@ boolean isHttp11ProxyAvailable() {
 
     @Override
     public int hashCode() {
-      return Objects.hash(discoveryMechanisms, lbConfig);
+      return Objects.hash(discoveryMechanism, lbConfig);
     }
 
     @Override
@@ -101,14 +100,14 @@ public boolean equals(Object o) {
         return false;
       }
       ClusterResolverConfig that = (ClusterResolverConfig) o;
-      return discoveryMechanisms.equals(that.discoveryMechanisms)
+      return discoveryMechanism.equals(that.discoveryMechanism)
           && lbConfig.equals(that.lbConfig);
     }
 
     @Override
     public String toString() {
       return MoreObjects.toStringHelper(this)
-          .add("discoveryMechanisms", discoveryMechanisms)
+          .add("discoveryMechanism", discoveryMechanism)
           .add("lbConfig", lbConfig)
           .toString();
     }