Obsolete resource handling for read-cache-after-write (#3207)

Signed-off-by: Attila Mészáros <a_meszaros@apple.com>

Author: csviri

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

@@ -23,6 +23,7 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.function.Function;
@@ -40,6 +41,7 @@ public class ExecutorServiceManager {
   private ExecutorService executor;
   private ExecutorService workflowExecutor;
   private ExecutorService cachingExecutorService;
+  private ScheduledExecutorService scheduledExecutorService;
   private boolean started;
   private ConfigurationService configurationService;
 
@@ -126,10 +128,15 @@ public ExecutorService cachingExecutorService() {
     return cachingExecutorService;
   }
 
+  public ScheduledExecutorService scheduledExecutorService() {
+    return scheduledExecutorService;
+  }
+
   public void start(ConfigurationService configurationService) {
     if (!started) {
       this.configurationService = configurationService; // used to lazy init workflow executor
       this.cachingExecutorService = Executors.newCachedThreadPool();
+      this.scheduledExecutorService = Executors.newScheduledThreadPool(0);
       this.executor = new InstrumentedExecutorService(configurationService.getExecutorService());
       started = true;
     }

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

@@ -27,9 +27,11 @@
 import io.javaoperatorsdk.operator.processing.event.source.filter.OnAddFilter;
 import io.javaoperatorsdk.operator.processing.event.source.filter.OnDeleteFilter;
 import io.javaoperatorsdk.operator.processing.event.source.filter.OnUpdateFilter;
+import io.javaoperatorsdk.operator.processing.event.source.informer.TemporaryResourceCache;
 
 import static io.javaoperatorsdk.operator.api.reconciler.Constants.DEFAULT_COMPARABLE_RESOURCE_VERSION;
 import static io.javaoperatorsdk.operator.api.reconciler.Constants.DEFAULT_FOLLOW_CONTROLLER_NAMESPACE_CHANGES;
+import static io.javaoperatorsdk.operator.api.reconciler.Constants.DEFAULT_GHOST_RESOURCE_CHECK_INTERVAL_MILLIS;
 import static io.javaoperatorsdk.operator.api.reconciler.Constants.NO_LONG_VALUE_SET;
 import static io.javaoperatorsdk.operator.api.reconciler.Constants.NO_VALUE_SET;
 
@@ -139,4 +141,13 @@
    * @since 5.3.0
    */
   boolean comparableResourceVersions() default DEFAULT_COMPARABLE_RESOURCE_VERSION;
+
+  /**
+   * For read-cache-after-write consistency there are some corner cases where we need to check the
+   * caches see {@link TemporaryResourceCache} periodically. This is the period in milliseconds.
+   * Applicable only if {@link #comparableResourceVersions()} is true.
+   *
+   * @since 5.3.0
+   */
+  long ghostResourceCacheCheckInterval() default DEFAULT_GHOST_RESOURCE_CHECK_INTERVAL_MILLIS;
 }

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

@@ -15,6 +15,7 @@
  */
 package io.javaoperatorsdk.operator.api.config.informer;
 
+import java.time.Duration;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
@@ -53,7 +54,8 @@ public class InformerConfiguration<R extends HasMetadata> {
   private ItemStore<R> itemStore;
   private Long informerListLimit;
   private FieldSelector fieldSelector;
-  private boolean comparableResourceVersions;
+  private Boolean comparableResourceVersions;
+  private Duration ghostResourceCacheCheckInterval;
 
   protected InformerConfiguration(
       Class<R> resourceClass,
@@ -68,7 +70,8 @@ protected InformerConfiguration(
       ItemStore<R> itemStore,
       Long informerListLimit,
       FieldSelector fieldSelector,
-      boolean comparableResourceVersions) {
+      Boolean comparableResourceVersions,
+      Duration ghostResourceCacheCheckInterval) {
     this(resourceClass);
     this.name = name;
     this.namespaces = namespaces;
@@ -82,6 +85,7 @@ protected InformerConfiguration(
     this.informerListLimit = informerListLimit;
     this.fieldSelector = fieldSelector;
     this.comparableResourceVersions = comparableResourceVersions;
+    this.ghostResourceCacheCheckInterval = ghostResourceCacheCheckInterval;
   }
 
   private InformerConfiguration(Class<R> resourceClass) {
@@ -117,7 +121,8 @@ public static <R extends HasMetadata> InformerConfiguration<R>.Builder builder(
             original.itemStore,
             original.informerListLimit,
             original.fieldSelector,
-            original.comparableResourceVersions)
+            original.comparableResourceVersions,
+            original.ghostResourceCacheCheckInterval)
         .builder;
   }
 
@@ -296,6 +301,10 @@ public boolean isComparableResourceVersions() {
     return comparableResourceVersions;
   }
 
+  public Duration getGhostResourceCacheCheckInterval() {
+    return ghostResourceCacheCheckInterval;
+  }
+
   @SuppressWarnings("UnusedReturnValue")
   public class Builder {
 
@@ -310,6 +319,13 @@ public InformerConfiguration<R> buildForController() {
       }
       // to avoid potential NPE
       followControllerNamespaceChanges = false;
+      if (comparableResourceVersions == null) {
+        comparableResourceVersions = DEFAULT_COMPARABLE_RESOURCE_VERSION;
+      }
+
+      if (ghostResourceCacheCheckInterval == null) {
+        ghostResourceCacheCheckInterval = DEFAULT_GHOST_RESOURCE_CHECK_INTERVAL;
+      }
       return InformerConfiguration.this;
     }
 
@@ -321,6 +337,14 @@ public InformerConfiguration<R> build() {
       if (followControllerNamespaceChanges == null) {
         followControllerNamespaceChanges = DEFAULT_FOLLOW_CONTROLLER_NAMESPACE_CHANGES;
       }
+      if (comparableResourceVersions == null) {
+        comparableResourceVersions = DEFAULT_COMPARABLE_RESOURCE_VERSION;
+      }
+
+      if (ghostResourceCacheCheckInterval == null) {
+        ghostResourceCacheCheckInterval = DEFAULT_GHOST_RESOURCE_CHECK_INTERVAL;
+      }
+
       return InformerConfiguration.this;
     }
 
@@ -368,6 +392,8 @@ public InformerConfiguration<R>.Builder initFromAnnotation(
                     .map(f -> new FieldSelector.Field(f.path(), f.value(), f.negated()))
                     .toList()));
         withComparableResourceVersions(informerConfig.comparableResourceVersions());
+        withGhostResourceCacheCheckInterval(
+            Duration.ofMillis(informerConfig.ghostResourceCacheCheckInterval()));
       }
       return this;
     }
@@ -473,5 +499,10 @@ public Builder withComparableResourceVersions(boolean comparableResourceVersions
       InformerConfiguration.this.comparableResourceVersions = comparableResourceVersions;
       return this;
     }
+
+    public Builder withGhostResourceCacheCheckInterval(Duration ghostResourceCacheCheckInterval) {
+      InformerConfiguration.this.ghostResourceCacheCheckInterval = ghostResourceCacheCheckInterval;
+      return this;
+    }
   }
 }

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

