caching of resources from update

csviri · csviri · commit 1548361834de · 2020-12-02T15:07:58.000+01:00
diff --git a/operator-framework/src/main/java/io/javaoperatorsdk/operator/api/ResourceController.java b/operator-framework/src/main/java/io/javaoperatorsdk/operator/api/ResourceController.java
@@ -20,6 +20,8 @@ public interface ResourceController<R extends CustomResource> {
 
     /**
      * The implementation of this operation is required to be idempotent.
+     * Always use the UpdateControl object to make updates on custom resource if possible.
+     * Also always use the custom resource parameter (not the custom resource that might be in the events)
      *
      * @return The resource is updated in api server if the return value is present
      * within Optional. This the common use cases. However in cases, for example the operator is restarted,
diff --git a/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/DefaultEventHandler.java b/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/DefaultEventHandler.java
@@ -15,6 +15,7 @@
 import java.util.concurrent.locks.ReentrantLock;
 
 import static io.javaoperatorsdk.operator.EventListUtils.containsCustomResourceDeletedEvent;
+import static io.javaoperatorsdk.operator.processing.KubernetesResourceUtils.*;
 
 /**
  * Event handler that makes sure that events are processed in a "single threaded" way per resource UID, while buffering
@@ -89,13 +90,31 @@ void eventProcessingFinished(ExecutionScope executionScope, PostExecutionControl
             if (containsCustomResourceDeletedEvent(executionScope.getEvents())) {
                 cleanupAfterDeletedEvent(executionScope.getCustomResourceUid());
             } else {
+                cacheUpdatedResourceIfChanged(executionScope, postExecutionControl);
                 executeBufferedEvents(executionScope.getCustomResourceUid());
             }
         } finally {
             lock.unlock();
         }
     }
 
+    /**
+     * Here we try to cache the latest resource after an update. The goal is to solve a concurrency issue we sometimes see:
+     * 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-comming execution.
+     */
+    private void cacheUpdatedResourceIfChanged(ExecutionScope executionScope, PostExecutionControl postExecutionControl) {
+        if (postExecutionControl.customResourceUpdatedDuringExecution()) {
+            CustomResource originalCustomResource = executionScope.getCustomResource();
+            CustomResource customResourceAfterExecution = postExecutionControl.getUpdatedCustomResource().get();
+            if (!getVersion(originalCustomResource).equals(getVersion(customResourceAfterExecution))) {
+                this.customResourceCache.cacheResource(customResourceAfterExecution);
+            }
+        }
+    }
+
     private void cleanupAfterDeletedEvent(String customResourceUid) {
         defaultEventSourceManager.cleanup(customResourceUid);
         eventBuffer.cleanup(customResourceUid);
diff --git a/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/EventDispatcher.java b/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/EventDispatcher.java
@@ -91,17 +91,23 @@ private PostExecutionControl handleCreateOrUpdate(ExecutionScope 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()) {
-                    customResourceFacade.updateStatus(updateControl.getCustomResource());
+                    updatedCustomResource = customResourceFacade.updateStatus(updateControl.getCustomResource());
                 } else if (updateControl.isUpdateCustomResource()) {
-                    updateCustomResource(updateControl.getCustomResource());
+                    updatedCustomResource = updateCustomResource(updateControl.getCustomResource());
                 }
                 markLastGenerationProcessed(resource);
+                if (updatedCustomResource != null) {
+                    return PostExecutionControl.customResourceUpdated(updatedCustomResource);
+                } else {
+                    return PostExecutionControl.defaultDispatch();
+                }
             } else {
                 log.debug("Skipping event processing because generations: {} with version: {}",
                         getUID(resource), getVersion(resource));
+                return PostExecutionControl.defaultDispatch();
             }
-            return PostExecutionControl.defaultDispatch();
         }
     }
 
