latest resync version

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/InformerManager.java

@@ -235,6 +235,13 @@ private Optional<InformerWrapper<R>> getSource(String namespace) {
     return Optional.ofNullable(sources.get(namespace));
   }
 
+  String lastSyncResourceVersion(String namespace) {
+    return getSource(namespace)
+        .map(InformerWrapper::getInformer)
+        .orElseThrow()
+        .lastSyncResourceVersion();
+  }
+
   @Override
   public void addIndexers(Map<String, Function<R, List<String>>> indexers) {
     this.indexers.putAll(indexers);

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

@@ -238,4 +238,8 @@ public Status getStatus() {
   public String getTargetNamespace() {
     return namespaceIdentifier;
   }
+
+  public SharedIndexInformer<T> getInformer() {
+    return informer;
+  }
 }

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

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

@@ -59,8 +59,6 @@ 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 final ConcurrentHashMap<String, String> perNamespaceLatestResourceVersion;
-  private volatile String latestResourceVersion;
 
   private final long obsoleteResourceCheckInterval;
   private volatile long lastObsoleteResourceCheck = System.currentTimeMillis();
@@ -74,28 +72,20 @@ 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) {
@@ -153,9 +143,6 @@ private synchronized EventHandling onEvent(
     if (log.isDebugEnabled()) {
       log.debug("Processing event");
     }
-    if (!unknownState) {
-      setLatestResourceVersion(resource);
-    }
     var cached = cache.get(resourceId);
     EventHandling result = EventHandling.NEW;
     if (cached != null) {
@@ -236,6 +223,10 @@ public synchronized void putResource(T newResource) {
     }
   }
 
+  private String getLatestResourceVersion(String namespace) {
+    return managedInformerEventSource.manager().lastSyncResourceVersion(namespace);
+  }
+
   void checkObsoleteResources() {
     if (System.currentTimeMillis() > lastObsoleteResourceCheck + obsoleteResourceCheckInterval) {
       lastObsoleteResourceCheck = System.currentTimeMillis();
@@ -244,7 +235,8 @@ void checkObsoleteResources() {
       while (iterator.hasNext()) {
         var e = iterator.next();
         if (ReconcilerUtilsInternal.compareResourceVersions(
-                e.getValue().getMetadata().getResourceVersion(), latestResourceVersion)
+                e.getValue().getMetadata().getResourceVersion(),
+                getLatestResourceVersion(e.getValue().getMetadata().getNamespace()))
             < 0) iterator.remove();
         managedInformerEventSource.handleEvent(ResourceAction.DELETED, e.getValue(), null, true);
         log.debug("Removing obsolete resource with ID: {}", e.getKey());
@@ -255,23 +247,4 @@ 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 namespace", 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

@@ -376,8 +376,12 @@ private CountDownLatch sendForEventFilteringUpdate(Deployment deployment, int re
   }
 
   private void withRealTemporaryResourceCache() {
-    temporaryResourceCache =
-        spy(new TemporaryResourceCache<>(true, false, mock(ManagedInformerEventSource.class)));
+    var mes = mock(ManagedInformerEventSource.class);
+    var mim = mock(InformerManager.class);
+    when(mes.manager()).thenReturn(mim);
+    when(mim.lastSyncResourceVersion(any())).thenReturn("1");
+
+    temporaryResourceCache = spy(new TemporaryResourceCache<>(true, mes));
     informerEventSource.setTemporalResourceCache(temporaryResourceCache);
   }
 

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

@@ -29,18 +29,25 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 
 class TemporaryPrimaryResourceCacheTest {
 
   public static final String RESOURCE_VERSION = "2";
 
   private TemporaryResourceCache<ConfigMap> temporaryResourceCache;
+  private volatile String latestSyncVersion;
 
   @BeforeEach
   void setup() {
-    temporaryResourceCache =
-        new TemporaryResourceCache<>(true, false, mock(ManagedInformerEventSource.class));
+    var mes = mock(ManagedInformerEventSource.class);
+    var mim = mock(InformerManager.class);
+    when(mes.manager()).thenReturn(mim);
+    when(mim.lastSyncResourceVersion(any())).then(a -> latestSyncVersion);
+
+    temporaryResourceCache = new TemporaryResourceCache<>(true, mes);
   }
 
   @Test
@@ -49,7 +56,7 @@ void updateAddsTheResourceIntoCacheIfTheInformerHasThePreviousResourceVersion()
     var prevTestResource = testResource();
     prevTestResource.getMetadata().setResourceVersion("1");
 
-    temporaryResourceCache.putResource(testResource);
+    putResource(testResource);
 
     var cached = temporaryResourceCache.getResourceFromCache(ResourceID.fromResource(testResource));
     assertThat(cached).isPresent();
@@ -63,8 +70,8 @@ void updateNotAddsTheResourceIntoCacheIfLaterVersionKnown() {
         ResourceAction.ADDED,
         testResource.toBuilder().editMetadata().withResourceVersion("3").endMetadata().build(),
         null);
-
-    temporaryResourceCache.putResource(testResource);
+    latestSyncVersion = "3";
+    putResource(testResource);
 
     var cached = temporaryResourceCache.getResourceFromCache(ResourceID.fromResource(testResource));
     assertThat(cached).isNotPresent();
@@ -74,7 +81,7 @@ void updateNotAddsTheResourceIntoCacheIfLaterVersionKnown() {
   void addOperationAddsTheResourceIfInformerCacheStillEmpty() {
     var testResource = testResource();
 
-    temporaryResourceCache.putResource(testResource);
+    putResource(testResource);
 
     var cached = temporaryResourceCache.getResourceFromCache(ResourceID.fromResource(testResource));
     assertThat(cached).isPresent();
@@ -84,9 +91,9 @@ void addOperationAddsTheResourceIfInformerCacheStillEmpty() {
   void addOperationNotAddsTheResourceIfInformerCacheNotEmpty() {
     var testResource = testResource();
 
-    temporaryResourceCache.putResource(testResource);
+    putResource(testResource);
 
-    temporaryResourceCache.putResource(
+    putResource(
         new ConfigMapBuilder(testResource)
             .editMetadata()
             .withResourceVersion("1")
@@ -109,15 +116,15 @@ void removesResourceFromCache() {
             .endMetadata()
             .build(),
         null);
-
+    latestSyncVersion = "3";
     assertThat(temporaryResourceCache.getResourceFromCache(ResourceID.fromResource(testResource)))
         .isNotPresent();
   }
 
   @Test
   void nonComparableResourceVersionsDisables() {
     this.temporaryResourceCache =
-        new TemporaryResourceCache<>(false, false, mock(ManagedInformerEventSource.class));
+        new TemporaryResourceCache<>(false, mock(ManagedInformerEventSource.class));
 
     this.temporaryResourceCache.putResource(testResource());
 
@@ -131,7 +138,7 @@ void eventReceivedDuringFiltering() {
 
     temporaryResourceCache.startEventFilteringModify(ResourceID.fromResource(testResource));
 
-    temporaryResourceCache.putResource(testResource);
+    putResource(testResource);
     assertThat(temporaryResourceCache.getResourceFromCache(ResourceID.fromResource(testResource)))
         .isPresent();
 
@@ -153,7 +160,7 @@ void newerEventDuringFiltering() {
 
     temporaryResourceCache.startEventFilteringModify(ResourceID.fromResource(testResource));
 
-    temporaryResourceCache.putResource(testResource);
+    putResource(testResource);
     assertThat(temporaryResourceCache.getResourceFromCache(ResourceID.fromResource(testResource)))
         .isPresent();
 
@@ -177,7 +184,7 @@ void eventAfterFiltering() {
 
     temporaryResourceCache.startEventFilteringModify(ResourceID.fromResource(testResource));
 
-    temporaryResourceCache.putResource(testResource);
+    putResource(testResource);
     assertThat(temporaryResourceCache.getResourceFromCache(ResourceID.fromResource(testResource)))
         .isPresent();
 
@@ -204,7 +211,7 @@ void putBeforeEvent() {
 
     var nextResource = testResource();
     nextResource.getMetadata().setResourceVersion("3");
-    temporaryResourceCache.putResource(nextResource);
+    putResource(nextResource);
 
     // the result is obsolete
     result = temporaryResourceCache.onAddOrUpdateEvent(ResourceAction.UPDATED, nextResource, null);
@@ -225,7 +232,7 @@ void putBeforeEventWithEventFiltering() {
     var resourceId = ResourceID.fromResource(testResource);
 
     temporaryResourceCache.startEventFilteringModify(resourceId);
-    temporaryResourceCache.putResource(nextResource);
+    putResource(nextResource);
     temporaryResourceCache.doneEventFilterModify(resourceId, "3");
 
     // the result is obsolete
@@ -252,7 +259,7 @@ void putAfterEventWithEventFilteringNoPost() {
             ResourceAction.UPDATED, nextResource, testResource);
     // the result is deferred
     assertThat(result).isEqualTo(EventHandling.DEFER);
-    temporaryResourceCache.putResource(nextResource);
+    putResource(nextResource);
     var postEvent = temporaryResourceCache.doneEventFilterModify(resourceId, "3");
 
     // there is no post event because the done call claimed responsibility for rv 3
@@ -291,15 +298,20 @@ void rapidDeletion() {
             .endMetadata()
             .build(),
         false);
-    temporaryResourceCache.putResource(testResource);
+    latestSyncVersion = "3";
+    putResource(testResource);
 
     assertThat(temporaryResourceCache.getResourceFromCache(ResourceID.fromResource(testResource)))
         .isEmpty();
   }
 
+  private void putResource(ConfigMap testResource) {
+    temporaryResourceCache.putResource(testResource);
+  }
+
   private ConfigMap propagateTestResourceToCache() {
     var testResource = testResource();
-    temporaryResourceCache.putResource(testResource);
+    putResource(testResource);
     assertThat(temporaryResourceCache.getResourceFromCache(ResourceID.fromResource(testResource)))
         .isPresent();
     return testResource;