refactor: do not rely on DefaultEventHandler so much for EventMonitor

The goal is to not rely on DefaultEventHandler eventually. EventMonitor
was kept on DefaultEventHandler for backwards compatibility reason but
this should be moved to its own package along with the Metrics class
for v2

Author: metacosm

micrometer-support/src/main/java/io/javaoperatorsdk/operator/micrometer/MicrometerMetrics.java

@@ -4,14 +4,26 @@
 import java.util.Map;
 
 import io.javaoperatorsdk.operator.Metrics;
-import io.javaoperatorsdk.operator.Metrics.ControllerExecution;
+import io.javaoperatorsdk.operator.processing.DefaultEventHandler.EventMonitor;
+import io.javaoperatorsdk.operator.processing.event.Event;
 import io.micrometer.core.instrument.MeterRegistry;
 import io.micrometer.core.instrument.Timer;
 
 public class MicrometerMetrics implements Metrics {
 
   public static final String PREFIX = "operator.sdk.";
   private final MeterRegistry registry;
+  private final EventMonitor monitor = new EventMonitor() {
+    @Override
+    public void processedEvent(String uid, Event event) {
+      incrementProcessedEventsNumber();
+    }
+
+    @Override
+    public void failedEvent(String uid, Event event) {
+      incrementControllerRetriesNumber();
+    }
+  };
 
   public MicrometerMetrics(MeterRegistry registry) {
     this.registry = registry;
@@ -63,4 +75,9 @@ public void incrementProcessedEventsNumber() {
   public <T extends Map<?, ?>> T monitorSizeOf(T map, String name) {
     return registry.gaugeMapSize(PREFIX + name + ".size", Collections.emptyList(), map);
   }
+
+  @Override
+  public EventMonitor getEventMonitor() {
+    return monitor;
+  }
 }

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

@@ -2,6 +2,9 @@
 
 import java.util.Map;
 
+import io.javaoperatorsdk.operator.processing.DefaultEventHandler;
+import io.javaoperatorsdk.operator.processing.DefaultEventHandler.EventMonitor;
+
 public interface Metrics {
   Metrics NOOP = new Metrics() {};
 
@@ -27,4 +30,8 @@ default void incrementProcessedEventsNumber() {}
   default <T extends Map<?, ?>> T monitorSizeOf(T map, String name) {
     return map;
   }
+
+  default DefaultEventHandler.EventMonitor getEventMonitor() {
+    return EventMonitor.NOOP;
+  }
 }

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

@@ -18,9 +18,6 @@
 import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
 import io.javaoperatorsdk.operator.api.config.ExecutorServiceManager;
 import io.javaoperatorsdk.operator.processing.ConfiguredController;
-import io.javaoperatorsdk.operator.processing.DefaultEventHandler;
-import io.javaoperatorsdk.operator.processing.DefaultEventHandler.EventMonitor;
-import io.javaoperatorsdk.operator.processing.event.Event;
 
 @SuppressWarnings("rawtypes")
 public class Operator implements AutoCloseable {
@@ -32,17 +29,6 @@ public class Operator implements AutoCloseable {
   public Operator(KubernetesClient k8sClient, ConfigurationService configurationService) {
     this.k8sClient = k8sClient;
     this.configurationService = configurationService;
-    DefaultEventHandler.setEventMonitor(new EventMonitor() {
-      @Override
-      public void processedEvent(String uid, Event event) {
-        configurationService.getMetrics().incrementProcessedEventsNumber();
-      }
-
-      @Override
-      public void failedEvent(String uid, Event event) {
-        configurationService.getMetrics().incrementControllerRetriesNumber();
-      }
-    });
   }
 
   /** Adds a shutdown hook that automatically calls {@link #close()} when the app shuts down. */

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

@@ -14,6 +14,7 @@
 import org.slf4j.LoggerFactory;
 
 import io.fabric8.kubernetes.client.CustomResource;
+import io.javaoperatorsdk.operator.Metrics;
 import io.javaoperatorsdk.operator.api.RetryInfo;
 import io.javaoperatorsdk.operator.api.config.ConfigurationService;
 import io.javaoperatorsdk.operator.api.config.ExecutorServiceManager;
@@ -35,13 +36,9 @@
 public class DefaultEventHandler<R extends CustomResource<?, ?>> implements EventHandler {
 
   private static final Logger log = LoggerFactory.getLogger(DefaultEventHandler.class);
-  private static EventMonitor monitor = new EventMonitor() {
-    @Override
-    public void processedEvent(String uid, Event event) {}
 
-    @Override
-    public void failedEvent(String uid, Event event) {}
-  };
+  @Deprecated
+  private static EventMonitor monitor = EventMonitor.NOOP;
 
   private final EventBuffer eventBuffer;
   private final Set<String> underProcessing = new HashSet<>();
@@ -51,23 +48,25 @@ public void failedEvent(String uid, Event event) {}
   private final ExecutorService executor;
   private final String controllerName;
   private final ReentrantLock lock = new ReentrantLock();
+  private final EventMonitor eventMonitor;
   private volatile boolean running;
   private DefaultEventSourceManager<R> eventSourceManager;
 
   public DefaultEventHandler(ConfiguredController<R> controller) {
     this(ExecutorServiceManager.instance().executorService(),
         controller.getConfiguration().getName(),
         new EventDispatcher<>(controller),
-        GenericRetry.fromConfiguration(controller.getConfiguration().getRetryConfiguration()));
+        GenericRetry.fromConfiguration(controller.getConfiguration().getRetryConfiguration()),
+        controller.getConfiguration().getConfigurationService().getMetrics().getEventMonitor());
   }
 
   DefaultEventHandler(EventDispatcher<R> eventDispatcher, String relatedControllerName,
       Retry retry) {
-    this(null, relatedControllerName, eventDispatcher, retry);
+    this(null, relatedControllerName, eventDispatcher, retry, null);
   }
 
   private DefaultEventHandler(ExecutorService executor, String relatedControllerName,
-      EventDispatcher<R> eventDispatcher, Retry retry) {
+      EventDispatcher<R> eventDispatcher, Retry retry, EventMonitor monitor) {
     this.running = true;
     this.executor =
         executor == null
@@ -78,14 +77,44 @@ private DefaultEventHandler(ExecutorService executor, String relatedControllerNa
     this.eventDispatcher = eventDispatcher;
     this.retry = retry;
     this.eventBuffer = new EventBuffer();
+    this.eventMonitor = monitor != null ? monitor : EventMonitor.NOOP;
+  }
+
+  public void setEventSourceManager(DefaultEventSourceManager<R> eventSourceManager) {
+    this.eventSourceManager = eventSourceManager;
   }
 
+  /**
+   * @deprecated the EventMonitor to be used should now be retrieved from
+   *             {@link Metrics#getEventMonitor()}
+   * @param monitor
+   */
+  @Deprecated
   public static void setEventMonitor(EventMonitor monitor) {
     DefaultEventHandler.monitor = monitor;
   }
 
-  public void setEventSourceManager(DefaultEventSourceManager<R> eventSourceManager) {
-    this.eventSourceManager = eventSourceManager;
+  /*
+   * TODO: promote this interface to top-level, probably create a `monitoring` package?
+   */
+  public interface EventMonitor {
+    EventMonitor NOOP = new EventMonitor() {
+      @Override
+      public void processedEvent(String uid, Event event) {}
+
+      @Override
+      public void failedEvent(String uid, Event event) {}
+    };
+
+    void processedEvent(String uid, Event event);
+
+    void failedEvent(String uid, Event event);
+  }
+
+  private EventMonitor monitor() {
+    // todo: remove us of static monitor, only here for backwards compatibility
+    return DefaultEventHandler.monitor != EventMonitor.NOOP ? DefaultEventHandler.monitor
+        : eventMonitor;
   }
 
   @Override
@@ -102,6 +131,7 @@ public void handleEvent(Event event) {
       log.debug("Received event: {}", event);
 
       final Predicate<CustomResource> selector = event.getCustomResourcesSelector();
+      final var monitor = monitor();
       for (String uid : eventSourceManager.getLatestResourceUids(selector)) {
         eventBuffer.addEvent(uid, event);
         monitor.processedEvent(uid, event);
@@ -168,6 +198,7 @@ void eventProcessingFinished(
 
       if (retry != null && postExecutionControl.exceptionDuringExecution()) {
         handleRetryOnException(executionScope);
+        final var monitor = monitor();
         executionScope.getEvents()
             .forEach(e -> monitor.failedEvent(executionScope.getCustomResourceUid(), e));
         return;
@@ -296,11 +327,6 @@ private void unsetUnderExecution(String customResourceUid) {
     underProcessing.remove(customResourceUid);
   }
 
-  public interface EventMonitor {
-    void processedEvent(String uid, Event event);
-
-    void failedEvent(String uid, Event event);
-  }
 
   private class ControllerExecution implements Runnable {
     private final ExecutionScope<R> executionScope;