retry basic impl

Author: csviri

operator-framework/src/main/java/io/javaoperatorsdk/operator/Operator.java

@@ -15,6 +15,7 @@
 import io.javaoperatorsdk.operator.processing.CustomResourceCache;
 import io.javaoperatorsdk.operator.processing.event.DefaultEventSourceManager;
 import io.javaoperatorsdk.operator.processing.event.internal.CustomResourceEventSource;
+import io.javaoperatorsdk.operator.processing.retry.Retry;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -38,16 +39,16 @@ public Operator(KubernetesClient k8sClient) {
 
 
     public <R extends CustomResource> void registerControllerForAllNamespaces(ResourceController<R> controller) throws OperatorException {
-        registerController(controller, true);
+        registerController(controller, true, null);
     }
 
     public <R extends CustomResource> void registerController(ResourceController<R> controller, String... targetNamespaces) throws OperatorException {
-        registerController(controller, false, targetNamespaces);
+        registerController(controller, false, null, targetNamespaces);
     }
 
     @SuppressWarnings("rawtypes")
     private <R extends CustomResource> void registerController(ResourceController<R> controller,
-                                                               boolean watchAllNamespaces, String... targetNamespaces) throws OperatorException {
+                                                               boolean watchAllNamespaces, Retry retry, String... targetNamespaces) throws OperatorException {
         Class<R> resClass = getCustomResourceClass(controller);
         CustomResourceDefinitionContext crd = getCustomResourceDefinitionForController(controller);
         KubernetesDeserializer.registerCustomKind(crd.getVersion(), crd.getKind(), resClass);
@@ -58,8 +59,8 @@ private <R extends CustomResource> void registerController(ResourceController<R>
 
 
         CustomResourceCache customResourceCache = new CustomResourceCache();
-        DefaultEventHandler defaultEventHandler = new DefaultEventHandler(customResourceCache, eventDispatcher, controller.getClass().getName());
-        DefaultEventSourceManager eventSourceManager = new DefaultEventSourceManager(defaultEventHandler);
+        DefaultEventHandler defaultEventHandler = new DefaultEventHandler(customResourceCache, eventDispatcher, controller.getClass().getName(), retry);
+        DefaultEventSourceManager eventSourceManager = new DefaultEventSourceManager(defaultEventHandler, retry != null);
         defaultEventHandler.setDefaultEventSourceManager(eventSourceManager);
         eventDispatcher.setEventSourceManager(eventSourceManager);
 

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

@@ -5,16 +5,15 @@
 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.retry.Retry;
+import io.javaoperatorsdk.operator.processing.retry.RetryExecution;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.HashSet;
-import java.util.Optional;
-import java.util.Set;
+import java.util.*;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.locks.ReentrantLock;
-import java.util.function.Predicate;
 
 import static io.javaoperatorsdk.operator.EventListUtils.containsCustomResourceDeletedEvent;
 import static io.javaoperatorsdk.operator.processing.KubernetesResourceUtils.getUID;
@@ -34,13 +33,17 @@ public class DefaultEventHandler implements EventHandler {
     private final Set<String> underProcessing = new HashSet<>();
     private final ScheduledThreadPoolExecutor executor;
     private final EventDispatcher eventDispatcher;
+    private final Retry retry;
+    private final Map<String, RetryExecution> retryState = new HashMap<>();
     private DefaultEventSourceManager defaultEventSourceManager;
 
     private final ReentrantLock lock = new ReentrantLock();
 
-    public DefaultEventHandler(CustomResourceCache customResourceCache, EventDispatcher eventDispatcher, String relatedControllerName) {
+    public DefaultEventHandler(CustomResourceCache customResourceCache, EventDispatcher eventDispatcher, String relatedControllerName,
+                               Retry retry) {
         this.customResourceCache = customResourceCache;
         this.eventDispatcher = eventDispatcher;
+        this.retry = retry;
         eventBuffer = new EventBuffer();
         executor = new ScheduledThreadPoolExecutor(5, new ThreadFactory() {
             @Override
@@ -90,6 +93,13 @@ void eventProcessingFinished(ExecutionScope executionScope, PostExecutionControl
             lock.lock();
             log.debug("Event processing finished. Scope: {}", executionScope);
             unsetUnderExecution(executionScope.getCustomResourceUid());
+
+            if (retry != null && postExecutionControl.exceptionDuringExecution()) {
+                handleRetryOnException(executionScope, postExecutionControl);
+            } else if (retry != null) {
+                handleSuccessfulExecutionRegardingRetry(executionScope);
+            }
+
             if (containsCustomResourceDeletedEvent(executionScope.getEvents())) {
                 cleanupAfterDeletedEvent(executionScope.getCustomResourceUid());
             } else {
@@ -101,16 +111,49 @@ void eventProcessingFinished(ExecutionScope executionScope, PostExecutionControl
         }
     }
 
+    /**
+     * Regarding the events  there are 2 approaches we can take. Either retry always when there are new 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 executionScope, PostExecutionControl postExecutionControl) {
+        RetryExecution execution = getOrInitRetryExecution(executionScope);
+        boolean newEventsExists = eventBuffer.newEventsExists(executionScope.getCustomResourceUid());
+        eventBuffer.putBackEvents(executionScope.getCustomResourceUid(), executionScope.getEvents());
+
+        Optional<Long> nextDelay = execution.nextDelay();
+        if (newEventsExists) {
+            executeBufferedEvents(executionScope.getCustomResourceUid());
+            return;
+        }
+        nextDelay.ifPresent(delay ->
+                defaultEventSourceManager.getRetryTimerEventSource()
+                        .scheduleOnce(executionScope.getCustomResource(), delay));
+    }
+
+    private void handleSuccessfulExecutionRegardingRetry(ExecutionScope executionScope) {
+        retryState.remove(executionScope.getCustomResourceUid());
+        defaultEventSourceManager.getRetryTimerEventSource().cancelOnceSchedule(executionScope.getCustomResourceUid());
+    }
+
+    private RetryExecution getOrInitRetryExecution(ExecutionScope executionScope) {
+        RetryExecution retryExecution = retryState.get(executionScope.getCustomResourceUid());
+        if (retryExecution == null) {
+            retryExecution = retry.initExecution();
+            retryState.put(executionScope.getCustomResourceUid(), retryExecution);
+        }
+        return retryExecution;
+    }
+
     /**
      * Here we try to cache the latest resource after an update. The goal is to solve a concurrency issue we've seen:
      * If an execution is finished, where we updated a custom resource, but there are other events already buffered for next
      * execution, we might not get the newest custom resource from CustomResource event source in time. Thus we execute
      * the next batch of events but with a non up to date CR. Here we cache the latest CustomResource from the update
      * execution so we make sure its already used in the up-coming execution.
-     *
+     * <p>
      * Note that this is an improvement, not a bug fix. This situation can happen naturally, we just make the execution more
      * efficient, and avoid questions about conflicts.
-     *
+     * <p>
      * Note that without the conditional locking in the cache, there is a very minor chance that we would override an
      * additional change coming from a different client.
      */

operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/EventBuffer.java

@@ -15,6 +15,14 @@ public void addEvent(Event event) {
         crEvents.add(event);
     }
 
+    public boolean newEventsExists(String resourceId) {
+        return !events.get(resourceId).isEmpty();
+    }
+
+    public void putBackEvents(String resourceUid, List<Event> oldEvents) {
+        events.get(resourceUid).addAll(0, oldEvents);
+    }
+
     public boolean containsEvents(String customResourceId) {
         return events.get(customResourceId) != null;
     }

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

@@ -5,16 +5,12 @@
 import io.fabric8.kubernetes.client.dsl.MixedOperation;
 import io.fabric8.kubernetes.client.dsl.Resource;
 import io.javaoperatorsdk.operator.api.*;
-import io.javaoperatorsdk.operator.processing.event.Event;
 import io.javaoperatorsdk.operator.processing.event.EventList;
 import io.javaoperatorsdk.operator.processing.event.EventSourceManager;
-import io.javaoperatorsdk.operator.processing.event.internal.CustomResourceEvent;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.ArrayList;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
 
 import static io.javaoperatorsdk.operator.EventListUtils.containsCustomResourceDeletedEvent;
 import static io.javaoperatorsdk.operator.processing.KubernetesResourceUtils.*;
@@ -43,16 +39,16 @@ public void setEventSourceManager(EventSourceManager eventSourceManager) {
         this.eventSourceManager = eventSourceManager;
     }
 
-    public PostExecutionControl handleEvent(ExecutionScope event) {
+    public PostExecutionControl handleExecution(ExecutionScope executionScope) {
         try {
-            return handDispatch(event);
+            return handleDispatch(executionScope);
         } catch (RuntimeException e) {
-            log.error("Error during event processing {} failed.", event, e);
-            return PostExecutionControl.defaultDispatch();
+            log.error("Error during event processing {} failed.", executionScope, e);
+            return PostExecutionControl.exceptionDuringExecution(e);
         }
     }
 
-    private PostExecutionControl handDispatch(ExecutionScope executionScope) {
+    private PostExecutionControl handleDispatch(ExecutionScope executionScope) {
         CustomResource resource = executionScope.getCustomResource();
         log.debug("Handling events: {} for resource {}", executionScope.getEvents(), resource.getMetadata());
 

operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/ExecutionConsumer.java

@@ -19,7 +19,7 @@ class ExecutionConsumer implements Runnable {
 
     @Override
     public void run() {
-        PostExecutionControl postExecutionControl = eventDispatcher.handleEvent(executionScope);
+        PostExecutionControl postExecutionControl = eventDispatcher.handleExecution(executionScope);
         defaultEventHandler.eventProcessingFinished(executionScope, postExecutionControl);
     }
 

operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/PostExecutionControl.java

@@ -1,7 +1,6 @@
 package io.javaoperatorsdk.operator.processing;
 
 import io.fabric8.kubernetes.client.CustomResource;
-import io.javaoperatorsdk.operator.api.UpdateControl;
 
 import java.util.Optional;
 
@@ -11,21 +10,28 @@ public final class PostExecutionControl {
 
     private final CustomResource updatedCustomResource;
 
-    private PostExecutionControl(boolean onlyFinalizerHandled, CustomResource updatedCustomResource) {
+    private final RuntimeException runtimeException;
+
+    private PostExecutionControl(boolean onlyFinalizerHandled, CustomResource updatedCustomResource, RuntimeException runtimeException) {
         this.onlyFinalizerHandled = onlyFinalizerHandled;
         this.updatedCustomResource = updatedCustomResource;
+        this.runtimeException = runtimeException;
     }
 
     public static PostExecutionControl onlyFinalizerAdded() {
-        return new PostExecutionControl(true, null);
+        return new PostExecutionControl(true, null, null);
     }
 
     public static PostExecutionControl defaultDispatch() {
-        return new PostExecutionControl(false, null);
+        return new PostExecutionControl(false, null, null);
     }
 
     public static PostExecutionControl customResourceUpdated(CustomResource updatedCustomResource) {
-        return new PostExecutionControl(false, updatedCustomResource);
+        return new PostExecutionControl(false, updatedCustomResource, null);
+    }
+
+    public static PostExecutionControl exceptionDuringExecution(RuntimeException exception) {
+        return new PostExecutionControl(false, null, exception);
     }
 
     public boolean isOnlyFinalizerHandled() {
@@ -39,4 +45,12 @@ public Optional<CustomResource> getUpdatedCustomResource() {
     public boolean customResourceUpdatedDuringExecution() {
         return updatedCustomResource != null;
     }
+
+    public boolean exceptionDuringExecution() {
+        return runtimeException != null;
+    }
+
+    public Optional<RuntimeException> getRuntimeException() {
+        return Optional.ofNullable(runtimeException);
+    }
 }

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

@@ -2,6 +2,7 @@
 
 import io.javaoperatorsdk.operator.processing.DefaultEventHandler;
 import io.javaoperatorsdk.operator.processing.event.internal.CustomResourceEventSource;
+import io.javaoperatorsdk.operator.processing.event.internal.TimerEventSource;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -12,15 +13,22 @@
 
 public class DefaultEventSourceManager implements EventSourceManager {
 
+    public static final String RETRY_TIMER_EVENT_SOURCE_NAME = "retry-timer-event-source";
     private static final Logger log = LoggerFactory.getLogger(DefaultEventSourceManager.class);
 
+
     private final ReentrantLock lock = new ReentrantLock();
     private Map<String, EventSource> eventSources = new ConcurrentHashMap<>();
     private CustomResourceEventSource customResourceEventSource;
     private DefaultEventHandler defaultEventHandler;
+    private TimerEventSource retryTimerEventSource;
 
-    public DefaultEventSourceManager(DefaultEventHandler defaultEventHandler) {
+    public DefaultEventSourceManager(DefaultEventHandler defaultEventHandler, boolean supportRetry) {
         this.defaultEventHandler = defaultEventHandler;
+        if (supportRetry) {
+            this.retryTimerEventSource = new TimerEventSource();
+            registerEventSource(RETRY_TIMER_EVENT_SOURCE_NAME, retryTimerEventSource);
+        }
     }
 
     public void registerCustomResourceEventSource(CustomResourceEventSource customResourceEventSource) {
@@ -36,15 +44,15 @@ public <T extends EventSource> void registerEventSource(String name, T eventSour
             if (currentEventSource != null) {
                 throw new IllegalStateException("Event source with name already registered. Event source name: " + name);
             }
-            eventSources.put(name,eventSource);
+            eventSources.put(name, eventSource);
             eventSource.setEventHandler(defaultEventHandler);
         } finally {
             lock.unlock();
         }
     }
 
     @Override
-    public Optional<EventSource> deRegisterCustomResourceFromEventSource(String eventSourceName,String customResourceUid) {
+    public Optional<EventSource> deRegisterCustomResourceFromEventSource(String eventSourceName, String customResourceUid) {
         try {
             lock.lock();
             EventSource eventSource = this.eventSources.get(eventSourceName);
@@ -60,13 +68,17 @@ public Optional<EventSource> deRegisterCustomResourceFromEventSource(String even
         }
     }
 
+    public TimerEventSource getRetryTimerEventSource() {
+        return retryTimerEventSource;
+    }
+
     @Override
     public Map<String, EventSource> getRegisteredEventSources() {
         return Collections.unmodifiableMap(eventSources);
     }
 
     public void cleanup(String customResourceUid) {
-        getRegisteredEventSources().keySet().forEach(k -> deRegisterCustomResourceFromEventSource(k,customResourceUid));
+        getRegisteredEventSources().keySet().forEach(k -> deRegisterCustomResourceFromEventSource(k, customResourceUid));
         eventSources.remove(customResourceUid);
     }
 

operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/retry/GenericRetry.java

@@ -16,7 +16,7 @@ public static GenericRetry defaultLimitedExponentialRetry() {
     }
 
     public static GenericRetry noRetry() {
-        return new GenericRetry().setMaxAttempts(1);
+        return new GenericRetry().setMaxAttempts(0);
     }
 
     public static GenericRetry every10second10TimesRetry() {

operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/retry/GenericRetryExecution.java

@@ -15,14 +15,7 @@ public GenericRetryExecution(GenericRetry genericRetry) {
         this.currentInterval = genericRetry.getInitialInterval();
     }
 
-    /**
-     * Note that first attempt is always 0. Since this implementation is tailored for event scheduling.
-     */
     public Optional<Long> nextDelay() {
-        if (lastAttemptIndex == 0) {
-            lastAttemptIndex++;
-            return Optional.of(0L);
-        }
         if (genericRetry.getMaxAttempts() > -1 && lastAttemptIndex >= genericRetry.getMaxAttempts()) {
             return Optional.empty();
         }