wip

Signed-off-by: Attila Mészáros <[email protected]>

Author: csviri

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/ControllerEventSource.java

@@ -81,7 +81,7 @@ public synchronized void start() {
     }
   }
 
-  public void eventReceived(
+  public synchronized void eventReceived(
       ResourceAction action,
       T resource,
       T oldResource,
@@ -90,11 +90,12 @@ public void eventReceived(
     try {
       if (log.isDebugEnabled()) {
         log.debug(
-            "Event received for resource: {} version: {} uuid: {} action: {}",
+            "Event received for resource: {} version: {} uuid: {} action: {} filter event: {}",
             ResourceID.fromResource(resource),
             getVersion(resource),
             resource.getMetadata().getUid(),
-            action);
+            action,
+            filterEvent);
         log.trace("Event Old resource: {},\n new resource: {}", oldResource, resource);
       }
       MDCUtils.addResourceInfo(resource);

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

@@ -96,12 +96,12 @@ public R eventFilteringUpdateAndCacheResource(R resourceToUpdate, UnaryOperator<
       log.debug("Update and cache: {}", id);
     }
     try {
-      temporaryResourceCache.startModifying(id);
+      temporaryResourceCache.startEventFilterModifying(id);
       var updated = updateMethod.apply(resourceToUpdate);
       handleRecentResourceUpdate(id, updated, resourceToUpdate);
       return updated;
     } finally {
-      temporaryResourceCache.doneModifying(id);
+      temporaryResourceCache.doneEventFilterModifying(id);
     }
   }
 

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

@@ -17,6 +17,7 @@
 
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.locks.ReentrantLock;
 
@@ -55,13 +56,14 @@ public class TemporaryResourceCache<T extends HasMetadata> {
   private final Map<ResourceID, T> cache = new ConcurrentHashMap<>();
   private final boolean comparableResourceVersions;
   private final Map<ResourceID, ReentrantLock> activelyModifying = new ConcurrentHashMap<>();
+  private final Set<ResourceID> skipFiltering = ConcurrentHashMap.newKeySet();
   private String latestResourceVersion;
 
   public TemporaryResourceCache(boolean comparableResourceVersions) {
     this.comparableResourceVersions = comparableResourceVersions;
   }
 
-  public void startModifying(ResourceID id) {
+  public void startEventFilterModifying(ResourceID id) {
     if (!comparableResourceVersions) {
       return;
     }
@@ -78,7 +80,7 @@ public void startModifying(ResourceID id) {
         .lock();
   }
 
-  public void doneModifying(ResourceID id) {
+  public void doneEventFilterModifying(ResourceID id) {
     if (!comparableResourceVersions) {
       return;
     }
@@ -102,35 +104,38 @@ public boolean onAddOrUpdateEvent(T resource) {
   }
 
   private boolean onEvent(T resource, boolean unknownState) {
-    ReentrantLock lock = activelyModifying.get(ResourceID.fromResource(resource));
+    var resourceId = ResourceID.fromResource(resource);
+    ReentrantLock lock = activelyModifying.get(resourceId);
     if (lock != null) {
       lock.lock(); // wait for the modification to finish
       lock.unlock(); // simply unlock as the event is guaranteed after the modification
     }
-    boolean[] known = new boolean[1];
+    boolean[] filter = new boolean[1];
     synchronized (this) {
       if (!unknownState) {
         latestResourceVersion = resource.getMetadata().getResourceVersion();
       }
       cache.computeIfPresent(
-          ResourceID.fromResource(resource),
+          resourceId,
           (id, cached) -> {
             boolean remove = unknownState;
             if (!unknownState) {
               int comp = ReconcileUtils.compareResourceVersions(resource, cached);
               if (comp >= 0) {
                 remove = true;
               }
-              if (comp <= 0) {
-                known[0] = true;
+              if (comp < 0) {
+                filter[0] = true;
+              } else {
+                filter[0] = skipFiltering.remove(resourceId);
               }
             }
             if (remove) {
               return null;
             }
             return cached;
           });
-      return known[0];
+      return filter[0];
     }
   }
 
@@ -178,10 +183,17 @@ public synchronized void putResource(T newResource) {
           newResource.getMetadata().getResourceVersion(),
           resourceId);
       cache.put(resourceId, newResource);
+      if (!isFilteringModification(resourceId)) {
+        skipFiltering.add(resourceId);
+      }
     }
   }
 
   public synchronized Optional<T> getResourceFromCache(ResourceID resourceID) {
     return Optional.ofNullable(cache.get(resourceID));
   }
+
+  private boolean isFilteringModification(ResourceID resourceId) {
+    return activelyModifying.containsKey(resourceId);
+  }
 }

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

@@ -122,15 +122,15 @@ void nonComparableResourceVersionsDisables() {
   void lockedEventBeforePut() throws Exception {
     var testResource = testResource();
 
-    temporaryResourceCache.startModifying(ResourceID.fromResource(testResource));
+    temporaryResourceCache.startEventFilterModifying(ResourceID.fromResource(testResource));
 
     ExecutorService ex = Executors.newSingleThreadExecutor();
     try {
       var result = ex.submit(() -> temporaryResourceCache.onAddOrUpdateEvent(testResource));
 
       temporaryResourceCache.putResource(testResource);
       assertThat(result.isDone()).isFalse();
-      temporaryResourceCache.doneModifying(ResourceID.fromResource(testResource));
+      temporaryResourceCache.doneEventFilterModifying(ResourceID.fromResource(testResource));
       assertThat(result.get(10, TimeUnit.SECONDS)).isTrue();
     } finally {
       ex.shutdownNow();