scheduler algorithm improvements, unit tests

Author: csviri

operator-framework/pom.xml

@@ -66,5 +66,11 @@
             <version>2.11.2</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.assertj</groupId>
+            <artifactId>assertj-core</artifactId>
+            <version>3.4.1</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 </project>

operator-framework/src/main/java/com/github/containersolutions/operator/processing/CustomResourceEvent.java

@@ -9,13 +9,14 @@
 
 import java.util.Optional;
 
-class CustomResourceEvent {
+public class CustomResourceEvent {
 
-    public static final int MAX_RETRY_COUNT = 5;
+    public static final long INITIAL_BACK_OFF_INTERVAL = 2000L;
 
+    public static final int MAX_RETRY_COUNT = 5;
+    public static final double BACK_OFF_MULTIPLIER = 1.5;
     private final static Logger log = LoggerFactory.getLogger(CustomResourceEvent.class);
-
-    private final static BackOffExecution backOff = new ExponentialBackOff(2000L, 1.5).start();
+    private final static BackOffExecution backOff = new ExponentialBackOff(INITIAL_BACK_OFF_INTERVAL, BACK_OFF_MULTIPLIER).start();
 
     private final Watcher.Action action;
     private final CustomResource resource;
@@ -49,9 +50,7 @@ public String resourceUid() {
     }
 
     public Boolean sameResourceAs(CustomResourceEvent otherEvent) {
-        return getResource().getKind().equals(otherEvent.getResource().getKind()) &&
-                getResource().getApiVersion().equals(otherEvent.getResource().getApiVersion()) &&
-                getResource().getMetadata().getName().equals(otherEvent.getResource().getMetadata().getName());
+        return getResource().getMetadata().getUid().equals(otherEvent.getResource().getMetadata().getUid());
     }
 
     public Boolean isSameResourceAndNewerVersion(CustomResourceEvent otherEvent) {

operator-framework/src/main/java/com/github/containersolutions/operator/processing/EventScheduler.java

@@ -23,25 +23,24 @@
  * <ul>
  *   <li>Only 1 event should be processed at a time for same custom resource
  *   (metadata.name is the id, but kind and api should be taken into account)</li>
- *   <li>Done - If event processing fails it should be rescheduled with retry - with limited number of retried
+ *   <li>If event processing fails it should be rescheduled with retry - with limited number of retried
  *   and exponential time slacks (pluggable reschedule strategy in future?)</li>
  *   <li>if there are multiple events received for the same resource process only the last one. (Others can be discarded)
  *   User resourceVersion to check which is the latest. Put the new one at the and of the queue?
  *   </li>
  *   <li>Done - Avoid starvation, so on retry put back resource at the end of the queue.</li>
  *   <li>The selecting event from a queue should not be naive. So for example:
  *     If we cannot pick the last event because an event for that resource is currently processing just gor for the next one.
- *     (Maybe is good to represent this queue with a list.) Or if and event is rescheduled
- *     (skip if there is not enough time left since last execution)
  *   </li>
- *   <li>Impossible, scheduled chosen - Threading approach thus thread pool size and/or implementation should be configurable</li>
- *   <li>see also: https://github.com/ContainerSolutions/java-operator-sdk/issues/34</li>
+ *   <li>Threading approach thus thread pool size and/or implementation should be configurable</li>
  * </ul>
  *
  * @param <R>
  */
 
-
+/**
+ *
+ **/
 public class EventScheduler<R extends CustomResource> implements Watcher<R> {
 
     private final static Logger log = LoggerFactory.getLogger(EventScheduler.class);
@@ -77,12 +76,11 @@ void scheduleEvent(CustomResourceEvent newEvent) {
             // we have to lock since the fabric8 client event handling is multi-threaded,
             // so in the following part could be a race condition when multiple events are received for same resource.
             lock.lock();
+            if (eventStore.processedNewerVersionBefore(newEvent)) {
+                log.debug("Skipping event processing since was processed event with newer version before. {}", newEvent);
+                return;
+            }
             if (newEvent.getAction() == Action.DELETED) {
-                // this is a tricky situation, do we want to process only events which are marked for deletion?
-                // or just ignore the problem. Note that marked for deletion event should already be the last event either
-                // under processing, or scheduled for it.
-                // There could be some corner case when we do have a event which we received before marked for deletion,
-                //   and did not received the marked for deletion, but this is such corner case that for sake of simplicity will ignore this.
                 return;
             }
             // if there is an event waiting for to be scheduled we just replace that.

operator-framework/src/main/java/com/github/containersolutions/operator/processing/EventStore.java

@@ -11,6 +11,8 @@ public class EventStore {
 
     private final static Logger log = LoggerFactory.getLogger(EventScheduler.class);
 
+    private final Map<String, Long> lastResourceVersion = new HashMap<>();
+
     private final Map<String, CustomResourceEvent> eventsNotScheduledYet = new HashMap<>();
     private final Map<String, ResourceScheduleHolder> eventsScheduledForProcessing = new HashMap<>();
     private final Map<String, CustomResourceEvent> eventsUnderProcessing = new HashMap<>();
@@ -24,8 +26,9 @@ public CustomResourceEvent removeEventNotScheduledYet(String uid) {
         return eventsNotScheduledYet.remove(uid);
     }
 
-    public void addOrReplaceEventAsNotScheduledYet(CustomResourceEvent newEvent) {
-        eventsNotScheduledYet.put(newEvent.resourceUid(), newEvent);
+    public void addOrReplaceEventAsNotScheduledYet(CustomResourceEvent event) {
+        eventsNotScheduledYet.put(event.resourceUid(), event);
+        updateLatestResourceVersionProcessed(event);
     }
 
     public boolean containsOlderVersionOfEventUnderProcessing(CustomResourceEvent newEvent) {
@@ -37,6 +40,11 @@ public boolean containsEventScheduledForProcessing(String uid) {
         return eventsScheduledForProcessing.containsKey(uid);
     }
 
+    public void addEventUnderProcessing(CustomResourceEvent event) {
+        eventsUnderProcessing.put(event.resourceUid(), event);
+        updateLatestResourceVersionProcessed(event);
+    }
+
     public ResourceScheduleHolder getEventScheduledForProcessing(String uid) {
         return eventsScheduledForProcessing.get(uid);
     }
@@ -47,16 +55,31 @@ public ResourceScheduleHolder removeEventScheduledForProcessing(String uid) {
 
     public void addEventScheduledForProcessing(ResourceScheduleHolder resourceScheduleHolder) {
         eventsScheduledForProcessing.put(resourceScheduleHolder.getCustomResourceEvent().resourceUid(), resourceScheduleHolder);
-    }
-
-    public void addEventUnderProcessing(CustomResourceEvent event) {
-        eventsUnderProcessing.put(event.resourceUid(), event);
+        updateLatestResourceVersionProcessed(resourceScheduleHolder.getCustomResourceEvent());
     }
 
     public CustomResourceEvent removeEventUnderProcessing(String uid) {
         return eventsUnderProcessing.remove(uid);
     }
 
+    private void updateLatestResourceVersionProcessed(CustomResourceEvent event) {
+        Long current = lastResourceVersion.get(event.resourceUid());
+        long received = Long.parseLong(event.getResource().getMetadata().getResourceVersion());
+        if (current == null || received > current) {
+            lastResourceVersion.put(event.resourceUid(), received);
+        }
+    }
+
+    public boolean processedNewerVersionBefore(CustomResourceEvent customResourceEvent) {
+        Long lastVersionProcessed = lastResourceVersion.get(customResourceEvent.resourceUid());
+        if (lastVersionProcessed == null) {
+            return false;
+        } else {
+            return lastVersionProcessed > Long.parseLong(customResourceEvent.getResource()
+                    .getMetadata().getResourceVersion());
+        }
+    }
+
     public static class ResourceScheduleHolder {
         private CustomResourceEvent customResourceEvent;
         private ScheduledFuture<?> scheduledFuture;