diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/dependent/DependentResource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/dependent/DependentResource.java index affeb96bbf..8230dc4cf9 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/dependent/DependentResource.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/dependent/DependentResource.java @@ -44,7 +44,7 @@ public interface DependentResource { * @param eventSourceContext context of event source initialization * @return an optional event source */ - default Optional> eventSource( + default Optional> eventSource( EventSourceContext

eventSourceContext) { return Optional.empty(); } diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/dependent/RecentOperationEventFilter.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/dependent/RecentOperationEventFilter.java deleted file mode 100644 index d48343e57c..0000000000 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/dependent/RecentOperationEventFilter.java +++ /dev/null @@ -1,11 +0,0 @@ -package io.javaoperatorsdk.operator.api.reconciler.dependent; - -import io.javaoperatorsdk.operator.processing.event.ResourceID; - -public interface RecentOperationEventFilter extends RecentOperationCacheFiller { - - void prepareForCreateOrUpdateEventFiltering(ResourceID resourceID, R resource); - - void cleanupOnCreateOrUpdateEventFiltering(ResourceID resourceID); - -} 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 972b61cd94..04aa5631cf 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 @@ -34,7 +34,7 @@ protected AbstractEventSourceHolderDependentResource(Class resourceType) { this.resourceType = resourceType; } - public Optional> eventSource(EventSourceContext

context) { + public Optional eventSource(EventSourceContext

context) { // some sub-classes (e.g. KubernetesDependentResource) can have their event source created // before this method is called in the managed case, so only create the event source if it // hasn't already been set. @@ -67,9 +67,8 @@ public void resolveEventSource(EventSourceRetriever

eventSourceRetriever) { * @param context for event sources * @return event source instance */ - @SuppressWarnings("unchecked") public T initEventSource(EventSourceContext

context) { - return (T) eventSource(context).orElseThrow(); + return eventSource(context).orElseThrow(); } @Override @@ -96,7 +95,7 @@ protected void applyFilters() { this.eventSource.setGenericFilter(genericFilter); } - public Optional> eventSource() { + public Optional eventSource() { return Optional.ofNullable(eventSource); } diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentResource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentResource.java index 8d32892e10..5fedd0899d 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentResource.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentResource.java @@ -1,6 +1,6 @@ package io.javaoperatorsdk.operator.processing.dependent.kubernetes; -import java.util.HashMap; +import java.util.Map; import java.util.Optional; import java.util.Set; @@ -8,7 +8,6 @@ import org.slf4j.LoggerFactory; import io.fabric8.kubernetes.api.model.HasMetadata; -import io.fabric8.kubernetes.api.model.Namespaced; import io.fabric8.kubernetes.client.KubernetesClient; import io.fabric8.kubernetes.client.dsl.Resource; import io.javaoperatorsdk.operator.OperatorException; @@ -103,29 +102,6 @@ public void configureWith(InformerEventSource informerEventSource) { setEventSource(informerEventSource); } - - protected R handleCreate(R desired, P primary, Context

context) { - ResourceID resourceID = ResourceID.fromResource(desired); - try { - prepareEventFiltering(desired, resourceID); - return super.handleCreate(desired, primary, context); - } catch (RuntimeException e) { - cleanupAfterEventFiltering(resourceID); - throw e; - } - } - - protected R handleUpdate(R actual, R desired, P primary, Context

context) { - ResourceID resourceID = ResourceID.fromResource(desired); - try { - prepareEventFiltering(desired, resourceID); - return super.handleUpdate(actual, desired, primary, context); - } catch (RuntimeException e) { - cleanupAfterEventFiltering(resourceID); - throw e; - } - } - @SuppressWarnings("unused") public R create(R target, P primary, Context

context) { if (useSSA(context)) { @@ -137,6 +113,7 @@ public R create(R target, P primary, Context

context) { target.getMetadata().setResourceVersion("1"); } } + addMetadata(false, null, target, primary); final var resource = prepare(target, primary, "Creating"); return useSSA(context) ? resource @@ -152,6 +129,7 @@ public R update(R actual, R target, P primary, Context

context) { actual.getMetadata().getResourceVersion()); } R updatedResource; + addMetadata(false, actual, target, primary); if (useSSA(context)) { updatedResource = prepare(target, primary, "Updating") .fieldManager(context.getControllerConfiguration().fieldManager()) @@ -165,31 +143,50 @@ public R update(R actual, R target, P primary, Context

context) { return updatedResource; } + @Override public Result match(R actualResource, P primary, Context

context) { final var desired = desired(primary, context); + return match(actualResource, desired, primary, updaterMatcher, context); + } + + @SuppressWarnings({"unused", "unchecked"}) + public Result match(R actualResource, R desired, P primary, Context

context) { + return match(actualResource, desired, primary, + (ResourceUpdaterMatcher) GenericResourceUpdaterMatcher + .updaterMatcherFor(actualResource.getClass()), + context); + } + + public Result match(R actualResource, R desired, P primary, ResourceUpdaterMatcher matcher, + Context

context) { final boolean matches; + addMetadata(true, actualResource, desired, primary); if (useSSA(context)) { - addReferenceHandlingMetadata(desired, primary); matches = SSABasedGenericKubernetesResourceMatcher.getInstance() .matches(actualResource, desired, context); } else { - matches = updaterMatcher.matches(actualResource, desired, context); + matches = matcher.matches(actualResource, desired, context); } return Result.computed(matches, desired); } - @SuppressWarnings("unused") - public Result match(R actualResource, R desired, P primary, Context

context) { - if (useSSA(context)) { - addReferenceHandlingMetadata(desired, primary); - var matches = SSABasedGenericKubernetesResourceMatcher.getInstance() - .matches(actualResource, desired, context); - return Result.computed(matches, desired); - } else { - return GenericKubernetesResourceMatcher - .match(desired, actualResource, true, - false, false, context); + protected void addMetadata(boolean forMatch, R actualResource, final R target, P primary) { + if (forMatch) { // keep the current + String actual = actualResource.getMetadata().getAnnotations() + .get(InformerEventSource.PREVIOUS_ANNOTATION_KEY); + Map annotations = target.getMetadata().getAnnotations(); + if (actual != null) { + annotations.put(InformerEventSource.PREVIOUS_ANNOTATION_KEY, actual); + } else { + annotations.remove(InformerEventSource.PREVIOUS_ANNOTATION_KEY); + } + } else { // set a new one + eventSource().orElseThrow().addPreviousAnnotation( + Optional.ofNullable(actualResource).map(r -> r.getMetadata().getResourceVersion()) + .orElse(null), + target); } + addReferenceHandlingMetadata(target, primary); } private boolean useSSA(Context

context) { @@ -197,6 +194,7 @@ private boolean useSSA(Context

context) { .ssaBasedCreateUpdateMatchForDependentResources(); } + @Override protected void handleDelete(P primary, R secondary, Context

context) { if (secondary != null) { client.resource(secondary).delete(); @@ -214,13 +212,7 @@ protected Resource prepare(R desired, P primary, String actionName) { desired.getClass(), ResourceID.fromResource(desired)); - addReferenceHandlingMetadata(desired, primary); - - if (desired instanceof Namespaced) { - return client.resource(desired).inNamespace(desired.getMetadata().getNamespace()); - } else { - return client.resource(desired); - } + return client.resource(desired); } protected void addReferenceHandlingMetadata(R desired, P primary) { @@ -254,7 +246,7 @@ protected InformerEventSource createEventSource(EventSourceContext

cont "Using default configuration for {} KubernetesDependentResource, call configureWith to provide configuration", resourceType().getSimpleName()); } - return (InformerEventSource) eventSource().orElseThrow(); + return eventSource().orElseThrow(); } private boolean useDefaultAnnotationsToIdentifyPrimary() { @@ -263,10 +255,6 @@ private boolean useDefaultAnnotationsToIdentifyPrimary() { private void addDefaultSecondaryToPrimaryMapperAnnotations(R desired, P primary) { var annotations = desired.getMetadata().getAnnotations(); - if (annotations == null) { - annotations = new HashMap<>(); - desired.getMetadata().setAnnotations(annotations); - } annotations.put(Mappers.DEFAULT_ANNOTATION_FOR_NAME, primary.getMetadata().getName()); var primaryNamespaces = primary.getMetadata().getNamespace(); if (primaryNamespaces != null) { @@ -294,16 +282,6 @@ protected R desired(P primary, Context

context) { return super.desired(primary, context); } - private void prepareEventFiltering(R desired, ResourceID resourceID) { - ((InformerEventSource) eventSource().orElseThrow()) - .prepareForCreateOrUpdateEventFiltering(resourceID, desired); - } - - private void cleanupAfterEventFiltering(ResourceID resourceID) { - ((InformerEventSource) eventSource().orElseThrow()) - .cleanupOnCreateOrUpdateEventFiltering(resourceID); - } - @Override public Optional> configuration() { return Optional.ofNullable(kubernetesDependentResourceConfig); diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/EventRecorder.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/EventRecorder.java deleted file mode 100644 index 5d23d870aa..0000000000 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/EventRecorder.java +++ /dev/null @@ -1,72 +0,0 @@ -package io.javaoperatorsdk.operator.processing.event.source.informer; - -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; -import java.util.Map; - -import io.fabric8.kubernetes.api.model.HasMetadata; -import io.javaoperatorsdk.operator.processing.event.ResourceID; - -public class EventRecorder { - - private final Map> resourceEvents = new HashMap<>(); - - public void startEventRecording(ResourceID resourceID) { - resourceEvents.putIfAbsent(resourceID, new ArrayList<>(5)); - } - - public boolean isRecordingFor(ResourceID resourceID) { - return resourceEvents.get(resourceID) != null; - } - - public void stopEventRecording(ResourceID resourceID) { - resourceEvents.remove(resourceID); - } - - public void recordEvent(R resource) { - resourceEvents.get(ResourceID.fromResource(resource)).add(resource); - } - - public boolean containsEventWithResourceVersion(ResourceID resourceID, - String resourceVersion) { - List events = resourceEvents.get(resourceID); - if (events == null) { - return false; - } - if (events.isEmpty()) { - return false; - } else { - return events.stream() - .anyMatch(e -> e.getMetadata().getResourceVersion().equals(resourceVersion)); - } - } - - public boolean containsEventWithVersionButItsNotLastOne( - ResourceID resourceID, String resourceVersion) { - List resources = resourceEvents.get(resourceID); - if (resources == null) { - throw new IllegalStateException( - "Null events list, this is probably a result of invalid usage of the " + - "InformerEventSource. Resource ID: " + resourceID); - } - if (resources.isEmpty()) { - throw new IllegalStateException("No events for resource id: " + resourceID); - } - return !resources - .get(resources.size() - 1) - .getMetadata() - .getResourceVersion() - .equals(resourceVersion); - } - - public R getLastEvent(ResourceID resourceID) { - List resources = resourceEvents.get(resourceID); - if (resources == null) { - throw new IllegalStateException( - "Null events list, this is probably a result of invalid usage of the " + - "InformerEventSource. Resource ID: " + resourceID); - } - return resources.get(resources.size() - 1); - } -} 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 8cca524464..1fb7d61b4e 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 @@ -1,10 +1,6 @@ package io.javaoperatorsdk.operator.processing.event.source.informer; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.Optional; -import java.util.Set; +import java.util.*; import java.util.function.Function; import java.util.stream.Collectors; @@ -18,7 +14,6 @@ import io.javaoperatorsdk.operator.api.config.ConfigurationService; import io.javaoperatorsdk.operator.api.config.informer.InformerConfiguration; import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext; -import io.javaoperatorsdk.operator.api.reconciler.dependent.RecentOperationEventFilter; import io.javaoperatorsdk.operator.processing.event.Event; import io.javaoperatorsdk.operator.processing.event.EventHandler; import io.javaoperatorsdk.operator.processing.event.ResourceID; @@ -72,17 +67,18 @@ */ public class InformerEventSource extends ManagedInformerEventSource> - implements ResourceEventHandler, RecentOperationEventFilter { + implements ResourceEventHandler { + + public static String PREVIOUS_ANNOTATION_KEY = "javaoperatorsdk.io/previous"; private static final Logger log = LoggerFactory.getLogger(InformerEventSource.class); private final InformerConfiguration configuration; - // always called from a synchronized method - private final EventRecorder eventRecorder = new EventRecorder<>(); // we need direct control for the indexer to propagate the just update resource also to the index private final PrimaryToSecondaryIndex primaryToSecondaryIndex; private final PrimaryToSecondaryMapper

primaryToSecondaryMapper; private Map>> indexerBuffer = new HashMap<>(); + private final String id = UUID.randomUUID().toString(); public InformerEventSource( InformerConfiguration configuration, EventSourceContext

context) { @@ -154,12 +150,8 @@ public void onDelete(R resource, boolean b) { private synchronized void onAddOrUpdate(Operation operation, R newObject, R oldObject, Runnable superOnOp) { var resourceID = ResourceID.fromResource(newObject); - if (eventRecorder.isRecordingFor(resourceID)) { - log.debug("Recording event for: {}", resourceID); - eventRecorder.recordEvent(newObject); - return; - } - if (temporaryCacheHasResourceWithSameVersionAs(newObject)) { + + if (canSkipEvent(newObject, oldObject, resourceID)) { log.debug( "Skipping event propagation for {}, since was a result of a reconcile action. Resource ID: {}", operation, @@ -179,16 +171,33 @@ private synchronized void onAddOrUpdate(Operation operation, R newObject, R oldO } } - private boolean temporaryCacheHasResourceWithSameVersionAs(R resource) { - var resourceID = ResourceID.fromResource(resource); + private boolean canSkipEvent(R newObject, R oldObject, ResourceID resourceID) { var res = temporaryResourceCache.getResourceFromCache(resourceID); - return res.map(r -> { - boolean resVersionsEqual = r.getMetadata().getResourceVersion() - .equals(resource.getMetadata().getResourceVersion()); - log.debug("Resource found in temporal cache for id: {} resource versions equal: {}", - resourceID, resVersionsEqual); - return resVersionsEqual; - }).orElse(false); + if (res.isEmpty()) { + return isEventKnownFromAnnotation(newObject, oldObject); + } + boolean resVersionsEqual = newObject.getMetadata().getResourceVersion() + .equals(res.get().getMetadata().getResourceVersion()); + log.debug("Resource found in temporal cache for id: {} resource versions equal: {}", + resourceID, resVersionsEqual); + return resVersionsEqual; + } + + private boolean isEventKnownFromAnnotation(R newObject, R oldObject) { + String previous = newObject.getMetadata().getAnnotations().get(PREVIOUS_ANNOTATION_KEY); + boolean known = false; + if (previous != null) { + String[] parts = previous.split(","); + if (id.equals(parts[0])) { + if (oldObject == null && parts.length == 1) { + known = true; + } else if (oldObject != null && parts.length == 2 + && oldObject.getMetadata().getResourceVersion().equals(parts[1])) { + known = true; + } + } + } + return known; } private void propagateEvent(R object) { @@ -239,99 +248,24 @@ public InformerConfiguration getConfiguration() { @Override public synchronized void handleRecentResourceUpdate(ResourceID resourceID, R resource, R previousVersionOfResource) { - handleRecentCreateOrUpdate(Operation.UPDATE, resource, previousVersionOfResource, - () -> super.handleRecentResourceUpdate(resourceID, resource, previousVersionOfResource)); + handleRecentCreateOrUpdate(Operation.UPDATE, resource, previousVersionOfResource); } @Override public synchronized void handleRecentResourceCreate(ResourceID resourceID, R resource) { - handleRecentCreateOrUpdate(Operation.ADD, resource, null, - () -> super.handleRecentResourceCreate(resourceID, resource)); + handleRecentCreateOrUpdate(Operation.ADD, resource, null); } - private void handleRecentCreateOrUpdate(Operation operation, R resource, R oldResource, - Runnable runnable) { - primaryToSecondaryIndex.onAddOrUpdate(resource); - if (eventRecorder.isRecordingFor(ResourceID.fromResource(resource))) { - handleRecentResourceOperationAndStopEventRecording(operation, resource, oldResource); - } else { - runnable.run(); - } - } - - /** - * There can be the following cases: - *

- * - * @param newResource just created or updated resource - */ - private void handleRecentResourceOperationAndStopEventRecording(Operation operation, - R newResource, R oldResource) { - ResourceID resourceID = ResourceID.fromResource(newResource); - try { - if (!eventRecorder.containsEventWithResourceVersion( - resourceID, newResource.getMetadata().getResourceVersion())) { - log.debug( - "Did not found event in buffer with target version and resource id: {}", resourceID); - temporaryResourceCache.unconditionallyCacheResource(newResource); - } else { - // if the resource is not added to the temp cache, it is cleared, since - // the cache is cleared by subsequent events after updates, but if those did not receive - // the temp cache is still filled at this point with an old resource - log.debug("Cleaning temporary cache for resource id: {}", resourceID); - temporaryResourceCache.removeResourceFromCache(newResource); - if (eventRecorder.containsEventWithVersionButItsNotLastOne( - resourceID, newResource.getMetadata().getResourceVersion())) { - R lastEvent = eventRecorder.getLastEvent(resourceID); - - log.debug( - "Found events in event buffer but the target event is not last for id: {}. Propagating event.", - resourceID); - if (eventAcceptedByFilter(operation, newResource, oldResource)) { - propagateEvent(lastEvent); - } - } - } - } finally { - log.debug("Stopping event recording for: {}", resourceID); - eventRecorder.stopEventRecording(resourceID); - } + private void handleRecentCreateOrUpdate(Operation operation, R newResource, R oldResource) { + primaryToSecondaryIndex.onAddOrUpdate(newResource); + temporaryResourceCache.putResource(newResource, Optional.ofNullable(oldResource) + .map(r -> r.getMetadata().getResourceVersion()).orElse(null)); } private boolean useSecondaryToPrimaryIndex() { return this.primaryToSecondaryMapper == null; } - @Override - public synchronized void prepareForCreateOrUpdateEventFiltering(ResourceID resourceID, - R resource) { - log.debug("Starting event recording for: {}", resourceID); - eventRecorder.startEventRecording(resourceID); - } - - /** - * Mean to be called to clean up in case of an exception from the client. Usually in a catch - * block. - * - * @param resourceID to cleanup - */ - @Override - public synchronized void cleanupOnCreateOrUpdateEventFiltering(ResourceID resourceID) { - log.debug("Stopping event recording for: {}", resourceID); - eventRecorder.stopEventRecording(resourceID); - } - @Override public boolean allowsNamespaceChanges() { return getConfiguration().followControllerNamespaceChanges(); @@ -361,6 +295,7 @@ private boolean acceptedByDeleteFilters(R resource, boolean b) { // Since this event source instance is created by the user, the ConfigurationService is actually // injected after it is registered. Some of the subcomponents are initialized at that time here. + @Override public void setConfigurationService(ConfigurationService configurationService) { super.setConfigurationService(configurationService); @@ -368,6 +303,7 @@ public void setConfigurationService(ConfigurationService configurationService) { indexerBuffer = null; } + @Override public void addIndexers(Map>> indexers) { if (indexerBuffer == null) { throw new OperatorException("Cannot add indexers after InformerEventSource started."); @@ -375,4 +311,16 @@ public void addIndexers(Map>> indexers) { indexerBuffer.putAll(indexers); } + /** + * Add an annotation to the resource so that the subsequent will be omitted + * + * @param resourceVersion null if there is no prior version + * @param target mutable resource that will be returned + */ + public R addPreviousAnnotation(String resourceVersion, R target) { + target.getMetadata().getAnnotations().put(PREVIOUS_ANNOTATION_KEY, + id + Optional.ofNullable(resourceVersion).map(rv -> "," + rv).orElse("")); + return target; + } + } 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 92a317096c..a7d3e5caa7 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 @@ -91,7 +91,7 @@ public void stop() { @Override public void handleRecentResourceUpdate(ResourceID resourceID, R resource, R previousVersionOfResource) { - temporaryResourceCache.putUpdatedResource(resource, + temporaryResourceCache.putResource(resource, previousVersionOfResource.getMetadata().getResourceVersion()); } @@ -128,8 +128,10 @@ void setTemporalResourceCache(TemporaryResourceCache temporaryResourceCache) this.temporaryResourceCache = temporaryResourceCache; } + @Override public abstract void addIndexers(Map>> indexers); + @Override public List byIndex(String indexName, String indexKey) { return manager().byIndex(indexName, indexKey); } 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 8c3d56c008..233b409f3f 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 @@ -41,40 +41,37 @@ public TemporaryResourceCache(ManagedInformerEventSource managedInforme this.managedInformerEventSource = managedInformerEventSource; } - public synchronized void removeResourceFromCache(T resource) { - cache.remove(ResourceID.fromResource(resource)); - } - - public synchronized void unconditionallyCacheResource(T newResource) { - putToCache(newResource, null); + public synchronized Optional removeResourceFromCache(T resource) { + return Optional.ofNullable(cache.remove(ResourceID.fromResource(resource))); } public synchronized void putAddedResource(T newResource) { - ResourceID resourceID = ResourceID.fromResource(newResource); - if (managedInformerEventSource.get(resourceID).isEmpty()) { - log.debug("Putting resource to cache with ID: {}", resourceID); - putToCache(newResource, resourceID); - } else { - log.debug("Won't put resource into cache found already informer cache: {}", resourceID); - } + putResource(newResource, null); } - public synchronized void putUpdatedResource(T newResource, String previousResourceVersion) { + /** + * put the item into the cache if the previousResourceVersion matches the current state. If not + * the currently cached item is removed. + * + * @param previousResourceVersion null indicates an add + */ + public synchronized void putResource(T newResource, String previousResourceVersion) { var resourceId = ResourceID.fromResource(newResource); - var informerCacheResource = managedInformerEventSource.get(resourceId); - if (informerCacheResource.isEmpty()) { - log.debug("No cached value present for resource: {}", newResource); - return; - } - // if this is not true that means the cache was already updated - if (informerCacheResource.get().getMetadata().getResourceVersion() - .equals(previousResourceVersion)) { - log.debug("Putting resource to temporal cache with id: {}", resourceId); + var cachedResource = getResourceFromCache(resourceId) + .orElse(managedInformerEventSource.get(resourceId).orElse(null)); + + if ((previousResourceVersion == null && cachedResource == null) + || (cachedResource != null && previousResourceVersion != null + && cachedResource.getMetadata().getResourceVersion() + .equals(previousResourceVersion))) { + log.debug( + "Temporarily moving ahead to target version {} for resource id: {}", + newResource.getMetadata().getResourceVersion(), resourceId); putToCache(newResource, resourceId); } else { - // if something is in cache it's surely obsolete now - log.debug("Trying to remove an obsolete resource from cache for id: {}", resourceId); - cache.remove(resourceId); + if (cache.remove(resourceId) != null) { + log.debug("Removed an obsolete resource from cache for id: {}", resourceId); + } } } diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/EventRecorderTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/EventRecorderTest.java deleted file mode 100644 index 556ad089ff..0000000000 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/EventRecorderTest.java +++ /dev/null @@ -1,81 +0,0 @@ -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.ObjectMeta; -import io.javaoperatorsdk.operator.processing.event.ResourceID; - -import static org.assertj.core.api.Assertions.assertThat; - -class EventRecorderTest { - - public static final String RESOURCE_VERSION = "0"; - public static final String RESOURCE_VERSION1 = "1"; - - EventRecorder eventRecorder = new EventRecorder<>(); - - ConfigMap testConfigMap = testConfigMap(RESOURCE_VERSION); - ConfigMap testConfigMap2 = testConfigMap(RESOURCE_VERSION1); - - ResourceID id = ResourceID.fromResource(testConfigMap); - - @Test - void recordsEvents() { - - assertThat(eventRecorder.isRecordingFor(id)).isFalse(); - - eventRecorder.startEventRecording(id); - assertThat(eventRecorder.isRecordingFor(id)).isTrue(); - - eventRecorder.recordEvent(testConfigMap); - - eventRecorder.stopEventRecording(id); - assertThat(eventRecorder.isRecordingFor(id)).isFalse(); - } - - @Test - void getsLastRecorded() { - eventRecorder.startEventRecording(id); - - eventRecorder.recordEvent(testConfigMap); - eventRecorder.recordEvent(testConfigMap2); - - assertThat(eventRecorder.getLastEvent(id)).isEqualTo(testConfigMap2); - } - - @Test - void checksContainsWithResourceVersion() { - eventRecorder.startEventRecording(id); - - eventRecorder.recordEvent(testConfigMap); - eventRecorder.recordEvent(testConfigMap2); - - assertThat(eventRecorder.containsEventWithResourceVersion(id, RESOURCE_VERSION)).isTrue(); - assertThat(eventRecorder.containsEventWithResourceVersion(id, RESOURCE_VERSION1)).isTrue(); - assertThat(eventRecorder.containsEventWithResourceVersion(id, "xxx")).isFalse(); - } - - @Test - void checkLastItemVersion() { - eventRecorder.startEventRecording(id); - - eventRecorder.recordEvent(testConfigMap); - eventRecorder.recordEvent(testConfigMap2); - - assertThat(eventRecorder.containsEventWithVersionButItsNotLastOne(id, RESOURCE_VERSION)) - .isTrue(); - assertThat(eventRecorder.containsEventWithVersionButItsNotLastOne(id, RESOURCE_VERSION1)) - .isFalse(); - } - - ConfigMap testConfigMap(String resourceVersion) { - ConfigMap configMap = new ConfigMap(); - configMap.setMetadata(new ObjectMeta()); - configMap.getMetadata().setName("test"); - configMap.getMetadata().setResourceVersion(resourceVersion); - - return configMap; - } - -} 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 69e26f0b35..7acecc7099 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 @@ -8,6 +8,7 @@ import io.fabric8.kubernetes.api.model.ObjectMeta; import io.fabric8.kubernetes.api.model.apps.Deployment; +import io.fabric8.kubernetes.api.model.apps.DeploymentBuilder; import io.fabric8.kubernetes.client.KubernetesClient; import io.javaoperatorsdk.operator.MockKubernetesClient; import io.javaoperatorsdk.operator.OperatorException; @@ -36,7 +37,6 @@ class InformerEventSourceTest { private static final String PREV_RESOURCE_VERSION = "0"; private static final String DEFAULT_RESOURCE_VERSION = "1"; - private static final String NEXT_RESOURCE_VERSION = "2"; private InformerEventSource informerEventSource; private final KubernetesClient clientMock = MockKubernetesClient.client(Deployment.class); @@ -78,126 +78,48 @@ void skipsEventPropagationIfResourceWithSameVersionInResourceCache() { } @Test - void propagateEventAndRemoveResourceFromTempCacheIfResourceVersionMismatch() { - Deployment cachedDeployment = testDeployment(); - cachedDeployment.getMetadata().setResourceVersion(PREV_RESOURCE_VERSION); - when(temporaryResourceCacheMock.getResourceFromCache(any())) - .thenReturn(Optional.of(cachedDeployment)); - - - informerEventSource.onUpdate(cachedDeployment, testDeployment()); + void skipsAddEventPropagationViaAnnotation() { + informerEventSource.onAdd(informerEventSource.addPreviousAnnotation(null, testDeployment())); - verify(eventHandlerMock, times(1)).handleEvent(any()); - verify(temporaryResourceCacheMock, times(1)).removeResourceFromCache(any()); + verify(eventHandlerMock, never()).handleEvent(any()); } @Test - void notPropagatesEventIfAfterUpdateReceivedJustTheRelatedEvent() { - var testDeployment = testDeployment(); - var prevTestDeployment = testDeployment(); - prevTestDeployment.getMetadata().setResourceVersion(PREV_RESOURCE_VERSION); - + void skipsUpdateEventPropagationViaAnnotation() { + informerEventSource.onUpdate(testDeployment(), + informerEventSource.addPreviousAnnotation("1", testDeployment())); - informerEventSource - .prepareForCreateOrUpdateEventFiltering(ResourceID.fromResource(testDeployment), - testDeployment); - informerEventSource.onUpdate(prevTestDeployment, testDeployment); - informerEventSource.handleRecentResourceUpdate(ResourceID.fromResource(testDeployment), - testDeployment, prevTestDeployment); - - verify(eventHandlerMock, times(0)).handleEvent(any()); - verify(temporaryResourceCacheMock, times(0)).unconditionallyCacheResource(any()); + verify(eventHandlerMock, never()).handleEvent(any()); } - @Test - void notPropagatesEventIfAfterCreateReceivedJustTheRelatedEvent() { - var testDeployment = testDeployment(); - - informerEventSource - .prepareForCreateOrUpdateEventFiltering(ResourceID.fromResource(testDeployment), - testDeployment); - informerEventSource.onAdd(testDeployment); - informerEventSource.handleRecentResourceCreate(ResourceID.fromResource(testDeployment), - testDeployment); - - verify(eventHandlerMock, times(0)).handleEvent(any()); - verify(temporaryResourceCacheMock, times(0)).unconditionallyCacheResource(any()); + void processEventPropagationWithoutAnnotation() { + informerEventSource.onUpdate(testDeployment(), testDeployment()); + + verify(eventHandlerMock, times(1)).handleEvent(any()); } @Test - void propagatesEventIfNewEventReceivedAfterTheCurrentTargetEvent() { - var testDeployment = testDeployment(); - var prevTestDeployment = testDeployment(); - prevTestDeployment.getMetadata().setResourceVersion(PREV_RESOURCE_VERSION); - var nextTestDeployment = testDeployment(); - nextTestDeployment.getMetadata().setResourceVersion(NEXT_RESOURCE_VERSION); - - informerEventSource - .prepareForCreateOrUpdateEventFiltering(ResourceID.fromResource(testDeployment), - testDeployment); - informerEventSource.onUpdate(prevTestDeployment, testDeployment); - informerEventSource.onUpdate(testDeployment, nextTestDeployment); - informerEventSource.handleRecentResourceUpdate(ResourceID.fromResource(testDeployment), - testDeployment, prevTestDeployment); + void processEventPropagationWithIncorrectAnnotation() { + informerEventSource.onAdd(new DeploymentBuilder(testDeployment()).editMetadata() + .addToAnnotations(InformerEventSource.PREVIOUS_ANNOTATION_KEY, "invalid") + .endMetadata().build()); verify(eventHandlerMock, times(1)).handleEvent(any()); - verify(temporaryResourceCacheMock, times(0)).unconditionallyCacheResource(any()); } @Test - void notPropagatesEventIfMoreReceivedButTheLastIsTheUpdated() { - var testDeployment = testDeployment(); - var prevTestDeployment = testDeployment(); - prevTestDeployment.getMetadata().setResourceVersion(PREV_RESOURCE_VERSION); - var prevPrevTestDeployment = testDeployment(); - prevPrevTestDeployment.getMetadata().setResourceVersion("-1"); - - informerEventSource - .prepareForCreateOrUpdateEventFiltering(ResourceID.fromResource(testDeployment), - testDeployment); - informerEventSource.onUpdate(prevPrevTestDeployment, prevTestDeployment); - informerEventSource.onUpdate(prevTestDeployment, testDeployment); - informerEventSource.handleRecentResourceUpdate(ResourceID.fromResource(testDeployment), - testDeployment, prevTestDeployment); - - verify(eventHandlerMock, times(0)).handleEvent(any()); - verify(temporaryResourceCacheMock, times(0)).unconditionallyCacheResource(any()); - } + void propagateEventAndRemoveResourceFromTempCacheIfResourceVersionMismatch() { + Deployment cachedDeployment = testDeployment(); + cachedDeployment.getMetadata().setResourceVersion(PREV_RESOURCE_VERSION); + when(temporaryResourceCacheMock.getResourceFromCache(any())) + .thenReturn(Optional.of(cachedDeployment)); - @Test - void putsResourceOnTempCacheIfNoEventRecorded() { - var testDeployment = testDeployment(); - var prevTestDeployment = testDeployment(); - prevTestDeployment.getMetadata().setResourceVersion(PREV_RESOURCE_VERSION); - - informerEventSource - .prepareForCreateOrUpdateEventFiltering(ResourceID.fromResource(testDeployment), - testDeployment); - informerEventSource.handleRecentResourceUpdate(ResourceID.fromResource(testDeployment), - testDeployment, prevTestDeployment); - - verify(eventHandlerMock, times(0)).handleEvent(any()); - verify(temporaryResourceCacheMock, times(1)).unconditionallyCacheResource(any()); - } - @Test - void putsResourceOnTempCacheIfNoEventRecordedWithSameResourceVersion() { - var testDeployment = testDeployment(); - var prevTestDeployment = testDeployment(); - prevTestDeployment.getMetadata().setResourceVersion(PREV_RESOURCE_VERSION); - var prevPrevTestDeployment = testDeployment(); - prevPrevTestDeployment.getMetadata().setResourceVersion("-1"); - - informerEventSource - .prepareForCreateOrUpdateEventFiltering(ResourceID.fromResource(testDeployment), - testDeployment); - informerEventSource.onUpdate(prevPrevTestDeployment, prevTestDeployment); - informerEventSource.handleRecentResourceUpdate(ResourceID.fromResource(testDeployment), - testDeployment, prevTestDeployment); - - verify(eventHandlerMock, times(0)).handleEvent(any()); - verify(temporaryResourceCacheMock, times(1)).unconditionallyCacheResource(any()); + informerEventSource.onUpdate(cachedDeployment, testDeployment()); + + verify(eventHandlerMock, times(1)).handleEvent(any()); + verify(temporaryResourceCacheMock, times(1)).removeResourceFromCache(any()); } @Test 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 f848e26cec..4d5bdf0dfd 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 @@ -30,7 +30,7 @@ void updateAddsTheResourceIntoCacheIfTheInformerHasThePreviousResourceVersion() prevTestResource.getMetadata().setResourceVersion("0"); when(informerEventSource.get(any())).thenReturn(Optional.of(prevTestResource)); - temporaryResourceCache.putUpdatedResource(testResource, "0"); + temporaryResourceCache.putResource(testResource, "0"); var cached = temporaryResourceCache.getResourceFromCache(ResourceID.fromResource(testResource)); assertThat(cached).isPresent(); @@ -43,7 +43,7 @@ void updateNotAddsTheResourceIntoCacheIfTheInformerHasOtherVersion() { informerCachedResource.getMetadata().setResourceVersion("x"); when(informerEventSource.get(any())).thenReturn(Optional.of(informerCachedResource)); - temporaryResourceCache.putUpdatedResource(testResource, "0"); + temporaryResourceCache.putResource(testResource, "0"); var cached = temporaryResourceCache.getResourceFromCache(ResourceID.fromResource(testResource)); assertThat(cached).isNotPresent(); diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/createupdateeventfilter/CreateUpdateEventFilterTestReconciler.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/createupdateeventfilter/CreateUpdateEventFilterTestReconciler.java index ef2ddfc4ec..6c38b2dd09 100644 --- a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/createupdateeventfilter/CreateUpdateEventFilterTestReconciler.java +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/createupdateeventfilter/CreateUpdateEventFilterTestReconciler.java @@ -16,7 +16,7 @@ import io.javaoperatorsdk.operator.api.reconciler.Reconciler; import io.javaoperatorsdk.operator.api.reconciler.UpdateControl; import io.javaoperatorsdk.operator.junit.KubernetesClientAware; -import io.javaoperatorsdk.operator.processing.event.ResourceID; +import io.javaoperatorsdk.operator.processing.dependent.kubernetes.CRUDKubernetesDependentResource; import io.javaoperatorsdk.operator.processing.event.source.EventSource; import io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource; @@ -26,10 +26,35 @@ public class CreateUpdateEventFilterTestReconciler EventSourceInitializer, KubernetesClientAware { + private static final class DirectConfigMapDependentResource + extends + CRUDKubernetesDependentResource { + + private ConfigMap desired; + + private DirectConfigMapDependentResource(Class resourceType) { + super(resourceType); + } + + @Override + protected ConfigMap desired(CreateUpdateEventFilterTestCustomResource primary, + Context context) { + return desired; + } + + @Override + public void setEventSource( + io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource eventSource) { + super.setEventSource(eventSource); + } + } + public static final String CONFIG_MAP_TEST_DATA_KEY = "key"; private KubernetesClient client; private final AtomicInteger numberOfExecutions = new AtomicInteger(0); private InformerEventSource informerEventSource; + private DirectConfigMapDependentResource configMapDR = + new DirectConfigMapDependentResource(ConfigMap.class); @Override public UpdateControl reconcile( @@ -44,43 +69,14 @@ public UpdateControl reconcile( .withName(resource.getMetadata().getName()) .get(); if (configMap == null) { - var configMapToCreate = createConfigMap(resource); - final var resourceID = ResourceID.fromResource(configMapToCreate); - try { - informerEventSource.prepareForCreateOrUpdateEventFiltering(resourceID, configMapToCreate); - configMap = - client - .configMaps() - .inNamespace(resource.getMetadata().getNamespace()) - .resource(configMapToCreate) - .create(); - informerEventSource.handleRecentResourceCreate(resourceID, configMap); - } catch (RuntimeException e) { - informerEventSource - .cleanupOnCreateOrUpdateEventFiltering(resourceID); - throw e; - } + configMapDR.desired = createConfigMap(resource); + configMapDR.reconcile(resource, context); } else { - ResourceID resourceID = ResourceID.fromResource(configMap); if (!Objects.equals( configMap.getData().get(CONFIG_MAP_TEST_DATA_KEY), resource.getSpec().getValue())) { configMap.getData().put(CONFIG_MAP_TEST_DATA_KEY, resource.getSpec().getValue()); - try { - informerEventSource - .prepareForCreateOrUpdateEventFiltering(resourceID, configMap); - var newConfigMap = - client - .configMaps() - .inNamespace(resource.getMetadata().getNamespace()) - .resource(configMap) - .replace(); - informerEventSource.handleRecentResourceUpdate(resourceID, - newConfigMap, configMap); - } catch (RuntimeException e) { - informerEventSource - .cleanupOnCreateOrUpdateEventFiltering(resourceID); - throw e; - } + configMapDR.desired = configMap; + configMapDR.reconcile(resource, context); } } return UpdateControl.noUpdate(); @@ -109,6 +105,10 @@ public Map prepareEventSources( .build(); informerEventSource = new InformerEventSource<>(informerConfiguration, client); + + this.configMapDR.setKubernetesClient(context.getClient()); + this.configMapDR.setEventSource(informerEventSource); + return EventSourceInitializer.nameEventSources(informerEventSource); } diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/primaryindexer/DependentPrimaryIndexerTestReconciler.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/primaryindexer/DependentPrimaryIndexerTestReconciler.java index 6d79d4ee56..e123500c42 100644 --- a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/primaryindexer/DependentPrimaryIndexerTestReconciler.java +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/primaryindexer/DependentPrimaryIndexerTestReconciler.java @@ -12,8 +12,8 @@ import io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependentResource; import io.javaoperatorsdk.operator.processing.event.ResourceID; import io.javaoperatorsdk.operator.processing.event.source.IndexerResourceCache; -import io.javaoperatorsdk.operator.processing.event.source.ResourceEventSource; import io.javaoperatorsdk.operator.processing.event.source.SecondaryToPrimaryMapper; +import io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource; @ControllerConfiguration(dependents = @Dependent( type = DependentPrimaryIndexerTestReconciler.ReadOnlyConfigMapDependent.class)) @@ -39,7 +39,7 @@ public Set toPrimaryResourceIDs(ConfigMap dependentResource) { } @Override - public Optional> eventSource( + public Optional> eventSource( EventSourceContext context) { cache = context.getPrimaryCache(); cache.addIndexer(CONFIG_MAP_RELATION_INDEXER, indexer);