fix(core): Support the eviction of stale cache key identifiers

This PR attempts to clean up situations where the `:members` set for a
given cached type contains keys that no longer exist.

In particular, a `CachingAgent` can optionally return a pattern glob for
each authoritative type.

Any `:members` key matching the pattern glob BUT not included in the
most recent caching cycle will be evicted.

This has been enabled on a select few `aws` caching agents.

cats/cats-core/cats-core.gradle

@@ -1,3 +1,5 @@
 dependencies {
+  compile spinnaker.dependency('slf4jApi')
+
   testCompile project(":cats:cats-test")
 }

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

@@ -16,15 +16,23 @@
 
 package com.netflix.spinnaker.cats.agent;
 
+import com.netflix.spinnaker.cats.cache.CacheData;
 import com.netflix.spinnaker.cats.provider.ProviderCache;
 import com.netflix.spinnaker.cats.provider.ProviderRegistry;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashSet;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
 
 /**
  * A CachingAgent loads one or more types of data.
- *
+ * <p>
  * The data set for a caching agent is scoped to the provider and agent type. For example
  * an agent might load clusters for the AWS provider, and be scoped to a particular account
  * and region.
@@ -44,11 +52,16 @@ public interface CachingAgent extends Agent {
    */
   CacheResult loadData(ProviderCache providerCache);
 
+  default Optional<Map<String, String>> getCacheKeyPatterns() {
+    return Optional.empty();
+  }
+
   default AgentExecution getAgentExecution(ProviderRegistry providerRegistry) {
     return new CacheExecution(providerRegistry);
   }
 
   class CacheExecution implements AgentExecution {
+    private final Logger log = LoggerFactory.getLogger(CacheExecution.class);
     private final ProviderRegistry providerRegistry;
 
     public CacheExecution(ProviderRegistry providerRegistry) {
@@ -77,6 +90,37 @@ public void storeAgentResult(Agent agent, CacheResult result) {
         }
       }
 
+
+      Optional<Map<String, String>> cacheKeyPatterns = cachingAgent.getCacheKeyPatterns();
+      if (cacheKeyPatterns.isPresent()) {
+        for (String type : authoritative) {
+          String cacheKeyPatternForType = cacheKeyPatterns.get().get(type);
+          if (cacheKeyPatternForType != null) {
+            try {
+              Set<String> cachedIdentifiersForType = result.getCacheResults().get(type)
+                .stream()
+                .map(CacheData::getId)
+                .collect(Collectors.toSet());
+
+              Collection<String> evictableIdentifiers = cache.filterIdentifiers(type, cacheKeyPatternForType)
+                .stream()
+                .filter(i -> !cachedIdentifiersForType.contains(i))
+                .collect(Collectors.toSet());
+
+              // any key that existed previously but was not re-cached by this agent is considered evictable
+              if (!evictableIdentifiers.isEmpty()) {
+                Collection<String> evictionsForType = result.getEvictions().computeIfAbsent(type, evictableKeys -> new ArrayList<>());
+                evictionsForType.addAll(evictableIdentifiers);
+
+                log.info("Evicting stale identifiers: {}", evictableIdentifiers);
+              }
+            } catch (Exception e) {
+              log.error("Failed to check for stale identifiers (type: {}, pattern: {}, agent: {})", type, cacheKeyPatternForType, agent, e);
+            }
+          }
+        }
+      }
+
       cache.putCacheResult(agent.getAgentType(), authoritative, result);
     }
   }

cats/cats-core/src/test/groovy/com/netflix/spinnaker/cats/agent/CacheExecutionSpec.groovy

