updating obsolete event handling

Author: shawkins

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

@@ -31,6 +31,7 @@
 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.ManagedInformerEventSource;
+import io.javaoperatorsdk.operator.processing.event.source.informer.TemporaryResourceCache.EventHandling;
 
 import static io.javaoperatorsdk.operator.ReconcilerUtilsInternal.handleKubernetesClientException;
 import static io.javaoperatorsdk.operator.processing.KubernetesResourceUtils.getVersion;
@@ -138,15 +139,15 @@ private boolean isAcceptedByFilters(ResourceAction action, T resource, T oldReso
 
   @Override
   public void onAdd(T resource) {
-    var obsoleteResourceVersion = temporaryResourceCache.onAddOrUpdateEvent(resource);
-    handleEvent(ResourceAction.ADDED, resource, null, null, obsoleteResourceVersion);
+    var handling = temporaryResourceCache.onAddOrUpdateEvent(resource);
+    handleEvent(ResourceAction.ADDED, resource, null, null, handling != EventHandling.NEW);
   }
 
   @Override
   public void onUpdate(T oldCustomResource, T newCustomResource) {
-    var knownResourceVersion = temporaryResourceCache.onAddOrUpdateEvent(newCustomResource);
+    var handling = temporaryResourceCache.onAddOrUpdateEvent(newCustomResource);
     handleEvent(
-        ResourceAction.UPDATED, newCustomResource, oldCustomResource, null, knownResourceVersion);
+        ResourceAction.UPDATED, newCustomResource, oldCustomResource, null, handling != EventHandling.NEW);
   }
 
   @Override

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

@@ -33,6 +33,7 @@
 import io.javaoperatorsdk.operator.processing.event.ResourceID;
 import io.javaoperatorsdk.operator.processing.event.source.PrimaryToSecondaryMapper;
 import io.javaoperatorsdk.operator.processing.event.source.controller.ResourceAction;
+import io.javaoperatorsdk.operator.processing.event.source.informer.TemporaryResourceCache.EventHandling;
 
 import static io.javaoperatorsdk.operator.api.reconciler.Constants.DEFAULT_COMPARABLE_RESOURCE_VERSION;
 
