simple algorithm, refined 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/informer/EventFilterSupport.java

@@ -19,15 +19,10 @@
 import java.util.Map;
 import java.util.Optional;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import io.javaoperatorsdk.operator.processing.event.ResourceID;
 
 public class EventFilterSupport {
 
-  private static final Logger log = LoggerFactory.getLogger(EventFilterSupport.class);
-
   private final Map<ResourceID, EventFilterWindow> eventFilterWindows = new HashMap<>();
   private Long lastKnownVersionBeforeRelist = null;
 
@@ -45,7 +40,7 @@ public synchronized Optional<GenericResourceEvent> doneEventFilterModify(Resourc
     return check(ed, resourceID);
   }
 
-  public synchronized Optional<GenericResourceEvent> processRelevantEvent(
+  public synchronized Optional<GenericResourceEvent> processEvent(
       ResourceID resourceId, GenericResourceEvent genericResourceEvent) {
     var ed = eventFilterWindows.get(resourceId);
     if (ed != null) {
@@ -71,10 +66,6 @@ public synchronized void addToOwnResourceVersions(ResourceID resourceId, String
   }
 
   public synchronized void handleGhostResourceRemoval(ResourceID resourceId) {
-    var ed = eventFilterWindows.get(resourceId);
-    if (ed != null && !ed.canRemoved()) {
-      return;
-    }
     eventFilterWindows.remove(resourceId);
   }
 
@@ -84,10 +75,12 @@ synchronized Map<ResourceID, EventFilterWindow> getEventFilterWindows() {
   }
 
   public synchronized void setStartingReList(String lastKnownVersion) {
+    lastKnownVersionBeforeRelist = Long.parseLong(lastKnownVersion);
     eventFilterWindows.values().forEach(au -> au.setReListStartedFrom(lastKnownVersion));
   }
 
   public synchronized void setRelistFinished(String syncResourceVersions) {
+    lastKnownVersionBeforeRelist = null;
     eventFilterWindows.values().forEach(EventFilterWindow::setReListFinished);
   }
 

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

@@ -24,7 +24,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.processing.event.source.ResourceAction;
 
 /**
@@ -37,16 +36,10 @@ class EventFilterWindow {
   private final SortedMap<Long, GenericResourceEvent> relatedEvents = new TreeMap<>();
   private final SortedSet<Long> ownResourceVersions = new TreeSet<>();
   private Long lastResourceVersionBeforeReList;
-  private boolean affectedByReList;
   private int activeUpdates = 0;
-  private boolean ownRvEverAdded = false;
-  private int ownRvCount = 0;
-  private Long lastEmittedResourceRv;
-  private Long lastSeenRelatedRv;
 
   public EventFilterWindow(Long lastResourceVersionBeforeReList) {
     this.lastResourceVersionBeforeReList = lastResourceVersionBeforeReList;
-    this.affectedByReList = lastResourceVersionBeforeReList != null;
   }
 
   // Before we run this method
@@ -69,179 +62,138 @@ public synchronized Optional<GenericResourceEvent> check() {
     if (relatedEvents.isEmpty()) {
       return Optional.empty();
     }
-
-    long maxRelatedRv = relatedEvents.lastKey();
-
-    // While an in-flight write hasn't recorded its own RV yet, events past
-    // the highest known own RV may still turn out to be that write's echo —
-    // restrict the synth window so they're held until either the RV arrives
-    // or the write completes. ownRvCount is monotonic across cleanups so
-    // already-recorded RVs are not re-classified as "pending" once forgotten.
-    Long cutoff;
-    if (activeUpdates > ownRvCount) {
-      if (ownResourceVersions.isEmpty()) {
-        return Optional.empty();
-      }
-      cutoff = ownResourceVersions.last();
-    } else {
-      cutoff = maxRelatedRv;
+    if (activeUpdates == 0 && ownResourceVersions.isEmpty()) {
+      return eventForRangeAndClear(relatedEvents, ownResourceVersions);
     }
-
-    var windowMap = relatedEvents.headMap(cutoff + 1);
-    if (windowMap.isEmpty()) {
-      return Optional.empty();
+    if (ownResourceVersions.isEmpty()
+        && getFirstRelatedEvent().getAction().equals(ResourceAction.DELETED)) {
+      return eventForRangeAndClear(relatedEvents, ownResourceVersions);
     }
 
-    boolean foundForeign = false;
-    for (var entry : windowMap.entrySet()) {
-      if (!isOwnEcho(entry.getKey(), entry.getValue())) {
-        foundForeign = true;
+    var lastEventVersion = getLastRelatedEvent().getResourceVersion();
+    var numberOwnUpdatesSelected = 0;
+    long lastOwnVersion = -1;
+    for (long ownVersion : ownResourceVersions) {
+      if (ownVersion <= lastEventVersion) {
+        numberOwnUpdatesSelected++;
+        lastOwnVersion = ownVersion;
+      } else {
         break;
       }
     }
+    if (numberOwnUpdatesSelected > 0) {
+      if (numberOwnUpdatesSelected == ownResourceVersions.size() && activeUpdates == 0) {
+        return eventForRangeAndClear(relatedEvents, ownResourceVersions);
+      } else {
+        if (numberOwnUpdatesSelected < ownResourceVersions.size()) {
+          return eventForRangeAndClear(
+              relatedEvents.headMap(ownResourceVersions.tailSet(lastOwnVersion + 1).first()),
+              ownResourceVersions.headSet(lastOwnVersion + 1));
+        } else
+          return eventForRangeAndClear(
+              relatedEvents.headMap(lastOwnVersion + 1),
+              ownResourceVersions.headSet(lastOwnVersion + 1));
+      }
+    }
+    return Optional.empty();
+  }
 
-    Long prevSeen = lastSeenRelatedRv;
-    Optional<GenericResourceEvent> result = Optional.empty();
-
-    // Emit if there is a foreign event in the window, or if a previously emitted
-    // event already advanced the reconciler's view and a *new* event (not one we
-    // already saw at a prior check) now moves it further. ReList also forces an
-    // emit since it may have hidden events while it was running.
-    boolean shouldEmit =
-        foundForeign
-            || (lastEmittedResourceRv != null && (prevSeen == null || cutoff > prevSeen))
-            || affectedByReList;
+  // it has responsibility to clear those ranges and emit event if needed
+  Optional<GenericResourceEvent> eventForRangeAndClear(
+      SortedMap<Long, GenericResourceEvent> events, SortedSet<Long> ownResourceVersions) {
+    if (events.isEmpty()) {
+      return Optional.empty();
+    }
+    var isAnyEventFromReList =
+        events.values().stream().anyMatch(GenericResourceEvent::isPartOfReList);
 
-    if (shouldEmit) {
-      // Synthesize only from events that are *new* since the last check;
-      // carryover events (RV ≤ prevSeen) were already considered before and
-      // should not drive the synthesized event's resource versions.
-      var synthWindow = prevSeen == null ? windowMap : windowMap.tailMap(prevSeen + 1);
+    var first = getFirstRelatedEvent(events);
+    if (events.size() > 1 && first.getAction() == ResourceAction.DELETED) {
+      events.remove(events.firstKey());
+      first = getFirstRelatedEvent(events);
+    }
 
-      // When affected by a reList, treat events at or before the reList boundary
-      // as captured *during* relist and not informative — only events strictly
-      // after the boundary drive the synthesized output.
-      var effectiveWindow =
-          affectedByReList && lastResourceVersionBeforeReList != null
-              ? synthWindow.tailMap(lastResourceVersionBeforeReList + 1)
-              : synthWindow;
+    if (events.keySet().equals(ownResourceVersions) && !isAnyEventFromReList) {
+      GenericResourceEvent res = null;
+      var lastEvent = getLastRelatedEvent(events);
+      if (lastEvent.getAction() == ResourceAction.DELETED) {
+        res = lastEvent;
+      }
+      events.clear();
+      ownResourceVersions.clear();
+      return Optional.ofNullable(res);
+    }
 
-      if (!effectiveWindow.isEmpty()) {
-        var firstEvent = effectiveWindow.get(effectiveWindow.firstKey());
-        var lastEvent = effectiveWindow.get(effectiveWindow.lastKey());
+    if (events.size() == 1) {
+      ownResourceVersions.clear();
+      var res = Optional.of(events.values().iterator().next());
+      events.clear();
+      return res;
+    }
+    var lastEvent = getLastRelatedEvent(events);
+    if (lastEvent.getAction() == ResourceAction.DELETED) {
+      events.clear();
+      ownResourceVersions.clear();
+      return Optional.of(lastEvent);
+    }
 
-        // Identify the last DELETE in the synth window; a DELETE marks the
-        // boundary of the "current life" of the resource — anything before it
-        // represents a state that no longer exists.
-        GenericResourceEvent lastDelete = null;
-        boolean hasForeign = false;
-        boolean allForeignAreDeletes = true;
-        for (var entry : effectiveWindow.entrySet()) {
-          var ev = entry.getValue();
-          if (ev.getAction() == ResourceAction.DELETED) {
-            lastDelete = ev;
-          }
-          if (!isOwnEcho(entry.getKey(), ev)) {
-            hasForeign = true;
-            if (ev.getAction() != ResourceAction.DELETED) {
-              allForeignAreDeletes = false;
-            }
-          }
-        }
-        boolean lastIsOwnEcho = isOwnEcho(effectiveWindow.lastKey(), lastEvent);
-        boolean reListBeforeFirstOwn =
-            affectedByReList
-                && !ownResourceVersions.isEmpty()
-                && lastResourceVersionBeforeReList != null
-                && lastResourceVersionBeforeReList < ownResourceVersions.first();
+    var res =
+        Optional.of(
+            new GenericResourceEvent(
+                ResourceAction.UPDATED,
+                lastEvent.getResource().orElseThrow(),
+                first.getPreviousResource().isEmpty()
+                    ? first.getResource().orElseThrow()
+                    : first.getPreviousResource().orElseThrow(),
+                null));
+    events.clear();
+    ownResourceVersions.clear();
+    return res;
+  }
 
-        if (affectedByReList && (hasForeign || reListBeforeFirstOwn)) {
-          // ReList obscured part of the timeline AND something happened that
-          // wasn't purely our own activity — surface a DELETE with
-          // lastStateUnknown=true so the reconciler knows the latest known
-          // state is uncertain.
-          HasMetadata deleted = lastEvent.getResource().orElseThrow();
-          result =
-              Optional.of(new GenericResourceEvent(ResourceAction.DELETED, deleted, null, true));
-          lastEmittedResourceRv = cutoff;
-        } else if (!affectedByReList && hasForeign && allForeignAreDeletes && lastIsOwnEcho) {
-          // The synth window represents a delete-then-our-recreate sequence:
-          // the only foreign activity was DELETE(s) and the resource is back
-          // under our control. Nothing for the reconciler to know about.
-        } else if (effectiveWindow.size() == 1) {
-          result = Optional.of(firstEvent);
-          lastEmittedResourceRv = cutoff;
-        } else if (lastEvent.getAction() == ResourceAction.DELETED) {
-          result = Optional.of(lastEvent);
-          lastEmittedResourceRv = cutoff;
-        } else if (lastDelete != null) {
-          // A DELETE happened in the middle and the resource was recreated/updated
-          // afterwards. Synth UPDATED with previous = the deleted state.
-          HasMetadata previous = lastDelete.getResource().orElseThrow();
-          HasMetadata latest = lastEvent.getResource().orElseThrow();
-          result =
-              Optional.of(new GenericResourceEvent(ResourceAction.UPDATED, latest, previous, null));
-          lastEmittedResourceRv = cutoff;
-        } else {
-          HasMetadata previous =
-              firstEvent
-                  .getPreviousResource()
-                  .orElseGet(() -> firstEvent.getResource().orElseThrow());
-          HasMetadata latest = lastEvent.getResource().orElseThrow();
-          result =
-              Optional.of(new GenericResourceEvent(ResourceAction.UPDATED, latest, previous, null));
-          lastEmittedResourceRv = cutoff;
-        }
-      }
+  private GenericResourceEvent getFirstRelatedEvent() {
+    return getFirstRelatedEvent(relatedEvents);
+  }
 
-      if (affectedByReList) {
-        affectedByReList = false;
-        lastResourceVersionBeforeReList = null;
-      }
-    }
+  private GenericResourceEvent getFirstRelatedEvent(SortedMap<Long, GenericResourceEvent> subMap) {
+    return subMap.values().iterator().next();
+  }
 
-    lastSeenRelatedRv = prevSeen == null ? maxRelatedRv : Math.max(prevSeen, maxRelatedRv);
-    relatedEvents.headMap(cutoff + 1).clear();
-    ownResourceVersions.headSet(cutoff + 1).clear();
-    return result;
+  private GenericResourceEvent getLastRelatedEvent(SortedMap<Long, GenericResourceEvent> subMap) {
+    return subMap.get(subMap.lastKey());
   }
 
-  private boolean isOwnEcho(Long resourceVersion, GenericResourceEvent event) {
-    return event.getAction() != ResourceAction.DELETED
-        && ownResourceVersions.contains(resourceVersion);
+  private GenericResourceEvent getLastRelatedEvent() {
+    return getLastRelatedEvent(relatedEvents);
   }
 
   public synchronized boolean canRemoved() {
-    if (activeUpdates == 0 && ownResourceVersions.isEmpty() && ownRvEverAdded) {
-      if (!relatedEvents.isEmpty()) {
-        log.warn("Related events are not empty");
-      }
+    if (activeUpdates == 0 && ownResourceVersions.isEmpty() && relatedEvents.isEmpty()) {
       return true;
     }
     return false;
   }
 
   void addToOwnResourceVersions(String resourceVersion) {
     ownResourceVersions.add(Long.parseLong(resourceVersion));
-    ownRvEverAdded = true;
-    ownRvCount++;
   }
 
-  public void addRelatedEvent(GenericResourceEvent event) {
+  public synchronized void addRelatedEvent(GenericResourceEvent event) {
+    if (lastResourceVersionBeforeReList != null) {
+      event.setPartOfReList(true);
+    }
+
     relatedEvents.put(
         Long.parseLong(event.getResource().orElseThrow().getMetadata().getResourceVersion()),
         event);
   }
 
   public synchronized void setReListStartedFrom(String lastResourceVersionBeforeReList) {
     this.lastResourceVersionBeforeReList = Long.parseLong(lastResourceVersionBeforeReList);
-    this.affectedByReList = true;
   }
 
   public synchronized void setReListFinished() {
-    // Marker: relist has completed and check() may now process. The relist
-    // boundary (lastResourceVersionBeforeReList) is consumed by the next check
-    // and reset there along with affectedByReList.
+    lastResourceVersionBeforeReList = null;
   }
 
   public synchronized void increaseActiveUpdates() {

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

@@ -28,6 +28,7 @@ public class GenericResourceEvent extends ResourceEvent {
 
   private final HasMetadata previousResource;
   private final Boolean lastStateUnknow;
+  private boolean partOfReList = false;
 
   public GenericResourceEvent(
       ResourceAction action,
@@ -75,4 +76,16 @@ public boolean equals(Object o) {
   public int hashCode() {
     return Objects.hash(super.hashCode(), previousResource);
   }
+
+  public long getResourceVersion() {
+    return Long.parseLong(getResource().orElseThrow().getMetadata().getResourceVersion());
+  }
+
+  public boolean isPartOfReList() {
+    return partOfReList;
+  }
+
+  public void setPartOfReList(boolean partOfReList) {
+    this.partOfReList = partOfReList;
+  }
 }

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

@@ -119,7 +119,7 @@ private synchronized Optional<GenericResourceEvent> onEvent(
         cache.remove(resourceId);
       }
     }
-    return eventFilteringSupport.processRelevantEvent(resourceId, actualEvent);
+    return eventFilteringSupport.processEvent(resourceId, actualEvent);
   }
 
   static <T extends HasMetadata> GenericResourceEvent toGenericResourceEvent(