feat: ReconcileUtils for strongly consistent updates - alternative algortihm

Signed-off-by: Attila Mészáros <[email protected]>

Author: csviri

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/ControllerEventSource.java

@@ -81,7 +81,7 @@ public synchronized void start() {
     }
   }
 
-  public synchronized void eventReceived(
+  public synchronized void handleEvent(
       ResourceAction action,
       T resource,
       T oldResource,
@@ -138,13 +138,13 @@ private boolean isAcceptedByFilters(ResourceAction action, T resource, T oldReso
   @Override
   public void onAdd(T resource) {
     var knownResourceVersion = temporaryResourceCache.onAddOrUpdateEvent(resource);
-    eventReceived(ResourceAction.ADDED, resource, null, null, knownResourceVersion);
+    handleEvent(ResourceAction.ADDED, resource, null, null, knownResourceVersion);
   }
 
   @Override
   public void onUpdate(T oldCustomResource, T newCustomResource) {
     var knownResourceVersion = temporaryResourceCache.onAddOrUpdateEvent(newCustomResource);
-    eventReceived(
+    handleEvent(
         ResourceAction.UPDATED, newCustomResource, oldCustomResource, null, knownResourceVersion);
   }
 
@@ -153,7 +153,7 @@ public void onDelete(T resource, boolean deletedFinalStateUnknown) {
     temporaryResourceCache.onDeleteEvent(resource, deletedFinalStateUnknown);
     // delete event is quite special here, that requires special care, since we clean up caches on
     // delete event.
-    eventReceived(ResourceAction.DELETED, resource, null, deletedFinalStateUnknown, false);
+    handleEvent(ResourceAction.DELETED, resource, null, deletedFinalStateUnknown, false);
   }
 
   @Override

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

@@ -0,0 +1,70 @@
+/*
+ * Copyright Java Operator SDK Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *         http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.javaoperatorsdk.operator.processing.event.source.informer;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+
+import io.javaoperatorsdk.operator.processing.event.source.controller.ResourceEvent;
+
+class EventFilterDetails {
+
+  // initially should be created during event filtering update
+
+  private int activeUpdates = 0;
+  // todo might be just one
+  private final List<ResourceEvent> events = new ArrayList<>();
+  private int lastUpdatedResourceVersion = -1;
+
+  public int getActiveUpdates() {
+    return activeUpdates;
+  }
+
+  public void increaseActiveUpdates() {
+    activeUpdates = activeUpdates + 1;
+  }
+
+  public void decreaseActiveUpdates() {
+    activeUpdates = activeUpdates - 1;
+  }
+
+  public void recordEvent(ResourceEvent event) {
+    events.add(event);
+  }
+
+  public void setLastUpdatedResourceVersion(String version) {
+    var parsed = Integer.parseInt(version);
+    if (parsed > lastUpdatedResourceVersion) {
+      lastUpdatedResourceVersion = parsed;
+    }
+  }
+
+  public Optional<ResourceEvent> getLatestEventAfterLastUpdateEvent() {
+    if (events.isEmpty()) return Optional.empty();
+    var latest = events.get(events.size() - 1);
+    if (Integer.parseInt(latest.getResource().orElseThrow().getMetadata().getResourceVersion())
+        > lastUpdatedResourceVersion) {
+      return Optional.of(latest);
+    } else {
+      return Optional.empty();
+    }
+  }
+
+  public boolean isFilteringDone() {
+    return activeUpdates == 0;
+  }
+}

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

@@ -32,6 +32,7 @@
 import io.javaoperatorsdk.operator.processing.event.EventHandler;
 import io.javaoperatorsdk.operator.processing.event.ResourceID;
 import io.javaoperatorsdk.operator.processing.event.source.PrimaryToSecondaryMapper;
+import io.javaoperatorsdk.operator.processing.event.source.controller.ResourceAction;
 
 import static io.javaoperatorsdk.operator.api.reconciler.Constants.DEFAULT_COMPARABLE_RESOURCE_VERSION;
 
@@ -136,6 +137,16 @@ public synchronized void onDelete(R resource, boolean b) {
     }
   }
 
+  @Override
+  public void handleEvent(
+      ResourceAction action,
+      R resource,
+      R oldResource,
+      Boolean deletedFinalStateUnknown,
+      boolean filterEvent) {
+    propagateEvent(resource);
+  }
+
   @Override
   public synchronized void start() {
     super.start();

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

@@ -42,6 +42,8 @@
 import io.javaoperatorsdk.operator.health.Status;
 import io.javaoperatorsdk.operator.processing.event.ResourceID;
 import io.javaoperatorsdk.operator.processing.event.source.*;
+import io.javaoperatorsdk.operator.processing.event.source.controller.ResourceAction;
+import io.javaoperatorsdk.operator.processing.event.source.controller.ResourceDeleteEvent;
 
 @SuppressWarnings("rawtypes")
 public abstract class ManagedInformerEventSource<
@@ -104,16 +106,36 @@ public R eventFilteringUpdateAndCacheResource(R resourceToUpdate, UnaryOperator<
     if (log.isDebugEnabled()) {
       log.debug("Update and cache: {}", id);
     }
+    R updatedResource = null;
     try {
       temporaryResourceCache.startEventFilteringModify(id);
-      var updated = updateMethod.apply(resourceToUpdate);
-      handleRecentResourceUpdate(id, updated, resourceToUpdate);
-      return updated;
+      updatedResource = updateMethod.apply(resourceToUpdate);
+      handleRecentResourceUpdate(id, updatedResource, resourceToUpdate);
+      return updatedResource;
     } finally {
-      temporaryResourceCache.doneEventFilterModify(id);
+      var res =
+          temporaryResourceCache.doneEventFilterModify(
+              id,
+              updatedResource == null ? null : updatedResource.getMetadata().getResourceVersion());
+      res.ifPresent(
+          r ->
+              handleEvent(
+                  r.getAction(),
+                  (R) r.getResource().orElseThrow(),
+                  null,
+                  !(r instanceof ResourceDeleteEvent)
+                      || ((ResourceDeleteEvent) r).isDeletedFinalStateUnknown(),
+                  false));
     }
   }
 
+  public abstract void handleEvent(
+      ResourceAction action,
+      R resource,
+      R oldResource,
+      Boolean deletedFinalStateUnknown,
+      boolean filterEvent);
+
   @SuppressWarnings("unchecked")
   @Override
   public synchronized void start() {