fix(core): Optimize CleanupPendingOnDemandCachesAgent (#2508)

Switched to using `exists` vs `get` with smaller partitions.
spinnaker · Apr 13, 2018 · e9fc0a8 · e9fc0a8
1 parent d82edae
commit e9fc0a8
Showing 1 changed file with 44 additions and 14 deletions.
diff --git a/...roovy/com/netflix/spinnaker/clouddriver/core/agent/CleanupPendingOnDemandCachesAgent.java b/...roovy/com/netflix/spinnaker/clouddriver/core/agent/CleanupPendingOnDemandCachesAgent.java
@@ -16,6 +16,8 @@
 
 package com.netflix.spinnaker.clouddriver.core.agent;
 
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
 import com.netflix.spinnaker.cats.agent.RunnableAgent;
 import com.netflix.spinnaker.cats.module.CatsModule;
 import com.netflix.spinnaker.cats.provider.Provider;
@@ -26,12 +28,16 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.context.ApplicationContext;
+import redis.clients.jedis.Response;
 import redis.clients.jedis.ScanParams;
 import redis.clients.jedis.ScanResult;
 
+import java.util.ArrayList;
 import java.util.Collection;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
@@ -88,33 +94,43 @@ void run(Collection<Provider> providers) {
        .filter(s -> !s.equals("_ALL_"))
        .collect(Collectors.toList());
 
-     List existingOnDemandKeys;
+     Map<String, Response<Boolean>> existingOnDemandKeys = new HashMap<>();
      if (redisClientDelegate.supportsMultiKeyPipelines()) {
-       existingOnDemandKeys = redisClientDelegate.withMultiKeyPipeline(pipeline -> {
-         onDemandKeys.forEach(k -> pipeline.get(provider.getProviderName() + ":onDemand:attributes:" + k));
-         return pipeline.syncAndReturnAll();
+       redisClientDelegate.withMultiKeyPipeline(pipeline -> {
+         for (List<String> partition : Iterables.partition(onDemandKeys, redisCacheOptions.getMaxDelSize())) {
+           for (String id : partition) {
+             existingOnDemandKeys.put(id, pipeline.exists(provider.getProviderName() + ":onDemand:attributes:" + id));
+           }
+         }
+         pipeline.sync();
        });
      } else {
-       existingOnDemandKeys = redisClientDelegate.withCommandsClient(client -> {
-         return onDemandKeys.stream()
-           .map(k -> client.get(provider.getProviderName() + "onDemand:attributes:" + k))
-           .collect(Collectors.toList());
+       redisClientDelegate.withCommandsClient(client -> {
+         onDemandKeys.stream()
+           .filter(k -> client.exists(provider.getProviderName() + "onDemand:attributes:" + k))
+           .forEach(k -> existingOnDemandKeys.put(k, new StaticResponse(Boolean.TRUE)));
        });
      }
 
-     Set<String> onDemandKeysToRemove = new HashSet<>();
-     for (int i = 0; i < onDemandKeys.size(); i++) {
-       if (existingOnDemandKeys.get(i) == null) {
-         onDemandKeysToRemove.add(onDemandKeys.get(i));
+     List<String> onDemandKeysToRemove = new ArrayList<>();
+     for (String onDemandKey : onDemandKeys) {
+       if (!existingOnDemandKeys.containsKey(onDemandKey) || !existingOnDemandKeys.get(onDemandKey).get()) {
+         onDemandKeysToRemove.add(onDemandKey);
        }
      }
 
      if (!onDemandKeysToRemove.isEmpty()) {
        log.info("Removing {} from {}", onDemandKeysToRemove.size(), onDemandSetName);
        log.debug("Removing {} from {}", onDemandKeysToRemove, onDemandSetName);
 
-       redisClientDelegate.withCommandsClient(client -> {
-         client.srem(onDemandSetName, onDemandKeysToRemove.toArray(new String[onDemandKeysToRemove.size()]));
+
+       redisClientDelegate.withMultiKeyPipeline(pipeline -> {
+         for (List<String> idPartition : Lists.partition(onDemandKeysToRemove, redisCacheOptions.getMaxDelSize())) {
+           String[] ids = idPartition.toArray(new String[idPartition.size()]);
+           pipeline.srem(onDemandSetName, ids);
+         }
+
+         pipeline.sync();
        });
      }
     });
@@ -147,4 +163,18 @@ private Set<String> scanMembers(String setKey) {
   private CatsModule getCatsModule() {
     return applicationContext.getBean(CatsModule.class);
   }
+
+  private static class StaticResponse extends Response<Boolean> {
+    private final Boolean value;
+
+    StaticResponse(Boolean value) {
+      super(null);
+      this.value = value;
+    }
+
+    @Override
+    public Boolean get() {
+      return value;
+    }
+  }
 }