rls: support routeLookupChannelServiceConfig in RLS lb config

Implementing the latest change for RLS lb config.

```
The configuration for the LB policy will be of the following form:

{
  "routeLookupConfig": <JSON form of RouteLookupConfig proto>,
  "routeLookupChannelServiceConfig": {...service config JSON...},
  "childPolicy": [
    {"<policy name>": {...child policy config...}}
  ],
  "childPolicyConfigTargetFieldName": "<name of field>"
}
```

>If the routeLookupChannelServiceConfig field is present, we will pass the specified service config to the RLS control plane channel, and we will disable fetching service config via that channel's resolver.

rls/src/main/java/io/grpc/rls/CachingRlsLbClient.java

@@ -23,8 +23,6 @@ import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Converter;
 import com.google.common.base.MoreObjects;
 import com.google.common.base.MoreObjects.ToStringHelper;
-import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableMap;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.SettableFuture;
 import io.grpc.ChannelLogger;
@@ -83,13 +81,6 @@ final class CachingRlsLbClient {
   private static final Converter<RouteLookupResponse, io.grpc.lookup.v1.RouteLookupResponse>
       RESPONSE_CONVERTER = new RouteLookupResponseConverter().reverse();
 
-  // System property to use direct path enabled OobChannel, by default direct path is enabled.
-  private static final String RLS_ENABLE_OOB_CHANNEL_DIRECTPATH_PROPERTY =
-      "io.grpc.rls.CachingRlsLbClient.enable_oobchannel_directpath";
-  @VisibleForTesting
-  static boolean enableOobChannelDirectPath =
-      Boolean.parseBoolean(System.getProperty(RLS_ENABLE_OOB_CHANNEL_DIRECTPATH_PROPERTY, "false"));
-
   // All cache status changes (pending, backoff, success) must be under this lock
   private final Object lock = new Object();
   // LRU cache based on access order (BACKOFF and actual data will be here)
@@ -158,14 +149,14 @@ final class CachingRlsLbClient {
     ManagedChannelBuilder<?> rlsChannelBuilder = helper.createResolvingOobChannelBuilder(
         rlsConfig.getLookupService(), helper.getUnsafeChannelCredentials());
     rlsChannelBuilder.overrideAuthority(helper.getAuthority());
-    if (enableOobChannelDirectPath) {
-      Map<String, ?> directPathServiceConfig =
-          getDirectPathServiceConfig(rlsConfig.getLookupService());
+    Map<String, ?> routeLookupChannelServiceConfig =
+        lbPolicyConfig.getRouteLookupChannelServiceConfig();
+    if (routeLookupChannelServiceConfig != null) {
       logger.log(
           ChannelLogLevel.DEBUG,
-          "RLS channel direct path enabled. RLS channel service config: {0}",
-          directPathServiceConfig);
-      rlsChannelBuilder.defaultServiceConfig(directPathServiceConfig);
+          "RLS channel service config: {0}",
+          routeLookupChannelServiceConfig);
+      rlsChannelBuilder.defaultServiceConfig(routeLookupChannelServiceConfig);
       rlsChannelBuilder.disableServiceConfigLookUp();
     }
     rlsChannel = rlsChannelBuilder.build();
@@ -184,21 +175,6 @@ final class CachingRlsLbClient {
     logger.log(ChannelLogLevel.DEBUG, "CachingRlsLbClient created");
   }
 
-  private static ImmutableMap<String, Object> getDirectPathServiceConfig(String serviceName) {
-    ImmutableMap<String, Object> pickFirstStrategy =
-        ImmutableMap.<String, Object>of("pick_first", ImmutableMap.of());
-
-    ImmutableMap<String, Object> childPolicy =
-        ImmutableMap.<String, Object>of(
-            "childPolicy", ImmutableList.of(pickFirstStrategy),
-            "serviceName", serviceName);
-
-    ImmutableMap<String, Object> grpcLbPolicy =
-        ImmutableMap.<String, Object>of("grpclb", childPolicy);
-
-    return ImmutableMap.<String, Object>of("loadBalancingConfig", ImmutableList.of(grpcLbPolicy));
-  }
-
   @CheckReturnValue
   private ListenableFuture<RouteLookupResponse> asyncRlsCall(RouteLookupRequest request) {
     final SettableFuture<RouteLookupResponse> response = SettableFuture.create();

rls/src/main/java/io/grpc/rls/LbPolicyConfiguration.java

@@ -48,11 +48,15 @@ import javax.annotation.Nullable;
 final class LbPolicyConfiguration {
 
   private final RouteLookupConfig routeLookupConfig;
+  @Nullable
+  private final Map<String, ?> routeLookupChannelServiceConfig;
   private final ChildLoadBalancingPolicy policy;
 
   LbPolicyConfiguration(
-      RouteLookupConfig routeLookupConfig, ChildLoadBalancingPolicy policy) {
+      RouteLookupConfig routeLookupConfig, @Nullable Map<String, ?> routeLookupChannelServiceConfig,
+      ChildLoadBalancingPolicy policy) {
     this.routeLookupConfig = checkNotNull(routeLookupConfig, "routeLookupConfig");
+    this.routeLookupChannelServiceConfig = routeLookupChannelServiceConfig;
     this.policy = checkNotNull(policy, "policy");
   }
 
@@ -60,6 +64,11 @@ final class LbPolicyConfiguration {
     return routeLookupConfig;
   }
 
+  @Nullable
+  Map<String, ?> getRouteLookupChannelServiceConfig() {
+    return routeLookupChannelServiceConfig;
+  }
+
   ChildLoadBalancingPolicy getLoadBalancingPolicy() {
     return policy;
   }
@@ -74,18 +83,20 @@ final class LbPolicyConfiguration {
     }
     LbPolicyConfiguration that = (LbPolicyConfiguration) o;
     return Objects.equals(routeLookupConfig, that.routeLookupConfig)
+        && Objects.equals(routeLookupChannelServiceConfig, that.routeLookupChannelServiceConfig)
         && Objects.equals(policy, that.policy);
   }
 
   @Override
   public int hashCode() {
-    return Objects.hash(routeLookupConfig, policy);
+    return Objects.hash(routeLookupConfig, routeLookupChannelServiceConfig, policy);
   }
 
   @Override
   public String toString() {
     return MoreObjects.toStringHelper(this)
         .add("routeLookupConfig", routeLookupConfig)
+        .add("routeLookupChannelServiceConfig", routeLookupChannelServiceConfig)
         .add("policy", policy)
         .toString();
   }

rls/src/main/java/io/grpc/rls/RlsLoadBalancerProvider.java

@@ -62,12 +62,15 @@ public final class RlsLoadBalancerProvider extends LoadBalancerProvider {
     try {
       RouteLookupConfig routeLookupConfig = new RouteLookupConfigConverter()
           .convert(JsonUtil.getObject(rawLoadBalancingConfigPolicy, "routeLookupConfig"));
+      Map<String, ?> routeLookupChannelServiceConfig =
+          JsonUtil.getObject(rawLoadBalancingConfigPolicy, "routeLookupChannelServiceConfig");
       ChildLoadBalancingPolicy lbPolicy = ChildLoadBalancingPolicy
           .create(
               JsonUtil.getString(rawLoadBalancingConfigPolicy, "childPolicyConfigTargetFieldName"),
               JsonUtil.checkObjectList(
                   checkNotNull(JsonUtil.getList(rawLoadBalancingConfigPolicy, "childPolicy"))));
-      return ConfigOrError.fromConfig(new LbPolicyConfiguration(routeLookupConfig, lbPolicy));
+      return ConfigOrError.fromConfig(
+          new LbPolicyConfiguration(routeLookupConfig, routeLookupChannelServiceConfig, lbPolicy));
     } catch (Exception e) {
       return ConfigOrError.fromError(
           Status.INVALID_ARGUMENT

rls/src/test/java/io/grpc/rls/CachingRlsLbClientTest.java

@@ -90,7 +90,6 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import javax.annotation.Nonnull;
 import org.junit.After;
-import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -144,19 +143,12 @@ public class CachingRlsLbClientTest {
       mock(Helper.class, AdditionalAnswers.delegatesTo(new FakeHelper()));
   private final FakeThrottler fakeThrottler = new FakeThrottler();
   private final LbPolicyConfiguration lbPolicyConfiguration =
-      new LbPolicyConfiguration(ROUTE_LOOKUP_CONFIG, childLbPolicy);
+      new LbPolicyConfiguration(ROUTE_LOOKUP_CONFIG, null, childLbPolicy);
 
   private CachingRlsLbClient rlsLbClient;
-  private boolean existingEnableOobChannelDirectPath;
   private Map<String, ?> rlsChannelServiceConfig;
   private String rlsChannelOverriddenAuthority;
 
-  @Before
-  public void setUp() throws Exception {
-    existingEnableOobChannelDirectPath = CachingRlsLbClient.enableOobChannelDirectPath;
-    CachingRlsLbClient.enableOobChannelDirectPath = false;
-  }
-
   private void setUpRlsLbClient() {
     rlsLbClient =
         CachingRlsLbClient.newBuilder()
@@ -173,7 +165,6 @@ public class CachingRlsLbClientTest {
   @After
   public void tearDown() throws Exception {
     rlsLbClient.close();
-    CachingRlsLbClient.enableOobChannelDirectPath = existingEnableOobChannelDirectPath;
     assertWithMessage(
             "On client shut down, RlsLoadBalancer must shut down with all its child loadbalancers.")
         .that(lbProvider.loadBalancers).isEmpty();
@@ -244,9 +235,29 @@ public class CachingRlsLbClientTest {
   }
 
   @Test
-  public void rls_overDirectPath() throws Exception {
-    CachingRlsLbClient.enableOobChannelDirectPath = true;
-    setUpRlsLbClient();
+  public void rls_withCustomRlsChannelServiceConfig() throws Exception {
+    Map<String, ?> routeLookupChannelServiceConfig =
+        ImmutableMap.of(
+            "loadBalancingConfig",
+            ImmutableList.of(ImmutableMap.of(
+                "grpclb",
+                ImmutableMap.of(
+                    "childPolicy",
+                    ImmutableList.of(ImmutableMap.of("pick_first", ImmutableMap.of())),
+                    "serviceName",
+                    "service1"))));
+    LbPolicyConfiguration lbPolicyConfiguration = new LbPolicyConfiguration(
+        ROUTE_LOOKUP_CONFIG, routeLookupChannelServiceConfig, childLbPolicy);
+    rlsLbClient =
+        CachingRlsLbClient.newBuilder()
+            .setBackoffProvider(fakeBackoffProvider)
+            .setResolvedAddressesFactory(resolvedAddressFactory)
+            .setEvictionListener(evictionListener)
+            .setHelper(helper)
+            .setLbPolicyConfig(lbPolicyConfiguration)
+            .setThrottler(fakeThrottler)
+            .setTimeProvider(fakeTimeProvider)
+            .build();
     RouteLookupRequest routeLookupRequest = new RouteLookupRequest(ImmutableMap.of(
         "server", "bigtable.googleapis.com", "service-key", "foo", "method-key", "bar"));
     rlsServerImpl.setLookupTable(
@@ -265,16 +276,7 @@ public class CachingRlsLbClientTest {
     assertThat(resp.hasData()).isTrue();
 
     assertThat(rlsChannelOverriddenAuthority).isEqualTo("bigtable.googleapis.com:443");
-    assertThat(rlsChannelServiceConfig).isEqualTo(
-        ImmutableMap.of(
-            "loadBalancingConfig",
-            ImmutableList.of(ImmutableMap.of(
-                "grpclb",
-                ImmutableMap.of(
-                    "childPolicy",
-                    ImmutableList.of(ImmutableMap.of("pick_first", ImmutableMap.of())),
-                    "serviceName",
-                    "service1")))));
+    assertThat(rlsChannelServiceConfig).isEqualTo(routeLookupChannelServiceConfig);
   }
 
   @Test

rls/src/test/java/io/grpc/rls/RlsLoadBalancerTest.java

@@ -118,16 +118,12 @@ public class RlsLoadBalancerTest {
   private MethodDescriptor<Object, Object> fakeSearchMethod;
   private MethodDescriptor<Object, Object> fakeRescueMethod;
   private RlsLoadBalancer rlsLb;
-  private boolean existingEnableOobChannelDirectPath;
   private String defaultTarget = "defaultTarget";
 
   @Before
   public void setUp() throws Exception {
     MockitoAnnotations.initMocks(this);
 
-    existingEnableOobChannelDirectPath = CachingRlsLbClient.enableOobChannelDirectPath;
-    CachingRlsLbClient.enableOobChannelDirectPath = false;
-
     fakeSearchMethod =
         MethodDescriptor.newBuilder()
             .setFullMethodName("com.google/Search")
@@ -168,7 +164,6 @@ public class RlsLoadBalancerTest {
   @After
   public void tearDown() throws Exception {
     rlsLb.shutdown();
-    CachingRlsLbClient.enableOobChannelDirectPath = existingEnableOobChannelDirectPath;
   }
 
   @Test