Obsolete resource handling for read-cache-after-write

Signed-off-by: Attila Mészáros <a_meszaros@apple.com>

Author: csviri

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/ManagedInformerEventSource.java

@@ -153,7 +153,7 @@ public synchronized void start() {
     if (isRunning()) {
       return;
     }
-    temporaryResourceCache = new TemporaryResourceCache<>(comparableResourceVersions);
+    temporaryResourceCache = new TemporaryResourceCache<>(comparableResourceVersions, this);
     this.cache = new InformerManager<>(client, configuration, this);
     cache.setControllerConfiguration(controllerConfiguration);
     cache.addIndexers(indexers);

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/TemporaryResourceCache.java

@@ -54,21 +54,39 @@
 public class TemporaryResourceCache<T extends HasMetadata> {
 
   private static final Logger log = LoggerFactory.getLogger(TemporaryResourceCache.class);
+  public static final long DEFAULT_OBSOLETE_RESOURCE_CHECK_INTERVAL = 1000 * 60 * 3L;
 
   private final Map<ResourceID, T> cache = new ConcurrentHashMap<>();
+  private final Map<ResourceID, EventFilterDetails> activeUpdates = new HashMap<>();
   private final boolean comparableResourceVersions;
   private String latestResourceVersion;
 
-  private final Map<ResourceID, EventFilterDetails> activeUpdates = new HashMap<>();
+  private final long obsoleteResourceCheckInterval;
+  private volatile long lastObsoleteResourceCheck = System.currentTimeMillis();
+  private ManagedInformerEventSource<T, ?, ?> managedInformerEventSource;
 
   public enum EventHandling {
     DEFER,
     OBSOLETE,
     NEW
   }
 
-  public TemporaryResourceCache(boolean comparableResourceVersions) {
+  public TemporaryResourceCache(
+      boolean comparableResourceVersions,
+      ManagedInformerEventSource<T, ?, ?> managedInformerEventSource) {
+    this(
+        comparableResourceVersions,
+        DEFAULT_OBSOLETE_RESOURCE_CHECK_INTERVAL,
+        managedInformerEventSource);
+  }
+
+  TemporaryResourceCache(
+      boolean comparableResourceVersions,
+      long obsoleteResourceCheckInterval,
+      ManagedInformerEventSource<T, ?, ?> managedInformerEventSource) {
     this.comparableResourceVersions = comparableResourceVersions;
+    this.obsoleteResourceCheckInterval = obsoleteResourceCheckInterval;
+    this.managedInformerEventSource = managedInformerEventSource;
   }
 
   public synchronized void startEventFilteringModify(ResourceID resourceID) {
@@ -148,6 +166,7 @@ private synchronized EventHandling onEvent(
         result = EventHandling.OBSOLETE;
       }
     }
+    checkObsoleteResources();
     var ed = activeUpdates.get(resourceId);
     if (ed != null && result != EventHandling.OBSOLETE) {
       log.debug("Setting last event for id: {} delete: {}", resourceId, delete);
@@ -208,6 +227,23 @@ public synchronized void putResource(T newResource) {
     }
   }
 
+  void checkObsoleteResources() {
+    if (System.currentTimeMillis() > lastObsoleteResourceCheck + obsoleteResourceCheckInterval) {
+      lastObsoleteResourceCheck = System.currentTimeMillis();
+      log.debug("Checking for obsolete resources.");
+      var iterator = cache.entrySet().iterator();
+      while (iterator.hasNext()) {
+        var e = iterator.next();
+        if (ReconcilerUtilsInternal.compareResourceVersions(
+                e.getValue().getMetadata().getResourceVersion(), latestResourceVersion)
+            < 0) iterator.remove();
+        // todo propagate event
+        managedInformerEventSource.handleEvent(ResourceAction.DELETED, e.getValue(), null, true);
+        log.debug("Removing obsolete resource with ID: {}", e.getKey());
+      }
+    }
+  }
+
   public synchronized Optional<T> getResourceFromCache(ResourceID resourceID) {
     return Optional.ofNullable(cache.get(resourceID));
   }

operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerEventSourceTest.java

@@ -376,7 +376,8 @@ private CountDownLatch sendForEventFilteringUpdate(Deployment deployment, int re
   }
 
   private void withRealTemporaryResourceCache() {
-    temporaryResourceCache = spy(new TemporaryResourceCache<>(true));
+    temporaryResourceCache =
+        spy(new TemporaryResourceCache<>(true, mock(ManagedInformerEventSource.class)));
     informerEventSource.setTemporalResourceCache(temporaryResourceCache);
   }
 

operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/TemporaryPrimaryResourceCacheTest.java

@@ -29,6 +29,7 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
 
 class TemporaryPrimaryResourceCacheTest {
 
@@ -38,7 +39,8 @@ class TemporaryPrimaryResourceCacheTest {
 
   @BeforeEach
   void setup() {
-    temporaryResourceCache = new TemporaryResourceCache<>(true);
+    temporaryResourceCache =
+        new TemporaryResourceCache<>(true, mock(ManagedInformerEventSource.class));
   }
 
   @Test
@@ -114,7 +116,8 @@ void removesResourceFromCache() {
 
   @Test
   void nonComparableResourceVersionsDisables() {
-    this.temporaryResourceCache = new TemporaryResourceCache<>(false);
+    this.temporaryResourceCache =
+        new TemporaryResourceCache<>(false, mock(ManagedInformerEventSource.class));
 
     this.temporaryResourceCache.putResource(testResource());
 
@@ -123,7 +126,7 @@ void nonComparableResourceVersionsDisables() {
   }
 
   @Test
-  void eventReceivedDuringFiltering() throws Exception {
+  void eventReceivedDuringFiltering() {
     var testResource = testResource();
 
     temporaryResourceCache.startEventFilteringModify(ResourceID.fromResource(testResource));