Send EDS response in ADS mode when Envoy reconnect and asks with Empty version

Signed-off-by: Lukasz Dziedziak <[email protected]>

Author: Lukasz Dziedziak

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

@@ -3,8 +3,10 @@
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Sets;
 import com.google.protobuf.Message;
+import io.envoyproxy.envoy.api.v2.ClusterLoadAssignment;
 import io.envoyproxy.envoy.api.v2.DiscoveryRequest;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Objects;
@@ -18,6 +20,7 @@
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
 import javax.annotation.concurrent.GuardedBy;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -261,21 +264,42 @@ private Response createResponse(DiscoveryRequest request, Map<String, ? extends
 
   private boolean respond(Watch watch, Snapshot snapshot, T group) {
     Map<String, ? extends Message> snapshotResources = snapshot.resources(watch.request().getTypeUrl());
+    Map<String, ClusterLoadAssignment> snapshotWithMissingResources = Collections.emptyMap();
 
     if (!watch.request().getResourceNamesList().isEmpty() && watch.ads()) {
       Collection<String> missingNames = watch.request().getResourceNamesList().stream()
           .filter(name -> !snapshotResources.containsKey(name))
           .collect(Collectors.toList());
 
-      if (!missingNames.isEmpty()) {
+      // When Envoy receive CDS response and reconnect to new instance of control-plane before EDS was sent, Envoy might
+      // stack in warming phase. New instance of control-plane might not have cluster in snapshot and won't be able to
+      // respond. First request from Envoy contains empty string version info.
+      if (!missingNames.isEmpty()
+          && watch.request().getTypeUrl().equals(Resources.ENDPOINT_TYPE_URL)
+          && watch.request().getVersionInfo().equals("")) {
+        LOGGER.info("adding missing resources [{}] to response for {} in ADS mode from node {} at version {}",
+            String.join(", ", missingNames),
+            watch.request().getTypeUrl(),
+            group,
+            snapshot.version(watch.request().getTypeUrl(), watch.request().getResourceNamesList())
+        );
+        snapshotWithMissingResources = new HashMap<>(missingNames.size() + snapshotResources.size());
+        for (String missingName : missingNames) {
+          snapshotWithMissingResources.put(
+              missingName,
+              ClusterLoadAssignment.newBuilder().setClusterName(missingName).build()
+          );
+          snapshotWithMissingResources.putAll(
+              (Map<? extends String, ? extends ClusterLoadAssignment>) snapshotResources);
+        }
+      } else if (!missingNames.isEmpty()) {
         LOGGER.info(
             "not responding in ADS mode for {} from node {} at version {} for request [{}] since [{}] not in snapshot",
             watch.request().getTypeUrl(),
             group,
             snapshot.version(watch.request().getTypeUrl(), watch.request().getResourceNamesList()),
             String.join(", ", watch.request().getResourceNamesList()),
             String.join(", ", missingNames));
-
         return false;
       }
     }
@@ -287,11 +311,18 @@ private boolean respond(Watch watch, Snapshot snapshot, T group) {
         group,
         watch.request().getVersionInfo(),
         version);
-
-    Response response = createResponse(
-        watch.request(),
-        snapshotResources,
-        version);
+    Response response;
+    if (!snapshotWithMissingResources.isEmpty()) {
+      response = createResponse(
+          watch.request(),
+          snapshotWithMissingResources,
+          version);
+    } else {
+      response = createResponse(
+          watch.request(),
+          snapshotResources,
+          version);
+    }
 
     try {
       watch.respond(response);

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

@@ -14,6 +14,7 @@
 import io.envoyproxy.envoy.api.v2.core.Node;
 import java.util.Collections;
 import java.util.LinkedList;
+import java.util.List;
 import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.ThreadLocalRandom;
@@ -73,13 +74,50 @@ public void invalidNamesListShouldReturnWatcherWithNoResponseInAdsMode() {
             .setNode(Node.getDefaultInstance())
             .setTypeUrl(Resources.ENDPOINT_TYPE_URL)
             .addResourceNames("none")
+            .setVersionInfo("123")
             .build(),
         Collections.emptySet(),
         responseTracker);
 
     assertThatWatchIsOpenWithNoResponses(new WatchAndTracker(watch, responseTracker));
   }
 
+  @Test
+  public void invalidNamesListShouldReturnWatcherWithResponseInAdsModeWhenVersionIsEmpty() {
+    SimpleCache<String> cache = new SimpleCache<>(new SingleNodeGroup());
+
+    cache.setSnapshot(SingleNodeGroup.GROUP, MULTIPLE_RESOURCES_SNAPSHOT2);
+
+    ResponseTracker responseTracker = new ResponseTracker();
+
+    Watch watch = cache.createWatch(
+        true,
+        DiscoveryRequest.newBuilder()
+            .setNode(Node.getDefaultInstance())
+            .setTypeUrl(Resources.ENDPOINT_TYPE_URL)
+            .addAllResourceNames(MULTIPLE_RESOURCES_SNAPSHOT2.resources(Resources.ENDPOINT_TYPE_URL).keySet())
+            .addResourceNames("none")
+            .setVersionInfo("")
+            .build(),
+        Collections.emptySet(),
+        responseTracker);
+
+    assertThat(responseTracker.responses).isNotEmpty();
+
+    Response response = responseTracker.responses.getFirst();
+
+    assertThat(response).isNotNull();
+    assertThat(response.version()).isEqualTo(MULTIPLE_RESOURCES_SNAPSHOT2.version(watch.request().getTypeUrl()));
+
+    List<ClusterLoadAssignment> expectedResponse = new LinkedList<>((List<ClusterLoadAssignment>)
+        MULTIPLE_RESOURCES_SNAPSHOT2.resources(watch.request().getTypeUrl()).values());
+    // Because versionInfo in request is empty we should send all requested resources to don't leave Envoy
+    // in warming state.
+    expectedResponse.add(ClusterLoadAssignment.newBuilder().setClusterName("none").build());
+
+    assertThat(response.resources().toArray(new Message[0])).containsExactlyElementsOf(expectedResponse);
+  }
+
   @Test
   public void invalidNamesListShouldReturnWatcherWithResponseInXdsMode() {
     SimpleCache<String> cache = new SimpleCache<>(new SingleNodeGroup());
@@ -402,7 +440,8 @@ public void clearSnapshotWithWatches() {
             .setTypeUrl("")
             .build(),
         Collections.emptySet(),
-        r -> { });
+        r -> {
+        });
 
     // clearSnapshot should fail and the snapshot should be left untouched
     assertThat(cache.clearSnapshot(SingleNodeGroup.GROUP)).isFalse();
@@ -428,7 +467,8 @@ public void groups() {
             .setTypeUrl("")
             .build(),
         Collections.emptySet(),
-        r -> { });
+        r -> {
+        });
 
     assertThat(cache.groups()).containsExactly(SingleNodeGroup.GROUP);
   }