eventing refinements mentioned on #2012 (#2034)

* refinements mentioned on #2012

provides two options
- to control if the annotation is used (to omit events that come too
quickly)
- to parse the resource version (to keep the cache up-to-date and omit
events if they come too slowly)

Signed-off-by: Steven Hawkins <shawkins@redhat.com>

Author: shawkins

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

@@ -364,4 +364,31 @@ default Set<Class<? extends HasMetadata>> defaultNonSSAResource() {
     return Set.of(ConfigMap.class, Secret.class);
   }
 
+  /**
+   * 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
+   *
+   * @since 4.5.0
+   */
+  default boolean previousAnnotationForDependentResourcesEventFiltering() {
+    return true;
+  }
+
+  /**
+   * If the event logic should parse the resourceVersion to determine the ordering of dependent
+   * resource events. This is typically not needed.
+   * <p>
+   * Disabled by default as Kubernetes does not support, and discourages, this interpretation of
+   * resourceVersions. Enable only if your api server event processing seems to lag the operator
+   * logic and you want to further minimize the the amount of work done / updates issued by the
+   * operator.
+   *
+   * @since 4.5.0
+   */
+  default boolean parseResourceVersionsForEventFilteringAndCaching() {
+    return false;
+  }
+
 }

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

@@ -37,6 +37,8 @@ public class ConfigurationServiceOverrider {
   private ResourceClassResolver resourceClassResolver;
   private Boolean ssaBasedCreateUpdateMatchForDependentResources;
   private Set<Class<? extends HasMetadata>> defaultNonSSAResource;
+  private Boolean previousAnnotationForDependentResources;
+  private Boolean parseResourceVersions;
 
   ConfigurationServiceOverrider(ConfigurationService original) {
     this.original = original;
@@ -158,6 +160,18 @@ public ConfigurationServiceOverrider withDefaultNonSSAResource(
     return this;
   }
 
+  public ConfigurationServiceOverrider withPreviousAnnotationForDependentResources(
+      boolean value) {
+    this.previousAnnotationForDependentResources = value;
+    return this;
+  }
+
+  public ConfigurationServiceOverrider wihtParseResourceVersions(
+      boolean value) {
+    this.parseResourceVersions = value;
+    return this;
+  }
+
   public ConfigurationService build() {
     return new BaseConfigurationService(original.getVersion(), cloner, client) {
       @Override
@@ -270,6 +284,20 @@ public Set<Class<? extends HasMetadata>> defaultNonSSAResource() {
         return defaultNonSSAResource != null ? defaultNonSSAResource
             : super.defaultNonSSAResource();
       }
+
+      @Override
+      public boolean previousAnnotationForDependentResourcesEventFiltering() {
+        return previousAnnotationForDependentResources != null
+            ? previousAnnotationForDependentResources
+            : super.previousAnnotationForDependentResourcesEventFiltering();
+      }
+
+      @Override
+      public boolean parseResourceVersionsForEventFilteringAndCaching() {
+        return parseResourceVersions != null
+            ? parseResourceVersions
+            : super.parseResourceVersionsForEventFilteringAndCaching();
+      }
     };
   }
 

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

@@ -62,6 +62,7 @@ public SecondaryToPrimaryMapper<R> getSecondaryToPrimaryMapper() {
       return secondaryToPrimaryMapper;
     }
 
+    @Override
     public Optional<OnDeleteFilter<? super R>> onDeleteFilter() {
       return Optional.ofNullable(onDeleteFilter);
     }
@@ -95,12 +96,15 @@ public <P extends HasMetadata> PrimaryToSecondaryMapper<P> getPrimaryToSecondary
    */
   SecondaryToPrimaryMapper<R> getSecondaryToPrimaryMapper();
 
+  @Override
   Optional<OnAddFilter<? super R>> onAddFilter();
 
+  @Override
   Optional<OnUpdateFilter<? super R>> onUpdateFilter();
 
   Optional<OnDeleteFilter<? super R>> onDeleteFilter();
 
+  @Override
   Optional<GenericFilter<? super R>> genericFilter();
 
   <P extends HasMetadata> PrimaryToSecondaryMapper<P> getPrimaryToSecondaryMapper();

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

@@ -113,7 +113,7 @@ public R create(R desired, P primary, Context<P> context) {
         desired.getMetadata().setResourceVersion("1");
       }
     }
-    addMetadata(false, null, desired, primary);
+    addMetadata(false, null, desired, primary, context);
     sanitizeDesired(desired, null, primary, context);
     final var resource = prepare(desired, primary, "Creating");
     return useSSA(context)
@@ -130,7 +130,7 @@ public R update(R actual, R desired, P primary, Context<P> context) {
           actual.getMetadata().getResourceVersion());
     }
     R updatedResource;
