diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/PrimaryUpdateAndCacheUtils.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/PrimaryUpdateAndCacheUtils.java index f74cd49ee7..2570c00b42 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/PrimaryUpdateAndCacheUtils.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/PrimaryUpdateAndCacheUtils.java @@ -193,7 +193,7 @@ public static

P updateAndCacheResource( .eventSourceRetriever() .getControllerEventSource() .handleRecentResourceUpdate( - ResourceID.fromResource(resourceToUpdate), updated, resourceToUpdate); + ResourceID.fromResource(resourceToUpdate), updated, resourceToUpdate, false); return updated; } catch (KubernetesClientException e) { log.trace("Exception during patch for resource: {}", resourceToUpdate); diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/ResourceOperations.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/ResourceOperations.java index b9ef475509..c3259e65b4 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/ResourceOperations.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/ResourceOperations.java @@ -79,7 +79,8 @@ public R serverSideApply(R resource) { .withForce(true) .withFieldManager(context.getControllerConfiguration().fieldManager()) .withPatchType(PatchType.SERVER_SIDE_APPLY) - .build())); + .build()), + false); } /** @@ -113,7 +114,8 @@ public R serverSideApply( .withFieldManager(context.getControllerConfiguration().fieldManager()) .withPatchType(PatchType.SERVER_SIDE_APPLY) .build()), - informerEventSource); + informerEventSource, + false); } /** @@ -144,7 +146,8 @@ public R serverSideApplyStatus(R resource) { .withForce(true) .withFieldManager(context.getControllerConfiguration().fieldManager()) .withPatchType(PatchType.SERVER_SIDE_APPLY) - .build())); + .build()), + true); } /** @@ -174,7 +177,8 @@ public P serverSideApplyPrimary(P resource) { .withFieldManager(context.getControllerConfiguration().fieldManager()) .withPatchType(PatchType.SERVER_SIDE_APPLY) .build()), - context.eventSourceRetriever().getControllerEventSource()); + context.eventSourceRetriever().getControllerEventSource(), + false); } /** @@ -205,7 +209,8 @@ public P serverSideApplyPrimaryStatus(P resource) { .withFieldManager(context.getControllerConfiguration().fieldManager()) .withPatchType(PatchType.SERVER_SIDE_APPLY) .build()), - context.eventSourceRetriever().getControllerEventSource()); + context.eventSourceRetriever().getControllerEventSource(), + true); } /** @@ -222,7 +227,7 @@ public P serverSideApplyPrimaryStatus(P resource) { * @param resource type */ public R update(R resource) { - return resourcePatch(resource, r -> context.getClient().resource(r).update()); + return resourcePatch(resource, r -> context.getClient().resource(r).update(), false); } /** @@ -245,7 +250,7 @@ public R update( return update(resource); } return resourcePatch( - resource, r -> context.getClient().resource(r).update(), informerEventSource); + resource, r -> context.getClient().resource(r).update(), informerEventSource, false); } /** @@ -262,7 +267,7 @@ public R update( * @param resource type */ public R create(R resource) { - return resourcePatch(resource, r -> context.getClient().resource(r).create()); + return resourcePatch(resource, r -> context.getClient().resource(r).create(), false); } /** @@ -285,7 +290,7 @@ public R create( return create(resource); } return resourcePatch( - resource, r -> context.getClient().resource(r).create(), informerEventSource); + resource, r -> context.getClient().resource(r).create(), informerEventSource, false); } /** @@ -304,7 +309,7 @@ public R create( * @param resource type */ public R updateStatus(R resource) { - return resourcePatch(resource, r -> context.getClient().resource(r).updateStatus()); + return resourcePatch(resource, r -> context.getClient().resource(r).updateStatus(), true); } /** @@ -325,7 +330,8 @@ public P updatePrimary(P resource) { return resourcePatch( resource, r -> context.getClient().resource(r).update(), - context.eventSourceRetriever().getControllerEventSource()); + context.eventSourceRetriever().getControllerEventSource(), + false); } /** @@ -346,7 +352,8 @@ public P updatePrimaryStatus(P resource) { return resourcePatch( resource, r -> context.getClient().resource(r).updateStatus(), - context.eventSourceRetriever().getControllerEventSource()); + context.eventSourceRetriever().getControllerEventSource(), + true); } /** @@ -367,7 +374,7 @@ public P updatePrimaryStatus(P resource) { * @param resource type */ public R jsonPatch(R resource, UnaryOperator unaryOperator) { - return resourcePatch(resource, r -> context.getClient().resource(r).edit(unaryOperator)); + return resourcePatch(resource, r -> context.getClient().resource(r).edit(unaryOperator), false); } /** @@ -388,7 +395,8 @@ public R jsonPatch(R resource, UnaryOperator unaryOpe * @param resource type */ public R jsonPatchStatus(R resource, UnaryOperator unaryOperator) { - return resourcePatch(resource, r -> context.getClient().resource(r).editStatus(unaryOperator)); + return resourcePatch( + resource, r -> context.getClient().resource(r).editStatus(unaryOperator), true); } /** @@ -410,7 +418,8 @@ public P jsonPatchPrimary(P resource, UnaryOperator

unaryOperator) { return resourcePatch( resource, r -> context.getClient().resource(r).edit(unaryOperator), - context.eventSourceRetriever().getControllerEventSource()); + context.eventSourceRetriever().getControllerEventSource(), + false); } /** @@ -432,7 +441,8 @@ public P jsonPatchPrimaryStatus(P resource, UnaryOperator

unaryOperator) { return resourcePatch( resource, r -> context.getClient().resource(r).editStatus(unaryOperator), - context.eventSourceRetriever().getControllerEventSource()); + context.eventSourceRetriever().getControllerEventSource(), + true); } /** @@ -452,7 +462,7 @@ public P jsonPatchPrimaryStatus(P resource, UnaryOperator

unaryOperator) { * @param resource type */ public R jsonMergePatch(R resource) { - return resourcePatch(resource, r -> context.getClient().resource(r).patch()); + return resourcePatch(resource, r -> context.getClient().resource(r).patch(), false); } /** @@ -471,7 +481,7 @@ public R jsonMergePatch(R resource) { * @param resource type */ public R jsonMergePatchStatus(R resource) { - return resourcePatch(resource, r -> context.getClient().resource(r).patchStatus()); + return resourcePatch(resource, r -> context.getClient().resource(r).patchStatus(), true); } /** @@ -493,7 +503,8 @@ public P jsonMergePatchPrimary(P resource) { return resourcePatch( resource, r -> context.getClient().resource(r).patch(), - context.eventSourceRetriever().getControllerEventSource()); + context.eventSourceRetriever().getControllerEventSource(), + false); } /** @@ -515,13 +526,14 @@ public P jsonMergePatchPrimaryStatus(P resource) { return resourcePatch( resource, r -> context.getClient().resource(r).patchStatus(), - context.eventSourceRetriever().getControllerEventSource()); + context.eventSourceRetriever().getControllerEventSource(), + true); } /** * Utility method to patch a resource and cache the result. Automatically discovers the event * source for the resource type and delegates to {@link #resourcePatch(HasMetadata, UnaryOperator, - * ManagedInformerEventSource)}. + * ManagedInformerEventSource,boolean)}. * * @param resource resource to patch * @param updateOperation operation to perform (update, patch, edit, etc.) @@ -530,7 +542,8 @@ public P jsonMergePatchPrimaryStatus(P resource) { * @throws IllegalStateException if no event source or multiple event sources are found */ @SuppressWarnings({"rawtypes", "unchecked"}) - public R resourcePatch(R resource, UnaryOperator updateOperation) { + public R resourcePatch( + R resource, UnaryOperator updateOperation, boolean isStatusChange) { var esList = context.eventSourceRetriever().getEventSourcesFor(resource.getClass()); if (esList.isEmpty()) { @@ -544,7 +557,8 @@ public R resourcePatch(R resource, UnaryOperator upda es.name()); } if (es instanceof ManagedInformerEventSource mes) { - return resourcePatch(resource, updateOperation, (ManagedInformerEventSource) mes); + return resourcePatch( + resource, updateOperation, (ManagedInformerEventSource) mes, isStatusChange); } else { throw new IllegalStateException( "Target event source must be a subclass off " @@ -564,8 +578,11 @@ public R resourcePatch(R resource, UnaryOperator upda * @param resource type */ public R resourcePatch( - R resource, UnaryOperator updateOperation, ManagedInformerEventSource ies) { - return ies.eventFilteringUpdateAndCacheResource(resource, updateOperation); + R resource, + UnaryOperator updateOperation, + ManagedInformerEventSource ies, + boolean isStatusChange) { + return ies.eventFilteringUpdateAndCacheResource(resource, updateOperation, isStatusChange); } /** diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/dependent/RecentOperationCacheFiller.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/dependent/RecentOperationCacheFiller.java index 9cf1fce287..dfaf787ed0 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/dependent/RecentOperationCacheFiller.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/dependent/RecentOperationCacheFiller.java @@ -21,5 +21,6 @@ public interface RecentOperationCacheFiller { void handleRecentResourceCreate(ResourceID resourceID, R resource); - void handleRecentResourceUpdate(ResourceID resourceID, R resource, R previousVersionOfResource); + void handleRecentResourceUpdate( + ResourceID resourceID, R resource, R previousVersionOfResource, boolean isStatusChange); } diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/AbstractEventSourceHolderDependentResource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/AbstractEventSourceHolderDependentResource.java index 269a9b2279..bfdca29377 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/AbstractEventSourceHolderDependentResource.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/AbstractEventSourceHolderDependentResource.java @@ -133,7 +133,7 @@ protected void onCreated(P primary, R created, Context

context) { protected void onUpdated(P primary, R updated, R actual, Context

context) { if (isCacheFillerEventSource) { recentOperationCacheFiller() - .handleRecentResourceUpdate(ResourceID.fromResource(primary), updated, actual); + .handleRecentResourceUpdate(ResourceID.fromResource(primary), updated, actual, false); } } diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/ExternalResourceCachingEventSource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/ExternalResourceCachingEventSource.java index 8a4c476443..867d420a9e 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/ExternalResourceCachingEventSource.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/ExternalResourceCachingEventSource.java @@ -221,7 +221,7 @@ public synchronized void handleRecentResourceCreate(ResourceID primaryID, R reso @Override public synchronized void handleRecentResourceUpdate( - ResourceID primaryID, R resource, R previousVersionOfResource) { + ResourceID primaryID, R resource, R previousVersionOfResource, boolean isStatusChange) { var actualValues = cache.get(primaryID); if (actualValues != null) { var resourceId = resourceIDMapper.idFor(resource); diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/EventFilterSupport.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/EventFilterSupport.java index 16dc9ebe2c..c6535f3582 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/EventFilterSupport.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/EventFilterSupport.java @@ -87,11 +87,12 @@ private Optional check( return res; } - public synchronized void addToOwnResourceVersions(ResourceID resourceId, String resourceVersion) { + public synchronized void addToOwnResourceVersions( + ResourceID resourceId, String resourceVersion, boolean isStatusChange) { var window = eventFilterWindows.get(resourceId); if (window != null) { log.debug("Recording own resourceVersion. id={}, rv={}", resourceId, resourceVersion); - window.addToOwnUpdateVersions(resourceVersion); + window.addToOwnUpdateVersions(resourceVersion, isStatusChange); } else { log.debug( "addToOwnResourceVersions: no active window for id={}, rv={} (skipped)", diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/EventFilterWindow.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/EventFilterWindow.java index 826551656e..7e45903018 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/EventFilterWindow.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/EventFilterWindow.java @@ -58,7 +58,8 @@ class EventFilterWindow { private static final Logger log = LoggerFactory.getLogger(EventFilterWindow.class); private final SortedMap relatedEvents = new TreeMap<>(); - private final SortedSet ownUpdateVersions = new TreeSet<>(); + private final SortedSet ownUpdates = new TreeSet<>(); + private final SortedMap statusChanges = new TreeMap<>(); private boolean reListOnGoing; private int activeUpdates = 0; @@ -82,7 +83,7 @@ public synchronized Optional check() { private String snapshotState() { return String.format( "relatedEvents=%s, ownResourceVersions=%s, activeUpdates=%d, reListOnGoing=%s", - relatedEvents.keySet(), ownUpdateVersions, activeUpdates, reListOnGoing); + relatedEvents.keySet(), ownUpdates, activeUpdates, reListOnGoing); } private Optional doCheck() { @@ -91,21 +92,20 @@ private Optional doCheck() { return Optional.empty(); } // cleanup events which are not related to our updates - if (activeUpdates == 0 && ownUpdateVersions.isEmpty()) { - return eventForRangeAndClear(relatedEvents, ownUpdateVersions); + if (activeUpdates == 0 && ownUpdates.isEmpty()) { + return eventForRangeAndClear(relatedEvents, ownUpdates, statusChanges); } // this is a special case that if we receive a delete event we // early clean it up, since we don't do filtering for deletes - if (ownUpdateVersions.isEmpty() - && getFirstRelatedEvent().getAction().equals(ResourceAction.DELETED)) { - return eventForRangeAndClear(relatedEvents, ownUpdateVersions); + if (ownUpdates.isEmpty() && getFirstRelatedEvent().getAction().equals(ResourceAction.DELETED)) { + return eventForRangeAndClear(relatedEvents, ownUpdates, statusChanges); } var lastEventVersion = relatedEvents.lastKey(); var numberOwnUpdatesSelected = 0; long lastOwnVersion = -1; // we find the last own update version for which we have event for // so those are the once we are going to clear our in this execution - for (long ownVersion : ownUpdateVersions) { + for (long ownVersion : ownUpdates) { if (ownVersion <= lastEventVersion) { numberOwnUpdatesSelected++; lastOwnVersion = ownVersion; @@ -120,18 +120,19 @@ && getFirstRelatedEvent().getAction().equals(ResourceAction.DELETED)) { // So If we have own version [1,3] and events [1,2,3,4] and active updates = 0 // we select all events [1,2,3,4] because for the active // update we might add own version 4. - if (numberOwnUpdatesSelected == ownUpdateVersions.size() && activeUpdates == 0) { - return eventForRangeAndClear(relatedEvents, ownUpdateVersions); + if (numberOwnUpdatesSelected == ownUpdates.size() && activeUpdates == 0) { + return eventForRangeAndClear(relatedEvents, ownUpdates, statusChanges); } else { // if we select only a subset of own updates, we select related events // up to the next own version (what is not selected). // So If we have own updates version [1,3,5] and events [1,2,3,4] // we select all those events (also 4) that happened before own version 5 // for which we don't have event yet. - if (numberOwnUpdatesSelected < ownUpdateVersions.size()) { + if (numberOwnUpdatesSelected < ownUpdates.size()) { return eventForRangeAndClear( - relatedEvents.headMap(ownUpdateVersions.tailSet(lastOwnVersion + 1).first()), - ownUpdateVersions.headSet(lastOwnVersion + 1)); + relatedEvents.headMap(ownUpdates.tailSet(lastOwnVersion + 1).first()), + ownUpdates.headSet(lastOwnVersion + 1), + statusChanges.headMap(lastOwnVersion + 1)); } else // this is essentially when we numberOwnUpdatesSelected == ownUpdateVersions.size() but // with active update > 0. In that case we: @@ -140,7 +141,8 @@ && getFirstRelatedEvent().getAction().equals(ResourceAction.DELETED)) { // update we might add own version 4. return eventForRangeAndClear( relatedEvents.headMap(lastOwnVersion + 1), - ownUpdateVersions.headSet(lastOwnVersion + 1)); + ownUpdates.headSet(lastOwnVersion + 1), + statusChanges.headMap(lastOwnVersion + 1)); } } return Optional.empty(); @@ -148,7 +150,9 @@ && getFirstRelatedEvent().getAction().equals(ResourceAction.DELETED)) { // calculates and clears events and own resources for a sorted range of events and own resources Optional eventForRangeAndClear( - SortedMap events, SortedSet ownResourceVersions) { + SortedMap events, + SortedSet ownResourceVersions, + SortedMap statusChanges) { if (events.isEmpty()) { return Optional.empty(); @@ -158,6 +162,7 @@ Optional eventForRangeAndClear( if (lastEvent.getAction() == ResourceAction.DELETED) { events.clear(); ownResourceVersions.clear(); + statusChanges.clear(); return Optional.of(lastEvent); } @@ -179,16 +184,19 @@ Optional eventForRangeAndClear( } // if all updates are related to own updates we don't return event. - // - if (events.keySet().equals(ownResourceVersions) && !isAnyEventFromReList) { + if (events.keySet().equals(ownResourceVersions) + && !isAnyEventFromReList + && eventCorrespondsToOurUpdates(events, statusChanges)) { events.clear(); ownResourceVersions.clear(); + statusChanges.clear(); return Optional.empty(); } // if only one event we return that if (events.size() == 1) { ownResourceVersions.clear(); + statusChanges.clear(); var res = Optional.of(events.values().iterator().next()); events.clear(); return res; @@ -207,9 +215,33 @@ Optional eventForRangeAndClear( null)); events.clear(); ownResourceVersions.clear(); + statusChanges.clear(); return res; } + // We only treat the events as pure echoes of our own writes if, for every UPDATED event, whether + // it is a status-only change matches how the corresponding own update at the same resource + // version was recorded. If an update event that shares a resource version with one of our own + // updates is not the same kind of change (e.g. it is a spec change while our own write was a + // status update), it cannot be the echo of our write: it is a concurrent external change that + // happened to be read back by a no-op own write, and it must surface instead of being filtered + // out. Add/delete events are creation/removal echoes of our own writes and do not carry a + // meaningful status-change distinction, so they are not used to discriminate here. + private boolean eventCorrespondsToOurUpdates( + SortedMap events, SortedMap statusChanges) { + for (var entry : events.entrySet()) { + var event = entry.getValue(); + if (event.getAction() != ResourceAction.UPDATED) { + continue; + } + var ownUpdateStatusChange = statusChanges.get(entry.getKey()); + if (ownUpdateStatusChange == null || event.isStatusChange() != ownUpdateStatusChange) { + return false; + } + } + return true; + } + private ExtendedResourceEvent getFirstRelatedEvent() { return getFirstRelatedEvent(relatedEvents); } @@ -224,14 +256,16 @@ private ExtendedResourceEvent getLastRelatedEvent(SortedMap getRelatedEvents() { } synchronized SortedSet getOwnResourceVersions() { - return ownUpdateVersions; + return ownUpdates; } } 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/ExtendedResourceEvent.java index a10945c9cf..26ed93b05a 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/ExtendedResourceEvent.java @@ -88,4 +88,27 @@ public boolean isPartOfReList() { public void setPartOfReList(boolean partOfReList) { this.partOfReList = partOfReList; } + + public boolean isStatusChange() { + if (getAction() != ResourceAction.UPDATED) { + return false; + } + // Without a previous resource we cannot observe a spec/metadata change, so we cannot claim this + // is anything but a status change. Treating it as a status change keeps such events filtered as + // our own echoes (the historical behavior) rather than surfacing them spuriously. + if (previousResource == null) { + return true; + } + var actualMeta = getResource().orElseThrow().getMetadata(); + var prevMeta = previousResource.getMetadata(); + if (!Objects.equals(prevMeta.getGeneration(), actualMeta.getGeneration())) { + return false; + } + return actualMeta.getAnnotations().equals(prevMeta.getAnnotations()) + && actualMeta.getLabels().equals(prevMeta.getLabels()) + && actualMeta.getFinalizers().equals(prevMeta.getFinalizers()) + && actualMeta.getOwnerReferences().equals(prevMeta.getOwnerReferences()) + && Objects.equals(actualMeta.getDeletionTimestamp(), prevMeta.getDeletionTimestamp()) + && actualMeta.getAdditionalProperties().equals(prevMeta.getAdditionalProperties()); + } } 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 b03a22e894..0bd52894a7 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 @@ -260,24 +260,25 @@ public Set getSecondaryResources(P primary) { @Override public void handleRecentResourceUpdate( - ResourceID resourceID, R resource, R previousVersionOfResource) { - handleRecentCreateOrUpdate(resource, previousVersionOfResource); + ResourceID resourceID, R resource, R previousVersionOfResource, boolean isStatusChange) { + handleRecentCreateOrUpdate(resource, previousVersionOfResource, isStatusChange); } @Override public void handleRecentResourceCreate(ResourceID resourceID, R resource) { - handleRecentCreateOrUpdate(resource, null); + handleRecentCreateOrUpdate(resource, null, false); } @Override protected Set cacheUpdateAndGetRelatedPrimaryIDs( - R updatedResource, R previousResource) { - return handleRecentCreateOrUpdate(updatedResource, previousResource); + R updatedResource, R previousResource, boolean isStatusChange) { + return handleRecentCreateOrUpdate(updatedResource, previousResource, isStatusChange); } - private Set handleRecentCreateOrUpdate(R newResource, R previousVersion) { + private Set handleRecentCreateOrUpdate( + R newResource, R previousVersion, boolean isStatusChange) { var relatedPrimaryIds = primaryToSecondaryIndex.onAddOrUpdate(newResource, previousVersion); - temporaryResourceCache.putResource(newResource); + temporaryResourceCache.putResource(newResource, isStatusChange); return relatedPrimaryIds; } 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 5a239a7377..c125e3b46b 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 @@ -94,7 +94,8 @@ public void changeNamespaces(Set namespaces) { * reconciliation. */ @SuppressWarnings("unchecked") - public R eventFilteringUpdateAndCacheResource(R resourceToUpdate, UnaryOperator updateMethod) { + public R eventFilteringUpdateAndCacheResource( + R resourceToUpdate, UnaryOperator updateMethod, boolean isStatusChange) { ResourceID id = ResourceID.fromResource(resourceToUpdate); log.debug("Starting event filtering and caching update for id={}", id); R updatedResource = null; @@ -102,7 +103,8 @@ public R eventFilteringUpdateAndCacheResource(R resourceToUpdate, UnaryOperator< try { temporaryResourceCache.startEventFilteringModify(id); updatedResource = updateMethod.apply(resourceToUpdate); - relatedPrimaryIds = cacheUpdateAndGetRelatedPrimaryIDs(updatedResource, resourceToUpdate); + relatedPrimaryIds = + cacheUpdateAndGetRelatedPrimaryIDs(updatedResource, resourceToUpdate, isStatusChange); log.debug( "Caching resource update successful. id={}, rv={}", id, @@ -177,13 +179,13 @@ public void onBeforeList(String lastSyncResourceVersion) { @Override public void handleRecentResourceUpdate( - ResourceID resourceID, R resource, R previousVersionOfResource) { - temporaryResourceCache.putResource(resource); + ResourceID resourceID, R resource, R previousVersionOfResource, boolean isStatusChange) { + temporaryResourceCache.putResource(resource, isStatusChange); } @Override public void handleRecentResourceCreate(ResourceID resourceID, R resource) { - temporaryResourceCache.putResource(resource); + temporaryResourceCache.putResource(resource, false); } /** @@ -195,8 +197,8 @@ public void handleRecentResourceCreate(ResourceID resourceID, R resource) { * internal to those event sources instead of leaking it into {@link RecentOperationCacheFiller}. */ protected Set cacheUpdateAndGetRelatedPrimaryIDs( - R updatedResource, R previousResource) { - handleRecentResourceUpdate(null, updatedResource, previousResource); + R updatedResource, R previousResource, boolean isStatusChange) { + handleRecentResourceUpdate(null, updatedResource, previousResource, isStatusChange); return Collections.emptySet(); } 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 a557fd1fc5..461ba20d78 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 @@ -138,7 +138,7 @@ static ExtendedResourceEvent toGenericResourceEvent( } /** 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) { + public synchronized void putResource(T newResource, boolean isStatusChange) { if (!comparableResourceVersions) { return; } @@ -166,7 +166,7 @@ public synchronized void putResource(T newResource) { var cachedResource = getResourceFromCache(resourceId).orElse(null); eventFilteringSupport.addToOwnResourceVersions( - resourceId, newResource.getMetadata().getResourceVersion()); + resourceId, newResource.getMetadata().getResourceVersion(), isStatusChange); // check against the latestResourceVersion processed by the TemporaryResourceCache // If the resource is older, then we can safely ignore. diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/reconciler/ResourceOperationsTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/reconciler/ResourceOperationsTest.java index 8d0176cd4a..24efc27178 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/reconciler/ResourceOperationsTest.java +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/reconciler/ResourceOperationsTest.java @@ -84,7 +84,7 @@ void addsFinalizer() { // Mock successful finalizer addition when(controllerEventSource.eventFilteringUpdateAndCacheResource( - any(), any(UnaryOperator.class))) + any(), any(UnaryOperator.class), anyBoolean())) .thenAnswer( invocation -> { var res = TestUtils.testCustomResource1(); @@ -99,7 +99,7 @@ void addsFinalizer() { assertThat(result.hasFinalizer(FINALIZER_NAME)).isTrue(); assertThat(result.getMetadata().getResourceVersion()).isEqualTo("2"); verify(controllerEventSource, times(1)) - .eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class)); + .eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class), anyBoolean()); } @Test @@ -111,7 +111,7 @@ void addsFinalizerWithSSA() { // Mock successful SSA finalizer addition when(controllerEventSource.eventFilteringUpdateAndCacheResource( - any(), any(UnaryOperator.class))) + any(), any(UnaryOperator.class), anyBoolean())) .thenAnswer( invocation -> { var res = TestUtils.testCustomResource1(); @@ -126,7 +126,7 @@ void addsFinalizerWithSSA() { assertThat(result.hasFinalizer(FINALIZER_NAME)).isTrue(); assertThat(result.getMetadata().getResourceVersion()).isEqualTo("2"); verify(controllerEventSource, times(1)) - .eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class)); + .eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class), anyBoolean()); } @Test @@ -139,7 +139,7 @@ void removesFinalizer() { // Mock successful finalizer removal when(controllerEventSource.eventFilteringUpdateAndCacheResource( - any(), any(UnaryOperator.class))) + any(), any(UnaryOperator.class), anyBoolean())) .thenAnswer( invocation -> { var res = TestUtils.testCustomResource1(); @@ -154,7 +154,7 @@ void removesFinalizer() { assertThat(result.hasFinalizer(FINALIZER_NAME)).isFalse(); assertThat(result.getMetadata().getResourceVersion()).isEqualTo("2"); verify(controllerEventSource, times(1)) - .eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class)); + .eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class), anyBoolean()); } @Test @@ -166,7 +166,7 @@ void retriesAddingFinalizerWithoutSSA() { // First call throws conflict, second succeeds when(controllerEventSource.eventFilteringUpdateAndCacheResource( - any(), any(UnaryOperator.class))) + any(), any(UnaryOperator.class), anyBoolean())) .thenThrow(new KubernetesClientException("Conflict", 409, null)) .thenAnswer( invocation -> { @@ -184,7 +184,7 @@ void retriesAddingFinalizerWithoutSSA() { assertThat(result).isNotNull(); assertThat(result.hasFinalizer(FINALIZER_NAME)).isTrue(); verify(controllerEventSource, times(2)) - .eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class)); + .eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class), anyBoolean()); verify(resourceOp, times(1)).get(); } @@ -198,7 +198,7 @@ void nullResourceIsGracefullyHandledOnFinalizerRemovalRetry() { // First call throws conflict when(controllerEventSource.eventFilteringUpdateAndCacheResource( - any(), any(UnaryOperator.class))) + any(), any(UnaryOperator.class), anyBoolean())) .thenThrow(new KubernetesClientException("Conflict", 409, null)); // Return null on retry (resource was deleted) @@ -207,7 +207,7 @@ void nullResourceIsGracefullyHandledOnFinalizerRemovalRetry() { resourceOperations.removeFinalizer(FINALIZER_NAME); verify(controllerEventSource, times(1)) - .eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class)); + .eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class), anyBoolean()); verify(resourceOp, times(1)).get(); } @@ -221,7 +221,7 @@ void retriesFinalizerRemovalWithFreshResource() { // First call throws unprocessable (422), second succeeds when(controllerEventSource.eventFilteringUpdateAndCacheResource( - any(), any(UnaryOperator.class))) + any(), any(UnaryOperator.class), anyBoolean())) .thenThrow(new KubernetesClientException("Unprocessable", 422, null)) .thenAnswer( invocation -> { @@ -243,7 +243,7 @@ void retriesFinalizerRemovalWithFreshResource() { assertThat(result.getMetadata().getResourceVersion()).isEqualTo("3"); assertThat(result.hasFinalizer(FINALIZER_NAME)).isFalse(); verify(controllerEventSource, times(2)) - .eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class)); + .eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class), anyBoolean()); verify(resourceOp, times(1)).get(); } @@ -261,15 +261,16 @@ void resourcePatchWithSingleEventSource() { when(context.eventSourceRetriever()).thenReturn(eventSourceRetriever); when(eventSourceRetriever.getEventSourcesFor(TestCustomResource.class)) .thenReturn(List.of(managedEventSource)); - when(managedEventSource.eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class))) + when(managedEventSource.eventFilteringUpdateAndCacheResource( + any(), any(UnaryOperator.class), anyBoolean())) .thenReturn(updatedResource); - var result = resourceOperations.resourcePatch(resource, UnaryOperator.identity()); + var result = resourceOperations.resourcePatch(resource, UnaryOperator.identity(), false); assertThat(result).isNotNull(); assertThat(result.getMetadata().getResourceVersion()).isEqualTo("2"); verify(managedEventSource, times(1)) - .eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class)); + .eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class), anyBoolean()); } @Test @@ -284,7 +285,7 @@ void resourcePatchThrowsWhenNoEventSourceFound() { var exception = assertThrows( IllegalStateException.class, - () -> resourceOperations.resourcePatch(resource, UnaryOperator.identity())); + () -> resourceOperations.resourcePatch(resource, UnaryOperator.identity(), true)); assertThat(exception.getMessage()).contains("No event source found for type"); } @@ -300,10 +301,10 @@ void resourcePatchUsesFirstEventSourceIfMultipleEventSourcesPresent() { when(eventSourceRetriever.getEventSourcesFor(TestCustomResource.class)) .thenReturn(List.of(eventSource1, eventSource2)); - resourceOperations.resourcePatch(resource, UnaryOperator.identity()); + resourceOperations.resourcePatch(resource, UnaryOperator.identity(), false); verify(eventSource1, times(1)) - .eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class)); + .eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class), anyBoolean()); } @Test @@ -319,7 +320,7 @@ void resourcePatchThrowsWhenEventSourceIsNotManagedInformer() { var exception = assertThrows( IllegalStateException.class, - () -> resourceOperations.resourcePatch(resource, UnaryOperator.identity())); + () -> resourceOperations.resourcePatch(resource, UnaryOperator.identity(), false)); assertThat(exception.getMessage()).contains("Target event source must be a subclass off"); assertThat(exception.getMessage()).contains("ManagedInformerEventSource"); diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/EventProcessorTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/EventProcessorTest.java index f7864f2f16..e82d1104f3 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/EventProcessorTest.java +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/EventProcessorTest.java @@ -54,6 +54,7 @@ import static io.javaoperatorsdk.operator.TestUtils.testCustomResource1; import static org.assertj.core.api.Assertions.assertThat; import static org.awaitility.Awaitility.await; +import static org.mockito.ArgumentMatchers.anyBoolean; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.ArgumentMatchers.isNull; import static org.mockito.Mockito.after; @@ -351,7 +352,8 @@ void notUpdatesEventSourceHandlerIfResourceUpdated() { eventProcessorWithRetry.eventProcessingFinished(executionScope, postExecutionControl); - verify(controllerEventSourceMock, times(0)).handleRecentResourceUpdate(any(), any(), any()); + verify(controllerEventSourceMock, times(0)) + .handleRecentResourceUpdate(any(), any(), any(), anyBoolean()); } @Test diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/EventFilterTestUtils.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/EventFilterTestUtils.java index 72bcac0f54..cf6b7d1ad8 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/EventFilterTestUtils.java +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/EventFilterTestUtils.java @@ -45,7 +45,8 @@ public static CountDownLatch sendForEventFilteringUpdate } catch (InterruptedException e) { throw new RuntimeException(e); } - })); + }, + true)); sendOnGoingLatch.await(); return latch; } catch (InterruptedException e) { diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/EventFilterSupportTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/EventFilterSupportTest.java index 239794e1b1..8ceb158221 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/EventFilterSupportTest.java +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/EventFilterSupportTest.java @@ -61,7 +61,7 @@ void doneEventFilterModifyEmptyWhenNoEventingDetail() { @Test void doneEventFilterModifyRemovesDetailWhenRemovable() { support.startEventFilteringModify(RESOURCE_ID); - support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION)); + support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION), true); assertThat(support.processEvent(RESOURCE_ID, updateEvent(FIRST_OWN_VERSION))).isEmpty(); var res = support.doneEventFilterModify(RESOURCE_ID); @@ -76,7 +76,7 @@ void scenarioWithSurroundingEvent() { .hasValueSatisfying(e -> assertThat(e.getAction()).isEqualTo(UPDATED)); support.startEventFilteringModify(RESOURCE_ID); - support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION)); + support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION), true); assertThat(support.processEvent(RESOURCE_ID, updateEvent(FIRST_OWN_VERSION))).isEmpty(); assertThat(support.processEvent(RESOURCE_ID, updateEvent(FIRST_OWN_VERSION + 1))).isEmpty(); @@ -101,7 +101,7 @@ void processEventPropagatesWhenNoEventingDetail() { @Test void processEventHoldsOwnEcho() { support.startEventFilteringModify(RESOURCE_ID); - support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION)); + support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION), true); var res = support.processEvent(RESOURCE_ID, updateEvent(FIRST_OWN_VERSION)); @@ -111,7 +111,7 @@ void processEventHoldsOwnEcho() { @Test void processEventEmitsSynthForForeignEvent() { support.startEventFilteringModify(RESOURCE_ID); - support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION)); + support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION), true); support.processEvent(RESOURCE_ID, updateEvent(FIRST_OWN_VERSION - 1)); var res = support.processEvent(RESOURCE_ID, updateEvent(FIRST_OWN_VERSION)); @@ -122,7 +122,7 @@ void processEventEmitsSynthForForeignEvent() { @Test void processEventEmitsAddedForeignVerbatim() { support.startEventFilteringModify(RESOURCE_ID); - support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION)); + support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION), true); var addEvent = addEvent(FIRST_OWN_VERSION); var updateEvent = addEvent(FIRST_OWN_VERSION + 1); support.processEvent(RESOURCE_ID, addEvent); @@ -137,7 +137,7 @@ void processEventEmitsAddedForeignVerbatim() { @Test void addToOwnResourceVersionsIsNoOpWithoutEventingDetail() { - support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION)); + support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION), true); assertThat(support.isActiveUpdateFor(RESOURCE_ID)).isFalse(); } @@ -164,7 +164,7 @@ void handleGhostResourceRemovalIsNoOpForUnknownResource() { @Test void fullLifecycleOwnWriteOnlyEmitsNothingAndCleansUp() { support.startEventFilteringModify(RESOURCE_ID); - support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION)); + support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION), true); assertThat(support.processEvent(RESOURCE_ID, updateEvent(FIRST_OWN_VERSION))).isEmpty(); var res = support.doneEventFilterModify(RESOURCE_ID); @@ -176,7 +176,7 @@ void fullLifecycleOwnWriteOnlyEmitsNothingAndCleansUp() { @Test void fullLifecycleForeignBeforeOwnEchoEmitsSynth() { support.startEventFilteringModify(RESOURCE_ID); - support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION)); + support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION), true); var foreign = updateEvent(FIRST_OWN_VERSION - 1); assertThat(support.processEvent(RESOURCE_ID, foreign)).isEmpty(); @@ -190,7 +190,7 @@ void fullLifecycleForeignBeforeOwnEchoEmitsSynth() { @Test void oneOwnVersionNoEvent() { support.startEventFilteringModify(RESOURCE_ID); - support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION)); + support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION), true); assertThat(support.doneEventFilterModify(RESOURCE_ID)).isEmpty(); // own RV recorded but no echo arrived yet → window stays @@ -202,7 +202,7 @@ void oneOwnVersionNoEvent() { @Test void oneOwnVersionEventReceivedEventForIt() { support.startEventFilteringModify(RESOURCE_ID); - support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION)); + support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION), true); assertThat(support.processEvent(RESOURCE_ID, updateEvent(FIRST_OWN_VERSION))).isEmpty(); assertThat(support.isActiveUpdateFor(RESOURCE_ID)).isTrue(); @@ -214,7 +214,7 @@ void oneOwnVersionEventReceivedEventForIt() { @Test void receivedAsFirstAddEventReturnTheSameEventIfThatIsOnlyRelevant() { support.startEventFilteringModify(RESOURCE_ID); - support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION)); + support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION), true); assertThat(support.processEvent(RESOURCE_ID, addEvent(FIRST_OWN_VERSION))).isEmpty(); } @@ -222,7 +222,7 @@ void receivedAsFirstAddEventReturnTheSameEventIfThatIsOnlyRelevant() { @Test void oneOwnVersionAdditionalEventReceivedBeforeIt() { support.startEventFilteringModify(RESOURCE_ID); - support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION)); + support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION), true); assertThat(support.processEvent(RESOURCE_ID, updateEvent(FIRST_OWN_VERSION - 1))).isEmpty(); assertThat(support.processEvent(RESOURCE_ID, updateEvent(FIRST_OWN_VERSION))).isPresent(); @@ -235,9 +235,9 @@ void oneOwnVersionAdditionalEventReceivedBeforeIt() { @Test void twoOwnVersionEventReceivedEventOnlyForFirstThenForSecond() { support.startEventFilteringModify(RESOURCE_ID); - support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION)); + support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION), true); support.startEventFilteringModify(RESOURCE_ID); - support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION + 1)); + support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION + 1), true); assertThat(support.processEvent(RESOURCE_ID, updateEvent(FIRST_OWN_VERSION))).isEmpty(); var window = support.getEventFilterWindows().get(RESOURCE_ID); @@ -255,9 +255,9 @@ void twoOwnVersionEventReceivedEventOnlyForFirstThenForSecond() { @Test void twoOwnVersionEventReceivedOne() { support.startEventFilteringModify(RESOURCE_ID); - support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION)); + support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION), true); support.startEventFilteringModify(RESOURCE_ID); - support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION + 1)); + support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION + 1), true); assertThat(support.processEvent(RESOURCE_ID, updateEvent(FIRST_OWN_VERSION))).isEmpty(); @@ -273,7 +273,7 @@ void twoOwnVersionEventReceivedOne() { @Test void receivedAddEventAfterOurUpdate() { support.startEventFilteringModify(RESOURCE_ID); - support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION)); + support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION), true); assertThat(support.processEvent(RESOURCE_ID, addEvent(FIRST_OWN_VERSION + 1))).isEmpty(); assertThat(support.doneEventFilterModify(RESOURCE_ID)) @@ -284,7 +284,7 @@ void receivedAddEventAfterOurUpdate() { @Test void receivedAddEventAfterOurUpdateDone() { support.startEventFilteringModify(RESOURCE_ID); - support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION)); + support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION), true); assertThat(support.doneEventFilterModify(RESOURCE_ID)).isEmpty(); // Window remains because own=[5] is non-empty. Late ADDED arrives after done. @@ -317,7 +317,7 @@ void propagateEventIfNoOwnResourceAndNoActiveUpdate() { @Test void receiveEventAfterEventForOwnUpdate() { support.startEventFilteringModify(RESOURCE_ID); - support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION + 1)); + support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION + 1), true); assertThat(support.processEvent(RESOURCE_ID, updateEvent(FIRST_OWN_VERSION))).isEmpty(); assertThat(support.processEvent(RESOURCE_ID, updateEvent(FIRST_OWN_VERSION + 1))) @@ -332,7 +332,7 @@ void receiveEventAfterEventForOwnUpdate() { @Test void doNotIncludeAfterEventForFirstOwnUpdateIfOtherOwnUpdateIsActive() { support.startEventFilteringModify(RESOURCE_ID); - support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION + 1)); + support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION + 1), true); support.startEventFilteringModify(RESOURCE_ID); assertThat(support.processEvent(RESOURCE_ID, updateEvent(FIRST_OWN_VERSION))).isEmpty(); @@ -340,7 +340,7 @@ void doNotIncludeAfterEventForFirstOwnUpdateIfOtherOwnUpdateIsActive() { assertThat(support.processEvent(RESOURCE_ID, updateEvent(FIRST_OWN_VERSION + 2))).isEmpty(); support.doneEventFilterModify(RESOURCE_ID); - support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION + 2)); + support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION + 2), true); support.doneEventFilterModify(RESOURCE_ID); assertThat(support.isActiveUpdateFor(RESOURCE_ID)).isFalse(); @@ -349,9 +349,9 @@ void doNotIncludeAfterEventForFirstOwnUpdateIfOtherOwnUpdateIsActive() { @Test void assertMultipleUpdatesAndIntermediateEventBetween() { support.startEventFilteringModify(RESOURCE_ID); - support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION)); + support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION), true); support.startEventFilteringModify(RESOURCE_ID); - support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION + 2)); + support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION + 2), true); assertThat(support.processEvent(RESOURCE_ID, updateEvent(FIRST_OWN_VERSION))).isEmpty(); assertThat(support.processEvent(RESOURCE_ID, updateEvent(FIRST_OWN_VERSION + 1))).isEmpty(); @@ -366,9 +366,9 @@ void assertMultipleUpdatesAndIntermediateEventBetween() { @Test void receiveIntermediateBetweenTwoOwnUpdates() { support.startEventFilteringModify(RESOURCE_ID); - support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION)); + support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION), true); support.startEventFilteringModify(RESOURCE_ID); - support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION + 2)); + support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION + 2), true); assertThat(support.processEvent(RESOURCE_ID, updateEvent(FIRST_OWN_VERSION))).isEmpty(); assertThat(support.processEvent(RESOURCE_ID, updateEvent(FIRST_OWN_VERSION + 1))).isEmpty(); @@ -387,7 +387,7 @@ void receiveIntermediateBetweenTwoOwnUpdates() { @Test void deleteEventAsLastEvent_simpleCase() { support.startEventFilteringModify(RESOURCE_ID); - support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION)); + support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION), true); assertThat(support.processEvent(RESOURCE_ID, deleteEvent(FIRST_OWN_VERSION))) .hasValueSatisfying(e -> assertThat(e.getAction()).isEqualTo(DELETED)); @@ -400,7 +400,7 @@ void deleteEventAsLastEvent_simpleCase() { @Test void deleteEventBeforeOurUpdate() { support.startEventFilteringModify(RESOURCE_ID); - support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION)); + support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION), true); assertThat(support.processEvent(RESOURCE_ID, deleteEvent(FIRST_OWN_VERSION - 1))).isEmpty(); assertThat(support.processEvent(RESOURCE_ID, addEvent(FIRST_OWN_VERSION))).isEmpty(); @@ -412,7 +412,7 @@ void deleteEventBeforeOurUpdate() { @Test void deleteEventOnMiddleOfOwnUpdate() { support.startEventFilteringModify(RESOURCE_ID); - support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION + 2)); + support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION + 2), true); assertThat(support.processEvent(RESOURCE_ID, updateEvent(FIRST_OWN_VERSION))).isEmpty(); assertThat(support.processEvent(RESOURCE_ID, deleteEvent(FIRST_OWN_VERSION + 1))).isEmpty(); @@ -426,7 +426,7 @@ void deleteEventOnMiddleOfOwnUpdate() { @Test void deleteEventAsAdditionalEventAfterOwnUpdates() { support.startEventFilteringModify(RESOURCE_ID); - support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION)); + support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION), true); assertThat(support.processEvent(RESOURCE_ID, updateEvent(FIRST_OWN_VERSION))).isEmpty(); assertThat(support.processEvent(RESOURCE_ID, deleteEvent(FIRST_OWN_VERSION + 1))) @@ -439,7 +439,7 @@ void deleteEventAsAdditionalEventAfterOwnUpdates() { @Test void additionalDeleteEvent() { support.startEventFilteringModify(RESOURCE_ID); - support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION + 1)); + support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION + 1), true); assertThat(support.processEvent(RESOURCE_ID, updateEvent(FIRST_OWN_VERSION))).isEmpty(); assertThat(support.processEvent(RESOURCE_ID, updateEvent(FIRST_OWN_VERSION + 1))) @@ -457,14 +457,14 @@ void additionalDeleteEvent() { @Test void deleteEventInMiddleTwoUpdates() { support.startEventFilteringModify(RESOURCE_ID); - support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION)); + support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION), true); assertThat(support.processEvent(RESOURCE_ID, updateEvent(FIRST_OWN_VERSION))).isEmpty(); assertThat(support.processEvent(RESOURCE_ID, deleteEvent(FIRST_OWN_VERSION + 1))) .hasValueSatisfying(e -> assertThat(e.getAction()).isEqualTo(DELETED)); support.startEventFilteringModify(RESOURCE_ID); - support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION + 2)); + support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION + 2), true); assertThat(support.processEvent(RESOURCE_ID, addEvent(FIRST_OWN_VERSION + 2))).isEmpty(); assertThat(support.doneEventFilterModify(RESOURCE_ID)).isEmpty(); @@ -476,9 +476,9 @@ void deleteEventInMiddleTwoUpdates() { @Test void deleteEventAfterTwoUpdates() { support.startEventFilteringModify(RESOURCE_ID); - support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION)); + support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION), true); support.startEventFilteringModify(RESOURCE_ID); - support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION + 1)); + support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION + 1), true); assertThat(support.processEvent(RESOURCE_ID, updateEvent(FIRST_OWN_VERSION))).isEmpty(); assertThat(support.processEvent(RESOURCE_ID, updateEvent(FIRST_OWN_VERSION + 1))).isEmpty(); @@ -494,7 +494,7 @@ void deleteEventAfterTwoUpdates() { @Test void reListBeforeUpdateStarted() { support.startEventFilteringModify(RESOURCE_ID); - support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION)); + support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION), true); support.setStartingReList(); assertThat(support.processEvent(RESOURCE_ID, updateEvent(FIRST_OWN_VERSION))) .hasValueSatisfying(e -> assertThat(e.getAction()).isEqualTo(UPDATED)); @@ -508,7 +508,7 @@ void reListBeforeUpdateStarted() { @Test void reListHappensAfterUpdate() { support.startEventFilteringModify(RESOURCE_ID); - support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION)); + support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION), true); assertThat(support.processEvent(RESOURCE_ID, updateEvent(FIRST_OWN_VERSION))).isEmpty(); support.setStartingReList(); assertThat(support.processEvent(RESOURCE_ID, updateEvent(FIRST_OWN_VERSION + 1))).isEmpty(); @@ -522,11 +522,11 @@ void reListHappensAfterUpdate() { @Test void reListBetweenTwoUpdates() { support.startEventFilteringModify(RESOURCE_ID); - support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION)); + support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION), true); assertThat(support.processEvent(RESOURCE_ID, updateEvent(FIRST_OWN_VERSION))).isEmpty(); support.startEventFilteringModify(RESOURCE_ID); - support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION + 1)); + support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION + 1), true); support.setStartingReList(); assertThat(support.processEvent(RESOURCE_ID, updateEvent(FIRST_OWN_VERSION + 1))) .hasValueSatisfying(e -> assertThat(e.getAction()).isEqualTo(UPDATED)); @@ -544,7 +544,7 @@ void ghostRemovalDuringActiveUpdateClearsWindow() { // A ghost cleanup arriving while an own write is in flight wipes the window // outright (current semantics — see EventFilterSupport.handleGhostResourceRemoval). support.startEventFilteringModify(RESOURCE_ID); - support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION)); + support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION), true); support.processEvent(RESOURCE_ID, updateEvent(FIRST_OWN_VERSION - 1)); assertThat(support.isActiveUpdateFor(RESOURCE_ID)).isTrue(); @@ -560,7 +560,7 @@ void ghostRemovalDuringActiveUpdateClearsWindow() { void ghostRemovalAfterEventsHaveBeenHeldDropsThem() { // Held foreign events that haven't yet been emitted are discarded by ghost removal. support.startEventFilteringModify(RESOURCE_ID); - support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION)); + support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION), true); assertThat(support.processEvent(RESOURCE_ID, updateEvent(FIRST_OWN_VERSION - 2))).isEmpty(); assertThat(support.processEvent(RESOURCE_ID, updateEvent(FIRST_OWN_VERSION - 1))).isEmpty(); var window = support.getEventFilterWindows().get(RESOURCE_ID); @@ -576,7 +576,7 @@ void ghostRemovalDuringReListAffectsOnlyTargetResource() { // Ghost removal targeting one resource doesn't disturb a parallel reList window // for another resource. support.startEventFilteringModify(RESOURCE_ID); - support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION)); + support.addToOwnResourceVersions(RESOURCE_ID, s(FIRST_OWN_VERSION), true); support.startEventFilteringModify(OTHER_RESOURCE_ID); support.setStartingReList(); 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 367c4fa4f3..01172a7591 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 @@ -37,7 +37,7 @@ class EventFilterWindowTest { @Test void oneOwnVersionNoEvent() { eventFilterWindow.increaseActiveUpdates(); - eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION)); + eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION), true); assertThat(eventFilterWindow.check()).isEmpty(); assertThat(eventFilterWindow.canBeRemoved()).isFalse(); @@ -49,7 +49,7 @@ void oneOwnVersionNoEvent() { @Test void oneOwnVersionEventReceivedEventForIt() { eventFilterWindow.increaseActiveUpdates(); - eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION)); + eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION), true); eventFilterWindow.addRelatedEvent(updateEvent(FIRST_OWN_VERSION)); // check also cleans up the current since we received event for our own resource @@ -63,7 +63,7 @@ void oneOwnVersionEventReceivedEventForIt() { @Test void receivedAsFirstAddEventReturnTheSameEventIfThatIsOnlyRelevant() { eventFilterWindow.increaseActiveUpdates(); - eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION)); + eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION), true); eventFilterWindow.addRelatedEvent(addEvent(FIRST_OWN_VERSION)); assertThat(eventFilterWindow.check()).isEmpty(); @@ -72,7 +72,7 @@ void receivedAsFirstAddEventReturnTheSameEventIfThatIsOnlyRelevant() { @Test void oneOwnVersionAdditionalEventReceivedBeforeIt() { eventFilterWindow.increaseActiveUpdates(); - eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION)); + eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION), true); eventFilterWindow.addRelatedEvent(updateEvent(FIRST_OWN_VERSION - 1)); eventFilterWindow.addRelatedEvent(updateEvent(FIRST_OWN_VERSION)); @@ -88,9 +88,9 @@ void oneOwnVersionAdditionalEventReceivedBeforeIt() { @Test void twoOwnVersionEventReceivedEventOnlyForFirstThenForSecond() { eventFilterWindow.increaseActiveUpdates(); - eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION)); + eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION), true); eventFilterWindow.increaseActiveUpdates(); - eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION + 1)); + eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION + 1), true); eventFilterWindow.addRelatedEvent(updateEvent(FIRST_OWN_VERSION)); @@ -114,9 +114,9 @@ void twoOwnVersionEventReceivedEventOnlyForFirstThenForSecond() { @Test void twoOwnVersionEventReceivedOne() { eventFilterWindow.increaseActiveUpdates(); - eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION)); + eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION), true); eventFilterWindow.increaseActiveUpdates(); - eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION + 1)); + eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION + 1), true); eventFilterWindow.addRelatedEvent(updateEvent(FIRST_OWN_VERSION)); @@ -135,7 +135,7 @@ void twoOwnVersionEventReceivedOne() { @Test void receivedAddEventAfterOurUpdate() { eventFilterWindow.increaseActiveUpdates(); - eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION)); + eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION), true); eventFilterWindow.addRelatedEvent(addEvent(FIRST_OWN_VERSION + 1)); eventFilterWindow.decreaseActiveUpdates(); @@ -150,7 +150,7 @@ void receivedAddEventAfterOurUpdate() { @Test void receivedAddEventAfterOurUpdateDone() { eventFilterWindow.increaseActiveUpdates(); - eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION)); + eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION), true); eventFilterWindow.decreaseActiveUpdates(); eventFilterWindow.addRelatedEvent(addEvent(FIRST_OWN_VERSION + 1)); assertThat(eventFilterWindow.check()) @@ -187,7 +187,7 @@ void propagateEventIfNoOwnResourceAndNoActiveUpdate() { @Test void receiveEventAfterEventForOwnUpdate() { eventFilterWindow.increaseActiveUpdates(); - eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION + 1)); + eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION + 1), true); eventFilterWindow.addRelatedEvent(updateEvent(FIRST_OWN_VERSION)); eventFilterWindow.addRelatedEvent(updateEvent(FIRST_OWN_VERSION + 1)); @@ -208,7 +208,7 @@ void receiveEventAfterEventForOwnUpdate() { @Test void doNotIncludeAfterEventForFirstOwnUpdateIfOtherOwnUpdateIsActive() { eventFilterWindow.increaseActiveUpdates(); - eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION + 1)); + eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION + 1), true); eventFilterWindow.increaseActiveUpdates(); @@ -224,7 +224,7 @@ void doNotIncludeAfterEventForFirstOwnUpdateIfOtherOwnUpdateIsActive() { eventFilterWindow.decreaseActiveUpdates(); assertThat(eventFilterWindow.getRelatedEvents()).isNotEmpty(); - eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION + 2)); + eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION + 2), true); assertThat(eventFilterWindow.check()).isEmpty(); @@ -237,9 +237,9 @@ void doNotIncludeAfterEventForFirstOwnUpdateIfOtherOwnUpdateIsActive() { @Test void assertMultipleUpdatesAndIntermediateEventBetween() { eventFilterWindow.increaseActiveUpdates(); - eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION)); + eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION), true); eventFilterWindow.increaseActiveUpdates(); - eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION + 2)); + eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION + 2), true); eventFilterWindow.addRelatedEvent(updateEvent(FIRST_OWN_VERSION)); eventFilterWindow.addRelatedEvent(updateEvent(FIRST_OWN_VERSION + 1)); @@ -259,9 +259,9 @@ void assertMultipleUpdatesAndIntermediateEventBetween() { @Test void receiveIntermediateBetweenTwoOwnUpdates() { eventFilterWindow.increaseActiveUpdates(); - eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION)); + eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION), true); eventFilterWindow.increaseActiveUpdates(); - eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION + 2)); + eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION + 2), true); eventFilterWindow.addRelatedEvent(updateEvent(FIRST_OWN_VERSION)); eventFilterWindow.addRelatedEvent(updateEvent(FIRST_OWN_VERSION + 1)); @@ -287,7 +287,7 @@ void receiveIntermediateBetweenTwoOwnUpdates() { @Test void deleteEventAsLastEvent_simpleCase() { eventFilterWindow.increaseActiveUpdates(); - eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION)); + eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION), true); eventFilterWindow.addRelatedEvent(deleteEvent(FIRST_OWN_VERSION)); assertThat(eventFilterWindow.check()).hasValueSatisfying(this::assertDeleteEvent); assertThat(eventFilterWindow.canBeRemoved()).isFalse(); @@ -300,7 +300,7 @@ void deleteEventAsLastEvent_simpleCase() { @Test void deleteEventBeforeOurUpdate() { eventFilterWindow.increaseActiveUpdates(); - eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION)); + eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION), true); eventFilterWindow.addRelatedEvent(deleteEvent(FIRST_OWN_VERSION - 1)); eventFilterWindow.addRelatedEvent(addEvent(FIRST_OWN_VERSION)); @@ -315,7 +315,7 @@ void deleteEventBeforeOurUpdate() { @Test void deleteEventOnMiddleOfOwnUpdate() { eventFilterWindow.increaseActiveUpdates(); - eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION + 2)); + eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION + 2), true); eventFilterWindow.addRelatedEvent(updateEvent(FIRST_OWN_VERSION)); eventFilterWindow.addRelatedEvent(deleteEvent(FIRST_OWN_VERSION + 1)); eventFilterWindow.addRelatedEvent(addEvent(FIRST_OWN_VERSION + 2)); @@ -332,7 +332,7 @@ void deleteEventOnMiddleOfOwnUpdate() { @Test void deleteEventAsAdditionalEventAfterOwnUpdates() { eventFilterWindow.increaseActiveUpdates(); - eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION)); + eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION), true); eventFilterWindow.addRelatedEvent(updateEvent(FIRST_OWN_VERSION)); eventFilterWindow.addRelatedEvent(deleteEvent(FIRST_OWN_VERSION + 1)); @@ -349,7 +349,7 @@ void deleteEventAsAdditionalEventAfterOwnUpdates() { @Test void additionalDeleteEvent() { eventFilterWindow.increaseActiveUpdates(); - eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION + 1)); + eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION + 1), true); eventFilterWindow.addRelatedEvent(updateEvent(FIRST_OWN_VERSION)); eventFilterWindow.addRelatedEvent(updateEvent(FIRST_OWN_VERSION + 1)); eventFilterWindow.addRelatedEvent(deleteEvent(FIRST_OWN_VERSION + 2)); @@ -370,7 +370,7 @@ void additionalDeleteEvent() { @Test void additionalEventAndDeleteEvent() { eventFilterWindow.increaseActiveUpdates(); - eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION)); + eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION), true); eventFilterWindow.addRelatedEvent(updateEvent(FIRST_OWN_VERSION)); eventFilterWindow.addRelatedEvent(updateEvent(FIRST_OWN_VERSION + 1)); @@ -387,7 +387,7 @@ void additionalEventAndDeleteEvent() { @Test void deleteEventInMiddleTwoUpdates() { eventFilterWindow.increaseActiveUpdates(); - eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION)); + eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION), true); eventFilterWindow.addRelatedEvent(updateEvent(FIRST_OWN_VERSION)); assertThat(eventFilterWindow.check()).isEmpty(); @@ -400,7 +400,7 @@ void deleteEventInMiddleTwoUpdates() { .hasValueSatisfying(e -> assertDeleteEvent(e, FIRST_OWN_VERSION + 1)); assertEmptyState(); - eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION + 2)); + eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION + 2), true); eventFilterWindow.addRelatedEvent(addEvent(FIRST_OWN_VERSION + 2)); // delete event should be skipped in these cases and taking directly the last event assertThat(eventFilterWindow.check()).isEmpty(); @@ -415,10 +415,10 @@ void deleteEventInMiddleTwoUpdates() { @Test void deleteEventAfterTwoUpdates() { eventFilterWindow.increaseActiveUpdates(); - eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION)); + eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION), true); eventFilterWindow.increaseActiveUpdates(); - eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION + 1)); + eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION + 1), true); eventFilterWindow.addRelatedEvent(updateEvent(FIRST_OWN_VERSION)); eventFilterWindow.addRelatedEvent(updateEvent(FIRST_OWN_VERSION + 1)); @@ -439,10 +439,10 @@ void deleteEventAfterTwoUpdates() { @Test void deleteEventAfterTwoUpdatesFinished() { eventFilterWindow.increaseActiveUpdates(); - eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION)); + eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION), true); eventFilterWindow.increaseActiveUpdates(); - eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION + 1)); + eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION + 1), true); eventFilterWindow.addRelatedEvent(updateEvent(FIRST_OWN_VERSION)); eventFilterWindow.addRelatedEvent(updateEvent(FIRST_OWN_VERSION + 1)); @@ -463,7 +463,7 @@ void deleteEventAfterTwoUpdatesFinished() { @Test void reListBeforeUpdateStarted() { eventFilterWindow.increaseActiveUpdates(); - eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION)); + eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION), true); eventFilterWindow.setReListStarted(); eventFilterWindow.addRelatedEvent(updateEvent(FIRST_OWN_VERSION)); eventFilterWindow.setReListFinished(); @@ -479,7 +479,7 @@ void reListBeforeUpdateStarted() { @Test void reListHappensAfterUpdate() { eventFilterWindow.increaseActiveUpdates(); - eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION)); + eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION), true); eventFilterWindow.addRelatedEvent(updateEvent(FIRST_OWN_VERSION)); eventFilterWindow.setReListStarted(); eventFilterWindow.addRelatedEvent(updateEvent(FIRST_OWN_VERSION + 1)); @@ -497,11 +497,11 @@ void reListHappensAfterUpdate() { @Test void reListBetweenTwoUpdates() { eventFilterWindow.increaseActiveUpdates(); - eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION)); + eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION), true); eventFilterWindow.addRelatedEvent(updateEvent(FIRST_OWN_VERSION)); eventFilterWindow.increaseActiveUpdates(); - eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION + 1)); + eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION + 1), true); eventFilterWindow.setReListStarted(); eventFilterWindow.addRelatedEvent(updateEvent(FIRST_OWN_VERSION + 1)); eventFilterWindow.setReListFinished(); @@ -525,7 +525,7 @@ void combinedCaseWithEarlyEvent() { // the window wedges (canRemoved stays false because relatedEvents is not // empty) and the reconciler never learns about the foreign change. eventFilterWindow.increaseActiveUpdates(); - eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION)); + eventFilterWindow.addToOwnUpdateVersions(s(FIRST_OWN_VERSION), true); eventFilterWindow.addRelatedEvent(updateEvent(FIRST_OWN_VERSION - 2)); // Held while the write is in flight. @@ -545,6 +545,50 @@ void combinedCaseWithEarlyEvent() { assertThat(eventFilterWindow.canBeRemoved()).isTrue(); } + @Test + void foreignUpdateCollidingWithNoOpOwnWriteVersionMustNotBeSwallowed() { + long foreignVersion = FIRST_OWN_VERSION; // rv of the generation 3 external change + + // generation 2 reconciliation starts + eventFilterWindow.increaseActiveUpdates(); + + eventFilterWindow.addRelatedEvent(specChangeUpdateEvent(foreignVersion)); + assertThat(eventFilterWindow.check()).isEmpty(); + + eventFilterWindow.addToOwnUpdateVersions(s(foreignVersion), true); + + eventFilterWindow.decreaseActiveUpdates(); + + // the foreign generation 3 change must surface, not be swallowed as an echo of our write + assertThat(eventFilterWindow.check()) + .as("foreign update sharing an rv with a no-op own write must still surface") + .hasValueSatisfying(e -> assertUpdateEvent(e, foreignVersion)); + + assertThat(eventFilterWindow.canBeRemoved()).isTrue(); + assertEmptyState(); + } + + @Test + void foreignUpdateCollidingWithNoOpOwnWriteVersionMustNotBeSwallowed2() { + long foreignVersion = FIRST_OWN_VERSION; // rv of the generation 3 external change + + eventFilterWindow.increaseActiveUpdates(); + assertThat(eventFilterWindow.check()).isEmpty(); + + eventFilterWindow.addToOwnUpdateVersions(s(foreignVersion), true); + + eventFilterWindow.addRelatedEvent(specChangeUpdateEvent(foreignVersion)); + + eventFilterWindow.decreaseActiveUpdates(); + + assertThat(eventFilterWindow.check()) + .as("foreign update sharing an rv with a no-op own write must still surface") + .hasValueSatisfying(e -> assertUpdateEvent(e, foreignVersion)); + + assertThat(eventFilterWindow.canBeRemoved()).isTrue(); + assertEmptyState(); + } + void assertUpdateEvent(ExtendedResourceEvent event, Long resourceVersion) { assertUpdateEvent(event, resourceVersion, resourceVersion - 1); } @@ -584,6 +628,13 @@ ExtendedResourceEvent updateEvent(long version) { UPDATED, testResource(version), testResource(version - 1), null); } + // An update that also bumps the generation, i.e. an external spec change (not an own + // status/metadata write, which never changes the generation). + ExtendedResourceEvent specChangeUpdateEvent(long version) { + return new ExtendedResourceEvent( + UPDATED, testResource(version, 2L), testResource(version - 1, 1L), null); + } + ExtendedResourceEvent addEvent(long version) { return new ExtendedResourceEvent(ADDED, testResource(version), null, null); } @@ -593,12 +644,17 @@ ExtendedResourceEvent deleteEvent(long version) { } ConfigMap testResource(Long version) { + return testResource(version, null); + } + + ConfigMap testResource(Long version, Long generation) { var cm = new ConfigMap(); cm.setMetadata( new ObjectMetaBuilder() .withName(RESOURCE_ID.getName()) .withNamespace(RESOURCE_ID.getNamespace().orElseThrow()) .withResourceVersion(version.toString()) + .withGeneration(generation) .build()); return cm; } diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/ExtendedResourceEventTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/ExtendedResourceEventTest.java new file mode 100644 index 0000000000..2306f6b979 --- /dev/null +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/ExtendedResourceEventTest.java @@ -0,0 +1,90 @@ +/* + * 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.processing.event.source.informer; + +import org.junit.jupiter.api.Test; + +import io.fabric8.kubernetes.api.model.ConfigMap; +import io.fabric8.kubernetes.api.model.ObjectMetaBuilder; + +import static io.javaoperatorsdk.operator.processing.event.source.ResourceAction.ADDED; +import static io.javaoperatorsdk.operator.processing.event.source.ResourceAction.DELETED; +import static io.javaoperatorsdk.operator.processing.event.source.ResourceAction.UPDATED; +import static org.assertj.core.api.Assertions.assertThat; + +class ExtendedResourceEventTest { + + @Test + void statusChangeWhenOnlyResourceVersionChangedAndGenerationAndMetadataUnchanged() { + var event = + new ExtendedResourceEvent( + UPDATED, resource("2", 1L, "app"), resource("1", 1L, "app"), null); + + assertThat(event.isStatusChange()).isTrue(); + } + + @Test + void notStatusChangeWhenGenerationChanged() { + var event = + new ExtendedResourceEvent( + UPDATED, resource("2", 2L, "app"), resource("1", 1L, "app"), null); + + assertThat(event.isStatusChange()).isFalse(); + } + + @Test + void notStatusChangeWhenLabelsChanged() { + var event = + new ExtendedResourceEvent( + UPDATED, resource("2", 1L, "changed"), resource("1", 1L, "app"), null); + + assertThat(event.isStatusChange()).isFalse(); + } + + @Test + void statusChangeWhenUpdatedEventHasNoPreviousResource() { + var event = new ExtendedResourceEvent(UPDATED, resource("2", 1L, "app"), null, null); + + assertThat(event.isStatusChange()).isTrue(); + } + + @Test + void notStatusChangeForAddEvent() { + var event = new ExtendedResourceEvent(ADDED, resource("1", 1L, "app"), null, null); + + assertThat(event.isStatusChange()).isFalse(); + } + + @Test + void notStatusChangeForDeleteEvent() { + var event = new ExtendedResourceEvent(DELETED, resource("1", 1L, "app"), null, true); + + assertThat(event.isStatusChange()).isFalse(); + } + + ConfigMap resource(String resourceVersion, Long generation, String labelValue) { + var cm = new ConfigMap(); + cm.setMetadata( + new ObjectMetaBuilder() + .withName("id1") + .withNamespace("default") + .withResourceVersion(resourceVersion) + .withGeneration(generation) + .addToLabels("app", labelValue) + .build()); + return cm; + } +} 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 c62a1d1a3a..0a7f5cf503 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 @@ -123,7 +123,7 @@ void propagatesEventAndEvictsTempCacheOnVersionMismatch() { Deployment cachedDeployment = testDeployment(); cachedDeployment.getMetadata().setResourceVersion(PREV_RESOURCE_VERSION); - temporaryResourceCache.putResource(cachedDeployment); + temporaryResourceCache.putResource(cachedDeployment, true); informerEventSource.onUpdate(cachedDeployment, testDeployment()); @@ -478,7 +478,7 @@ void checkGhostResourcesPropagatesDeleteForMissingTempCacheEntry() { when(manager.get(any())).thenReturn(Optional.empty()); when(informerEventSource.manager()).thenReturn(manager); - tempCache.putResource(ghost); + tempCache.putResource(ghost, true); assertThat(tempCache.getResources()).containsKey(ResourceID.fromResource(ghost)); // Informer's last-sync moves past the temp cache entry's RV and the resource @@ -509,7 +509,7 @@ void checkGhostResourcesKeepsResourcePresentInInformerCache() { when(manager.get(any())).thenReturn(Optional.of(resource)); when(informerEventSource.manager()).thenReturn(manager); - tempCache.putResource(resource); + tempCache.putResource(resource, true); when(manager.lastSyncResourceVersion(any())).thenReturn("5"); tempCache.checkGhostResources(); @@ -542,7 +542,7 @@ void ghostCleanupDiscardsOrphanFilterWindow() { // Open a filter window for an in-flight write, then close it (our update returned but // the watch event never arrived). tempCache.startEventFilteringModify(resourceId); - tempCache.putResource(ghost); + tempCache.putResource(ghost, true); tempCache.doneEventFilterModify(resourceId); assertThat(tempCache.getEventFilterSupport().isActiveUpdateFor(resourceId)) .as("filter window must persist while ownResourceVersions is non-empty") @@ -580,7 +580,7 @@ void ghostCleanupSyntheticDeleteRespectsOnDeleteFilter() throws Exception { when(manager.get(any())).thenReturn(Optional.empty()); when(informerEventSource.manager()).thenReturn(manager); - tempCache.putResource(ghost); + tempCache.putResource(ghost, true); when(manager.lastSyncResourceVersion(any())).thenReturn("5"); tempCache.checkGhostResources(); @@ -610,7 +610,7 @@ void ghostCleanupRetainsActiveFilterWindowWhenResourcePresentInInformer() { when(informerEventSource.manager()).thenReturn(manager); tempCache.startEventFilteringModify(resourceId); - tempCache.putResource(resource); + tempCache.putResource(resource, true); tempCache.doneEventFilterModify(resourceId); when(manager.lastSyncResourceVersion(any())).thenReturn("5"); @@ -752,7 +752,7 @@ void filteringUpdateMapsUpdatedResourceToPrimariesOnlyOnce() { new ExtendedResourceEvent( ResourceAction.UPDATED, updated, resourceToUpdate, false))); - informerEventSource.eventFilteringUpdateAndCacheResource(resourceToUpdate, r -> updated); + informerEventSource.eventFilteringUpdateAndCacheResource(resourceToUpdate, r -> updated, true); verify(secondaryToPrimaryMapper, times(1)).toPrimaryResourceIDs(updated); verify(eventHandlerMock, times(1)).handleEvent(any()); @@ -773,7 +773,7 @@ void filteringUpdateFallsBackToMapperWhenNoPrimaryToSecondaryIndex() { new ExtendedResourceEvent( ResourceAction.UPDATED, updated, resourceToUpdate, false))); - informerEventSource.eventFilteringUpdateAndCacheResource(resourceToUpdate, r -> updated); + informerEventSource.eventFilteringUpdateAndCacheResource(resourceToUpdate, r -> updated, true); verify(secondaryToPrimaryMapper, times(1)).toPrimaryResourceIDs(updated); verify(secondaryToPrimaryMapper, times(1)).toPrimaryResourceIDs(resourceToUpdate); 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 158e42180f..da33dca92b 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 @@ -64,7 +64,7 @@ void updateAddsTheResourceIntoCacheIfTheInformerHasThePreviousResourceVersion() var prevTestResource = testResource(); prevTestResource.getMetadata().setResourceVersion("1"); - temporaryResourceCache.putResource(testResource); + temporaryResourceCache.putResource(testResource, true); var cached = temporaryResourceCache.getResourceFromCache(ResourceID.fromResource(testResource)); assertThat(cached).isPresent(); @@ -79,7 +79,7 @@ void updateNotAddsTheResourceIntoCacheIfLaterVersionKnown() { testResource.toBuilder().editMetadata().withResourceVersion("3").endMetadata().build(), null); latestSyncVersion = "3"; - temporaryResourceCache.putResource(testResource); + temporaryResourceCache.putResource(testResource, true); var cached = temporaryResourceCache.getResourceFromCache(ResourceID.fromResource(testResource)); assertThat(cached).isNotPresent(); @@ -89,7 +89,7 @@ void updateNotAddsTheResourceIntoCacheIfLaterVersionKnown() { void addOperationAddsTheResourceIfInformerCacheStillEmpty() { var testResource = testResource(); - temporaryResourceCache.putResource(testResource); + temporaryResourceCache.putResource(testResource, true); var cached = temporaryResourceCache.getResourceFromCache(ResourceID.fromResource(testResource)); assertThat(cached).isPresent(); @@ -99,14 +99,15 @@ void addOperationAddsTheResourceIfInformerCacheStillEmpty() { void addOperationNotAddsTheResourceIfInformerCacheNotEmpty() { var testResource = testResource(); - temporaryResourceCache.putResource(testResource); + temporaryResourceCache.putResource(testResource, true); when(managedInformerEventSource.get(any())).thenReturn(Optional.of(testResource)); temporaryResourceCache.putResource( new ConfigMapBuilder(testResource) .editMetadata() .withResourceVersion("1") .endMetadata() - .build()); + .build(), + true); var cached = temporaryResourceCache.getResourceFromCache(ResourceID.fromResource(testResource)); assertThat(cached.orElseThrow().getMetadata().getResourceVersion()).isEqualTo(RESOURCE_VERSION); @@ -134,7 +135,7 @@ void nonComparableResourceVersionsDisables() { this.temporaryResourceCache = new TemporaryResourceCache<>(false, mock(ManagedInformerEventSource.class)); - this.temporaryResourceCache.putResource(testResource()); + this.temporaryResourceCache.putResource(testResource(), true); assertThat(temporaryResourceCache.getResourceFromCache(ResourceID.fromResource(testResource()))) .isEmpty(); @@ -146,7 +147,7 @@ void eventReceivedDuringFiltering() { temporaryResourceCache.startEventFilteringModify(ResourceID.fromResource(testResource)); - temporaryResourceCache.putResource(testResource); + temporaryResourceCache.putResource(testResource, true); assertThat(temporaryResourceCache.getResourceFromCache(ResourceID.fromResource(testResource))) .isPresent(); @@ -168,7 +169,7 @@ void eventAfterFiltering() { temporaryResourceCache.startEventFilteringModify(ResourceID.fromResource(testResource)); - temporaryResourceCache.putResource(testResource); + temporaryResourceCache.putResource(testResource, true); assertThat(temporaryResourceCache.getResourceFromCache(ResourceID.fromResource(testResource))) .isPresent(); @@ -199,7 +200,7 @@ void putBeforeEventWithEventFiltering() { var resourceId = ResourceID.fromResource(testResource); temporaryResourceCache.startEventFilteringModify(resourceId); - temporaryResourceCache.putResource(nextResource); + temporaryResourceCache.putResource(nextResource, true); temporaryResourceCache.doneEventFilterModify(resourceId); latestSyncVersion = "3"; @@ -233,7 +234,7 @@ void putAfterEventWithEventFilteringNoPost() { ResourceAction.UPDATED, nextResource, testResource); assertThat(result).isEmpty(); - temporaryResourceCache.putResource(nextResource); + temporaryResourceCache.putResource(nextResource, true); var postEvent = temporaryResourceCache.doneEventFilterModify(resourceId); // there is no post event because the done call claimed responsibility for rv 3 @@ -269,7 +270,7 @@ void intermediateEventPropagatedWhenNoActiveUpdate() { var newer = testResource(); newer.getMetadata().setResourceVersion("3"); - temporaryResourceCache.putResource(newer); + temporaryResourceCache.putResource(newer, true); assertThat(temporaryResourceCache.getResourceFromCache(ResourceID.fromResource(olderEvent))) .isPresent(); @@ -298,7 +299,7 @@ void rapidDeletion() { .build(), false); latestSyncVersion = "3"; - temporaryResourceCache.putResource(testResource); + temporaryResourceCache.putResource(testResource, true); assertThat(temporaryResourceCache.getResourceFromCache(ResourceID.fromResource(testResource))) .isEmpty(); @@ -307,7 +308,7 @@ void rapidDeletion() { @Test void removalOfGhostResources() { var tr = testResource(); - this.temporaryResourceCache.putResource(tr); + this.temporaryResourceCache.putResource(tr, true); // ghost check should not remove when latestSyncVersion is not ahead temporaryResourceCache.checkGhostResources(); @@ -324,7 +325,7 @@ void removalOfGhostResources() { @Test void ghostResourceIsNotRemovedIfLatestSyncVersionIsOlder() { - this.temporaryResourceCache.putResource(testResource()); + this.temporaryResourceCache.putResource(testResource(), true); latestSyncVersion = "1"; temporaryResourceCache.checkGhostResources(); @@ -335,7 +336,7 @@ void ghostResourceIsNotRemovedIfLatestSyncVersionIsOlder() { @Test void ghostRemovalRemovesResourcesOnNotFollowedNamespaces() { var tr = testResource(); - temporaryResourceCache.putResource(tr); + temporaryResourceCache.putResource(tr, true); assertThat(temporaryResourceCache.getResourceFromCache(ResourceID.fromResource(tr))) .isPresent(); @@ -358,7 +359,7 @@ void doNotCacheResourceOnPutIfNamespaceIsNotFollowedAnymore() { when(mim.isWatchingNamespace("default")).thenReturn(false); var tr = testResource(); - temporaryResourceCache.putResource(tr); + temporaryResourceCache.putResource(tr, true); assertThat(temporaryResourceCache.getResourceFromCache(ResourceID.fromResource(tr))).isEmpty(); } @@ -371,7 +372,7 @@ void unknownStateDeleteEvictsTempCacheEvenWhenOlder() { .withResourceVersion("5") .endMetadata() .build(); - temporaryResourceCache.putResource(newer); + temporaryResourceCache.putResource(newer, true); var olderUnknownState = testResource(); var result = temporaryResourceCache.onDeleteEvent(olderUnknownState, true); @@ -384,7 +385,7 @@ void unknownStateDeleteEvictsTempCacheEvenWhenOlder() { private ConfigMap propagateTestResourceToCache() { var testResource = testResource(); - temporaryResourceCache.putResource(testResource); + temporaryResourceCache.putResource(testResource, true); assertThat(temporaryResourceCache.getResourceFromCache(ResourceID.fromResource(testResource))) .isPresent(); return testResource; diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/readcacheafterwrite/deletionduringstatusupdate/DeletionDuringStatusUpdateReconciler.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/readcacheafterwrite/deletionduringstatusupdate/DeletionDuringStatusUpdateReconciler.java index feb0509e72..0c34a70bd4 100644 --- a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/readcacheafterwrite/deletionduringstatusupdate/DeletionDuringStatusUpdateReconciler.java +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/readcacheafterwrite/deletionduringstatusupdate/DeletionDuringStatusUpdateReconciler.java @@ -64,7 +64,8 @@ public UpdateControl reconcile( r.getMetadata().setResourceVersion(null); return context.getClient().resource(r).patchStatus(); }, - context.eventSourceRetriever().getControllerEventSource()); + context.eventSourceRetriever().getControllerEventSource(), + true); return UpdateControl.noUpdate(); } diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/readcacheafterwrite/externalupdateduringownupdate/ExternalUpdateDuringOwnUpdateReconciler.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/readcacheafterwrite/externalupdateduringownupdate/ExternalUpdateDuringOwnUpdateReconciler.java index e5c956dc4e..10d4a1f4bd 100644 --- a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/readcacheafterwrite/externalupdateduringownupdate/ExternalUpdateDuringOwnUpdateReconciler.java +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/readcacheafterwrite/externalupdateduringownupdate/ExternalUpdateDuringOwnUpdateReconciler.java @@ -68,7 +68,8 @@ public UpdateControl reconcile( r.getMetadata().setResourceVersion(null); return context.getClient().resource(r).patchStatus(); }, - context.eventSourceRetriever().getControllerEventSource()); + context.eventSourceRetriever().getControllerEventSource(), + false); } else { var labels = resource.getMetadata().getLabels(); if (labels != null && EXTERNAL_LABEL_VALUE.equals(labels.get(EXTERNAL_LABEL_KEY))) { 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 13e8e72d74..a327235750 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 @@ -114,7 +114,8 @@ public UpdateControl reconcile( .withFieldManager(fieldManager) .withPatchType(PatchType.SERVER_SIDE_APPLY) .build()); - }); + }, + false); // 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); diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/readcacheafterwrite/specchangeduringstatuspatch/SpecChangeDuringStatusPatchCustomResource.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/readcacheafterwrite/specchangeduringstatuspatch/SpecChangeDuringStatusPatchCustomResource.java new file mode 100644 index 0000000000..dff33b454d --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/readcacheafterwrite/specchangeduringstatuspatch/SpecChangeDuringStatusPatchCustomResource.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.readcacheafterwrite.specchangeduringstatuspatch; + +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("scdsp") +public class SpecChangeDuringStatusPatchCustomResource + extends CustomResource + implements Namespaced {} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/readcacheafterwrite/specchangeduringstatuspatch/SpecChangeDuringStatusPatchIT.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/readcacheafterwrite/specchangeduringstatuspatch/SpecChangeDuringStatusPatchIT.java new file mode 100644 index 0000000000..a7467db395 --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/readcacheafterwrite/specchangeduringstatuspatch/SpecChangeDuringStatusPatchIT.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.readcacheafterwrite.specchangeduringstatuspatch; + +import java.time.Duration; +import java.util.concurrent.TimeUnit; + +import org.junit.jupiter.api.RepeatedTest; +import org.junit.jupiter.api.extension.RegisterExtension; + +import io.fabric8.kubernetes.api.model.ObjectMetaBuilder; +import io.fabric8.kubernetes.client.dsl.base.PatchContext; +import io.fabric8.kubernetes.client.dsl.base.PatchType; +import io.javaoperatorsdk.operator.junit.LocallyRunOperatorExtension; + +import static io.javaoperatorsdk.operator.baseapi.readcacheafterwrite.specchangeduringstatuspatch.SpecChangeDuringStatusPatchReconciler.STATUS_VALUE; +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; + +/** + * Reproduces a concurrent spec change happening while the controller patches its own status. When + * the reconciler patches the status it opens an event filtering window so it does not re-trigger + * itself. This test changes the spec on the cluster while that window is open and verifies that the + * spec change is still reconciled - it must not be silently absorbed together with the controller's + * own status update. + */ +class SpecChangeDuringStatusPatchIT { + + static final String RESOURCE_NAME = "test-resource"; + static final String SPEC_VALUE = "initial"; + public static final String UPDATED_SPEC_VALUE = "updated-val"; + + SpecChangeDuringStatusPatchReconciler reconciler = new SpecChangeDuringStatusPatchReconciler(); + + @RegisterExtension + LocallyRunOperatorExtension extension = + LocallyRunOperatorExtension.builder().withReconciler(reconciler).build(); + + @RepeatedTest(10) + void specChangeDuringStatusPatchIsReconciled() throws InterruptedException { + var res = extension.create(testResource()); + var statusRes = testResource(); + statusRes.getMetadata().setNamespace(res.getMetadata().getNamespace()); + extension + .getKubernetesClient() + .resource(statusRes) + .status() + .patch( + new PatchContext.Builder() + .withForce(true) + .withFieldManager( + SpecChangeDuringStatusPatchReconciler.class.getSimpleName().toLowerCase()) + .withPatchType(PatchType.SERVER_SIDE_APPLY) + .build()); + + // wait until the reconciler is inside its own status patch, holding the filtering window open + assertThat(reconciler.statusPatchStartedLatch.await(30, TimeUnit.SECONDS)) + .as("reconciler should enter its own status patch operation") + .isTrue(); + + // change the spec on the cluster while the controller's status patch is still in flight + var current = extension.get(SpecChangeDuringStatusPatchCustomResource.class, RESOURCE_NAME); + current.getSpec().setValue(UPDATED_SPEC_VALUE); + extension.replace(current); + + // let the reconciler finish its own status patch + reconciler.specChangeDoneLatch.countDown(); + + // the spec change must be picked up by a fresh reconciliation and not lost with the own update + await() + .atMost(Duration.ofSeconds(5)) + .untilAsserted( + () -> { + assertThat(reconciler.numberOfExecutions.get()).isGreaterThanOrEqualTo(2); + assertThat(reconciler.lastObservedSpecValue.get()) + .as("a later reconciliation must observe the externally-applied spec change") + .isEqualTo(UPDATED_SPEC_VALUE); + }); + + // sanity check: the status the controller set is still present after the concurrent spec change + var updated = extension.get(SpecChangeDuringStatusPatchCustomResource.class, RESOURCE_NAME); + assertThat(updated.getSpec().getValue()).isEqualTo(UPDATED_SPEC_VALUE); + assertThat(updated.getStatus()).isNotNull(); + assertThat(updated.getStatus().getValue()).isEqualTo(STATUS_VALUE); + } + + static SpecChangeDuringStatusPatchCustomResource testResource() { + var r = new SpecChangeDuringStatusPatchCustomResource(); + r.setMetadata(new ObjectMetaBuilder().withName(RESOURCE_NAME).build()); + r.setSpec(new SpecChangeDuringStatusPatchSpec().setValue(SPEC_VALUE)); + r.setStatus(new SpecChangeDuringStatusPatchStatus().setValue(STATUS_VALUE)); + return r; + } +} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/readcacheafterwrite/specchangeduringstatuspatch/SpecChangeDuringStatusPatchReconciler.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/readcacheafterwrite/specchangeduringstatuspatch/SpecChangeDuringStatusPatchReconciler.java new file mode 100644 index 0000000000..33d809046b --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/readcacheafterwrite/specchangeduringstatuspatch/SpecChangeDuringStatusPatchReconciler.java @@ -0,0 +1,72 @@ +/* + * 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.readcacheafterwrite.specchangeduringstatuspatch; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + +import io.javaoperatorsdk.operator.api.reconciler.Context; +import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration; +import io.javaoperatorsdk.operator.api.reconciler.Reconciler; +import io.javaoperatorsdk.operator.api.reconciler.UpdateControl; + +/** + * On the first reconciliation the reconciler patches its own status, but keeps the event filtering + * window (opened by {@link + * io.javaoperatorsdk.operator.api.reconciler.ResourceOperations#resourcePatch}) open until the test + * signals that it has changed the spec on the cluster. This reproduces the race where a spec change + * lands while the controller's own status patch is in flight: the spec change event must still + * propagate as a fresh reconciliation, it must not be absorbed as if it were our own status update. + */ +@ControllerConfiguration +public class SpecChangeDuringStatusPatchReconciler + implements Reconciler { + + static final String STATUS_VALUE = "reconciled"; + + final AtomicInteger numberOfExecutions = new AtomicInteger(); + final CountDownLatch statusPatchStartedLatch = new CountDownLatch(1); + final CountDownLatch specChangeDoneLatch = new CountDownLatch(1); + final AtomicReference lastObservedSpecValue = new AtomicReference<>(); + + @Override + public UpdateControl reconcile( + SpecChangeDuringStatusPatchCustomResource resource, + Context context) { + int execution = numberOfExecutions.incrementAndGet(); + lastObservedSpecValue.set(resource.getSpec().getValue()); + + if (execution == 1) { + resource.setStatus(new SpecChangeDuringStatusPatchStatus().setValue(STATUS_VALUE)); + resource.getMetadata().setResourceVersion(null); + // Patch our own status, but hold the filtering window open with a hook that lets the test + // change the spec on the cluster WHILE the status patch is still in flight. + statusPatchStartedLatch.countDown(); + try { + if (!specChangeDoneLatch.await(30, TimeUnit.SECONDS)) { + throw new IllegalStateException("timed out waiting for external spec change"); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException(e); + } + return UpdateControl.patchStatus(resource); + } + return UpdateControl.noUpdate(); + } +} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/readcacheafterwrite/specchangeduringstatuspatch/SpecChangeDuringStatusPatchSpec.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/readcacheafterwrite/specchangeduringstatuspatch/SpecChangeDuringStatusPatchSpec.java new file mode 100644 index 0000000000..64a6e7f0ef --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/readcacheafterwrite/specchangeduringstatuspatch/SpecChangeDuringStatusPatchSpec.java @@ -0,0 +1,30 @@ +/* + * 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.readcacheafterwrite.specchangeduringstatuspatch; + +public class SpecChangeDuringStatusPatchSpec { + + private String value; + + public String getValue() { + return value; + } + + public SpecChangeDuringStatusPatchSpec setValue(String value) { + this.value = value; + return this; + } +} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/readcacheafterwrite/specchangeduringstatuspatch/SpecChangeDuringStatusPatchStatus.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/readcacheafterwrite/specchangeduringstatuspatch/SpecChangeDuringStatusPatchStatus.java new file mode 100644 index 0000000000..4cb857a4cc --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/readcacheafterwrite/specchangeduringstatuspatch/SpecChangeDuringStatusPatchStatus.java @@ -0,0 +1,30 @@ +/* + * 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.readcacheafterwrite.specchangeduringstatuspatch; + +public class SpecChangeDuringStatusPatchStatus { + + private String value; + + public String getValue() { + return value; + } + + public SpecChangeDuringStatusPatchStatus setValue(String value) { + this.value = value; + return this; + } +} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/dependent/externalstate/ExternalStateReconciler.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/dependent/externalstate/ExternalStateReconciler.java index 4f4cab80d7..73e3769adf 100644 --- a/operator-framework/src/test/java/io/javaoperatorsdk/operator/dependent/externalstate/ExternalStateReconciler.java +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/dependent/externalstate/ExternalStateReconciler.java @@ -87,7 +87,7 @@ private void updateExternalResource( var newResource = new ExternalResource(externalResource.getId(), resource.getSpec().getData()); externalService.update(newResource); externalResourceEventSource.handleRecentResourceUpdate( - ResourceID.fromResource(resource), newResource, externalResource); + ResourceID.fromResource(resource), newResource, externalResource, false); } private void createExternalResource( @@ -110,7 +110,7 @@ private void createExternalResource( // This is critical in this case, since on next reconciliation if it would not be in the cache // it would be created again. configMapEventSource.eventFilteringUpdateAndCacheResource( - configMap, toCreate -> context.resourceOperations().serverSideApply(toCreate)); + configMap, toCreate -> context.resourceOperations().serverSideApply(toCreate), false); externalResourceEventSource.handleRecentResourceCreate(primaryID, createdResource); }