xds: implement least_request load balancing policy (#8739)

Implements least_request_experimental as defined by
[A48](https://github.com/grpc/proposal/blob/master/A48-xds-least-request-lb-policy.md)

These tests are mostly just a copy of
RoundRobinLoadBalancerTest.
The main difference is currently in the pickerLeastRequest test case.
All other tests should be the same.

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

@@ -32,6 +32,7 @@
 import io.grpc.xds.CdsLoadBalancerProvider.CdsConfig;
 import io.grpc.xds.ClusterResolverLoadBalancerProvider.ClusterResolverConfig;
 import io.grpc.xds.ClusterResolverLoadBalancerProvider.ClusterResolverConfig.DiscoveryMechanism;
+import io.grpc.xds.LeastRequestLoadBalancer.LeastRequestConfig;
 import io.grpc.xds.RingHashLoadBalancer.RingHashConfig;
 import io.grpc.xds.XdsClient.CdsResourceWatcher;
 import io.grpc.xds.XdsClient.CdsUpdate;
@@ -190,6 +191,10 @@ private void handleClusterDiscovered() {
         lbProvider = lbRegistry.getProvider("ring_hash_experimental");
         lbConfig = new RingHashConfig(root.result.minRingSize(), root.result.maxRingSize());
       }
+      if (root.result.lbPolicy() == LbPolicy.LEAST_REQUEST) {
+        lbProvider = lbRegistry.getProvider("least_request_experimental");
+        lbConfig = new LeastRequestConfig(root.result.choiceCount());
+      }
       if (lbProvider == null) {
         lbProvider = lbRegistry.getProvider("round_robin");
         lbConfig = null;

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

@@ -44,6 +44,7 @@
 import io.envoyproxy.envoy.config.cluster.v3.Cluster.CustomClusterType;
 import io.envoyproxy.envoy.config.cluster.v3.Cluster.DiscoveryType;
 import io.envoyproxy.envoy.config.cluster.v3.Cluster.LbPolicy;
+import io.envoyproxy.envoy.config.cluster.v3.Cluster.LeastRequestLbConfig;
 import io.envoyproxy.envoy.config.cluster.v3.Cluster.RingHashLbConfig;
 import io.envoyproxy.envoy.config.core.v3.HttpProtocolOptions;
 import io.envoyproxy.envoy.config.core.v3.RoutingPriority;
@@ -140,6 +141,8 @@ final class ClientXdsClient extends XdsClient implements XdsResponseHandler, Res
   @VisibleForTesting
   static final long DEFAULT_RING_HASH_LB_POLICY_MAX_RING_SIZE = 8 * 1024 * 1024L;
   @VisibleForTesting
+  static final int DEFAULT_LEAST_REQUEST_CHOICE_COUNT = 2;
+  @VisibleForTesting
   static final long MAX_RING_HASH_LB_POLICY_RING_SIZE = 8 * 1024 * 1024L;
   @VisibleForTesting
   static final String AGGREGATE_CLUSTER_TYPE_NAME = "envoy.clusters.aggregate";
@@ -161,6 +164,11 @@ final class ClientXdsClient extends XdsClient implements XdsResponseHandler, Res
   static boolean enableRouteLookup =
       !Strings.isNullOrEmpty(System.getenv("GRPC_EXPERIMENTAL_XDS_RLS_LB"))
           && Boolean.parseBoolean(System.getenv("GRPC_EXPERIMENTAL_XDS_RLS_LB"));
+  @VisibleForTesting
+  static boolean enableLeastRequest =
+      !Strings.isNullOrEmpty(System.getenv("GRPC_EXPERIMENTAL_ENABLE_LEAST_REQUEST"))
+          ? Boolean.parseBoolean(System.getenv("GRPC_EXPERIMENTAL_ENABLE_LEAST_REQUEST"))
+          : Boolean.parseBoolean(System.getProperty("io.grpc.xds.experimentalEnableLeastRequest"));
   private static final String TYPE_URL_HTTP_CONNECTION_MANAGER_V2 =
       "type.googleapis.com/envoy.config.filter.network.http_connection_manager.v2"
           + ".HttpConnectionManager";
@@ -1616,6 +1624,17 @@ static CdsUpdate processCluster(Cluster cluster, Set<String> retainedEdsResource
       updateBuilder.ringHashLbPolicy(minRingSize, maxRingSize);
     } else if (cluster.getLbPolicy() == LbPolicy.ROUND_ROBIN) {
       updateBuilder.roundRobinLbPolicy();
+    } else if (enableLeastRequest && cluster.getLbPolicy() == LbPolicy.LEAST_REQUEST) {
+      LeastRequestLbConfig lbConfig =  cluster.getLeastRequestLbConfig();
+      int choiceCount =
+              lbConfig.hasChoiceCount()
+                ? lbConfig.getChoiceCount().getValue()
+                : DEFAULT_LEAST_REQUEST_CHOICE_COUNT;
+      if (choiceCount < DEFAULT_LEAST_REQUEST_CHOICE_COUNT) {
+        throw new ResourceInvalidException(
+                "Cluster " + cluster.getName() + ": invalid least_request_lb_config: " + lbConfig);
+      }
+      updateBuilder.leastRequestLbPolicy(choiceCount);
     } else {
       throw new ResourceInvalidException(
           "Cluster " + cluster.getName() + ": unsupported lb policy: " + cluster.getLbPolicy());

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

@@ -672,7 +672,7 @@ private static PriorityChildConfig generateDnsBasedPriorityChildConfig(
    * Generates configs to be used in the priority LB policy for priorities in an EDS cluster.
    *
    * <p>priority LB -> cluster_impl LB (one per priority) -> (weighted_target LB
-   * -> round_robin (one per locality)) / ring_hash_experimental
+   * -> round_robin / least_request_experimental (one per locality)) / ring_hash_experimental
    */
   private static Map<String, PriorityChildConfig> generateEdsBasedPriorityChildConfigs(
       String cluster, @Nullable String edsServiceName, @Nullable ServerInfo lrsServerInfo,
@@ -684,13 +684,14 @@ private static Map<String, PriorityChildConfig> generateEdsBasedPriorityChildCon
     for (String priority : prioritizedLocalityWeights.keySet()) {
       PolicySelection leafPolicy =  endpointLbPolicy;
       // Depending on the endpoint-level load balancing policy, different LB hierarchy may be
-      // created. If the endpoint-level LB policy is round_robin, it creates a two-level LB
-      // hierarchy: a locality-level LB policy that balances load according to locality weights
-      // followed by an endpoint-level LB policy that simply rounds robin the endpoints within
-      // the locality. If the endpoint-level LB policy is ring_hash_experimental, it creates
-      // a unified LB policy that balances load by weighing the product of each endpoint's weight
-      // and the weight of the locality it belongs to.
-      if (endpointLbPolicy.getProvider().getPolicyName().equals("round_robin")) {
+      // created. If the endpoint-level LB policy is round_robin or least_request_experimental,
+      // it creates a two-level LB hierarchy: a locality-level LB policy that balances load
+      // according to locality weights followed by an endpoint-level LB policy that balances load
+      // between endpoints within the locality. If the endpoint-level LB policy is
+      // ring_hash_experimental, it creates a unified LB policy that balances load by weighing the
+      // product of each endpoint's weight and the weight of the locality it belongs to.
+      if (endpointLbPolicy.getProvider().getPolicyName().equals("round_robin")
+          || endpointLbPolicy.getProvider().getPolicyName().equals("least_request_experimental")) {
         Map<Locality, Integer> localityWeights = prioritizedLocalityWeights.get(priority);
         Map<String, WeightedPolicySelection> targets = new HashMap<>();
         for (Locality locality : localityWeights.keySet()) {

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

@@ -68,7 +68,8 @@ public LoadBalancer newLoadBalancer(Helper helper) {
   static final class ClusterResolverConfig {
     // Ordered list of clusters to be resolved.
     final List<DiscoveryMechanism> discoveryMechanisms;
-    // Endpoint-level load balancing policy with config (round_robin or ring_hash_experimental).
+    // Endpoint-level load balancing policy with config
+    // (round_robin, least_request_experimental or ring_hash_experimental).
     final PolicySelection lbPolicy;
 
     ClusterResolverConfig(List<DiscoveryMechanism> discoveryMechanisms, PolicySelection lbPolicy) {