Send responses in specific order (#128)

Signed-off-by: Lukasz Dziedziak <lukasz.dziedziak@allegro.pl>

Author: lukidzi

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

@@ -212,26 +212,32 @@ public synchronized void setSnapshot(T group, Snapshot snapshot) {
       return;
     }
 
-    status.watchesRemoveIf((id, watch) -> {
-      String version = snapshot.version(watch.request().getTypeUrl(), watch.request().getResourceNamesList());
-
-      if (!watch.request().getVersionInfo().equals(version)) {
-        if (LOGGER.isDebugEnabled()) {
-          LOGGER.debug("responding to open watch {}[{}] with new version {}",
-              id,
-              String.join(", ", watch.request().getResourceNamesList()),
-              version);
+    // Responses should be in specific order and TYPE_URLS has a list of resources in the right order.
+    for (String typeUrl : Resources.TYPE_URLS) {
+      status.watchesRemoveIf((id, watch) -> {
+        if (!watch.request().getTypeUrl().equals(typeUrl)) {
+          return false;
         }
+        String version = snapshot.version(watch.request().getTypeUrl(), watch.request().getResourceNamesList());
+
+        if (!watch.request().getVersionInfo().equals(version)) {
+          if (LOGGER.isDebugEnabled()) {
+            LOGGER.debug("responding to open watch {}[{}] with new version {}",
+                id,
+                String.join(", ", watch.request().getResourceNamesList()),
+                version);
+          }
 
-        respond(watch, snapshot, group);
+          respond(watch, snapshot, group);
 
-        // Discard the watch. A new watch will be created for future snapshots once envoy ACKs the response.
-        return true;
-      }
+          // Discard the watch. A new watch will be created for future snapshots once envoy ACKs the response.
+          return true;
+        }
 
-      // Do not discard the watch. The request version is the same as the snapshot version, so we wait to respond.
-      return false;
-    });
+        // Do not discard the watch. The request version is the same as the snapshot version, so we wait to respond.
+        return false;
+      });
+    }
   }
 
   /**

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

@@ -13,6 +13,7 @@
 import io.envoyproxy.envoy.api.v2.auth.Secret;
 import io.envoyproxy.envoy.api.v2.core.Node;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.LinkedList;
 import java.util.Map;
 import java.util.UUID;
@@ -165,25 +166,35 @@ public void successfullyWatchAllResourceTypesWithSetBeforeWatchWithRequestVersio
 
     cache.setSnapshot(SingleNodeGroup.GROUP, SNAPSHOT1);
 
-    Map<String, WatchAndTracker> watches = Resources.TYPE_URLS.stream()
-        .collect(Collectors.toMap(
-            typeUrl -> typeUrl,
-            typeUrl -> {
-              ResponseTracker responseTracker = new ResponseTracker();
-
-              Watch watch = cache.createWatch(
-                  ADS,
-                  DiscoveryRequest.newBuilder()
-                      .setNode(Node.getDefaultInstance())
-                      .setTypeUrl(typeUrl)
-                      .setVersionInfo(SNAPSHOT1.version(typeUrl))
-                      .addAllResourceNames(SNAPSHOT1.resources(typeUrl).keySet())
-                      .build(),
-                  SNAPSHOT2.resources(typeUrl).keySet(),
-                  responseTracker);
-
-              return new WatchAndTracker(watch, responseTracker);
-            }));
+    ResponseOrderTracker responseOrderTracker = new ResponseOrderTracker();
+
+    HashMap<String, WatchAndTracker> watches = new HashMap<>();
+
+    for (int i = 0; i < 2; ++i) {
+      watches.putAll(Resources.TYPE_URLS.stream()
+          .collect(Collectors.toMap(
+              typeUrl -> typeUrl,
+              typeUrl -> {
+                ResponseTracker responseTracker = new ResponseTracker();
+
+                Watch watch = cache.createWatch(
+                    ADS,
+                    DiscoveryRequest.newBuilder()
+                        .setNode(Node.getDefaultInstance())
+                        .setTypeUrl(typeUrl)
+                        .setVersionInfo(SNAPSHOT1.version(typeUrl))
+                        .addAllResourceNames(SNAPSHOT1.resources(typeUrl).keySet())
+                        .build(),
+                    SNAPSHOT2.resources(typeUrl).keySet(),
+                    r -> {
+                      responseTracker.accept(r);
+                      responseOrderTracker.accept(r);
+                    });
+
+                return new WatchAndTracker(watch, responseTracker);
+              }))
+      );
+    }
 
     // The request version matches the current snapshot version, so the watches shouldn't receive any responses.
     for (String typeUrl : Resources.TYPE_URLS) {
@@ -195,6 +206,12 @@ public void successfullyWatchAllResourceTypesWithSetBeforeWatchWithRequestVersio
     for (String typeUrl : Resources.TYPE_URLS) {
       assertThatWatchReceivesSnapshot(watches.get(typeUrl), SNAPSHOT2);
     }
+
+    // Verify that CDS and LDS always get triggered before EDS and RDS respectively.
+    assertThat(responseOrderTracker.responseTypes).containsExactly(Resources.CLUSTER_TYPE_URL,
+        Resources.CLUSTER_TYPE_URL, Resources.ENDPOINT_TYPE_URL, Resources.ENDPOINT_TYPE_URL,
+        Resources.LISTENER_TYPE_URL, Resources.LISTENER_TYPE_URL, Resources.ROUTE_TYPE_URL,
+        Resources.ROUTE_TYPE_URL, Resources.SECRET_TYPE_URL, Resources.SECRET_TYPE_URL);
   }
 
   @Test
@@ -457,6 +474,16 @@ private static class ResponseTracker implements Consumer<Response> {
     public void accept(Response response) {
       responses.add(response);
     }
+
+  }
+
+  private static class ResponseOrderTracker implements Consumer<Response> {
+
+    private final LinkedList<String> responseTypes = new LinkedList<>();
+
+    @Override public void accept(Response response) {
+      responseTypes.add(response.request().getTypeUrl());
+    }
   }
 
   private static class SingleNodeGroup implements NodeGroup<String> {