improve: using fabric8 re-list callback to cover event filtering edge cases (#3448)

Fabric8 client now supports callback onBeforeList in Informers, that
allows to enhance edge case support for event filtering:
Since on re-list some events could be lost, we will trigger the reconciliation in such scenarios.

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

@@ -155,16 +155,14 @@ public synchronized void stop() {
 
   @Override
   public void onList(String resourceVersion, boolean remainedEmpty) {
-    //  re-list supported by fabric8 client https://github.com/fabric8io/kubernetes-client/pull/7899
-    //    temporaryResourceCache.setRelistFinished(resourceVersion);
+    temporaryResourceCache.setRelistFinished();
     temporaryResourceCache.checkGhostResources();
   }
 
-  //  @Override (enable when
-  //  re-list supported by fabric8 client https://github.com/fabric8io/kubernetes-client/pull/7899
-  //  public void onBeforeList(String lastSyncResourceVersion) {
-  //    temporaryResourceCache.setOngoingRelist(lastSyncResourceVersion);
-  //  }
+  @Override
+  public void onBeforeList(String lastSyncResourceVersion) {
+    temporaryResourceCache.setOngoingRelist();
+  }
 
   @Override
   public void handleRecentResourceUpdate(

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

@@ -265,11 +265,11 @@ EventFilterSupport getEventFilterSupport() {
     return eventFilteringSupport;
   }
 
-  public void setOngoingRelist(String lastKnownSyncVersion) {
+  public void setOngoingRelist() {
     eventFilteringSupport.setStartingReList();
   }
 
-  public void setRelistFinished(String syncResourceVersions) {
+  public void setRelistFinished() {
     eventFilteringSupport.setRelistFinished();
   }
 }

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

@@ -15,7 +15,6 @@
  */
 package io.javaoperatorsdk.operator.processing.event.source.informer;
 
-import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 
 import io.fabric8.kubernetes.api.model.ConfigMap;
@@ -385,25 +384,6 @@ void additionalEventAndDeleteEvent() {
     assertThat(eventFilterWindow.canBeRemoved()).isTrue();
   }
 
-  @Test
-  @Disabled("should be part of event filter support")
-  void additionalEventAndDeleteEventNoUpdate() {
-    eventFilterWindow.increaseActiveUpdates();
-    eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION));
-    eventFilterWindow.addRelatedEvent(updateEvent(FIRST_OWN_VERSION));
-    eventFilterWindow.addRelatedEvent(updateEvent(FIRST_OWN_VERSION + 1));
-    eventFilterWindow.addRelatedEvent(deleteEvent(FIRST_OWN_VERSION + 2));
-
-    assertThat(eventFilterWindow.check())
-        .hasValueSatisfying(e -> assertDeleteEvent(e, FIRST_OWN_VERSION + 2));
-    assertThat(eventFilterWindow.check()).isEmpty();
-
-    assertEmptyState();
-    eventFilterWindow.decreaseActiveUpdates();
-
-    assertThat(eventFilterWindow.canBeRemoved()).isTrue();
-  }
-
   @Test
   void deleteEventInMiddleTwoUpdates() {
     eventFilterWindow.increaseActiveUpdates();

operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/readcacheafterwrite/onrelistfilter/OnRelistFilterIT.java

@@ -17,7 +17,6 @@
 
 import java.time.Duration;
 
-import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
 
@@ -37,7 +36,6 @@
  *   <li>re-list starts WHILE the update window is open — own write is propagated
  * </ul>
  */
-@Disabled("enable when fabric8 supports relist")
 class OnRelistFilterIT {
 
   static final String RESOURCE_NAME = "test-resource";

operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/readcacheafterwrite/onrelistfilter/OnRelistFilterReconciler.java

@@ -15,8 +15,11 @@
  */
 package io.javaoperatorsdk.operator.baseapi.readcacheafterwrite.onrelistfilter;
 
+import java.time.Duration;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 
@@ -32,6 +35,7 @@
 import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
 import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
 import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
+import io.javaoperatorsdk.operator.processing.event.ResourceID;
 import io.javaoperatorsdk.operator.processing.event.source.EventSource;
 import io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource;
 
@@ -78,7 +82,13 @@ public UpdateControl<OnRelistFilterCustomResource> reconcile(
         case NO_RELIST -> context.resourceOperations().serverSideApply(cm, configMapEventSource);
         case RELIST_AROUND_UPDATE -> {
           configMapEventSource.simulateOnBeforeList();
-          context.resourceOperations().serverSideApply(cm, configMapEventSource);
+          var applied = context.resourceOperations().serverSideApply(cm, configMapEventSource);
+          // Make the simulation deterministic: the own-write watch event arrives asynchronously,
+          // so we must wait for it to be received (and buffered into the still-open re-list
+          // window, where it is tagged as part of the re-list) BEFORE the re-list finishes.
+          // Otherwise onList may clear the window's re-list flag before the event lands and the
+          // event would be filtered as an own write — the race this test originally flaked on.
+          configMapEventSource.awaitWatchEventReceived(applied);
           configMapEventSource.simulateOnList();
         }
         case RELIST_COMPLETES_BEFORE_UPDATE -> {
@@ -90,20 +100,24 @@ public UpdateControl<OnRelistFilterCustomResource> reconcile(
           // Drive the event-filtering update path manually so we can fire onBeforeList AFTER the
           // window has been opened by startEventFilteringModify but BEFORE the SSA hits the API.
           var fieldManager = context.getControllerConfiguration().fieldManager();
-          configMapEventSource.eventFilteringUpdateAndCacheResource(
-              cm,
-              r -> {
-                configMapEventSource.simulateOnBeforeList();
-                return context
-                    .getClient()
-                    .resource(r)
-                    .patch(
-                        new PatchContext.Builder()
-                            .withForce(true)
-                            .withFieldManager(fieldManager)
-                            .withPatchType(PatchType.SERVER_SIDE_APPLY)
-                            .build());
-              });
+          var applied =
+              configMapEventSource.eventFilteringUpdateAndCacheResource(
+                  cm,
+                  r -> {
+                    configMapEventSource.simulateOnBeforeList();
+                    return context
+                        .getClient()
+                        .resource(r)
+                        .patch(
+                            new PatchContext.Builder()
+                                .withForce(true)
+                                .withFieldManager(fieldManager)
+                                .withPatchType(PatchType.SERVER_SIDE_APPLY)
+                                .build());
+                  });
+          // See RELIST_AROUND_UPDATE: wait for the own-write event to be buffered while the
+          // re-list is still in progress, so it is tagged as part of the re-list and propagated.
+          configMapEventSource.awaitWatchEventReceived(applied);
           configMapEventSource.simulateOnList();
         }
       }
@@ -154,14 +168,60 @@ private static ConfigMap prepareConfigMap(OnRelistFilterCustomResource p) {
   static class RelistAwareInformerEventSource<R extends HasMetadata, P extends HasMetadata>
       extends InformerEventSource<R, P> {
 
+    // Highest resourceVersion the informer has actually delivered (as a watch event) per resource.
+    // Lets a test block until the event for its own write has been received and processed.
+    private final ConcurrentMap<ResourceID, Long> latestReceivedVersion = new ConcurrentHashMap<>();
+
     RelistAwareInformerEventSource(
         InformerEventSourceConfiguration<R> configuration, EventSourceContext<P> context) {
       super(configuration, context);
     }
 
+    @Override
+    public void onAdd(R newResource) {
+      super.onAdd(newResource);
+      recordReceived(newResource);
+    }
+
+    @Override
+    public void onUpdate(R oldResource, R newResource) {
+      super.onUpdate(oldResource, newResource);
+      recordReceived(newResource);
+    }
+
+    private void recordReceived(R resource) {
+      latestReceivedVersion.merge(
+          ResourceID.fromResource(resource),
+          Long.parseLong(resource.getMetadata().getResourceVersion()),
+          Math::max);
+    }
+
+    /**
+     * Blocks until the informer has delivered a watch event for the given resource at a
+     * resourceVersion at least as recent as the one supplied (i.e. our own write has come back
+     * through the watch). Calling {@code super.onAdd/onUpdate} before recording guarantees the
+     * event is already buffered in the event-filter window by the time this returns.
+     */
+    void awaitWatchEventReceived(R resource) {
+      var id = ResourceID.fromResource(resource);
+      var target = Long.parseLong(resource.getMetadata().getResourceVersion());
+      var deadline = System.nanoTime() + Duration.ofSeconds(10).toNanos();
+      while (latestReceivedVersion.getOrDefault(id, -1L) < target) {
+        if (System.nanoTime() > deadline) {
+          throw new IllegalStateException(
+              "Timed out waiting for watch event with rv>=" + target + " for " + id);
+        }
+        try {
+          Thread.sleep(20);
+        } catch (InterruptedException e) {
+          Thread.currentThread().interrupt();
+          throw new IllegalStateException(e);
+        }
+      }
+    }
+
     void simulateOnBeforeList() {
-      // uncomment when fabric8 supports re-list
-      //      onBeforeList(null);
+      onBeforeList(null);
     }
 
     void simulateOnList() {

operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/readcacheafterwrite/ownsecondaryupdate/OwnSecondaryUpdateIT.java

@@ -17,7 +17,6 @@
 
 import java.time.Duration;
 
-import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
 
@@ -33,7 +32,6 @@
  * the secondary are filtered and do NOT trigger additional reconciliations. Counterpart to {@code
  * ExternalSecondaryUpdateIT}, which asserts the opposite for third-party updates.
  */
-@Disabled("enable if re-list notification supported by fabric8 client")
 class OwnSecondaryUpdateIT {
 
   static final String RESOURCE_NAME = "test-resource";