feat: notify Metrics when a controller's event processor starts (#3485)

Add Metrics.eventProcessingStarted(Controller), invoked in Controller
  whenever the event processor begins accepting events (both on inline
  start and after a deferred start, e.g. once leader election succeeds),
  so implementations can measure operator startup latency.

  Wire it through AggregatedMetrics so composed Metrics instances all
  receive the callback, and implement it in MicrometerMetrics/V2 as a
  per-controller gauge recording JVM uptime at the moment processing
  starts.

Author: jiangzho

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);