feat: notify Metrics when a controller's event processor starts

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

@@ -15,6 +15,7 @@
  */
 package io.javaoperatorsdk.operator.monitoring.micrometer;
 
+import java.lang.management.ManagementFactory;
 import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executors;
@@ -54,6 +55,7 @@ public class MicrometerMetrics implements Metrics {
   private static final String RECONCILIATIONS_STARTED = RECONCILIATIONS + "started";
   private static final String RECONCILIATIONS_EXECUTIONS = PREFIX + RECONCILIATIONS + "executions.";
   private static final String RECONCILIATIONS_QUEUE_SIZE = PREFIX + RECONCILIATIONS + "queue.size.";
+  private static final String PROCESSING_STARTED_LATENCY = PREFIX + "processing.started.latency.";
   private static final String NAME = "name";
   private static final String NAMESPACE = "namespace";
   private static final String GROUP = "group";
@@ -147,6 +149,16 @@ public void controllerRegistered(Controller<? extends HasMetadata> controller) {
     gauges.put(controllerQueueName, controllerQueueSize);
   }
 
+  @Override
+  public void eventProcessingStarted(Controller<? extends HasMetadata> controller) {
+    final var configuration = controller.getConfiguration();
+    final var name = configuration.getName();
+    final var tags = new ArrayList<Tag>(3);
+    addGVKTags(GroupVersionKind.gvkFor(configuration.getResourceClass()), tags, false);
+    registry.gauge(
+        PROCESSING_STARTED_LATENCY + name, tags, ManagementFactory.getRuntimeMXBean().getUptime());
+  }
+
   @Override
   public <T> T timeControllerExecution(ControllerExecution<T> execution) {
     final var name = execution.controllerName();

micrometer-support/src/main/java/io/javaoperatorsdk/operator/monitoring/micrometer/MicrometerMetricsV2.java

@@ -15,6 +15,7 @@
  */
 package io.javaoperatorsdk.operator.monitoring.micrometer;
 
+import java.lang.management.ManagementFactory;
 import java.time.Duration;
 import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
@@ -59,6 +60,7 @@ public class MicrometerMetricsV2 implements Metrics {
   public static final String RECONCILIATIONS_EXECUTIONS_GAUGE = RECONCILIATIONS + "active";
   public static final String RECONCILIATIONS_QUEUE_SIZE_GAUGE = RECONCILIATIONS + "queue";
   public static final String NUMBER_OF_RESOURCE_GAUGE = "custom_resources";
+  public static final String PROCESSING_STARTED_LATENCY_GAUGE = "processing.started.latency";
 
   public static final String RECONCILIATION_EXECUTION_DURATION =
       RECONCILIATIONS + "execution.duration";
@@ -145,6 +147,15 @@ public void controllerRegistered(Controller<? extends HasMetadata> controller) {
     executionTimers.put(name, timer);
   }
 
+  @Override
+  public void eventProcessingStarted(Controller<? extends HasMetadata> controller) {
+    final var name = controller.getConfiguration().getName();
+    final var tags = new ArrayList<Tag>();
+    addControllerNameTag(name, tags);
+    registry.gauge(
+        PROCESSING_STARTED_LATENCY_GAUGE, tags, ManagementFactory.getRuntimeMXBean().getUptime());
+  }
+
   private String numberOfResourcesRefName(String name) {
     return NUMBER_OF_RESOURCE_GAUGE + name;
   }

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/monitoring/AggregatedMetrics.java

@@ -70,6 +70,11 @@ public void controllerRegistered(Controller<? extends HasMetadata> controller) {
     metricsList.forEach(metrics -> metrics.controllerRegistered(controller));
   }
 
+  @Override
+  public void eventProcessingStarted(Controller<? extends HasMetadata> controller) {
+    metricsList.forEach(metrics -> metrics.eventProcessingStarted(controller));
+  }
+
   @Override
   public void eventReceived(Event event, Map<String, Object> metadata) {
     metricsList.forEach(metrics -> metrics.eventReceived(event, metadata));

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

@@ -41,6 +41,17 @@ public interface Metrics {
    */
   default void controllerRegistered(Controller<? extends HasMetadata> controller) {}
 
+  /**
+   * Called when the event processor of the specified controller has started and begins processing
+   * events. Until this point, events received from the event sources are deferred rather than
+   * reconciled (see the {@code EventProcessor}); in leader-election setups this is only called once
+   * leadership has been acquired. This callback can be used to measure how long the operator takes
+   * to start accepting events.
+   *
+   * @param controller the controller whose event processor has started
+   */
+  default void eventProcessingStarted(Controller<? extends HasMetadata> controller) {}
+
   /**
    * Called when an event has been accepted by the SDK from an event source, which would potentially
    * trigger the Reconciler.

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

@@ -383,6 +383,7 @@ public synchronized void start(boolean startEventProcessor) throws OperatorExcep
       eventSourceManager.start();
       if (startEventProcessor) {
         eventProcessor.start();
+        metrics.eventProcessingStarted(this);
       }
       log.info("'{}' controller started", controllerName);
     } catch (MissingCRDException e) {
@@ -429,6 +430,7 @@ public synchronized void changeNamespaces(Set<String> namespaces) {
 
   public synchronized void startEventProcessing() {
     eventProcessor.start();
+    metrics.eventProcessingStarted(this);
     log.info("Started event processing for controller: {}", configuration.getName());
   }
 

operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/ControllerTest.java

@@ -28,6 +28,7 @@
 import io.javaoperatorsdk.operator.api.config.ConfigurationService;
 import io.javaoperatorsdk.operator.api.config.MockControllerConfiguration;
 import io.javaoperatorsdk.operator.api.config.workflow.WorkflowSpec;
+import io.javaoperatorsdk.operator.api.monitoring.Metrics;
 import io.javaoperatorsdk.operator.api.reconciler.Cleaner;
 import io.javaoperatorsdk.operator.api.reconciler.DefaultContext;
 import io.javaoperatorsdk.operator.api.reconciler.DeleteControl;
@@ -68,6 +69,42 @@ void crdShouldNotBeCheckedForNativeResources() {
     verify(client, never()).apiextensions();
   }
 
+  @Test
+  void notifiesMetricsWhenEventProcessorStarts() {
+    final var client = MockKubernetesClient.client(Secret.class);
+    final var metrics = mock(Metrics.class);
+    final var configurationService =
+        ConfigurationService.newOverriddenConfigurationService(
+            new BaseConfigurationService(), o -> o.withMetrics(metrics));
+    final var configuration =
+        MockControllerConfiguration.forResource(Secret.class, configurationService);
+    final var controller = new Controller<Secret>(reconciler, configuration, client);
+
+    // Inline start (event processing started synchronously, e.g. no leader election).
+    controller.start(true);
+    verify(metrics, times(1)).eventProcessingStarted(controller);
+
+    // Deferred start (event processing started later, e.g. after acquiring leadership).
+    controller.startEventProcessing();
+    verify(metrics, times(2)).eventProcessingStarted(controller);
+  }
+
+  @Test
+  void doesNotNotifyMetricsWhenEventProcessorNotStarted() {
+    final var client = MockKubernetesClient.client(Secret.class);
+    final var metrics = mock(Metrics.class);
+    final var configurationService =
+        ConfigurationService.newOverriddenConfigurationService(
+            new BaseConfigurationService(), o -> o.withMetrics(metrics));
+    final var configuration =
+        MockControllerConfiguration.forResource(Secret.class, configurationService);
+    final var controller = new Controller<Secret>(reconciler, configuration, client);
+
+    // e.g. leader election enabled: event sources start but event processing is deferred.
+    controller.start(false);
+    verify(metrics, never()).eventProcessingStarted(controller);
+  }
+
   @Test
   void crdShouldNotBeCheckedForCustomResourcesIfDisabled() {
     final var client = MockKubernetesClient.client(TestCustomResource.class);