diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/ControllerEventSource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/ControllerEventSource.java index 07d59e039a..7afb62ea64 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/ControllerEventSource.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/ControllerEventSource.java @@ -31,8 +31,8 @@ import io.javaoperatorsdk.operator.processing.event.source.ResourceAction; 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.GenericResourceEvent; 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.event.source.controller.InternalEventFilters.*; @@ -84,7 +84,7 @@ protected synchronized void handleEvent( try { if (log.isDebugEnabled()) { log.debug("Event received with action: {}", action); - log.trace("Event Old resource: {},\n new resource: {}", oldResource, resource); + log.debug("Event Old resource: {},\n new resource: {}", oldResource, resource); } MDCUtils.addResourceInfo(resource); controller.getEventSourceManager().broadcastOnResourceEvent(action, resource, oldResource); @@ -141,11 +141,22 @@ private void handleOnAddOrUpdate( ResourceAction action, T oldCustomResource, T newCustomResource) { var handling = temporaryResourceCache.onAddOrUpdateEvent(action, newCustomResource, oldCustomResource); - if (handling == EventHandling.NEW) { - handleEvent(action, newCustomResource, oldCustomResource, null); - } else if (log.isDebugEnabled()) { - log.debug("{} event propagation for action: {}", handling, action); - } + handling.ifPresentOrElse( + this::handleEvent, + () -> { + if (log.isDebugEnabled()) { + log.debug("{} event propagation for action: {}", handling, action); + } + }); + } + + @SuppressWarnings("unchecked") + private void handleEvent(GenericResourceEvent r) { + handleEvent( + r.getAction(), + (T) r.getResource().orElseThrow(), + (T) r.getPreviousResource().orElse(null), + r.getLastStateUnknow()); } @Override @@ -154,10 +165,10 @@ public synchronized void onDelete(T resource, boolean deletedFinalStateUnknown) resource, ResourceAction.DELETED, () -> { - temporaryResourceCache.onDeleteEvent(resource, deletedFinalStateUnknown); + var res = temporaryResourceCache.onDeleteEvent(resource, deletedFinalStateUnknown); // delete event is quite special here, that requires special care, since we clean up // caches on delete event. - handleEvent(ResourceAction.DELETED, resource, null, deletedFinalStateUnknown); + res.ifPresent(this::handleEvent); }); } diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/EventFilterDetails.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/EventFilterDetails.java index b747c69dff..b9d12f9f10 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/EventFilterDetails.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/EventFilterDetails.java @@ -15,18 +15,23 @@ */ package io.javaoperatorsdk.operator.processing.event.source.informer; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; import java.util.Optional; +import java.util.Set; import java.util.function.UnaryOperator; +import java.util.stream.Collectors; import io.fabric8.kubernetes.api.model.HasMetadata; import io.javaoperatorsdk.operator.ReconcilerUtilsInternal; -import io.javaoperatorsdk.operator.processing.event.source.controller.ResourceEvent; +import io.javaoperatorsdk.operator.processing.event.source.ResourceAction; class EventFilterDetails { private int activeUpdates = 0; - private ResourceEvent lastEvent; - private String lastOwnUpdatedResourceVersion; + private final List relatedEvents = new ArrayList<>(5); + private final Set allOwnResourceVersions = new HashSet<>(5); public void increaseActiveUpdates() { activeUpdates = activeUpdates + 1; @@ -37,36 +42,70 @@ public void increaseActiveUpdates() { * controller to prevent race condition and send event from {@link * ManagedInformerEventSource#eventFilteringUpdateAndCacheResource(HasMetadata, UnaryOperator)} */ - public boolean decreaseActiveUpdates(String updatedResourceVersion) { - if (updatedResourceVersion != null - && (lastOwnUpdatedResourceVersion == null - || ReconcilerUtilsInternal.compareResourceVersions( - updatedResourceVersion, lastOwnUpdatedResourceVersion) - > 0)) { - lastOwnUpdatedResourceVersion = updatedResourceVersion; - } - + public boolean decreaseActiveUpdates() { activeUpdates = activeUpdates - 1; return activeUpdates == 0; } - public void setLastEvent(ResourceEvent event) { - lastEvent = event; + public int getActiveUpdates() { + return activeUpdates; + } + + public boolean isNoActiveUpdate() { + return activeUpdates == 0; + } + + void addToOwnResourceVersions(String updateVersion) { + allOwnResourceVersions.add(updateVersion); + } + + public void addRelatedEvent(GenericResourceEvent event) { + relatedEvents.add(event); } - public Optional getLatestEventAfterLastUpdateEvent() { - if (lastEvent != null - && (lastOwnUpdatedResourceVersion == null - || ReconcilerUtilsInternal.compareResourceVersions( - lastEvent.getResource().orElseThrow().getMetadata().getResourceVersion(), - lastOwnUpdatedResourceVersion) - > 0)) { - return Optional.of(lastEvent); + public Optional prepareSummaryEventIfNotOwnEventsPresent() { + if (relatedEvents.isEmpty()) { + return Optional.empty(); + } + if (allOwnResourceVersions.containsAll(relatedEventResourceVersions())) { + return Optional.empty(); + } + var deleteEvent = + relatedEvents.stream().filter(e -> e.getAction() == ResourceAction.DELETED).findFirst(); + if (deleteEvent.isPresent()) { + return deleteEvent; } - return Optional.empty(); + if (relatedEvents.size() == 1) { + return Optional.of(relatedEvents.get(0)); + } + var firstEvent = relatedEvents.get(0); + var firstResource = + firstEvent.getPreviousResource().orElseGet(() -> firstEvent.getResource().orElseThrow()); + + return Optional.of( + new GenericResourceEvent( + ResourceAction.UPDATED, + relatedEvents.get(relatedEvents.size() - 1).getResource().orElseThrow(), + firstResource, + null)); } - public int getActiveUpdates() { - return activeUpdates; + private Set relatedEventResourceVersions() { + return relatedEvents.stream() + .map(e -> e.getResource().orElseThrow().getMetadata().getResourceVersion()) + .collect(Collectors.toSet()); + } + + public boolean newerOrEqualEventReceivedForOwnLastUpdate() { + if (allOwnResourceVersions.isEmpty()) { + return true; + } + String lastOwn = + allOwnResourceVersions.stream() + .reduce((a, b) -> ReconcilerUtilsInternal.compareResourceVersions(a, b) >= 0 ? a : b) + .orElseThrow(); + return relatedEvents.stream() + .map(e -> e.getResource().orElseThrow().getMetadata().getResourceVersion()) + .anyMatch(rv -> ReconcilerUtilsInternal.compareResourceVersions(rv, lastOwn) >= 0); } } diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/ExtendedResourceEvent.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/GenericResourceEvent.java similarity index 81% rename from operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/ExtendedResourceEvent.java rename to operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/GenericResourceEvent.java index 5d30d1b0e1..c6911f48cc 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/ExtendedResourceEvent.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/GenericResourceEvent.java @@ -24,26 +24,32 @@ import io.javaoperatorsdk.operator.processing.event.source.controller.ResourceEvent; /** Used only for resource event filtering. */ -public class ExtendedResourceEvent extends ResourceEvent { +public class GenericResourceEvent extends ResourceEvent { private final HasMetadata previousResource; + private final Boolean lastStateUnknow; - public ExtendedResourceEvent( + public GenericResourceEvent( ResourceAction action, - ResourceID resourceID, HasMetadata latestResource, - HasMetadata previousResource) { - super(action, resourceID, latestResource); + HasMetadata previousResource, + Boolean lastStateUnknow) { + super(action, ResourceID.fromResource(latestResource), latestResource); this.previousResource = previousResource; + this.lastStateUnknow = lastStateUnknow; } public Optional getPreviousResource() { return Optional.ofNullable(previousResource); } + public Boolean getLastStateUnknow() { + return lastStateUnknow; + } + @Override public String toString() { - return "ExtendedResourceEvent{" + return "GenericResourceEvent{" + getPreviousResource() .map(r -> "previousResourceVersion=" + r.getMetadata().getResourceVersion()) .orElse("") @@ -61,7 +67,7 @@ public String toString() { public boolean equals(Object o) { if (o == null || getClass() != o.getClass()) return false; if (!super.equals(o)) return false; - ExtendedResourceEvent that = (ExtendedResourceEvent) o; + GenericResourceEvent that = (GenericResourceEvent) o; return Objects.equals(previousResource, that.previousResource); } diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerEventSource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerEventSource.java index 93d3eb5e80..d0cec2e112 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerEventSource.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerEventSource.java @@ -33,7 +33,6 @@ import io.javaoperatorsdk.operator.processing.event.ResourceID; import io.javaoperatorsdk.operator.processing.event.source.PrimaryToSecondaryMapper; import io.javaoperatorsdk.operator.processing.event.source.ResourceAction; -import io.javaoperatorsdk.operator.processing.event.source.informer.TemporaryResourceCache.EventHandling; /** * Wraps informer(s) so they are connected to the eventing system of the framework. Note that since @@ -154,20 +153,24 @@ private synchronized void onAddOrUpdate(ResourceAction action, R newObject, R ol var eventHandling = temporaryResourceCache.onAddOrUpdateEvent(action, newObject, oldObject); - if (eventHandling != EventHandling.NEW) { - log.debug( - "{} event propagation", eventHandling == EventHandling.DEFER ? "Deferring" : "Skipping"); + if (eventHandling.isEmpty()) { + log.debug("Deferring event propagation"); } else if (eventAcceptedByFilter(action, newObject, oldObject)) { log.debug( - "Propagating event for {}, resource with same version not result of a reconciliation.", + "Propagating event for {}, resource with same version not result of a our update.", action); - propagateEvent(newObject); + var event = eventHandling.get(); + handleEvent( + event.getAction(), + (R) event.getResource().orElseThrow(), + (R) event.getPreviousResource().orElse(null), + event.getLastStateUnknow()); } else { log.debug("Event filtered out for operation: {}, resourceID: {}", action, resourceID); } } - private void propagateEvent(R object) { + protected void propagateEvent(R object) { var primaryResourceIdSet = configuration().getSecondaryToPrimaryMapper().toPrimaryResourceIDs(object); if (primaryResourceIdSet.isEmpty()) { 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 f021101229..9dc487215a 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 @@ -46,7 +46,6 @@ import io.javaoperatorsdk.operator.processing.event.ResourceID; import io.javaoperatorsdk.operator.processing.event.source.*; import io.javaoperatorsdk.operator.processing.event.source.ResourceAction; -import io.javaoperatorsdk.operator.processing.event.source.controller.ResourceDeleteEvent; @SuppressWarnings("rawtypes") public abstract class ManagedInformerEventSource< @@ -101,45 +100,19 @@ public R eventFilteringUpdateAndCacheResource(R resourceToUpdate, UnaryOperator< try { temporaryResourceCache.startEventFilteringModify(id); updatedResource = updateMethod.apply(resourceToUpdate); - log.debug("Resource update successful"); handleRecentResourceUpdate(id, updatedResource, resourceToUpdate); + log.debug("Caching resource update successful"); return updatedResource; } finally { - var res = - temporaryResourceCache.doneEventFilterModify( - id, - updatedResource == null ? null : updatedResource.getMetadata().getResourceVersion()); - var updatedForLambda = updatedResource; + var res = temporaryResourceCache.doneEventFilterModify(id); res.ifPresentOrElse( r -> { - R latestResource = (R) r.getResource().orElseThrow(); - // as previous resource version we use the one from successful update, since - // we process new event here only if that is more recent then the event from our update. - // Note that this is equivalent with the scenario when an informer watch connection - // would reconnect and loose some events in between. - // If that update was not successful we still record the previous version from the - // actual event in the ExtendedResourceEvent. - R extendedResourcePrevVersion = - (r instanceof ExtendedResourceEvent) - ? (R) ((ExtendedResourceEvent) r).getPreviousResource().orElse(null) - : null; - R prevVersionOfResource = - updatedForLambda != null ? updatedForLambda : extendedResourcePrevVersion; - if (log.isDebugEnabled()) { - log.debug( - "Previous resource version: {} resource from update present: {}" - + " extendedPrevResource present: {}", - prevVersionOfResource.getMetadata().getResourceVersion(), - updatedForLambda != null, - extendedResourcePrevVersion != null); - } + log.debug("Propagating not own event"); handleEvent( r.getAction(), - latestResource, - prevVersionOfResource, - (r instanceof ResourceDeleteEvent) - ? ((ResourceDeleteEvent) r).isDeletedFinalStateUnknown() - : null); + (R) r.getResource().orElseThrow(), + (R) r.getPreviousResource().orElse(null), + r.getLastStateUnknow()); }, () -> log.debug("No new event present after the filtering update")); } 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 405f52cc8d..51ca7516ba 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 @@ -29,8 +29,6 @@ import io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependentResource; import io.javaoperatorsdk.operator.processing.event.ResourceID; import io.javaoperatorsdk.operator.processing.event.source.ResourceAction; -import io.javaoperatorsdk.operator.processing.event.source.controller.ResourceDeleteEvent; -import io.javaoperatorsdk.operator.processing.event.source.controller.ResourceEvent; /** * Temporal cache is used to solve the problem for {@link KubernetesDependentResource} that is, when @@ -62,12 +60,6 @@ public class TemporaryResourceCache { private final ManagedInformerEventSource managedInformerEventSource; - public enum EventHandling { - DEFER, - OBSOLETE, - NEW - } - public TemporaryResourceCache( boolean comparableResourceVersions, ManagedInformerEventSource managedInformerEventSource) { @@ -83,58 +75,49 @@ public synchronized void startEventFilteringModify(ResourceID resourceID) { ed.increaseActiveUpdates(); } - public synchronized Optional doneEventFilterModify( - ResourceID resourceID, String updatedResourceVersion) { + public synchronized Optional doneEventFilterModify(ResourceID resourceID) { if (!comparableResourceVersions) { return Optional.empty(); } var ed = activeUpdates.get(resourceID); - if (ed == null || !ed.decreaseActiveUpdates(updatedResourceVersion)) { - log.debug( - "Active updates {} for resource id: {}", - ed != null ? ed.getActiveUpdates() : 0, - 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(); } - activeUpdates.remove(resourceID); - var res = ed.getLatestEventAfterLastUpdateEvent(); - log.debug( - "Zero active updates for resource id: {}; event after update event: {}; updated resource" - + " version: {}", - resourceID, - res.isPresent(), - updatedResourceVersion); - return res; } - public void onDeleteEvent(T resource, boolean unknownState) { - onEvent(ResourceAction.DELETED, resource, null, unknownState, true); + public Optional onDeleteEvent(T resource, boolean unknownState) { + return onEvent(ResourceAction.DELETED, resource, null, unknownState); } - public EventHandling onAddOrUpdateEvent( + public Optional onAddOrUpdateEvent( ResourceAction action, T resource, T prevResourceVersion) { - return onEvent(action, resource, prevResourceVersion, false, false); + return onEvent(action, resource, prevResourceVersion, null); } - private synchronized EventHandling onEvent( - ResourceAction action, - T resource, - T prevResourceVersion, - boolean unknownState, - boolean delete) { + private synchronized Optional onEvent( + ResourceAction action, T resource, T prevResourceVersion, Boolean unknownState) { + GenericResourceEvent actualEvent = + toGenericResourceEvent(action, resource, prevResourceVersion, unknownState); if (!comparableResourceVersions) { - return EventHandling.NEW; + return Optional.of(actualEvent); } - var resourceId = ResourceID.fromResource(resource); if (log.isDebugEnabled()) { log.debug("Processing event"); } var cached = cache.get(resourceId); - EventHandling result = EventHandling.NEW; + Optional result = Optional.of(actualEvent); if (cached != null) { int comp = ReconcilerUtilsInternal.compareResourceVersions(resource, cached); - if (comp >= 0 || unknownState) { + if (comp >= 0 || Boolean.TRUE.equals(unknownState)) { log.debug( "Removing resource from temp cache. comparison: {} unknown state: {}", comp, @@ -143,24 +126,39 @@ private synchronized EventHandling onEvent( // we propagate event only for our update or newer other can be discarded since we know we // will receive // additional event - result = comp == 0 ? EventHandling.OBSOLETE : EventHandling.NEW; + if (comp == 0) { + result = Optional.empty(); + } } else { - result = EventHandling.OBSOLETE; + // in this case we received and event that might be in some edge case that was + // already used in reconciler or after that, but before our updated resource version. + // That would be hard to distinguish, so for those we are propagating the event further. + log.debug("Received intermediate event."); } } - var ed = activeUpdates.get(resourceId); - if (ed != null && result != EventHandling.OBSOLETE) { - log.debug("Setting last event for id: {} delete: {}", resourceId, delete); - ed.setLastEvent( - delete - ? new ResourceDeleteEvent(ResourceAction.DELETED, resourceId, resource, unknownState) - : new ExtendedResourceEvent(action, resourceId, resource, prevResourceVersion)); - return EventHandling.DEFER; + var au = activeUpdates.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 Optional.empty(); } else { + log.debug("No active recording, event handling: {}", result); return result; } } + static GenericResourceEvent toGenericResourceEvent( + ResourceAction action, T resource, T prevResourceVersion, Boolean unknownState) { + return new GenericResourceEvent(action, resource, prevResourceVersion, unknownState); + } + /** put the item into the cache if it's for a later state than what has already been observed. */ public synchronized void putResource(T newResource) { if (!comparableResourceVersions) { @@ -208,6 +206,9 @@ 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)) + .ifPresent( + au -> au.addToOwnResourceVersions(newResource.getMetadata().getResourceVersion())); if (cachedResource == null || ReconcilerUtilsInternal.compareResourceVersions(newResource, cachedResource) > 0) { diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/controller/ControllerEventSourceTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/controller/ControllerEventSourceTest.java index 4528fa8a83..a7765da4fa 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/controller/ControllerEventSourceTest.java +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/controller/ControllerEventSourceTest.java @@ -35,12 +35,14 @@ import io.javaoperatorsdk.operator.processing.Controller; import io.javaoperatorsdk.operator.processing.event.EventHandler; import io.javaoperatorsdk.operator.processing.event.EventSourceManager; +import io.javaoperatorsdk.operator.processing.event.ResourceID; import io.javaoperatorsdk.operator.processing.event.source.AbstractEventSourceTestBase; import io.javaoperatorsdk.operator.processing.event.source.EventFilterTestUtils; import io.javaoperatorsdk.operator.processing.event.source.ResourceAction; import io.javaoperatorsdk.operator.processing.event.source.filter.GenericFilter; import io.javaoperatorsdk.operator.processing.event.source.filter.OnAddFilter; import io.javaoperatorsdk.operator.processing.event.source.filter.OnUpdateFilter; +import io.javaoperatorsdk.operator.processing.event.source.informer.TemporaryResourceCache; import io.javaoperatorsdk.operator.sample.simple.TestCustomResource; import static io.javaoperatorsdk.operator.processing.event.source.EventFilterTestUtils.withResourceVersion; @@ -227,6 +229,74 @@ void eventFilteringExceptionDuringUpdate() { expectHandleEvent(2, 1); } + @Test + void propagatesIntermediateEventForExternalUpdateDuringFiltering() { + // Causal-dependency scenario: a third party updated the resource between our read and + // our write. The informer delivers that update during our active filter; since its + // resource version is NOT one of our own writes, it must be propagated. + var src = new TestableControllerEventSource(new TestController(null, null, null)); + setUpSource(src, true, controllerConfig); + + var resourceId = ResourceID.fromResource(TestUtils.testCustomResource1()); + + // first filter writes rv 4 (our own); a second concurrent filter keeps the + // active-updates window open while the event below is processed + var latch1 = sendForEventFilteringUpdate(4); + var latch2 = sendForEventFilteringUpdate(testResourceWithVersion(4), 5); + + latch1.countDown(); + awaitCachedResourceVersion(src.tempCache(), resourceId, "4"); + + // external update with rv 3 (older than our cached rv 4) — must propagate + source.onUpdate(testResourceWithVersion(2), testResourceWithVersion(3)); + latch2.countDown(); + source.onUpdate(testResourceWithVersion(3), testResourceWithVersion(5)); + + await().untilAsserted(() -> verify(eventHandler, times(1)).handleEvent(any())); + } + + @Test + void doesNotPropagateIntermediateEventForOurOwnIntermediateUpdate() { + // Two consecutive own writes (rv 3 then rv 4) within an open filter window: an event + // for the older own version must be deferred since it's recognized as our own. A + // third concurrent filter keeps the active-updates window open while the event below + // is processed. + var src = new TestableControllerEventSource(new TestController(null, null, null)); + setUpSource(src, true, controllerConfig); + + var resourceId = ResourceID.fromResource(TestUtils.testCustomResource1()); + + var latch1 = sendForEventFilteringUpdate(3); + var latch2 = sendForEventFilteringUpdate(testResourceWithVersion(3), 4); + var latch3 = sendForEventFilteringUpdate(testResourceWithVersion(4), 5); + + latch1.countDown(); + awaitCachedResourceVersion(src.tempCache(), resourceId, "3"); + latch2.countDown(); + awaitCachedResourceVersion(src.tempCache(), resourceId, "4"); + + // event for our own rv 3 (older than cached rv 4) — must be deferred + source.onUpdate(testResourceWithVersion(2), testResourceWithVersion(3)); + + verify(eventHandler, never()).handleEvent(any()); + + latch3.countDown(); + } + + private void awaitCachedResourceVersion( + TemporaryResourceCache cache, + ResourceID resourceId, + String resourceVersion) { + await() + .untilAsserted( + () -> + assertThat( + cache + .getResourceFromCache(resourceId) + .map(r -> r.getMetadata().getResourceVersion())) + .hasValue(resourceVersion)); + } + private void expectHandleEvent(int newResourceVersion, int oldResourceVersion) { await() .untilAsserted( @@ -247,7 +317,7 @@ private void expectHandleEvent(int newResourceVersion, int oldResourceVersion) { .isEqualTo("" + oldResourceVersion); return true; }), - isNull()); + any()); }); } @@ -330,4 +400,15 @@ public TestConfiguration( false); } } + + private static class TestableControllerEventSource + extends ControllerEventSource { + TestableControllerEventSource(Controller controller) { + super(controller); + } + + TemporaryResourceCache tempCache() { + return temporaryResourceCache; + } + } } diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerEventSourceTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerEventSourceTest.java index fe78bd3147..65bb3f0fea 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerEventSourceTest.java +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerEventSourceTest.java @@ -25,11 +25,11 @@ import java.util.stream.Stream; import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.RepeatedTest; import org.junit.jupiter.api.Test; import io.fabric8.kubernetes.api.model.ObjectMeta; import io.fabric8.kubernetes.api.model.apps.Deployment; -import io.fabric8.kubernetes.api.model.apps.DeploymentBuilder; import io.fabric8.kubernetes.client.KubernetesClient; import io.fabric8.kubernetes.client.KubernetesClientException; import io.javaoperatorsdk.operator.MockKubernetesClient; @@ -46,7 +46,6 @@ import io.javaoperatorsdk.operator.processing.event.source.EventFilterTestUtils; import io.javaoperatorsdk.operator.processing.event.source.ResourceAction; import io.javaoperatorsdk.operator.processing.event.source.SecondaryToPrimaryMapper; -import io.javaoperatorsdk.operator.processing.event.source.informer.TemporaryResourceCache.EventHandling; import io.javaoperatorsdk.operator.sample.simple.TestCustomResource; import static io.javaoperatorsdk.operator.api.reconciler.Constants.DEFAULT_NAMESPACES_SET; @@ -71,7 +70,8 @@ class InformerEventSourceTest { private static final String PREV_RESOURCE_VERSION = "0"; - private static final String DEFAULT_RESOURCE_VERSION = "1"; + private static final String DEFAULT_RESOURCE_VERSION = "2"; + public static final int REPEAT_COUNT = 10; private InformerEventSource informerEventSource; private final KubernetesClient clientMock = MockKubernetesClient.client(Deployment.class); @@ -112,43 +112,6 @@ public synchronized void start() {} informerEventSource.setTemporalResourceCache(temporaryResourceCache); } - @Test - void skipsEventPropagation() { - when(temporaryResourceCache.getResourceFromCache(any())) - .thenReturn(Optional.of(testDeployment())); - - when(temporaryResourceCache.onAddOrUpdateEvent(any(), any(), any())) - .thenReturn(EventHandling.OBSOLETE); - - informerEventSource.onAdd(testDeployment()); - informerEventSource.onUpdate(testDeployment(), testDeployment()); - - verify(eventHandlerMock, never()).handleEvent(any()); - } - - @Test - void processEventPropagationWithoutAnnotation() { - when(temporaryResourceCache.onAddOrUpdateEvent(any(), any(), any())) - .thenReturn(EventHandling.NEW); - informerEventSource.onUpdate(testDeployment(), testDeployment()); - - verify(eventHandlerMock, times(1)).handleEvent(any()); - } - - @Test - void processEventPropagationWithIncorrectAnnotation() { - when(temporaryResourceCache.onAddOrUpdateEvent(any(), any(), any())) - .thenReturn(EventHandling.NEW); - informerEventSource.onAdd( - new DeploymentBuilder(testDeployment()) - .editMetadata() - .addToAnnotations(InformerEventSource.PREVIOUS_ANNOTATION_KEY, "invalid") - .endMetadata() - .build()); - - verify(eventHandlerMock, times(1)).handleEvent(any()); - } - @Test void propagateEventAndRemoveResourceFromTempCacheIfResourceVersionMismatch() { withRealTemporaryResourceCache(); @@ -205,19 +168,21 @@ void filtersOnDeleteEvents() { verify(eventHandlerMock, never()).handleEvent(any()); } - @Test + @RepeatedTest(REPEAT_COUNT) void handlesPrevResourceVersionForUpdate() { withRealTemporaryResourceCache(); - CountDownLatch latch = sendForEventFilteringUpdate(2); + CountDownLatch latch = sendForEventFilteringUpdate(3); informerEventSource.onUpdate( - deploymentWithResourceVersion(2), deploymentWithResourceVersion(3)); + deploymentWithResourceVersion(1), deploymentWithResourceVersion(2)); latch.countDown(); + informerEventSource.onUpdate( + deploymentWithResourceVersion(2), deploymentWithResourceVersion(3)); - expectHandleEvent(3, 2); + expectHandleAddEvent(3, 1); } - @Test + @RepeatedTest(REPEAT_COUNT) void handlesPrevResourceVersionForUpdateInCaseOfException() { withRealTemporaryResourceCache(); @@ -232,10 +197,10 @@ void handlesPrevResourceVersionForUpdateInCaseOfException() { deploymentWithResourceVersion(1), deploymentWithResourceVersion(2)); latch.countDown(); - expectHandleEvent(2, 1); + expectHandleAddEvent(2, 1); } - @Test + @RepeatedTest(REPEAT_COUNT) void handlesPrevResourceVersionForUpdateInCaseOfMultipleUpdates() { withRealTemporaryResourceCache(); @@ -247,10 +212,10 @@ void handlesPrevResourceVersionForUpdateInCaseOfMultipleUpdates() { withResourceVersion(testDeployment(), 3), withResourceVersion(testDeployment(), 4)); latch.countDown(); - expectHandleEvent(4, 2); + expectHandleAddEvent(4, 2); } - @Test + @RepeatedTest(REPEAT_COUNT) void doesNotPropagateEventIfReceivedBeforeUpdate() { withRealTemporaryResourceCache(); @@ -262,89 +227,96 @@ void doesNotPropagateEventIfReceivedBeforeUpdate() { assertNoEventProduced(); } - @Test - void filterAddEventBeforeUpdate() { - withRealTemporaryResourceCache(); - - CountDownLatch latch = sendForEventFilteringUpdate(2); - informerEventSource.onAdd(deploymentWithResourceVersion(1)); - latch.countDown(); - - assertNoEventProduced(); - } - - @Test + @RepeatedTest(REPEAT_COUNT) void multipleCachingFilteringUpdates() { withRealTemporaryResourceCache(); - CountDownLatch latch = sendForEventFilteringUpdate(2); + CountDownLatch latch = sendForEventFilteringUpdate(3); CountDownLatch latch2 = - sendForEventFilteringUpdate(withResourceVersion(testDeployment(), 2), 3); + sendForEventFilteringUpdate(withResourceVersion(testDeployment(), 3), 4); informerEventSource.onUpdate( - deploymentWithResourceVersion(1), deploymentWithResourceVersion(2)); + deploymentWithResourceVersion(2), deploymentWithResourceVersion(3)); latch.countDown(); latch2.countDown(); informerEventSource.onUpdate( - deploymentWithResourceVersion(2), deploymentWithResourceVersion(3)); + deploymentWithResourceVersion(3), deploymentWithResourceVersion(4)); assertNoEventProduced(); } - @Test + @RepeatedTest(REPEAT_COUNT) void multipleCachingFilteringUpdates_variant2() { withRealTemporaryResourceCache(); - CountDownLatch latch = sendForEventFilteringUpdate(2); + CountDownLatch latch = sendForEventFilteringUpdate(3); CountDownLatch latch2 = - sendForEventFilteringUpdate(withResourceVersion(testDeployment(), 2), 3); + sendForEventFilteringUpdate(withResourceVersion(testDeployment(), 3), 4); informerEventSource.onUpdate( - deploymentWithResourceVersion(1), deploymentWithResourceVersion(2)); + deploymentWithResourceVersion(2), deploymentWithResourceVersion(3)); latch.countDown(); informerEventSource.onUpdate( - deploymentWithResourceVersion(2), deploymentWithResourceVersion(3)); + deploymentWithResourceVersion(3), deploymentWithResourceVersion(4)); latch2.countDown(); assertNoEventProduced(); } - @Test + @RepeatedTest(REPEAT_COUNT) void multipleCachingFilteringUpdates_variant3() { withRealTemporaryResourceCache(); - CountDownLatch latch = sendForEventFilteringUpdate(2); + CountDownLatch latch = sendForEventFilteringUpdate(3); CountDownLatch latch2 = - sendForEventFilteringUpdate(withResourceVersion(testDeployment(), 2), 3); + sendForEventFilteringUpdate(withResourceVersion(testDeployment(), 3), 4); latch.countDown(); - informerEventSource.onUpdate( - deploymentWithResourceVersion(1), deploymentWithResourceVersion(2)); informerEventSource.onUpdate( deploymentWithResourceVersion(2), deploymentWithResourceVersion(3)); + informerEventSource.onUpdate( + deploymentWithResourceVersion(4), deploymentWithResourceVersion(4)); latch2.countDown(); assertNoEventProduced(); } - @Test + @RepeatedTest(REPEAT_COUNT) void multipleCachingFilteringUpdates_variant4() { withRealTemporaryResourceCache(); - CountDownLatch latch = sendForEventFilteringUpdate(2); + CountDownLatch latch = sendForEventFilteringUpdate(3); CountDownLatch latch2 = - sendForEventFilteringUpdate(withResourceVersion(testDeployment(), 2), 3); + sendForEventFilteringUpdate(withResourceVersion(testDeployment(), 3), 4); - informerEventSource.onUpdate( - deploymentWithResourceVersion(1), deploymentWithResourceVersion(2)); informerEventSource.onUpdate( deploymentWithResourceVersion(2), deploymentWithResourceVersion(3)); + informerEventSource.onUpdate( + deploymentWithResourceVersion(3), deploymentWithResourceVersion(4)); latch.countDown(); latch2.countDown(); assertNoEventProduced(); } - @Test + @RepeatedTest(REPEAT_COUNT) + void multipleCachingFilteringUpdates_variant5() { + withRealTemporaryResourceCache(); + + CountDownLatch latch = sendForEventFilteringUpdate(3); + CountDownLatch latch2 = + sendForEventFilteringUpdate(withResourceVersion(testDeployment(), 3), 4); + latch.countDown(); + latch2.countDown(); + + informerEventSource.onUpdate( + deploymentWithResourceVersion(2), deploymentWithResourceVersion(3)); + informerEventSource.onUpdate( + deploymentWithResourceVersion(3), deploymentWithResourceVersion(4)); + + assertNoEventProduced(); + } + + @RepeatedTest(REPEAT_COUNT) void ghostCheckRemovesCachedResourceDuringFilteringUpdate() { var mes = mock(ManagedInformerEventSource.class); var mim = mock(InformerManager.class); @@ -370,11 +342,11 @@ void ghostCheckRemovesCachedResourceDuringFilteringUpdate() { assertThat(temporaryResourceCache.getResourceFromCache(resourceId)).isEmpty(); // complete the filtering update - the resource should not reappear - temporaryResourceCache.doneEventFilterModify(resourceId, "2"); + temporaryResourceCache.doneEventFilterModify(resourceId); assertThat(temporaryResourceCache.getResourceFromCache(resourceId)).isEmpty(); } - @Test + @RepeatedTest(REPEAT_COUNT) void ghostCheckRunsConcurrentlyWithPutResource() { var mes = mock(ManagedInformerEventSource.class); var mim = mock(InformerManager.class); @@ -405,7 +377,7 @@ void ghostCheckRunsConcurrentlyWithPutResource() { .isPresent(); } - @Test + @RepeatedTest(REPEAT_COUNT) void filteringUpdateAndGhostCheckWithNamespaceChange() { var mes = mock(ManagedInformerEventSource.class); var mim = mock(InformerManager.class); @@ -430,7 +402,7 @@ void filteringUpdateAndGhostCheckWithNamespaceChange() { assertThat(temporaryResourceCache.getResourceFromCache(resourceId)).isEmpty(); // complete the filtering update - var doneResult = temporaryResourceCache.doneEventFilterModify(resourceId, "2"); + var doneResult = temporaryResourceCache.doneEventFilterModify(resourceId); // resource was already cleaned by ghost check, so no deferred event assertThat(doneResult).isEmpty(); @@ -439,16 +411,101 @@ void filteringUpdateAndGhostCheckWithNamespaceChange() { assertThat(temporaryResourceCache.getResourceFromCache(resourceId)).isEmpty(); } + @RepeatedTest(REPEAT_COUNT) + void propagatesIntermediateEventForExternalUpdateDuringFiltering() { + // Causal-dependency fix: another controller updated the resource between our read + // and our write. The informer delivers that update during our active filter; since + // its resource version is NOT one of our own writes, it must be propagated. + withRealTemporaryResourceCache(); + + var resourceId = ResourceID.fromResource(testDeployment()); + + // first filter writes rv 4 (our own); a second concurrent filter keeps the + // active-updates window open so the event below hits the active path + var latch1 = sendForEventFilteringUpdate(4); + var latch2 = sendForEventFilteringUpdate(deploymentWithResourceVersion(4), 5); + + latch1.countDown(); + awaitCachedResourceVersion(resourceId, "4"); + + // external update with rv 3 (older than our cached rv 4) — must propagate + informerEventSource.onUpdate( + deploymentWithResourceVersion(2), deploymentWithResourceVersion(3)); + latch2.countDown(); + informerEventSource.onUpdate( + deploymentWithResourceVersion(4), deploymentWithResourceVersion(5)); + + expectHandleAddEvent(5, 2); + } + + @RepeatedTest(REPEAT_COUNT) + void doesNotPropagateIntermediateEventForOurOwnIntermediateUpdate() { + // Two consecutive own writes (rv 3 then rv 4) within an open filter window: an + // event for the older own version must be deferred since it's recognized as our own. + // A third concurrent filter keeps the active-updates window open while the event + // below is processed. + withRealTemporaryResourceCache(); + + var resourceId = ResourceID.fromResource(testDeployment()); + + var latch1 = sendForEventFilteringUpdate(3); + var latch2 = sendForEventFilteringUpdate(deploymentWithResourceVersion(3), 4); + var latch3 = sendForEventFilteringUpdate(deploymentWithResourceVersion(4), 5); + + latch1.countDown(); + awaitCachedResourceVersion(resourceId, "3"); + latch2.countDown(); + awaitCachedResourceVersion(resourceId, "4"); + + // event for our own rv 3 (older than cached rv 4) — must be deferred + informerEventSource.onUpdate( + deploymentWithResourceVersion(2), deploymentWithResourceVersion(3)); + + verify(eventHandlerMock, never()).handleEvent(any()); + + latch3.countDown(); + } + + private void awaitCachedResourceVersion(ResourceID resourceId, String resourceVersion) { + await() + .untilAsserted( + () -> + assertThat( + temporaryResourceCache + .getResourceFromCache(resourceId) + .map(d -> d.getMetadata().getResourceVersion())) + .hasValue(resourceVersion)); + } + private void assertNoEventProduced() { await() - .pollDelay(Duration.ofMillis(50)) - .timeout(Duration.ofMillis(51)) + .pollDelay(Duration.ofMillis(70)) + .timeout(Duration.ofMillis(71)) + .untilAsserted(() -> verify(informerEventSource, never()).propagateEvent(any())); + } + + private void expectHandleAddEvent(int newResourceVersion) { + await() + .atMost(Duration.ofSeconds(1)) .untilAsserted( - () -> verify(informerEventSource, never()).handleEvent(any(), any(), any(), any())); + () -> { + verify(informerEventSource, times(1)) + .handleEvent( + eq(ResourceAction.ADDED), + argThat( + newResource -> { + assertThat(newResource.getMetadata().getResourceVersion()) + .isEqualTo("" + newResourceVersion); + return true; + }), + isNull(), + any()); + }); } - private void expectHandleEvent(int newResourceVersion, int oldResourceVersion) { + private void expectHandleAddEvent(int newResourceVersion, int oldResourceVersion) { await() + .atMost(Duration.ofSeconds(1)) .untilAsserted( () -> { verify(informerEventSource, times(1)) @@ -466,7 +523,7 @@ private void expectHandleEvent(int newResourceVersion, int oldResourceVersion) { .isEqualTo("" + oldResourceVersion); return true; }), - isNull()); + any()); }); } @@ -483,6 +540,7 @@ private void withRealTemporaryResourceCache() { var mes = mock(ManagedInformerEventSource.class); var mim = mock(InformerManager.class); when(mes.manager()).thenReturn(mim); + when(mim.isWatchingNamespace(any())).thenReturn(true); when(mim.lastSyncResourceVersion(any())).thenReturn("1"); temporaryResourceCache = spy(new TemporaryResourceCache<>(true, mes)); diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/TemporaryResourceCacheTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/TemporaryResourceCacheTest.java index 9a58b83f88..edae142770 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/TemporaryResourceCacheTest.java +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/TemporaryResourceCacheTest.java @@ -25,7 +25,6 @@ import io.fabric8.kubernetes.api.model.ObjectMetaBuilder; import io.javaoperatorsdk.operator.processing.event.ResourceID; import io.javaoperatorsdk.operator.processing.event.source.ResourceAction; -import io.javaoperatorsdk.operator.processing.event.source.informer.TemporaryResourceCache.EventHandling; import static org.assertj.core.api.Assertions.assertThat; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -155,7 +154,7 @@ void eventReceivedDuringFiltering() { .isEmpty(); var doneRes = - temporaryResourceCache.doneEventFilterModify(ResourceID.fromResource(testResource), "2"); + temporaryResourceCache.doneEventFilterModify(ResourceID.fromResource(testResource)); assertThat(doneRes).isEmpty(); assertThat(temporaryResourceCache.getResourceFromCache(ResourceID.fromResource(testResource))) @@ -179,7 +178,7 @@ void newerEventDuringFiltering() { .isEmpty(); var doneRes = - temporaryResourceCache.doneEventFilterModify(ResourceID.fromResource(testResource), "2"); + temporaryResourceCache.doneEventFilterModify(ResourceID.fromResource(testResource)); assertThat(doneRes).isPresent(); assertThat(temporaryResourceCache.getResourceFromCache(ResourceID.fromResource(testResource))) @@ -197,7 +196,7 @@ void eventAfterFiltering() { .isPresent(); var doneRes = - temporaryResourceCache.doneEventFilterModify(ResourceID.fromResource(testResource), "2"); + temporaryResourceCache.doneEventFilterModify(ResourceID.fromResource(testResource)); assertThat(doneRes).isEmpty(); assertThat(temporaryResourceCache.getResourceFromCache(ResourceID.fromResource(testResource))) @@ -215,14 +214,14 @@ void putBeforeEvent() { // first ensure an event is not known var result = temporaryResourceCache.onAddOrUpdateEvent(ResourceAction.ADDED, testResource, null); - assertThat(result).isEqualTo(EventHandling.NEW); + assertThat(result).isPresent(); var nextResource = testResource(); nextResource.getMetadata().setResourceVersion("3"); temporaryResourceCache.putResource(nextResource); result = temporaryResourceCache.onAddOrUpdateEvent(ResourceAction.UPDATED, nextResource, null); - assertThat(result).isEqualTo(EventHandling.OBSOLETE); + assertThat(result).isEmpty(); } @Test @@ -232,7 +231,7 @@ void putBeforeEventWithEventFiltering() { // first ensure an event is not known var result = temporaryResourceCache.onAddOrUpdateEvent(ResourceAction.ADDED, testResource, null); - assertThat(result).isEqualTo(EventHandling.NEW); + assertThat(result).isPresent(); latestSyncVersion = RESOURCE_VERSION; var nextResource = testResource(); @@ -241,11 +240,11 @@ void putBeforeEventWithEventFiltering() { temporaryResourceCache.startEventFilteringModify(resourceId); temporaryResourceCache.putResource(nextResource); - temporaryResourceCache.doneEventFilterModify(resourceId, "3"); + temporaryResourceCache.doneEventFilterModify(resourceId); latestSyncVersion = "3"; result = temporaryResourceCache.onAddOrUpdateEvent(ResourceAction.UPDATED, nextResource, null); - assertThat(result).isEqualTo(EventHandling.OBSOLETE); + assertThat(result).isEmpty(); } @Test @@ -255,7 +254,14 @@ void putAfterEventWithEventFilteringNoPost() { // first ensure an event is not known var result = temporaryResourceCache.onAddOrUpdateEvent(ResourceAction.ADDED, testResource, null); - assertThat(result).isEqualTo(EventHandling.NEW); + assertThat(result) + .hasValueSatisfying( + v -> { + assertThat(v.getAction()).isEqualTo(ResourceAction.ADDED); + assertThat(v.getPreviousResource()).isEmpty(); + assertThat(v.getResource()).contains(testResource); + assertThat(v.getLastStateUnknow()).isNull(); + }); var nextResource = testResource(); nextResource.getMetadata().setResourceVersion("3"); @@ -265,10 +271,10 @@ void putAfterEventWithEventFilteringNoPost() { result = temporaryResourceCache.onAddOrUpdateEvent( ResourceAction.UPDATED, nextResource, testResource); - // the result is deferred - assertThat(result).isEqualTo(EventHandling.DEFER); + assertThat(result).isEmpty(); + temporaryResourceCache.putResource(nextResource); - var postEvent = temporaryResourceCache.doneEventFilterModify(resourceId, "3"); + var postEvent = temporaryResourceCache.doneEventFilterModify(resourceId); // there is no post event because the done call claimed responsibility for rv 3 assertTrue(postEvent.isEmpty()); @@ -280,20 +286,88 @@ void putAfterEventWithEventFilteringWithPost() { var resourceId = ResourceID.fromResource(testResource); temporaryResourceCache.startEventFilteringModify(resourceId); - // this should be a corner case - watch had a hard reset since the start of the + // this should be a corner case - watch had a hard reset since the start // of the update operation, such that 4 rv event is seen prior to the update // completing with the 3 rv. var nextResource = testResource(); nextResource.getMetadata().setResourceVersion("4"); var result = temporaryResourceCache.onAddOrUpdateEvent(ResourceAction.ADDED, nextResource, null); - assertThat(result).isEqualTo(EventHandling.DEFER); + assertThat(result).isEmpty(); - var postEvent = temporaryResourceCache.doneEventFilterModify(resourceId, "3"); + var postEvent = temporaryResourceCache.doneEventFilterModify(resourceId); assertTrue(postEvent.isPresent()); } + @Test + void intermediateEventPropagatedWhenNoActiveUpdate() { + // Cache holds a newer version from a prior own write; no active filter is in progress. + // An older event arriving used to be OBSOLETE; now it must be propagated as INTERMEDIATE + // so callers can react to changes that happened between read and write. + var olderEvent = testResource(); + var newer = testResource(); + newer.getMetadata().setResourceVersion("3"); + + temporaryResourceCache.putResource(newer); + assertThat(temporaryResourceCache.getResourceFromCache(ResourceID.fromResource(olderEvent))) + .isPresent(); + + var result = + temporaryResourceCache.onAddOrUpdateEvent(ResourceAction.UPDATED, olderEvent, null); + + assertThat(result) + .hasValueSatisfying( + e -> { + assertThat(e.getResource().orElseThrow()).isEqualTo(olderEvent); + assertThat(e.getPreviousResource()).isNotPresent(); + assertThat(e.getAction()).isEqualTo(ResourceAction.UPDATED); + }); + } + + @Test + void intermediateEventRecorded() { + // Causal-dependency scenario: a third party updated the resource between our read and + // our write. Its version arrives as an event but is NOT in our own resource versions, + // so it must be propagated (INTERMEDIATE), not deferred. + var external = testResource(); // rv=2 — written by another controller + var resourceId = ResourceID.fromResource(external); + + temporaryResourceCache.startEventFilteringModify(resourceId); + + var ourUpdate = testResource(); + ourUpdate.getMetadata().setResourceVersion("3"); + temporaryResourceCache.putResource(ourUpdate); + + var result = temporaryResourceCache.onAddOrUpdateEvent(ResourceAction.UPDATED, external, null); + + assertThat(result).isEmpty(); + } + + @Test + void intermediateEventDeferredWhenItIsOurOwnIntermediateUpdate() { + // Two consecutive own writes within the same filter window: the older one's event + // arrives after the newer one is cached. Because the version is recorded as our own, + // the event must be DEFERred rather than propagated. + var testResource = testResource(); + var resourceId = ResourceID.fromResource(testResource); + + temporaryResourceCache.startEventFilteringModify(resourceId); + + var ourFirst = testResource(); // rv=2 + temporaryResourceCache.putResource(ourFirst); + + var ourSecond = testResource(); + ourSecond.getMetadata().setResourceVersion("3"); + + temporaryResourceCache.startEventFilteringModify(resourceId); + temporaryResourceCache.putResource(ourSecond); + + var result = temporaryResourceCache.onAddOrUpdateEvent(ResourceAction.UPDATED, ourFirst, null); + + assertThat(result).isEmpty(); + } + @Test void rapidDeletion() { var testResource = testResource(); diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/deletionduringstatusupdate/DeletionDuringStatusUpdateCustomResource.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/deletionduringstatusupdate/DeletionDuringStatusUpdateCustomResource.java new file mode 100644 index 0000000000..5cb1170c34 --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/deletionduringstatusupdate/DeletionDuringStatusUpdateCustomResource.java @@ -0,0 +1,28 @@ +/* + * Copyright Java Operator SDK Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.javaoperatorsdk.operator.baseapi.deletionduringstatusupdate; + +import io.fabric8.kubernetes.api.model.Namespaced; +import io.fabric8.kubernetes.client.CustomResource; +import io.fabric8.kubernetes.model.annotation.Group; +import io.fabric8.kubernetes.model.annotation.ShortNames; +import io.fabric8.kubernetes.model.annotation.Version; + +@Group("sample.javaoperatorsdk") +@Version("v1") +@ShortNames("ddsu") +public class DeletionDuringStatusUpdateCustomResource + extends CustomResource implements Namespaced {} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/deletionduringstatusupdate/DeletionDuringStatusUpdateIT.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/deletionduringstatusupdate/DeletionDuringStatusUpdateIT.java new file mode 100644 index 0000000000..7574dd07b4 --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/deletionduringstatusupdate/DeletionDuringStatusUpdateIT.java @@ -0,0 +1,107 @@ +/* + * Copyright Java Operator SDK Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.javaoperatorsdk.operator.baseapi.deletionduringstatusupdate; + +import java.time.Duration; +import java.util.Collections; +import java.util.concurrent.TimeUnit; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; + +import io.fabric8.kubernetes.api.model.ObjectMetaBuilder; +import io.javaoperatorsdk.operator.junit.LocallyRunOperatorExtension; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; + +/** + * Regression test for: deletion event dropped when resource is deleted concurrently with a status + * update. + */ +class DeletionDuringStatusUpdateIT { + + static final String RESOURCE_NAME = "test-resource"; + + @RegisterExtension + LocallyRunOperatorExtension extension = + LocallyRunOperatorExtension.builder() + .withReconciler(new DeletionDuringStatusUpdateReconciler()) + .build(); + + @AfterEach + void forceCleanup() { + // If the test failed, remove the finalizer so the resource can be deleted + var res = extension.get(DeletionDuringStatusUpdateCustomResource.class, RESOURCE_NAME); + if (res != null) { + res.getMetadata().setFinalizers(Collections.emptyList()); + extension.replace(res); + extension.delete(res); + } + } + + @Test + void deletionDuringStatusUpdateTriggersCleanup() throws InterruptedException { + var reconciler = extension.getReconcilerOfType(DeletionDuringStatusUpdateReconciler.class); + + extension.create(testResource()); + + // Wait until the reconciler is inside the update operation (active-update window is open) + assertThat(reconciler.patchStartedLatch.await(30, TimeUnit.SECONDS)) + .as("reconciler should enter the patch update operation") + .isTrue(); + + // Issue delete — K8s sets deletionTimestamp while the active-update window is open + extension.delete(testResource()); + + // Wait for deletionTimestamp to be confirmed on the resource in K8s + await() + .atMost(Duration.ofSeconds(30)) + .until( + () -> { + var res = + extension.get(DeletionDuringStatusUpdateCustomResource.class, RESOURCE_NAME); + return res != null && res.isMarkedForDeletion(); + }); + + // Signal the reconciler to proceed with the actual PATCH. K8s will merge deletionTimestamp + // into the response - the deletion event (lower RV) is now deferred and will be dropped + // without the fix. + reconciler.deleteConfirmedLatch.countDown(); + + // cleanup() must be called — the deletion must not be silently lost + assertThat(reconciler.cleanupCalledLatch.await(30, TimeUnit.SECONDS)) + .as("cleanup() must be called after the status update that races with the delete") + .isTrue(); + + // Resource must eventually disappear (finalizer removed) + await() + .atMost(Duration.ofSeconds(30)) + .untilAsserted( + () -> + assertThat( + extension.get( + DeletionDuringStatusUpdateCustomResource.class, RESOURCE_NAME)) + .isNull()); + } + + DeletionDuringStatusUpdateCustomResource testResource() { + var resource = new DeletionDuringStatusUpdateCustomResource(); + resource.setMetadata(new ObjectMetaBuilder().withName(RESOURCE_NAME).build()); + return resource; + } +} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/deletionduringstatusupdate/DeletionDuringStatusUpdateReconciler.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/deletionduringstatusupdate/DeletionDuringStatusUpdateReconciler.java new file mode 100644 index 0000000000..db05321ee7 --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/deletionduringstatusupdate/DeletionDuringStatusUpdateReconciler.java @@ -0,0 +1,79 @@ +/* + * Copyright Java Operator SDK Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.javaoperatorsdk.operator.baseapi.deletionduringstatusupdate; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import io.javaoperatorsdk.operator.api.reconciler.Cleaner; +import io.javaoperatorsdk.operator.api.reconciler.Context; +import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration; +import io.javaoperatorsdk.operator.api.reconciler.DeleteControl; +import io.javaoperatorsdk.operator.api.reconciler.Reconciler; +import io.javaoperatorsdk.operator.api.reconciler.UpdateControl; + +@ControllerConfiguration +public class DeletionDuringStatusUpdateReconciler + implements Reconciler, + Cleaner { + + final CountDownLatch patchStartedLatch = new CountDownLatch(1); + final CountDownLatch deleteConfirmedLatch = new CountDownLatch(1); + final CountDownLatch cleanupCalledLatch = new CountDownLatch(1); + + @Override + public UpdateControl reconcile( + DeletionDuringStatusUpdateCustomResource resource, + Context context) + throws InterruptedException { + if (resource.isMarkedForDeletion()) { + return UpdateControl.noUpdate(); + } + + var status = new DeletionDuringStatusUpdateStatus(); + status.setReady(true); + resource.setStatus(status); + + context + .resourceOperations() + .resourcePatch( + resource, + r -> { + patchStartedLatch.countDown(); + try { + if (!deleteConfirmedLatch.await(30, TimeUnit.SECONDS)) { + throw new RuntimeException("Timed out waiting for delete confirmation"); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException(e); + } + r.getMetadata().setResourceVersion(null); + return context.getClient().resource(r).patchStatus(); + }, + context.eventSourceRetriever().getControllerEventSource()); + + return UpdateControl.noUpdate(); + } + + @Override + public DeleteControl cleanup( + DeletionDuringStatusUpdateCustomResource resource, + Context context) { + cleanupCalledLatch.countDown(); + return DeleteControl.defaultDelete(); + } +} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/deletionduringstatusupdate/DeletionDuringStatusUpdateStatus.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/deletionduringstatusupdate/DeletionDuringStatusUpdateStatus.java new file mode 100644 index 0000000000..52da516d00 --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/deletionduringstatusupdate/DeletionDuringStatusUpdateStatus.java @@ -0,0 +1,29 @@ +/* + * Copyright Java Operator SDK Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.javaoperatorsdk.operator.baseapi.deletionduringstatusupdate; + +public class DeletionDuringStatusUpdateStatus { + + private boolean ready; + + public boolean isReady() { + return ready; + } + + public void setReady(boolean ready) { + this.ready = ready; + } +}