@@ -15,6 +15,7 @@
  */
 package io.javaoperatorsdk.operator.api.config.informer;
 
+import java.time.Duration;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
@@ -33,7 +34,6 @@
 import io.javaoperatorsdk.operator.processing.event.source.filter.OnUpdateFilter;
 import io.javaoperatorsdk.operator.processing.event.source.informer.Mappers;
 
-import static io.javaoperatorsdk.operator.api.reconciler.Constants.DEFAULT_COMPARABLE_RESOURCE_VERSION;
 import static io.javaoperatorsdk.operator.api.reconciler.Constants.SAME_AS_CONTROLLER_NAMESPACES_SET;
 import static io.javaoperatorsdk.operator.api.reconciler.Constants.WATCH_ALL_NAMESPACE_SET;
 import static io.javaoperatorsdk.operator.api.reconciler.Constants.WATCH_CURRENT_NAMESPACE_SET;
@@ -97,21 +97,18 @@ class DefaultInformerEventSourceConfiguration<R extends HasMetadata>
     private final GroupVersionKind groupVersionKind;
     private final InformerConfiguration<R> informerConfig;
     private final KubernetesClient kubernetesClient;
-    private final boolean comparableResourceVersion;
 
     protected DefaultInformerEventSourceConfiguration(
         GroupVersionKind groupVersionKind,
         PrimaryToSecondaryMapper<?> primaryToSecondaryMapper,
         SecondaryToPrimaryMapper<R> secondaryToPrimaryMapper,
         InformerConfiguration<R> informerConfig,
-        KubernetesClient kubernetesClient,
-        boolean comparableResourceVersion) {
+        KubernetesClient kubernetesClient) {
       this.informerConfig = Objects.requireNonNull(informerConfig);
       this.groupVersionKind = groupVersionKind;
       this.primaryToSecondaryMapper = primaryToSecondaryMapper;
       this.secondaryToPrimaryMapper = secondaryToPrimaryMapper;
       this.kubernetesClient = kubernetesClient;
-      this.comparableResourceVersion = comparableResourceVersion;
     }
 
     @Override
@@ -139,11 +136,6 @@ public Optional<GroupVersionKind> getGroupVersionKind() {
     public Optional<KubernetesClient> getKubernetesClient() {
       return Optional.ofNullable(kubernetesClient);
     }
-
-    @Override
-    public boolean comparableResourceVersion() {
-      return this.comparableResourceVersion;
-    }
   }
 
   @SuppressWarnings({"unused", "UnusedReturnValue"})
