Send EDS response after CDS even if EDS hasn't change (#131)

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

Author: lukidzi

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

@@ -16,14 +16,17 @@ public interface ConfigWatcher {
   /**
    * Returns a new configuration resource {@link Watch} for the given discovery request.
    *
-   * @param ads is the watch for an ADS request?
-   * @param request the discovery request (node, names, etc.) to use to generate the watch
+   * @param ads                is the watch for an ADS request?
+   * @param request            the discovery request (node, names, etc.) to use to generate the watch
    * @param knownResourceNames resources that are already known to the caller
-   * @param responseConsumer the response handler, used to process outgoing response messages
+   * @param responseConsumer   the response handler, used to process outgoing response messages
+   * @param hasClusterChanged  Indicates if EDS should be sent immediately, even if version has not been changed.
+   *                           Supported in ADS mode.
    */
   Watch createWatch(
       boolean ads,
       DiscoveryRequest request,
       Set<String> knownResourceNames,
-      Consumer<Response> responseConsumer);
+      Consumer<Response> responseConsumer,
+      boolean hasClusterChanged);
 }

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

@@ -1,5 +1,6 @@
 package io.envoyproxy.controlplane.cache;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Sets;
 import com.google.protobuf.Message;
@@ -81,6 +82,14 @@ public boolean clearSnapshot(T group) {
     }
   }
 
+  public Watch createWatch(
+      boolean ads,
+      DiscoveryRequest request,
+      Set<String> knownResourceNames,
+      Consumer<Response> responseConsumer) {
+    return createWatch(ads, request, knownResourceNames, responseConsumer, false);
+  }
+
   /**
    * {@inheritDoc}
    */
@@ -89,7 +98,8 @@ public Watch createWatch(
       boolean ads,
       DiscoveryRequest request,
       Set<String> knownResourceNames,
-      Consumer<Response> responseConsumer) {
+      Consumer<Response> responseConsumer,
+      boolean hasClusterChanged) {
 
     T group = groups.hash(request.getNode());
     // even though we're modifying, we take a readLock to allow multiple watches to be created in parallel since it
@@ -121,6 +131,10 @@ public Watch createWatch(
 
             return watch;
           }
+        } else if (hasClusterChanged && request.getTypeUrl().equals(Resources.ENDPOINT_TYPE_URL)) {
+          respond(watch, snapshot, group);
+
+          return watch;
         }
       }
 
@@ -213,6 +227,25 @@ public synchronized void setSnapshot(T group, Snapshot snapshot) {
     }
 
     // Responses should be in specific order and TYPE_URLS has a list of resources in the right order.
