feat(cats): Dynomite-friendly ClusteredSortAgentScheduler (#2393)

build.gradle

@@ -39,7 +39,7 @@ allprojects {
   apply plugin: 'groovy'
 
   ext {
-    spinnakerDependenciesVersion = project.hasProperty('spinnakerDependenciesVersion') ? project.property('spinnakerDependenciesVersion') : '0.142.2'
+    spinnakerDependenciesVersion = project.hasProperty('spinnakerDependenciesVersion') ? project.property('spinnakerDependenciesVersion') : '0.145.0'
   }
 
   def checkLocalVersions = [spinnakerDependenciesVersion: spinnakerDependenciesVersion]

cats/cats-core/src/main/java/com/netflix/spinnaker/cats/agent/CachingAgent.java

@@ -68,6 +68,7 @@ public CacheExecution(ProviderRegistry providerRegistry) {
       this.providerRegistry = providerRegistry;
     }
 
+    @Override
     public void executeAgent(Agent agent) {
       storeAgentResult(agent, executeAgentWithoutStore(agent));
     }

cats/cats-dynomite/src/main/java/com/netflix/spinnaker/cats/dynomite/DynomiteUtils.java

@@ -0,0 +1,40 @@
+/*
+ * Copyright 2018 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.netflix.spinnaker.cats.dynomite;
+
+import com.netflix.dyno.connectionpool.exception.DynoException;
+import com.netflix.spinnaker.kork.dynomite.DynomiteClientDelegate;
+import net.jodah.failsafe.RetryPolicy;
+import redis.clients.jedis.exceptions.JedisException;
+
+import java.util.Arrays;
+import java.util.concurrent.TimeUnit;
+
+public class DynomiteUtils {
+
+  private DynomiteUtils() {}
+
+  public static RetryPolicy greedyRetryPolicy(long delayMs) {
+    return new RetryPolicy()
+      .retryOn(Arrays.asList(
+        JedisException.class,
+        DynoException.class,
+        DynomiteClientDelegate.ClientDelegateException.class
+      ))
+      .withDelay(delayMs, TimeUnit.MILLISECONDS)
+      .withMaxRetries(3);
+  }
+}

cats/cats-dynomite/src/main/java/com/netflix/spinnaker/cats/dynomite/ExcessiveDynoFailureRetries.java

@@ -0,0 +1,22 @@
+/*
+ * Copyright 2018 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.netflix.spinnaker.cats.dynomite;
+
+public class ExcessiveDynoFailureRetries extends RuntimeException {
+  public ExcessiveDynoFailureRetries(String message, Throwable cause) {
+    super(message, cause);
+  }
+}

cats/cats-dynomite/src/main/java/com/netflix/spinnaker/cats/dynomite/cache/DynomiteCache.java

@@ -25,10 +25,11 @@
 import com.netflix.spinnaker.cats.cache.DefaultCacheData;
 import com.netflix.spinnaker.cats.compression.CompressionStrategy;
 import com.netflix.spinnaker.cats.compression.NoopCompression;
+import com.netflix.spinnaker.cats.dynomite.DynomiteUtils;
+import com.netflix.spinnaker.cats.dynomite.ExcessiveDynoFailureRetries;
 import com.netflix.spinnaker.cats.redis.cache.AbstractRedisCache;
 import com.netflix.spinnaker.cats.redis.cache.RedisCacheOptions;
 import com.netflix.spinnaker.kork.dynomite.DynomiteClientDelegate;
-import com.netflix.spinnaker.kork.dynomite.DynomiteClientDelegate.ClientDelegateException;
 import net.jodah.failsafe.Failsafe;
 import net.jodah.failsafe.RetryPolicy;
 import org.slf4j.Logger;
@@ -38,18 +39,9 @@
 
 import java.io.IOException;
 import java.time.Duration;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.LinkedHashSet;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 import java.util.Map.Entry;
-import java.util.Optional;
-import java.util.Set;
 import java.util.concurrent.ThreadLocalRandom;
-import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 
@@ -95,10 +87,7 @@ class NOOP implements CacheMetrics {}
   private final Logger log = LoggerFactory.getLogger(getClass());
 
   // TODO rz - Make retry policy configurable
-  private static final RetryPolicy REDIS_RETRY_POLICY = new RetryPolicy()
-    .retryOn(Arrays.asList(JedisException.class, DynoException.class, ClientDelegateException.class))
-    .withDelay(500, TimeUnit.MILLISECONDS)
-    .withMaxRetries(3);
+  private static final RetryPolicy REDIS_RETRY_POLICY = DynomiteUtils.greedyRetryPolicy(500);
 
   private final CacheMetrics cacheMetrics;
 
@@ -467,10 +456,4 @@ protected String allOfTypeId(String type) {
   protected String allOfTypeReindex(String type) {
     return format("{%s:%s}:members.2", prefix, type);
   }
-
-  private static class ExcessiveDynoFailureRetries extends RuntimeException {
-    ExcessiveDynoFailureRetries(String message, Throwable cause) {
-      super(message, cause);
-    }
-  }
 }