xds: generate EDS LB config with hardcoded locality picking policy (#7443)

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

@@ -20,9 +20,9 @@ import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkNotNull;
 import static io.grpc.ConnectivityState.TRANSIENT_FAILURE;
 import static io.grpc.xds.XdsLbPolicies.EDS_POLICY_NAME;
+import static io.grpc.xds.XdsLbPolicies.WEIGHTED_TARGET_POLICY_NAME;
 
 import com.google.common.annotations.VisibleForTesting;
-import com.google.common.collect.ImmutableMap;
 import io.grpc.EquivalentAddressGroup;
 import io.grpc.InternalLogId;
 import io.grpc.LoadBalancer;
@@ -198,18 +198,20 @@ final class CdsLoadBalancer extends LoadBalancer {
             xdsClient, newUpdate.getClusterName(), newUpdate.getEdsServiceName(),
             newUpdate.getLbPolicy(), newUpdate.getLrsServerName() != null);
       }
+      // FIXME(chengyuanzhang): handle error correctly to avoid being unnecessarily fragile.
       checkArgument(
           newUpdate.getLbPolicy().equals("round_robin"), "can only support round_robin policy");
-
-      LoadBalancerProvider lbProvider = lbRegistry.getProvider(newUpdate.getLbPolicy());
-      Object lbConfig =
-          lbProvider.parseLoadBalancingPolicyConfig(ImmutableMap.<String, Object>of()).getConfig();
+      LoadBalancerProvider endpointPickingPolicyProvider =
+          lbRegistry.getProvider(newUpdate.getLbPolicy());
+      LoadBalancerProvider localityPickingPolicyProvider =
+          lbRegistry.getProvider(WEIGHTED_TARGET_POLICY_NAME);  // hardcode to weighted-target
       final EdsConfig edsConfig =
           new EdsConfig(
               /* clusterName = */ newUpdate.getClusterName(),
               /* edsServiceName = */ newUpdate.getEdsServiceName(),
               /* lrsServerName = */ newUpdate.getLrsServerName(),
-              new PolicySelection(lbProvider, ImmutableMap.<String, Object>of(), lbConfig));
+              new PolicySelection(localityPickingPolicyProvider, null, null /* by EDS policy */),
+              new PolicySelection(endpointPickingPolicyProvider, null, null));
       if (isXdsSecurityEnabled()) {
         updateSslContextProviderSupplier(newUpdate.getUpstreamTlsContext());
       }

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

@@ -17,7 +17,6 @@
 package io.grpc.xds;
 
 import static com.google.common.base.Preconditions.checkNotNull;
-import static io.grpc.xds.XdsLbPolicies.WEIGHTED_TARGET_POLICY_NAME;
 
 import com.google.common.base.MoreObjects;
 import com.google.common.base.Objects;
@@ -25,7 +24,6 @@ import io.grpc.Internal;
 import io.grpc.LoadBalancer;
 import io.grpc.LoadBalancer.Helper;
 import io.grpc.LoadBalancerProvider;
-import io.grpc.LoadBalancerRegistry;
 import io.grpc.NameResolver.ConfigOrError;
 import io.grpc.internal.ServiceConfigUtil.PolicySelection;
 import java.util.Map;
