xds: implement routing in xDS resolver with config selector API (#7275)

Name resolver implementation for performing xDS request routing before the call is made: the resolver emits a config selector to the Channel to let calls make routing decision before delegating to the corresponding cluster load balancer's picker.

This is a branched xDS name resolver implementation. It will replace the existing xDS resolver once the Channel's integration for using config selector is done.

Author: voidzcy

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

@@ -24,7 +24,6 @@
 import static io.grpc.xds.XdsSubchannelPickers.BUFFER_PICKER;
 
 import com.google.common.annotations.VisibleForTesting;
-import io.grpc.CallOptions;
 import io.grpc.ConnectivityState;
 import io.grpc.InternalLogId;
 import io.grpc.LoadBalancer;
@@ -51,9 +50,6 @@ class ClusterManagerLoadBalancer extends LoadBalancer {
 
   @VisibleForTesting
   static final int DELAYED_CHILD_DELETION_TIME_MINUTES = 15;
-  @VisibleForTesting
-  static final CallOptions.Key<String> ROUTING_CLUSTER_NAME_KEY =
-      CallOptions.Key.create("io.grpc.xds.ROUTING_CLUSTER_NAME_KEY");
 
   private final Map<String, ChildLbState> childLbStates = new HashMap<>();
   private final Helper helper;
@@ -148,7 +144,8 @@ private void updateOverallBalancingState() {
       SubchannelPicker picker = new SubchannelPicker() {
         @Override
         public PickResult pickSubchannel(PickSubchannelArgs args) {
-          String clusterName = args.getCallOptions().getOption(ROUTING_CLUSTER_NAME_KEY);
+          String clusterName =
+              args.getCallOptions().getOption(XdsNameResolver2.CLUSTER_SELECTION_KEY);
           SubchannelPicker delegate = childPickers.get(clusterName);
           if (delegate == null) {
             return

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

@@ -1153,6 +1153,9 @@ static StructOrError<RouteAction> fromEnvoyProtoRouteAction(
         case WEIGHTED_CLUSTERS:
           List<io.envoyproxy.envoy.config.route.v3.WeightedCluster.ClusterWeight> clusterWeights
               = proto.getWeightedClusters().getClustersList();
+          if (clusterWeights.isEmpty()) {
+            return StructOrError.fromError("No cluster found in weighted cluster list");
+          }
           weightedClusters = new ArrayList<>();
           for (io.envoyproxy.envoy.config.route.v3.WeightedCluster.ClusterWeight clusterWeight
               : clusterWeights) {

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

@@ -45,6 +45,7 @@ final class RouteMatch {
     this.headerMatchers = headerMatchers;
   }
 
+  @VisibleForTesting
   RouteMatch(@Nullable String pathPrefixMatch, @Nullable String pathExactMatch) {
     this(
         new PathMatcher(pathExactMatch, pathPrefixMatch, null),

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

@@ -29,6 +29,7 @@
 import io.grpc.Status;
 import io.grpc.alts.GoogleDefaultChannelBuilder;
 import io.grpc.internal.ObjectPool;
+import io.grpc.xds.Bootstrapper.BootstrapInfo;
 import io.grpc.xds.Bootstrapper.ChannelCreds;
 import io.grpc.xds.Bootstrapper.ServerInfo;
 import io.grpc.xds.EnvoyProtoData.DropOverload;
@@ -549,6 +550,7 @@ void removeClientStats(String clusterName, @Nullable String clusterServiceName)
     throw new UnsupportedOperationException();
   }
 
+  // TODO(chengyuanzhang): eliminate this factory
   abstract static class XdsClientFactory {
     abstract XdsClient createXdsClient();
   }
@@ -685,4 +687,8 @@ boolean isUseProtocolV3() {
       return useProtocolV3;
     }
   }
+
+  interface XdsClientPoolFactory {
+    ObjectPool<XdsClient> newXdsClientObjectPool(BootstrapInfo bootstrapInfo);
+  }
 }