xds: ClusterManagerLB must update child configuration

While child LB policies are unlikey to change for each cluster name (RLS
returns regular cluster names, so should be unique), and the
configuration for CDS policies won't change, RLS configuration can
definitely change.

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

@@ -23,11 +23,11 @@ import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.MoreObjects;
 import io.grpc.ConnectivityState;
 import io.grpc.InternalLogId;
-import io.grpc.LoadBalancerProvider;
+import io.grpc.LoadBalancer;
 import io.grpc.Status;
 import io.grpc.SynchronizationContext;
 import io.grpc.SynchronizationContext.ScheduledHandle;
-import io.grpc.internal.ServiceConfigUtil.PolicySelection;
+import io.grpc.util.GracefulSwitchLoadBalancer;
 import io.grpc.util.MultiChildLoadBalancer;
 import io.grpc.xds.ClusterManagerLoadBalancerProvider.ClusterManagerConfig;
 import io.grpc.xds.client.XdsLogger;
@@ -71,7 +71,10 @@ class ClusterManagerLoadBalancer extends MultiChildLoadBalancer {
 
   @Override
   protected ResolvedAddresses getChildAddresses(Object key, ResolvedAddresses resolvedAddresses,
-      Object childConfig) {
+      Object unusedChildConfig) {
+    ClusterManagerConfig config = (ClusterManagerConfig)
+        resolvedAddresses.getLoadBalancingPolicyConfig();
+    Object childConfig = config.childPolicies.get(key);
     return resolvedAddresses.toBuilder().setLoadBalancingPolicyConfig(childConfig).build();
   }
 
@@ -81,13 +84,12 @@ class ClusterManagerLoadBalancer extends MultiChildLoadBalancer {
         resolvedAddresses.getLoadBalancingPolicyConfig();
     Map<Object, ChildLbState> newChildPolicies = new HashMap<>();
     if (config != null) {
-      for (Entry<String, PolicySelection> entry : config.childPolicies.entrySet()) {
-        ChildLbState child = getChildLbState(entry.getKey());
+      for (String key : config.childPolicies.keySet()) {
+        ChildLbState child = getChildLbState(key);
         if (child == null) {
-          child = new ClusterManagerLbState(entry.getKey(),
-              entry.getValue().getProvider(), entry.getValue().getConfig());
+          child = new ClusterManagerLbState(key, GracefulSwitchLoadBalancerFactory.INSTANCE, null);
         }
-        newChildPolicies.put(entry.getKey(), child);
+        newChildPolicies.put(key, child);
       }
     }
     logger.log(
@@ -108,8 +110,8 @@ class ClusterManagerLoadBalancer extends MultiChildLoadBalancer {
           resolvedAddresses.getLoadBalancingPolicyConfig();
       ClusterManagerConfig lastConfig = (ClusterManagerConfig)
           lastResolvedAddresses.getLoadBalancingPolicyConfig();
-      Map<String, PolicySelection> adjChildPolicies = new HashMap<>(config.childPolicies);
-      for (Entry<String, PolicySelection> entry : lastConfig.childPolicies.entrySet()) {
+      Map<String, Object> adjChildPolicies = new HashMap<>(config.childPolicies);
+      for (Entry<String, Object> entry : lastConfig.childPolicies.entrySet()) {
         ClusterManagerLbState state = (ClusterManagerLbState) getChildLbState(entry.getKey());
         if (adjChildPolicies.containsKey(entry.getKey())) {
           if (state.deletionTimer != null) {
@@ -202,9 +204,9 @@ class ClusterManagerLoadBalancer extends MultiChildLoadBalancer {
     @Nullable
     ScheduledHandle deletionTimer;
 
-    public ClusterManagerLbState(Object key, LoadBalancerProvider policyProvider,
+    public ClusterManagerLbState(Object key, LoadBalancer.Factory policyFactory,
         Object childConfig) {
-      super(key, policyProvider, childConfig);
+      super(key, policyFactory, childConfig);
     }
 
     @Override
@@ -237,8 +239,8 @@ class ClusterManagerLoadBalancer extends MultiChildLoadBalancer {
         public void run() {
           ClusterManagerConfig config = (ClusterManagerConfig)
               lastResolvedAddresses.getLoadBalancingPolicyConfig();
-          Map<String, PolicySelection> childPolicies = new HashMap<>(config.childPolicies);
-          PolicySelection removed = childPolicies.remove(getKey());
+          Map<String, Object> childPolicies = new HashMap<>(config.childPolicies);
+          Object removed = childPolicies.remove(getKey());
           assert removed != null;
           config = new ClusterManagerConfig(childPolicies);
           lastResolvedAddresses =
@@ -276,4 +278,13 @@ class ClusterManagerLoadBalancer extends MultiChildLoadBalancer {
       }
     }
   }
+
+  static final class GracefulSwitchLoadBalancerFactory extends LoadBalancer.Factory {
+    static final LoadBalancer.Factory INSTANCE = new GracefulSwitchLoadBalancerFactory();
+
+    @Override
+    public LoadBalancer newLoadBalancer(LoadBalancer.Helper helper) {
+      return new GracefulSwitchLoadBalancer(helper);
+    }
+  }
 }

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

@@ -26,12 +26,9 @@ import io.grpc.LoadBalancerRegistry;
 import io.grpc.NameResolver.ConfigOrError;
 import io.grpc.Status;
 import io.grpc.internal.JsonUtil;
-import io.grpc.internal.ServiceConfigUtil;
-import io.grpc.internal.ServiceConfigUtil.LbConfig;
-import io.grpc.internal.ServiceConfigUtil.PolicySelection;
+import io.grpc.util.GracefulSwitchLoadBalancer;
 import java.util.Collections;
 import java.util.LinkedHashMap;
-import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import javax.annotation.Nullable;
@@ -73,7 +70,7 @@ public class ClusterManagerLoadBalancerProvider extends LoadBalancerProvider {
 
   @Override
   public ConfigOrError parseLoadBalancingPolicyConfig(Map<String, ?> rawConfig) {
-    Map<String, PolicySelection> parsedChildPolicies = new LinkedHashMap<>();
+    Map<String, Object> parsedChildPolicies = new LinkedHashMap<>();
     try {
       Map<String, ?> childPolicies = JsonUtil.getObject(rawConfig, "childPolicy");
       if (childPolicies == null || childPolicies.isEmpty()) {
@@ -86,27 +83,19 @@ public class ClusterManagerLoadBalancerProvider extends LoadBalancerProvider {
           return ConfigOrError.fromError(Status.INTERNAL.withDescription(
               "No config for child " + name + " in cluster_manager LB policy: " + rawConfig));
         }
-        List<LbConfig> childConfigCandidates =
-            ServiceConfigUtil.unwrapLoadBalancingConfigList(
-                JsonUtil.getListOfObjects(childPolicy, "lbPolicy"));
-        if (childConfigCandidates == null || childConfigCandidates.isEmpty()) {
-          return ConfigOrError.fromError(Status.INTERNAL.withDescription(
-              "No config specified for child " + name + " in cluster_manager Lb policy: "
-                  + rawConfig));
-        }
         LoadBalancerRegistry registry =
             lbRegistry != null ? lbRegistry : LoadBalancerRegistry.getDefaultRegistry();
-        ConfigOrError selectedConfig =
-            ServiceConfigUtil.selectLbPolicyFromList(childConfigCandidates, registry);
-        if (selectedConfig.getError() != null) {
-          Status error = selectedConfig.getError();
+        ConfigOrError childConfig = GracefulSwitchLoadBalancer.parseLoadBalancingPolicyConfig(
+            JsonUtil.getListOfObjects(childPolicy, "lbPolicy"), registry);
+        if (childConfig.getError() != null) {
+          Status error = childConfig.getError();
           return ConfigOrError.fromError(
               Status.INTERNAL
                   .withCause(error.getCause())
                   .withDescription(error.getDescription())
-                  .augmentDescription("Failed to select config for child " + name));
+                  .augmentDescription("Failed to parse config for child " + name));
         }
-        parsedChildPolicies.put(name, (PolicySelection) selectedConfig.getConfig());
+        parsedChildPolicies.put(name, childConfig.getConfig());
       }
     } catch (RuntimeException e) {
       return ConfigOrError.fromError(
@@ -122,9 +111,9 @@ public class ClusterManagerLoadBalancerProvider extends LoadBalancerProvider {
   }
 
   static class ClusterManagerConfig {
-    final Map<String, PolicySelection> childPolicies;
+    final Map<String, Object> childPolicies;
 
-    ClusterManagerConfig(Map<String, PolicySelection> childPolicies) {
+    ClusterManagerConfig(Map<String, Object> childPolicies) {
       this.childPolicies = Collections.unmodifiableMap(childPolicies);
     }
 

xds/src/test/java/io/grpc/xds/ClusterManagerLoadBalancerProviderTest.java

@@ -26,7 +26,7 @@ import io.grpc.NameResolver.ConfigOrError;
 import io.grpc.Status;
 import io.grpc.Status.Code;
 import io.grpc.internal.JsonParser;
-import io.grpc.internal.ServiceConfigUtil.PolicySelection;
+import io.grpc.util.GracefulSwitchLoadBalancer;
 import io.grpc.xds.ClusterManagerLoadBalancerProvider.ClusterManagerConfig;
 import java.io.IOException;
 import java.util.Map;
@@ -133,10 +133,9 @@ public class ClusterManagerLoadBalancerProviderTest {
     assertThat(config.childPolicies)
         .containsExactly(
             "child1",
-            new PolicySelection(
-                lbProviderFoo, fooConfig),
+            GracefulSwitchLoadBalancer.createLoadBalancingPolicyConfig(lbProviderFoo, fooConfig),
             "child2",
-            new PolicySelection(lbProviderBar, barConfig));
+            GracefulSwitchLoadBalancer.createLoadBalancingPolicyConfig(lbProviderBar, barConfig));
   }
 
   @Test

xds/src/test/java/io/grpc/xds/ClusterManagerLoadBalancerTest.java

@@ -52,10 +52,11 @@ import io.grpc.Status.Code;
 import io.grpc.SynchronizationContext;
 import io.grpc.internal.FakeClock;
 import io.grpc.internal.PickSubchannelArgsImpl;
-import io.grpc.internal.ServiceConfigUtil.PolicySelection;
 import io.grpc.testing.TestMethodDescriptors;
+import io.grpc.util.GracefulSwitchLoadBalancer;
 import io.grpc.xds.ClusterManagerLoadBalancerProvider.ClusterManagerConfig;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.LinkedHashMap;
@@ -288,16 +289,27 @@ public class ClusterManagerLoadBalancerTest {
             .build());
   }
 
+  // Prevent ClusterManagerLB from detecting different providers even when the configuration is the
+  // same.
+  private Map<List<Object>, FakeLoadBalancerProvider> fakeLoadBalancerProviderCache
+      = new HashMap<>();
+
   private ClusterManagerConfig buildConfig(Map<String, String> childPolicies, boolean failing) {
-    Map<String, PolicySelection> childPolicySelections = new LinkedHashMap<>();
+    Map<String, Object> childConfigs = new LinkedHashMap<>();
     for (String name : childPolicies.keySet()) {
       String childPolicyName = childPolicies.get(name);
       Object childConfig = lbConfigInventory.get(name);
-      PolicySelection policy =
-          new PolicySelection(new FakeLoadBalancerProvider(childPolicyName, failing), childConfig);
-      childPolicySelections.put(name, policy);
+      FakeLoadBalancerProvider lbProvider =
+          fakeLoadBalancerProviderCache.get(Arrays.asList(childPolicyName, failing));
+      if (lbProvider == null) {
+        lbProvider = new FakeLoadBalancerProvider(childPolicyName, failing);
+        fakeLoadBalancerProviderCache.put(Arrays.asList(childPolicyName, failing), lbProvider);
+      }
+      Object policy =
+          GracefulSwitchLoadBalancer.createLoadBalancingPolicyConfig(lbProvider, childConfig);
+      childConfigs.put(name, policy);
     }
-    return new ClusterManagerConfig(childPolicySelections);
+    return new ClusterManagerConfig(childConfigs);
   }
 
   private static PickResult pickSubchannel(SubchannelPicker picker, String clusterName) {