feat(core): Support for loadObjects (#545)

This PR allows a `StorageService` implementation to support bulk loading of objects. Default implementation raises an `UnsupportedOperationException` and triggers the legacy rxJava-backed loads.
spinnaker · Jun 17, 2019 · 1b48c2c · 1b48c2c
1 parent f33e188
commit 1b48c2c
Show file tree

Hide file tree

Showing 2 changed files with 74 additions and 47 deletions.
diff --git a/front50-core/src/main/groovy/com/netflix/spinnaker/front50/model/StorageService.java b/front50-core/src/main/groovy/com/netflix/spinnaker/front50/model/StorageService.java
@@ -19,6 +19,7 @@
 import com.netflix.spinnaker.front50.exception.NotFoundException;
 import java.time.Duration;
 import java.util.Collection;
+import java.util.List;
 import java.util.Map;
 
 public interface StorageService {
@@ -35,6 +36,11 @@ default boolean supportsEventing(ObjectType objectType) {
   <T extends Timestamped> T loadObject(ObjectType objectType, String objectKey)
       throws NotFoundException;
 
+  default <T extends Timestamped> List<T> loadObjects(
+      ObjectType objectType, List<String> objectKeys) {
+    throw new UnsupportedOperationException();
+  }
+
   void deleteObject(ObjectType objectType, String objectKey);
 
   default void bulkDeleteObjects(ObjectType objectType, Collection<String> objectKeys) {

diff --git a/front50-core/src/main/groovy/com/netflix/spinnaker/front50/model/StorageServiceSupport.java b/front50-core/src/main/groovy/com/netflix/spinnaker/front50/model/StorageServiceSupport.java
@@ -31,6 +31,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Function;
 import java.util.function.ToDoubleFunction;
 import java.util.stream.Collectors;
 import javax.annotation.PostConstruct;
@@ -354,59 +355,79 @@ private Set<T> fetchAllItems(Set<T> existingItems) {
       log.debug("Modified object keys: {}", value("keys", modifiedKeys));
     }
 
-    Observable.from(modifiedKeys)
-        .buffer(10)
-        .flatMap(
-            ids ->
-                Observable.from(ids)
-                    .flatMap(
-                        entry -> {
-                          try {
-                            String key = entry.getKey();
-                            T object = (T) service.loadObject(objectType, key);
-
-                            Long expectedLastModifiedTime = keyUpdateTime.get(key);
-                            Long currentLastModifiedTime = object.getLastModified();
+    try {
+      List<String> objectKeys =
+          modifiedKeys.stream().map(Map.Entry::getKey).collect(Collectors.toList());
+      List<T> objects = service.loadObjects(objectType, objectKeys);
+
+      Map<String, T> objectsById =
+          objects.stream()
+              .collect(Collectors.toMap(Timestamped::getId, Function.identity(), (o1, o2) -> o1));
+
+      for (String objectKey : objectKeys) {
+        if (objectsById.containsKey(objectKey)) {
+          resultMap.put(objectKey, objectsById.get(objectKey));
+        } else {
+          // equivalent to the NotFoundException handling in the exceptional case below
+          resultMap.remove(keyToId.get(objectKey));
+          numRemoved.getAndIncrement();
+        }
+      }
+    } catch (UnsupportedOperationException e) {
+      Observable.from(modifiedKeys)
+          .buffer(10)
+          .flatMap(
+              ids ->
+                  Observable.from(ids)
+                      .flatMap(
+                          entry -> {
+                            try {
+                              String key = entry.getKey();
+                              T object = (T) service.loadObject(objectType, key);
+
+                              Long expectedLastModifiedTime = keyUpdateTime.get(key);
+                              Long currentLastModifiedTime = object.getLastModified();
+
+                              if (expectedLastModifiedTime != null
+                                  && currentLastModifiedTime != null) {
+                                if (currentLastModifiedTime < expectedLastModifiedTime) {
+                                  log.warn(
+                                      "Unexpected stale read for {} (current: {}, expected: {})",
+                                      key,
+                                      new Date(currentLastModifiedTime),
+                                      new Date(expectedLastModifiedTime));
+                                }
+                              }
 
-                            if (expectedLastModifiedTime != null
-                                && currentLastModifiedTime != null) {
-                              if (currentLastModifiedTime < expectedLastModifiedTime) {
+                              if (!key.equals(buildObjectKey(object))) {
+                                mismatchedIdCounter.increment();
                                 log.warn(
-                                    "Unexpected stale read for {} (current: {}, expected: {})",
+                                    "{} '{}' has non-matching id '{}'",
+                                    objectType.group,
                                     key,
-                                    new Date(currentLastModifiedTime),
-                                    new Date(expectedLastModifiedTime));
+                                    buildObjectKey(object));
+                                // Should return Observable.empty() to skip caching, but will wait
+                                // until the
+                                // logging has been present for a release.
                               }
-                            }
 
-                            if (!key.equals(buildObjectKey(object))) {
-                              mismatchedIdCounter.increment();
-                              log.warn(
-                                  "{} '{}' has non-matching id '{}'",
-                                  objectType.group,
-                                  key,
-                                  buildObjectKey(object));
-                              // Should return Observable.empty() to skip caching, but will wait
-                              // until the
-                              // logging has been present for a release.
+                              return Observable.just(object);
+                            } catch (NotFoundException e2) {
+                              resultMap.remove(keyToId.get(entry.getKey()));
+                              numRemoved.getAndIncrement();
+                              return Observable.empty();
                             }
-
-                            return Observable.just(object);
-                          } catch (NotFoundException e) {
-                            resultMap.remove(keyToId.get(entry.getKey()));
-                            numRemoved.getAndIncrement();
-                            return Observable.empty();
-                          }
-                        })
-                    .subscribeOn(scheduler))
-        .subscribeOn(scheduler)
-        .toList()
-        .toBlocking()
-        .single()
-        .forEach(
-            item -> {
-              resultMap.put(buildObjectKey(item), item);
-            });
+                          })
+                      .subscribeOn(scheduler))
+          .subscribeOn(scheduler)
+          .toList()
+          .toBlocking()
+          .single()
+          .forEach(
+              item -> {
+                resultMap.put(buildObjectKey(item), item);
+              });
+    }
 
     Set<T> result = resultMap.values().stream().collect(Collectors.toSet());
     this.lastRefreshedTime.set(refreshTime);