-    addMetadata(false, actual, desired, primary);
+    addMetadata(false, actual, desired, primary, context);
     sanitizeDesired(desired, actual, primary, context);
     if (useSSA(context)) {
       updatedResource = prepare(desired, primary, "Updating")
@@ -163,7 +163,7 @@ public Result<R> match(R actualResource, R desired, P primary, Context<P> contex
   public Result<R> match(R actualResource, R desired, P primary, ResourceUpdaterMatcher<R> matcher,
       Context<P> context) {
     final boolean matches;
-    addMetadata(true, actualResource, desired, primary);
+    addMetadata(true, actualResource, desired, primary, context);
     if (useSSA(context)) {
       matches = SSABasedGenericKubernetesResourceMatcher.getInstance()
           .matches(actualResource, desired, context);
@@ -173,8 +173,9 @@ public Result<R> match(R actualResource, R desired, P primary, ResourceUpdaterMa
     return Result.computed(matches, desired);
   }
 
-  protected void addMetadata(boolean forMatch, R actualResource, final R target, P primary) {
-    if (forMatch) { // keep the current
+  protected void addMetadata(boolean forMatch, R actualResource, final R target, P primary,
+      Context<P> context) {
+    if (forMatch) { // keep the current previous annotation
       String actual = actualResource.getMetadata().getAnnotations()
           .get(InformerEventSource.PREVIOUS_ANNOTATION_KEY);
       Map<String, String> annotations = target.getMetadata().getAnnotations();
@@ -183,7 +184,7 @@ protected void addMetadata(boolean forMatch, R actualResource, final R target, P
       } else {
         annotations.remove(InformerEventSource.PREVIOUS_ANNOTATION_KEY);
       }
-    } else { // set a new one
+    } else if (usePreviousAnnotation(context)) { // set a new one
       eventSource().orElseThrow().addPreviousAnnotation(
           Optional.ofNullable(actualResource).map(r -> r.getMetadata().getResourceVersion())
               .orElse(null),
@@ -208,6 +209,11 @@ protected boolean useSSA(Context<P> context) {
         .ssaBasedCreateUpdateMatchForDependentResources());
   }
 
+  private boolean usePreviousAnnotation(Context<P> context) {
+    return context.getControllerConfiguration().getConfigurationService()
+        .previousAnnotationForDependentResourcesEventFiltering();
+  }
+
   @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/controller/ControllerResourceEventSource.java

@@ -32,12 +32,12 @@ public class ControllerResourceEventSource<T extends HasMetadata>
 
   @SuppressWarnings({"unchecked", "rawtypes"})
   public ControllerResourceEventSource(Controller<T> controller) {
-    super(controller.getCRClient(), controller.getConfiguration());
+    super(controller.getCRClient(), controller.getConfiguration(), false);
     this.controller = controller;
 
     final var config = controller.getConfiguration();
     OnUpdateFilter internalOnUpdateFilter =
-        (OnUpdateFilter<T>) onUpdateFinalizerNeededAndApplied(controller.useFinalizer(),
+        onUpdateFinalizerNeededAndApplied(controller.useFinalizer(),
             config.getFinalizerName())
             .or(onUpdateGenerationAware(config.isGenerationAware()))
             .or(onUpdateMarkedForDeletion());

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

@@ -81,12 +81,18 @@ public class InformerEventSource<R extends HasMetadata, P extends HasMetadata>
 
   public InformerEventSource(
       InformerConfiguration<R> configuration, EventSourceContext<P> context) {
-    this(configuration, context.getClient());
+    this(configuration, context.getClient(),
+        context.getControllerConfiguration().getConfigurationService()
+            .parseResourceVersionsForEventFilteringAndCaching());
   }
 
   public InformerEventSource(InformerConfiguration<R> configuration, KubernetesClient client) {
-    super(client.resources(configuration.getResourceClass()), configuration);
+    this(configuration, client, false);
+  }
 
+  public InformerEventSource(InformerConfiguration<R> configuration, KubernetesClient client,
+      boolean parseResourceVersions) {
+    super(client.resources(configuration.getResourceClass()), configuration, parseResourceVersions);
     // If there is a primary to secondary mapper there is no need for primary to secondary index.
     primaryToSecondaryMapper = configuration.getPrimaryToSecondaryMapper();
     if (primaryToSecondaryMapper == null) {
@@ -169,6 +175,9 @@ private synchronized void onAddOrUpdate(Operation operation, R newObject, R oldO
   }
 
   private boolean canSkipEvent(R newObject, R oldObject, ResourceID resourceID) {
+    if (temporaryResourceCache.isKnownResourceVersion(newObject)) {
+      return true;
+    }
     var res = temporaryResourceCache.getResourceFromCache(resourceID);
     if (res.isEmpty()) {
       return isEventKnownFromAnnotation(newObject, oldObject);
@@ -275,7 +284,6 @@ public boolean allowsNamespaceChanges() {
     return configuration().followControllerNamespaceChanges();
   }
 
-
   private boolean eventAcceptedByFilter(Operation operation, R newObject, R oldObject) {
     if (genericFilter != null && !genericFilter.accept(newObject)) {
       return false;

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

@@ -42,26 +42,27 @@ public abstract class ManagedInformerEventSource<R extends HasMetadata, P extend
   protected MixedOperation<R, KubernetesResourceList<R>, Resource<R>> client;
 
   protected ManagedInformerEventSource(
-      MixedOperation<R, KubernetesResourceList<R>, Resource<R>> client, C configuration) {
+      MixedOperation<R, KubernetesResourceList<R>, Resource<R>> client, C configuration,
+      boolean parseResourceVersions) {
     super(configuration.getResourceClass());
     this.client = client;
-    temporaryResourceCache = new TemporaryResourceCache<>(this);
+    temporaryResourceCache = new TemporaryResourceCache<>(this, parseResourceVersions);
     this.cache = new InformerManager<>(client, configuration, this);
   }
 
   @Override
   public void onAdd(R resource) {
-    temporaryResourceCache.removeResourceFromCache(resource);
+    temporaryResourceCache.onEvent(resource, false);
   }
 
   @Override
   public void onUpdate(R oldObj, R newObj) {
-    temporaryResourceCache.removeResourceFromCache(newObj);
+    temporaryResourceCache.onEvent(newObj, false);
   }
 
   @Override
   public void onDelete(R obj, boolean deletedFinalStateUnknown) {
-    temporaryResourceCache.removeResourceFromCache(obj);
+    temporaryResourceCache.onEvent(obj, deletedFinalStateUnknown);
   }
 
   protected InformerManager<R, C> manager() {
@@ -127,6 +128,7 @@ void setTemporalResourceCache(TemporaryResourceCache<R> temporaryResourceCache)
     this.temporaryResourceCache = temporaryResourceCache;
   }
 
+  @Override
   public void addIndexers(Map<String, Function<R, List<String>>> indexers) {
     cache.addIndexers(indexers);
   }

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

@@ -1,13 +1,17 @@
 package io.javaoperatorsdk.operator.processing.event.source.informer;
 
+import java.util.Collections;
+import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.javaoperatorsdk.operator.api.config.informer.InformerConfiguration;
 import io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependentResource;
 import io.javaoperatorsdk.operator.processing.event.ResourceID;
 
@@ -33,16 +37,33 @@
 public class TemporaryResourceCache<T extends HasMetadata> {
 
   private static final Logger log = LoggerFactory.getLogger(TemporaryResourceCache.class);
+  private static final int MAX_RESOURCE_VERSIONS = 256;
 
   private final Map<ResourceID, T> cache = new ConcurrentHashMap<>();
   private final ManagedInformerEventSource<T, ?, ?> managedInformerEventSource;
+  private final boolean parseResourceVersions;
+  private final Set<String> knownResourceVersions;
 
-  public TemporaryResourceCache(ManagedInformerEventSource<T, ?, ?> managedInformerEventSource) {
+  public TemporaryResourceCache(ManagedInformerEventSource<T, ?, ?> managedInformerEventSource,
+      boolean parseResourceVersions) {
     this.managedInformerEventSource = managedInformerEventSource;
+    this.parseResourceVersions = parseResourceVersions;
+    if (parseResourceVersions) {
+      knownResourceVersions = Collections.newSetFromMap(new LinkedHashMap<String, Boolean>() {
+        @Override
+        protected boolean removeEldestEntry(java.util.Map.Entry<String, Boolean> eldest) {
+          return size() >= MAX_RESOURCE_VERSIONS;
+        }
+      });
+    } else {
+      knownResourceVersions = null;
+    }
   }
 
-  public synchronized Optional<T> removeResourceFromCache(T resource) {
-    return Optional.ofNullable(cache.remove(ResourceID.fromResource(resource)));
+  public synchronized void onEvent(T resource, boolean unknownState) {
+    cache.computeIfPresent(ResourceID.fromResource(resource),
+        (id, cached) -> (unknownState || !isLaterResourceVersion(id, cached, resource)) ? null
+            : cached);
   }
 
   public synchronized void putAddedResource(T newResource) {
@@ -56,23 +77,50 @@ public synchronized void putAddedResource(T newResource) {
    * @param previousResourceVersion null indicates an add
    */
   public synchronized void putResource(T newResource, String previousResourceVersion) {
+    if (knownResourceVersions != null) {
+      knownResourceVersions.add(newResource.getMetadata().getResourceVersion());
+    }
     var resourceId = ResourceID.fromResource(newResource);
     var cachedResource = getResourceFromCache(resourceId)
         .orElse(managedInformerEventSource.get(resourceId).orElse(null));
 
     if ((previousResourceVersion == null && cachedResource == null)
-        || (cachedResource != null && previousResourceVersion != null
-            && cachedResource.getMetadata().getResourceVersion()
-                .equals(previousResourceVersion))) {
+        || (cachedResource != null
+            && (cachedResource.getMetadata().getResourceVersion().equals(previousResourceVersion))
+            || isLaterResourceVersion(resourceId, newResource, cachedResource))) {
       log.debug(
           "Temporarily moving ahead to target version {} for resource id: {}",
           newResource.getMetadata().getResourceVersion(), resourceId);
       putToCache(newResource, resourceId);
-    } else {
-      if (cache.remove(resourceId) != null) {
-        log.debug("Removed an obsolete resource from cache for id: {}", resourceId);
+    } else if (cache.remove(resourceId) != null) {
+      log.debug("Removed an obsolete resource from cache for id: {}", resourceId);
+    }
+  }
+
+  public boolean isKnownResourceVersion(T resource) {
+    return knownResourceVersions != null
+        && knownResourceVersions.contains(resource.getMetadata().getResourceVersion());
+  }
+
+  /**
+   * @return true if {@link InformerConfiguration#parseResourceVersions()} is enabled and the
+   *         resourceVersion of newResource is numerically greater than cachedResource, otherwise
+   *         false
+   */
+  private boolean isLaterResourceVersion(ResourceID resourceId, T newResource, T cachedResource) {
+    try {
+      if (parseResourceVersions
+          && Long.parseLong(newResource.getMetadata().getResourceVersion()) > Long
+              .parseLong(cachedResource.getMetadata().getResourceVersion())) {
+        return true;
       }
+    } catch (NumberFormatException e) {
+      log.debug(
+          "Could not compare resourceVersions {} and {} for {}",
+          newResource.getMetadata().getResourceVersion(),
+          cachedResource.getMetadata().getResourceVersion(), resourceId);
     }
+    return false;
   }
 
   private void putToCache(T resource, ResourceID resourceID) {

operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerEventSourceTest.java

@@ -119,7 +119,7 @@ void propagateEventAndRemoveResourceFromTempCacheIfResourceVersionMismatch() {
     informerEventSource.onUpdate(cachedDeployment, testDeployment());
 
     verify(eventHandlerMock, times(1)).handleEvent(any());
-    verify(temporaryResourceCacheMock, times(1)).removeResourceFromCache(any());
+    verify(temporaryResourceCacheMock, times(1)).onEvent(testDeployment(), false);
   }
 
   @Test