@@ -74,21 +72,6 @@ public class EdsLoadBalancerProvider extends LoadBalancerProvider {
     final PolicySelection localityPickingPolicy;
     final PolicySelection endpointPickingPolicy;
 
-    // TODO(chengyuanzhang): delete me.
-    EdsConfig(
-        String clusterName,
-        @Nullable String edsServiceName,
-        @Nullable String lrsServerName,
-        PolicySelection endpointPickingPolicy) {
-      this.clusterName = checkNotNull(clusterName, "clusterName");
-      this.edsServiceName = edsServiceName;
-      this.lrsServerName = lrsServerName;
-      this.endpointPickingPolicy = checkNotNull(endpointPickingPolicy, "endpointPickingPolicy");
-      LoadBalancerProvider provider =
-          LoadBalancerRegistry.getDefaultRegistry().getProvider(WEIGHTED_TARGET_POLICY_NAME);
-      localityPickingPolicy = new PolicySelection(provider, null, null);
-    }
-
     EdsConfig(
         String clusterName,
         @Nullable String edsServiceName,

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

@@ -89,6 +89,7 @@ public class CdsLoadBalancerTest {
     MockitoAnnotations.initMocks(this);
 
     LoadBalancerRegistry registry = new LoadBalancerRegistry();
+    registry.register(new FakeLoadBalancerProvider(XdsLbPolicies.WEIGHTED_TARGET_POLICY_NAME));
     registry.register(new FakeLoadBalancerProvider(XdsLbPolicies.EDS_POLICY_NAME));
     registry.register(new FakeLoadBalancerProvider("round_robin"));
     ObjectPool<XdsClient> xdsClientPool = new ObjectPool<XdsClient>() {
@@ -135,6 +136,8 @@ public class CdsLoadBalancerTest {
     assertThat(edsConfig.clusterName).isEqualTo(CLUSTER);
     assertThat(edsConfig.edsServiceName).isNull();
     assertThat(edsConfig.lrsServerName).isNull();
+    assertThat(edsConfig.localityPickingPolicy.getProvider().getPolicyName())
+        .isEqualTo(XdsLbPolicies.WEIGHTED_TARGET_POLICY_NAME);  // hardcoded to weighted-target
     assertThat(edsConfig.endpointPickingPolicy.getProvider().getPolicyName())
         .isEqualTo("round_robin");
   }
@@ -174,6 +177,8 @@ public class CdsLoadBalancerTest {
     assertThat(edsConfig.clusterName).isEqualTo(CLUSTER);
     assertThat(edsConfig.edsServiceName).isNull();
     assertThat(edsConfig.lrsServerName).isNull();
+    assertThat(edsConfig.localityPickingPolicy.getProvider().getPolicyName())
+        .isEqualTo(XdsLbPolicies.WEIGHTED_TARGET_POLICY_NAME);  // hardcoded to weighted-target
     assertThat(edsConfig.endpointPickingPolicy.getProvider().getPolicyName())
         .isEqualTo("round_robin");
 
@@ -185,6 +190,8 @@ public class CdsLoadBalancerTest {
     assertThat(edsConfig.clusterName).isEqualTo(CLUSTER);
     assertThat(edsConfig.edsServiceName).isEqualTo(edsService);
     assertThat(edsConfig.lrsServerName).isEqualTo(loadReportServer);
+    assertThat(edsConfig.localityPickingPolicy.getProvider().getPolicyName())
+        .isEqualTo(XdsLbPolicies.WEIGHTED_TARGET_POLICY_NAME);  // hardcoded to weighted-target
     assertThat(edsConfig.endpointPickingPolicy.getProvider().getPolicyName())
         .isEqualTo("round_robin");
   }

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

@@ -143,6 +143,8 @@ public class EdsLoadBalancerTest {
     }
   };
 
+  private final PolicySelection fakeLocalityPickingPolicy =
+      new PolicySelection(mock(LoadBalancerProvider.class), null, null);
   private final PolicySelection fakeEndpointPickingPolicy =
       new PolicySelection(mock(LoadBalancerProvider.class), null, new Object());
 
@@ -272,7 +274,7 @@ public class EdsLoadBalancerTest {
 
   @Test
   public void handleNameResolutionErrorBeforeAndAfterEdsWorkding() {
-    deliverResolvedAddresses(null, null, fakeEndpointPickingPolicy);
+    deliverResolvedAddresses(null, null, fakeLocalityPickingPolicy, fakeEndpointPickingPolicy);
 
     // handleResolutionError() before receiving any endpoint update.
     edsLb.handleNameResolutionError(Status.DATA_LOSS.withDescription("fake status"));
@@ -300,7 +302,8 @@ public class EdsLoadBalancerTest {
   public void handleEdsServiceNameChange() {
     assertThat(childHelpers).isEmpty();
 
-    deliverResolvedAddresses("edsServiceName1", null, fakeEndpointPickingPolicy);
+    deliverResolvedAddresses("edsServiceName1", null, fakeLocalityPickingPolicy,
+        fakeEndpointPickingPolicy);
     ClusterLoadAssignment clusterLoadAssignment =
         buildClusterLoadAssignmentV2("edsServiceName1",
             ImmutableList.of(
@@ -320,7 +323,8 @@ public class EdsLoadBalancerTest {
     assertLatestConnectivityState(CONNECTING);
 
     // Change edsServicename to edsServiceName2.
-    deliverResolvedAddresses("edsServiceName2", null, fakeEndpointPickingPolicy);
+    deliverResolvedAddresses("edsServiceName2", null, fakeLocalityPickingPolicy,
+        fakeEndpointPickingPolicy);
     // The old balancer was not READY, so it will be shutdown immediately.
     verify(childBalancer1).shutdown();
 
@@ -350,7 +354,8 @@ public class EdsLoadBalancerTest {
     assertLatestSubchannelPicker(subchannel2);
 
     // Change edsServiceName to edsServiceName3.
-    deliverResolvedAddresses("edsServiceName3", null, fakeEndpointPickingPolicy);
+    deliverResolvedAddresses("edsServiceName3", null, fakeLocalityPickingPolicy,
+        fakeEndpointPickingPolicy);
     clusterLoadAssignment =
         buildClusterLoadAssignmentV2("edsServiceName3",
             ImmutableList.of(
@@ -376,7 +381,8 @@ public class EdsLoadBalancerTest {
     assertLatestConnectivityState(CONNECTING);
 
     // Change edsServiceName to edsServiceName4.
-    deliverResolvedAddresses("edsServiceName4", null, fakeEndpointPickingPolicy);
+    deliverResolvedAddresses("edsServiceName4", null, fakeLocalityPickingPolicy,
+        fakeEndpointPickingPolicy);
     verify(childBalancer3).shutdown();
 
     clusterLoadAssignment =
@@ -404,7 +410,8 @@ public class EdsLoadBalancerTest {
     assertLatestSubchannelPicker(subchannel4);
 
     // Change edsServiceName to edsServiceName5.
-    deliverResolvedAddresses("edsServiceName5", null, fakeEndpointPickingPolicy);
+    deliverResolvedAddresses("edsServiceName5", null, fakeLocalityPickingPolicy,
+        fakeEndpointPickingPolicy);
     clusterLoadAssignment =
         buildClusterLoadAssignmentV2("edsServiceName5",
             ImmutableList.of(
@@ -439,7 +446,7 @@ public class EdsLoadBalancerTest {
 
   @Test
   public void edsResourceUpdate_allDrop() {
-    deliverResolvedAddresses(null, null, fakeEndpointPickingPolicy);
+    deliverResolvedAddresses(null, null, fakeLocalityPickingPolicy, fakeEndpointPickingPolicy);
 
     ClusterLoadAssignment clusterLoadAssignment = buildClusterLoadAssignmentV2(
         CLUSTER_NAME,
@@ -488,7 +495,7 @@ public class EdsLoadBalancerTest {
 
   @Test
   public void edsResourceUpdate_localityAssignmentChange() {
-    deliverResolvedAddresses(null, null, fakeEndpointPickingPolicy);
+    deliverResolvedAddresses(null, null, fakeLocalityPickingPolicy, fakeEndpointPickingPolicy);
 
     LbEndpoint endpoint11 = buildLbEndpointV2("addr11.example.com", 8011, HEALTHY, 11);
     LbEndpoint endpoint12 = buildLbEndpointV2("addr12.example.com", 8012, HEALTHY, 12);
@@ -575,7 +582,8 @@ public class EdsLoadBalancerTest {
     edsLb =
         new EdsLoadBalancer(helper, lbRegistry, localityStoreFactory, bootstrapper, channelFactory);
 
-    deliverResolvedAddresses("edsServiceName1", null, fakeEndpointPickingPolicy);
+    deliverResolvedAddresses("edsServiceName1", null, fakeLocalityPickingPolicy,
+        fakeEndpointPickingPolicy);
     assertThat(localityStores).hasSize(1);
     LocalityStore localityStore = localityStores.peekLast();
 
@@ -612,7 +620,8 @@ public class EdsLoadBalancerTest {
     verify(localityStore).updateLocalityStore(endpointUpdate.getLocalityLbEndpointsMap());
 
     // Change cluster name.
-    deliverResolvedAddresses("edsServiceName2", null, fakeEndpointPickingPolicy);
+    deliverResolvedAddresses("edsServiceName2", null, fakeLocalityPickingPolicy,
+        fakeEndpointPickingPolicy);
     assertThat(localityStores).hasSize(2);
     localityStore = localityStores.peekLast();
 
@@ -635,7 +644,7 @@ public class EdsLoadBalancerTest {
 
   @Test
   public void edsResourceNotExist() {
-    deliverResolvedAddresses(null, null, fakeEndpointPickingPolicy);
+    deliverResolvedAddresses(null, null, fakeLocalityPickingPolicy, fakeEndpointPickingPolicy);
 
     // Forwarding 20 seconds so that the xds client will deem EDS resource not available.
     fakeClock.forwardTime(20, TimeUnit.SECONDS);
@@ -649,7 +658,7 @@ public class EdsLoadBalancerTest {
 
   @Test
   public void edsResourceRemoved() {
-    deliverResolvedAddresses(null, null, fakeEndpointPickingPolicy);
+    deliverResolvedAddresses(null, null, fakeLocalityPickingPolicy, fakeEndpointPickingPolicy);
     ClusterLoadAssignment clusterLoadAssignment =
         buildClusterLoadAssignmentV2(CLUSTER_NAME,
             ImmutableList.of(
@@ -695,7 +704,7 @@ public class EdsLoadBalancerTest {
 
   @Test
   public void transientError_noPreviousEndpointUpdateReceived() {
-    deliverResolvedAddresses(null, null, fakeEndpointPickingPolicy);
+    deliverResolvedAddresses(null, null, fakeLocalityPickingPolicy, fakeEndpointPickingPolicy);
 
     // Forwarding 20 seconds so that the xds client will deem EDS resource not available.
     fakeClock.forwardTime(20, TimeUnit.SECONDS);
@@ -704,7 +713,7 @@ public class EdsLoadBalancerTest {
 
   @Test
   public void transientError_withPreviousEndpointUpdateReceived() {
-    deliverResolvedAddresses(null, null, fakeEndpointPickingPolicy);
+    deliverResolvedAddresses(null, null, fakeLocalityPickingPolicy, fakeEndpointPickingPolicy);
     // Endpoint update received.
     ClusterLoadAssignment clusterLoadAssignment =
         buildClusterLoadAssignmentV2(CLUSTER_NAME,
@@ -751,9 +760,11 @@ public class EdsLoadBalancerTest {
   private void deliverResolvedAddresses(
       @Nullable String edsServiceName,
       @Nullable String lrsServerName,
+      PolicySelection locaityPickingPolicy,
       PolicySelection endpointPickingPolicy) {
     EdsConfig config =
-        new EdsConfig(CLUSTER_NAME, edsServiceName, lrsServerName, endpointPickingPolicy);
+        new EdsConfig(CLUSTER_NAME, edsServiceName, lrsServerName, locaityPickingPolicy,
+            endpointPickingPolicy);
     ResolvedAddresses.Builder resolvedAddressBuilder = ResolvedAddresses.newBuilder()
         .setAddresses(ImmutableList.<EquivalentAddressGroup>of())
         .setLoadBalancingPolicyConfig(config);