xds: Fix cluster selection races when updating config selector

ejona86 · web-flow · commit d82613a74cbe · 2025-03-07T10:33:35.000-08:00
Listener2.onResult() doesn't require running in the sync context, so when called from the sync context it is guaranteed not to do its processing immediately (instead, it schedules work into the sync context). The code was doing an update dance: 1) update service config to add new cluster, 2) update config selector to use new cluster, 3) update service config to remove old clusters. But the onResult() wasn't being processed immediately, so the actual execution order was 2, 1, 3 which has a small window where RPCs will fail. But onResult2() does run immediately. And since ca4819a, updateBalancingState() updates the picker immediately. cleanUpRoutes() was also racy because it updated the routingConfig before swapping to the new config selector, so RPCs could fail saying there was no route instead of the useful error message. Even with the opposite order, some RPCs may be executing the while loop of selectConfig(), trying to acquire a cluster. The code unreffed the clusters before updating the routingConfig, so those RPCs could go into a tight loop until the routingConfig was updated. Also, once the routingConfig was updated to EMPTY those RPCs would similarly see the wrong error message. To give the correct error message, selectConfig() must fail such RPCs directly, and once it can do that there's no need to stop using the config selector in error cases. This has the benefit of fewer moving parts and more consistent threading among cases. The added test was able to detect the race 2% of the time. The slower the code/machine, the more reliable the test failed. ca4819a along with this commit reduced it to 0 failures in 1000 runs. Discovered when investigating b/394850611
diff --git a/xds/src/main/java/io/grpc/xds/XdsNameResolver.java b/xds/src/main/java/io/grpc/xds/XdsNameResolver.java
@@ -132,7 +132,7 @@ final class XdsNameResolver extends NameResolver {
   // NamedFilterConfig.filterStateKey -> filter_instance.
   private final HashMap<String, Filter> activeFilters = new HashMap<>();
 
-  private volatile RoutingConfig routingConfig = RoutingConfig.EMPTY;
+  private volatile RoutingConfig routingConfig;
   private Listener2 listener;
   private ObjectPool<XdsClient> xdsClientPool;
   private XdsClient xdsClient;
@@ -306,7 +306,7 @@ private void updateResolutionResult() {
 
     if (logger.isLoggable(XdsLogLevel.INFO)) {
       logger.log(
-          XdsLogLevel.INFO, "Generated service config:\n{0}", new Gson().toJson(rawServiceConfig));
+          XdsLogLevel.INFO, "Generated service config: {0}", new Gson().toJson(rawServiceConfig));
     }
     ConfigOrError parsedServiceConfig = serviceConfigParser.parseServiceConfig(rawServiceConfig);
     Attributes attrs =
@@ -320,7 +320,7 @@ private void updateResolutionResult() {
             .setAttributes(attrs)
             .setServiceConfig(parsedServiceConfig)
             .build();
-    listener.onResult(result);
+    listener.onResult2(result);
     receivedConfig = true;
   }
 
@@ -395,6 +395,9 @@ public Result selectConfig(PickSubchannelArgs args) {
       String path = "/" + args.getMethodDescriptor().getFullMethodName();
       do {
         routingCfg = routingConfig;
+        if (routingCfg.errorStatus != null) {
+          return Result.forError(routingCfg.errorStatus);
+        }
         selectedRoute = null;
         for (RouteData route : routingCfg.routes) {
           if (RoutingUtils.matchRoute(route.routeMatch, path, headers, random)) {
@@ -626,19 +629,6 @@ private static String prefixedClusterSpecifierPluginName(String pluginName) {
     return "cluster_specifier_plugin:" + pluginName;
   }
 
-  private static final class FailingConfigSelector extends InternalConfigSelector {
-    private final Result result;
-
-    public FailingConfigSelector(Status error) {
-      this.result = Result.forError(error);
-    }
-
-    @Override
-    public Result selectConfig(PickSubchannelArgs args) {
-      return result;
-    }
-  }
-
   private class ResolveState implements ResourceWatcher<XdsListenerResource.LdsUpdate> {
     private final ConfigOrError emptyServiceConfig =
         serviceConfigParser.parseServiceConfig(Collections.<String, Object>emptyMap());
@@ -835,13 +825,13 @@ private void updateRoutes(List<VirtualHost> virtualHosts, long httpMaxStreamDura
         }
       }
       // Update service config to include newly added clusters.
-      if (shouldUpdateResult) {
+      if (shouldUpdateResult && routingConfig != null) {
         updateResolutionResult();
+        shouldUpdateResult = false;
       }
       // Make newly added clusters selectable by config selector and deleted clusters no longer
       // selectable.
       routingConfig = new RoutingConfig(httpMaxStreamDurationNano, routesData.build());
-      shouldUpdateResult = false;
       for (String cluster : deletedClusters) {
         int count = clusterRefs.get(cluster).refCount.decrementAndGet();
         if (count == 0) {
@@ -893,6 +883,9 @@ private ClientInterceptor createFilters(
     }
 
     private void cleanUpRoutes(String error) {
+      String errorWithNodeId =
+          error + ", xDS node ID: " + xdsClient.getBootstrapInfo().node().getId();
+      routingConfig = new RoutingConfig(Status.UNAVAILABLE.withDescription(errorWithNodeId));
       if (existingClusters != null) {
         for (String cluster : existingClusters) {
           int count = clusterRefs.get(cluster).refCount.decrementAndGet();
@@ -902,17 +895,12 @@ private void cleanUpRoutes(String error) {
         }
         existingClusters = null;
       }
-      routingConfig = RoutingConfig.EMPTY;
+
       // Without addresses the default LB (normally pick_first) should become TRANSIENT_FAILURE, and
-      // the config selector handles the error message itself. Once the LB API allows providing
-      // failure information for addresses yet still providing a service config, the config seector
-      // could be avoided.
-      String errorWithNodeId =
-          error + ", xDS node ID: " + xdsClient.getBootstrapInfo().node().getId();
-      listener.onResult(ResolutionResult.newBuilder()
+      // the config selector handles the error message itself.
+      listener.onResult2(ResolutionResult.newBuilder()
           .setAttributes(Attributes.newBuilder()
-            .set(InternalConfigSelector.KEY,
-              new FailingConfigSelector(Status.UNAVAILABLE.withDescription(errorWithNodeId)))
+            .set(InternalConfigSelector.KEY, configSelector)
             .build())
           .setServiceConfig(emptyServiceConfig)
           .build());
@@ -983,12 +971,19 @@ public void onResourceDoesNotExist(final String resourceName) {
   private static class RoutingConfig {
     private final long fallbackTimeoutNano;
     final ImmutableList<RouteData> routes;
-
-    private static final RoutingConfig EMPTY = new RoutingConfig(0, ImmutableList.of());
+    final Status errorStatus;
 
     private RoutingConfig(long fallbackTimeoutNano, ImmutableList<RouteData> routes) {
       this.fallbackTimeoutNano = fallbackTimeoutNano;
       this.routes = checkNotNull(routes, "routes");
+      this.errorStatus = null;
+    }
+
+    private RoutingConfig(Status errorStatus) {
+      this.fallbackTimeoutNano = 0;
+      this.routes = null;
+      this.errorStatus = checkNotNull(errorStatus, "errorStatus");
+      checkArgument(!errorStatus.isOk(), "errorStatus should not be okay");
     }
   }
 
diff --git a/xds/src/test/java/io/grpc/xds/FakeControlPlaneXdsIntegrationTest.java b/xds/src/test/java/io/grpc/xds/FakeControlPlaneXdsIntegrationTest.java
@@ -19,9 +19,12 @@
 
 import static com.google.common.truth.Truth.assertThat;
 import static io.grpc.xds.DataPlaneRule.ENDPOINT_HOST_NAME;
+import static io.grpc.xds.XdsTestControlPlaneService.ADS_TYPE_URL_CDS;
+import static io.grpc.xds.XdsTestControlPlaneService.ADS_TYPE_URL_EDS;
 import static org.junit.Assert.assertEquals;
 
 import com.github.xds.type.v3.TypedStruct;
+import com.google.common.collect.ImmutableMap;
 import com.google.protobuf.Any;
 import com.google.protobuf.Struct;
 import com.google.protobuf.Value;
@@ -36,11 +39,17 @@
 import io.envoyproxy.envoy.config.endpoint.v3.Endpoint;
 import io.envoyproxy.envoy.config.endpoint.v3.LbEndpoint;
 import io.envoyproxy.envoy.config.endpoint.v3.LocalityLbEndpoints;
+import io.envoyproxy.envoy.config.route.v3.Route;
+import io.envoyproxy.envoy.config.route.v3.RouteAction;
+import io.envoyproxy.envoy.config.route.v3.RouteConfiguration;
+import io.envoyproxy.envoy.config.route.v3.RouteMatch;
+import io.envoyproxy.envoy.config.route.v3.VirtualHost;
 import io.envoyproxy.envoy.extensions.load_balancing_policies.wrr_locality.v3.WrrLocality;
 import io.grpc.CallOptions;
 import io.grpc.Channel;
 import io.grpc.ClientCall;
 import io.grpc.ClientInterceptor;
+import io.grpc.ClientStreamTracer;
 import io.grpc.ForwardingClientCall.SimpleForwardingClientCall;
 import io.grpc.ForwardingClientCallListener;
 import io.grpc.LoadBalancerRegistry;
@@ -89,8 +98,7 @@ public void pingPong() throws Exception {
     ManagedChannel channel = dataPlane.getManagedChannel();
     SimpleServiceGrpc.SimpleServiceBlockingStub blockingStub = SimpleServiceGrpc.newBlockingStub(
         channel);
-    SimpleRequest request = SimpleRequest.newBuilder()
-        .build();
+    SimpleRequest request = SimpleRequest.getDefaultInstance();
     SimpleResponse goldenResponse = SimpleResponse.newBuilder()
         .setResponseMessage("Hi, xDS! Authority= test-server")
         .build();
@@ -104,8 +112,7 @@ public void pingPong_edsEndpoint_authorityOverride() throws Exception {
       ManagedChannel channel = dataPlane.getManagedChannel();
       SimpleServiceGrpc.SimpleServiceBlockingStub blockingStub = SimpleServiceGrpc.newBlockingStub(
           channel);
-      SimpleRequest request = SimpleRequest.newBuilder()
-          .build();
+      SimpleRequest request = SimpleRequest.getDefaultInstance();
       SimpleResponse goldenResponse = SimpleResponse.newBuilder()
           .setResponseMessage("Hi, xDS! Authority= " + ENDPOINT_HOST_NAME)
           .build();
@@ -145,8 +152,7 @@ public void pingPong_metadataLoadBalancer() throws Exception {
       // We add an interceptor to catch the response headers from the server.
       SimpleServiceGrpc.SimpleServiceBlockingStub blockingStub = SimpleServiceGrpc.newBlockingStub(
           dataPlane.getManagedChannel()).withInterceptors(responseHeaderInterceptor);
-      SimpleRequest request = SimpleRequest.newBuilder()
-          .build();
+      SimpleRequest request = SimpleRequest.getDefaultInstance();
       SimpleResponse goldenResponse = SimpleResponse.newBuilder()
           .setResponseMessage("Hi, xDS! Authority= test-server")
           .build();
@@ -160,6 +166,100 @@ public void pingPong_metadataLoadBalancer() throws Exception {
     }
   }
 
+  // Try to trigger "UNAVAILABLE: CDS encountered error: unable to find available subchannel for
+  // cluster cluster:cluster1" race, if XdsNameResolver updates its ConfigSelector before
+  // cluster_manager config.
+  @Test
+  public void changeClusterForRoute() throws Exception {
+    // Start with route to cluster0
+    InetSocketAddress edsInetSocketAddress
+        = (InetSocketAddress) dataPlane.getServer().getListenSockets().get(0);
+    controlPlane.getService().setXdsConfig(
+        ADS_TYPE_URL_EDS,
+        ImmutableMap.of(
+          "eds-service-0",
+          ControlPlaneRule.buildClusterLoadAssignment(
+              edsInetSocketAddress.getHostName(), "", edsInetSocketAddress.getPort(),
+              "eds-service-0"),
+          "eds-service-1",
+          ControlPlaneRule.buildClusterLoadAssignment(
+              edsInetSocketAddress.getHostName(), "", edsInetSocketAddress.getPort(),
+              "eds-service-1")));
+    controlPlane.getService().setXdsConfig(
+        ADS_TYPE_URL_CDS,
+        ImmutableMap.of(
+            "cluster0",
+            ControlPlaneRule.buildCluster("cluster0", "eds-service-0"),
+            "cluster1",
+            ControlPlaneRule.buildCluster("cluster1", "eds-service-1")));
+    controlPlane.setRdsConfig(RouteConfiguration.newBuilder()
+        .setName("route-config.googleapis.com")
+        .addVirtualHosts(VirtualHost.newBuilder()
+          .addDomains("test-server")
+          .addRoutes(Route.newBuilder()
+            .setMatch(RouteMatch.newBuilder().setPrefix("/").build())
+            .setRoute(RouteAction.newBuilder().setCluster("cluster0").build())
+            .build())
+          .build())
+        .build());
+
+    class ClusterClientStreamTracer extends ClientStreamTracer {
+      boolean usedCluster1;
+
+      @Override
+      public void addOptionalLabel(String key, String value) {
+        if ("grpc.lb.backend_service".equals(key)) {
+          usedCluster1 = "cluster1".equals(value);
+        }
+      }
+    }
+
+    ClusterClientStreamTracer tracer = new ClusterClientStreamTracer();
+    ClientStreamTracer.Factory tracerFactory = new ClientStreamTracer.Factory() {
+      @Override
+      public ClientStreamTracer newClientStreamTracer(
+          ClientStreamTracer.StreamInfo info, Metadata headers) {
+        return tracer;
+      }
+    };
+    ClientInterceptor tracerInterceptor = new ClientInterceptor() {
+      @Override
+      public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
+          MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
+        return next.newCall(method, callOptions.withStreamTracerFactory(tracerFactory));
+      }
+    };
+    SimpleServiceGrpc.SimpleServiceBlockingStub stub = SimpleServiceGrpc
+        .newBlockingStub(dataPlane.getManagedChannel())
+        .withInterceptors(tracerInterceptor);
+    SimpleRequest request = SimpleRequest.getDefaultInstance();
+    SimpleResponse goldenResponse = SimpleResponse.newBuilder()
+        .setResponseMessage("Hi, xDS! Authority= test-server")
+        .build();
+    assertThat(stub.unaryRpc(request)).isEqualTo(goldenResponse);
+    assertThat(tracer.usedCluster1).isFalse();
+
+    // Check for errors when swapping route to cluster1
+    controlPlane.setRdsConfig(RouteConfiguration.newBuilder()
+        .setName("route-config.googleapis.com")
+        .addVirtualHosts(VirtualHost.newBuilder()
+          .addDomains("test-server")
+          .addRoutes(Route.newBuilder()
+            .setMatch(RouteMatch.newBuilder().setPrefix("/").build())
+            .setRoute(RouteAction.newBuilder().setCluster("cluster1").build())
+            .build())
+          .build())
+        .build());
+
+    for (int j = 0; j < 10; j++) {
+      stub.unaryRpc(request);
+      if (tracer.usedCluster1) {
+        break;
+      }
+    }
+    assertThat(tracer.usedCluster1).isTrue();
+  }
+
   // Captures response headers from the server.
   private static class ResponseHeaderClientInterceptor implements ClientInterceptor {
     Metadata reponseHeaders;
@@ -199,8 +299,7 @@ public void pingPong_ringHash() {
     ManagedChannel channel = dataPlane.getManagedChannel();
     SimpleServiceGrpc.SimpleServiceBlockingStub blockingStub = SimpleServiceGrpc.newBlockingStub(
         channel);
-    SimpleRequest request = SimpleRequest.newBuilder()
-        .build();
+    SimpleRequest request = SimpleRequest.getDefaultInstance();
     SimpleResponse goldenResponse = SimpleResponse.newBuilder()
         .setResponseMessage("Hi, xDS! Authority= test-server")
         .build();
@@ -231,8 +330,7 @@ public void pingPong_logicalDns_authorityOverride() {
       ManagedChannel channel = dataPlane.getManagedChannel();
       SimpleServiceGrpc.SimpleServiceBlockingStub blockingStub = SimpleServiceGrpc.newBlockingStub(
           channel);
-      SimpleRequest request = SimpleRequest.newBuilder()
-          .build();
+      SimpleRequest request = SimpleRequest.getDefaultInstance();
       SimpleResponse goldenResponse = SimpleResponse.newBuilder()
           .setResponseMessage("Hi, xDS! Authority= localhost:" + serverAddress.getPort())
           .build();
diff --git a/xds/src/test/java/io/grpc/xds/XdsNameResolverTest.java b/xds/src/test/java/io/grpc/xds/XdsNameResolverTest.java
