Temporal resource cache in Event Source

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/InformerConfiguration.java

@@ -37,6 +37,7 @@ protected DefaultInformerConfiguration(ConfigurationService service, String labe
           Objects.requireNonNullElseGet(associatedWith, () -> ResourceID::fromResource);
     }
 
+
     public PrimaryResourcesRetriever<R> getPrimaryResourcesRetriever() {
       return secondaryToPrimaryResourcesIdSet;
     }

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/Controller.java

@@ -214,7 +214,7 @@ public void start() throws OperatorException {
       }
 
       final var context = new EventSourceContext<>(
-          eventSourceManager.getControllerResourceEventSource().getResourceCache(),
+          eventSourceManager.getControllerResourceEventSource(),
           configurationService(), kubernetesClient);
 
       prepareEventSources(context).forEach(eventSourceManager::registerEventSource);

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependent.java

@@ -32,5 +32,4 @@
    * @return the label selector
    */
   String labelSelector() default EMPTY_STRING;
-
 }

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentResource.java

@@ -75,30 +75,46 @@ private void configureWith(ConfigurationService configService, String labelSelec
             .withPrimaryResourcesRetriever(primaryResourcesRetriever)
             .withAssociatedSecondaryResourceIdentifier(secondaryResourceIdentifier)
             .build();
-    configureWith(configService, new InformerEventSource<>(ic, client), addOwnerReference);
+    configureWith(new InformerEventSource<>(ic, client), addOwnerReference);
   }
 
   /**
    * Use to share informers between event more resources.
-   *
-   * @param configurationService get configs
+   * 
    * @param informerEventSource informer to use
    * @param addOwnerReference to the created resource
    */
