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