diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/ManagedInformerEventSource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/ManagedInformerEventSource.java index a9c6818565..bb28e153d1 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/ManagedInformerEventSource.java +++ b/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( diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/TemporaryResourceCache.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/TemporaryResourceCache.java index 8879493a2a..e7d6f6f55d 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/TemporaryResourceCache.java +++ b/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(); } } diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/EventFilterWindowTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/EventFilterWindowTest.java index 70f6d1621c..367c4fa4f3 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/EventFilterWindowTest.java +++ b/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(); diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/readcacheafterwrite/onrelistfilter/OnRelistFilterIT.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/readcacheafterwrite/onrelistfilter/OnRelistFilterIT.java index df8d7c2591..8d2ff01cb7 100644 --- a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/readcacheafterwrite/onrelistfilter/OnRelistFilterIT.java +++ b/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 @@ *
  • re-list starts WHILE the update window is open — own write is propagated * */ -@Disabled("enable when fabric8 supports relist") class OnRelistFilterIT { static final String RESOURCE_NAME = "test-resource"; diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/readcacheafterwrite/onrelistfilter/OnRelistFilterReconciler.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/readcacheafterwrite/onrelistfilter/OnRelistFilterReconciler.java index 287141e4d1..13e8e72d74 100644 --- a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/readcacheafterwrite/onrelistfilter/OnRelistFilterReconciler.java +++ b/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 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 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 extends InformerEventSource { + // 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 latestReceivedVersion = new ConcurrentHashMap<>(); + RelistAwareInformerEventSource( InformerEventSourceConfiguration configuration, EventSourceContext

    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() { diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/readcacheafterwrite/ownsecondaryupdate/OwnSecondaryUpdateIT.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/readcacheafterwrite/ownsecondaryupdate/OwnSecondaryUpdateIT.java index dfa5b899fe..eaa8f14c69 100644 --- a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/readcacheafterwrite/ownsecondaryupdate/OwnSecondaryUpdateIT.java +++ b/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";