Add flag to guard xds fallback

Author: larry-safran

census/src/test/java/io/grpc/census/CensusModulesTest.java

@@ -56,6 +56,7 @@
 import io.grpc.ClientInterceptors;
 import io.grpc.ClientStreamTracer;
 import io.grpc.Context;
+import io.grpc.KnownLength;
 import io.grpc.Metadata;
 import io.grpc.MethodDescriptor;
 import io.grpc.ServerCall;
@@ -99,6 +100,7 @@
 import io.opencensus.trace.Tracer;
 import io.opencensus.trace.propagation.BinaryFormat;
 import io.opencensus.trace.propagation.SpanContextParseException;
+import java.io.IOException;
 import java.io.InputStream;
 import java.util.HashSet;
 import java.util.List;
@@ -136,7 +138,7 @@ public class CensusModulesTest {
       ClientStreamTracer.StreamInfo.newBuilder()
           .setCallOptions(CallOptions.DEFAULT.withOption(NAME_RESOLUTION_DELAYED, 10L)).build();
 
-  private static class StringInputStream extends InputStream {
+  private static class StringInputStream extends InputStream implements KnownLength {
     final String string;
 
     StringInputStream(String string) {
@@ -149,6 +151,11 @@ public int read() {
       // passed to the InProcess server and consumed by MARSHALLER.parse().
       throw new UnsupportedOperationException("Should not be called");
     }
+
+    @Override
+    public int available() throws IOException {
+      return string == null ? 0 : string.length();
+    }
   }
 
   private static final MethodDescriptor.Marshaller<String> MARSHALLER =

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

@@ -41,6 +41,10 @@
 @Internal
 public abstract class BootstrapperImpl extends Bootstrapper {
 
+  public static final String GRPC_EXPERIMENTAL_XDS_FALLBACK =
+      "GRPC_EXPERIMENTAL_XDS_FALLBACK";
+  private static Boolean xdsFallbackEnabled = null;
+
   // Client features.
   @VisibleForTesting
   public static final String CLIENT_FEATURE_DISABLE_OVERPROVISIONING =
@@ -59,11 +63,21 @@ protected BootstrapperImpl() {
     logger = XdsLogger.withLogId(InternalLogId.allocate("bootstrapper", null));
   }
 
+  // Delayed initialization of xdsFallbackEnabled to allow for flag initialization.
+  public static boolean isEnabledXdsFallback() {
+    if (xdsFallbackEnabled == null) {
+      xdsFallbackEnabled = GrpcUtil.getFlag(GRPC_EXPERIMENTAL_XDS_FALLBACK, false);
+    }
+    return xdsFallbackEnabled;
+  }
+
   protected abstract String getJsonContent() throws IOException, XdsInitializationException;
 
   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.
    */
@@ -102,6 +116,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 && !isEnabledXdsFallback()) {
+      servers = ImmutableList.of(servers.get(0));
+    }
     builder.servers(servers);
 
     Node.Builder nodeBuilder = Node.newBuilder();
@@ -208,6 +225,9 @@ protected BootstrapInfo.Builder bootstrapBuilder(Map<String, ?> rawData)
         if (rawAuthorityServers == null || rawAuthorityServers.isEmpty()) {
           authorityServers = servers;
         } else {
+          if (rawAuthorityServers.size() > 1 && !isEnabledXdsFallback()) {
+            rawAuthorityServers = ImmutableList.of(rawAuthorityServers.get(0));
+          }
           authorityServers = parseServerInfos(rawAuthorityServers, logger);
         }
         authorityInfoMapBuilder.put(

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

@@ -402,11 +402,7 @@ public void run() {
     public void onStatusReceived(final Status status) {
       lastStateWasReady = false;
       syncContext.execute(() -> {
-        if (status.isOk()) {
-          handleRpcStreamClosed(Status.UNAVAILABLE.withDescription(CLOSED_BY_SERVER));
-        } else {
-          handleRpcStreamClosed(status);
-        }
+        handleRpcStreamClosed(status);
       });
     }
 
@@ -425,7 +421,7 @@ final void handleRpcResponse(XdsResourceType<?> type, String versionInfo, List<A
       processingTracker.onComplete();
     }
 
-    private void handleRpcStreamClosed(Status error) {
+    private void handleRpcStreamClosed(Status status) {
       if (closed) {
         return;
       }
@@ -440,15 +436,18 @@ private void handleRpcStreamClosed(Status error) {
       // concurrently with the stopwatch and schedule.
       long delayNanos = scheduleRpcRetry();
 
-      checkArgument(!error.isOk(), "unexpected OK status");
-      String errorMsg = error.getDescription() != null
-          && error.getDescription().equals(CLOSED_BY_SERVER)
-              ? "ADS stream closed with status {0}: {1}. Cause: {2}"
-              : "ADS stream failed with status {0}: {1}. Cause: {2}";
-      logger.log(
-          XdsLogLevel.ERROR, errorMsg, error.getCode(), error.getDescription(), error.getCause());
+      if (status.isOk()) {
+        logger.log(XdsLogLevel.WARNING, "ADS stream closed, backoff {0} ns", delayNanos);
+      } else {
+        String errorMsg = status.getDescription() != null
+            && status.getDescription().equals(CLOSED_BY_SERVER)
+                ? "ADS stream closed with status {0}: {1}. Cause: {2}"
+                : "ADS stream failed with status {0}: {1}. Cause: {2}";
+        logger.log(
+            XdsLogLevel.ERROR, errorMsg, status.getCode(), status.getDescription(), status.getCause());
+      }
       closed = true;
-      xdsResponseHandler.handleStreamClosed(error);
+      xdsResponseHandler.handleStreamClosed(status);
       cleanUp();
 
       logger.log(XdsLogLevel.INFO, "Retry ADS stream in {0} ns", delayNanos);

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

@@ -19,6 +19,7 @@
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkNotNull;
 import static io.grpc.xds.client.Bootstrapper.XDSTP_SCHEME;
+import static io.grpc.xds.client.ControlPlaneClient.CLOSED_BY_SERVER;
 import static io.grpc.xds.client.XdsResourceType.ParsedResource;
 import static io.grpc.xds.client.XdsResourceType.ValidatedResourceUpdate;
 
@@ -996,28 +997,30 @@ public void handleResourceResponse(
     }
 
     @Override
-    public void handleStreamClosed(Status error) {
+    public void handleStreamClosed(Status status) {
       syncContext.throwIfNotInThisSynchronizationContext();
       ControlPlaneClient cpcForThisStream = serverCpClientMap.get(serverInfo);
       cleanUpResourceTimers(cpcForThisStream);
 
-      boolean hadError = false;
+      Status error = status.isOk() ? Status.UNAVAILABLE.withDescription(CLOSED_BY_SERVER) : status;
+      boolean checkForFallback = !status.isOk() && BootstrapperImpl.isEnabledXdsFallback();
+
       for (Map<String, ResourceSubscriber<? extends ResourceUpdate>> subscriberMap :
           resourceSubscribers.values()) {
         for (ResourceSubscriber<? extends ResourceUpdate> subscriber : subscriberMap.values()) {
           if (subscriber.hasResult()
               || (subscriber.cpcFixed && !cpcForThisStream.equals(subscriber.controlPlaneClient))) {
             continue;
           }
-          if (!hadError) {
-            logger.log(XdsLogLevel.WARNING, "ADS stream closed with error: {0}", error);
+          if (checkForFallback) {
             // try to fallback to lower priority control plane client
             if (doFallbackForAuthority(cpcForThisStream,
                 serverInfo, subscriber.serverInfos, subscriber.authority)) {
               return;
             }
+            checkForFallback = false;
           }
-          hadError = true;
+
           subscriber.onError(error, null);
         }
       }

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

@@ -65,6 +65,7 @@ public class GrpcBootstrapperImplTest {
   @Before
   public void setUp() {
     saveEnvironment();
+    System.setProperty(BootstrapperImpl.GRPC_EXPERIMENTAL_XDS_FALLBACK, "true");
     bootstrapper.bootstrapPathFromEnvVar = BOOTSTRAP_FILE_PATH;
   }
 

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

@@ -30,6 +30,7 @@
 import com.google.common.collect.ImmutableMap;
 import io.grpc.Status;
 import io.grpc.internal.ObjectPool;
+import io.grpc.xds.client.BootstrapperImpl;
 import io.grpc.xds.client.XdsClient;
 import io.grpc.xds.client.XdsClientImpl;
 import io.grpc.xds.client.XdsInitializationException;
@@ -119,6 +120,7 @@ public void onResourceDoesNotExist(String resourceName) {
 
   @Before
   public void setUp() throws XdsInitializationException {
+    System.setProperty(BootstrapperImpl.GRPC_EXPERIMENTAL_XDS_FALLBACK, "true");
     setAdsConfig(mainTdServer, MAIN_SERVER);
     setAdsConfig(fallbackServer, FALLBACK_SERVER);