@@ -128,13 +134,14 @@ private PostExecutionControl handleDelete(CustomResource resource, Context conte
         DeleteControl deleteControl = controller.deleteResource(resource, context);
         boolean hasFinalizer = ControllerUtils.hasGivenFinalizer(resource, resourceFinalizer);
         if (deleteControl == DeleteControl.DEFAULT_DELETE && hasFinalizer) {
-            removeFinalizer(resource);
+            CustomResource customResource = removeFinalizer(resource);
             cleanup(resource);
+            return PostExecutionControl.customResourceUpdated(customResource);
         } else {
             log.debug("Skipping finalizer remove for resource: {} with version: {}. delete control: {}, hasFinalizer: {} ",
                     getUID(resource), getVersion(resource), deleteControl, hasFinalizer);
+            return PostExecutionControl.defaultDispatch();
         }
-        return PostExecutionControl.defaultDispatch();
     }
 
     public boolean largerGenerationThenProcessedBefore(CustomResource resource) {
@@ -165,23 +172,23 @@ private void updateCustomResourceWithFinalizer(CustomResource resource) {
         replace(resource);
     }
 
-    private void updateCustomResource(CustomResource resource) {
+    private CustomResource updateCustomResource(CustomResource resource) {
         log.debug("Updating resource: {} with version: {}", getUID(resource),
                 getVersion(resource));
         log.trace("Resource before update: {}", resource);
-        replace(resource);
+        return replace(resource);
     }
 
 
-    private void removeFinalizer(CustomResource resource) {
+    private CustomResource removeFinalizer(CustomResource resource) {
         log.debug("Removing finalizer on resource: {} with version: {}", getUID(resource), getVersion(resource));
         resource.getMetadata().getFinalizers().remove(resourceFinalizer);
-        customResourceFacade.replaceWithLock(resource);
+        return customResourceFacade.replaceWithLock(resource);
     }
 
-    private void replace(CustomResource resource) {
+    private CustomResource replace(CustomResource resource) {
         log.debug("Trying to replace resource {}, version: {}", resource.getMetadata().getName(), resource.getMetadata().getResourceVersion());
-        customResourceFacade.replaceWithLock(resource);
+        return customResourceFacade.replaceWithLock(resource);
     }
 
     private void addFinalizerIfNotPresent(CustomResource resource) {
@@ -207,9 +214,9 @@ public CustomResourceFacade(MixedOperation<?, ?, ?, Resource<CustomResource, ?>>
             this.resourceOperation = resourceOperation;
         }
 
-        public void updateStatus(CustomResource resource) {
+        public CustomResource updateStatus(CustomResource resource) {
             log.trace("Updating status for resource: {}", resource);
-            resourceOperation.inNamespace(resource.getMetadata().getNamespace())
+            return resourceOperation.inNamespace(resource.getMetadata().getNamespace())
                     .withName(resource.getMetadata().getName())
                     .updateStatus(resource);
         }
diff --git a/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/PostExecutionControl.java b/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/PostExecutionControl.java
@@ -1,5 +1,6 @@
 package io.javaoperatorsdk.operator.processing;
 
+import io.fabric8.kubernetes.client.CustomResource;
 import io.javaoperatorsdk.operator.api.UpdateControl;
 
 import java.util.Optional;
@@ -8,29 +9,34 @@ public final class PostExecutionControl {
 
     private final boolean onlyFinalizerHandled;
 
-    private final UpdateControl updateControl;
+    private final CustomResource updatedCustomResource;
 
-    private PostExecutionControl(boolean onlyFinalizerHandled, UpdateControl updateControl) {
+    private PostExecutionControl(boolean onlyFinalizerHandled, CustomResource updatedCustomResource) {
         this.onlyFinalizerHandled = onlyFinalizerHandled;
-        this.updateControl = updateControl;
+        this.updatedCustomResource = updatedCustomResource;
     }
 
     public static PostExecutionControl onlyFinalizerAdded() {
-        return new PostExecutionControl(true,null);
+        return new PostExecutionControl(true, null);
     }
 
     public static PostExecutionControl defaultDispatch() {
-        return new PostExecutionControl(false,null);
+        return new PostExecutionControl(false, null);
     }
-    public static PostExecutionControl dispatchWithUpdateControl(UpdateControl updateControl) {
-        return new PostExecutionControl(false,updateControl);
+
+    public static PostExecutionControl customResourceUpdated(CustomResource updatedCustomResource) {
+        return new PostExecutionControl(false, updatedCustomResource);
     }
 
     public boolean isOnlyFinalizerHandled() {
         return onlyFinalizerHandled;
     }
 
-    public Optional<UpdateControl> getUpdateControl() {
-        return Optional.ofNullable(updateControl);
+    public Optional<CustomResource> getUpdatedCustomResource() {
+        return Optional.ofNullable(updatedCustomResource);
+    }
+
+    public boolean customResourceUpdatedDuringExecution() {
+        return updatedCustomResource != null;
     }
 }
diff --git a/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/event/internal/CustomResourceEventSource.java b/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/event/internal/CustomResourceEventSource.java
@@ -76,6 +76,18 @@ public void eventReceived(Watcher.Action action, CustomResource customResource)
         eventHandler.handleEvent(new CustomResourceEvent(action, customResource, this));
     }
 
+    public void refreshCachedCustomResource(String resourceId) {
+        CustomResource customResource = resourceCache.getLatestResource(resourceId).get();
+        CustomResourceOperationsImpl client = (CustomResourceOperationsImpl) this.client;
+
+        CustomResource upToDateCR = (CustomResource)
+                ((CustomResourceOperationsImpl) client.inNamespace(customResource.getMetadata()
+                        .getNamespace())
+                        .withName(customResource.getMetadata().getName()))
+                        .fromServer().get();
+        resourceCache.cacheResource(upToDateCR);
+    }
+
     @Override
     public void onClose(KubernetesClientException e) {
         if (e == null) {

Original file line number	Diff line number	Diff line change
`@@ -20,6 +20,8 @@ public interface ResourceController<R extends CustomResource> {`
`20`	`20`
`21`	`21`	`/**`
`22`	`22`	`* The implementation of this operation is required to be idempotent.`
	`23`	`+ * Always use the UpdateControl object to make updates on custom resource if possible.`
	`24`	`+ * Also always use the custom resource parameter (not the custom resource that might be in the events)`
`23`	`25`	`*`
`24`	`26`	`* @return The resource is updated in api server if the return value is present`
`25`	`27`	`* within Optional. This the common use cases. However in cases, for example the operator is restarted,`