working bounded cache

Author: csviri

caffein-bounded-cache-support/pom.xml

@@ -42,6 +42,18 @@
       <artifactId>crd-generator-apt</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.apache.logging.log4j</groupId>
+      <artifactId>log4j-slf4j-impl</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.logging.log4j</groupId>
+      <artifactId>log4j-core</artifactId>
+      <version>${log4j.version}</version>
+      <type>test-jar</type>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 
   <build>

caffein-bounded-cache-support/src/main/java/io/javaoperatorsdk/operator/processing/event/source/cache/CaffeinBoundedCache.java

@@ -30,14 +30,4 @@ public R put(K key, R object) {
     cache.put(key, object);
     return object;
   }
-
-  @Override
-  public Set<K> keys() {
-    return cache.asMap().keySet();
-  }
-
-  @Override
-  public Set<R> values() {
-    return new HashSet<>(cache.asMap().values());
-  }
 }

caffein-bounded-cache-support/src/main/java/io/javaoperatorsdk/operator/processing/event/source/cache/CaffeinBoundedItemStore.java

@@ -0,0 +1,4 @@
+package io.javaoperatorsdk.operator.processing.event.source.cache;
+
+public class CaffeinBoundedItemStore {
+}

caffein-bounded-cache-support/src/test/java/io/javaoperatorsdk/operator/processing/event/source/cache/CaffeinBoundedCacheIT.java

