moving generation handling to custom resource event source. Better handling of cached resource updates.

Author: csviri

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

@@ -54,7 +54,7 @@ private <R extends CustomResource> void registerController(ResourceController<R>
         String finalizer = ControllerUtils.getFinalizer(controller);
         MixedOperation client = k8sClient.customResources(crd, resClass, CustomResourceList.class, ControllerUtils.getCustomResourceDoneableClass(controller));
         EventDispatcher eventDispatcher = new EventDispatcher(controller,
-                finalizer, new EventDispatcher.CustomResourceFacade(client), ControllerUtils.getGenerationEventProcessing(controller));
+                finalizer, new EventDispatcher.CustomResourceFacade(client));
 
 
         CustomResourceCache customResourceCache = new CustomResourceCache();
@@ -68,7 +68,7 @@ private <R extends CustomResource> void registerController(ResourceController<R>
         controller.init(eventSourceManager);
         CustomResourceEventSource customResourceEventSource
                 = createCustomResourceEventSource(client, customResourceCache, watchAllNamespaces, targetNamespaces,
-                defaultEventHandler);
+                defaultEventHandler, ControllerUtils.getGenerationEventProcessing(controller));
         eventSourceManager.registerCustomResourceEventSource(customResourceEventSource);
 
 
@@ -80,10 +80,11 @@ private CustomResourceEventSource createCustomResourceEventSource(MixedOperation
                                                                       CustomResourceCache customResourceCache,
                                                                       boolean watchAllNamespaces,
                                                                       String[] targetNamespaces,
-                                                                      DefaultEventHandler defaultEventHandler) {
+                                                                      DefaultEventHandler defaultEventHandler,
+                                                                      boolean generationAware) {
         CustomResourceEventSource customResourceEventSource = watchAllNamespaces ?
-                CustomResourceEventSource.customResourceEventSourceForAllNamespaces(customResourceCache, client) :
-                CustomResourceEventSource.customResourceEventSourceForTargetNamespaces(customResourceCache, client, targetNamespaces);
+                CustomResourceEventSource.customResourceEventSourceForAllNamespaces(customResourceCache, client, generationAware) :
+                CustomResourceEventSource.customResourceEventSourceForTargetNamespaces(customResourceCache, client, targetNamespaces, generationAware);
 
         customResourceEventSource.setEventHandler(defaultEventHandler);
 

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

@@ -1,17 +1,42 @@
 package io.javaoperatorsdk.operator.processing;
 
 import io.fabric8.kubernetes.client.CustomResource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
+import java.util.HashMap;
 import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Predicate;
 
 public class CustomResourceCache {
+    private static final Logger log = LoggerFactory.getLogger(CustomResourceCache.class);
 
     private final Map<String, CustomResource> resources = new ConcurrentHashMap<>();
+    private final Lock lock = new ReentrantLock();
 
     public void cacheResource(CustomResource resource) {
-        resources.put(resource.getMetadata().getUid(), resource);
+        try {
+            lock.lock();
+            resources.put(KubernetesResourceUtils.getUID(resource), resource);
+        } finally {
+            lock.unlock();
+        }
+    }
+
+    public void cacheResource(CustomResource resource, Predicate<CustomResource> predicate) {
+        try {
+            lock.lock();
+            if (predicate.test(resources.get(KubernetesResourceUtils.getUID(resource)))) {
+                log.trace("Update cache after condition is true: {}", resource);
+                resources.put(resource.getMetadata().getUid(), resource);
+            }
+        } finally {
+            lock.unlock();
+        }
     }
 
     public Optional<CustomResource> getLatestResource(String uuid) {

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

@@ -14,6 +14,7 @@
 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;
@@ -101,23 +102,30 @@ void eventProcessingFinished(ExecutionScope executionScope, PostExecutionControl
     }
 
     /**
-     * Here we try to cache the latest resource after an update. The goal is to solve a concurrency issue we sometimes see:
+     * 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
+     * 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.
+     *
+     * 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.
+     *
+     * 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.
      */
     private void cacheUpdatedResourceIfChanged(ExecutionScope executionScope, PostExecutionControl postExecutionControl) {
         if (postExecutionControl.customResourceUpdatedDuringExecution()) {
             CustomResource originalCustomResource = executionScope.getCustomResource();
             CustomResource customResourceAfterExecution = postExecutionControl.getUpdatedCustomResource().get();
-            CustomResource cachedVersion = this.customResourceCache.getLatestResource(getUID(customResourceAfterExecution)).get();
             String originalResourceVersion = getVersion(originalCustomResource);
-            if (getVersion(cachedVersion).equals(originalResourceVersion) && !originalResourceVersion.equals(getVersion(customResourceAfterExecution))) {
-                log.debug("Updating custom resource cache from update response for resource uid: {} new version: {} old version: {}",
-                        getUID(originalCustomResource), getVersion(customResourceAfterExecution), getVersion(originalCustomResource));
-                this.customResourceCache.cacheResource(customResourceAfterExecution);
-            }
+
+            log.debug("Trying to update resource cache from update response for resource uid: {} new version: {} old version: {}",
+                    getUID(originalCustomResource), getVersion(customResourceAfterExecution), getVersion(originalCustomResource));
+            this.customResourceCache.cacheResource(customResourceAfterExecution, customResource ->
+                    getVersion(customResource).equals(originalResourceVersion)
+                            && !originalResourceVersion.equals(getVersion(customResourceAfterExecution))
+            );
         }
     }
 

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

@@ -29,17 +29,14 @@ public class EventDispatcher {
     private final ResourceController controller;
     private final String resourceFinalizer;
     private final CustomResourceFacade customResourceFacade;
-    private final boolean generationAware;
-    private final Map<String, Long> lastGenerationProcessedSuccessfully = new ConcurrentHashMap<>();
     private EventSourceManager eventSourceManager;
 
     public EventDispatcher(ResourceController controller,
                            String finalizer,
-                           CustomResourceFacade customResourceFacade, boolean generationAware) {
+                           CustomResourceFacade customResourceFacade) {
         this.controller = controller;
         this.customResourceFacade = customResourceFacade;
         this.resourceFinalizer = finalizer;
-        this.generationAware = generationAware;
     }
 
     public void setEventSourceManager(EventSourceManager eventSourceManager) {
@@ -60,7 +57,6 @@ private PostExecutionControl handDispatch(ExecutionScope executionScope) {
         log.debug("Handling events: {} for resource {}", executionScope.getEvents(), resource.getMetadata());
 
         if (containsCustomResourceDeletedEvent(executionScope.getEvents())) {
-            cleanup(executionScope.getCustomResource());
             log.debug("Skipping dispatch processing because of a Delete event: {} with version: {}",
                     getUID(resource), getVersion(resource));
             return PostExecutionControl.defaultDispatch();
@@ -87,55 +83,30 @@ private PostExecutionControl handleCreateOrUpdate(ExecutionScope executionScope,
             updateCustomResourceWithFinalizer(resource);
             return PostExecutionControl.onlyFinalizerAdded();
         } else {
-            if (!skipBecauseOfGenerations(executionScope)) {
-                log.debug("Executing createOrUpdate for resource {} with version: {} with execution scope: {}",
-                        getUID(resource), getVersion(resource), executionScope);
-                UpdateControl<? extends CustomResource> updateControl = controller.createOrUpdateResource(resource, context);
-                CustomResource updatedCustomResource = null;
-                if (updateControl.isUpdateStatusSubResource()) {
-                    updatedCustomResource = customResourceFacade.updateStatus(updateControl.getCustomResource());
-                } else if (updateControl.isUpdateCustomResource()) {
-                    updatedCustomResource = updateCustomResource(updateControl.getCustomResource());
-                }
-                markLastGenerationProcessed(resource);
-                if (updatedCustomResource != null) {
-                    return PostExecutionControl.customResourceUpdated(updatedCustomResource);
-                } else {
-                    return PostExecutionControl.defaultDispatch();
-                }
+            log.debug("Executing createOrUpdate for resource {} with version: {} with execution scope: {}",
+                    getUID(resource), getVersion(resource), executionScope);
+            UpdateControl<? extends CustomResource> updateControl = controller.createOrUpdateResource(resource, context);
+            CustomResource updatedCustomResource = null;
+            if (updateControl.isUpdateStatusSubResource()) {
+                updatedCustomResource = customResourceFacade.updateStatus(updateControl.getCustomResource());
+            } else if (updateControl.isUpdateCustomResource()) {
+                updatedCustomResource = updateCustomResource(updateControl.getCustomResource());
+            }
+            if (updatedCustomResource != null) {
+                return PostExecutionControl.customResourceUpdated(updatedCustomResource);
             } else {
-                log.debug("Skipping event processing because generations: {} with version: {}",
-                        getUID(resource), getVersion(resource));
                 return PostExecutionControl.defaultDispatch();
             }
         }
     }
 
-    private boolean skipBecauseOfGenerations(ExecutionScope executionScope) {
-        if (!generationAware) {
-            return false;
-        }
-        if (executionScope.getEvents().size() == 1) {
-            Event event = executionScope.getEvents().get(0);
-            if (event instanceof CustomResourceEvent) {
-                Long actualGeneration = executionScope.getCustomResource().getMetadata().getGeneration();
-                Long lastGeneration = lastGenerationProcessedSuccessfully.get(executionScope.getCustomResourceUid());
-                if (lastGeneration == null) {
-                    return false;
-                }
-                return actualGeneration <= lastGeneration;
-            }
-        }
-        return false;
-    }
 
     private PostExecutionControl handleDelete(CustomResource resource, Context context) {
         log.debug("Executing delete for resource: {} with version: {}", getUID(resource), getVersion(resource));
         DeleteControl deleteControl = controller.deleteResource(resource, context);
         boolean hasFinalizer = ControllerUtils.hasGivenFinalizer(resource, resourceFinalizer);
         if (deleteControl == DeleteControl.DEFAULT_DELETE && hasFinalizer) {
             CustomResource customResource = removeFinalizer(resource);
-            cleanup(resource);
             return PostExecutionControl.customResourceUpdated(customResource);
         } else {
             log.debug("Skipping finalizer remove for resource: {} with version: {}. delete control: {}, hasFinalizer: {} ",
@@ -144,27 +115,6 @@ private PostExecutionControl handleDelete(CustomResource resource, Context conte
         }
     }
 
-    public boolean largerGenerationThenProcessedBefore(CustomResource resource) {
-        Long lastGeneration = lastGenerationProcessedSuccessfully.get(resource.getMetadata().getUid());
-        if (lastGeneration == null) {
-            return true;
-        } else {
-            return resource.getMetadata().getGeneration() > lastGeneration;
-        }
-    }
-
-    private void cleanup(CustomResource resource) {
-        if (generationAware) {
-            lastGenerationProcessedSuccessfully.remove(resource.getMetadata().getUid());
-        }
-    }
-
-    private void markLastGenerationProcessed(CustomResource resource) {
-        if (generationAware) {
-            lastGenerationProcessedSuccessfully.put(resource.getMetadata().getUid(), resource.getMetadata().getGeneration());
-        }
-    }
-
     private void updateCustomResourceWithFinalizer(CustomResource resource) {
         log.debug("Adding finalizer for resource: {} version: {}", getUID(resource),
                 getVersion(resource));
@@ -201,10 +151,6 @@ private void addFinalizerIfNotPresent(CustomResource resource) {
         }
     }
 
-    private boolean markedForDeletion(CustomResource resource) {
-        return resource.getMetadata().getDeletionTimestamp() != null && !resource.getMetadata().getDeletionTimestamp().isEmpty();
-    }
-
     // created to support unit testing
     public static class CustomResourceFacade {
 

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

@@ -1,6 +1,7 @@
 package io.javaoperatorsdk.operator.processing;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.fabric8.kubernetes.client.CustomResource;
 
 public class KubernetesResourceUtils {
 
@@ -12,4 +13,8 @@ public static String getVersion(HasMetadata customResource) {
         return customResource.getMetadata().getResourceVersion();
     }
 
+
+    public static boolean markedForDeletion(CustomResource resource) {
+        return resource.getMetadata().getDeletionTimestamp() != null && !resource.getMetadata().getDeletionTimestamp().isEmpty();
+    }
 }