diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventProcessor.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventProcessor.java index ddc0f73a27..374beb91e9 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventProcessor.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventProcessor.java @@ -543,6 +543,7 @@ public void run() { // change thread name for easier debugging final var thread = Thread.currentThread(); final var name = thread.getName(); + boolean reconciliationStarted = false; try { // we try to get the most up-to-date resource from cache var actualResource = cache.get(resourceID); @@ -567,19 +568,27 @@ public void run() { } } else { log.debug("Skipping execution; primary resource missing from cache"); + // the submission has to be completed, otherwise the resource stays marked as under + // processing and no further reconciliation would ever be submitted for it + eventProcessingFinished(executionScope, PostExecutionControl.defaultDispatch()); return; } } actualResource.ifPresent(executionScope::setResource); MDCUtils.addResourceInfo(executionScope.getResource()); metrics.reconciliationStarted(executionScope.getResource(), metricsMetadata); + reconciliationStarted = true; thread.setName("ReconcilerExecutor-" + controllerName() + "-" + thread.getId()); PostExecutionControl

postExecutionControl = reconciliationDispatcher.handleExecution(executionScope); eventProcessingFinished(executionScope, postExecutionControl); } finally { - metrics.reconciliationFinished( - executionScope.getResource(), executionScope.getRetryInfo(), metricsMetadata); + // only report the reconciliation as finished if it was reported as started, otherwise + // gauges tracking in-flight reconciliations drift on every skipped execution + if (reconciliationStarted) { + metrics.reconciliationFinished( + executionScope.getResource(), executionScope.getRetryInfo(), metricsMetadata); + } // restore original name thread.setName(name); MDCUtils.removeResourceInfo(); diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/EventProcessorTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/EventProcessorTest.java index f7864f2f16..941da8be4c 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/EventProcessorTest.java +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/EventProcessorTest.java @@ -134,6 +134,53 @@ void skipProcessingIfLatestCustomResourceNotInCache() { verify(reconciliationDispatcherMock, timeout(50).times(0)).handleExecution(any()); } + @Test + void completesSubmissionIfResourceDisappearsFromCacheBeforeExecution() { + var resourceID = new ResourceID(UUID.randomUUID().toString(), TEST_NAMESPACE); + var customResource = testCustomResource(resourceID); + // present when the execution is submitted, but gone by the time the executor thread runs + when(controllerEventSourceMock.get(eq(resourceID))) + .thenReturn(Optional.of(customResource)) + .thenReturn(Optional.empty()); + + eventProcessor.handleEvent( + new ResourceEvent(ResourceAction.UPDATED, resourceID, customResource)); + + await() + .atMost(Duration.ofSeconds(1)) + .untilAsserted(() -> assertThat(eventProcessor.isUnderProcessing(resourceID)).isFalse()); + verify(reconciliationDispatcherMock, times(0)).handleExecution(any()); + } + + @Test + void doesNotReportReconciliationFinishedIfItNeverStarted() { + var processorWithMetrics = + spy( + new EventProcessor( + controllerConfiguration(null, rateLimiterMock), + reconciliationDispatcherMock, + eventSourceManagerMock, + metricsMock)); + processorWithMetrics.start(); + when(processorWithMetrics.retryEventSource()).thenReturn(retryTimerEventSourceMock); + + var resourceID = new ResourceID(UUID.randomUUID().toString(), TEST_NAMESPACE); + var customResource = testCustomResource(resourceID); + when(controllerEventSourceMock.get(eq(resourceID))) + .thenReturn(Optional.of(customResource)) + .thenReturn(Optional.empty()); + + processorWithMetrics.handleEvent( + new ResourceEvent(ResourceAction.UPDATED, resourceID, customResource)); + + await() + .atMost(Duration.ofSeconds(1)) + .untilAsserted( + () -> assertThat(processorWithMetrics.isUnderProcessing(resourceID)).isFalse()); + verify(metricsMock, times(0)).reconciliationStarted(any(), any()); + verify(metricsMock, times(0)).reconciliationFinished(any(), any(), any()); + } + @Test void ifExecutionInProgressWaitsUntilItsFinished() { ResourceID resourceUid = eventAlreadyUnderProcessing();