feat: strong consistency variant for informer list and byIndex (#3316)

Signed-off-by: Attila Mészáros <a_meszaros@apple.com>
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Chris Laprun <metacosm@users.noreply.github.com>

Author: 3 people

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

@@ -16,9 +16,13 @@
 package io.javaoperatorsdk.operator.api.reconciler;
 
 import java.util.List;
+import java.util.stream.Stream;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 
 public interface IndexedResourceCache<T extends HasMetadata> extends ResourceCache<T> {
+
   List<T> byIndex(String indexName, String indexKey);
+
+  Stream<T> byIndexStream(String indexName, String indexKey);
 }

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

@@ -37,7 +37,8 @@
 /**
  * Provides useful operations to manipulate resources (server-side apply, patch, etc.) in an
  * idiomatic way, in particular to make sure that the latest version of the resource is present in
- * the caches for the next reconciliation.
+ * the caches for the next reconciliation. In other words provides read-cache-after-write
+ * consistency.
  *
  * @param <P> the resource type on which this object operates
  */

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

@@ -256,12 +256,13 @@ public void addIndexers(Map<String, Function<R, List<String>>> indexers) {
     this.indexers.putAll(indexers);
   }
 
+  public Stream<R> byIndexStream(String indexName, String indexKey) {
+    return sources.values().stream().map(s -> s.byIndex(indexName, indexKey)).flatMap(List::stream);
+  }
+
   @Override
   public List<R> byIndex(String indexName, String indexKey) {
-    return sources.values().stream()
-        .map(s -> s.byIndex(indexName, indexKey))
-        .flatMap(List::stream)
-        .collect(Collectors.toList());
+    return byIndexStream(indexName, indexKey).collect(Collectors.toList());
   }
 
   @Override

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

@@ -195,6 +195,11 @@ public List<T> byIndex(String indexName, String indexKey) {
     return informer.getIndexer().byIndex(indexName, indexKey);
   }
 
+  @Override
+  public Stream<T> byIndexStream(String indexName, String indexKey) {
+    return byIndex(indexName, indexKey).stream();
+  }
+
   @Override
   public String toString() {
     return informerInfo() + " (" + informer + ')';

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

@@ -23,6 +23,7 @@
 import java.util.function.Function;
 import java.util.function.Predicate;
 import java.util.function.UnaryOperator;
+import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
 import org.slf4j.Logger;
@@ -111,7 +112,6 @@ public R eventFilteringUpdateAndCacheResource(R resourceToUpdate, UnaryOperator<
       res.ifPresentOrElse(
           r -> {
             R latestResource = (R) r.getResource().orElseThrow();
-
             // as previous resource version we use the one from successful update, since
             // we process new event here only if that is more recent then the event from our update.
             // Note that this is equivalent with the scenario when an informer watch connection
@@ -222,11 +222,6 @@ public Optional<R> getCachedValue(ResourceID resourceID) {
     return get(resourceID);
   }
 
-  @Override
-  public Stream<R> list(String namespace, Predicate<R> predicate) {
-    return manager().list(namespace, predicate);
-  }
-
   void setTemporalResourceCache(TemporaryResourceCache<R> temporaryResourceCache) {
     this.temporaryResourceCache = temporaryResourceCache;
   }
@@ -239,19 +234,134 @@ public void addIndexers(Map<String, Function<R, List<String>>> indexers) {
     this.indexers.putAll(indexers);
   }
 
+  @Override
+  public Stream<R> list(String namespace, Predicate<R> predicate) {
+    return manager().list(namespace, predicate);
+  }
+
+  @Override
+  public Stream<R> list(Predicate<R> predicate) {
+    return cache.list(predicate);
+  }
+
   @Override
   public List<R> byIndex(String indexName, String indexKey) {
     return manager().byIndex(indexName, indexKey);
   }
 
-  @Override
-  public Stream<ResourceID> keys() {
-    return cache.keys();
+  public Stream<R> byIndexStream(String indexName, String indexKey) {
+    return manager().byIndexStream(indexName, indexKey);
+  }
+
+  /**
+   * Like {@link #list(String, Predicate)} but for read-cache-after-write consistency. This is
+   * useful when resources are updated using {@link
+   * io.javaoperatorsdk.operator.api.reconciler.ResourceOperations}.
+   */
+  public Stream<R> listWithStrongConsistency(String namespace, Predicate<R> predicate) {
+    return mergeWithWithTempCacheResources(
+        manager().list(namespace, predicate), namespace, predicate);
+  }
+
+  /**
+   * Like {@link #list(Predicate)} but for read-cache-after-write consistency. This is useful when
+   * resources are updated using {@link
+   * io.javaoperatorsdk.operator.api.reconciler.ResourceOperations}.
+   */
+  public Stream<R> listWithStrongConsistency(Predicate<R> predicate) {
+    return mergeWithWithTempCacheResources(cache.list(predicate), null, predicate);
+  }
+
+  /**
+   * Like {@link #byIndexStream(String, String)} but for read-cache-after-write consistency. This is
+   * useful when resources are updated using {@link
+   * io.javaoperatorsdk.operator.api.reconciler.ResourceOperations}.
+   */
+  public Stream<R> byIndexStreamWithStrongConsistency(String indexName, String indexKey) {
+    return mergeWithWithTempCacheResources(
+        manager().byIndexStream(indexName, indexKey), indexName, indexKey);
+  }
+
+  private Stream<R> mergeWithWithTempCacheResources(
+      Stream<R> stream, String indexName, String indexKey) {
+    return mergeWithWithTempCacheResources(stream, null, null, indexName, indexKey);
+  }
+
+  private Stream<R> mergeWithWithTempCacheResources(
+      Stream<R> stream, String namespace, Predicate<R> predicate) {
+    return mergeWithWithTempCacheResources(stream, namespace, predicate, null, null);
+  }
+
+  private Stream<R> mergeWithWithTempCacheResources(
+      Stream<R> stream,
+      String namespace,
+      Predicate<R> predicate,
+      String indexName,
+      String indexKey) {
+    if (!comparableResourceVersions || temporaryResourceCache.isEmpty()) {
+      return stream;
+    }
+    var allTempResources = temporaryResourceCache.getResources();
+    Map<ResourceID, R> tempResources;
+    if (namespace == null && predicate == null) {
+      tempResources = new HashMap<>(allTempResources);
+    } else {
+      // filtering the temp cache according the user input (predicate, namespace)
+      tempResources =
+          allTempResources.entrySet().stream()
+              .filter(
+                  e -> {
+                    if (namespace != null) {
+                      var res =
+                          e.getKey().getNamespace().map(ns -> ns.equals(namespace)).orElse(false);
+                      if (!res) return false;
+                    }
+                    if (predicate != null) {
+                      return predicate.test(e.getValue());
+                    }
+                    return true;
+                  })
+              .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+    }
+    if (tempResources.isEmpty()) {
+      return stream;
+    }
+    var upToDateList =
+        stream
+            .map(
+                r -> {
+                  var resourceID = ResourceID.fromResource(r);
+                  // removing the id from the related temp resources
+                  // this is important so we can detect ghost resources:
+                  // all that remains is ghost resource
+                  var tempResource = tempResources.remove(resourceID);
+                  // using the latest version
+                  if (tempResource != null
+                      && ReconcilerUtilsInternal.compareResourceVersions(tempResource, r) > 0) {
+                    return tempResource;
+                  }
+                  return r;
+                })
+            .toList();
+    Stream<R> tempResourceStream;
+    // ghost resource handling
+    if (indexName != null && indexKey != null) {
+      var indexer = indexers.get(indexName);
+      if (indexer == null) {
+        throw new IllegalArgumentException("Indexer not found for: " + indexName);
+      }
+      // we check if the ghost resource is part of the index
+      tempResourceStream =
+          tempResources.values().stream().filter(r -> indexer.apply(r).contains(indexKey));
+    } else {
+      tempResourceStream = tempResources.values().stream();
+    }
+    return Stream.concat(tempResourceStream, upToDateList.stream());
   }
 
   @Override
-  public Stream<R> list(Predicate<R> predicate) {
-    return cache.list(predicate);
+  public Stream<ResourceID> keys() {
+    return cache.keys();
   }
 
   @Override

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

@@ -275,4 +275,12 @@ private void checkGhostResources() {
   public synchronized Optional<T> getResourceFromCache(ResourceID resourceID) {
     return Optional.ofNullable(cache.get(resourceID));
   }
+
+  synchronized boolean isEmpty() {
+    return cache.isEmpty();
+  }
+
+  synchronized Map<ResourceID, T> getResources() {
+    return new HashMap<>(cache);
+  }
 }