Event source started before events are processed (simple/naive syncing approach) (#794)

Author: csviri

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

@@ -1,6 +1,11 @@
 package io.javaoperatorsdk.operator.processing.event;
 
 import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static io.javaoperatorsdk.operator.processing.event.EventMarker.EventingState.NO_EVENT_PRESENT;
 
 /**
  * Manages the state of received events. Basically there can be only three distinct states relevant
@@ -23,7 +28,7 @@ public enum EventingState {
 
   private EventingState getEventingState(ResourceID resourceID) {
     EventingState actualState = eventingState.get(resourceID);
-    return actualState == null ? EventingState.NO_EVENT_PRESENT : actualState;
+    return actualState == null ? NO_EVENT_PRESENT : actualState;
   }
 
   private void setEventingState(ResourceID resourceID, EventingState state) {
@@ -46,7 +51,7 @@ public void unMarkEventReceived(ResourceID resourceID) {
     switch (actualState) {
       case EVENT_PRESENT:
         setEventingState(resourceID,
-            EventingState.NO_EVENT_PRESENT);
+            NO_EVENT_PRESENT);
         break;
       case DELETE_EVENT_PRESENT:
         throw new IllegalStateException("Cannot unmark delete event.");
@@ -72,10 +77,18 @@ public boolean eventPresent(ResourceID resourceID) {
 
   public boolean noEventPresent(ResourceID resourceID) {
     var actualState = getEventingState(resourceID);
-    return actualState == EventingState.NO_EVENT_PRESENT;
+    return actualState == NO_EVENT_PRESENT;
   }
 
   public void cleanup(ResourceID resourceID) {
     eventingState.remove(resourceID);
   }
+
+  public List<ResourceID> resourceIDsWithEventPresent() {
+    return eventingState.entrySet().stream()
+        .filter(e -> e.getValue() != NO_EVENT_PRESENT)
+        .map(Map.Entry::getKey)
+        .collect(Collectors.toList());
+  }
+
 }

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

@@ -58,25 +58,38 @@ class EventProcessor<R extends HasMetadata> implements EventHandler, LifecycleAw
             eventSourceManager.getController().getConfiguration().getRetryConfiguration()),
         eventSourceManager.getController().getConfiguration().getConfigurationService() == null
             ? Metrics.NOOP
-            : eventSourceManager.getController().getConfiguration().getConfigurationService()
+            : eventSourceManager
+                .getController()
+                .getConfiguration()
+                .getConfigurationService()
                 .getMetrics(),
         eventSourceManager);
   }
 
-  EventProcessor(ReconciliationDispatcher<R> reconciliationDispatcher,
+  EventProcessor(
+      ReconciliationDispatcher<R> reconciliationDispatcher,
       EventSourceManager<R> eventSourceManager,
       String relatedControllerName,
       Retry retry) {
-    this(eventSourceManager.getControllerResourceEventSource().getResourceCache(), null,
+    this(
+        eventSourceManager.getControllerResourceEventSource().getResourceCache(),
+        null,
         relatedControllerName,
-        reconciliationDispatcher, retry, null, eventSourceManager);
+        reconciliationDispatcher,
+        retry,
+        null,
+        eventSourceManager);
   }
 
-  private EventProcessor(Cache<R> cache, ExecutorService executor,
+  private EventProcessor(
+      Cache<R> cache,
+      ExecutorService executor,
       String relatedControllerName,
-      ReconciliationDispatcher<R> reconciliationDispatcher, Retry retry, Metrics metrics,
+      ReconciliationDispatcher<R> reconciliationDispatcher,
+      Retry retry,
+      Metrics metrics,
       EventSourceManager<R> eventSourceManager) {
-    this.running = true;
+    this.running = false;
     this.executor =
         executor == null
             ? new ScheduledThreadPoolExecutor(
@@ -99,26 +112,31 @@ public void handleEvent(Event event) {
     lock.lock();
     try {
       log.debug("Received event: {}", event);
-      if (!this.running) {
-        log.debug("Skipping event: {} because the event handler is not started", event);
-        return;
-      }
+
       final var resourceID = event.getRelatedCustomResourceID();
       MDCUtils.addResourceIDInfo(resourceID);
       metrics.receivedEvent(event);
-
       handleEventMarking(event);
-      if (!eventMarker.deleteEventPresent(resourceID)) {
-        submitReconciliationExecution(resourceID);
-      } else {
-        cleanupForDeletedEvent(resourceID);
+      if (!this.running) {
+        // events are received and marked, but will be processed when started, see start() method.
+        log.debug("Skipping event: {} because the event processor is not started", event);
+        return;
       }
+      handleMarkedEventForResource(resourceID);
     } finally {
       lock.unlock();
       MDCUtils.removeResourceIDInfo();
     }
   }
 
+  private void handleMarkedEventForResource(ResourceID resourceID) {
+    if (!eventMarker.deleteEventPresent(resourceID)) {
+      submitReconciliationExecution(resourceID);
+    } else {
+      cleanupForDeletedEvent(resourceID);
+    }
+  }
+
   private void submitReconciliationExecution(ResourceID resourceID) {
     try {
       boolean controllerUnderExecution = isControllerUnderExecution(resourceID);
@@ -148,8 +166,8 @@ private void submitReconciliationExecution(ResourceID resourceID) {
   }
 
   private void handleEventMarking(Event event) {
-    if (event instanceof ResourceEvent &&
-        ((ResourceEvent) event).getAction() == ResourceAction.DELETED) {
+    if (event instanceof ResourceEvent
+        && ((ResourceEvent) event).getAction() == ResourceAction.DELETED) {
       eventMarker.markDeleteEventReceived(event);
     } else if (!eventMarker.deleteEventPresent(event.getRelatedCustomResourceID())) {
       eventMarker.markEventReceived(event);
@@ -177,10 +195,11 @@ void eventProcessingFinished(
       // 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 (isRetryConfigured() && postExecutionControl.exceptionDuringExecution() &&
-          !eventMarker.deleteEventPresent(resourceID)) {
-        handleRetryOnException(executionScope,
-            postExecutionControl.getRuntimeException().orElseThrow());
+      if (isRetryConfigured()
+          && postExecutionControl.exceptionDuringExecution()
+          && !eventMarker.deleteEventPresent(resourceID)) {
+        handleRetryOnException(
+            executionScope, postExecutionControl.getRuntimeException().orElseThrow());
         return;
       }
       cleanupOnSuccessfulExecution(executionScope);
@@ -195,8 +214,7 @@ void eventProcessingFinished(
             postponeReconciliationAndHandleCacheSyncEvent(resourceID);
           }
         } else {
-          reScheduleExecutionIfInstructed(postExecutionControl,
-              executionScope.getResource());
+          reScheduleExecutionIfInstructed(postExecutionControl, executionScope.getResource());
         }
       }
     } finally {
@@ -208,20 +226,26 @@ private void postponeReconciliationAndHandleCacheSyncEvent(ResourceID resourceID
     eventSourceManager.getControllerResourceEventSource().whitelistNextEvent(resourceID);
   }
 
-  private boolean isCacheReadyForInstantReconciliation(ExecutionScope<R> executionScope,
-      PostExecutionControl<R> postExecutionControl) {
+  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")));
+    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;
@@ -233,9 +257,10 @@ private boolean isCacheReadyForInstantReconciliation(ExecutionScope<R> execution
     return !cachedCustomResourceVersion.equals(originalResourceVersion);
   }
 
-  private void reScheduleExecutionIfInstructed(PostExecutionControl<R> postExecutionControl,
-      R customResource) {
-    postExecutionControl.getReScheduleDelay()
+  private void reScheduleExecutionIfInstructed(
+      PostExecutionControl<R> postExecutionControl, R customResource) {
+    postExecutionControl
+        .getReScheduleDelay()
         .ifPresent(delay -> retryEventSource().scheduleOnce(customResource, delay));
   }
 
@@ -248,16 +273,15 @@ TimerEventSource<R> retryEventSource() {
    * events (received meanwhile retry is in place or already in buffer) instantly or always wait
    * according to the retry timing if there was an exception.
    */
-  private void handleRetryOnException(ExecutionScope<R> executionScope,
-      RuntimeException exception) {
+  private void handleRetryOnException(
+      ExecutionScope<R> executionScope, RuntimeException exception) {
     RetryExecution execution = getOrInitRetryExecution(executionScope);
     var customResourceID = executionScope.getCustomResourceID();
     boolean eventPresent = eventMarker.eventPresent(customResourceID);
     eventMarker.markEventReceived(customResourceID);
 
     if (eventPresent) {
-      log.debug("New events exists for for resource id: {}",
-          customResourceID);
+      log.debug("New events exists for for resource id: {}", customResourceID);
       submitReconciliationExecution(customResourceID);
       return;
     }
@@ -277,8 +301,7 @@ private void handleRetryOnException(ExecutionScope<R> executionScope,
 
   private void cleanupOnSuccessfulExecution(ExecutionScope<R> executionScope) {
     log.debug(
-        "Cleanup for successful execution for resource: {}",
-        getName(executionScope.getResource()));
+        "Cleanup for successful execution for resource: {}", getName(executionScope.getResource()));
     if (isRetryConfigured()) {
       retryState.remove(executionScope.getCustomResourceID());
     }
@@ -330,11 +353,18 @@ public void start() throws OperatorException {
     lock.lock();
     try {
       this.running = true;
+      handleAlreadyMarkedEvents();
     } finally {
       lock.unlock();
     }
   }
 
+  private void handleAlreadyMarkedEvents() {
+    for (ResourceID resourceID : eventMarker.resourceIDsWithEventPresent()) {
+      handleMarkedEventForResource(resourceID);
+    }
+  }
+
   private class ControllerExecution implements Runnable {
     private final ExecutionScope<R> executionScope;
 

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

@@ -52,9 +52,18 @@ public EventSourceManager(Controller<R> controller) {
     registerEventSource(controllerEventSource);
   }
 
+  /**
+   * Starts the event sources first and then the processor. Note that it's not desired to start
+   * processing events while the event sources are not "synced". This not fully started and the
+   * caches propagated - although for non k8s related event sources this behavior might be different
+   * (see
+   * {@link io.javaoperatorsdk.operator.processing.event.source.polling.PerResourcePollingEventSource}).
+   *
+   * Now the event sources are also started sequentially, mainly because others might depend on
+   * {@link ControllerResourceEventSource} , which is started first.
+   */
   @Override
-  public void start() throws OperatorException {
-    eventProcessor.start();
+  public void start() {
     lock.lock();
     try {
       log.debug("Starting event sources.");
@@ -65,6 +74,7 @@ public void start() throws OperatorException {
           log.warn("Error starting {} -> {}", eventSource, e);
         }
       }
+      eventProcessor.start();
     } finally {
       lock.unlock();
     }

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

@@ -37,10 +37,16 @@ private static <T extends HasMetadata> PrimaryResourcesRetriever<T> fromMetadata
         return Collections.emptySet();
       } else {
         final var map = isLabel ? metadata.getLabels() : metadata.getAnnotations();
+        if (map == null) {
+          return Collections.emptySet();
+        }
+        var name = map.get(nameKey);
+        if (name == null) {
+          return Collections.emptySet();
+        }
         var namespace =
             namespaceKey == null ? resource.getMetadata().getNamespace() : map.get(namespaceKey);
-        return map != null ? Set.of(new ResourceID(map.get(nameKey), namespace))
-            : Collections.emptySet();
+        return Set.of(new ResourceID(name, namespace));
       }
     };
   }

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PerResourcePollingEventSource.java

@@ -119,6 +119,18 @@ public void run() {
     }
   }
 
+  /**
+   * When this event source is queried for the resource, it might not be fully "synced". Thus, the
+   * cache might not be propagated, therefore the supplier is checked for the resource too.
+   *
+   * @param primary resource of the controller
+   * @return the related resource for this event source
+   */
+  @Override
+  public Optional<T> getAssociated(R primary) {
+    return getValueFromCacheOrSupplier(ResourceID.fromResource(primary));
+  }
+
   /**
    *
    * @param resourceID of the target related resource

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingEventSource.java

@@ -53,6 +53,17 @@ public void stop() throws OperatorException {
     timer.cancel();
   }
 
+  /**
+   * See {@link PerResourcePollingEventSource} for more info.
+   * 
+   * @param primary custom resource
+   * @return related resource
+   */
+  @Override
+  public Optional<T> getAssociated(P primary) {
+    return getValueFromCacheOrSupplier(ResourceID.fromResource(primary));
+  }
+
   public Optional<T> getValueFromCacheOrSupplier(ResourceID resourceID) {
     var resource = getCachedValue(resourceID);
     if (resource.isPresent()) {

operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/EventMarkerTest.java

@@ -9,6 +9,7 @@ class EventMarkerTest {
 
   private final EventMarker eventMarker = new EventMarker();
   private ResourceID sampleResourceID = new ResourceID("test-name");
+  private ResourceID sampleResourceID2 = new ResourceID("test-name2");
 
   @Test
   public void returnsNoEventPresentIfNotMarkedYet() {
@@ -62,4 +63,16 @@ public void cannotMarkEventAfterDeleteEventReceived() {
     });
   }
 
+  @Test
+  public void listsResourceIDSWithEventsPresent() {
+    eventMarker.markEventReceived(sampleResourceID);
+    eventMarker.markEventReceived(sampleResourceID2);
+    eventMarker.unMarkEventReceived(sampleResourceID);
+
+    var res = eventMarker.resourceIDsWithEventPresent();
+
+    assertThat(res).hasSize(1);
+    assertThat(res).contains(sampleResourceID2);
+  }
+
 }