@@ -157,7 +149,6 @@ class Builder<R extends HasMetadata> {
     private PrimaryToSecondaryMapper<?> primaryToSecondaryMapper;
     private SecondaryToPrimaryMapper<R> secondaryToPrimaryMapper;
     private KubernetesClient kubernetesClient;
-    private boolean comparableResourceVersion = DEFAULT_COMPARABLE_RESOURCE_VERSION;
 
     private Builder(Class<R> resourceClass, Class<? extends HasMetadata> primaryResourceClass) {
       this(resourceClass, primaryResourceClass, null);
@@ -296,7 +287,13 @@ public Builder<R> withFieldSelector(FieldSelector fieldSelector) {
     }
 
     public Builder<R> withComparableResourceVersion(boolean comparableResourceVersion) {
-      this.comparableResourceVersion = comparableResourceVersion;
+      config.withComparableResourceVersions(comparableResourceVersion);
+      return this;
+    }
+
+    public Builder<R> withGhostResourceCacheCheckInterval(
+        Duration ghostResourceCacheCheckInterval) {
+      config.withGhostResourceCacheCheckInterval(ghostResourceCacheCheckInterval);
       return this;
     }
 
@@ -339,10 +336,7 @@ public InformerEventSourceConfiguration<R> build() {
                   HasMetadata.getKind(primaryResourceClass),
                   false)),
           config.build(),
-          kubernetesClient,
-          comparableResourceVersion);
+          kubernetesClient);
     }
   }
-
-  boolean comparableResourceVersion();
 }

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

@@ -15,6 +15,7 @@
  */
 package io.javaoperatorsdk.operator.api.reconciler;
 
+import java.time.Duration;
 import java.util.Collections;
 import java.util.Set;
 
@@ -43,5 +44,9 @@ public final class Constants {
   public static final boolean DEFAULT_FOLLOW_CONTROLLER_NAMESPACE_CHANGES = true;
   public static final boolean DEFAULT_COMPARABLE_RESOURCE_VERSION = true;
 
+  public static final long DEFAULT_GHOST_RESOURCE_CHECK_INTERVAL_MILLIS = 3 * 60 * 1000;
+  public static final Duration DEFAULT_GHOST_RESOURCE_CHECK_INTERVAL =
+      Duration.ofMillis(DEFAULT_GHOST_RESOURCE_CHECK_INTERVAL_MILLIS);
+
   private Constants() {}
 }

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

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

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

@@ -35,8 +35,6 @@
 import io.javaoperatorsdk.operator.processing.event.source.ResourceAction;
 import io.javaoperatorsdk.operator.processing.event.source.informer.TemporaryResourceCache.EventHandling;
 
-import static io.javaoperatorsdk.operator.api.reconciler.Constants.DEFAULT_COMPARABLE_RESOURCE_VERSION;
-
 /**
  * Wraps informer(s) so they are connected to the eventing system of the framework. Note that since
  * this is built on top of Fabric8 client Informers, it also supports caching resources using
@@ -58,29 +56,18 @@ public class InformerEventSource<R extends HasMetadata, P extends HasMetadata>
 
   public InformerEventSource(
       InformerEventSourceConfiguration<R> configuration, EventSourceContext<P> context) {
-    this(
-        configuration,
-        configuration.getKubernetesClient().orElse(context.getClient()),
-        configuration.comparableResourceVersion());
-  }
-
-  InformerEventSource(InformerEventSourceConfiguration<R> configuration, KubernetesClient client) {
-    this(configuration, client, DEFAULT_COMPARABLE_RESOURCE_VERSION);
+    this(configuration, configuration.getKubernetesClient().orElse(context.getClient()));
   }
 
   @SuppressWarnings({"unchecked", "rawtypes"})
-  private InformerEventSource(
-      InformerEventSourceConfiguration<R> configuration,
-      KubernetesClient client,
-      boolean comparableResourceVersions) {
+  InformerEventSource(InformerEventSourceConfiguration<R> configuration, KubernetesClient client) {
     super(
         configuration.name(),
         configuration
             .getGroupVersionKind()
             .map(gvk -> client.genericKubernetesResources(gvk.apiVersion(), gvk.getKind()))
             .orElseGet(() -> (MixedOperation) client.resources(configuration.getResourceClass())),
-        configuration,
-        comparableResourceVersions);
+        configuration);
     // If there is a primary to secondary mapper there is no need for primary to secondary index.
     primaryToSecondaryMapper = configuration.getPrimaryToSecondaryMapper();
     if (useSecondaryToPrimaryIndex()) {

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

@@ -230,11 +230,27 @@ private boolean isWatchingAllNamespaces() {
     return sources.containsKey(WATCH_ALL_NAMESPACES);
   }
 
+  public boolean isWatchingNamespace(String namespace) {
+    // for cluster scoped resources we can assume
+    // that we watch the whole cluster
+    if (namespace == null) {
+      return true;
+    }
+    if (isWatchingAllNamespaces()) {
+      return true;
+    }
+    return sources.containsKey(namespace);
+  }
+
   private Optional<InformerWrapper<R>> getSource(String namespace) {
     namespace = isWatchingAllNamespaces() || namespace == null ? WATCH_ALL_NAMESPACES : namespace;
     return Optional.ofNullable(sources.get(namespace));
   }
 
+  String lastSyncResourceVersion(String namespace) {
+    return getSource(namespace).orElseThrow().getLastSyncResourceVersion();
+  }
+
   @Override
   public void addIndexers(Map<String, Function<R, List<String>>> indexers) {
     this.indexers.putAll(indexers);