@@ -159,10 +160,12 @@ private synchronized void onAddOrUpdate(Operation operation, R newObject, R oldO
     primaryToSecondaryIndex.onAddOrUpdate(newObject);
     var resourceID = ResourceID.fromResource(newObject);
 
-    if (temporaryResourceCache.onAddOrUpdateEvent(newObject)) {
+    var eventHandling = temporaryResourceCache.onAddOrUpdateEvent(newObject);
+    
+    if (eventHandling != EventHandling.NEW) {
       log.debug(
-          "Skipping event propagation for {}, since was a result of a reconcile action. Resource"
-              + " ID: {}",
+          "{} event propagation for {}. Resource ID: {}",
+          eventHandling == EventHandling.DEFER ? "Deferring" : "Skipping",
           operation,
           ResourceID.fromResource(newObject));
     } else if (eventAcceptedByFilter(operation, newObject, oldObject)) {

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

@@ -60,6 +60,12 @@ public class TemporaryResourceCache<T extends HasMetadata> {
   private String latestResourceVersion;
 
   private final Map<ResourceID, EventFilterDetails> activeUpdates = new HashMap<>();
+  
+  public enum EventHandling {
+    DEFER,
+    OBSOLETE,
+    NEW
+  }
 
   public TemporaryResourceCache(boolean comparableResourceVersions) {
     this.comparableResourceVersions = comparableResourceVersions;
@@ -95,13 +101,13 @@ public void onDeleteEvent(T resource, boolean unknownState) {
   /**
    * @return true if the resourceVersion was obsolete
    */
-  public boolean onAddOrUpdateEvent(T resource) {
+  public EventHandling onAddOrUpdateEvent(T resource) {
     return onEvent(resource, false, false);
   }
 
-  private synchronized boolean onEvent(T resource, boolean unknownState, boolean delete) {
+  private synchronized EventHandling onEvent(T resource, boolean unknownState, boolean delete) {
     if (!comparableResourceVersions) {
-      return false;
+      return EventHandling.NEW;
     }
 
     var resourceId = ResourceID.fromResource(resource);
@@ -115,30 +121,30 @@ private synchronized boolean onEvent(T resource, boolean unknownState, boolean d
       latestResourceVersion = resource.getMetadata().getResourceVersion();
     }
     var cached = cache.get(resourceId);
-    boolean obsoleteEvent = false;
+    EventHandling result = EventHandling.NEW;
     int comp = 0;
     if (cached != null) {
       comp = ReconcileUtils.validateAndCompareResourceVersions(resource, cached);
-      if (comp > 0 || unknownState) {
+      if (comp >= 0 || unknownState) {
         cache.remove(resourceId);
         // we propagate event only for our update or newer other can be discarded since we know we
         // will receive
         // additional event
-        obsoleteEvent = false;
+        result = comp == 0 ? EventHandling.OBSOLETE : EventHandling.NEW;
       } else {
-        obsoleteEvent = true;
+        result = EventHandling.OBSOLETE;
       }
     }
     var ed = activeUpdates.get(resourceId);
-    if (ed != null) {
+    if (ed != null && result != EventHandling.OBSOLETE) {
       ed.setLastEvent(
           delete
               ? new ResourceDeleteEvent(ResourceAction.DELETED, resourceId, resource, unknownState)
               : new ResourceEvent(
                   ResourceAction.UPDATED, resourceId, resource)); // todo true action
-      return true;
+      return EventHandling.DEFER;
     } else {
-      return obsoleteEvent;
+      return result;
     }
   }
 

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

@@ -28,6 +28,7 @@
 import io.fabric8.kubernetes.api.model.ConfigMapBuilder;
 import io.fabric8.kubernetes.api.model.ObjectMetaBuilder;
 import io.javaoperatorsdk.operator.processing.event.ResourceID;
+import io.javaoperatorsdk.operator.processing.event.source.informer.TemporaryResourceCache.EventHandling;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -133,7 +134,7 @@ void lockedEventBeforePut() throws Exception {
       temporaryResourceCache.putResource(testResource);
       assertThat(result.isDone()).isFalse();
       temporaryResourceCache.doneEventFilterModify(ResourceID.fromResource(testResource), "3");
-      assertThat(result.get(10, TimeUnit.SECONDS)).isTrue();
+      assertThat(result.get(10, TimeUnit.SECONDS)).isEqualTo(EventHandling.NEW);
     } finally {
       ex.shutdownNow();
     }
@@ -145,15 +146,15 @@ void putBeforeEvent() {
 
     // first ensure an event is not known
     var result = temporaryResourceCache.onAddOrUpdateEvent(testResource);
-    assertThat(result).isFalse();
+    assertThat(result).isEqualTo(EventHandling.NEW);
 
     var nextResource = testResource();
     nextResource.getMetadata().setResourceVersion("3");
     temporaryResourceCache.putResource(nextResource);
 
-    // now expect an event with the matching resourceVersion to be known after the put
+    // the result is obsolete
     result = temporaryResourceCache.onAddOrUpdateEvent(nextResource);
-    assertThat(result).isTrue();
+    assertThat(result).isEqualTo(EventHandling.OBSOLETE);
   }
 
   @Test
@@ -162,7 +163,7 @@ void putBeforeEventWithEventFiltering() {
 
     // first ensure an event is not known
     var result = temporaryResourceCache.onAddOrUpdateEvent(testResource);
-    assertThat(result).isFalse();
+    assertThat(result).isEqualTo(EventHandling.NEW);
 
     var nextResource = testResource();
     nextResource.getMetadata().setResourceVersion("3");
@@ -172,9 +173,29 @@ void putBeforeEventWithEventFiltering() {
     temporaryResourceCache.putResource(nextResource);
     temporaryResourceCache.doneEventFilterModify(resourceId, "3");
 
-    // the result is false since the put was not part of event filtering update
+    // the result is obsolete
     result = temporaryResourceCache.onAddOrUpdateEvent(nextResource);
-    assertThat(result).isTrue();
+    assertThat(result).isEqualTo(EventHandling.OBSOLETE);
+  }
+  
+  @Test
+  void putAfterEventWithEventFiltering() {
+    var testResource = testResource();
+
+    // first ensure an event is not known
+    var result = temporaryResourceCache.onAddOrUpdateEvent(testResource);
+    assertThat(result).isEqualTo(EventHandling.NEW);
+
+    var nextResource = testResource();
+    nextResource.getMetadata().setResourceVersion("3");
+    var resourceId = ResourceID.fromResource(testResource);
+
+    temporaryResourceCache.startEventFilteringModify(resourceId);
+    result = temporaryResourceCache.onAddOrUpdateEvent(nextResource);
+    // the result is deferred
+    assertThat(result).isEqualTo(EventHandling.DEFER);
+    temporaryResourceCache.putResource(nextResource);
+    temporaryResourceCache.doneEventFilterModify(resourceId, "3");
   }
 
   @Test