-  public void configureWith(ConfigurationService configurationService,
+  public void configureWith(
       InformerEventSource<R, P> informerEventSource,
       boolean addOwnerReference) {
     this.informerEventSource = informerEventSource;
     this.addOwnerReference = addOwnerReference;
   }
 
   public void create(R target, P primary, Context context) {
-    prepare(target, primary, "Creating").create(target);
+    var resourceID = ResourceID.fromResource(target);
+    try {
+      informerEventSource.prepareForImminentCreateOrUpdateForResource(resourceID);
+      var created = prepare(target, primary, "Creating").create(target);
+      informerEventSource.handleRecentResourceCreate(created);
+    } catch (RuntimeException e) {
+      informerEventSource.cleanupOnUpdateAndCreate(resourceID);
+      throw e;
+    }
   }
 
   public void update(R actual, R target, P primary, Context context) {
-    var updatedActual = processor.replaceSpecOnActual(actual, target, context);
-    prepare(target, primary, "Updating").replace(updatedActual);
+    var resourceID = ResourceID.fromResource(target);
+    try {
+      var updatedActual = processor.replaceSpecOnActual(actual, target, context);
+      informerEventSource.prepareForImminentCreateOrUpdateForResource(resourceID);
+      var updated = prepare(target, primary, "Updating").replace(updatedActual);
+      informerEventSource.handleRecentResourceUpdate(updated,
+          actual.getMetadata().getResourceVersion());
+    } catch (RuntimeException e) {
+      informerEventSource.cleanupOnUpdateAndCreate(resourceID);
+      throw e;
+    }
   }
 
   public boolean match(R actualResource, R desiredResource, Context context) {

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventProcessor.java

@@ -29,7 +29,6 @@
 import io.javaoperatorsdk.operator.processing.retry.RetryExecution;
 
 import static io.javaoperatorsdk.operator.processing.KubernetesResourceUtils.getName;
-import static io.javaoperatorsdk.operator.processing.KubernetesResourceUtils.getVersion;
 
 class EventProcessor<R extends HasMetadata> implements EventHandler, LifecycleAware {
 
@@ -50,7 +49,7 @@ class EventProcessor<R extends HasMetadata> implements EventHandler, LifecycleAw
 
   EventProcessor(EventSourceManager<R> eventSourceManager) {
     this(
-        eventSourceManager.getControllerResourceEventSource().getResourceCache(),
+        eventSourceManager.getControllerResourceEventSource(),
         ExecutorServiceManager.instance().executorService(),
         eventSourceManager.getController().getConfiguration().getName(),
         new ReconciliationDispatcher<>(eventSourceManager.getController()),
@@ -73,7 +72,7 @@ class EventProcessor<R extends HasMetadata> implements EventHandler, LifecycleAw
       Retry retry,
       Metrics metrics) {
     this(
-        eventSourceManager.getControllerResourceEventSource().getResourceCache(),
+        eventSourceManager.getControllerResourceEventSource(),
         null,
         relatedControllerName,
         reconciliationDispatcher,
@@ -208,12 +207,12 @@ void eventProcessingFinished(
       if (eventMarker.deleteEventPresent(resourceID)) {
         cleanupForDeletedEvent(executionScope.getCustomResourceID());
       } else {
+        postExecutionControl.getUpdatedCustomResource().ifPresent(r -> {
+          eventSourceManager.getControllerResourceEventSource().handleRecentResourceUpdate(r,
+              executionScope.getResource().getMetadata().getResourceVersion());
+        });
         if (eventMarker.eventPresent(resourceID)) {
-          if (isCacheReadyForInstantReconciliation(executionScope, postExecutionControl)) {
-            submitReconciliationExecution(resourceID);
-          } else {
-            postponeReconciliationAndHandleCacheSyncEvent(resourceID);
-          }
+          submitReconciliationExecution(resourceID);
         } else {
           reScheduleExecutionIfInstructed(postExecutionControl, executionScope.getResource());
         }
@@ -223,41 +222,6 @@ void eventProcessingFinished(
     }
   }
 
-  private void postponeReconciliationAndHandleCacheSyncEvent(ResourceID resourceID) {
-    eventSourceManager.getControllerResourceEventSource().whitelistNextEvent(resourceID);
-  }
-
-  private boolean isCacheReadyForInstantReconciliation(
-      ExecutionScope<R> executionScope, PostExecutionControl<R> postExecutionControl) {
-    if (!postExecutionControl.customResourceUpdatedDuringExecution()) {
-      return true;
-    }
-    String originalResourceVersion = getVersion(executionScope.getResource());
-    String customResourceVersionAfterExecution =
-        getVersion(
-            postExecutionControl
-                .getUpdatedCustomResource()
-                .orElseThrow(
-                    () -> new IllegalStateException(
-                        "Updated custom resource must be present at this point of time")));
-    String cachedCustomResourceVersion =
-        getVersion(
-            cache
-                .get(executionScope.getCustomResourceID())
-                .orElseThrow(
-                    () -> new IllegalStateException(
-                        "Cached custom resource must be present at this point")));
-
-    if (cachedCustomResourceVersion.equals(customResourceVersionAfterExecution)) {
-      return true;
-    }
-    // If the cached resource version equals neither the version before nor after execution
-    // probably an update happened on the custom resource independent of the framework during
-    // reconciliation. We cannot tell at this point if it happened before our update or before.
-    // (Well we could if we would parse resource version, but that should not be done by definition)
-    return !cachedCustomResourceVersion.equals(originalResourceVersion);
-  }
-
   private void reScheduleExecutionIfInstructed(
       PostExecutionControl<R> postExecutionControl, R customResource) {
     postExecutionControl

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/ControllerResourceEventSource.java

@@ -13,7 +13,6 @@
 import io.javaoperatorsdk.operator.processing.Controller;
 import io.javaoperatorsdk.operator.processing.MDCUtils;
 import io.javaoperatorsdk.operator.processing.event.ResourceID;
-import io.javaoperatorsdk.operator.processing.event.source.ResourceCache;
 import io.javaoperatorsdk.operator.processing.event.source.informer.ManagedInformerEventSource;
 
 import static io.javaoperatorsdk.operator.processing.KubernetesResourceUtils.getName;
@@ -25,30 +24,19 @@ public class ControllerResourceEventSource<T extends HasMetadata>
     implements ResourceEventHandler<T> {
 
   public static final String ANY_NAMESPACE_MAP_KEY = "anyNamespace";
-
   private static final Logger log = LoggerFactory.getLogger(ControllerResourceEventSource.class);
 
   private final Controller<T> controller;
   private final ResourceEventFilter<T> filter;
-  private final OnceWhitelistEventFilterEventFilter<T> onceWhitelistEventFilterEventFilter;
 
   public ControllerResourceEventSource(Controller<T> controller) {
     super(controller.getCRClient(), controller.getConfiguration());
     this.controller = controller;
-
     var filters = new ResourceEventFilter[] {
         ResourceEventFilters.finalizerNeededAndApplied(),
         ResourceEventFilters.markedForDeletion(),
         ResourceEventFilters.generationAware(),
-        null
     };
-
-    if (controller.getConfiguration().isGenerationAware()) {
-      onceWhitelistEventFilterEventFilter = new OnceWhitelistEventFilterEventFilter<>();
-      filters[filters.length - 1] = onceWhitelistEventFilterEventFilter;
-    } else {
-      onceWhitelistEventFilterEventFilter = null;
-    }
     if (controller.getConfiguration().getEventFilter() != null) {
       filter = controller.getConfiguration().getEventFilter().and(ResourceEventFilters.or(filters));
     } else {
@@ -87,36 +75,22 @@ public void eventReceived(ResourceAction action, T resource, T oldResource) {
 
   @Override
   public void onAdd(T resource) {
+    super.onAdd(resource);
     eventReceived(ResourceAction.ADDED, resource, null);
   }
 
   @Override
   public void onUpdate(T oldCustomResource, T newCustomResource) {
+    super.onUpdate(oldCustomResource, newCustomResource);
     eventReceived(ResourceAction.UPDATED, newCustomResource, oldCustomResource);
   }
 
   @Override
   public void onDelete(T resource, boolean b) {
+    super.onDelete(resource, b);
     eventReceived(ResourceAction.DELETED, resource, null);
   }
 
-  public ResourceCache<T> getResourceCache() {
-    return manager();
-  }
-
-  /**
-   * This will ensure that the next event received after this method is called will not be filtered
-   * out.
-   *
-   * @param resourceID - to which the event is related
-   */
-  public void whitelistNextEvent(ResourceID resourceID) {
-    if (onceWhitelistEventFilterEventFilter != null) {
-      onceWhitelistEventFilterEventFilter.whitelistNextEvent(resourceID);
-    }
-  }
-
-
   private void handleKubernetesClientException(Exception e) {
     KubernetesClientException ke = (KubernetesClientException) e;
     if (404 == ke.getCode()) {

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/EventBuffer.java

@@ -0,0 +1,65 @@
+package io.javaoperatorsdk.operator.processing.event.source.informer;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.processing.event.ResourceID;
+
+public class EventBuffer<R extends HasMetadata> {
+
+  private final Map<ResourceID, ArrayList<R>> resourceEvents = new ConcurrentHashMap<>();
+
+  void startEventRecording(ResourceID resourceID) {
+    resourceEvents.putIfAbsent(resourceID, new ArrayList<>());
+  }
+
+  public boolean isEventsRecordedFor(ResourceID resourceID) {
+    return resourceEvents.get(resourceID) != null;
+  }
+
+  public void stopEventRecording(ResourceID resourceID) {
+    resourceEvents.remove(resourceID);
+  }
+
+  public void eventReceived(R resource) {
+    resourceEvents.get(ResourceID.fromResource(resource)).add(resource);
+  }
+
+  public boolean containsEventsFor(ResourceID resourceID) {
+    return !resourceEvents.get(resourceID).isEmpty();
+  }
+
+  public boolean containsEventWithResourceVersion(ResourceID resourceID, String resourceVersion) {
+    List<R> 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<R> resources = resourceEvents.get(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<R> resources = resourceEvents.get(resourceID);
+    return resources.get(resources.size() - 1);
+  }
+}