improvements and releated unit tests

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

@@ -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.*;
@@ -141,11 +141,22 @@ private void handleOnAddOrUpdate(
       ResourceAction action, T oldCustomResource, T newCustomResource) {
     var handling =
         temporaryResourceCache.onAddOrUpdateEvent(action, newCustomResource, oldCustomResource);
-    if (handling == EventHandling.PROPAGATE) {
-      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("{} event propagation for action: {}", handling, 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/EventFilterDetails.java

@@ -24,13 +24,14 @@
 import java.util.stream.Collectors;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.ReconcilerUtilsInternal;
 import io.javaoperatorsdk.operator.processing.event.source.ResourceAction;
 
 class EventFilterDetails {
 
   private int activeUpdates = 0;
-  private List<GenericResourceEvent> relatedEvents = new ArrayList<>();
-  private Set<String> allOwnResourceVersions = new HashSet<>();
+  private final List<GenericResourceEvent> relatedEvents = new ArrayList<>(5);
+  private final Set<String> allOwnResourceVersions = new HashSet<>(5);
 
   public void increaseActiveUpdates() {
     activeUpdates = activeUpdates + 1;
@@ -50,6 +51,10 @@ public int getActiveUpdates() {
     return activeUpdates;
   }
 
+  public boolean isNoActiveUpdate() {
+    return activeUpdates == 0;
+  }
+
   void addToOwnResourceVersions(String updateVersion) {
     allOwnResourceVersions.add(updateVersion);
   }
@@ -62,10 +67,7 @@ public Optional<GenericResourceEvent> prepareSummaryEventIfNotOwnEventsPresent()
     if (relatedEvents.isEmpty()) {
       return Optional.empty();
     }
-    if (allOwnResourceVersions.containsAll(
-        relatedEvents.stream()
-            .map(e -> e.getResource().orElseThrow().getMetadata().getResourceVersion())
-            .collect(Collectors.toSet()))) {
+    if (allOwnResourceVersions.containsAll(relatedEventResourceVersions())) {
       return Optional.empty();
     }
     var deleteEvent =
@@ -87,4 +89,23 @@ public Optional<GenericResourceEvent> prepareSummaryEventIfNotOwnEventsPresent()
             firstResource,
             null));
   }
+
+  private Set<String> relatedEventResourceVersions() {
+    return relatedEvents.stream()
+        .map(e -> e.getResource().orElseThrow().getMetadata().getResourceVersion())
+        .collect(Collectors.toSet());
+  }
+
+  public boolean newerOrEqualEventReceivedForOwnLastUpdate() {
+    if (allOwnResourceVersions.isEmpty()) {
+      return true;
+    }
+    String lastOwn =
+        allOwnResourceVersions.stream()
+            .reduce((a, b) -> ReconcilerUtilsInternal.compareResourceVersions(a, b) >= 0 ? a : b)
+            .orElseThrow();
+    return relatedEvents.stream()
+        .map(e -> e.getResource().orElseThrow().getMetadata().getResourceVersion())
+        .anyMatch(rv -> ReconcilerUtilsInternal.compareResourceVersions(rv, lastOwn) >= 0);
+  }
 }

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

@@ -33,7 +33,6 @@
 import io.javaoperatorsdk.operator.processing.event.ResourceID;
 import io.javaoperatorsdk.operator.processing.event.source.PrimaryToSecondaryMapper;
 import io.javaoperatorsdk.operator.processing.event.source.ResourceAction;
-import io.javaoperatorsdk.operator.processing.event.source.informer.TemporaryResourceCache.EventHandling;
 
 /**
  * Wraps informer(s) so they are connected to the eventing system of the framework. Note that since
@@ -154,20 +153,24 @@ private synchronized void onAddOrUpdate(ResourceAction action, R newObject, R ol
 
     var eventHandling = temporaryResourceCache.onAddOrUpdateEvent(action, newObject, oldObject);
 
-    if (eventHandling != EventHandling.PROPAGATE) {
-      log.debug(
-          "{} event propagation", eventHandling == EventHandling.IGNORE ? "Deferring" : "Skipping");
+    if (eventHandling.isEmpty()) {
+      log.debug("Deferring event propagation");
     } else if (eventAcceptedByFilter(action, newObject, oldObject)) {
       log.debug(
           "Propagating event for {}, resource with same version not result of a our update.",
           action);
-      propagateEvent(newObject);
+      var event = eventHandling.get();
+      handleEvent(
+          event.getAction(),
+          (R) event.getResource().orElseThrow(),
+          (R) event.getPreviousResource().orElse(null),
+          event.getLastStateUnknow());
     } else {
       log.debug("Event filtered out for operation: {}, resourceID: {}", action, resourceID);
     }
   }
 
-  private void propagateEvent(R object) {
+  protected void propagateEvent(R object) {
     var primaryResourceIdSet =
         configuration().getSecondaryToPrimaryMapper().toPrimaryResourceIDs(object);
     if (primaryResourceIdSet.isEmpty()) {

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

@@ -85,41 +85,44 @@ public synchronized Optional<GenericResourceEvent> doneEventFilterModify(Resourc
       return Optional.empty();
     }
     var ed = activeUpdates.get(resourceID);
-    if (ed == null || !ed.decreaseActiveUpdates()) {
-      log.debug(
-          "Active updates {} for resource id: {}",
-          ed != null ? ed.getActiveUpdates() : 0,
-          resourceID);
+    if (!ed.decreaseActiveUpdates()) {
+      log.debug("Active updates {} for resource id: {}", ed.getActiveUpdates(), resourceID);
+      return Optional.empty();
+    }
+
+    if (ed.newerOrEqualEventReceivedForOwnLastUpdate()) {
+      activeUpdates.remove(resourceID);
+      return ed.prepareSummaryEventIfNotOwnEventsPresent();
+    } else {
       return Optional.empty();
     }
-    activeUpdates.remove(resourceID);
-    return ed.prepareSummaryEventIfNotOwnEventsPresent();
   }
 
-  public void onDeleteEvent(T resource, boolean unknownState) {
-    onEvent(ResourceAction.DELETED, resource, null, unknownState);
+  public Optional<GenericResourceEvent> onDeleteEvent(T resource, boolean unknownState) {
+    return onEvent(ResourceAction.DELETED, resource, null, unknownState);
   }
 
-  public EventHandling onAddOrUpdateEvent(
+  public Optional<GenericResourceEvent> onAddOrUpdateEvent(
       ResourceAction action, T resource, T prevResourceVersion) {
-    return onEvent(action, resource, prevResourceVersion, false);
+    return onEvent(action, resource, prevResourceVersion, null);
   }
 
-  private synchronized EventHandling onEvent(
-      ResourceAction action, T resource, T prevResourceVersion, boolean unknownState) {
+  private synchronized Optional<GenericResourceEvent> onEvent(
+      ResourceAction action, T resource, T prevResourceVersion, Boolean unknownState) {
+    GenericResourceEvent actualEvent =
+        toGenericResourceEvent(action, resource, prevResourceVersion, unknownState);
     if (!comparableResourceVersions) {
-      return EventHandling.PROPAGATE;
+      return Optional.of(actualEvent);
     }
-
     var resourceId = ResourceID.fromResource(resource);
     if (log.isDebugEnabled()) {
       log.debug("Processing event");
     }
     var cached = cache.get(resourceId);
-    EventHandling result = EventHandling.PROPAGATE;
+    Optional<GenericResourceEvent> result = Optional.of(actualEvent);
     if (cached != null) {
       int comp = ReconcilerUtilsInternal.compareResourceVersions(resource, cached);
-      if (comp >= 0 || unknownState) {
+      if (comp >= 0 || Boolean.TRUE.equals(unknownState)) {
         log.debug(
             "Removing resource from temp cache. comparison: {} unknown state: {}",
             comp,
@@ -128,7 +131,9 @@ private synchronized EventHandling onEvent(
         // we propagate event only for our update or newer other can be discarded since we know we
         // will receive
         // additional event
-        result = comp == 0 ? EventHandling.IGNORE : EventHandling.PROPAGATE;
+        if (comp == 0) {
+          result = Optional.empty();
+        }
       } else {
         // 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.
@@ -141,13 +146,24 @@ private synchronized EventHandling onEvent(
       log.debug("Recording relevant event");
       au.addRelatedEvent(
           new GenericResourceEvent(action, resource, prevResourceVersion, unknownState));
-      return EventHandling.IGNORE;
+      // this is to cover the situation when we finished the filtering and caching update but
+      // did not receive events for our own updates yet.
+      if (au.isNoActiveUpdate() && au.newerOrEqualEventReceivedForOwnLastUpdate()) {
+        activeUpdates.remove(resourceId);
+        return au.prepareSummaryEventIfNotOwnEventsPresent();
+      }
+      return Optional.empty();
     } else {
-      log.debug("No active recornding, event handling: {}", result);
+      log.debug("No active recording, event handling: {}", result);
       return result;
     }
   }
 
+  static <T extends HasMetadata> GenericResourceEvent toGenericResourceEvent(
+      ResourceAction action, T resource, T prevResourceVersion, Boolean unknownState) {
+    return new GenericResourceEvent(action, resource, prevResourceVersion, unknownState);
+  }
+
   /** put the item into the cache if it's for a later state than what has already been observed. */
   public synchronized void putResource(T newResource) {
     if (!comparableResourceVersions) {

operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/controller/ControllerEventSourceTest.java

@@ -250,6 +250,7 @@ void propagatesIntermediateEventForExternalUpdateDuringFiltering() {
     // external update with rv 3 (older than our cached rv 4) — must propagate
     source.onUpdate(testResourceWithVersion(2), testResourceWithVersion(3));
     latch2.countDown();
+    source.onUpdate(testResourceWithVersion(3), testResourceWithVersion(5));
 
     await().untilAsserted(() -> verify(eventHandler, times(1)).handleEvent(any()));
   }