improve:

Signed-off-by: Attila Mészáros <[email protected]>

Author: csviri

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

@@ -65,7 +65,6 @@ public class InformerEventSource<R extends HasMetadata, P extends HasMetadata>
   private static final Logger log = LoggerFactory.getLogger(InformerEventSource.class);
   // we need direct control for the indexer to propagate the just update resource also to the index
   private final PrimaryToSecondaryMapper<P> primaryToSecondaryMapper;
-  private final TemporalPrimaryToSecondaryIndex<R> temporalPrimaryToSecondaryIndex;
   private final String id = UUID.randomUUID().toString();
 
   public InformerEventSource(
@@ -99,17 +98,13 @@ private InformerEventSource(
     // If there is a primary to secondary mapper there is no need for primary to secondary index.
     primaryToSecondaryMapper = configuration.getPrimaryToSecondaryMapper();
     if (useSecondaryToPrimaryIndex()) {
-      temporalPrimaryToSecondaryIndex =
-          new DefaultTemporalPrimaryToSecondaryIndex<>(configuration.getSecondaryToPrimaryMapper());
       addIndexers(
           Map.of(
               PRIMARY_TO_SECONDARY_INDEX_NAME,
               (R r) ->
                   configuration.getSecondaryToPrimaryMapper().toPrimaryResourceIDs(r).stream()
                       .map(InformerEventSource::resourceIdToString)
                       .toList()));
-    } else {
-      temporalPrimaryToSecondaryIndex = NOOPTemporalPrimaryToSecondaryIndex.getInstance();
     }
 
     final var informerConfig = configuration.getInformerConfig();
@@ -158,7 +153,6 @@ public void onDelete(R resource, boolean b) {
           ResourceID.fromResource(resource),
           resourceType().getSimpleName());
     }
-    temporalPrimaryToSecondaryIndex.cleanupForResource(resource);
     super.onDelete(resource, b);
     if (acceptedByDeleteFilters(resource, b)) {
       propagateEvent(resource);
@@ -168,7 +162,6 @@ public void onDelete(R resource, boolean b) {
   private synchronized void onAddOrUpdate(
       Operation operation, R newObject, R oldObject, Runnable superOnOp) {
     var resourceID = ResourceID.fromResource(newObject);
-    temporalPrimaryToSecondaryIndex.cleanupForResource(newObject);
     if (canSkipEvent(newObject, oldObject, resourceID)) {
       log.debug(
           "Skipping event propagation for {}, since was a result of a reconcile action. Resource"
@@ -260,7 +253,10 @@ public Set<R> getSecondaryResources(P primary) {
       // that we did not receive yet an event in the informer so the index would not
       // be updated. However, before reading it from temp IDs the event arrives and erases
       // the temp index. So in case of Add not id would be found.
-      var temporalIds = temporalPrimaryToSecondaryIndex.getSecondaryResources(primaryID);
+      var temporalIds =
+          temporaryResourceCache
+              .getTemporalPrimaryToSecondaryIndex()
+              .getSecondaryResources(primaryID);
       var resources = byIndex(PRIMARY_TO_SECONDARY_INDEX_NAME, resourceIdToString(primaryID));
 
       log.debug(
@@ -313,7 +309,6 @@ public synchronized void handleRecentResourceCreate(ResourceID resourceID, R res
   }
 
   private void handleRecentCreateOrUpdate(R newResource, R oldResource) {
-    temporalPrimaryToSecondaryIndex.explicitAddOrUpdate(newResource);
     temporaryResourceCache.putResource(
         newResource,
         Optional.ofNullable(oldResource)
@@ -370,4 +365,16 @@ private enum Operation {
   private static String resourceIdToString(ResourceID resourceID) {
     return resourceID.getName() + "#" + resourceID.getNamespace().orElse("$na");
   }
+
+  @Override
+  @SuppressWarnings({"unchecked", "rawtypes"})
+  protected TemporaryResourceCache<R> temporaryResourceCache() {
+    return new TemporaryResourceCache<>(
+        this,
+        useSecondaryToPrimaryIndex()
+            ? new DefaultTemporalPrimaryToSecondaryIndex(
+                configuration().getSecondaryToPrimaryMapper())
+            : NOOPTemporalPrimaryToSecondaryIndex.getInstance(),
+        parseResourceVersions);
+  }
 }

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

@@ -40,7 +40,7 @@ public abstract class ManagedInformerEventSource<
 
   private static final Logger log = LoggerFactory.getLogger(ManagedInformerEventSource.class);
   private InformerManager<R, C> cache;
-  private final boolean parseResourceVersions;
+  protected final boolean parseResourceVersions;
   private ControllerConfiguration<R> controllerConfiguration;
   private final C configuration;
   private final Map<String, Function<R, List<String>>> indexers = new HashMap<>();
@@ -87,7 +87,7 @@ public synchronized void start() {
     if (isRunning()) {
       return;
     }
-    temporaryResourceCache = new TemporaryResourceCache<>(this, parseResourceVersions);
+    temporaryResourceCache = temporaryResourceCache();
     this.cache = new InformerManager<>(client, configuration, this);
     cache.setControllerConfiguration(controllerConfiguration);
     cache.addIndexers(indexers);
@@ -133,6 +133,11 @@ public Optional<R> get(ResourceID resourceID) {
     }
   }
 
+  protected TemporaryResourceCache temporaryResourceCache() {
+    return new TemporaryResourceCache<>(
+        this, NOOPTemporalPrimaryToSecondaryIndex.getInstance(), parseResourceVersions);
+  }
+
   @SuppressWarnings("unused")
   public Optional<R> getCachedValue(ResourceID resourceID) {
     return get(resourceID);

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

@@ -31,47 +31,6 @@
  */
 public class TemporaryResourceCache<T extends HasMetadata> {
 
-  static class ExpirationCache<K> {
-    private final LinkedHashMap<K, Long> cache;
-    private final int ttlMs;
-
-    public ExpirationCache(int maxEntries, int ttlMs) {
-      this.ttlMs = ttlMs;
-      this.cache =
-          new LinkedHashMap<>() {
-            @Override
-            protected boolean removeEldestEntry(Map.Entry<K, Long> eldest) {
-              return size() > maxEntries;
-            }
-          };
-    }
-
-    public void add(K key) {
-      clean();
-      cache.putIfAbsent(key, System.currentTimeMillis());
-    }
-
-    public boolean contains(K key) {
-      clean();
-      return cache.get(key) != null;
-    }
-
-    void clean() {
-      if (!cache.isEmpty()) {
-        long currentTimeMillis = System.currentTimeMillis();
-        var iter = cache.entrySet().iterator();
-        // the order will already be from oldest to newest, clean a fixed number of entries to
-        // amortize the cost amongst multiple calls
-        for (int i = 0; i < 10 && iter.hasNext(); i++) {
-          var entry = iter.next();
-          if (currentTimeMillis - entry.getValue() > ttlMs) {
-            iter.remove();
-          }
-        }
-      }
-    }
-  }
-
   private static final Logger log = LoggerFactory.getLogger(TemporaryResourceCache.class);
 
   private final Map<ResourceID, T> cache = new ConcurrentHashMap<>();
@@ -81,12 +40,15 @@ void clean() {
   private final ManagedInformerEventSource<T, ?, ?> managedInformerEventSource;
   private final boolean parseResourceVersions;
   private final ExpirationCache<String> knownResourceVersions;
+  private final TemporalPrimaryToSecondaryIndex<T> temporalPrimaryToSecondaryIndex;
 
   public TemporaryResourceCache(
       ManagedInformerEventSource<T, ?, ?> managedInformerEventSource,
+      TemporalPrimaryToSecondaryIndex<T> temporalPrimaryToSecondaryIndex,
       boolean parseResourceVersions) {
     this.managedInformerEventSource = managedInformerEventSource;
     this.parseResourceVersions = parseResourceVersions;
+    this.temporalPrimaryToSecondaryIndex = temporalPrimaryToSecondaryIndex;
     if (parseResourceVersions) {
       // keep up to the 50000 add/updates for up to 5 minutes
       knownResourceVersions = new ExpirationCache<>(50000, 600000);
@@ -105,10 +67,14 @@ public synchronized void onAddOrUpdateEvent(T resource) {
   }
 
   synchronized void onEvent(T resource, boolean unknownState) {
-    cache.computeIfPresent(
-        ResourceID.fromResource(resource),
-        (id, cached) ->
-            (unknownState || !isLaterResourceVersion(id, cached, resource)) ? null : cached);
+    var res =
+        cache.computeIfPresent(
+            ResourceID.fromResource(resource),
+            (id, cached) ->
+                (unknownState || !isLaterResourceVersion(id, cached, resource)) ? null : cached);
+    if (res == null) {
+      temporalPrimaryToSecondaryIndex.cleanupForResource(resource);
+    }
   }
 
   public synchronized void putAddedResource(T newResource) {
@@ -154,6 +120,7 @@ public synchronized void putResource(T newResource, String previousResourceVersi
           newResource.getMetadata().getResourceVersion(),
           resourceId);
       cache.put(resourceId, newResource);
+      temporalPrimaryToSecondaryIndex.explicitAddOrUpdate(newResource);
     } else if (cache.remove(resourceId) != null) {
       log.debug("Removed an obsolete resource from cache for id: {}", resourceId);
     }
@@ -189,4 +156,49 @@ private boolean isLaterResourceVersion(ResourceID resourceId, T newResource, T c
   public synchronized Optional<T> getResourceFromCache(ResourceID resourceID) {
     return Optional.ofNullable(cache.get(resourceID));
   }
+
+  public TemporalPrimaryToSecondaryIndex<T> getTemporalPrimaryToSecondaryIndex() {
+    return temporalPrimaryToSecondaryIndex;
+  }
+
+  static class ExpirationCache<K> {
+    private final LinkedHashMap<K, Long> cache;
+    private final int ttlMs;
+
+    public ExpirationCache(int maxEntries, int ttlMs) {
+      this.ttlMs = ttlMs;
+      this.cache =
+          new LinkedHashMap<>() {
+            @Override
+            protected boolean removeEldestEntry(Map.Entry<K, Long> eldest) {
+              return size() > maxEntries;
+            }
+          };
+    }
+
+    public void add(K key) {
+      clean();
+      cache.putIfAbsent(key, System.currentTimeMillis());
+    }
+
+    public boolean contains(K key) {
+      clean();
+      return cache.get(key) != null;
+    }
+
+    void clean() {
+      if (!cache.isEmpty()) {
+        long currentTimeMillis = System.currentTimeMillis();
+        var iter = cache.entrySet().iterator();
+        // the order will already be from oldest to newest, clean a fixed number of entries to
+        // amortize the cost amongst multiple calls
+        for (int i = 0; i < 10 && iter.hasNext(); i++) {
+          var entry = iter.next();
+          if (currentTimeMillis - entry.getValue() > ttlMs) {
+            iter.remove();
+          }
+        }
+      }
+    }
+  }
 }

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

@@ -31,7 +31,9 @@ class TemporaryPrimaryResourceCacheTest {
   @BeforeEach
   void setup() {
     informerEventSource = mock(InformerEventSource.class);
-    temporaryResourceCache = new TemporaryResourceCache<>(informerEventSource, false);
+    temporaryResourceCache =
+        new TemporaryResourceCache<>(
+            informerEventSource, NOOPTemporalPrimaryToSecondaryIndex.getInstance(), false);
   }
 
   @Test
@@ -94,7 +96,9 @@ void removesResourceFromCache() {
 
   @Test
   void resourceVersionParsing() {
-    this.temporaryResourceCache = new TemporaryResourceCache<>(informerEventSource, true);
+    this.temporaryResourceCache =
+        new TemporaryResourceCache<>(
+            informerEventSource, NOOPTemporalPrimaryToSecondaryIndex.getInstance(), true);
 
     assertThat(temporaryResourceCache.isKnownResourceVersion(testResource())).isFalse();