+    respondWithSpecificOrder(group, snapshot, status);
+  }
+
+  /**
+   * {@inheritDoc}
+   */
+  @Override
+  public StatusInfo statusInfo(T group) {
+    readLock.lock();
+
+    try {
+      return statuses.get(group);
+    } finally {
+      readLock.unlock();
+    }
+  }
+
+  @VisibleForTesting
+  protected void respondWithSpecificOrder(T group, Snapshot snapshot, CacheStatusInfo<T> status) {
     for (String typeUrl : Resources.TYPE_URLS) {
       status.watchesRemoveIf((id, watch) -> {
         if (!watch.request().getTypeUrl().equals(typeUrl)) {
@@ -240,20 +273,6 @@ public synchronized void setSnapshot(T group, Snapshot snapshot) {
     }
   }
 
-  /**
-   * {@inheritDoc}
-   */
-  @Override
-  public StatusInfo statusInfo(T group) {
-    readLock.lock();
-
-    try {
-      return statuses.get(group);
-    } finally {
-      readLock.unlock();
-    }
-  }
-
   private Response createResponse(DiscoveryRequest request, Map<String, ? extends Message> resources, String version) {
     Collection<? extends Message> filtered = request.getResourceNamesList().isEmpty()
         ? resources.values()

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

@@ -96,14 +96,25 @@ public static Cluster createCluster(String clusterName, String address, int port
    * @param port port to use for the endpoint
    */
   public static ClusterLoadAssignment createEndpoint(String clusterName, int port) {
+    return createEndpoint(clusterName, LOCALHOST, port);
+  }
+
+  /**
+   * Returns a new test endpoint for the given cluster.
+   *
+   * @param clusterName name of the test cluster that is associated with this endpoint
+   * @param address ip address to use for the endpoint
+   * @param port port to use for the endpoint
+   */
+  public static ClusterLoadAssignment createEndpoint(String clusterName, String address, int port) {
     return ClusterLoadAssignment.newBuilder()
         .setClusterName(clusterName)
         .addEndpoints(LocalityLbEndpoints.newBuilder()
             .addLbEndpoints(LbEndpoint.newBuilder()
                 .setEndpoint(Endpoint.newBuilder()
                     .setAddress(Address.newBuilder()
                         .setSocketAddress(SocketAddress.newBuilder()
-                            .setAddress(LOCALHOST)
+                            .setAddress(address)
                             .setPortValue(port)
                             .setProtocol(Protocol.TCP))))))
         .build();

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

@@ -4,6 +4,7 @@
 import static org.assertj.core.api.Assertions.assertThat;
 
 import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Sets;
 import com.google.protobuf.Message;
 import io.envoyproxy.envoy.api.v2.Cluster;
 import io.envoyproxy.envoy.api.v2.ClusterLoadAssignment;
@@ -130,6 +131,33 @@ public void successfullyWatchAllResourceTypesWithSetBeforeWatch() {
     }
   }
 
+  @Test
+  public void shouldSendEdsWhenClusterChangedButEdsVersionDidnt() {
+    SimpleCache<String> cache = new SimpleCache<>(new SingleNodeGroup());
+
+    cache.setSnapshot(SingleNodeGroup.GROUP, SNAPSHOT1);
+
+    ResponseTracker responseTracker = new ResponseTracker();
+
+    Watch watch = cache.createWatch(
+        ADS,
+        DiscoveryRequest.newBuilder()
+            .setNode(Node.getDefaultInstance())
+            .setVersionInfo(VERSION1)
+            .setTypeUrl(Resources.ENDPOINT_TYPE_URL)
+            .addAllResourceNames(SNAPSHOT1.resources(Resources.ENDPOINT_TYPE_URL).keySet())
+            .build(),
+        Sets.newHashSet(""),
+        responseTracker,
+        true);
+
+    assertThat(watch.request().getTypeUrl()).isEqualTo(Resources.ENDPOINT_TYPE_URL);
+    assertThat(watch.request().getResourceNamesList()).containsExactlyElementsOf(
+        SNAPSHOT1.resources(Resources.ENDPOINT_TYPE_URL).keySet());
+
+    assertThatWatchReceivesSnapshot(new WatchAndTracker(watch, responseTracker), SNAPSHOT1);
+  }
+
   @Test
   public void successfullyWatchAllResourceTypesWithSetAfterWatch() {
     SimpleCache<String> cache = new SimpleCache<>(new SingleNodeGroup());

server/src/main/java/io/envoyproxy/controlplane/server/AdsDiscoveryRequestStreamObserver.java

@@ -66,6 +66,11 @@ LatestDiscoveryResponse latestResponse(String typeUrl) {
   @Override
   void setLatestResponse(String typeUrl, LatestDiscoveryResponse response) {
     latestResponse.put(typeUrl, response);
+    if (typeUrl.equals(Resources.CLUSTER_TYPE_URL)) {
+      hasClusterChanged = true;
+    } else if (typeUrl.equals(Resources.ENDPOINT_TYPE_URL)) {
+      hasClusterChanged = false;
+    }
   }
 
   @Override

server/src/main/java/io/envoyproxy/controlplane/server/DiscoveryRequestStreamObserver.java

@@ -30,6 +30,7 @@ public abstract class DiscoveryRequestStreamObserver implements StreamObserver<D
   private static final Logger LOGGER = LoggerFactory.getLogger(DiscoveryServer.class);
 
   final long streamId;
+  volatile boolean hasClusterChanged;
   private final String defaultTypeUrl;
   private final StreamObserver<DiscoveryResponse> responseObserver;
   private final Executor executor;
@@ -48,6 +49,7 @@ public abstract class DiscoveryRequestStreamObserver implements StreamObserver<D
     this.executor = executor;
     this.streamNonce = 0;
     this.discoverySever = discoveryServer;
+    this.hasClusterChanged = false;
   }
 
   @Override
@@ -83,7 +85,9 @@ public void onNext(DiscoveryRequest request) {
           ads(),
           request,
           ackedResources(requestTypeUrl),
-          r -> executor.execute(() -> send(r, requestTypeUrl))));
+          r -> executor.execute(() -> send(r, requestTypeUrl)),
+          hasClusterChanged
+      ));
     }
   }
 

server/src/test/java/io/envoyproxy/controlplane/server/DiscoveryServerAdsIT.java

@@ -52,7 +52,15 @@ public void onStreamResponse(long streamId, DiscoveryRequest request, DiscoveryR
 
       cache.setSnapshot(
           GROUP,
-          createSnapshot(true, "upstream", "upstream", EchoContainer.PORT, "listener0", LISTENER_PORT, "route0", "1"));
+          createSnapshot(true,
+              "upstream",
+              UPSTREAM.ipAddress(),
+              EchoContainer.PORT,
+              "listener0",
+              LISTENER_PORT,
+              "route0",
+              "1")
+      );
 
       DiscoveryServer server = new DiscoveryServer(callbacks, cache);