xds:Update logic to match A57 (#9745)

* xds:Change timer creation logic to wait until the adsStream is ready before creating the timer to mark resources absent.

* xds:When the ads stream is closed only send errors to subscribers that haven't yet gotten results to match spec.

* Use a blocking queue to avoid the 2-second sleep.
For some inexplicable reason the following call.verifyRequest fails only for the V2 test and only from command line not IDE unless there is some Thread.sleep, even if it is only 1-millis.

Author: larry-safran

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

@@ -36,6 +36,8 @@
 import io.grpc.SynchronizationContext;
 import io.grpc.SynchronizationContext.ScheduledHandle;
 import io.grpc.internal.BackoffPolicy;
+import io.grpc.stub.ClientCallStreamObserver;
+import io.grpc.stub.ClientResponseObserver;
 import io.grpc.stub.StreamObserver;
 import io.grpc.xds.Bootstrapper.ServerInfo;
 import io.grpc.xds.EnvoyProtoData.Node;
@@ -71,6 +73,7 @@ final class AbstractXdsClient {
   private final BackoffPolicy.Provider backoffPolicyProvider;
   private final Stopwatch stopwatch;
   private final Node bootstrapNode;
+  private final XdsClient.TimerLaunch timerLaunch;
 
   // 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
@@ -98,7 +101,8 @@ final class AbstractXdsClient {
       timeService,
       SynchronizationContext syncContext,
       BackoffPolicy.Provider backoffPolicyProvider,
-      Supplier<Stopwatch> stopwatchSupplier) {
+      Supplier<Stopwatch> stopwatchSupplier,
+      XdsClient.TimerLaunch timerLaunch) {
     this.serverInfo = checkNotNull(serverInfo, "serverInfo");
     this.channel = checkNotNull(xdsChannelFactory, "xdsChannelFactory").create(serverInfo);
     this.xdsResponseHandler = checkNotNull(xdsResponseHandler, "xdsResponseHandler");
@@ -108,6 +112,7 @@ final class AbstractXdsClient {
     this.timeService = checkNotNull(timeService, "timeService");
     this.syncContext = checkNotNull(syncContext, "syncContext");
     this.backoffPolicyProvider = checkNotNull(backoffPolicyProvider, "backoffPolicyProvider");
+    this.timerLaunch  = checkNotNull(timerLaunch, "timerLaunch");
     stopwatch = checkNotNull(stopwatchSupplier, "stopwatchSupplier").get();
     logId = InternalLogId.allocate("xds-client", serverInfo.target());
     logger = XdsLogger.withLogId(logId);
@@ -199,6 +204,22 @@ boolean isInBackoff() {
     return rpcRetryTimer != null && rpcRetryTimer.isPending();
   }
 
+  boolean isReady() {
+    return adsStream != null && adsStream.isReady();
+  }
+
+  /**
+   * Starts a timer for each requested resource that hasn't been responded to and
+   * has been waiting for the channel to get ready.
+   */
+  void readyHandler() {
+    if (!isReady()) {
+      return;
+    }
+
+    timerLaunch.startSubscriberTimersIfNeeded(serverInfo);
+  }
+
   /**
    * Establishes the RPC connection by creating a new RPC stream on the given channel for
    * xDS protocol communication.
@@ -262,6 +283,8 @@ private abstract class AbstractAdsStream {
 
     abstract void sendError(Exception error);
 
+    abstract boolean isReady();
+
     /**
      * Sends a discovery request with the given {@code versionInfo}, {@code nonce} and
      * {@code errorDetail}. Used for reacting to a specific discovery response. For
@@ -344,13 +367,26 @@ private void cleanUp() {
   private final class AdsStreamV2 extends AbstractAdsStream {
     private StreamObserver<io.envoyproxy.envoy.api.v2.DiscoveryRequest> requestWriter;
 
+    @Override
+    public boolean isReady() {
+      return requestWriter != null && ((ClientCallStreamObserver<?>) requestWriter).isReady();
+    }
+
     @Override
     void start() {
       io.envoyproxy.envoy.service.discovery.v2.AggregatedDiscoveryServiceGrpc
           .AggregatedDiscoveryServiceStub stub =
           io.envoyproxy.envoy.service.discovery.v2.AggregatedDiscoveryServiceGrpc.newStub(channel);
       StreamObserver<io.envoyproxy.envoy.api.v2.DiscoveryResponse> responseReaderV2 =
-          new StreamObserver<io.envoyproxy.envoy.api.v2.DiscoveryResponse>() {
+          new ClientResponseObserver<io.envoyproxy.envoy.api.v2.DiscoveryRequest,
+              io.envoyproxy.envoy.api.v2.DiscoveryResponse>() {
+
+            @Override
+            public void beforeStart(
+                ClientCallStreamObserver<io.envoyproxy.envoy.api.v2.DiscoveryRequest> reqStream) {
+              reqStream.setOnReadyHandler(AbstractXdsClient.this::readyHandler);
+            }
+
             @Override
             public void onNext(final io.envoyproxy.envoy.api.v2.DiscoveryResponse response) {
               syncContext.execute(new Runnable() {
@@ -427,11 +463,23 @@ void sendError(Exception error) {
   private final class AdsStreamV3 extends AbstractAdsStream {
     private StreamObserver<DiscoveryRequest> requestWriter;
 
+    @Override
+    public boolean isReady() {
+      return requestWriter != null && ((ClientCallStreamObserver<?>) requestWriter).isReady();
+    }
+
     @Override
     void start() {
       AggregatedDiscoveryServiceGrpc.AggregatedDiscoveryServiceStub stub =
           AggregatedDiscoveryServiceGrpc.newStub(channel);
-      StreamObserver<DiscoveryResponse> responseReader = new StreamObserver<DiscoveryResponse>() {
+      StreamObserver<DiscoveryResponse> responseReader =
+          new ClientResponseObserver<DiscoveryRequest,DiscoveryResponse>() {
+
+        @Override
+        public void beforeStart(ClientCallStreamObserver<DiscoveryRequest> requestStream) {
+          requestStream.setOnReadyHandler(AbstractXdsClient.this::readyHandler);
+        }
+
         @Override
         public void onNext(final DiscoveryResponse response) {
           syncContext.execute(new Runnable() {

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

@@ -373,4 +373,12 @@ Collection<String> getSubscribedResources(ServerInfo serverInfo,
 
     Map<String, XdsResourceType<?>> getSubscribedResourceTypesWithTypeUrl();
   }
+
+  interface TimerLaunch {
+    /**
+     * For all subscriber's for the specified server, if the resource hasn't yet been
+     * resolved then start a timer for it.
+     */
+    void startSubscriberTimersIfNeeded(ServerInfo serverInfo);
+  }
 }

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

@@ -47,6 +47,7 @@
 import io.grpc.xds.LoadStatsManager2.ClusterDropStats;
 import io.grpc.xds.LoadStatsManager2.ClusterLocalityStats;
 import io.grpc.xds.XdsClient.ResourceStore;
+import io.grpc.xds.XdsClient.TimerLaunch;
 import io.grpc.xds.XdsClient.XdsResponseHandler;
 import io.grpc.xds.XdsLogger.XdsLogLevel;
 import java.net.URI;
@@ -65,9 +66,10 @@
 import javax.annotation.Nullable;
 
 /**
- * XdsClient implementation for client side usages.
+ * XdsClient implementation.
  */
-final class XdsClientImpl extends XdsClient implements XdsResponseHandler, ResourceStore {
+final class XdsClientImpl extends XdsClient
+    implements XdsResponseHandler, ResourceStore, TimerLaunch {
 
   private static boolean LOG_XDS_NODE_ID = Boolean.parseBoolean(
       System.getenv("GRPC_LOG_XDS_NODE_ID"));
@@ -152,7 +154,8 @@ private void maybeCreateXdsChannelWithLrs(ServerInfo serverInfo) {
         timeService,
         syncContext,
         backoffPolicyProvider,
-        stopwatchSupplier);
+        stopwatchSupplier,
+        this);
     LoadReportClient lrsClient = new LoadReportClient(
         loadStatsManager, xdsChannel.channel(), context, serverInfo.useProtocolV3(),
         bootstrapInfo.node(), syncContext, timeService, backoffPolicyProvider, stopwatchSupplier);
@@ -188,7 +191,9 @@ public void handleStreamClosed(Status error) {
     for (Map<String, ResourceSubscriber<? extends ResourceUpdate>> subscriberMap :
         resourceSubscribers.values()) {
       for (ResourceSubscriber<? extends ResourceUpdate> subscriber : subscriberMap.values()) {
-        subscriber.onError(error);
+        if (!subscriber.hasResult()) {
+          subscriber.onError(error);
+        }
       }
     }
   }
@@ -386,6 +391,30 @@ public String toString() {
     return logId.toString();
   }
 
+  @Override
+  public void startSubscriberTimersIfNeeded(ServerInfo serverInfo) {
+    if (isShutDown()) {
+      return;
+    }
+
+    syncContext.execute(new Runnable() {
+      @Override
+      public void run() {
+        if (isShutDown()) {
+          return;
+        }
+
+        for (Map<String, ResourceSubscriber<?>> subscriberMap : resourceSubscribers.values()) {
+          for (ResourceSubscriber<?> subscriber : subscriberMap.values()) {
+            if (subscriber.serverInfo.equals(serverInfo) && subscriber.respTimer == null) {
+              subscriber.restartTimer();
+            }
+          }
+        }
+      }
+    });
+  }
+
   private void cleanUpResourceTimers() {
     for (Map<String, ResourceSubscriber<?>> subscriberMap : resourceSubscribers.values()) {
       for (ResourceSubscriber<?> subscriber : subscriberMap.values()) {
@@ -537,6 +566,10 @@ void restartTimer() {
       if (data != null || absent) {  // resource already resolved
         return;
       }
+      if (!xdsChannel.isReady()) { // When channel becomes ready, it will trigger a restartTimer
+        return;
+      }
+
       class ResourceNotFound implements Runnable {
         @Override
         public void run() {
@@ -554,6 +587,7 @@ public String toString() {
 
       // Initial fetch scheduled or rescheduled, transition metadata state to REQUESTED.
       metadata = ResourceMetadata.newResourceMetadataRequested();
+
       respTimer = syncContext.schedule(
           new ResourceNotFound(), INITIAL_RESOURCE_FETCH_TIMEOUT_SEC, TimeUnit.SECONDS,
           timeService);
@@ -585,6 +619,10 @@ boolean isWatched() {
       return !watchers.isEmpty();
     }
 
+    boolean hasResult() {
+      return data != null || absent;
+    }
+
     void onData(ParsedResource<T> parsedResource, String version, long updateTime) {
       if (respTimer != null && respTimer.isPending()) {
         respTimer.cancel();