From 55ebfd9bab3effe29094fb8c8d17167af40c497f Mon Sep 17 00:00:00 2001 From: Zhou JIANG Date: Thu, 9 Jul 2026 16:01:40 -0700 Subject: [PATCH] feat: notify Metrics when a controller's event processor starts 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. --- .../micrometer/MicrometerMetrics.java | 12 ++++++ .../micrometer/MicrometerMetricsV2.java | 11 ++++++ .../api/monitoring/AggregatedMetrics.java | 5 +++ .../operator/api/monitoring/Metrics.java | 11 ++++++ .../operator/processing/Controller.java | 2 + .../operator/processing/ControllerTest.java | 37 +++++++++++++++++++ 6 files changed, 78 insertions(+) diff --git a/micrometer-support/src/main/java/io/javaoperatorsdk/operator/monitoring/micrometer/MicrometerMetrics.java b/micrometer-support/src/main/java/io/javaoperatorsdk/operator/monitoring/micrometer/MicrometerMetrics.java index 60f58b411f..0c102e52c4 100644 --- a/micrometer-support/src/main/java/io/javaoperatorsdk/operator/monitoring/micrometer/MicrometerMetrics.java +++ b/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 controller) { gauges.put(controllerQueueName, controllerQueueSize); } + @Override + public void eventProcessingStarted(Controller controller) { + final var configuration = controller.getConfiguration(); + final var name = configuration.getName(); + final var tags = new ArrayList(3); + addGVKTags(GroupVersionKind.gvkFor(configuration.getResourceClass()), tags, false); + registry.gauge( + PROCESSING_STARTED_LATENCY + name, tags, ManagementFactory.getRuntimeMXBean().getUptime()); + } + @Override public T timeControllerExecution(ControllerExecution execution) { final var name = execution.controllerName(); diff --git a/micrometer-support/src/main/java/io/javaoperatorsdk/operator/monitoring/micrometer/MicrometerMetricsV2.java b/micrometer-support/src/main/java/io/javaoperatorsdk/operator/monitoring/micrometer/MicrometerMetricsV2.java index deb6c98c9e..b342061720 100644 --- a/micrometer-support/src/main/java/io/javaoperatorsdk/operator/monitoring/micrometer/MicrometerMetricsV2.java +++ b/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 controller) { executionTimers.put(name, timer); } + @Override + public void eventProcessingStarted(Controller controller) { + final var name = controller.getConfiguration().getName(); + final var tags = new ArrayList(); + addControllerNameTag(name, tags); + registry.gauge( + PROCESSING_STARTED_LATENCY_GAUGE, tags, ManagementFactory.getRuntimeMXBean().getUptime()); + } + private String numberOfResourcesRefName(String name) { return NUMBER_OF_RESOURCE_GAUGE + name; } diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/monitoring/AggregatedMetrics.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/monitoring/AggregatedMetrics.java index 6ae3ebe65a..374f4a8a45 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/monitoring/AggregatedMetrics.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/monitoring/AggregatedMetrics.java @@ -70,6 +70,11 @@ public void controllerRegistered(Controller controller) { metricsList.forEach(metrics -> metrics.controllerRegistered(controller)); } + @Override + public void eventProcessingStarted(Controller controller) { + metricsList.forEach(metrics -> metrics.eventProcessingStarted(controller)); + } + @Override public void eventReceived(Event event, Map metadata) { metricsList.forEach(metrics -> metrics.eventReceived(event, metadata)); diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/monitoring/Metrics.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/monitoring/Metrics.java index fbbb20d92d..67fda2eb12 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/monitoring/Metrics.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/monitoring/Metrics.java @@ -41,6 +41,17 @@ public interface Metrics { */ default void controllerRegistered(Controller 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 controller) {} + /** * Called when an event has been accepted by the SDK from an event source, which would potentially * trigger the Reconciler. diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/Controller.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/Controller.java index bc3a43a9a3..5df4a86f57 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/Controller.java +++ b/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 namespaces) { public synchronized void startEventProcessing() { eventProcessor.start(); + metrics.eventProcessingStarted(this); log.info("Started event processing for controller: {}", configuration.getName()); } diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/ControllerTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/ControllerTest.java index 9eeba7ab1a..91d60f7aa7 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/ControllerTest.java +++ b/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(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(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);