This is an automated email from the ASF dual-hosted git repository. dominikriemer pushed a commit to branch remove-max-retries-health-check in repository https://gitbox.apache.org/repos/asf/streampipes.git
commit aa6cab90b99fdd8361fda4f2530e0e8477553aaa Author: Dominik Riemer <[email protected]> AuthorDate: Mon Jul 20 17:52:28 2026 +0200 fix: Remove max retries from pipeline health check --- .../health/monitoring/PipelineHealthCheck.java | 98 +++++-------- .../health/monitoring/PipelineHealthCheckTest.java | 153 +++++++++++++++++++++ 2 files changed, 184 insertions(+), 67 deletions(-) diff --git a/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/PipelineHealthCheck.java b/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/PipelineHealthCheck.java index f3d88fad16..5d6b5263a4 100644 --- a/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/PipelineHealthCheck.java +++ b/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/PipelineHealthCheck.java @@ -41,19 +41,13 @@ import org.slf4j.LoggerFactory; import java.util.ArrayList; import java.util.Collections; -import java.util.HashMap; import java.util.List; -import java.util.Map; import java.util.Objects; -import java.util.concurrent.atomic.AtomicBoolean; import java.util.stream.Stream; public class PipelineHealthCheck implements HealthCheck { private static final Logger LOG = LoggerFactory.getLogger(PipelineHealthCheck.class); - private static final int MAX_FAILED_ATTEMPTS = 10; - - private static final Map<String, Integer> failedRestartAttempts = new HashMap<>(); private static final PipelinesStats pipelinesStats = new PipelinesStats(); private final HealthCheckData healthCheckData; @@ -109,7 +103,6 @@ public class PipelineHealthCheck implements HealthCheck { private void checkAndRestorePipelineElements() { healthCheckData.activeResources().runningPipelines().forEach(pipeline -> { - AtomicBoolean shouldUpdatePipeline = new AtomicBoolean(false); List<String> failedInstances = new ArrayList<>(); List<String> recoveredInstances = new ArrayList<>(); List<String> pipelineNotifications = new ArrayList<>(); @@ -121,41 +114,20 @@ public class PipelineHealthCheck implements HealthCheck { runningPipelineElements.forEach(pipelineElement -> { String instanceId = HealthCheckUtils.extractInstanceId(pipelineElement); if (isNowhereRunning(instanceId)) { - if (shouldRetry(instanceId)) { - shouldUpdatePipeline.set(true); - boolean success; - try { - var service = new ExtensionsServiceEndpointGenerator().selectService( - pipelineElement.getAppId(), - ExtensionsServiceEndpointUtils.getPipelineElementType(pipelineElement.getAppId()), - Collections.emptySet() - ); - new SecretService(new SecretDecrypter()).apply(pipelineElement); - pipelineElement.setSelectedEndpointUrl(service.getServiceUrl()); - pipelineElement.setSelectedServiceId(service.getSvcId()); - success = new InvokeExtensionRequest(requestManager, resourceManager) - .execute(pipelineElement, pipeline.getPipelineId()).isSuccess(); - new SecretService(new SecretEncrypter()).apply(pipelineElement); - } catch (NoServiceEndpointsAvailableException e) { - success = false; - } - if (!success) { - failedInstances.add(instanceId); - HealthCheckUtils.addFailedAttemptNotification(pipelineNotifications, pipelineElement); - increaseFailedAttempt(instanceId); - LOG.info("Could not restore pipeline element {} of pipeline {} ({}/{})", - pipelineElement.getName(), pipeline.getName(), failedRestartAttempts.get(instanceId), - MAX_FAILED_ATTEMPTS); - } else { - recoveredInstances.add(instanceId); - resetFailedAttempts(instanceId); - LOG.info("Successfully restored pipeline element {} of pipeline {}", - pipelineElement.getName(), pipeline.getName()); - } + boolean success = restorePipelineElement(pipelineElement, pipeline.getPipelineId()); + if (!success) { + failedInstances.add(instanceId); + HealthCheckUtils.addFailedAttemptNotification(pipelineNotifications, pipelineElement); + LOG.info("Could not restore pipeline element {} of pipeline {}", + pipelineElement.getName(), pipeline.getName()); + } else { + recoveredInstances.add(instanceId); + LOG.info("Successfully restored pipeline element {} of pipeline {}", + pipelineElement.getName(), pipeline.getName()); } } }); - if (shouldUpdatePipeline.get()) { + if (!failedInstances.isEmpty() || !recoveredInstances.isEmpty()) { var currentPipeline = resourceProvider.pipelineStorage().getElementById(pipeline.getPipelineId()); if (!failedInstances.isEmpty()) { currentPipeline.setHealthStatus(PipelineHealthStatus.FAILURE); @@ -182,40 +154,32 @@ public class PipelineHealthCheck implements HealthCheck { pipelinesStats.setElementCount(getElementsCount(healthCheckData.activeResources().allPipelines())); } - private boolean isNowhereRunning(String instanceId) { - return (healthCheckData.activeExtensionInstances().entrySet().stream() - .noneMatch(entry -> entry.getValue().runningPipelineElementInstanceIds().contains(instanceId))); - } - - private boolean shouldRetry(String instanceId) { - if (!failedRestartAttempts.containsKey(instanceId)) { - return true; - } else { - return failedRestartAttempts.get(instanceId) < MAX_FAILED_ATTEMPTS; + protected boolean restorePipelineElement(InvocableStreamPipesEntity pipelineElement, + String pipelineId) { + try { + var service = new ExtensionsServiceEndpointGenerator().selectService( + pipelineElement.getAppId(), + ExtensionsServiceEndpointUtils.getPipelineElementType(pipelineElement.getAppId()), + Collections.emptySet() + ); + new SecretService(new SecretDecrypter()).apply(pipelineElement); + pipelineElement.setSelectedEndpointUrl(service.getServiceUrl()); + pipelineElement.setSelectedServiceId(service.getSvcId()); + boolean success = new InvokeExtensionRequest(requestManager, resourceManager) + .execute(pipelineElement, pipelineId).isSuccess(); + new SecretService(new SecretEncrypter()).apply(pipelineElement); + return success; + } catch (NoServiceEndpointsAvailableException e) { + return false; } } - private void resetFailedAttempts(String instanceId) { - failedRestartAttempts.put(instanceId, 0); - } - - private void increaseFailedAttempt(String instanceId) { - if (!failedRestartAttempts.containsKey(instanceId)) { - failedRestartAttempts.put(instanceId, 1); - } else { - Integer currentAttempt = failedRestartAttempts.get(instanceId) + 1; - failedRestartAttempts.put(instanceId, currentAttempt); - } + private boolean isNowhereRunning(String instanceId) { + return (healthCheckData.activeExtensionInstances().entrySet().stream() + .noneMatch(entry -> entry.getValue().runningPipelineElementInstanceIds().contains(instanceId))); } private int getElementsCount(List<Pipeline> allPipelines) { return allPipelines.stream().mapToInt(pipeline -> pipeline.getActions().size()).sum(); } - - private String getInvocationUrl(InvocableStreamPipesEntity pipelineElement, - String baseUrl) { - return ExtensionsServiceEndpointUtils - .getPipelineElementType(pipelineElement) - .getInvocationUrl(baseUrl, pipelineElement.getAppId()); - } } diff --git a/streampipes-health-monitoring/src/test/java/org/apache/streampipes/health/monitoring/PipelineHealthCheckTest.java b/streampipes-health-monitoring/src/test/java/org/apache/streampipes/health/monitoring/PipelineHealthCheckTest.java new file mode 100644 index 0000000000..6f5dc89493 --- /dev/null +++ b/streampipes-health-monitoring/src/test/java/org/apache/streampipes/health/monitoring/PipelineHealthCheckTest.java @@ -0,0 +1,153 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ +package org.apache.streampipes.health.monitoring; + +import org.apache.streampipes.health.monitoring.model.ActiveResources; +import org.apache.streampipes.health.monitoring.model.HealthCheckData; +import org.apache.streampipes.health.monitoring.utils.HealthCheckUtils; +import org.apache.streampipes.model.base.InvocableStreamPipesEntity; +import org.apache.streampipes.model.graph.DataProcessorInvocation; +import org.apache.streampipes.model.pipeline.Pipeline; +import org.apache.streampipes.model.pipeline.PipelineHealthStatus; +import org.apache.streampipes.storage.api.pipeline.IPipelineStorage; + +import org.junit.jupiter.api.Test; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +public class PipelineHealthCheckTest { + + @Test + public void retriesRestorationBeyondPreviousLimit() { + var pipeline = pipeline(processor("missing", "Missing processor")); + var healthCheck = healthCheck(pipeline, Map.of("missing", false)); + + for (int i = 0; i < 11; i++) { + healthCheck.runCheck(); + } + + assertEquals(11, healthCheck.restoreAttempts("missing")); + assertEquals(PipelineHealthStatus.FAILURE, pipeline.getHealthStatus()); + } + + @Test + public void recoveredElementDoesNotMaskFailedElement() { + var pipeline = pipeline( + processor("recovered", "Recovered processor"), + processor("missing", "Missing processor") + ); + var healthCheck = healthCheck(pipeline, Map.of( + "recovered", true, + "missing", false + )); + + healthCheck.runCheck(); + + assertEquals(PipelineHealthStatus.FAILURE, pipeline.getHealthStatus()); + } + + @Test + public void pipelineIsHealthyWhenAllMissingElementsAreRestored() { + var pipeline = pipeline( + processor("first", "First processor"), + processor("second", "Second processor") + ); + pipeline.setHealthStatus(PipelineHealthStatus.FAILURE); + var healthCheck = healthCheck(pipeline, Map.of( + "first", true, + "second", true + )); + + healthCheck.runCheck(); + + assertEquals(PipelineHealthStatus.OK, pipeline.getHealthStatus()); + } + + private TestPipelineHealthCheck healthCheck(Pipeline pipeline, + Map<String, Boolean> restorationResults) { + IPipelineStorage pipelineStorage = mock(IPipelineStorage.class); + when(pipelineStorage.getElementById(pipeline.getPipelineId())).thenReturn(pipeline); + when(pipelineStorage.updateElement(any())).thenAnswer(invocation -> invocation.getArgument(0)); + + var resourceProvider = new ResourceProvider(pipelineStorage, null, null); + var activeResources = new ActiveResources( + List.of(pipeline), + List.of(pipeline), + List.of(), + List.of() + ); + var healthCheckData = new HealthCheckData( + resourceProvider, + activeResources, + Map.of(), + Map.of() + ); + + return new TestPipelineHealthCheck(healthCheckData, resourceProvider, restorationResults); + } + + private Pipeline pipeline(DataProcessorInvocation... processors) { + var pipeline = new Pipeline(); + pipeline.setPipelineId("pipeline-id"); + pipeline.setName("Pipeline"); + pipeline.setRunning(true); + pipeline.setHealthStatus(PipelineHealthStatus.OK); + pipeline.setSepas(List.of(processors)); + return pipeline; + } + + private DataProcessorInvocation processor(String instanceId, + String name) { + var processor = new DataProcessorInvocation(); + processor.setElementId("urn:streampipes.org:spi:" + instanceId); + processor.setName(name); + return processor; + } + + private static class TestPipelineHealthCheck extends PipelineHealthCheck { + + private final Map<String, Boolean> restorationResults; + private final Map<String, Integer> restorationAttempts = new HashMap<>(); + + TestPipelineHealthCheck(HealthCheckData healthCheckData, + ResourceProvider resourceProvider, + Map<String, Boolean> restorationResults) { + super(healthCheckData, null, resourceProvider, null); + this.restorationResults = restorationResults; + } + + @Override + protected boolean restorePipelineElement(InvocableStreamPipesEntity pipelineElement, + String pipelineId) { + var instanceId = HealthCheckUtils.extractInstanceId(pipelineElement); + restorationAttempts.merge(instanceId, 1, Integer::sum); + return restorationResults.get(instanceId); + } + + int restoreAttempts(String instanceId) { + return restorationAttempts.getOrDefault(instanceId, 0); + } + } +}