@@ -0,0 +1,85 @@
+/*
+ * Copyright 2017 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.agent
+
+import com.netflix.spinnaker.cats.cache.CacheData
+import com.netflix.spinnaker.cats.cache.DefaultCacheData
+import com.netflix.spinnaker.cats.provider.ProviderCache
+import com.netflix.spinnaker.cats.provider.ProviderRegistry
+import spock.lang.Specification
+
+import static com.netflix.spinnaker.cats.agent.AgentDataType.Authority.AUTHORITATIVE;
+
+class CacheExecutionSpec extends Specification {
+  def providerRegistry = Mock(ProviderRegistry)
+  def cachingAgent = Mock(CachingAgent)
+  def providerCache = Mock(ProviderCache)
+
+  void "should evict keys that were NOT cached by the responsible agent"() {
+    given:
+    def cacheExecution = new CachingAgent.CacheExecution(providerRegistry)
+    def result = new DefaultCacheResult([
+      "securityGroups": [new DefaultCacheData("securityGroups:foo:test:us-west-1", [:], [:])]
+    ], [:])
+
+    when:
+    cacheExecution.storeAgentResult(cachingAgent, result)
+
+    then:
+    1 * cachingAgent.getProvidedDataTypes() >> {
+      return [
+        AUTHORITATIVE.forType("securityGroups")
+      ]
+    }
+    1 * cachingAgent.getCacheKeyPatterns() >> {
+      return [
+        "securityGroups": "securityGroups:*:test:us-west-1"
+      ]
+    }
+    1 * providerCache.filterIdentifiers("securityGroups", "securityGroups:*:test:us-west-1") >> {
+      return [
+        "securityGroups:foo:test:us-west-1",
+        "securityGroups:bar:test:us-west-1"
+      ]
+    }
+    1 * providerRegistry.getProviderCache(_) >> { return providerCache }
+
+    result.evictions["securityGroups"] == ["securityGroups:bar:test:us-west-1"]
+  }
+
+  void "should skip stale keys check if agent supplies no cache key patterns"() {
+    given:
+    def cacheExecution = new CachingAgent.CacheExecution(providerRegistry)
+    def result = new DefaultCacheResult([
+      "securityGroups": [new DefaultCacheData("securityGroups:foo:test:us-west-1", [:], [:])]
+    ], [:])
+
+    when:
+    cacheExecution.storeAgentResult(cachingAgent, result)
+
+    then:
+    1 * cachingAgent.getProvidedDataTypes() >> {
+      return [
+        AUTHORITATIVE.forType("securityGroups")
+      ]
+    }
+    1 * cachingAgent.getCacheKeyPatterns() >> { return Optional.empty() }
+    1 * providerRegistry.getProviderCache(_) >> { return providerCache }
+
+    result.evictions.isEmpty()
+  }
+}

clouddriver-aws/src/main/groovy/com/netflix/spinnaker/clouddriver/aws/data/Keys.groovy

@@ -144,7 +144,11 @@ class Keys implements KeyParser {
 
   static String getServerGroupKey(String autoScalingGroupName, String account, String region) {
     Names names = Names.parseName(autoScalingGroupName)
-    "${ID}:${Namespace.SERVER_GROUPS}:${names.cluster}:${account}:${region}:${names.group}"
+    return getServerGroupKey(names.cluster, names.group, account, region)
+  }
+
+  static String getServerGroupKey(String cluster, String autoScalingGroupName, String account, String region) {
+    "${ID}:${Namespace.SERVER_GROUPS}:${cluster}:${account}:${region}:${autoScalingGroupName}"
   }
 
   static String getInstanceKey(String instanceId, String account, String region) {

clouddriver-aws/src/main/groovy/com/netflix/spinnaker/clouddriver/aws/provider/agent/AmazonLoadBalancerCachingAgent.groovy

@@ -47,6 +47,13 @@ class AmazonLoadBalancerCachingAgent extends AbstractAmazonLoadBalancerCachingAg
     super(amazonCloudProvider, amazonClientProvider, account, region, objectMapper, registry)
   }
 
+  @Override
+  Optional<Map<String, String>> getCacheKeyPatterns() {
+    return [
+      (LOAD_BALANCERS.ns): Keys.getLoadBalancerKey('*', account.name, region, '*', null)
+    ]
+  }
+
   @Override
   OnDemandAgent.OnDemandResult handle(ProviderCache providerCache, Map<String, ? extends Object> data) {
     if (!data.containsKey("loadBalancerName")) {

clouddriver-aws/src/main/groovy/com/netflix/spinnaker/clouddriver/aws/provider/agent/AmazonSecurityGroupCachingAgent.groovy

@@ -154,6 +154,15 @@ class AmazonSecurityGroupCachingAgent implements CachingAgent, OnDemandAgent, Ac
     buildCacheResult(providerCache, securityGroups, evictions, null)
   }
 
+  @Override
+  Optional<Map<String, String>> getCacheKeyPatterns() {
+    return Optional.of(
+      Collections.singletonMap(
+        SECURITY_GROUPS.ns, Keys.getSecurityGroupKey('*', '*', region, account.name, '*')
+      )
+    )
+  }
+
   @Override
   Collection<Map> pendingOnDemandRequests(ProviderCache providerCache) {
     return []

clouddriver-aws/src/main/groovy/com/netflix/spinnaker/clouddriver/aws/provider/agent/ClusterCachingAgent.groovy

@@ -128,6 +128,13 @@ class ClusterCachingAgent implements CachingAgent, OnDemandAgent, AccountAware,
     types
   }
 
+  @Override
+  Optional<Map<String, String>> getCacheKeyPatterns() {
+    return [
+      (SERVER_GROUPS.ns): Keys.getServerGroupKey('*', '*', account.name, region)
+    ]
+  }
+
   static class AmazonClients {
     final AmazonAutoScaling autoScaling
     final AmazonEC2 amazonEC2