improve: filter only own updates for read-after-write-conistency

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

@@ -31,8 +31,8 @@
 import io.javaoperatorsdk.operator.processing.event.source.ResourceAction;
 import io.javaoperatorsdk.operator.processing.event.source.filter.OnDeleteFilter;
 import io.javaoperatorsdk.operator.processing.event.source.filter.OnUpdateFilter;
+import io.javaoperatorsdk.operator.processing.event.source.informer.GenericResourceEvent;
 import io.javaoperatorsdk.operator.processing.event.source.informer.ManagedInformerEventSource;
-import io.javaoperatorsdk.operator.processing.event.source.informer.TemporaryResourceCache.EventHandling;
 
 import static io.javaoperatorsdk.operator.ReconcilerUtilsInternal.handleKubernetesClientException;
 import static io.javaoperatorsdk.operator.processing.event.source.controller.InternalEventFilters.*;
@@ -84,7 +84,7 @@ protected synchronized void handleEvent(
     try {
       if (log.isDebugEnabled()) {
         log.debug("Event received with action: {}", action);
-        log.trace("Event Old resource: {},\n new resource: {}", oldResource, resource);
+        log.debug("Event Old resource: {},\n new resource: {}", oldResource, resource);
       }
       MDCUtils.addResourceInfo(resource);
       controller.getEventSourceManager().broadcastOnResourceEvent(action, resource, oldResource);
@@ -141,11 +141,22 @@ private void handleOnAddOrUpdate(
       ResourceAction action, T oldCustomResource, T newCustomResource) {
     var handling =
         temporaryResourceCache.onAddOrUpdateEvent(action, newCustomResource, oldCustomResource);
-    if (handling == EventHandling.NEW) {
-      handleEvent(action, newCustomResource, oldCustomResource, null);
-    } else if (log.isDebugEnabled()) {
-      log.debug("{} event propagation for action: {}", handling, action);
-    }
+    handling.ifPresentOrElse(
+        this::handleEvent,
+        () -> {
+          if (log.isDebugEnabled()) {
+            log.debug("Skipping/deferring event propagation for action: {}", action);
+          }
+        });
+  }
+
+  @SuppressWarnings("unchecked")
+  private void handleEvent(GenericResourceEvent r) {
+    handleEvent(
+        r.getAction(),
+        (T) r.getResource().orElseThrow(),
+        (T) r.getPreviousResource().orElse(null),
+        r.getLastStateUnknow());
   }
 
   @Override
@@ -154,10 +165,10 @@ public synchronized void onDelete(T resource, boolean deletedFinalStateUnknown)
         resource,
         ResourceAction.DELETED,
         () -> {
-          temporaryResourceCache.onDeleteEvent(resource, deletedFinalStateUnknown);
+          var res = temporaryResourceCache.onDeleteEvent(resource, deletedFinalStateUnknown);
           // delete event is quite special here, that requires special care, since we clean up
           // caches on delete event.
-          handleEvent(ResourceAction.DELETED, resource, null, deletedFinalStateUnknown);
+          res.ifPresent(this::handleEvent);
         });
   }
 

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

@@ -0,0 +1,89 @@
+/*
+ * Copyright Java Operator SDK Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *         http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.javaoperatorsdk.operator.processing.event.source.informer;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+
+import io.javaoperatorsdk.operator.processing.event.ResourceID;
+
+public class EventFilterSupport {
+
+  private final Map<ResourceID, EventFilterWindow> eventFilterWindows = new HashMap<>();
+  private boolean ongoingReList = false;
+
+  public synchronized void startEventFilteringModify(ResourceID resourceID) {
+    var ed =
+        eventFilterWindows.computeIfAbsent(resourceID, id -> new EventFilterWindow(ongoingReList));
+    ed.increaseActiveUpdates();
+  }
+
+  public synchronized Optional<GenericResourceEvent> doneEventFilterModify(ResourceID resourceID) {
+    var ed = eventFilterWindows.get(resourceID);
+    if (ed == null) return Optional.empty();
+    ed.decreaseActiveUpdates();
+    return check(ed, resourceID);
+  }
+
+  public synchronized Optional<GenericResourceEvent> processEvent(
+      ResourceID resourceId, GenericResourceEvent genericResourceEvent) {
+    var ed = eventFilterWindows.get(resourceId);
+    if (ed != null) {
+      ed.addRelatedEvent(genericResourceEvent);
+      return check(ed, resourceId);
+    } else {
+      return Optional.of(genericResourceEvent);
+    }
+  }
+
+  private Optional<GenericResourceEvent> check(
+      EventFilterWindow eventFilterWindow, ResourceID resourceID) {
+    var res = eventFilterWindow.check();
+    if (eventFilterWindow.canBeRemoved()) {
+      eventFilterWindows.remove(resourceID);
+    }
+    return res;
+  }
+
+  public synchronized void addToOwnResourceVersions(ResourceID resourceId, String resourceVersion) {
+    Optional.ofNullable(eventFilterWindows.get(resourceId))
+        .ifPresent(au -> au.addToOwnResourceVersions(resourceVersion));
+  }
+
+  public synchronized void handleGhostResourceRemoval(ResourceID resourceId) {
+    eventFilterWindows.remove(resourceId);
+  }
+
+  // for testing purposes
+  synchronized Map<ResourceID, EventFilterWindow> getEventFilterWindows() {
+    return eventFilterWindows;
+  }
+
+  public synchronized void setStartingReList() {
+    ongoingReList = true;
+    eventFilterWindows.values().forEach(EventFilterWindow::setReListStarted);
+  }
+
+  public synchronized void setRelistFinished() {
+    ongoingReList = false;
+    eventFilterWindows.values().forEach(EventFilterWindow::setReListFinished);
+  }
+
+  public synchronized boolean isActiveUpdateFor(ResourceID resourceId) {
+    return eventFilterWindows.containsKey(resourceId);
+  }
+}