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

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/EventFilterDetails.java

@@ -32,6 +32,12 @@ class EventFilterDetails {
   private int activeUpdates = 0;
   private final List<GenericResourceEvent> relatedEvents = new ArrayList<>(5);
   private final Set<String> allOwnResourceVersions = new HashSet<>(5);
+  private boolean affectedByReList;
+  private volatile boolean reListSummaryEventSent = false;
+
+  public EventFilterDetails(boolean affectedByReList) {
+    this.affectedByReList = affectedByReList;
+  }
 
   public void increaseActiveUpdates() {
     activeUpdates = activeUpdates + 1;
@@ -63,13 +69,33 @@ public void addRelatedEvent(GenericResourceEvent event) {
     relatedEvents.add(event);
   }
 
-  public Optional<GenericResourceEvent> prepareSummaryEventIfNotOwnEventsPresent() {
+  public Optional<GenericResourceEvent> summaryEventForReList() {
+    if (!affectedByReList) {
+      throw new IllegalStateException(
+          "ReList summary event requested to detail not affected by relist");
+    }
+    if (reListSummaryEventSent) {
+      throw new IllegalStateException("ReList summary event already sent");
+    }
+    reListSummaryEventSent = true;
+    if (relatedEvents.isEmpty()) {
+      return Optional.empty();
+    }
+    return summaryEvent();
+  }
+
+  // todo unit tests for corner cases with empty collections
+  public Optional<GenericResourceEvent> summaryEvent() {
     if (relatedEvents.isEmpty()) {
       return Optional.empty();
     }
     if (allOwnResourceVersions.containsAll(relatedEventResourceVersions())) {
       return Optional.empty();
     }
+    return summaryEventInternal();
+  }
+
+  private Optional<GenericResourceEvent> summaryEventInternal() {
     var deleteEvent =
         relatedEvents.stream().filter(e -> e.getAction() == ResourceAction.DELETED).findFirst();
     if (deleteEvent.isPresent()) {
@@ -108,4 +134,16 @@ public boolean newerOrEqualEventReceivedForOwnLastUpdate() {
         .map(e -> e.getResource().orElseThrow().getMetadata().getResourceVersion())
         .anyMatch(rv -> ReconcilerUtilsInternal.compareResourceVersions(rv, lastOwn) >= 0);
   }
+
+  public boolean isAffectedByReList() {
+    return affectedByReList;
+  }
+
+  public void affectedByReList() {
+    this.affectedByReList = true;
+  }
+
+  public boolean isReListSummaryEventSent() {
+    return reListSummaryEventSent;
+  }
 }

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

@@ -146,9 +146,15 @@ public synchronized void stop() {
 
   @Override
   public void onList(String resourceVersion, boolean remainedEmpty) {
+    temporaryResourceCache.setRelistFinished();
     temporaryResourceCache.checkGhostResources();
   }
 
+  @Override
+  public void onBeforeList(String lastSyncResourceVersion) {
+    temporaryResourceCache.setOngoingRelist();
+  }
+
   @Override
   public void handleRecentResourceUpdate(
       ResourceID resourceID, R resource, R previousVersionOfResource) {

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

@@ -55,8 +55,9 @@ public class TemporaryResourceCache<T extends HasMetadata> {
   private static final Logger log = LoggerFactory.getLogger(TemporaryResourceCache.class);
 
   private final Map<ResourceID, T> cache = new ConcurrentHashMap<>();
-  private final Map<ResourceID, EventFilterDetails> activeUpdates = new HashMap<>();
+  private final Map<ResourceID, EventFilterDetails> cachingFilteringUpdates = new HashMap<>();
   private final boolean comparableResourceVersions;
+  private boolean informerOngoingRelist = false;
 
   private final ManagedInformerEventSource<T, ?, ?> managedInformerEventSource;
 
@@ -71,26 +72,22 @@ public synchronized void startEventFilteringModify(ResourceID resourceID) {
     if (!comparableResourceVersions) {
       return;
     }
-    var ed = activeUpdates.computeIfAbsent(resourceID, id -> new EventFilterDetails());
+    var ed =
+        cachingFilteringUpdates.computeIfAbsent(
+            resourceID, id -> new EventFilterDetails(informerOngoingRelist));
     ed.increaseActiveUpdates();
   }
 
   public synchronized Optional<GenericResourceEvent> doneEventFilterModify(ResourceID resourceID) {
     if (!comparableResourceVersions) {
       return Optional.empty();
     }
-    var ed = activeUpdates.get(resourceID);
+    var ed = cachingFilteringUpdates.get(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();
-    }
+    return finaleEventHandlingAndCleanup(resourceID, ed);
   }
 
   public Optional<GenericResourceEvent> onDeleteEvent(T resource, boolean unknownState) {
@@ -136,21 +133,20 @@ private synchronized Optional<GenericResourceEvent> onEvent(
         log.debug("Received intermediate event.");
       }
     }
-    var au = activeUpdates.get(resourceId);
+    var au = cachingFilteringUpdates.get(resourceId);
     if (au != null) {
       log.debug("Recording relevant event");
       au.addRelatedEvent(
           new GenericResourceEvent(action, resource, prevResourceVersion, unknownState));
       // 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 finaleEventHandlingAndCleanup(resourceId, au);
       }
       return Optional.empty();
     } else {
       log.debug("No active recording, event handling: {}", result);
-      return result;
+      return informerOngoingRelist ? Optional.of(actualEvent) : result;
     }
   }
 
@@ -206,7 +202,7 @@ public synchronized void putResource(T newResource) {
 
     // also make sure that we're later than the existing temporary entry
     var cachedResource = getResourceFromCache(resourceId).orElse(null);
-    Optional.ofNullable(activeUpdates.get(resourceId))
+    Optional.ofNullable(cachingFilteringUpdates.get(resourceId))
         .ifPresent(
             au -> au.addToOwnResourceVersions(newResource.getMetadata().getResourceVersion()));
 
@@ -264,6 +260,20 @@ public void checkGhostResources() {
     }
   }
 
+  private Optional<GenericResourceEvent> finaleEventHandlingAndCleanup(
+      ResourceID resourceID, EventFilterDetails ed) {
+    if (ed.newerOrEqualEventReceivedForOwnLastUpdate()) {
+      cachingFilteringUpdates.remove(resourceID);
+      if (ed.isAffectedByReList()) {
+        return ed.summaryEventForReList();
+      } else {
+        return ed.summaryEvent();
+      }
+    } else {
+      return Optional.empty();
+    }
+  }
+
   public synchronized Optional<T> getResourceFromCache(ResourceID resourceID) {
     return Optional.ofNullable(cache.get(resourceID));
   }
@@ -275,4 +285,13 @@ synchronized boolean isEmpty() {
   synchronized Map<ResourceID, T> getResources() {
     return Collections.unmodifiableMap(cache);
   }
+
+  public synchronized void setOngoingRelist() {
+    this.informerOngoingRelist = true;
+    cachingFilteringUpdates.values().forEach(EventFilterDetails::affectedByReList);
+  }
+
+  public synchronized void setRelistFinished() {
+    this.informerOngoingRelist = false;
+  }
 }

pom.xml

@@ -71,7 +71,7 @@
     <sonar.host.url>https://sonarcloud.io</sonar.host.url>
     <fabric8-httpclient-impl.name>jdk</fabric8-httpclient-impl.name>
     <junit.version>6.1.0</junit.version>
-    <fabric8-client.version>7.7.0</fabric8-client.version>
+    <fabric8-client.version>999-SNAPSHOT</fabric8-client.version>
     <slf4j.version>2.0.18</slf4j.version>
     <log4j.version>2.26.0</log4j.version>
     <mokito.version>5.23.0</mokito.version>