feat: monitor cache size

Author: metacosm

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/Metrics.java

@@ -1,5 +1,7 @@
 package io.javaoperatorsdk.operator;
 
+import java.util.Collections;
+import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import java.util.function.ToDoubleFunction;
 import java.util.function.ToLongFunction;
@@ -26,6 +28,7 @@
 
 public class Metrics {
   public static final Metrics NOOP = new Metrics(new NoopMeterRegistry(Clock.SYSTEM));
+  public static final String PREFIX = "operator.sdk.";
   private final MeterRegistry registry;
 
   public Metrics(MeterRegistry registry) {
@@ -44,7 +47,7 @@ public interface ControllerExecution<T> {
 
   public <T> T timeControllerExecution(ControllerExecution<T> execution) {
     final var name = execution.controllerName();
-    final var execName = "operator.sdk.controllers.execution." + execution.name();
+    final var execName = PREFIX + "controllers.execution." + execution.name();
     final var timer =
         Timer.builder(execName)
             .tags("controller", name)
@@ -68,25 +71,27 @@ public <T> T timeControllerExecution(ControllerExecution<T> execution) {
   }
 
   public void incrementControllerRetriesNumber() {
-
     registry
         .counter(
-            "operator.sdk.retry.on.exception", "retry", "retryCounter", "type",
+            PREFIX + "retry.on.exception", "retry", "retryCounter", "type",
             "retryException")
         .increment();
 
   }
 
   public void incrementProcessedEventsNumber() {
-
     registry
         .counter(
-            "operator.sdk.total.events.received", "events", "totalEvents", "type",
+            PREFIX + "total.events.received", "events", "totalEvents", "type",
             "eventsReceived")
         .increment();
 
   }
 
+  public <T extends Map<?, ?>> T monitorSizeOf(T map, String name) {
+    return registry.gaugeMapSize(PREFIX + name + ".size", Collections.emptyList(), map);
+  }
+
   public static class NoopMeterRegistry extends MeterRegistry {
     public NoopMeterRegistry(Clock clock) {
       super(clock);

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/CustomResourceCache.java

@@ -14,6 +14,7 @@
 import org.slf4j.LoggerFactory;
 
 import io.fabric8.kubernetes.client.CustomResource;
+import io.javaoperatorsdk.operator.Metrics;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -25,34 +26,34 @@
 public class CustomResourceCache {
 
   private static final Logger log = LoggerFactory.getLogger(CustomResourceCache.class);
+  private static final Predicate passthrough = o -> true;
 
   private final ObjectMapper objectMapper;
-  private final ConcurrentMap<String, CustomResource> resources = new ConcurrentHashMap<>();
+  private final ConcurrentMap<String, CustomResource> resources;
   private final Lock lock = new ReentrantLock();
 
   public CustomResourceCache() {
-    this(new ObjectMapper());
+    this(new ObjectMapper(), Metrics.NOOP);
   }
 
-  public CustomResourceCache(ObjectMapper objectMapper) {
+  public CustomResourceCache(ObjectMapper objectMapper, Metrics metrics) {
     this.objectMapper = objectMapper;
+    resources = metrics.monitorSizeOf(new ConcurrentHashMap<>(), "cache");
   }
 
   public void cacheResource(CustomResource resource) {
-    try {
-      lock.lock();
-      resources.put(KubernetesResourceUtils.getUID(resource), resource);
-    } finally {
-      lock.unlock();
-    }
+    cacheResource(resource, passthrough);
   }
 
   public void cacheResource(CustomResource resource, Predicate<CustomResource> predicate) {
     try {
       lock.lock();
-      if (predicate.test(resources.get(KubernetesResourceUtils.getUID(resource)))) {
-        log.trace("Update cache after condition is true: {}", getName(resource));
-        resources.put(getUID(resource), resource);
+      final var uid = getUID(resource);
+      if (predicate.test(resources.get(uid))) {
+        if (passthrough != predicate) {
+          log.trace("Update cache after condition is true: {}", getName(resource));
+        }
+        resources.put(uid, resource);
       }
     } finally {
       lock.unlock();

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/internal/CustomResourceEventSource.java

@@ -39,8 +39,9 @@ public class CustomResourceEventSource<T extends CustomResource<?, ?>> extends A
   public CustomResourceEventSource(ConfiguredController<T> controller) {
     this.controller = controller;
     this.watches = new LinkedList<>();
-    this.customResourceCache = new CustomResourceCache(
-        controller.getConfiguration().getConfigurationService().getObjectMapper());
+    final var configurationService = controller.getConfiguration().getConfigurationService();
+    this.customResourceCache = new CustomResourceCache(configurationService.getObjectMapper(),
+        configurationService.getMetrics());
   }
 
   @Override