latest resource version by informers

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,11 @@ public synchronized void start() {
     if (isRunning()) {
       return;
     }
-    temporaryResourceCache = new TemporaryResourceCache<>(comparableResourceVersions, this);
+    var watchesOnlyOneNamespace =
+        configuration.getInformerConfig().watchCurrentNamespace()
+            || configuration.getInformerConfig().watchAllNamespaces();
+    temporaryResourceCache =
+        new TemporaryResourceCache<>(comparableResourceVersions, !watchesOnlyOneNamespace, 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

@@ -59,7 +59,8 @@ public class TemporaryResourceCache<T extends HasMetadata> {
   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 ConcurrentHashMap<String, String> perNamespaceLatestResourceVersion;
+  private volatile String latestResourceVersion;
 
   private final long obsoleteResourceCheckInterval;
   private volatile long lastObsoleteResourceCheck = System.currentTimeMillis();
@@ -73,20 +74,28 @@ public enum EventHandling {
 
   public TemporaryResourceCache(
       boolean comparableResourceVersions,
+      boolean wrapsMultipleInformers,
       ManagedInformerEventSource<T, ?, ?> managedInformerEventSource) {
     this(
         comparableResourceVersions,
+        wrapsMultipleInformers,
         DEFAULT_OBSOLETE_RESOURCE_CHECK_INTERVAL,
         managedInformerEventSource);
   }
 
   TemporaryResourceCache(
       boolean comparableResourceVersions,
+      boolean wrapsMultipleInformers,
       long obsoleteResourceCheckInterval,
       ManagedInformerEventSource<T, ?, ?> managedInformerEventSource) {
     this.comparableResourceVersions = comparableResourceVersions;
     this.obsoleteResourceCheckInterval = obsoleteResourceCheckInterval;
     this.managedInformerEventSource = managedInformerEventSource;
+    if (wrapsMultipleInformers || !comparableResourceVersions) {
+      perNamespaceLatestResourceVersion = new ConcurrentHashMap<>();
+    } else {
+      perNamespaceLatestResourceVersion = null;
+    }
   }
 
   public synchronized void startEventFilteringModify(ResourceID resourceID) {
@@ -145,8 +154,7 @@ private synchronized EventHandling onEvent(
       log.debug("Processing event");
     }
     if (!unknownState) {
-      latestResourceVersion = resource.getMetadata().getResourceVersion();
-      log.debug("Setting latest resource version to: {}", latestResourceVersion);
+      setLatestResourceVersion(resource);
     }
     var cached = cache.get(resourceId);
     EventHandling result = EventHandling.NEW;
@@ -202,15 +210,16 @@ public synchronized void putResource(T newResource) {
     //
     // this also prevents resurrecting recently deleted entities for which the delete event
     // has already been processed
-    if (latestResourceVersion != null
+    var latestRV = getLatestResourceVersion(newResource.getMetadata().getNamespace());
+    if (latestRV != null
         && ReconcilerUtilsInternal.compareResourceVersions(
-                latestResourceVersion, newResource.getMetadata().getResourceVersion())
+                latestRV, newResource.getMetadata().getResourceVersion())
             > 0) {
       log.debug(
           "Resource {}: resourceVersion {} is not later than latest {}",
           resourceId,
           newResource.getMetadata().getResourceVersion(),
-          latestResourceVersion);
+          latestRV);
       return;
     }
 
@@ -246,4 +255,23 @@ void checkObsoleteResources() {
   public synchronized Optional<T> getResourceFromCache(ResourceID resourceID) {
     return Optional.ofNullable(cache.get(resourceID));
   }
+
+  private void setLatestResourceVersion(T resource) {
+    if (perNamespaceLatestResourceVersion == null) {
+      latestResourceVersion = resource.getMetadata().getResourceVersion();
+      log.debug("Setting latest resource version to: {}", latestResourceVersion);
+    } else {
+      perNamespaceLatestResourceVersion.put(
+          resource.getMetadata().getNamespace(), resource.getMetadata().getResourceVersion());
+      log.debug("Setting latest resource version to: {} for namesoace", latestResourceVersion);
+    }
+  }
+
+  public String getLatestResourceVersion(String namespace) {
+    if (perNamespaceLatestResourceVersion == null) {
+      return latestResourceVersion;
+    } else {
+      return perNamespaceLatestResourceVersion.get(namespace);
+    }
+  }
 }

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

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

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

@@ -40,7 +40,7 @@ class TemporaryPrimaryResourceCacheTest {
   @BeforeEach
   void setup() {
     temporaryResourceCache =
-        new TemporaryResourceCache<>(true, mock(ManagedInformerEventSource.class));
+        new TemporaryResourceCache<>(true, false, mock(ManagedInformerEventSource.class));
   }
 
   @Test
@@ -117,7 +117,7 @@ void removesResourceFromCache() {
   @Test
   void nonComparableResourceVersionsDisables() {
     this.temporaryResourceCache =
-        new TemporaryResourceCache<>(false, mock(ManagedInformerEventSource.class));
+        new TemporaryResourceCache<>(false, false, mock(ManagedInformerEventSource.class));
 
     this.temporaryResourceCache.putResource(testResource());