Fixed bug: The DiscoveryRequest with resource names may never be updated after xDS server is restarted with empty SnapshotCache. (#103)

Signed-off-by: Pin Zhang <zhangpin04@gmail.com>

Author: PinZhang

cache/src/main/java/io/envoyproxy/controlplane/cache/SimpleCache.java

@@ -145,7 +145,22 @@ public Watch createWatch(
       }
 
       // Otherwise, the watch may be responded immediately
-      respond(watch, snapshot, group);
+      boolean responded = respond(watch, snapshot, group);
+
+      if (!responded) {
+        watchCount++;
+
+        LOGGER.info("did not respond immediately, leaving open watch {} for {}[{}] from node {} for version {}",
+            watchCount,
+            request.getTypeUrl(),
+            String.join(", ", request.getResourceNamesList()),
+            group,
+            request.getVersionInfo());
+
+        status.setWatch(watchCount, watch);
+
+        watch.setStop(() -> status.removeWatch(watchCount));
+      }
 
       return watch;
 
@@ -245,7 +260,7 @@ private Response createResponse(DiscoveryRequest request, Map<String, ? extends
     return Response.create(request, filtered, version);
   }
 
-  private void respond(Watch watch, Snapshot snapshot, T group) {
+  private boolean respond(Watch watch, Snapshot snapshot, T group) {
     Map<String, ? extends Message> snapshotResources = snapshot.resources(watch.request().getTypeUrl());
 
     if (!watch.request().getResourceNamesList().isEmpty() && watch.ads()) {
@@ -262,7 +277,7 @@ private void respond(Watch watch, Snapshot snapshot, T group) {
             String.join(", ", watch.request().getResourceNamesList()),
             String.join(", ", missingNames));
 
-        return;
+        return false;
       }
     }
 
@@ -281,6 +296,7 @@ private void respond(Watch watch, Snapshot snapshot, T group) {
 
     try {
       watch.respond(response);
+      return true;
     } catch (WatchCancelledException e) {
       LOGGER.error(
           "failed to respond for {} from node {} at version {} with version {} because watch was already cancelled",
@@ -289,5 +305,7 @@ private void respond(Watch watch, Snapshot snapshot, T group) {
           watch.request().getVersionInfo(),
           version);
     }
+
+    return false;
   }
 }

cache/src/test/java/io/envoyproxy/controlplane/cache/SimpleCacheTest.java

@@ -1,5 +1,6 @@
 package io.envoyproxy.controlplane.cache;
 
+import static io.envoyproxy.controlplane.cache.Resources.ROUTE_TYPE_URL;
 import static org.assertj.core.api.Assertions.assertThat;
 
 import com.google.common.collect.ImmutableList;
@@ -350,6 +351,26 @@ public void watchesAreReleasedAfterCancel() {
     watches.values().forEach(w -> assertThat(w.watch.isCancelled()).isTrue());
   }
 
+  @Test
+  public void watchIsLeftOpenIfNotRespondedImmediately() {
+    SimpleCache<String> cache = new SimpleCache<>(new SingleNodeGroup());
+    cache.setSnapshot(SingleNodeGroup.GROUP, Snapshot.create(
+        ImmutableList.of(), ImmutableList.of(), ImmutableList.of(), ImmutableList.of(), ImmutableList.of(), VERSION1));
+
+    ResponseTracker responseTracker = new ResponseTracker();
+    Watch watch = cache.createWatch(
+        true,
+        DiscoveryRequest.newBuilder()
+            .setNode(Node.getDefaultInstance())
+            .setTypeUrl(ROUTE_TYPE_URL)
+            .addAllResourceNames(Collections.singleton(ROUTE_NAME))
+            .build(),
+        Collections.singleton(ROUTE_NAME),
+        responseTracker);
+
+    assertThatWatchIsOpenWithNoResponses(new WatchAndTracker(watch, responseTracker));
+  }
+
   @Test
   public void getSnapshot() {
     SimpleCache<String> cache = new SimpleCache<>(new SingleNodeGroup());