@@ -30,13 +30,13 @@ class CaffeinBoundedCacheIT {
   @RegisterExtension
   LocallyRunOperatorExtension extension =
       LocallyRunOperatorExtension.builder().withReconciler(new BoundedCacheTestReconciler(), o -> {
-        Cache<String, ConfigMap> cache = Caffeine.newBuilder()
+        Cache<String, BoundedCacheTestCustomResource> cache = Caffeine.newBuilder()
             .expireAfterAccess(1, TimeUnit.MINUTES)
             .maximumSize(1)
             .build();
-        BoundedItemStore<ConfigMap> boundedItemStore =
-            new BoundedItemStore<>(new KubernetesClientBuilder().build(), ConfigMap.class,
-                new CaffeinBoundedCache<>(cache));
+        BoundedItemStore<BoundedCacheTestCustomResource> boundedItemStore =
+            new BoundedItemStore<>(new KubernetesClientBuilder().build(),
+                new CaffeinBoundedCache<>(cache), BoundedCacheTestCustomResource.class);
         o.withItemStore(boundedItemStore);
       })
           .build();

caffein-bounded-cache-support/src/test/java/io/javaoperatorsdk/operator/processing/event/source/cache/sample/BoundedCacheTestCustomResource.java

@@ -1,5 +1,6 @@
 package io.javaoperatorsdk.operator.processing.event.source.cache.sample;
 
+import io.fabric8.kubernetes.api.model.Namespaced;
 import io.fabric8.kubernetes.client.CustomResource;
 import io.fabric8.kubernetes.model.annotation.Group;
 import io.fabric8.kubernetes.model.annotation.ShortNames;
@@ -9,5 +10,5 @@
 @Version("v1")
 @ShortNames("bct")
 public class BoundedCacheTestCustomResource
-    extends CustomResource<BoundedCacheTestSpec, BoundedCacheTestStatus> {
+    extends CustomResource<BoundedCacheTestSpec, BoundedCacheTestStatus> implements Namespaced {
 }

caffein-bounded-cache-support/src/test/java/io/javaoperatorsdk/operator/processing/event/source/cache/sample/BoundedCacheTestReconciler.java

@@ -3,6 +3,9 @@
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import io.fabric8.kubernetes.api.model.ConfigMap;
 import io.fabric8.kubernetes.api.model.ConfigMapBuilder;
 import io.fabric8.kubernetes.api.model.ObjectMetaBuilder;
@@ -22,6 +25,8 @@
 public class BoundedCacheTestReconciler implements Reconciler<BoundedCacheTestCustomResource>,
     EventSourceInitializer<BoundedCacheTestCustomResource>, KubernetesClientAware {
 
+  private static final Logger log = LoggerFactory.getLogger(BoundedCacheTestReconciler.class);
+
   public static final String DATA_KEY = "dataKey";
   private KubernetesClient client;
 
@@ -34,6 +39,7 @@ public UpdateControl<BoundedCacheTestCustomResource> reconcile(
         cm -> updateConfigMapIfNeeded(cm, resource),
         () -> createConfigMap(resource));
     ensureStatus(resource);
+    log.info("Reconciled: {}", resource.getMetadata().getName());
     return UpdateControl.patchStatus(resource);
   }
 
@@ -65,8 +71,8 @@ public Map<String, EventSource> prepareEventSources(
         .maximumSize(1)
         .build();
 
-    BoundedItemStore<ConfigMap> boundedItemStore = new BoundedItemStore<>(client, ConfigMap.class,
-        new CaffeinBoundedCache<>(cache));
+    BoundedItemStore<ConfigMap> boundedItemStore = new BoundedItemStore<>(client,
+        new CaffeinBoundedCache<>(cache), ConfigMap.class);
 
     var es = new InformerEventSource<>(InformerConfiguration.from(ConfigMap.class, context)
         .withItemStore(boundedItemStore)

caffein-bounded-cache-support/src/test/resources/log4j2.xml

@@ -0,0 +1,13 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<Configuration status="WARN">
+  <Appenders>
+    <Console name="Console" target="SYSTEM_OUT">
+      <PatternLayout pattern="%d %threadId %-30c{1.} [%-5level] %msg%n%throwable"/>
+    </Console>
+  </Appenders>
+  <Loggers>
+    <Root level="debug">
+      <AppenderRef ref="Console"/>
+    </Root>
+  </Loggers>
+</Configuration>

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/cache/BoundedCache.java

@@ -1,7 +1,5 @@
 package io.javaoperatorsdk.operator.processing.event.source.cache;
 
-import java.util.Set;
-
 // todo: rename to cache? (replace the old one)
 public interface BoundedCache<K, R> {
 
@@ -11,8 +9,4 @@ public interface BoundedCache<K, R> {
 
   R put(K key, R object);
 
-  Set<K> keys();
-
-  Set<R> values();
-
 }

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/cache/BoundedItemStore.java

@@ -1,60 +1,137 @@
 package io.javaoperatorsdk.operator.processing.event.source.cache;
 
-import java.util.HashSet;
+import java.lang.reflect.InvocationTargetException;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.Function;
 import java.util.stream.Stream;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.fabric8.kubernetes.api.model.ObjectMetaBuilder;
 import io.fabric8.kubernetes.client.KubernetesClient;
 import io.fabric8.kubernetes.client.informers.cache.Cache;
 import io.fabric8.kubernetes.client.informers.cache.ItemStore;
 
-public class BoundedItemStore<R extends HasMetadata> extends BoundedStore<String, R>
+public class BoundedItemStore<R extends HasMetadata>
     implements ItemStore<R> {
 
+  private Logger log = LoggerFactory.getLogger(BoundedItemStore.class);
+
+  private final ResourceFetcher<String, R> resourceFetcher;
+  private final BoundedCache<String, R> cache;
   private final Function<R, String> keyFunction;
+  private final Map<String, R> existingMinimalResources = new ConcurrentHashMap<>();
+  private final Class<R> resourceClass;
 
-  public BoundedItemStore(KubernetesClient client, Class<R> resourceClass,
-      BoundedCache<String, R> cache) {
-    this(new KubernetesResourceFetcher<>(resourceClass, client), cache, namespaceKeyFunc());
+  public BoundedItemStore(KubernetesClient client,
+      BoundedCache<String, R> cache, Class<R> resourceClass) {
+    this(client, cache, resourceClass, namespaceKeyFunc());
   }
 
-  public BoundedItemStore(KubernetesResourceFetcher<R> resourceFetcher,
-      BoundedCache<String, R> cache, Function<R, String> keyFunction) {
-    super(resourceFetcher, cache);
+  public BoundedItemStore(KubernetesClient client,
+      BoundedCache<String, R> cache,
+      Class<R> resourceClass,
+      Function<R, String> keyFunction) {
+    this.resourceFetcher = new KubernetesResourceFetcher<>(resourceClass, client);
+    this.cache = cache;
     this.keyFunction = keyFunction;
+    this.resourceClass = resourceClass;
   }
 
   @Override
   public String getKey(R obj) {
     return keyFunction.apply(obj);
   }
 
+  @Override
+  public synchronized R put(String key, R obj) {
+    var result = existingMinimalResources.get(key);
+    cache.put(key, obj);
+    existingMinimalResources.put(key, createMinimalResource(obj));
+    return result;
+  }
+
+  private R createMinimalResource(R obj) {
+    try {
+      R minimal = resourceClass.getConstructor().newInstance();
+      minimal.setMetadata(new ObjectMetaBuilder().build());
+      minimal.getMetadata().setName(obj.getMetadata().getName());
+      minimal.getMetadata().setNamespace(obj.getMetadata().getNamespace());
+      minimal.getMetadata().setResourceVersion(obj.getMetadata().getResourceVersion());
+      return minimal;
+    } catch (InstantiationException | IllegalAccessException | InvocationTargetException
+        | NoSuchMethodException e) {
+      throw new IllegalStateException(e);
+    }
+  }
+
+  @Override
+  public synchronized R remove(String key) {
+    var fullValue = cache.remove(key);
+    var minimalValue = existingMinimalResources.remove(key);
+    return fullValue != null ? fullValue : minimalValue;
+  }
+
   @Override
   public Stream<String> keySet() {
-    return super.keys();
+    return existingMinimalResources.keySet().stream();
   }
 
-  /**
-   * This is very inefficient but should not be called by the Informer or just before it's started
-   */
   @Override
   public Stream<R> values() {
-    var keys = cache.keys();
-    var values = cache.values();
-    var notPresentValueKeys = new HashSet<>(existingResources);
-    notPresentValueKeys.retainAll(keys);
-    var fetchedValues =
-        notPresentValueKeys.stream().map(k -> fetchAndCacheResourceIfStillNonPresent(k));
-    return Stream.concat(values.stream(), fetchedValues);
+    return existingMinimalResources.values().stream();
   }
 
   @Override
   public int size() {
-    return existingResources.size();
+    return existingMinimalResources.size();
+  }
+
+  @Override
+  public R get(String key) {
+    var res = cache.get(key);
+    if (res != null) {
+      return res;
+    }
+    if (!existingMinimalResources.containsKey(key)) {
+      return null;
+    } else {
+      return refreshMissingStateFromServer(key);
+    }
+  }
+
+  @Override
+  public boolean isFullState() {
+    return false;
   }
 
   public static <R extends HasMetadata> Function<R, String> namespaceKeyFunc() {
     return r -> Cache.namespaceKeyFunc(r.getMetadata().getNamespace(), r.getMetadata().getName());
   }
+
+  protected R refreshMissingStateFromServer(String key) {
+    log.debug("Fetching resource from server");
+    var newRes = resourceFetcher.fetchResource(key);
+    synchronized (this) {
+      log.debug("Fetched resource: {}", newRes);
+      if (newRes == null) {
+        existingMinimalResources.remove(key);
+        return null;
+      }
+      // Just want to put the fetched resource if there is still no resource published from
+      // different source. In case of informers actually multiple events might arrive, therefore non
+      // fetched resource should take always precedence.
+      var actual = cache.get(key);
+      if (actual == null) {
+        cache.put(key, newRes);
+        existingMinimalResources.put(key, createMinimalResource(newRes));
+        return newRes;
+      } else {
+        return actual;
+      }
+    }
+  }
 }