refinements mentioned on operator-framework#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: Steve Hawkins <[email protected]>

Author: shawkins

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

@@ -345,4 +345,31 @@ default boolean ssaBasedCreateUpdateMatchForDependentResources() {
     return true;
   }
 
+  /**
+   * If an 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 previousAnnotationForDependentResources() {
+    return true;
+  }
+
+  /**
+   * If the event logic should parse the resourceVersion to determine the ordering of events. This
+   * is typically not needed.
+   * <p>
+   * Disabled by default as Kubernetes does not officially support 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 parseResourceVersions() {
+    return false;
+  }
+
 }

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

@@ -35,6 +35,8 @@ public class ConfigurationServiceOverrider {
   private Duration cacheSyncTimeout;
   private ResourceClassResolver resourceClassResolver;
   private Boolean ssaBasedCreateUpdateMatchForDependentResources;
+  private Boolean previousAnnotationForDependentResources;
+  private Boolean parseResourceVersions;
 
   ConfigurationServiceOverrider(ConfigurationService original) {
     this.original = original;
@@ -150,6 +152,18 @@ public ConfigurationServiceOverrider withSSABasedCreateUpdateMatchForDependentRe
     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
@@ -256,6 +270,20 @@ public boolean ssaBasedCreateUpdateMatchForDependentResources() {
             ? ssaBasedCreateUpdateMatchForDependentResources
             : super.ssaBasedCreateUpdateMatchForDependentResources();
       }
+
+      @Override
+      public boolean previousAnnotationForDependentResources() {
+        return previousAnnotationForDependentResources != null
+            ? previousAnnotationForDependentResources
+            : super.previousAnnotationForDependentResources();
+      }
+
+      @Override
+      public boolean parseResourceVersions() {
+        return parseResourceVersions != null
+            ? parseResourceVersions
+            : super.parseResourceVersions();
+      }
     };
   }
 

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 target, P primary, Context<P> context) {
         target.getMetadata().setResourceVersion("1");
       }
     }
-    addMetadata(false, null, target, primary);
+    addMetadata(false, null, target, primary, context);
     final var resource = prepare(target, primary, "Creating");
     return useSSA(context)
         ? resource
@@ -129,7 +129,7 @@ public R update(R actual, R target, P primary, Context<P> context) {
           actual.getMetadata().getResourceVersion());
     }
     R updatedResource;
-    addMetadata(false, actual, target, primary);
+    addMetadata(false, actual, target, primary, context);
     if (useSSA(context)) {
       updatedResource = prepare(target, primary, "Updating")
           .fieldManager(context.getControllerConfiguration().fieldManager())
@@ -160,7 +160,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);
@@ -170,8 +170,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();
@@ -180,7 +181,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),
@@ -196,6 +197,11 @@ private boolean useSSA(Context<P> context) {
         .ssaBasedCreateUpdateMatchForDependentResources());
   }
 
+  private boolean usePreviousAnnotation(Context<P> context) {
+    return context.getControllerConfiguration().getConfigurationService()
+        .previousAnnotationForDependentResources();
+  }
+
   @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

@@ -69,6 +69,8 @@ public class InformerEventSource<R extends HasMetadata, P extends HasMetadata>
     extends ManagedInformerEventSource<R, P, InformerConfiguration<R>>
     implements ResourceEventHandler<R> {
 
+  private static final int MAX_RESOURCE_VERSIONS = 256;
+
   public static String PREVIOUS_ANNOTATION_KEY = "javaoperatorsdk.io/previous";
 
   private static final Logger log = LoggerFactory.getLogger(InformerEventSource.class);
@@ -78,14 +80,31 @@ public class InformerEventSource<R extends HasMetadata, P extends HasMetadata>
   private final PrimaryToSecondaryMapper<P> primaryToSecondaryMapper;
   private Map<String, Function<R, List<String>>> indexerBuffer = new HashMap<>();
   private final String id = UUID.randomUUID().toString();
+  private final Set<String> knownResourceVersions;
 
   public InformerEventSource(
       InformerConfiguration<R> configuration, EventSourceContext<P> context) {
-    this(configuration, context.getClient());
+    this(configuration, context.getClient(),
+        context.getControllerConfiguration().getConfigurationService().parseResourceVersions());
   }
 
   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 (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;
+    }
 
     // If there is a primary to secondary mapper there is no need for primary to secondary index.
     primaryToSecondaryMapper = configuration.getPrimaryToSecondaryMapper();
@@ -169,6 +188,10 @@ private synchronized void onAddOrUpdate(Operation operation, R newObject, R oldO
   }
 
   private boolean canSkipEvent(R newObject, R oldObject, ResourceID resourceID) {
+    if (knownResourceVersions != null
+        && knownResourceVersions.contains(newObject.getMetadata().getResourceVersion())) {
+      return true;
+    }
     var res = temporaryResourceCache.getResourceFromCache(resourceID);
     if (res.isEmpty()) {
       return isEventKnownFromAnnotation(newObject, oldObject);
@@ -262,6 +285,9 @@ public synchronized void handleRecentResourceCreate(ResourceID resourceID, R res
 
   private void handleRecentCreateOrUpdate(Operation operation, R newResource, R oldResource) {
     primaryToSecondaryIndex.onAddOrUpdate(newResource);
+    if (knownResourceVersions != null) {
+      knownResourceVersions.add(newResource.getMetadata().getResourceVersion());
+    }
     temporaryResourceCache.putResource(newResource, Optional.ofNullable(oldResource)
         .map(r -> r.getMetadata().getResourceVersion()).orElse(null));
   }
@@ -275,7 +301,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

@@ -8,6 +8,7 @@
 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;
 
@@ -36,13 +37,18 @@ public class TemporaryResourceCache<T extends HasMetadata> {
 
   private final Map<ResourceID, T> cache = new ConcurrentHashMap<>();
   private final ManagedInformerEventSource<T, ?, ?> managedInformerEventSource;
+  private final boolean parseResourceVersions;
 
-  public TemporaryResourceCache(ManagedInformerEventSource<T, ?, ?> managedInformerEventSource) {
+  public TemporaryResourceCache(ManagedInformerEventSource<T, ?, ?> managedInformerEventSource,
+      boolean parseResourceVersions) {
     this.managedInformerEventSource = managedInformerEventSource;
+    this.parseResourceVersions = parseResourceVersions;
   }
 
-  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) {
@@ -61,18 +67,37 @@ public synchronized void putResource(T newResource, String previousResourceVersi
         .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);
+    }
+  }
+
+  /**
+   * @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.compare(Long.parseLong(newResource.getMetadata().getResourceVersion()),
+              Long.parseLong(cachedResource.getMetadata().getResourceVersion())) > 0) {
+        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

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

@@ -21,7 +21,7 @@ class TemporaryResourceCacheTest {
   private final InformerEventSource<ConfigMap, ?> informerEventSource =
       mock(InformerEventSource.class);
   private final TemporaryResourceCache<ConfigMap> temporaryResourceCache =
-      new TemporaryResourceCache<>(informerEventSource);
+      new TemporaryResourceCache<>(informerEventSource, false);
 
   @Test
   void updateAddsTheResourceIntoCacheIfTheInformerHasThePreviousResourceVersion() {
@@ -75,7 +75,7 @@ void addOperationNotAddsTheResourceIfInformerCacheNotEmpty() {
   void removesResourceFromCache() {
     ConfigMap testResource = propagateTestResourceToCache();
 
-    temporaryResourceCache.removeResourceFromCache(testResource());
+    temporaryResourceCache.onEvent(testResource(), false);
 
     assertThat(temporaryResourceCache.getResourceFromCache(ResourceID.fromResource(testResource)))
         .isNotPresent();