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

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/controller/ControllerEventSource.java

@@ -141,7 +141,7 @@ private void handleOnAddOrUpdate(
       ResourceAction action, T oldCustomResource, T newCustomResource) {
     var handling =
         temporaryResourceCache.onAddOrUpdateEvent(action, newCustomResource, oldCustomResource);
-    if (handling == EventHandling.NEW) {
+    if (handling == EventHandling.NEW || handling == EventHandling.IN_BETWEEN) {
       handleEvent(action, newCustomResource, oldCustomResource, null);
     } else if (log.isDebugEnabled()) {
       log.debug("{} event propagation for action: {}", handling, action);

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

@@ -15,7 +15,9 @@
  */
 package io.javaoperatorsdk.operator.processing.event.source.informer;
 
+import java.util.HashSet;
 import java.util.Optional;
+import java.util.Set;
 import java.util.function.UnaryOperator;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
@@ -27,6 +29,7 @@ class EventFilterDetails {
   private int activeUpdates = 0;
   private ResourceEvent lastEvent;
   private String lastOwnUpdatedResourceVersion;
+  private Set<String> allOwnResourceVersions = new HashSet<>();
 
   public void increaseActiveUpdates() {
     activeUpdates = activeUpdates + 1;
@@ -69,4 +72,12 @@ public Optional<ResourceEvent> getLatestEventAfterLastUpdateEvent() {
   public int getActiveUpdates() {
     return activeUpdates;
   }
+
+  void addToOwnResourceVersions(String updateVersion) {
+    allOwnResourceVersions.add(updateVersion);
+  }
+
+  public boolean isOwnResourceVersions(String resourceVersion) {
+    return allOwnResourceVersions.contains(resourceVersion);
+  }
 }

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

@@ -154,12 +154,12 @@ private synchronized void onAddOrUpdate(ResourceAction action, R newObject, R ol
 
     var eventHandling = temporaryResourceCache.onAddOrUpdateEvent(action, newObject, oldObject);
 
-    if (eventHandling != EventHandling.NEW) {
+    if (eventHandling != EventHandling.NEW && eventHandling != EventHandling.IN_BETWEEN) {
       log.debug(
           "{} event propagation", eventHandling == EventHandling.DEFER ? "Deferring" : "Skipping");
     } else if (eventAcceptedByFilter(action, newObject, oldObject)) {
       log.debug(
-          "Propagating event for {}, resource with same version not result of a reconciliation.",
+          "Propagating event for {}, resource with same version not result of a our update.",
           action);
       propagateEvent(newObject);
     } else {

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

@@ -65,6 +65,7 @@ public class TemporaryResourceCache<T extends HasMetadata> {
   public enum EventHandling {
     DEFER,
     OBSOLETE,
+    IN_BETWEEN,
     NEW
   }
 
@@ -145,17 +146,30 @@ private synchronized EventHandling onEvent(
         // additional event
         result = comp == 0 ? EventHandling.OBSOLETE : EventHandling.NEW;
       } else {
-        result = EventHandling.OBSOLETE;
+        // in this case we received and event that might be in some edge case that was
+        // already used in reconciler or after that, but before our updated resource version.
+        // That would be hard to distinguish, so for those we are propagating the event further.
+        log.debug("Received in between event.");
+        result = EventHandling.IN_BETWEEN;
       }
     }
-    var ed = activeUpdates.get(resourceId);
-    if (ed != null && result != EventHandling.OBSOLETE) {
-      log.debug("Setting last event for id: {} delete: {}", resourceId, delete);
-      ed.setLastEvent(
-          delete
-              ? new ResourceDeleteEvent(ResourceAction.DELETED, resourceId, resource, unknownState)
-              : new ExtendedResourceEvent(action, resourceId, resource, prevResourceVersion));
-      return EventHandling.DEFER;
+    var au = activeUpdates.get(resourceId);
+    if (au != null) {
+      if (result == EventHandling.IN_BETWEEN) {
+        return au.isOwnResourceVersions(resource.getMetadata().getResourceVersion())
+            ? EventHandling.DEFER
+            : EventHandling.IN_BETWEEN;
+      }
+      if (result == EventHandling.NEW) {
+        log.debug("Setting last event for id: {} delete: {}", resourceId, delete);
+        au.setLastEvent(
+            delete
+                ? new ResourceDeleteEvent(
+                    ResourceAction.DELETED, resourceId, resource, unknownState)
+                : new ExtendedResourceEvent(action, resourceId, resource, prevResourceVersion));
+        return EventHandling.DEFER;
+      }
+      return result;
     } else {
       return result;
     }
@@ -216,6 +230,10 @@ public synchronized void putResource(T newResource) {
           newResource.getMetadata().getResourceVersion(),
           resourceId);
       cache.put(resourceId, newResource);
+      var au = activeUpdates.get(resourceId);
+      if (au != null) {
+        au.addToOwnResourceVersions(newResource.getMetadata().getResourceVersion());
+      }
     }
   }