cache: Per resource versioning (#94)

This adds the concept of ResourceVersionResolver that receives a list of resource names and should return a version for those resources.
This could be used by users of java-control-plane to implement per-resource versioning in EDS and RDS.
The adds new methods but does not break existing APIs.

Signed-off-by: Jakub Dyszkiewicz <[email protected]>
Signed-off-by: sschepens <[email protected]>

Author: sschepens

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

@@ -0,0 +1,27 @@
+package io.envoyproxy.controlplane.cache;
+
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * {@code ResourceVersionResolver} calculates a version for resources in {@link SnapshotCache}.
+ * It can be used to take advantage of Envoy's per-resource versioning.
+ */
+public interface ResourceVersionResolver {
+  /**
+   * Returns a version for resources.
+   *
+   * @param resourceNames list of resourceNames requested an empty list means all resources.
+   * @return version for the resources received
+   */
+  String version(List<String> resourceNames);
+
+  /**
+   * Returns a version for all resources.
+   *
+   * @return version for all the resources in the {@link SnapshotCache}.
+   */
+  default String version() {
+    return version(Collections.emptyList());
+  }
+}

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

@@ -100,7 +100,7 @@ public Watch createWatch(
       status.setLastWatchRequestTime(System.currentTimeMillis());
 
       Snapshot snapshot = snapshots.get(group);
-      String version = snapshot == null ? "" : snapshot.version(request.getTypeUrl());
+      String version = snapshot == null ? "" : snapshot.version(request.getTypeUrl(), request.getResourceNamesList());
 
       Watch watch = new Watch(ads, request, responseConsumer);
 
@@ -213,7 +213,7 @@ public void setSnapshot(T group, Snapshot snapshot) {
       }
 
       status.watchesRemoveIf((id, watch) -> {
-        String version = snapshot.version(watch.request().getTypeUrl());
+        String version = snapshot.version(watch.request().getTypeUrl(), watch.request().getResourceNamesList());
 
         if (!watch.request().getVersionInfo().equals(version)) {
           LOGGER.info("responding to open watch {}[{}] with new version {}",
@@ -273,15 +273,15 @@ private boolean respond(Watch watch, Snapshot snapshot, T group) {
             "not responding in ADS mode for {} from node {} at version {} for request [{}] since [{}] not in snapshot",
             watch.request().getTypeUrl(),
             group,
-            snapshot.version(watch.request().getTypeUrl()),
+            snapshot.version(watch.request().getTypeUrl(), watch.request().getResourceNamesList()),
             String.join(", ", watch.request().getResourceNamesList()),
             String.join(", ", missingNames));
 
         return false;
       }
     }
 
-    String version = snapshot.version(watch.request().getTypeUrl());
+    String version = snapshot.version(watch.request().getTypeUrl(), watch.request().getResourceNamesList());
 
     LOGGER.info("responding for {} from node {} at version {} with version {}",
         watch.request().getTypeUrl(),

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

@@ -16,6 +16,7 @@
 import io.envoyproxy.envoy.api.v2.RouteConfiguration;
 import io.envoyproxy.envoy.api.v2.auth.Secret;
 import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 
@@ -84,6 +85,40 @@ public static Snapshot create(
         SnapshotResources.create(secrets, secretsVersion));
   }
 
+  /**
+   * Returns a new {@link Snapshot} instance that has separate versions for each resource type.
+   *
+   * @param clusters the cluster resources in this snapshot
+   * @param clusterVersionResolver version resolver of the clusters in this snapshot
+   * @param endpoints the endpoint resources in this snapshot
+   * @param endpointVersionResolver version resolver of the endpoints in this snapshot
+   * @param listeners the listener resources in this snapshot
+   * @param listenerVersionResolver version resolver of listeners in this snapshot
+   * @param routes the route resources in this snapshot
+   * @param routeVersionResolver version resolver of the routes in this snapshot
+   * @param secrets the secret resources in this snapshot
+   * @param secretVersionResolver version resolver of the secrets in this snapshot
+   */
+  public static Snapshot create(
+      Iterable<Cluster> clusters,
+      ResourceVersionResolver clusterVersionResolver,
+      Iterable<ClusterLoadAssignment> endpoints,
+      ResourceVersionResolver endpointVersionResolver,
+      Iterable<Listener> listeners,
+      ResourceVersionResolver listenerVersionResolver,
+      Iterable<RouteConfiguration> routes,
+      ResourceVersionResolver routeVersionResolver,
+      Iterable<Secret> secrets,
+      ResourceVersionResolver secretVersionResolver) {
+
+    return new AutoValue_Snapshot(
+        SnapshotResources.create(clusters, clusterVersionResolver),
+        SnapshotResources.create(endpoints, endpointVersionResolver),
+        SnapshotResources.create(listeners, listenerVersionResolver),
+        SnapshotResources.create(routes, routeVersionResolver),
+        SnapshotResources.create(secrets, secretVersionResolver));
+  }
+
   /**
    * Creates an empty snapshot with the given version.
    *
@@ -170,21 +205,32 @@ public void ensureConsistent() throws SnapshotConsistencyException {
    * @param typeUrl the URL for the requested resource type
    */
   public String version(String typeUrl) {
+    return version(typeUrl, Collections.emptyList());
+  }
+
+  /**
+   * Returns the version in this snapshot for the given resource type.
+   *
+   * @param typeUrl the URL for the requested resource type
+   * @param resourceNames list of requested resource names,
+   *                      used to calculate a version for the given resources
+   */
+  public String version(String typeUrl, List<String> resourceNames) {
     if (Strings.isNullOrEmpty(typeUrl)) {
       return "";
     }
 
     switch (typeUrl) {
       case CLUSTER_TYPE_URL:
-        return clusters().version();
+        return clusters().version(resourceNames);
       case ENDPOINT_TYPE_URL:
-        return endpoints().version();
+        return endpoints().version(resourceNames);
       case LISTENER_TYPE_URL:
-        return listeners().version();
+        return listeners().version(resourceNames);
       case ROUTE_TYPE_URL:
-        return routes().version();
+        return routes().version(resourceNames);
       case SECRET_TYPE_URL:
-        return secrets().version();
+        return secrets().version(resourceNames);
       default:
         return "";
     }

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

@@ -3,6 +3,8 @@
 import com.google.auto.value.AutoValue;
 import com.google.common.collect.ImmutableMap;
 import com.google.protobuf.Message;
+
+import java.util.List;
 import java.util.Map;
 import java.util.stream.Collector;
 import java.util.stream.StreamSupport;
@@ -19,14 +21,34 @@ public abstract class SnapshotResources<T extends Message> {
    */
   public static <T extends Message> SnapshotResources<T> create(Iterable<T> resources, String version) {
     return new AutoValue_SnapshotResources<>(
-        StreamSupport.stream(resources.spliterator(), false)
-            .collect(
-                Collector.of(
-                    ImmutableMap.Builder<String, T>::new,
-                    (b, e) -> b.put(Resources.getResourceName(e), e),
-                    (b1, b2) -> b1.putAll(b2.build()),
-                    ImmutableMap.Builder::build)),
-        version);
+        resourcesMap(resources),
+        (r) -> version
+    );
+  }
+
+  /**
+   * Returns a new {@link SnapshotResources} instance with versions by resource name.
+   *
+   * @param resources the resources in this collection
+   * @param versionResolver version resolver for the resources in this collection
+   * @param <T> the type of resources in this collection
+   */
+  public static <T extends Message> SnapshotResources<T> create(
+      Iterable<T> resources,
+      ResourceVersionResolver versionResolver) {
+    return new AutoValue_SnapshotResources<>(
+        resourcesMap(resources),
+        versionResolver);
+  }
+
+  private static <T extends Message> ImmutableMap<String, T> resourcesMap(Iterable<T> resources) {
+    return StreamSupport.stream(resources.spliterator(), false)
+        .collect(
+            Collector.of(
+                ImmutableMap.Builder<String, T>::new,
+                (b, e) -> b.put(Resources.getResourceName(e), e),
+                (b1, b2) -> b1.putAll(b2.build()),
+                ImmutableMap.Builder::build));
   }
 
   /**
@@ -35,7 +57,24 @@ public static <T extends Message> SnapshotResources<T> create(Iterable<T> resour
   public abstract Map<String, T> resources();
 
   /**
-   * Returns the version associated with this resources in this collection.
+   * Returns the version associated with this all resources in this collection.
    */
-  public abstract String version();
+  public String version() {
+    return resourceVersionResolver().version();
+  }
+
+  /**
+   * Returns the version associated with the requested resources in this collection.
+   *
+   * @param resourceNames list of list of requested resources.
+   */
+  public String version(List<String> resourceNames) {
+    return resourceVersionResolver().version(resourceNames);
+  }
+
+  /**
+   * Returns the version resolver associated with this resources in this collection.
+   */
+  public abstract ResourceVersionResolver resourceVersionResolver();
+
 }

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

@@ -3,7 +3,10 @@
 import static org.assertj.core.api.Assertions.assertThat;
 
 import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+
 import io.envoyproxy.envoy.api.v2.Cluster;
+import java.util.Map;
 import java.util.UUID;
 import org.junit.Test;
 
@@ -28,4 +31,36 @@ public void createBuildsResourcesMapWithNameAndPopulatesVersion() {
 
     assertThat(snapshot.version()).isEqualTo(version);
   }
+
+  @Test
+  public void populatesVersionWithSeparateVersionPerCluster() {
+    final String aggregateVersion = UUID.randomUUID().toString();
+
+    final Map<String, String> versions = ImmutableMap.of(
+        CLUSTER0_NAME, UUID.randomUUID().toString(),
+        CLUSTER1_NAME, UUID.randomUUID().toString()
+    );
+
+    SnapshotResources<Cluster> snapshot = SnapshotResources.create(
+        ImmutableList.of(CLUSTER0, CLUSTER1), resourceNames -> {
+          if (resourceNames.size() != 1 || !versions.containsKey(resourceNames.get(0))) {
+            return aggregateVersion;
+          }
+          return versions.get(resourceNames.get(0));
+        }
+    );
+
+    // when no resource name provided, the aggregated version should be returned
+    assertThat(snapshot.version()).isEqualTo(aggregateVersion);
+
+    // when one resource name is provided, the cluster version should be returned
+    assertThat(snapshot.version(ImmutableList.of(CLUSTER0_NAME))).isEqualTo(versions.get(CLUSTER0_NAME));
+    assertThat(snapshot.version(ImmutableList.of(CLUSTER1_NAME))).isEqualTo(versions.get(CLUSTER1_NAME));
+
+    // when an unknown resource name is provided, the aggregated version should be returned
+    assertThat(snapshot.version(ImmutableList.of("unknown_cluster_name"))).isEqualTo(aggregateVersion);
+
+    // when multiple resource names are provided, the aggregated version should be returned
+    assertThat(snapshot.version(ImmutableList.of(CLUSTER1_NAME, CLUSTER1_NAME))).isEqualTo(aggregateVersion);
+  }
 }