Removing events from context (#596)

* fix: WIP

* fix: Addressing Custom Resource by Name and Namespace refactor + Informer Cache WIP

* fix: Build is fixed, (test failing)

* fix: Test fixes

* fix: minor update

* fix: EventSourceManager small fix

* fix: Unit tests fixed

* fix: DefaultEventHandler init from EventSourceManager

* fix: custom resource selector test improvement

* fix: wip test imrpovements

* fix: test improvements

* fix: further improvements

* fix: build

* feature: add mvn jar to gitignore

* Exposing CustomResourceEventSource and informers

* fix: cleanup

* fix: remove caching optimization since it not possible anymore with informer

* fix:  formatting

* refactor: make name/namespace final

* feature: Simple label selector support

* fix: formatting

* fix: code inspection reports

* fix: merge from v2

* fix: removed most deprecated apis

* wip: started to remove events from variouse layers

* fix: progress with implementation and tests

* fix: Updated informer mapping to CustomResourceID

* fix: imports

* fix: decorational changes

* fix: event marker unit test

* Default Event Handler Unit tests

* fix: fixes after merge

* fix: changes from code review

* fix: method naming

* Update operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/EventMarker.java

Co-authored-by: Chris Laprun <[email protected]>

* Update operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/EventMarker.java

Co-authored-by: Chris Laprun <[email protected]>

* fix: comment

* fix: fixes from merge

* fix: remove not used method

* fix: formatting

Co-authored-by: Chris Laprun <[email protected]>
Co-authored-by: Chris Laprun <[email protected]>

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/Context.java

@@ -3,11 +3,9 @@
 import java.util.Optional;
 
 import io.fabric8.kubernetes.client.CustomResource;
-import io.javaoperatorsdk.operator.processing.event.EventList;
 
 public interface Context<T extends CustomResource> {
 
-  EventList getEvents();
-
   Optional<RetryInfo> getRetryInfo();
+
 }

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/DefaultContext.java

@@ -3,21 +3,13 @@
 import java.util.Optional;
 
 import io.fabric8.kubernetes.client.CustomResource;
-import io.javaoperatorsdk.operator.processing.event.EventList;
 
 public class DefaultContext<T extends CustomResource> implements Context<T> {
 
   private final RetryInfo retryInfo;
-  private final EventList events;
 
-  public DefaultContext(EventList events, RetryInfo retryInfo) {
+  public DefaultContext(RetryInfo retryInfo) {
     this.retryInfo = retryInfo;
-    this.events = events;
-  }
-
-  @Override
-  public EventList getEvents() {
-    return events;
   }
 
   @Override

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/ResourceController.java

@@ -6,12 +6,15 @@
 public interface ResourceController<R extends CustomResource> {
 
   /**
+   * Note that this method is used in combination of finalizers. If automatic finalizer handling is
+   * turned off for the controller, this method is not called.
+   *
    * The implementation should delete the associated component(s). Note that this is method is
    * called when an object is marked for deletion. After it's executed the custom resource finalizer
    * is automatically removed by the framework; unless the return value is
    * {@link DeleteControl#noFinalizerRemoval()}, which indicates that the controller has determined
-   * that the resource should not be deleted yet, in which case it is up to the controller to
-   * restore the resource's status so that it's not marked for deletion anymore.
+   * that the resource should not be deleted yet. This is usually a corner case, when a cleanup is
+   * tried again eventually.
    *
    * <p>
    * It's important that this method be idempotent, as it could be called several times, depending

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

@@ -20,11 +20,12 @@
 import io.javaoperatorsdk.operator.processing.event.DefaultEventSourceManager;
 import io.javaoperatorsdk.operator.processing.event.Event;
 import io.javaoperatorsdk.operator.processing.event.EventHandler;
+import io.javaoperatorsdk.operator.processing.event.internal.CustomResourceEvent;
+import io.javaoperatorsdk.operator.processing.event.internal.ResourceAction;
 import io.javaoperatorsdk.operator.processing.retry.GenericRetry;
 import io.javaoperatorsdk.operator.processing.retry.Retry;
 import io.javaoperatorsdk.operator.processing.retry.RetryExecution;
 
-import static io.javaoperatorsdk.operator.EventListUtils.containsCustomResourceDeletedEvent;
 import static io.javaoperatorsdk.operator.processing.KubernetesResourceUtils.getName;
 
 /**
@@ -38,7 +39,6 @@ public class DefaultEventHandler<R extends CustomResource<?, ?>> implements Even
   @Deprecated
   private static EventMonitor monitor = EventMonitor.NOOP;
 
-  private final EventBuffer eventBuffer;
   private final Set<CustomResourceID> underProcessing = new HashSet<>();
   private final EventDispatcher<R> eventDispatcher;
   private final Retry retry;
@@ -50,6 +50,7 @@ public class DefaultEventHandler<R extends CustomResource<?, ?>> implements Even
   private volatile boolean running;
   private final ResourceCache<R> resourceCache;
   private DefaultEventSourceManager<R> eventSourceManager;
+  private final EventMarker eventMarker;
 
   public DefaultEventHandler(ConfiguredController<R> controller, ResourceCache<R> resourceCache) {
     this(
@@ -58,18 +59,20 @@ public DefaultEventHandler(ConfiguredController<R> controller, ResourceCache<R>
         controller.getConfiguration().getName(),
         new EventDispatcher<>(controller),
         GenericRetry.fromConfiguration(controller.getConfiguration().getRetryConfiguration()),
-        controller.getConfiguration().getConfigurationService().getMetrics().getEventMonitor());
+        controller.getConfiguration().getConfigurationService().getMetrics().getEventMonitor(),
+        new EventMarker());
   }
 
   DefaultEventHandler(EventDispatcher<R> eventDispatcher, ResourceCache<R> resourceCache,
       String relatedControllerName,
-      Retry retry) {
-    this(resourceCache, null, relatedControllerName, eventDispatcher, retry, null);
+      Retry retry, EventMarker eventMarker) {
+    this(resourceCache, null, relatedControllerName, eventDispatcher, retry, null, eventMarker);
   }
 
   private DefaultEventHandler(ResourceCache<R> resourceCache, ExecutorService executor,
       String relatedControllerName,
-      EventDispatcher<R> eventDispatcher, Retry retry, EventMonitor monitor) {
+      EventDispatcher<R> eventDispatcher, Retry retry, EventMonitor monitor,
+      EventMarker eventMarker) {
     this.running = true;
     this.executor =
         executor == null
@@ -79,9 +82,9 @@ private DefaultEventHandler(ResourceCache<R> resourceCache, ExecutorService exec
     this.controllerName = relatedControllerName;
     this.eventDispatcher = eventDispatcher;
     this.retry = retry;
-    eventBuffer = new EventBuffer();
     this.resourceCache = resourceCache;
     this.eventMonitor = monitor != null ? monitor : EventMonitor.NOOP;
+    this.eventMarker = eventMarker;
   }
 
   public void setEventSourceManager(DefaultEventSourceManager<R> eventSourceManager) {
@@ -113,71 +116,75 @@ private EventMonitor monitor() {
 
   @Override
   public void handleEvent(Event event) {
+    lock.lock();
     try {
-      lock.lock();
       log.debug("Received event: {}", event);
       if (!this.running) {
         log.debug("Skipping event: {} because the event handler is shutting down", event);
         return;
       }
       final var monitor = monitor();
-      eventBuffer.addEvent(event.getRelatedCustomResourceID(), event);
       monitor.processedEvent(event.getRelatedCustomResourceID(), event);
-      executeBufferedEvents(event.getRelatedCustomResourceID());
-    } finally {
-      lock.unlock();
-    }
-  }
 
-  @Override
-  public void close() {
-    try {
-      lock.lock();
-      this.running = false;
+      handleEventMarking(event);
+      if (!eventMarker.deleteEventPresent(event.getRelatedCustomResourceID())) {
+        submitReconciliationExecution(event.getRelatedCustomResourceID());
+      } else {
+        cleanupForDeletedEvent(event.getRelatedCustomResourceID());
+      }
     } finally {
       lock.unlock();
     }
   }
 
-  private boolean executeBufferedEvents(CustomResourceID customResourceUid) {
-    boolean newEventForResourceId = eventBuffer.containsEvents(customResourceUid);
+  private boolean submitReconciliationExecution(CustomResourceID customResourceUid) {
     boolean controllerUnderExecution = isControllerUnderExecution(customResourceUid);
     Optional<R> latestCustomResource =
         resourceCache.getCustomResource(customResourceUid);
 
-    if (!controllerUnderExecution && newEventForResourceId && latestCustomResource.isPresent()) {
+    if (!controllerUnderExecution
+        && latestCustomResource.isPresent()) {
       setUnderExecutionProcessing(customResourceUid);
       ExecutionScope executionScope =
           new ExecutionScope(
-              eventBuffer.getAndRemoveEventsForExecution(customResourceUid),
               latestCustomResource.get(),
               retryInfo(customResourceUid));
+      eventMarker.unMarkEventReceived(customResourceUid);
       log.debug("Executing events for custom resource. Scope: {}", executionScope);
       executor.execute(new ControllerExecution(executionScope));
       return true;
     } else {
       log.debug(
-          "Skipping executing controller for resource id: {}. Events in queue: {}."
+          "Skipping executing controller for resource id: {}."
               + " Controller in execution: {}. Latest CustomResource present: {}",
           customResourceUid,
-          newEventForResourceId,
           controllerUnderExecution,
           latestCustomResource.isPresent());
       if (latestCustomResource.isEmpty()) {
-        log.warn("no custom resource found in cache for CustomResourceID: {}", customResourceUid);
+        log.warn("no custom resource found in cache for CustomResourceID: {}",
+            customResourceUid);
       }
       return false;
     }
   }
 
+  private void handleEventMarking(Event event) {
+    if (event instanceof CustomResourceEvent &&
+        ((CustomResourceEvent) event).getAction() == ResourceAction.DELETED) {
+      eventMarker.markDeleteEventReceived(event);
+    } else if (!eventMarker.deleteEventPresent(event.getRelatedCustomResourceID())) {
+      eventMarker.markEventReceived(event);
+    }
+  }
+
   private RetryInfo retryInfo(CustomResourceID customResourceUid) {
     return retryState.get(customResourceUid);
   }
 
   void eventProcessingFinished(
       ExecutionScope<R> executionScope, PostExecutionControl<R> postExecutionControl) {
+    lock.lock();
     try {
-      lock.lock();
       if (!running) {
         return;
       }
@@ -188,23 +195,29 @@ void eventProcessingFinished(
           postExecutionControl);
       unsetUnderExecution(executionScope.getCustomResourceID());
 
-      if (retry != null && postExecutionControl.exceptionDuringExecution()) {
+      // If a delete event present at this phase, it was received during reconciliation.
+      // So we either removed the finalizer during reconciliation or we don't use finalizers.
+      // Either way we don't want to retry.
+      if (retry != null && postExecutionControl.exceptionDuringExecution() &&
+          !eventMarker.deleteEventPresent(executionScope.getCustomResourceID())) {
         handleRetryOnException(executionScope);
-        final var monitor = monitor();
-        executionScope.getEvents()
-            .forEach(e -> monitor.failedEvent(executionScope.getCustomResourceID(), e));
+        // todo revisit monitoring since events are not present anymore
+        // final var monitor = monitor(); executionScope.getEvents().forEach(e ->
+        // monitor.failedEvent(executionScope.getCustomResourceID(), e));
         return;
       }
 
       if (retry != null) {
-        markSuccessfulExecutionRegardingRetry(executionScope);
+        handleSuccessfulExecutionRegardingRetry(executionScope);
       }
-      if (containsCustomResourceDeletedEvent(executionScope.getEvents())) {
-        cleanupAfterDeletedEvent(executionScope.getCustomResourceID());
+      if (eventMarker.deleteEventPresent(executionScope.getCustomResourceID())) {
+        cleanupForDeletedEvent(executionScope.getCustomResourceID());
       } else {
-        var executed = executeBufferedEvents(executionScope.getCustomResourceID());
-        if (!executed) {
-          reScheduleExecutionIfInstructed(postExecutionControl, executionScope.getCustomResource());
+        if (eventMarker.eventPresent(executionScope.getCustomResourceID())) {
+          submitReconciliationExecution(executionScope.getCustomResourceID());
+        } else {
+          reScheduleExecutionIfInstructed(postExecutionControl,
+              executionScope.getCustomResource());
         }
       }
     } finally {
@@ -227,13 +240,13 @@ private void reScheduleExecutionIfInstructed(PostExecutionControl<R> postExecuti
   private void handleRetryOnException(ExecutionScope<R> executionScope) {
     RetryExecution execution = getOrInitRetryExecution(executionScope);
     var customResourceID = executionScope.getCustomResourceID();
-    boolean newEventsExists = eventBuffer
-        .newEventsExists(customResourceID);
-    eventBuffer.putBackEvents(customResourceID, executionScope.getEvents());
+    boolean eventPresent = eventMarker.eventPresent(customResourceID);
+    eventMarker.markEventReceived(customResourceID);
 
-    if (newEventsExists) {
-      log.debug("New events exists for for resource id: {}", customResourceID);
-      executeBufferedEvents(customResourceID);
+    if (eventPresent) {
+      log.debug("New events exists for for resource id: {}",
+          customResourceID);
+      submitReconciliationExecution(customResourceID);
       return;
     }
     Optional<Long> nextDelay = execution.nextDelay();
@@ -251,7 +264,7 @@ private void handleRetryOnException(ExecutionScope<R> executionScope) {
         () -> log.error("Exhausted retries for {}", executionScope));
   }
 
-  private void markSuccessfulExecutionRegardingRetry(ExecutionScope<R> executionScope) {
+  private void handleSuccessfulExecutionRegardingRetry(ExecutionScope<R> executionScope) {
     log.debug(
         "Marking successful execution for resource: {}",
         getName(executionScope.getCustomResource()));
@@ -270,9 +283,9 @@ private RetryExecution getOrInitRetryExecution(ExecutionScope<R> executionScope)
     return retryExecution;
   }
 
-  private void cleanupAfterDeletedEvent(CustomResourceID customResourceUid) {
+  private void cleanupForDeletedEvent(CustomResourceID customResourceUid) {
     eventSourceManager.cleanupForCustomResource(customResourceUid);
-    eventBuffer.cleanup(customResourceUid);
+    eventMarker.cleanup(customResourceUid);
   }
 
   private boolean isControllerUnderExecution(CustomResourceID customResourceUid) {
@@ -287,6 +300,15 @@ private void unsetUnderExecution(CustomResourceID customResourceUid) {
     underProcessing.remove(customResourceUid);
   }
 
+  @Override
+  public void close() {
+    lock.lock();
+    try {
+      this.running = false;
+    } finally {
+      lock.unlock();
+    }
+  }
 
   private class ControllerExecution implements Runnable {
     private final ExecutionScope<R> executionScope;

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

@@ -10,9 +10,7 @@
 import io.fabric8.kubernetes.client.dsl.Resource;
 import io.javaoperatorsdk.operator.api.*;
 import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
-import io.javaoperatorsdk.operator.processing.event.EventList;
 
-import static io.javaoperatorsdk.operator.EventListUtils.containsCustomResourceDeletedEvent;
 import static io.javaoperatorsdk.operator.processing.KubernetesResourceUtils.getName;
 import static io.javaoperatorsdk.operator.processing.KubernetesResourceUtils.getUID;
 import static io.javaoperatorsdk.operator.processing.KubernetesResourceUtils.getVersion;
@@ -55,15 +53,7 @@ public PostExecutionControl<R> handleExecution(ExecutionScope<R> executionScope)
 
   private PostExecutionControl<R> handleDispatch(ExecutionScope<R> executionScope) {
     R resource = executionScope.getCustomResource();
-    log.debug("Handling events: {} for resource {}", executionScope.getEvents(), getName(resource));
-
-    if (containsCustomResourceDeletedEvent(executionScope.getEvents())) {
-      log.debug(
-          "Skipping dispatch processing because of a Delete event: {} with version: {}",
-          getName(resource),
-          getVersion(resource));
-      return PostExecutionControl.defaultDispatch();
-    }
+    log.debug("Handling dispatch for resource {}", getName(resource));
 
     final var markedForDeletion = resource.isMarkedForDeletion();
     if (markedForDeletion && shouldNotDispatchToDelete(resource)) {
@@ -75,8 +65,7 @@ private PostExecutionControl<R> handleDispatch(ExecutionScope<R> executionScope)
     }
 
     Context<R> context =
-        new DefaultContext<>(
-            new EventList(executionScope.getEvents()), executionScope.getRetryInfo());
+        new DefaultContext<>(executionScope.getRetryInfo());
     if (markedForDeletion) {
       return handleDelete(resource, context);
     } else {