XDS client fallback

Author: larry-safran

xds/src/main/java/io/grpc/xds/client/BootstrapperImpl.java

@@ -41,6 +41,9 @@
 @Internal
 public abstract class BootstrapperImpl extends Bootstrapper {
 
+  public static final String GRPC_EXPERIMENTAL_XDS_FALLBACK =
+      "GRPC_EXPERIMENTAL_XDS_FALLBACK";
+
   // Client features.
   @VisibleForTesting
   public static final String CLIENT_FEATURE_DISABLE_OVERPROVISIONING =
@@ -52,6 +55,9 @@ public abstract class BootstrapperImpl extends Bootstrapper {
   private static final String SERVER_FEATURE_IGNORE_RESOURCE_DELETION = "ignore_resource_deletion";
   private static final String SERVER_FEATURE_TRUSTED_XDS_SERVER = "trusted_xds_server";
 
+  @VisibleForTesting
+  static boolean enableXdsFallback = GrpcUtil.getFlag(GRPC_EXPERIMENTAL_XDS_FALLBACK, false);
+
   protected final XdsLogger logger;
 
   protected FileReader reader = LocalFileReader.INSTANCE;
@@ -65,6 +71,7 @@ protected BootstrapperImpl() {
   protected abstract Object getImplSpecificConfig(Map<String, ?> serverConfig, String serverUri)
       throws XdsInitializationException;
 
+
   /**
    * Reads and parses bootstrap config. The config is expected to be in JSON format.
    */
@@ -103,6 +110,9 @@ protected BootstrapInfo.Builder bootstrapBuilder(Map<String, ?> rawData)
       throw new XdsInitializationException("Invalid bootstrap: 'xds_servers' does not exist.");
     }
     List<ServerInfo> servers = parseServerInfos(rawServerConfigs, logger);
+    if (servers.size() > 1 && !enableXdsFallback) {
+      servers = ImmutableList.of(servers.get(0));
+    }
     builder.servers(servers);
 
     Node.Builder nodeBuilder = Node.newBuilder();
@@ -209,6 +219,9 @@ protected BootstrapInfo.Builder bootstrapBuilder(Map<String, ?> rawData)
         if (rawAuthorityServers == null || rawAuthorityServers.isEmpty()) {
           authorityServers = servers;
         } else {
+          if (rawAuthorityServers.size() > 1 && !enableXdsFallback) {
+            rawAuthorityServers = ImmutableList.of(rawAuthorityServers.get(0));
+          }
           authorityServers = parseServerInfos(rawAuthorityServers, logger);
         }
         authorityInfoMapBuilder.put(

xds/src/main/java/io/grpc/xds/client/ControlPlaneClient.java

@@ -36,10 +36,8 @@
 import io.grpc.xds.client.Bootstrapper.ServerInfo;
 import io.grpc.xds.client.EnvoyProtoData.Node;
 import io.grpc.xds.client.XdsClient.ProcessingTracker;
-import io.grpc.xds.client.XdsClient.ResourceStore;
 import io.grpc.xds.client.XdsClient.XdsResponseHandler;
 import io.grpc.xds.client.XdsLogger.XdsLogLevel;
-import io.grpc.xds.client.XdsTransportFactory.EventHandler;
 import io.grpc.xds.client.XdsTransportFactory.StreamingCall;
 import io.grpc.xds.client.XdsTransportFactory.XdsTransport;
 import java.util.Collection;
@@ -65,40 +63,41 @@ final class ControlPlaneClient {
   private final ServerInfo serverInfo;
   private final XdsTransport xdsTransport;
   private final XdsResponseHandler xdsResponseHandler;
-  private final ResourceStore resourceStore;
+  private final XdsClient.ResourceStore resourceStore;
   private final ScheduledExecutorService timeService;
   private final BackoffPolicy.Provider backoffPolicyProvider;
   private final Stopwatch stopwatch;
   private final Node bootstrapNode;
-  private final XdsClient xdsClient;
 
   // Last successfully applied version_info for each resource type. Starts with empty string.
   // A version_info is used to update management server with client's most recent knowledge of
   // resources.
   private final Map<XdsResourceType<?>, String> versions = new HashMap<>();
 
   private boolean shutdown;
+  private boolean lastStateWasReady;
+  private boolean inError;
+
   @Nullable
   private AdsStream adsStream;
   @Nullable
   private BackoffPolicy retryBackoffPolicy;
   @Nullable
   private ScheduledHandle rpcRetryTimer;
-  private MessagePrettyPrinter messagePrinter;
+  private final MessagePrettyPrinter messagePrinter;
 
   /** An entity that manages ADS RPCs over a single channel. */
   ControlPlaneClient(
       XdsTransport xdsTransport,
       ServerInfo serverInfo,
       Node bootstrapNode,
       XdsResponseHandler xdsResponseHandler,
-      ResourceStore resourceStore,
+      XdsClient.ResourceStore resourceStore,
       ScheduledExecutorService
       timeService,
       SynchronizationContext syncContext,
       BackoffPolicy.Provider backoffPolicyProvider,
       Supplier<Stopwatch> stopwatchSupplier,
-      XdsClient xdsClient,
       MessagePrettyPrinter messagePrinter) {
     this.serverInfo = checkNotNull(serverInfo, "serverInfo");
     this.xdsTransport = checkNotNull(xdsTransport, "xdsTransport");
@@ -108,7 +107,6 @@ final class ControlPlaneClient {
     this.timeService = checkNotNull(timeService, "timeService");
     this.syncContext = checkNotNull(syncContext, "syncContext");
     this.backoffPolicyProvider = checkNotNull(backoffPolicyProvider, "backoffPolicyProvider");
-    this.xdsClient = checkNotNull(xdsClient, "xdsClient");
     this.messagePrinter = checkNotNull(messagePrinter, "messagePrinter");
     stopwatch = checkNotNull(stopwatchSupplier, "stopwatchSupplier").get();
     logId = InternalLogId.allocate("xds-client", serverInfo.target());
@@ -138,6 +136,10 @@ public String toString() {
     return logId.toString();
   }
 
+  public ServerInfo getServerInfo() {
+    return serverInfo;
+  }
+
   /**
    * Updates the resource subscription for the given resource type.
    */
@@ -148,7 +150,15 @@ void adjustResourceSubscription(XdsResourceType<?> resourceType) {
     }
     if (adsStream == null) {
       startRpcStream();
+      // when the stream becomes ready, it will send the discovery requests
+      return;
+    }
+
+    // We will do the rest of the method as part of the readyHandler when the stream is ready.
+    if (!lastStateWasReady) {
+      return;
     }
+
     Collection<String> resources = resourceStore.getSubscribedResources(serverInfo, resourceType);
     if (resources == null) {
       resources = Collections.emptyList();
@@ -203,25 +213,45 @@ boolean isInBackoff() {
 
   // Must be synchronized.
   boolean isReady() {
-    return adsStream != null && adsStream.call != null && adsStream.call.isReady();
+    return adsStream != null && adsStream.call != null
+        && adsStream.call.isReady() && !adsStream.closed;
+  }
+
+  boolean isResponseReceived() {
+    return adsStream != null && adsStream.responseReceived;
+  }
+
+  boolean isConnected() {
+    return lastStateWasReady;
+  }
+
+  boolean isInError() {
+    return inError;
   }
 
+
   /**
    * Starts a timer for each requested resource that hasn't been responded to and
    * has been waiting for the channel to get ready.
    */
   // Must be synchronized.
   void readyHandler() {
     if (!isReady()) {
+      logger.log(XdsLogLevel.DEBUG, "ADS stream ready handler called, but not ready {0}", logId);
       return;
     }
 
-    if (isInBackoff()) {
+    logger.log(XdsLogLevel.DEBUG, "ADS stream ready {0}", logId);
+
+    if (rpcRetryTimer != null) {
       rpcRetryTimer.cancel();
       rpcRetryTimer = null;
     }
 
-    xdsClient.startSubscriberTimersIfNeeded(serverInfo);
+    if (!lastStateWasReady) {
+      lastStateWasReady = true;
+      xdsResponseHandler.handleStreamRestarted(serverInfo);
+    }
   }
 
   /**
@@ -232,27 +262,50 @@ void readyHandler() {
   private void startRpcStream() {
     checkState(adsStream == null, "Previous adsStream has not been cleared yet");
     adsStream = new AdsStream();
+    adsStream.start();
     logger.log(XdsLogLevel.INFO, "ADS stream started");
     stopwatch.reset().start();
   }
 
+  void sendDiscoveryRequests() {
+    if (adsStream == null) {
+      startRpcStream();
+      // when the stream becomes ready, it will send the discovery requests
+      return;
+    }
+
+    if (isConnected()) {
+      adjustAllResourceSubscriptions();
+    }
+  }
+
+  void adjustAllResourceSubscriptions() {
+    if (isInBackoff()) {
+      return;
+    }
+
+    Set<XdsResourceType<?>> subscribedResourceTypes =
+        new HashSet<>(resourceStore.getSubscribedResourceTypesWithTypeUrl().values());
+
+    for (XdsResourceType<?> type : subscribedResourceTypes) {
+      adjustResourceSubscription(type);
+    }
+  }
+
   @VisibleForTesting
   public final class RpcRetryTask implements Runnable {
     @Override
     public void run() {
-      if (shutdown) {
+      logger.log(XdsLogLevel.DEBUG, "Retry timeout. Restart ADS stream {0}", logId);
+      if (shutdown || isReady()) {
         return;
       }
-      startRpcStream();
-      Set<XdsResourceType<?>> subscribedResourceTypes =
-          new HashSet<>(resourceStore.getSubscribedResourceTypesWithTypeUrl().values());
-      for (XdsResourceType<?> type : subscribedResourceTypes) {
-        Collection<String> resources = resourceStore.getSubscribedResources(serverInfo, type);
-        if (resources != null) {
-          adsStream.sendDiscoveryRequest(type, resources);
-        }
+
+      if (adsStream == null) {
+        startRpcStream();
       }
-      xdsResponseHandler.handleStreamRestarted(serverInfo);
+
+      // handling CPC management is triggered in readyHandler
     }
   }
 
@@ -262,7 +315,7 @@ XdsResourceType<?> fromTypeUrl(String typeUrl) {
     return resourceStore.getSubscribedResourceTypesWithTypeUrl().get(typeUrl);
   }
 
-  private class AdsStream implements EventHandler<DiscoveryResponse> {
+  private class AdsStream implements XdsTransportFactory.EventHandler<DiscoveryResponse> {
     private boolean responseReceived;
     private boolean closed;
     // Response nonce for the most recently received discovery responses of each resource type.
@@ -279,6 +332,9 @@ private class AdsStream implements EventHandler<DiscoveryResponse> {
     private AdsStream() {
       this.call = xdsTransport.createStreamingCall(methodDescriptor.getFullMethodName(),
           methodDescriptor.getRequestMarshaller(), methodDescriptor.getResponseMarshaller());
+    }
+
+    void start() {
       call.start(this);
     }
 
@@ -363,10 +419,13 @@ public void onStatusReceived(final Status status) {
     final void handleRpcResponse(XdsResourceType<?> type, String versionInfo, List<Any> resources,
                                  String nonce) {
       checkNotNull(type, "type");
+
       if (closed) {
         return;
       }
+
       responseReceived = true;
+      inError = false;
       respNonces.put(type, nonce);
       ProcessingTracker processingTracker = new ProcessingTracker(
           () -> call.startRecvMessage(), syncContext);
@@ -376,6 +435,10 @@ final void handleRpcResponse(XdsResourceType<?> type, String versionInfo, List<A
     }
 
     private void handleRpcStreamClosed(Status status) {
+      if (this == adsStream || adsStream == null) {
+        lastStateWasReady = false;
+      }
+
       if (closed) {
         return;
       }
@@ -385,13 +448,16 @@ private void handleRpcStreamClosed(Status status) {
         // has never been initialized.
         retryBackoffPolicy = backoffPolicyProvider.get();
       }
+
       // FakeClock in tests isn't thread-safe. Schedule the retry timer before notifying callbacks
       // to avoid TSAN races, since tests may wait until callbacks are called but then would run
       // concurrently with the stopwatch and schedule.
+
       long elapsed = stopwatch.elapsed(TimeUnit.NANOSECONDS);
       long delayNanos = Math.max(0, retryBackoffPolicy.nextBackoffNanos() - elapsed);
-      rpcRetryTimer = syncContext.schedule(
-          new RpcRetryTask(), delayNanos, TimeUnit.NANOSECONDS, timeService);
+
+      rpcRetryTimer =
+          syncContext.schedule(new RpcRetryTask(), delayNanos, TimeUnit.NANOSECONDS, timeService);
 
       Status newStatus = status;
       if (responseReceived) {
@@ -410,6 +476,7 @@ private void handleRpcStreamClosed(Status status) {
       } else {
         // If the ADS stream is closed without ever having received a response from the server, then
         // the XdsClient should consider that a connectivity error (see gRFC A57).
+        inError = true;
         if (status.isOk()) {
           newStatus = Status.UNAVAILABLE.withDescription(
               "ADS stream closed with OK before receiving a response");
@@ -420,10 +487,8 @@ private void handleRpcStreamClosed(Status status) {
       }
 
       closed = true;
-      xdsResponseHandler.handleStreamClosed(newStatus);
+      xdsResponseHandler.handleStreamClosed(newStatus, !responseReceived);
       cleanUp();
-
-      logger.log(XdsLogLevel.INFO, "Retry ADS stream in {0} ns", delayNanos);
     }
 
     private void close(Exception error) {
@@ -441,4 +506,5 @@ private void cleanUp() {
       }
     }
   }
+
 }

xds/src/main/java/io/grpc/xds/client/XdsClient.java

@@ -298,14 +298,6 @@ public Object getSecurityConfig() {
     throw new UnsupportedOperationException();
   }
 
-  /**
-   * For all subscriber's for the specified server, if the resource hasn't yet been
-   * resolved then start a timer for it.
-   */
-  protected void startSubscriberTimersIfNeeded(ServerInfo serverInfo) {
-    throw new UnsupportedOperationException();
-  }
-
   /**
    * Returns a {@link ListenableFuture} to the snapshot of the subscribed resources as
    * they are at the moment of the call.
@@ -407,25 +399,34 @@ void handleResourceResponse(
 
     /** Called when the ADS stream is closed passively. */
     // Must be synchronized.
-    void handleStreamClosed(Status error);
+    void handleStreamClosed(Status error, boolean shouldTryFallback);
 
-    /** Called when the ADS stream has been recreated. */
-    // Must be synchronized.
+    /** Called when the ADS stream has established communication with the xds server.
+     * Is expected to manage the ControlPlanClients and cache updates associated with
+     * Moving to or from a fallback server.
+     *
+     * <p>Must be synchronized.
+     */
     void handleStreamRestarted(ServerInfo serverInfo);
   }
 
   public interface ResourceStore {
+
     /**
-     * Returns the collection of resources currently subscribing to or {@code null} if not
-     * subscribing to any resources for the given type.
+     * Returns the collection of resources currently subscribed to which have an authority matching
+     * one of those for which the ControlPlaneClient associated with the specified ServerInfo is
+     * the active one, or {@code null} if no such resources are currently subscribed to.
      *
      * <p>Note an empty collection indicates subscribing to resources of the given type with
      * wildcard mode.
+     *
+     * @param serverInfo the xds server to get the resources from
+     * @param type       the type of the resources that should be retrieved
      */
     // Must be synchronized.
     @Nullable
-    Collection<String> getSubscribedResources(ServerInfo serverInfo,
-                                              XdsResourceType<? extends ResourceUpdate> type);
+    Collection<String> getSubscribedResources(
+        ServerInfo serverInfo, XdsResourceType<? extends ResourceUpdate> type);
 
     Map<String, XdsResourceType<?>> getSubscribedResourceTypesWithTypeUrl();
   }