add allowDefaultEmptyEdsUpdate in server startup configs (#287)

Signed-off-by: Yifei Sun <ysun@squareup.com>
Co-authored-by: Yifei Sun <ysun@squareup.com>

Author: ysun-2023

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

@@ -22,13 +22,18 @@ public interface ConfigWatcher {
    * @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.
+   *
+   * @param allowDefaultEmptyEdsUpdate indicates if default empty EDS response should be sent when some clusters
+   *                                   in request are missing in snapshot. Supported in ADS mode.
+   *
    */
   Watch createWatch(
       boolean ads,
       XdsRequest request,
       Set<String> knownResourceNames,
       Consumer<Response> responseConsumer,
-      boolean hasClusterChanged);
+      boolean hasClusterChanged,
+      boolean allowDefaultEmptyEdsUpdate);
 
   /**
    * Returns a new configuration resource {@link Watch} for the given discovery request.

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

@@ -9,14 +9,15 @@
 import com.google.common.collect.Sets;
 import com.google.protobuf.Message;
 import io.envoyproxy.controlplane.cache.Resources.ResourceType;
-
+import io.envoyproxy.envoy.config.endpoint.v3.ClusterLoadAssignment;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
+import java.util.UUID;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReadWriteLock;
@@ -26,16 +27,18 @@
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import javax.annotation.concurrent.GuardedBy;
-
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 /**
  * {@code SimpleCache} provides a default implementation of {@link SnapshotCache}. It maintains a single versioned
  * {@link Snapshot} per node group. For the protocol to work correctly in ADS mode, EDS/RDS requests are responded to
- * only when all resources in the snapshot xDS response are named as part of the request. It is expected that the CDS
- * response names all EDS clusters, and the LDS response names all RDS routes in a snapshot, to ensure that Envoy makes
- * the request for all EDS clusters or RDS routes eventually.
+ * only when all resources in the snapshot xDS response are named as part of the request by default. It is expected
+ * that the CDS response names all EDS clusters, and the LDS response names all RDS routes in a snapshot, to ensure
+ * that Envoy makes the request for all EDS clusters or RDS routes eventually.
+ *<p/>
+ * when allowIncompleteEdsUpdate is true, we will send EDS response even if some clusters names are missing in the
+ * snapshot in ADS.
  *
  * <p>The snapshot can be partial, e.g. only include RDS or EDS resources.
  */
@@ -93,7 +96,7 @@ public Watch createWatch(
       XdsRequest request,
       Set<String> knownResourceNames,
       Consumer<Response> responseConsumer) {
-    return createWatch(ads, request, knownResourceNames, responseConsumer, false);
+    return createWatch(ads, request, knownResourceNames, responseConsumer, false, false);
   }
 
   /**
@@ -105,7 +108,8 @@ public Watch createWatch(
       XdsRequest request,
       Set<String> knownResourceNames,
       Consumer<Response> responseConsumer,
-      boolean hasClusterChanged) {
+      boolean hasClusterChanged,
+      boolean allowDefaultEmptyEdsUpdate) {
     ResourceType requestResourceType = request.getResourceType();
     Preconditions.checkNotNull(requestResourceType, "unsupported type URL %s",
         request.getTypeUrl());
@@ -124,7 +128,7 @@ public Watch createWatch(
       String version = snapshot == null ? "" : snapshot.version(requestResourceType,
           request.getResourceNamesList());
 
-      Watch watch = new Watch(ads, request, responseConsumer);
+      Watch watch = new Watch(ads, allowDefaultEmptyEdsUpdate, request, responseConsumer);
 
       if (snapshot != null) {
         Set<String> requestedResources = ImmutableSet.copyOf(request.getResourceNamesList());
@@ -440,13 +444,14 @@ protected void respondWithSpecificOrder(T group,
   }
 
   private Response createResponse(XdsRequest request, Map<String, VersionedResource<?>> resources,
-                                  String version) {
+                                  String version, boolean allowDefaultResource) {
     Collection<? extends Message> filtered = request.getResourceNamesList().isEmpty()
         ? resources.values().stream()
         .map(VersionedResource::resource)
         .collect(Collectors.toList())
         : request.getResourceNamesList().stream()
-        .map(resources::get)
+        .map(name -> resources.getOrDefault(name,
+            allowDefaultResource ? defaultResource(name, request.getResourceType()) : null))
         .filter(Objects::nonNull)
         .map(VersionedResource::resource)
         .collect(Collectors.toList());
@@ -458,21 +463,35 @@ private boolean respond(Watch watch, U snapshot, T group) {
     Map<String, VersionedResource<?>> snapshotResources =
         snapshot.versionedResources(watch.request().getResourceType());
 
+    boolean allowDefaultResource = false;
     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()) {
-        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().getResourceType(), watch.request().getResourceNamesList()),
-            String.join(", ", watch.request().getResourceNamesList()),
-            String.join(", ", missingNames));
+        if (watch.allowDefaultEmptyEdsUpdate() && watch.request().getResourceType().equals(ResourceType.ENDPOINT)) {
+          allowDefaultResource = true;
+          LOGGER.info(
+              "responding with empty ClusterLoadAssignments in ADS mode for {} from node {} at version {} "
+                  + "for request [{}] and [{}] not in snapshot",
+              watch.request().getTypeUrl(),
+              group,
+              snapshot.version(watch.request().getResourceType(), watch.request().getResourceNamesList()),
+              String.join(", ", watch.request().getResourceNamesList()),
+              String.join(", ", missingNames));
+        } else {
+          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().getResourceType(), watch.request().getResourceNamesList()),
+              String.join(", ", watch.request().getResourceNamesList()),
+              String.join(", ", missingNames));
 
-        return false;
+          return false;
+        }
       }
     }
 
@@ -488,7 +507,8 @@ private boolean respond(Watch watch, U snapshot, T group) {
     Response response = createResponse(
         watch.request(),
         snapshotResources,
-        version);
+        version,
+        allowDefaultResource);
 
     try {
       watch.respond(response);
@@ -573,6 +593,14 @@ private ResponseState respondDelta(DeltaWatch watch,
     return ResponseState.CANCELLED;
   }
 
+  private VersionedResource<?> defaultResource(String resourceName, ResourceType resourceType) {
+    if (resourceType.equals(ResourceType.ENDPOINT)) {
+      return VersionedResource.create(ClusterLoadAssignment.newBuilder().setClusterName(resourceName).build(),
+          UUID.randomUUID().toString());
+    }
+    throw new IllegalArgumentException(String.format("no default resource for resourceType: [%s]", resourceType));
+  }
+
   private enum ResponseState {
     RESPONDED,
     UNRESPONDED,

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

@@ -2,6 +2,7 @@
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.protobuf.Any;
+import com.google.protobuf.Duration;
 import com.google.protobuf.util.Durations;
 import io.envoyproxy.envoy.config.cluster.v3.Cluster;
 import io.envoyproxy.envoy.config.core.v3.Address;
@@ -50,6 +51,7 @@ public class TestResources {
   public static Cluster createCluster(String clusterName) {
     ConfigSource edsSource =
         ConfigSource.newBuilder()
+            .setInitialFetchTimeout(Duration.newBuilder().setSeconds(0))
             .setAds(AggregatedConfigSource.getDefaultInstance())
             .setResourceApiVersion(ApiVersion.V3)
             .build();

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

@@ -8,17 +8,22 @@
  */
 public class Watch extends AbstractWatch<XdsRequest, Response> {
   private final boolean ads;
+  private final boolean allowDefaultEmptyEdsUpdate;
 
   /**
    * Construct a watch.
    *
-   * @param ads              is this watch for an ADS request?
-   * @param request          the original request for the watch
-   * @param responseConsumer handler for outgoing response messages
+   * @param ads                      is this watch for an ADS request?
+   * @param allowDefaultEmptyEdsUpdate whether allow responding incomplete EDS response
+   *                                 when some clusters are missing in the snapshot
+   * @param request                  the original request for the watch
+   * @param responseConsumer         handler for outgoing response messages
    */
-  public Watch(boolean ads, XdsRequest request, Consumer<Response> responseConsumer) {
+  public Watch(boolean ads, boolean allowDefaultEmptyEdsUpdate,
+      XdsRequest request, Consumer<Response> responseConsumer) {
     super(request, responseConsumer);
     this.ads = ads;
+    this.allowDefaultEmptyEdsUpdate = allowDefaultEmptyEdsUpdate;
   }
 
   /**
@@ -28,4 +33,7 @@ public boolean ads() {
     return ads;
   }
 
+  public boolean allowDefaultEmptyEdsUpdate() {
+    return allowDefaultEmptyEdsUpdate;
+  }
 }

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

@@ -44,20 +44,21 @@ public void lastWatchRequestTimeReturnsExpectedValueIfSet() {
   @Test
   public void numWatchesReturnsExpectedSize() {
     final boolean ads = ThreadLocalRandom.current().nextBoolean();
+    final boolean allowDefaultEmptyEdsUpdate = ThreadLocalRandom.current().nextBoolean();
     final long watchId1 = ThreadLocalRandom.current().nextLong(10000, 50000);
     final long watchId2 = ThreadLocalRandom.current().nextLong(50000, 100000);
 
     CacheStatusInfo<Node> info = new CacheStatusInfo<>(Node.getDefaultInstance());
 
     assertThat(info.numWatches()).isZero();
 
-    info.setWatch(watchId1, new Watch(ads,
+    info.setWatch(watchId1, new Watch(ads, allowDefaultEmptyEdsUpdate,
         XdsRequest.create(DiscoveryRequest.getDefaultInstance()), r -> { }));
 
     assertThat(info.numWatches()).isEqualTo(1);
     assertThat(info.watchIds()).containsExactlyInAnyOrder(watchId1);
 
-    info.setWatch(watchId2, new Watch(ads,
+    info.setWatch(watchId2, new Watch(ads, allowDefaultEmptyEdsUpdate,
         XdsRequest.create(DiscoveryRequest.getDefaultInstance()), r -> { }));
 
     assertThat(info.numWatches()).isEqualTo(2);
@@ -72,14 +73,15 @@ public void numWatchesReturnsExpectedSize() {
   @Test
   public void watchesRemoveIfRemovesExpectedWatches() {
     final boolean ads = ThreadLocalRandom.current().nextBoolean();
+    final boolean allowDefaultEmptyEdsUpdate = ThreadLocalRandom.current().nextBoolean();
     final long watchId1 = ThreadLocalRandom.current().nextLong(10000, 50000);
     final long watchId2 = ThreadLocalRandom.current().nextLong(50000, 100000);
 
     CacheStatusInfo<Node> info = new CacheStatusInfo<>(Node.getDefaultInstance());
 
-    info.setWatch(watchId1, new Watch(ads,
+    info.setWatch(watchId1, new Watch(ads, allowDefaultEmptyEdsUpdate,
         XdsRequest.create(DiscoveryRequest.getDefaultInstance()), r -> { }));
-    info.setWatch(watchId2, new Watch(ads,
+    info.setWatch(watchId2, new Watch(ads, allowDefaultEmptyEdsUpdate,
         XdsRequest.create(DiscoveryRequest.getDefaultInstance()), r -> { }));
 
     assertThat(info.numWatches()).isEqualTo(2);
@@ -94,15 +96,16 @@ public void watchesRemoveIfRemovesExpectedWatches() {
   @Test
   public void testConcurrentSetWatchAndRemove() {
     final boolean ads = ThreadLocalRandom.current().nextBoolean();
+    final boolean allowDefaultEmptyEdsUpdate = ThreadLocalRandom.current().nextBoolean();
     final int watchCount = 50;
 
     CacheStatusInfo<Node> info = new CacheStatusInfo<>(Node.getDefaultInstance());
 
     Collection<Long> watchIds = LongStream.range(0, watchCount).boxed().collect(Collectors.toList());
 
     watchIds.parallelStream().forEach(watchId -> {
-      Watch watch = new Watch(ads, XdsRequest.create(DiscoveryRequest.getDefaultInstance()),
-          r -> { });
+      Watch watch = new Watch(ads, allowDefaultEmptyEdsUpdate,
+          XdsRequest.create(DiscoveryRequest.getDefaultInstance()), r -> { });
 
       info.setWatch(watchId, watch);
     });

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

@@ -19,18 +19,21 @@ public class WatchTest {
   @Test
   public void adsReturnsGivenValue() {
     final boolean ads = ThreadLocalRandom.current().nextBoolean();
+    final boolean allowDefaultEmptyEdsUpdate = ThreadLocalRandom.current().nextBoolean();
 
-    Watch watch = new Watch(ads, XdsRequest.create(DiscoveryRequest.getDefaultInstance()),
-        r -> { });
+    Watch watch = new Watch(ads, allowDefaultEmptyEdsUpdate,
+        XdsRequest.create(DiscoveryRequest.getDefaultInstance()), r -> { });
 
     assertThat(watch.ads()).isEqualTo(ads);
   }
 
   @Test
   public void isCancelledTrueAfterCancel() {
     final boolean ads = ThreadLocalRandom.current().nextBoolean();
+    final boolean allowDefaultEmptyEdsUpdate = ThreadLocalRandom.current().nextBoolean();
 
-    Watch watch = new Watch(ads, XdsRequest.create(DiscoveryRequest.getDefaultInstance()), r -> { });
+    Watch watch = new Watch(ads, allowDefaultEmptyEdsUpdate,
+        XdsRequest.create(DiscoveryRequest.getDefaultInstance()), r -> { });
 
     assertThat(watch.isCancelled()).isFalse();
 
@@ -42,10 +45,12 @@ public void isCancelledTrueAfterCancel() {
   @Test
   public void cancelWithStopCallsStop() {
     final boolean ads = ThreadLocalRandom.current().nextBoolean();
+    final boolean allowDefaultEmptyEdsUpdate = ThreadLocalRandom.current().nextBoolean();
 
     AtomicInteger stopCount = new AtomicInteger();
 
-    Watch watch = new Watch(ads, XdsRequest.create(DiscoveryRequest.getDefaultInstance()), r -> { });
+    Watch watch = new Watch(ads, allowDefaultEmptyEdsUpdate,
+        XdsRequest.create(DiscoveryRequest.getDefaultInstance()), r -> { });
 
     watch.setStop(stopCount::getAndIncrement);
 
@@ -62,6 +67,7 @@ public void cancelWithStopCallsStop() {
   @Test
   public void responseHandlerExecutedForResponsesUntilCancelled() {
     final boolean ads = ThreadLocalRandom.current().nextBoolean();
+    final boolean allowDefaultEmptyEdsUpdate = ThreadLocalRandom.current().nextBoolean();
 
     Response response1 = Response.create(
         XdsRequest.create(DiscoveryRequest.getDefaultInstance()),
@@ -80,7 +86,8 @@ public void responseHandlerExecutedForResponsesUntilCancelled() {
 
     List<Response> responses = new LinkedList<>();
 
-    Watch watch = new Watch(ads, XdsRequest.create(DiscoveryRequest.getDefaultInstance()), responses::add);
+    Watch watch = new Watch(ads, allowDefaultEmptyEdsUpdate,
+        XdsRequest.create(DiscoveryRequest.getDefaultInstance()), responses::add);
 
     try {
       watch.respond(response1);

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

@@ -108,6 +108,36 @@ public void invalidNamesListShouldReturnWatcherWithNoResponseInAdsMode() {
     assertThatWatchIsOpenWithNoResponses(new WatchAndTracker(watch, responseTracker));
   }
 
+  @Test
+  public void invalidNamesListShouldReturnWatcherWithDefaultEmptyResponseInAdsModeAndAllowDefaultEmptyEdsUpdate() {
+    SimpleCache<String> cache = new SimpleCache<>(new SingleNodeGroup());
+
+    cache.setSnapshot(SingleNodeGroup.GROUP, SNAPSHOT1);
+
+    ResponseTracker responseTracker = new ResponseTracker();
+
+    Watch watch = cache.createWatch(
+        true,
+        XdsRequest.create(DiscoveryRequest.newBuilder()
+            .setNode(Node.getDefaultInstance())
+            .setTypeUrl(Resources.V3.ENDPOINT_TYPE_URL)
+            .addResourceNames("none")
+            .build()),
+        Collections.emptySet(),
+        responseTracker,
+        false,
+        true);
+
+    assertThat(watch.isCancelled()).isFalse();
+    Assertions.assertThat(responseTracker.responses).isNotEmpty();
+    Assertions.assertThat(responseTracker.responses.size()).isEqualTo(1);
+    Message[] messages = responseTracker.responses.getFirst().resources().toArray(new Message[0]);
+
+    assertThat(messages).containsExactlyElementsOf(
+        Collections.singleton(ClusterLoadAssignment.newBuilder()
+            .setClusterName("none").build()));
+  }
+
   @Test
   public void invalidNamesListShouldReturnWatcherWithResponseInXdsMode() {
     SimpleCache<String> cache = new SimpleCache<>(new SingleNodeGroup());
@@ -175,7 +205,8 @@ public void shouldSendEdsWhenClusterChangedButEdsVersionDidnt() {
             .build()),
         Sets.newHashSet(""),
         responseTracker,
-        true);
+        true,
+        false);
 
     assertThat(watch.request().getTypeUrl()).isEqualTo(Resources.V3.ENDPOINT_TYPE_URL);
     assertThat(watch.request().getResourceNamesList()).containsExactlyElementsOf(

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

@@ -0,0 +1,8 @@
+package io.envoyproxy.controlplane.server;
+
+public class DefaultStartupConfigs implements StartupConfigs {
+  @Override
+  public boolean allowDefaultEmptyEdsUpdate() {
+    return false;
+  }
+}