enabling parse versions by default

aligning more

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

Author: shawkins

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

@@ -491,18 +491,16 @@ default Set<Class<? extends HasMetadata>> withPreviousAnnotationForDependentReso
 
   /**
    * If the event logic should parse the resourceVersion to determine the ordering of dependent
-   * resource events. This is typically not needed.
+   * resource events.
    *
-   * <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 amount of work done / updates issued by the
-   * operator.
+   * <p>Enabled by default as Kubernetes does support this interpretation of resourceVersions.
+   * Disable only if your api server provides non comparable resource versions..
    *
    * @return if resource version should be parsed (as integer)
    * @since 4.5.0
    */
   default boolean parseResourceVersionsForEventFilteringAndCaching() {
-    return false;
+    return true;
   }
 
   /**

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/PrimaryUpdateAndCacheUtils.java

@@ -451,6 +451,11 @@ public static <P extends HasMetadata> P addFinalizerWithSSA(
     }
   }
 
+  public static int compareResourceVersions(HasMetadata h1, HasMetadata h2) {
+    return compareResourceVersions(
+        h1.getMetadata().getResourceVersion(), h2.getMetadata().getResourceVersion());
+  }
+
   public static int compareResourceVersions(String v1, String v2) {
     var v1Length = v1.length();
     if (v1Length == 0) {

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

@@ -47,7 +47,7 @@ public class ControllerEventSource<T extends HasMetadata>
 
   @SuppressWarnings({"unchecked", "rawtypes"})
   public ControllerEventSource(Controller<T> controller) {
-    super(NAME, controller.getCRClient(), controller.getConfiguration(), false);
+    super(NAME, controller.getCRClient(), controller.getConfiguration(), true);
     this.controller = controller;
 
     final var config = controller.getConfiguration();

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

@@ -92,7 +92,7 @@ public InformerEventSource(
   }
 
   InformerEventSource(InformerEventSourceConfiguration<R> configuration, KubernetesClient client) {
-    this(configuration, client, false);
+    this(configuration, client, true);
   }
 
   @SuppressWarnings({"unchecked", "rawtypes"})
@@ -207,21 +207,8 @@ private synchronized void onAddOrUpdate(
   }
 
   private boolean canSkipEvent(R newObject, R oldObject, ResourceID resourceID) {
-    var res = temporaryResourceCache.getResourceFromCache(resourceID);
-    if (res.isEmpty()) {
-      return isEventKnownFromAnnotation(newObject, oldObject);
-    }
-    boolean resVersionsEqual =
-        newObject
-            .getMetadata()
-            .getResourceVersion()
-            .equals(res.get().getMetadata().getResourceVersion());
-    log.debug(
-        "Resource found in temporal cache for id: {} resource versions equal: {}",
-        resourceID,
-        resVersionsEqual);
-    return resVersionsEqual
-        || temporaryResourceCache.isLaterResourceVersion(resourceID, res.get(), newObject);
+    return temporaryResourceCache.canSkipEvent(resourceID, newObject)
+        || isEventKnownFromAnnotation(newObject, oldObject);
   }
 
   private boolean isEventKnownFromAnnotation(R newObject, R oldObject) {

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

@@ -221,6 +221,11 @@ 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/InformerWrapper.java

@@ -156,6 +156,10 @@ public Optional<T> get(ResourceID resourceID) {
     return Optional.ofNullable(cache.getByKey(getKey(resourceID)));
   }
 
+  public String getLastSyncResourceVersion() {
+    return this.informer.lastSyncResourceVersion();
+  }
+
   private String getKey(ResourceID resourceID) {
     return Cache.namespaceKeyFunc(resourceID.getNamespace().orElse(null), resourceID.getName());
   }

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

@@ -34,6 +34,7 @@
 import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
 import io.javaoperatorsdk.operator.api.config.Informable;
 import io.javaoperatorsdk.operator.api.config.NamespaceChangeable;
+import io.javaoperatorsdk.operator.api.reconciler.PrimaryUpdateAndCacheUtils;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.RecentOperationCacheFiller;
 import io.javaoperatorsdk.operator.health.InformerHealthIndicator;
 import io.javaoperatorsdk.operator.health.InformerWrappingEventSourceHealthIndicator;
@@ -133,19 +134,28 @@ public void handleRecentResourceCreate(ResourceID resourceID, R resource) {
 
   @Override
   public Optional<R> get(ResourceID resourceID) {
+    var res = cache.get(resourceID);
     Optional<R> resource = temporaryResourceCache.getResourceFromCache(resourceID);
-    if (resource.isPresent()) {
-      log.debug("Resource found in temporary cache for Resource ID: {}", resourceID);
+    if (parseResourceVersions
+        && resource.isPresent()
+        && res.filter(
+                r ->
+                    PrimaryUpdateAndCacheUtils.compareResourceVersions(r, resource.orElseThrow())
+                        > 0)
+            .isEmpty()) {
+      log.debug("Latest resource found in temporary cache for Resource ID: {}", resourceID);
       return resource;
-    } else {
-      log.debug(
-          "Resource not found in temporary cache reading it from informer cache,"
-              + " for Resource ID: {}",
-          resourceID);
-      var res = cache.get(resourceID);
-      log.debug("Resource found in cache: {} for id: {}", res.isPresent(), resourceID);
-      return res;
     }
+    log.debug(
+        "Resource not found, or older, in temporary cache. Found in informer cache {}, for"
+            + " Resource ID: {}",
+        res.isPresent(),
+        resourceID);
+    return res;
+  }
+
+  public Optional<String> getLastSyncResourceVersion(Optional<String> namespace) {
+    return cache.getLastSyncResourceVersion(namespace);
   }
 
   @SuppressWarnings("unused")

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

@@ -25,6 +25,7 @@
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.api.config.ConfigurationService;
+import io.javaoperatorsdk.operator.api.reconciler.PrimaryUpdateAndCacheUtils;
 import io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependentResource;
 import io.javaoperatorsdk.operator.processing.event.ResourceID;
 
@@ -130,6 +131,10 @@ public synchronized void putAddedResource(T newResource) {
    * @param previousResourceVersion null indicates an add
    */
   public synchronized void putResource(T newResource, String previousResourceVersion) {
+    if (!parseResourceVersions) {
+      return;
+    }
+
     var resourceId = ResourceID.fromResource(newResource);
     var cachedResource = managedInformerEventSource.get(resourceId).orElse(null);
 
@@ -185,6 +190,14 @@ public boolean isLaterResourceVersion(ResourceID resourceId, T newResource, T ca
     }
     return false;
   }
+  
+  public boolean canSkipEvent(ResourceID resourceID, T resource) {
+    return parseResourceVersions
+        && getResourceFromCache(resourceID)
+            .filter(
+                cached -> PrimaryUpdateAndCacheUtils.compareResourceVersions(cached, resource) >= 0)
+            .isPresent();
+  }
 
   public synchronized Optional<T> getResourceFromCache(ResourceID resourceID) {
     return Optional.ofNullable(cache.get(resourceID));