remove the use of the previous annotation via locking

Signed-off-by: Steve Hawkins <[email protected]>

Author: shawkins

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationService.java

@@ -447,19 +447,6 @@ default Set<Class<? extends HasMetadata>> defaultNonSSAResource() {
     return defaultNonSSAResources();
   }
 
-  /**
-   * If a javaoperatorsdk.io/previous annotation should be used so that the operator sdk can detect
-   * events from its own updates of dependent resources and then filter them.
-   *
-   * <p>Disable this if you want to react to your own dependent resource updates
-   *
-   * @return if special annotation should be used for dependent resource to filter events
-   * @since 4.5.0
-   */
-  default boolean previousAnnotationForDependentResourcesEventFiltering() {
-    return true;
-  }
-
   /**
    * For dependent resources, the framework can add an annotation to filter out events resulting
    * directly from the framework's operation. There are, however, some resources that do not follow

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationServiceOverrider.java

@@ -331,13 +331,6 @@ public Set<Class<? extends HasMetadata>> defaultNonSSAResources() {
             defaultNonSSAResource, ConfigurationService::defaultNonSSAResources);
       }
 
-      @Override
-      public boolean previousAnnotationForDependentResourcesEventFiltering() {
-        return overriddenValueOrDefault(
-            previousAnnotationForDependentResources,
-            ConfigurationService::previousAnnotationForDependentResourcesEventFiltering);
-      }
-
       @Override
       public boolean parseResourceVersionsForEventFilteringAndCaching() {
         return overriddenValueOrDefault(

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentResource.java

@@ -55,7 +55,6 @@ public abstract class KubernetesDependentResource<R extends HasMetadata, P exten
   private final boolean garbageCollected = this instanceof GarbageCollected;
   private KubernetesDependentResourceConfig<R> kubernetesDependentResourceConfig;
   private volatile Boolean useSSA;
-  private volatile Boolean usePreviousAnnotationForEventFiltering;
 
   public KubernetesDependentResource() {}
 
@@ -72,6 +71,27 @@ public void configureWith(KubernetesDependentResourceConfig<R> config) {
     this.kubernetesDependentResourceConfig = config;
   }
 
+  @Override
+  protected R handleCreate(R desired, P primary, Context<P> context) {
+    return eventSource()
+        .orElseThrow()
+        .updateAndCacheResource(
+            desired,
+            context,
+            toCreate -> KubernetesDependentResource.super.handleCreate(toCreate, primary, context));
+  }
+
+  @Override
+  protected R handleUpdate(R actual, R desired, P primary, Context<P> context) {
+    return eventSource()
+        .orElseThrow()
+        .updateAndCacheResource(
+            desired,
+            context,
+            toUpdate ->
+                KubernetesDependentResource.super.handleUpdate(actual, toUpdate, primary, context));
+  }
+
   @SuppressWarnings("unused")
   public R create(R desired, P primary, Context<P> context) {
     if (useSSA(context)) {
@@ -158,14 +178,6 @@ protected void addMetadata(
       } else {
         annotations.remove(InformerEventSource.PREVIOUS_ANNOTATION_KEY);
       }
-    } else if (usePreviousAnnotation(context)) { // set a new one
-      eventSource()
-          .orElseThrow()
-          .addPreviousAnnotation(
-              Optional.ofNullable(actualResource)
-                  .map(r -> r.getMetadata().getResourceVersion())
-                  .orElse(null),
-              target);
     }
     addReferenceHandlingMetadata(target, primary);
   }
@@ -181,22 +193,6 @@ protected boolean useSSA(Context<P> context) {
     return useSSA;
   }
 
-  private boolean usePreviousAnnotation(Context<P> context) {
-    if (usePreviousAnnotationForEventFiltering == null) {
-      usePreviousAnnotationForEventFiltering =
-          context
-                  .getControllerConfiguration()
-                  .getConfigurationService()
-                  .previousAnnotationForDependentResourcesEventFiltering()
-              && !context
-                  .getControllerConfiguration()
-                  .getConfigurationService()
-                  .withPreviousAnnotationForDependentResourcesBlocklist()
-                  .contains(this.resourceType());
-    }
-    return usePreviousAnnotationForEventFiltering;
-  }
-
   @Override
   protected void handleDelete(P primary, R secondary, Context<P> context) {
     if (secondary != null) {

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

@@ -17,7 +17,7 @@
 
 import java.util.Optional;
 import java.util.Set;
-import java.util.UUID;
+import java.util.function.UnaryOperator;
 import java.util.stream.Collectors;
 
 import org.slf4j.Logger;
@@ -78,7 +78,6 @@ public class InformerEventSource<R extends HasMetadata, P extends HasMetadata>
   // we need direct control for the indexer to propagate the just update resource also to the index
   private final PrimaryToSecondaryIndex<R> primaryToSecondaryIndex;
   private final PrimaryToSecondaryMapper<P> primaryToSecondaryMapper;
-  private final String id = UUID.randomUUID().toString();
 
   public InformerEventSource(
       InformerEventSourceConfiguration<R> configuration, EventSourceContext<P> context) {
@@ -122,6 +121,22 @@ private InformerEventSource(
     genericFilter = informerConfig.getGenericFilter();
   }
 
+  public R updateAndCacheResource(
+      R resourceToUpdate, Context<?> context, UnaryOperator<R> updateMethod) {
+    ResourceID id = ResourceID.fromResource(resourceToUpdate);
+    if (log.isDebugEnabled()) {
+      log.debug("Update and cache: {}", id);
+    }
+    try {
+      temporaryResourceCache.startModifying(id);
+      var updated = updateMethod.apply(resourceToUpdate);
+      handleRecentResourceUpdate(id, updated, resourceToUpdate);
+      return updated;
+    } finally {
+      temporaryResourceCache.doneModifying(id);
+    }
+  }
+
   @Override
   public void onAdd(R newResource) {
     if (log.isDebugEnabled()) {
@@ -131,9 +146,7 @@ public void onAdd(R newResource) {
           resourceType().getSimpleName(),
           newResource.getMetadata().getResourceVersion());
     }
-    primaryToSecondaryIndex.onAddOrUpdate(newResource);
-    onAddOrUpdate(
-        Operation.ADD, newResource, null, () -> InformerEventSource.super.onAdd(newResource));
+    onAddOrUpdate(Operation.ADD, newResource, null);
   }
 
   @Override
@@ -146,16 +159,11 @@ public void onUpdate(R oldObject, R newObject) {
           newObject.getMetadata().getResourceVersion(),
           oldObject.getMetadata().getResourceVersion());
     }
-    primaryToSecondaryIndex.onAddOrUpdate(newObject);
-    onAddOrUpdate(
-        Operation.UPDATE,
-        newObject,
-        oldObject,
-        () -> InformerEventSource.super.onUpdate(oldObject, newObject));
+    onAddOrUpdate(Operation.UPDATE, newObject, oldObject);
   }
 
   @Override
-  public void onDelete(R resource, boolean b) {
+  public synchronized void onDelete(R resource, boolean b) {
     if (log.isDebugEnabled()) {
       log.debug(
           "On delete event received for resource id: {} type: {}",
@@ -177,53 +185,26 @@ public synchronized void start() {
     manager().list().forEach(primaryToSecondaryIndex::onAddOrUpdate);
   }
 
-  private synchronized void onAddOrUpdate(
-      Operation operation, R newObject, R oldObject, Runnable superOnOp) {
+  private synchronized void onAddOrUpdate(Operation operation, R newObject, R oldObject) {
+    primaryToSecondaryIndex.onAddOrUpdate(newObject);
     var resourceID = ResourceID.fromResource(newObject);
 
-    if (canSkipEvent(newObject, oldObject, resourceID)) {
+    if (temporaryResourceCache.onAddOrUpdateEvent(newObject)) {
       log.debug(
           "Skipping event propagation for {}, since was a result of a reconcile action. Resource"
               + " ID: {}",
           operation,
           ResourceID.fromResource(newObject));
-      superOnOp.run();
+    } else if (eventAcceptedByFilter(operation, newObject, oldObject)) {
+      log.debug(
+          "Propagating event for {}, resource with same version not result of a reconciliation."
+              + " Resource ID: {}",
+          operation,
+          resourceID);
+      propagateEvent(newObject);
     } else {
-      superOnOp.run();
-      if (eventAcceptedByFilter(operation, newObject, oldObject)) {
-        log.debug(
-            "Propagating event for {}, resource with same version not result of a reconciliation."
-                + " Resource ID: {}",
-            operation,
-            resourceID);
-        propagateEvent(newObject);
-      } else {
-        log.debug("Event filtered out for operation: {}, resourceID: {}", operation, resourceID);
-      }
-    }
-  }
-
-  private boolean canSkipEvent(R newObject, R oldObject, ResourceID resourceID) {
-    return temporaryResourceCache.canSkipEvent(resourceID, newObject)
-        || isEventKnownFromAnnotation(newObject, oldObject);
-  }
-
-  private boolean isEventKnownFromAnnotation(R newObject, R oldObject) {
-    String previous = newObject.getMetadata().getAnnotations().get(PREVIOUS_ANNOTATION_KEY);
-    boolean known = false;
-    if (previous != null) {
-      String[] parts = previous.split(",");
-      if (id.equals(parts[0])) {
-        if (oldObject == null && parts.length == 1) {
-          known = true;
-        } else if (oldObject != null
-            && parts.length == 2
-            && oldObject.getMetadata().getResourceVersion().equals(parts[1])) {
-          known = true;
-        }
-      }
+      log.debug("Event filtered out for operation: {}, resourceID: {}", operation, resourceID);
     }
-    return known;
   }
 
   private void propagateEvent(R object) {
@@ -273,13 +254,13 @@ public Set<R> getSecondaryResources(P primary) {
   }
 
   @Override
-  public synchronized void handleRecentResourceUpdate(
+  public void handleRecentResourceUpdate(
       ResourceID resourceID, R resource, R previousVersionOfResource) {
     handleRecentCreateOrUpdate(Operation.UPDATE, resource, previousVersionOfResource);
   }
 
   @Override
-  public synchronized void handleRecentResourceCreate(ResourceID resourceID, R resource) {
+  public void handleRecentResourceCreate(ResourceID resourceID, R resource) {
     handleRecentCreateOrUpdate(Operation.ADD, resource, null);
   }
 
@@ -313,22 +294,6 @@ private boolean acceptedByDeleteFilters(R resource, boolean b) {
         && (genericFilter == null || genericFilter.accept(resource));
   }
 
-  /**
-   * Add an annotation to the resource so that the subsequent will be omitted
-   *
-   * @param resourceVersion null if there is no prior version
-   * @param target mutable resource that will be returned
-   */
-  public R addPreviousAnnotation(String resourceVersion, R target) {
-    target
-        .getMetadata()
-        .getAnnotations()
-        .put(
-            PREVIOUS_ANNOTATION_KEY,
-            id + Optional.ofNullable(resourceVersion).map(rv -> "," + rv).orElse(""));
-    return target;
-  }
-
   private enum Operation {
     ADD,
     UPDATE

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

@@ -221,11 +221,6 @@ public Optional<R> get(ResourceID resourceID) {
                     : r);
   }
 
-  public Optional<String> getLastSyncResourceVersion(Optional<String> namespace) {
-    return getSource(namespace.orElse(WATCH_ALL_NAMESPACES))
-        .map(source -> source.getLastSyncResourceVersion());
-  }
-
   @Override
   public Stream<ResourceID> keys() {
     return sources.values().stream().flatMap(Cache::keys);

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

@@ -103,7 +103,7 @@ public synchronized void start() {
     if (isRunning()) {
       return;
     }
-    temporaryResourceCache = new TemporaryResourceCache<>(this, parseResourceVersions);
+    temporaryResourceCache = new TemporaryResourceCache<>(parseResourceVersions);
     this.cache = new InformerManager<>(client, configuration, this);
     cache.setControllerConfiguration(controllerConfiguration);
     cache.addIndexers(indexers);
@@ -153,10 +153,6 @@ public Optional<R> get(ResourceID resourceID) {
     return res;
   }
 
-  public Optional<String> getLastSyncResourceVersion(Optional<String> namespace) {
-    return cache.getLastSyncResourceVersion(namespace);
-  }
-
   @SuppressWarnings("unused")
   public Optional<R> getCachedValue(ResourceID resourceID) {
     return get(resourceID);