Removing events from context

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/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,45 +116,41 @@ 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;
+      markEvent(event);
+      if (eventMarker.isEventPresent(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 newEventForResourceId = eventMarker.isEventPresent(customResourceUid);
     boolean controllerUnderExecution = isControllerUnderExecution(customResourceUid);
     Optional<R> latestCustomResource =
         resourceCache.getCustomResource(customResourceUid);
 
-    if (!controllerUnderExecution && newEventForResourceId && latestCustomResource.isPresent()) {
+    if (!controllerUnderExecution && newEventForResourceId
+        && 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;
@@ -164,20 +163,30 @@ private boolean executeBufferedEvents(CustomResourceID customResourceUid) {
           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 markEvent(Event event) {
+    if (event instanceof CustomResourceEvent &&
+        ((CustomResourceEvent) event).getAction() == ResourceAction.DELETED) {
+      eventMarker.markDeleteEventReceived(event);
+    } else if (!eventMarker.isDeleteEventPresent(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;
       }
@@ -190,28 +199,34 @@ void eventProcessingFinished(
 
       if (retry != null && postExecutionControl.exceptionDuringExecution()) {
         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 (readyForCleanup(executionScope.getCustomResourceID())) {
+        cleanupForDeletedEvent(executionScope.getCustomResourceID());
       } else {
-        var executed = executeBufferedEvents(executionScope.getCustomResourceID());
+        var executed = submitReconciliationExecution(executionScope.getCustomResourceID());
         if (!executed) {
-          reScheduleExecutionIfInstructed(postExecutionControl, executionScope.getCustomResource());
+          reScheduleExecutionIfInstructed(postExecutionControl,
+              executionScope.getCustomResource());
         }
       }
     } finally {
       lock.unlock();
     }
   }
 
+  private boolean readyForCleanup(CustomResourceID customResourceID) {
+    return eventMarker
+        .getEventingState(customResourceID) == EventMarker.EventingState.ONLY_DELETE_EVENT_PRESENT;
+  }
+
   private void reScheduleExecutionIfInstructed(PostExecutionControl<R> postExecutionControl,
       R customResource) {
     postExecutionControl.getReScheduleDelay().ifPresent(delay -> eventSourceManager
@@ -227,13 +242,17 @@ 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());
+    EventMarker.EventingState eventingState =
+        eventMarker.getEventingState(customResourceID);
+    boolean newEventsExists =
+        eventingState == EventMarker.EventingState.EVENT_PRESENT ||
+            eventingState == EventMarker.EventingState.DELETE_AND_NON_DELETE_EVENT_PRESENT;
+    eventMarker.markEventReceived(customResourceID);
 
     if (newEventsExists) {
-      log.debug("New events exists for for resource id: {}", customResourceID);
-      executeBufferedEvents(customResourceID);
+      log.debug("New events exists for for resource id: {}",
+          customResourceID);
+      submitReconciliationExecution(customResourceID);
       return;
     }
     Optional<Long> nextDelay = execution.nextDelay();
@@ -251,7 +270,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 +289,9 @@ private RetryExecution getOrInitRetryExecution(ExecutionScope<R> executionScope)
     return retryExecution;
   }
 
-  private void cleanupAfterDeletedEvent(CustomResourceID customResourceUid) {
+  private void cleanupForDeletedEvent(CustomResourceID customResourceUid) {
     eventSourceManager.cleanup(customResourceUid);
-    eventBuffer.cleanup(customResourceUid);
+    eventMarker.cleanup(customResourceUid);
   }
 
   private boolean isControllerUnderExecution(CustomResourceID customResourceUid) {
@@ -287,6 +306,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

@@ -14,9 +14,7 @@
 import io.javaoperatorsdk.operator.api.ResourceController;
 import io.javaoperatorsdk.operator.api.UpdateControl;
 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;
@@ -59,15 +57,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)) {
@@ -79,8 +69,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 {