diff --git a/docs/content/en/docs/documentation/dependent-resource-and-workflows/dependent-resources.md b/docs/content/en/docs/documentation/dependent-resource-and-workflows/dependent-resources.md index 8974c41f2a..538c52c00c 100644 --- a/docs/content/en/docs/documentation/dependent-resource-and-workflows/dependent-resources.md +++ b/docs/content/en/docs/documentation/dependent-resource-and-workflows/dependent-resources.md @@ -480,6 +480,17 @@ also be created, one per dependent resource. See [integration test](https://github.com/operator-framework/java-operator-sdk/blob/main/operator-framework/src/test/java/io/javaoperatorsdk/operator/dependent/externalstate/externalstatebulkdependent) as a sample. +Note that an external resource and the state resource referencing it cannot be created atomically: +the external resource has to be created first, since its identifier is what gets stored in the +state. If the resources are fetched based on the state - which is usually the case, since the +identifier is only known from the state - a poll happening in between the two steps cannot see the +new external resource yet. JOSDK keeps such a recently created resource in the cache for the next +update to avoid creating a duplicate of it, but for a resource that takes longer to become visible, +it is recommended to resolve the actual resources from the state resources in +`BulkDependentResource.getSecondaryResources`, as done in the integration test above. The state +resources are managed by an `InformerEventSource`, thus are always up-to-date regarding the +operator's own changes. + ## GenericKubernetesResource based Dependent Resources In rare circumstances resource handling where there is no class representation or just typeless handling might be 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 109e83b413..d7a50620e5 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 @@ -67,6 +67,28 @@ public abstract class ExternalResourceCachingEventSource> cache = new ConcurrentHashMap<>(); + /** + * The resources written by the reconciler ({@link #handleRecentResourceCreate(ResourceID, + * Object)} and {@link #handleRecentResourceUpdate(ResourceID, Object, Object)}) that were not + * seen yet in a subsequent update of the whole resource set of a primary. Such an update might + * have been created (polled or received) before the resource was actually written, thus not + * containing the new state yet. Since these updates are handled as the full actual state, the + * write would be lost from the cache; the next reconciliation would then create a duplicate of an + * already created resource, or repeat an already executed update. Note that a mark is dropped on + * the first update, so a resource really deleted or changed in the meantime is not retained + * indefinitely. + * + * @see #retainUnconfirmedWrites(ResourceID, Map) + */ + private final Map>> unconfirmedWrites = + new ConcurrentHashMap<>(); + + /** + * A resource written by the reconciler and the state it replaced, which is {@code null} in case + * the resource was created. + */ + private record RecentWrite(R written, R replaced) {} + protected ExternalResourceCachingEventSource( Class resourceClass, ResourceIDMapper resourceIDMapper) { this(null, resourceClass, resourceIDMapper); @@ -86,6 +108,7 @@ protected ExternalResourceCachingEventSource( } protected synchronized void handleDelete(ResourceID primaryID) { + unconfirmedWrites.remove(primaryID); var res = cache.remove(primaryID); if (res != null && deleteAcceptedByFilter(res.values())) { getEventHandler().handleEvent(new Event(primaryID)); @@ -105,6 +128,13 @@ protected synchronized void handleDelete(ResourceID primaryID, Set resourceI if (!isRunning()) { return; } + var unconfirmed = unconfirmedWrites.get(primaryID); + if (unconfirmed != null) { + unconfirmed.keySet().removeAll(resourceIDs); + if (unconfirmed.isEmpty()) { + unconfirmedWrites.remove(primaryID); + } + } var cachedValues = cache.get(primaryID); List removedResources = cachedValues == null @@ -131,7 +161,16 @@ protected synchronized void handleResources(ResourceID primaryID, Set newReso protected synchronized void handleResources(Map> allNewResources) { var toDelete = cache.keySet().stream().filter(k -> !allNewResources.containsKey(k)).toList(); - toDelete.forEach(this::handleDelete); + toDelete.forEach( + primaryID -> { + if (unconfirmedWrites.containsKey(primaryID)) { + // handled as an empty update, so that a recently written resource, that this update + // could not see yet, is not removed from the cache + handleResources(primaryID, Collections.emptySet()); + } else { + handleDelete(primaryID); + } + }); allNewResources.forEach(this::handleResources); } @@ -148,6 +187,7 @@ protected synchronized void handleResources( } var newResourcesMap = newResources.stream().collect(Collectors.toMap(resourceIDMapper::idFor, r -> r)); + retainUnconfirmedWrites(primaryID, newResourcesMap); cache.put(primaryID, newResourcesMap); if (propagateEvent && !newResourcesMap.equals(cachedResources) @@ -156,6 +196,34 @@ && acceptedByFiler(cachedResources, newResourcesMap)) { } } + /** + * Keeps the resources written since the received update was created, thus missing from it. An + * update is considered stale for a written resource if it does not contain it at all - which is + * the expected case for a create - or if it still contains the state that the write replaced. Any + * other state is a change that happened outside of the reconciler, so it is accepted as the + * actual state. + * + * @see #unconfirmedWrites + */ + private void retainUnconfirmedWrites(ResourceID primaryID, Map newResourcesMap) { + var unconfirmed = unconfirmedWrites.remove(primaryID); + if (unconfirmed == null) { + return; + } + unconfirmed.forEach( + (id, write) -> { + var newResource = newResourcesMap.get(id); + if (newResource == null || newResource.equals(write.replaced())) { + log.debug( + "Retaining recently written resource missing from the update. Primary ID: {}," + + " resource ID: {}", + primaryID, + id); + newResourcesMap.put(id, write.written()); + } + }); + } + private boolean acceptedByFiler(Map cachedResourceMap, Map newResourcesMap) { var addedResources = new HashMap<>(newResourcesMap); @@ -217,6 +285,7 @@ public synchronized void handleRecentResourceCreate(ResourceID primaryID, R reso } else { actualValues.computeIfAbsent(resourceId, r -> resource); } + markUnconfirmedWrite(primaryID, resourceId, new RecentWrite<>(resource, null)); } @Override @@ -226,12 +295,18 @@ public synchronized void handleRecentResourceUpdate( if (actualValues != null) { var resourceId = resourceIDMapper.idFor(resource); R actualResource = actualValues.get(resourceId); - if (actualResource.equals(previousVersionOfResource)) { + if (actualResource != null && actualResource.equals(previousVersionOfResource)) { actualValues.put(resourceId, resource); + markUnconfirmedWrite( + primaryID, resourceId, new RecentWrite<>(resource, previousVersionOfResource)); } } } + private void markUnconfirmedWrite(ResourceID primaryID, ID resourceId, RecentWrite write) { + unconfirmedWrites.computeIfAbsent(primaryID, id -> new HashMap<>()).put(resourceId, write); + } + @Override public Set getSecondaryResources(P primary) { return getSecondaryResources(ResourceID.fromResource(primary)); diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/ExternalResourceCachingEventSourceTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/ExternalResourceCachingEventSourceTest.java index 889cc4da75..f106257197 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/ExternalResourceCachingEventSourceTest.java +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/ExternalResourceCachingEventSourceTest.java @@ -15,6 +15,7 @@ */ package io.javaoperatorsdk.operator.processing.event.source; +import java.util.Map; import java.util.Set; import org.junit.jupiter.api.BeforeEach; @@ -211,6 +212,107 @@ void genericFilteringEvents() { verify(eventHandler, times(0)).handleEvent(any()); } + @Test + void retainsRecentlyCreatedResourceMissingFromUpdate() { + source.handleResources(primaryID1(), Set.of(testResource1())); + source.handleRecentResourceCreate(primaryID1(), testResource2()); + + // the update was created before the resource, thus does not contain it yet + source.handleResources(primaryID1(), Set.of(testResource1())); + + assertThat(source.getSecondaryResources(primaryID1())) + .containsExactlyInAnyOrder(testResource1(), testResource2()); + // no event for the retained resource, only the initial add event + verify(eventHandler, times(1)).handleEvent(new Event(primaryID1())); + } + + @Test + void retainsRecentlyCreatedResourceOnlyForASingleUpdate() { + source.handleResources(primaryID1(), Set.of(testResource1())); + source.handleRecentResourceCreate(primaryID1(), testResource2()); + source.handleResources(primaryID1(), Set.of(testResource1())); + + // this update is created after the resource, so it is really deleted meanwhile + source.handleResources(primaryID1(), Set.of(testResource1())); + + assertThat(source.getSecondaryResources(primaryID1())).containsExactly(testResource1()); + verify(eventHandler, times(2)).handleEvent(new Event(primaryID1())); + } + + @Test + void doesNotRetainRecentlyCreatedResourceDeletedBeforeTheUpdate() { + source.handleRecentResourceCreate(primaryID1(), testResource2()); + source.handleDelete(primaryID1(), testResource2()); + + source.handleResources(primaryID1(), Set.of(testResource1())); + + assertThat(source.getSecondaryResources(primaryID1())).containsExactly(testResource1()); + } + + @Test + void retainsRecentlyCreatedResourceMissingFromWholeCacheUpdate() { + source.handleRecentResourceCreate(primaryID1(), testResource1()); + + source.handleResources(Map.of()); + + assertThat(source.getSecondaryResources(primaryID1())).containsExactly(testResource1()); + + source.handleResources(Map.of()); + + assertThat(source.getSecondaryResources(primaryID1())).isEmpty(); + } + + @Test + void retainsRecentlyUpdatedResourceMissingFromUpdate() { + source.handleResources(primaryID1(), Set.of(testResource1())); + source.handleRecentResourceUpdate(primaryID1(), changedTestResource1(), testResource1()); + + // the update was created before the resource was updated, thus still contains the old state + source.handleResources(primaryID1(), Set.of(testResource1())); + + assertThat(source.getSecondaryResources(primaryID1())).containsExactly(changedTestResource1()); + // no event for the retained resource, only the initial add event + verify(eventHandler, times(1)).handleEvent(new Event(primaryID1())); + } + + @Test + void retainsRecentlyUpdatedResourceOnlyForASingleUpdate() { + source.handleResources(primaryID1(), Set.of(testResource1())); + source.handleRecentResourceUpdate(primaryID1(), changedTestResource1(), testResource1()); + source.handleResources(primaryID1(), Set.of(testResource1())); + + // this update is created after the resource was updated, so it was really changed meanwhile + source.handleResources(primaryID1(), Set.of(testResource1())); + + assertThat(source.getSecondaryResources(primaryID1())).containsExactly(testResource1()); + verify(eventHandler, times(2)).handleEvent(new Event(primaryID1())); + } + + @Test + void doesNotRetainRecentlyUpdatedResourceChangedOutsideOfTheReconciler() { + var externallyChanged = testResource1().setValue("externallyChangedValue"); + source.handleResources(primaryID1(), Set.of(testResource1())); + source.handleRecentResourceUpdate(primaryID1(), changedTestResource1(), testResource1()); + + source.handleResources(primaryID1(), Set.of(externallyChanged)); + + assertThat(source.getSecondaryResources(primaryID1())).containsExactly(externallyChanged); + } + + @Test + void retainsRecentlyUpdatedResourceInWholeCacheUpdate() { + source.handleResources(primaryID1(), Set.of(testResource1())); + source.handleRecentResourceUpdate(primaryID1(), changedTestResource1(), testResource1()); + + source.handleResources(Map.of(primaryID1(), Set.of(testResource1()))); + + assertThat(source.getSecondaryResources(primaryID1())).containsExactly(changedTestResource1()); + } + + private static SampleExternalResource changedTestResource1() { + return testResource1().setValue("changedValue"); + } + public static class TestExternalCachingEventSource extends ExternalResourceCachingEventSource { public TestExternalCachingEventSource() { diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/dependent/externalstate/externalstatebulkdependent/BulkDependentResourceExternalWithState.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/dependent/externalstate/externalstatebulkdependent/BulkDependentResourceExternalWithState.java index ba08f7fdfa..eb699f225e 100644 --- a/operator-framework/src/test/java/io/javaoperatorsdk/operator/dependent/externalstate/externalstatebulkdependent/BulkDependentResourceExternalWithState.java +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/dependent/externalstate/externalstatebulkdependent/BulkDependentResourceExternalWithState.java @@ -135,12 +135,21 @@ public Map desiredResources( return res; } + /** + * Resolves the actual resources from the persisted state instead of the polled cache. An external + * resource and the state referencing it cannot be created atomically, so a poll happening in + * between replaces the cached resources with the ones it can already see, dropping the freshly + * created one. The next reconciliation would then create a duplicate external resource that no + * state references anymore, thus is leaked. The state itself is read-after-write consistent, + * since it is managed through an {@link + * io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource}. + */ @Override public Map getSecondaryResources( ExternalStateBulkDependentCustomResource primary, Context context) { - var resources = context.getSecondaryResources(ExternalResource.class); - return resources.stream().collect(Collectors.toMap(this::externalResourceIndex, r -> r)); + return fetchResources(primary).stream() + .collect(Collectors.toMap(this::externalResourceIndex, r -> r)); } @Override