scheduled cleanup of obsolete resources, improved get

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

Author: csviri

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ExecutorServiceManager.java

@@ -23,6 +23,7 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.function.Function;
@@ -39,7 +40,7 @@ public class ExecutorServiceManager {
   private static final Logger log = LoggerFactory.getLogger(ExecutorServiceManager.class);
   private ExecutorService executor;
   private ExecutorService workflowExecutor;
-  private ExecutorService cachingExecutorService;
+  private ScheduledExecutorService cachingExecutorService;
   private boolean started;
   private ConfigurationService configurationService;
 
@@ -122,14 +123,14 @@ private synchronized void lazyInitWorkflowExecutorService() {
     }
   }
 
-  public ExecutorService cachingExecutorService() {
+  public ScheduledExecutorService cachingExecutorService() {
     return cachingExecutorService;
   }
 
   public void start(ConfigurationService configurationService) {
     if (!started) {
       this.configurationService = configurationService; // used to lazy init workflow executor
-      this.cachingExecutorService = Executors.newCachedThreadPool();
+      this.cachingExecutorService = Executors.newScheduledThreadPool(0);
       this.executor = new InstrumentedExecutorService(configurationService.getExecutorService());
       started = true;
     }

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

@@ -153,7 +153,14 @@ public synchronized void start() {
     if (isRunning()) {
       return;
     }
-    temporaryResourceCache = new TemporaryResourceCache<>(comparableResourceVersions, this);
+    temporaryResourceCache =
+        new TemporaryResourceCache<>(
+            comparableResourceVersions,
+            controllerConfiguration
+                .getConfigurationService()
+                .getExecutorServiceManager()
+                .cachingExecutorService(),
+            this);
     this.cache = new InformerManager<>(client, configuration, this);
     cache.setControllerConfiguration(controllerConfiguration);
     cache.addIndexers(indexers);
@@ -192,9 +199,10 @@ public Optional<R> get(ResourceID resourceID) {
     var res = cache.get(resourceID);
     if (comparableResourceVersions
         && resource.isPresent()
-        && res.filter(
-                r -> ReconcilerUtilsInternal.compareResourceVersions(r, resource.orElseThrow()) > 0)
-            .isEmpty()) {
+        && ReconcilerUtilsInternal.compareResourceVersions(
+                resource.get().getMetadata().getResourceVersion(),
+                manager().lastSyncResourceVersion(resource.get().getMetadata().getNamespace()))
+            > 0) {
       log.debug("Latest resource found in temporary cache for Resource ID: {}", resourceID);
       return resource;
     }

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

@@ -19,6 +19,8 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -72,20 +74,30 @@ public enum EventHandling {
 
   public TemporaryResourceCache(
       boolean comparableResourceVersions,
+      ScheduledExecutorService obsoleteCheckExecutor,
       ManagedInformerEventSource<T, ?, ?> managedInformerEventSource) {
     this(
         comparableResourceVersions,
         DEFAULT_OBSOLETE_RESOURCE_CHECK_INTERVAL,
+        obsoleteCheckExecutor,
         managedInformerEventSource);
   }
 
   TemporaryResourceCache(
       boolean comparableResourceVersions,
       long obsoleteResourceCheckInterval,
+      ScheduledExecutorService obsoleteCheckExecutor,
       ManagedInformerEventSource<T, ?, ?> managedInformerEventSource) {
     this.comparableResourceVersions = comparableResourceVersions;
     this.obsoleteResourceCheckInterval = obsoleteResourceCheckInterval;
     this.managedInformerEventSource = managedInformerEventSource;
+    if (comparableResourceVersions) {
+      obsoleteCheckExecutor.scheduleWithFixedDelay(
+          this::checkObsoleteResources,
+          obsoleteResourceCheckInterval,
+          obsoleteResourceCheckInterval,
+          TimeUnit.MILLISECONDS);
+    }
   }
 
   public synchronized void startEventFilteringModify(ResourceID resourceID) {
@@ -161,7 +173,6 @@ 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);
@@ -227,7 +238,7 @@ private String getLatestResourceVersion(String namespace) {
     return managedInformerEventSource.manager().lastSyncResourceVersion(namespace);
   }
 
-  void checkObsoleteResources() {
+  private void checkObsoleteResources() {
     if (System.currentTimeMillis() > lastObsoleteResourceCheck + obsoleteResourceCheckInterval) {
       lastObsoleteResourceCheck = System.currentTimeMillis();
       log.debug("Checking for obsolete resources.");

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

@@ -19,6 +19,7 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ScheduledExecutorService;
 
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -381,7 +382,8 @@ private void withRealTemporaryResourceCache() {
     when(mes.manager()).thenReturn(mim);
     when(mim.lastSyncResourceVersion(any())).thenReturn("1");
 
-    temporaryResourceCache = spy(new TemporaryResourceCache<>(true, mes));
+    temporaryResourceCache =
+        spy(new TemporaryResourceCache<>(true, mock(ScheduledExecutorService.class), mes));
     informerEventSource.setTemporalResourceCache(temporaryResourceCache);
   }
 

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

@@ -16,6 +16,7 @@
 package io.javaoperatorsdk.operator.processing.event.source.informer;
 
 import java.util.Map;
+import java.util.concurrent.ScheduledExecutorService;
 
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -46,8 +47,8 @@ void setup() {
     var mim = mock(InformerManager.class);
     when(mes.manager()).thenReturn(mim);
     when(mim.lastSyncResourceVersion(any())).then(a -> latestSyncVersion);
-
-    temporaryResourceCache = new TemporaryResourceCache<>(true, mes);
+    temporaryResourceCache =
+        new TemporaryResourceCache<>(true, mock(ScheduledExecutorService.class), mes);
   }
 
   @Test
@@ -124,7 +125,8 @@ void removesResourceFromCache() {
   @Test
   void nonComparableResourceVersionsDisables() {
     this.temporaryResourceCache =
-        new TemporaryResourceCache<>(false, mock(ManagedInformerEventSource.class));
+        new TemporaryResourceCache<>(
+            false, mock(ScheduledExecutorService.class), mock(ManagedInformerEventSource.class));
 
     this.temporaryResourceCache.putResource(testResource());