Fix flapping warming cluster integration tests (#145)

slonka · web-flow · commit d61954420d41 · 2020-09-24T10:17:28.000+02:00
* Try to fix flapping warming cluster integration tests

Signed-off-by: slonka &lt;slonka@users.noreply.github.com&gt;

* Remove delay, add assertion on 503

Signed-off-by: slonka &lt;slonka@users.noreply.github.com&gt;
diff --git a/server/src/test/java/io/envoyproxy/controlplane/server/V2DiscoveryServerAdsWarmingClusterIT.java b/server/src/test/java/io/envoyproxy/controlplane/server/V2DiscoveryServerAdsWarmingClusterIT.java
@@ -1,16 +1,13 @@
 package io.envoyproxy.controlplane.server;
 
-import static io.envoyproxy.controlplane.server.V2TestSnapshots.createSnapshot;
 import static io.envoyproxy.envoy.api.v2.core.ApiVersion.V2;
 import static io.restassured.RestAssured.given;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.awaitility.Awaitility.await;
 import static org.hamcrest.Matchers.containsString;
 
 import com.google.protobuf.util.Durations;
-import io.envoyproxy.controlplane.cache.CacheStatusInfo;
 import io.envoyproxy.controlplane.cache.NodeGroup;
-import io.envoyproxy.controlplane.cache.Resources;
 import io.envoyproxy.controlplane.cache.TestResources;
 import io.envoyproxy.controlplane.cache.v2.SimpleCache;
 import io.envoyproxy.controlplane.cache.v2.Snapshot;
@@ -26,10 +23,7 @@
 import io.envoyproxy.envoy.api.v2.core.Node;
 import io.grpc.netty.NettyServerBuilder;
 import io.restassured.http.ContentType;
-import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import org.junit.ClassRule;
 import org.junit.Test;
@@ -43,7 +37,7 @@ public class V2DiscoveryServerAdsWarmingClusterIT {
   private static final String CONFIG = "envoy/ads.v2.config.yaml";
   private static final String GROUP = "key";
   private static final Integer LISTENER_PORT = 10000;
-  private static final CustomCache<String> cache = new CustomCache<>(new NodeGroup<String>() {
+  private static final SimpleCache<String> cache = new SimpleCache<>(new NodeGroup<String>() {
     @Override public String hash(Node node) {
       return GROUP;
     }
@@ -60,7 +54,6 @@ public class V2DiscoveryServerAdsWarmingClusterIT {
   private static final NettyGrpcServerRule ADS = new NettyGrpcServerRule() {
     @Override
     protected void configureServerBuilder(NettyServerBuilder builder) {
-      ExecutorService executorService = Executors.newSingleThreadExecutor();
       final DiscoveryServerCallbacks callbacks = new DiscoveryServerCallbacks() {
             @Override
             public void onStreamOpen(long streamId, String typeUrl) {
@@ -82,7 +75,6 @@ public void onV3StreamRequest(long streamId,
             public void onStreamResponse(long streamId, DiscoveryRequest request, DiscoveryResponse response) {
               // Here we update a Snapshot with working cluster, but we change only CDS version, not EDS version.
               // This change allows to test if EDS will be sent anyway after CDS was sent.
-              createSnapshotWithWorkingClusterWithTheSameEdsVersion(request, executorService);
               onStreamResponseLatch.countDown();
             }
           };
@@ -131,30 +123,35 @@ public void validateTestRequestToEchoServerViaEnvoy() throws InterruptedExceptio
 
     String baseUri = String.format("http://%s:%d", ENVOY.getContainerIpAddress(), ENVOY.getMappedPort(LISTENER_PORT));
 
+    await().atMost(5, TimeUnit.SECONDS).ignoreExceptions().untilAsserted(
+        () -> given().baseUri(baseUri).contentType(ContentType.TEXT)
+            .when().get("/")
+            .then().statusCode(503));
+
+    // Here we update a Snapshot with working cluster, but we change only CDS version, not EDS version.
+    // This change allows to test if EDS will be sent anyway after CDS was sent.
+    createSnapshotWithWorkingClusterWithTheSameEdsVersion();
+
     await().atMost(5, TimeUnit.SECONDS).ignoreExceptions().untilAsserted(
         () -> given().baseUri(baseUri).contentType(ContentType.TEXT)
             .when().get("/")
             .then().statusCode(200)
             .and().body(containsString(UPSTREAM.response)));
   }
 
-  private static void createSnapshotWithWorkingClusterWithTheSameEdsVersion(DiscoveryRequest request,
-                                                                            ExecutorService executorService) {
-    if (request.getTypeUrl().equals(Resources.V2.CLUSTER_TYPE_URL)) {
-      executorService.submit(() -> cache.setSnapshot(
-          GROUP,
-          createSnapshot(true,
-              "upstream",
-              UPSTREAM.ipAddress(),
-              EchoContainer.PORT,
-              "listener0",
-              LISTENER_PORT,
-              "route0",
-              "2"))
-      );
-    }
+  private static void createSnapshotWithWorkingClusterWithTheSameEdsVersion() {
+    cache.setSnapshot(GROUP,
+        V2TestSnapshots.createSnapshot(true,
+            "upstream",
+            UPSTREAM.ipAddress(),
+            EchoContainer.PORT,
+            "listener0",
+            LISTENER_PORT,
+            "route0",
+            "2"));
   }
 
+
   private static Snapshot createSnapshotWithNotWorkingCluster(boolean ads,
                                                               String clusterName,
                                                               String endpointAddress,
@@ -197,34 +194,21 @@ private static Snapshot createSnapshotWithNotWorkingCluster(boolean ads,
   }
 
 
-  /**
-   * Code has been copied from io.envoyproxy.controlplane.cache.SimpleCache to show specific case when
-   * Envoy might stuck with warming cluster. Class has changed lines from method respondWithSpecificOrder which are
-   * responsible for responding for watches. Because to reproduce this problem we need a lot of connected Envoy's and
-   * changes to snapshot it is easier to reproduce this way.
+  /*
+   * In the previous versions of this tests we had a copied SimpleCache with respondWithSpecificOrder removed.
+   * With new versions of Envoy hitting this edge-case became highly improbable.
+   * Now this test checks only if a CDS change will also send EDS.
+   * 1. Envoy connects to control-plane
+   * 2. Snapshot already exists in control-plane <- other instance share same group
+   * 3. Control-plane respond with CDS in createWatch method
+   * 4. There is snapshot update which change CDS and EDS versions
+   * 5. Envoy sends EDS request
+   * 6. Control-plane respond with EDS in createWatch method
+   * 7. Envoy resume CDS and EDS requests.
+   * 8. Envoy sends request CDS
+   * 9. Control plane respond with CDS in createWatch method
+   * 10. Envoy sends EDS requests
+   * 11. Control plane doesn't respond because version hasn't changed
+   * 12. Cluster of service stays in warming phase
    */
-  static class CustomCache<T> extends SimpleCache<T> {
-
-    public CustomCache(NodeGroup<T> groups) {
-      super(groups);
-    }
-
-    @Override
-    protected void respondWithSpecificOrder(T group, Snapshot snapshot,
-        ConcurrentMap<Resources.ResourceType, CacheStatusInfo<T>> status) {
-      // This code has been removed to show specific case which is hard to reproduce in integration test:
-      //      1. Envoy connects to control-plane
-      //      2. Snapshot already exists in control-plane <- other instance share same group
-      //      3. Control-plane respond with CDS in createWatch method
-      //      4. There is snapshot update which change CDS and EDS versions
-      //      5. Envoy sends EDS request
-      //      6. Control-plane respond with EDS in createWatch method
-      //      7. Envoy resume CDS and EDS requests.
-      //      8. Envoy sends request CDS
-      //      9. Control plane respond with CDS in createWatch method
-      //      10. Envoy sends EDS requests
-      //      11. Control plane doesn't respond because version hasn't changed
-      //      12. Cluster of service stays in warming phase
-    }
-  }
 }
diff --git a/server/src/test/java/io/envoyproxy/controlplane/server/V3DiscoveryServerAdsWarmingClusterIT.java b/server/src/test/java/io/envoyproxy/controlplane/server/V3DiscoveryServerAdsWarmingClusterIT.java
@@ -8,9 +8,7 @@
 import static org.hamcrest.Matchers.containsString;
 
 import com.google.protobuf.util.Durations;
-import io.envoyproxy.controlplane.cache.CacheStatusInfo;
 import io.envoyproxy.controlplane.cache.NodeGroup;
-import io.envoyproxy.controlplane.cache.Resources;
 import io.envoyproxy.controlplane.cache.TestResources;
 import io.envoyproxy.controlplane.cache.v3.SimpleCache;
 import io.envoyproxy.controlplane.cache.v3.Snapshot;
@@ -26,10 +24,7 @@
 import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse;
 import io.grpc.netty.NettyServerBuilder;
 import io.restassured.http.ContentType;
-import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import org.junit.ClassRule;
 import org.junit.Test;
@@ -43,7 +38,7 @@ public class V3DiscoveryServerAdsWarmingClusterIT {
   private static final String CONFIG = "envoy/ads.v3.config.yaml";
   private static final String GROUP = "key";
   private static final Integer LISTENER_PORT = 10000;
-  private static final CustomCache<String> cache = new CustomCache<>(new NodeGroup<String>() {
+  private static final SimpleCache<String> cache = new SimpleCache<>(new NodeGroup<String>() {
     @Override public String hash(Node node) {
       throw new IllegalStateException("Unexpected v2 request in v3 test");
     }
@@ -60,7 +55,6 @@ public class V3DiscoveryServerAdsWarmingClusterIT {
   private static final NettyGrpcServerRule ADS = new NettyGrpcServerRule() {
     @Override
     protected void configureServerBuilder(NettyServerBuilder builder) {
-      ExecutorService executorService = Executors.newSingleThreadExecutor();
       final DiscoveryServerCallbacks callbacks = new DiscoveryServerCallbacks() {
         @Override
         public void onStreamOpen(long streamId, String typeUrl) {
@@ -87,9 +81,6 @@ public void onStreamResponse(long streamId, io.envoyproxy.envoy.api.v2.Discovery
         @Override
         public void onV3StreamResponse(long streamId, DiscoveryRequest request,
             DiscoveryResponse response) {
-          // Here we update a Snapshot with working cluster, but we change only CDS version, not EDS version.
-          // This change allows to test if EDS will be sent anyway after CDS was sent.
-          createSnapshotWithWorkingClusterWithTheSameEdsVersion(request, executorService);
           onStreamResponseLatch.countDown();
         }
       };
@@ -138,28 +129,33 @@ public void validateTestRequestToEchoServerViaEnvoy() throws InterruptedExceptio
 
     String baseUri = String.format("http://%s:%d", ENVOY.getContainerIpAddress(), ENVOY.getMappedPort(LISTENER_PORT));
 
+    await().atMost(5, TimeUnit.SECONDS).ignoreExceptions().untilAsserted(
+        () -> given().baseUri(baseUri).contentType(ContentType.TEXT)
+            .when().get("/")
+            .then().statusCode(503));
+
+    // Here we update a Snapshot with working cluster, but we change only CDS version, not EDS version.
+    // This change allows to test if EDS will be sent anyway after CDS was sent.
+    createSnapshotWithWorkingClusterWithTheSameEdsVersion();
+
     await().atMost(5, TimeUnit.SECONDS).ignoreExceptions().untilAsserted(
         () -> given().baseUri(baseUri).contentType(ContentType.TEXT)
             .when().get("/")
             .then().statusCode(200)
             .and().body(containsString(UPSTREAM.response)));
   }
 
-  private static void createSnapshotWithWorkingClusterWithTheSameEdsVersion(DiscoveryRequest request,
-      ExecutorService executorService) {
-    if (request.getTypeUrl().equals(Resources.V3.CLUSTER_TYPE_URL)) {
-      executorService.submit(() -> cache.setSnapshot(
-          GROUP,
-          createSnapshot(true,
-              "upstream",
-              UPSTREAM.ipAddress(),
-              EchoContainer.PORT,
-              "listener0",
-              LISTENER_PORT,
-              "route0",
-              "2"))
-      );
-    }
+  private static void createSnapshotWithWorkingClusterWithTheSameEdsVersion() {
+    cache.setSnapshot(
+        GROUP,
+        createSnapshot(true,
+            "upstream",
+            UPSTREAM.ipAddress(),
+            EchoContainer.PORT,
+            "listener0",
+            LISTENER_PORT,
+            "route0",
+            "2"));
   }
 
   private static Snapshot createSnapshotWithNotWorkingCluster(boolean ads,
@@ -204,35 +200,21 @@ private static Snapshot createSnapshotWithNotWorkingCluster(boolean ads,
         "2");
   }
 
-
-  /**
-   * Code has been copied from io.envoyproxy.controlplane.cache.SimpleCache to show specific case when
-   * Envoy might stuck with warming cluster. Class has changed lines from method respondWithSpecificOrder which are
-   * responsible for responding for watches. Because to reproduce this problem we need a lot of connected Envoy's and
-   * changes to snapshot it is easier to reproduce this way.
+  /*
+   * In the previous versions of this tests we had a copied SimpleCache with respondWithSpecificOrder removed.
+   * With new versions of Envoy hitting this edge-case became highly improbable.
+   * Now this test checks only if a CDS change will also send EDS.
+   * 1. Envoy connects to control-plane
+   * 2. Snapshot already exists in control-plane <- other instance share same group
+   * 3. Control-plane respond with CDS in createWatch method
+   * 4. There is snapshot update which change CDS and EDS versions
+   * 5. Envoy sends EDS request
+   * 6. Control-plane respond with EDS in createWatch method
+   * 7. Envoy resume CDS and EDS requests.
+   * 8. Envoy sends request CDS
+   * 9. Control plane respond with CDS in createWatch method
+   * 10. Envoy sends EDS requests
+   * 11. Control plane doesn't respond because version hasn't changed
+   * 12. Cluster of service stays in warming phase
    */
-  static class CustomCache<T> extends SimpleCache<T> {
-
-    public CustomCache(NodeGroup<T> groups) {
-      super(groups);
-    }
-
-    @Override
-    protected void respondWithSpecificOrder(T group, Snapshot snapshot,
-        ConcurrentMap<Resources.ResourceType, CacheStatusInfo<T>> status) {
-      // This code has been removed to show specific case which is hard to reproduce in integration test:
-      //      1. Envoy connects to control-plane
-      //      2. Snapshot already exists in control-plane <- other instance share same group
-      //      3. Control-plane respond with CDS in createWatch method
-      //      4. There is snapshot update which change CDS and EDS versions
-      //      5. Envoy sends EDS request
-      //      6. Control-plane respond with EDS in createWatch method
-      //      7. Envoy resume CDS and EDS requests.
-      //      8. Envoy sends request CDS
-      //      9. Control plane respond with CDS in createWatch method
-      //      10. Envoy sends EDS requests
-      //      11. Control plane doesn't respond because version hasn't changed
-      //      12. Cluster of service stays in warming phase
-    }
-  }
 }
