Delta XDS - improved (#166)

* delta xds Signed-off-by: Sebastian Schepens <sebastian.schepens@mercadolibre.com> Signed-off-by: slonka <slonka@users.noreply.github.com> * fixes Signed-off-by: Sebastian Schepens <sebastian.schepens@mercadolibre.com> Signed-off-by: slonka <slonka@users.noreply.github.com> * method naming Signed-off-by: Sebastian Schepens <sebastian.schepens@mercadolibre.com> Signed-off-by: slonka <slonka@users.noreply.github.com> * Building version of non-breaking changes Signed-off-by: slonka <slonka@users.noreply.github.com> * Get integration tests passing Signed-off-by: slonka <slonka@users.noreply.github.com> * Add delta tests back Signed-off-by: slonka <slonka@users.noreply.github.com> * Rewrite V3DiscoveryServerAdsDeltaResourcesIT to have a 2 second poll delay to account for potential delays in writing to the wire (we don't want that to happen) Signed-off-by: slonka <slonka@users.noreply.github.com> * Add comment to trigger PR build Signed-off-by: slonka <slonka@users.noreply.github.com> * Get V3DiscoveryServerXdsDeltaResourcesIT passing Signed-off-by: slonka <slonka@users.noreply.github.com> * Update protobuf to envoy 1.16.0 (#155) * Update protobuf to envoy 1.16.0 Signed-off-by: Lukasz Jedryczka <lukasz.jedryczka@allegro.pl> * Fixing test by setting -boostrap-version 2 flag Signed-off-by: Lukasz Jedryczka <lukasz.jedryczka@allegro.pl> * Update protobuf to envoy 1.16.0 Signed-off-by: Lukasz Jedryczka <lukasz.jedryczka@allegro.pl> * Information about update envoy image version in README.md Signed-off-by: wookieJ <lukaszjedryczka.biuro@gmail.com> Signed-off-by: slonka <slonka@users.noreply.github.com> * release: prepare release v0.1.25 Signed-off-by: slonka <slonka@users.noreply.github.com> * release: prepare for next development iteration Signed-off-by: slonka <slonka@users.noreply.github.com> * release: prepare release v0.1.26 Signed-off-by: slonka <slonka@users.noreply.github.com> * release: prepare for next development iteration Signed-off-by: slonka <slonka@users.noreply.github.com> * Bump nexus release plugin timeout to 20 minutes (#156) Signed-off-by: slonka <slonka@users.noreply.github.com> * release: prepare release v0.1.27 Signed-off-by: slonka <slonka@users.noreply.github.com> * release: prepare for next development iteration Signed-off-by: slonka <slonka@users.noreply.github.com> * ci: fixes javadoc and jacoco plugin issues (#158) Signed-off-by: karthik <listaction@gmail.com> Signed-off-by: slonka <slonka@users.noreply.github.com> * Update api to v1 17 (#159) * Update protobuf to envoy 1.17 Signed-off-by: Lukasz Dziedziak <lukasz.dziedziak@allegro.pl> * Use v2 version Signed-off-by: Lukasz Dziedziak <lukasz.dziedziak@allegro.pl> * Support V2/V3 in Envoy - remove V2 in separate PR Signed-off-by: Lukasz Dziedziak <lukasz.dziedziak@allegro.pl> * UDPA download - split directory create/copy Signed-off-by: Lukasz Dziedziak <lukasz.dziedziak@allegro.pl> Signed-off-by: slonka <slonka@users.noreply.github.com> * Fix references to main branch after rename (#160) Signed-off-by: slonka <slonka@users.noreply.github.com> * release: prepare release v0.1.28 Signed-off-by: slonka <slonka@users.noreply.github.com> * release: prepare for next development iteration Signed-off-by: slonka <slonka@users.noreply.github.com> * Change version to 0.1.29-delta-xds-slonka-SNAPSHOT Signed-off-by: slonka <slonka@users.noreply.github.com> * Revert ads configs to main Signed-off-by: slonka <slonka@users.noreply.github.com> * Hash bytes array not string Signed-off-by: slonka <slonka@users.noreply.github.com> * Remove unused import Signed-off-by: slonka <slonka@users.noreply.github.com> * Revert snapshot name Signed-off-by: slonka <slonka@users.noreply.github.com> * Remove respondDeltaTracked since it's not used anywhere Signed-off-by: slonka <slonka@users.noreply.github.com> * Delta xds non breaking hash bytes refactor (#181) * refactor delta xds Signed-off-by: radoslaw.chrzanowski <radoslaw.chrzanowski@allegro.pl> * refactor setDeltaWatch and remove code duplication Signed-off-by: radoslaw.chrzanowski <radoslaw.chrzanowski@allegro.pl> * resolve TODO in SnapshotTest Signed-off-by: radoslaw.chrzanowski <radoslaw.chrzanowski@allegro.pl> * Change version to 0.1.29-delta-xds-slonka-SNAPSHOT Signed-off-by: radoslaw.chrzanowski <radoslaw.chrzanowski@allegro.pl> * DEPLOY_BRANCH added for snapshot deploy Signed-off-by: radoslaw.chrzanowski <radoslaw.chrzanowski@allegro.pl> * changes after merge master with new envoy api Signed-off-by: radoslaw.chrzanowski <radoslaw.chrzanowski@allegro.pl> * make CacheStatusInfoAggregator public Signed-off-by: radoslaw.chrzanowski <radoslaw.chrzanowski@allegro.pl> * make GroupCacheStatusInfo and MutableStatusInfo public Signed-off-by: radoslaw.chrzanowski <radoslaw.chrzanowski@allegro.pl> * refactor SimpleCache Signed-off-by: radoslaw.chrzanowski <radoslaw.chrzanowski@allegro.pl> * create resources map only once Signed-off-by: radoslaw.chrzanowski <radoslaw.chrzanowski@allegro.pl> * build snapshot with improved performance Signed-off-by: radoslaw.chrzanowski <radoslaw.chrzanowski@allegro.pl> * refactor creating snapshot resources due to performance improvements Signed-off-by: radoslaw.chrzanowski <radoslaw.chrzanowski@allegro.pl> * create hash version from string Signed-off-by: radoslaw.chrzanowski <radoslaw.chrzanowski@allegro.pl> * remove custom snapshot version and deploy branch env Signed-off-by: radoslaw.chrzanowski <radoslaw.chrzanowski@allegro.pl> * fix test after merge Signed-off-by: radoslaw.chrzanowski <radoslaw.chrzanowski@allegro.pl> * fix V3DeltaDiscoveryServerCallbacks description Signed-off-by: radoslaw.chrzanowski <radoslaw.chrzanowski@allegro.pl> Signed-off-by: Sebastian Schepens <sebastian.schepens@mercadolibre.com> Signed-off-by: slonka <slonka@users.noreply.github.com> Signed-off-by: wookieJ <lukaszjedryczka.biuro@gmail.com> Signed-off-by: karthik <listaction@gmail.com> Signed-off-by: Lukasz Dziedziak <lukasz.dziedziak@allegro.pl> Signed-off-by: radoslaw.chrzanowski <radoslaw.chrzanowski@allegro.pl> Co-authored-by: Sebastian Schepens <sebastian.schepens@mercadolibre.com> Co-authored-by: mgajda <mgajda@hubspot.com> Co-authored-by: Łukasz Jędryczka <34069409+wookieJ@users.noreply.github.com> Co-authored-by: envoy-bot <envoy-bot@users.noreply.github.com> Co-authored-by: Karthik Ram <listaction@gmail.com> Co-authored-by: Łukasz Dziedziak <lukasz.dziedziak@allegro.pl> Co-authored-by: Radek Chrzanowski <ferdudas97@gmail.com> Co-authored-by: radoslaw.chrzanowski <radoslaw.chrzanowski@allegro.pl>
envoyproxy · Sep 9, 2022 · f54d234 · f54d234
1 parent 76383ff
commit f54d234
Show file tree

Hide file tree

Showing 48 changed files with 2,418 additions and 447 deletions.
diff --git a/cache/src/main/java/io/envoyproxy/controlplane/cache/AbstractWatch.java b/cache/src/main/java/io/envoyproxy/controlplane/cache/AbstractWatch.java
@@ -0,0 +1,73 @@
+package io.envoyproxy.controlplane.cache;
+
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+import java.util.function.Consumer;
+
+public abstract class AbstractWatch<V, T> {
+
+  private static final AtomicIntegerFieldUpdater<AbstractWatch> isCancelledUpdater =
+      AtomicIntegerFieldUpdater.newUpdater(AbstractWatch.class, "isCancelled");
+  private final V request;
+  private final Consumer<T> responseConsumer;
+  private volatile int isCancelled = 0;
+  private Runnable stop;
+
+  /**
+   * Construct a watch.
+   *
+   * @param request          the original request for the watch
+   * @param responseConsumer handler for outgoing response messages
+   */
+  public AbstractWatch(V request, Consumer<T> responseConsumer) {
+    this.request = request;
+    this.responseConsumer = responseConsumer;
+  }
+
+  /**
+   * Cancel the watch. A watch must be cancelled in order to complete its resource stream and free resources. Cancel
+   * may be called multiple times, with each subsequent call being a no-op.
+   */
+  public void cancel() {
+    if (isCancelledUpdater.compareAndSet(this, 0, 1)) {
+      if (stop != null) {
+        stop.run();
+      }
+    }
+  }
+
+  /**
+   * Returns boolean indicating whether or not the watch has been cancelled.
+   */
+  public boolean isCancelled() {
+    return isCancelledUpdater.get(this) == 1;
+  }
+
+  /**
+   * Returns the original request for the watch.
+   */
+  public V request() {
+    return request;
+  }
+
+  /**
+   * Sends the given response to the watch's response handler.
+   *
+   * @param response the response to be handled
+   * @throws WatchCancelledException if the watch has already been cancelled
+   */
+  public void respond(T response) throws WatchCancelledException {
+    if (isCancelled()) {
+      throw new WatchCancelledException();
+    }
+
+    responseConsumer.accept(response);
+  }
+
+  /**
+   * Sets the callback method to be executed when the watch is cancelled. Even if cancel is executed multiple times, it
+   * ensures that this stop callback is only executed once.
+   */
+  public void setStop(Runnable stop) {
+    this.stop = stop;
+  }
+}
diff --git a/cache/src/main/java/io/envoyproxy/controlplane/cache/Cache.java b/cache/src/main/java/io/envoyproxy/controlplane/cache/Cache.java
@@ -20,5 +20,5 @@ public interface Cache<T> extends ConfigWatcher {
    *
    * @param group the node group whose status is being fetched
    */
-  StatusInfo statusInfo(T group);
+  StatusInfo<T> statusInfo(T group);
 }
diff --git a/cache/src/main/java/io/envoyproxy/controlplane/cache/CacheStatusInfo.java b/cache/src/main/java/io/envoyproxy/controlplane/cache/CacheStatusInfo.java
@@ -1,94 +1,14 @@
 package io.envoyproxy.controlplane.cache;
 
-import com.google.common.collect.ImmutableSet;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-import java.util.function.BiFunction;
 import javax.annotation.concurrent.ThreadSafe;
 
 /**
  * {@code CacheStatusInfo} provides a default implementation of {@link StatusInfo} for use in {@link Cache}
  * implementations.
  */
 @ThreadSafe
-public class CacheStatusInfo<T> implements StatusInfo<T> {
-
-  private final T nodeGroup;
-
-  private final ConcurrentMap<Long, Watch> watches = new ConcurrentHashMap<>();
-  private volatile long lastWatchRequestTime;
-
+public class CacheStatusInfo<T> extends MutableStatusInfo<T, Watch> {
   public CacheStatusInfo(T nodeGroup) {
-    this.nodeGroup = nodeGroup;
-  }
-
-  /**
-   * {@inheritDoc}
-   */
-  @Override
-  public long lastWatchRequestTime() {
-    return lastWatchRequestTime;
-  }
-
-  /**
-   * {@inheritDoc}
-   */
-  @Override
-  public T nodeGroup() {
-    return nodeGroup;
-  }
-
-  /**
-   * {@inheritDoc}
-   */
-  @Override
-  public int numWatches() {
-    return watches.size();
-  }
-
-  /**
-   * Removes the given watch from the tracked collection of watches.
-   *
-   * @param watchId the ID for the watch that should be removed
-   */
-  public void removeWatch(long watchId) {
-    watches.remove(watchId);
-  }
-
-  /**
-   * Sets the timestamp of the last discovery watch request.
-   *
-   * @param lastWatchRequestTime the latest watch request timestamp
-   */
-  public void setLastWatchRequestTime(long lastWatchRequestTime) {
-    this.lastWatchRequestTime = lastWatchRequestTime;
-  }
-
-  /**
-   * Adds the given watch to the tracked collection of watches.
-   *
-   * @param watchId the ID for the watch that should be added
-   * @param watch   the watch that should be added
-   */
-  public void setWatch(long watchId, Watch watch) {
-    watches.put(watchId, watch);
-  }
-
-  /**
-   * Returns the set of IDs for all watched currently being tracked.
-   */
-  public Set<Long> watchIds() {
-    return ImmutableSet.copyOf(watches.keySet());
-  }
-
-  /**
-   * Iterate over all tracked watches and execute the given function. If it returns {@code true}, then the watch is
-   * removed from the tracked collection. If it returns {@code false}, then the watch is not removed.
-   *
-   * @param filter the function to execute on each watch
-   */
-  public void watchesRemoveIf(BiFunction<Long, Watch, Boolean> filter) {
-    watches.entrySet().removeIf(entry -> filter.apply(entry.getKey(), entry.getValue()));
+    super(nodeGroup);
   }
 }
diff --git a/cache/src/main/java/io/envoyproxy/controlplane/cache/CacheStatusInfoAggregator.java b/cache/src/main/java/io/envoyproxy/controlplane/cache/CacheStatusInfoAggregator.java
@@ -0,0 +1,77 @@
+package io.envoyproxy.controlplane.cache;
+
+import java.util.Collection;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+public class CacheStatusInfoAggregator<T> {
+  private final ConcurrentMap<T, ConcurrentMap<Resources.ResourceType, CacheStatusInfo<T>>> statuses =
+      new ConcurrentHashMap<>();
+  private final ConcurrentMap<T, ConcurrentMap<Resources.ResourceType, DeltaCacheStatusInfo<T>>> deltaStatuses =
+      new ConcurrentHashMap<>();
+
+  public Collection<T> groups() {
+    return Stream.concat(statuses.keySet().stream(), deltaStatuses.keySet().stream()).collect(Collectors.toSet());
+  }
+
+  public void remove(T group) {
+    statuses.remove(group);
+    deltaStatuses.remove(group);
+  }
+
+  /**
+   * Returns map of delta status infos for group identifier.
+   *
+   * @param group group identifier.
+   */
+  public Map<Resources.ResourceType, DeltaCacheStatusInfo<T>> getDeltaStatus(T group) {
+    return deltaStatuses.getOrDefault(group, new ConcurrentHashMap<>());
+  }
+
+  /**
+   * Returns map of status infos for group identifier.
+   *
+   * @param group group identifier.
+   */
+  public Map<Resources.ResourceType, CacheStatusInfo<T>> getStatus(T group) {
+    return statuses.getOrDefault(group, new ConcurrentHashMap<>());
+  }
+
+  /**
+   * Check if statuses for specific group have any watcher.
+   *
+   * @param group group identifier.
+   * @return true if statuses for specific group have any watcher.
+   */
+  public boolean hasStatuses(T group) {
+    Map<Resources.ResourceType, CacheStatusInfo<T>> status = getStatus(group);
+    Map<Resources.ResourceType, DeltaCacheStatusInfo<T>> deltaStatus = getDeltaStatus(group);
+    return status.values().stream().mapToLong(CacheStatusInfo::numWatches).sum()
+        + deltaStatus.values().stream().mapToLong(DeltaCacheStatusInfo::numWatches).sum() > 0;
+  }
+
+  /**
+   * Returns delta status info for group identifier and creates new one if it doesn't exist.
+   *
+   * @param group        group identifier.
+   * @param resourceType resource type.
+   */
+  public DeltaCacheStatusInfo<T> getOrAddDeltaStatusInfo(T group, Resources.ResourceType resourceType) {
+    return deltaStatuses.computeIfAbsent(group, g -> new ConcurrentHashMap<>())
+        .computeIfAbsent(resourceType, s -> new DeltaCacheStatusInfo<>(group));
+  }
+
+  /**
+   * Returns status info for group identifier and creates new one if it doesn't exist.
+   *
+   * @param group        group identifier.
+   * @param resourceType resource type.
+   */
+  public CacheStatusInfo<T> getOrAddStatusInfo(T group, Resources.ResourceType resourceType) {
+    return statuses.computeIfAbsent(group, g -> new ConcurrentHashMap<>())
+        .computeIfAbsent(resourceType, s -> new CacheStatusInfo<>(group));
+  }
+}
diff --git a/cache/src/main/java/io/envoyproxy/controlplane/cache/ConfigWatcher.java b/cache/src/main/java/io/envoyproxy/controlplane/cache/ConfigWatcher.java
@@ -1,5 +1,6 @@
 package io.envoyproxy.controlplane.cache;
 
+import java.util.Map;
 import java.util.Set;
 import java.util.function.Consumer;
 import javax.annotation.concurrent.ThreadSafe;
@@ -28,4 +29,25 @@ Watch createWatch(
       Set<String> knownResourceNames,
       Consumer<Response> responseConsumer,
       boolean hasClusterChanged);
+
+  /**
+   * Returns a new configuration resource {@link Watch} for the given discovery request.
+   *
+   * @param request           the discovery request (node, names, etc.) to use to generate the watch
+   * @param requesterVersion  the last version applied by the requester
+   * @param resourceVersions  resources that are already known to the requester
+   * @param pendingResources  resources that the caller is waiting for
+   * @param isWildcard        indicates if the stream is in wildcard mode
+   * @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.
+   */
+  DeltaWatch createDeltaWatch(
+      DeltaXdsRequest request,
+      String requesterVersion,
+      Map<String, String> resourceVersions,
+      Set<String> pendingResources,
+      boolean isWildcard,
+      Consumer<DeltaResponse> responseConsumer,
+      boolean hasClusterChanged);
 }
diff --git a/cache/src/main/java/io/envoyproxy/controlplane/cache/DeltaCacheStatusInfo.java b/cache/src/main/java/io/envoyproxy/controlplane/cache/DeltaCacheStatusInfo.java
@@ -0,0 +1,8 @@
+package io.envoyproxy.controlplane.cache;
+
+public class DeltaCacheStatusInfo<T> extends MutableStatusInfo<T, DeltaWatch> {
+
+  public DeltaCacheStatusInfo(T nodeGroup) {
+    super(nodeGroup);
+  }
+}
diff --git a/cache/src/main/java/io/envoyproxy/controlplane/cache/DeltaResponse.java b/cache/src/main/java/io/envoyproxy/controlplane/cache/DeltaResponse.java
@@ -0,0 +1,41 @@
+package io.envoyproxy.controlplane.cache;
+
+import com.google.auto.value.AutoValue;
+import com.google.protobuf.Message;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * {@code Response} is a data class that contains the response for an assumed configuration type.
+ */
+@AutoValue
+public abstract class DeltaResponse {
+
+  public static DeltaResponse create(DeltaXdsRequest request,
+      Map<String, VersionedResource<?>> resources,
+      List<String> removedResources,
+      String version) {
+    return new AutoValue_DeltaResponse(request, resources, removedResources, version);
+  }
+
+  /**
+   * Returns the original request associated with the response.
+   */
+  public abstract DeltaXdsRequest request();
+
+  /**
+   * Returns the resources to include in the response.
+   */
+  public abstract Map<String, VersionedResource<? extends Message>> resources();
+
+  /**
+   * Returns the removed resources to include in the response.
+   */
+  public abstract List<String> removedResources();
+
+  /**
+   * Returns the version of the resources as tracked by the cache for the given type. Envoy responds with this version
+   * as an acknowledgement.
+   */
+  public abstract String version();
+}