This is an automated email from the ASF dual-hosted git repository.
dominikriemer pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/streampipes.git
The following commit(s) were added to refs/heads/dev by this push:
new 9f56b5386f fix: Remove max retries health check (#4739)
9f56b5386f is described below
commit 9f56b5386f30effc88d4745df95f8d00cc8c1167
Author: Dominik Riemer <[email protected]>
AuthorDate: Wed Jul 22 07:21:35 2026 +0200
fix: Remove max retries health check (#4739)
---
.../pe/InvocablePipelineElementManagement.java | 2 +-
.../health/monitoring/ExtensionHealthCheck.java | 27 ++-
.../health/monitoring/PipelineHealthCheck.java | 150 +++++++-----
.../health/monitoring/PipelineRecoveryBackoff.java | 107 +++++++++
.../health/monitoring/utils/HealthCheckUtils.java | 6 +
.../health/monitoring/PipelineHealthCheckTest.java | 256 +++++++++++++++++++++
.../monitoring/PipelineRecoveryBackoffTest.java | 100 ++++++++
.../streampipes/service/core/PostStartupTask.java | 7 +-
.../service/core/StreamPipesCoreApplication.java | 23 +-
.../pipeline-operation-status.component.html | 93 +++++---
.../pipeline-operation-status.component.scss | 119 +++++++++-
.../pipeline-operation-status.component.ts | 31 ++-
ui/src/scss/sp/_variables.scss | 1 +
ui/src/scss/sp/main.scss | 2 +-
14 files changed, 810 insertions(+), 114 deletions(-)
diff --git
a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/pe/InvocablePipelineElementManagement.java
b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/pe/InvocablePipelineElementManagement.java
index 4e67c458e9..66069e40d6 100644
---
a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/pe/InvocablePipelineElementManagement.java
+++
b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/pe/InvocablePipelineElementManagement.java
@@ -64,7 +64,7 @@ public abstract class InvocablePipelineElementManagement<
public Response invokeRuntime(String appId, K graph) {
if (isDebug()) {
- LOG.info("SP_DEBUG env variable is set - overriding broker hostname and
port for local development");
+ LOG.debug("SP_DEBUG env variable is set - overriding broker hostname and
port for local development");
graph = createGroundingDebugInformation(graph);
}
diff --git
a/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/ExtensionHealthCheck.java
b/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/ExtensionHealthCheck.java
index 265cca67ea..02a8724f2c 100644
---
a/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/ExtensionHealthCheck.java
+++
b/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/ExtensionHealthCheck.java
@@ -27,6 +27,7 @@ import
org.apache.streampipes.storage.api.system.IExtensionsServiceStorage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
@@ -40,17 +41,35 @@ public class ExtensionHealthCheck implements Runnable {
private final IExtensionsServiceStorage extensionsServiceStorage;
private final SpResourceManager resourceManager;
private final List<HealthCheck> registeredHealthChecks;
+ private final PipelineRecoveryBackoff pipelineRecoveryBackoff;
public ExtensionHealthCheck(ResourceProvider resourceProvider,
IExtensionsServiceStorage
extensionsServiceStorage,
ExtensionServiceRequestManager
extensionRequestManager,
SpResourceManager resourceManager,
List<HealthCheck> registeredHealthChecks) {
+ this(
+ resourceProvider,
+ extensionsServiceStorage,
+ extensionRequestManager,
+ resourceManager,
+ registeredHealthChecks,
+ PipelineRecoveryBackoff.DEFAULT_INITIAL_DELAY
+ );
+ }
+
+ public ExtensionHealthCheck(ResourceProvider resourceProvider,
+ IExtensionsServiceStorage
extensionsServiceStorage,
+ ExtensionServiceRequestManager
extensionRequestManager,
+ SpResourceManager resourceManager,
+ List<HealthCheck> registeredHealthChecks,
+ Duration healthCheckInterval) {
this.resourceProvider = resourceProvider;
this.extensionsServiceStorage = extensionsServiceStorage;
this.extensionRequestManager = extensionRequestManager;
this.resourceManager = resourceManager;
this.registeredHealthChecks = registeredHealthChecks;
+ this.pipelineRecoveryBackoff = new
PipelineRecoveryBackoff(healthCheckInterval);
}
@Override
@@ -100,7 +119,13 @@ public class ExtensionHealthCheck implements Runnable {
protected List<HealthCheck> getBuiltInHealthChecks(HealthCheckData
healthCheckData) {
return List.of(
- new PipelineHealthCheck(healthCheckData, extensionRequestManager,
resourceProvider, resourceManager),
+ new PipelineHealthCheck(
+ healthCheckData,
+ extensionRequestManager,
+ resourceProvider,
+ resourceManager,
+ pipelineRecoveryBackoff
+ ),
new AdapterHealthCheck(healthCheckData)
);
}
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..55fbcc8947 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,34 +41,40 @@ 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.Set;
+import java.util.stream.Collectors;
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;
private final ExtensionServiceRequestManager requestManager;
private final ResourceProvider resourceProvider;
private final SpResourceManager resourceManager;
+ private final PipelineRecoveryBackoff recoveryBackoff;
public PipelineHealthCheck(HealthCheckData healthCheckData,
ExtensionServiceRequestManager requestManager,
ResourceProvider resourceProvider,
SpResourceManager resourceManager) {
+ this(healthCheckData, requestManager, resourceProvider, resourceManager,
new PipelineRecoveryBackoff());
+ }
+
+ PipelineHealthCheck(HealthCheckData healthCheckData,
+ ExtensionServiceRequestManager requestManager,
+ ResourceProvider resourceProvider,
+ SpResourceManager resourceManager,
+ PipelineRecoveryBackoff recoveryBackoff) {
this.healthCheckData = healthCheckData;
this.requestManager = requestManager;
this.resourceProvider = resourceProvider;
this.resourceManager = resourceManager;
+ this.recoveryBackoff = recoveryBackoff;
}
@Override
@@ -108,9 +114,9 @@ public class PipelineHealthCheck implements HealthCheck {
}
private void checkAndRestorePipelineElements() {
+ recoveryBackoff.retainOnly(getActiveRecoveryKeys());
healthCheckData.activeResources().runningPipelines().forEach(pipeline -> {
- AtomicBoolean shouldUpdatePipeline = new AtomicBoolean(false);
- List<String> failedInstances = new ArrayList<>();
+ List<String> missingInstances = new ArrayList<>();
List<String> recoveredInstances = new ArrayList<>();
List<String> pipelineNotifications = new ArrayList<>();
List<InvocableStreamPipesEntity> runningPipelineElements = Stream.concat(
@@ -121,46 +127,39 @@ 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;
- }
+ missingInstances.add(instanceId);
+ if (recoveryBackoff.isAttemptDue(pipeline.getPipelineId(),
instanceId)) {
+ boolean success = restorePipelineElement(pipelineElement,
pipeline.getPipelineId());
if (!success) {
- failedInstances.add(instanceId);
+ var state =
recoveryBackoff.recordFailure(pipeline.getPipelineId(), 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);
+ logFailedRecovery(pipeline, pipelineElement, state);
} else {
+ missingInstances.remove(instanceId);
recoveredInstances.add(instanceId);
- resetFailedAttempts(instanceId);
- LOG.info("Successfully restored pipeline element {} of pipeline
{}",
- pipelineElement.getName(), pipeline.getName());
+ int previousFailures =
recoveryBackoff.reset(pipeline.getPipelineId(), instanceId);
+ logSuccessfulRecovery(pipeline, pipelineElement,
previousFailures);
}
+ } else {
+
HealthCheckUtils.addPendingRecoveryNotification(pipelineNotifications,
pipelineElement);
+ }
+ } else {
+ int previousFailures =
recoveryBackoff.reset(pipeline.getPipelineId(), instanceId);
+ if (previousFailures > 0) {
+ recoveredInstances.add(instanceId);
+ LOG.info("Pipeline element {} of pipeline {} is running again
after {} failed recovery attempts",
+ pipelineElement.getName(), pipeline.getName(),
previousFailures);
}
}
});
- if (shouldUpdatePipeline.get()) {
+ boolean recoveredBeforeThisCheck = missingInstances.isEmpty()
+ && pipeline.getHealthStatus() == PipelineHealthStatus.FAILURE;
+ if (!missingInstances.isEmpty() || !recoveredInstances.isEmpty() ||
recoveredBeforeThisCheck) {
var currentPipeline =
resourceProvider.pipelineStorage().getElementById(pipeline.getPipelineId());
- if (!failedInstances.isEmpty()) {
+ if (!missingInstances.isEmpty()) {
currentPipeline.setHealthStatus(PipelineHealthStatus.FAILURE);
pipelinesStats.failedIncrease();
- } else if (!recoveredInstances.isEmpty()) {
+ } else {
currentPipeline.setHealthStatus(PipelineHealthStatus.OK);
pipelinesStats.attentionRequiredIncrease();
}
@@ -182,40 +181,73 @@ 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 Set<PipelineRecoveryBackoff.RecoveryKey> getActiveRecoveryKeys() {
+ return healthCheckData.activeResources().runningPipelines().stream()
+ .flatMap(pipeline -> Stream.concat(pipeline.getSepas().stream(),
pipeline.getActions().stream())
+ .map(pipelineElement -> new PipelineRecoveryBackoff.RecoveryKey(
+ pipeline.getPipelineId(),
+ HealthCheckUtils.extractInstanceId(pipelineElement)
+ )))
+ .collect(Collectors.toSet());
}
- private boolean shouldRetry(String instanceId) {
- if (!failedRestartAttempts.containsKey(instanceId)) {
- return true;
+ private void logFailedRecovery(Pipeline pipeline,
+ InvocableStreamPipesEntity pipelineElement,
+ PipelineRecoveryBackoff.RecoveryState state) {
+ var logMessage = "Could not restore pipeline element {} of pipeline {} on
attempt {}; "
+ + "next attempt is eligible in {} seconds";
+ var delaySeconds = state.delay().toSeconds();
+ if (state.failedAttempts() == 1) {
+ LOG.warn(logMessage, pipelineElement.getName(), pipeline.getName(),
state.failedAttempts(), delaySeconds);
+ } else if (isPowerOfTwo(state.failedAttempts())) {
+ LOG.info(logMessage, pipelineElement.getName(), pipeline.getName(),
state.failedAttempts(), delaySeconds);
} else {
- return failedRestartAttempts.get(instanceId) < MAX_FAILED_ATTEMPTS;
+ LOG.debug(logMessage, pipelineElement.getName(), pipeline.getName(),
state.failedAttempts(), delaySeconds);
}
}
- private void resetFailedAttempts(String instanceId) {
- failedRestartAttempts.put(instanceId, 0);
+ private void logSuccessfulRecovery(Pipeline pipeline,
+ InvocableStreamPipesEntity
pipelineElement,
+ int previousFailures) {
+ if (previousFailures == 0) {
+ LOG.info("Successfully restored pipeline element {} of pipeline {}",
+ pipelineElement.getName(), pipeline.getName());
+ } else {
+ LOG.info("Successfully restored pipeline element {} of pipeline {} after
{} failed attempts",
+ pipelineElement.getName(), pipeline.getName(), previousFailures);
+ }
}
- 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 isPowerOfTwo(int value) {
+ return (value & (value - 1)) == 0;
+ }
+
+ 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 int getElementsCount(List<Pipeline> allPipelines) {
- return allPipelines.stream().mapToInt(pipeline ->
pipeline.getActions().size()).sum();
+ private boolean isNowhereRunning(String instanceId) {
+ return (healthCheckData.activeExtensionInstances().entrySet().stream()
+ .noneMatch(entry ->
entry.getValue().runningPipelineElementInstanceIds().contains(instanceId)));
}
- private String getInvocationUrl(InvocableStreamPipesEntity pipelineElement,
- String baseUrl) {
- return ExtensionsServiceEndpointUtils
- .getPipelineElementType(pipelineElement)
- .getInvocationUrl(baseUrl, pipelineElement.getAppId());
+ private int getElementsCount(List<Pipeline> allPipelines) {
+ return allPipelines.stream().mapToInt(pipeline ->
pipeline.getActions().size()).sum();
}
}
diff --git
a/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/PipelineRecoveryBackoff.java
b/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/PipelineRecoveryBackoff.java
new file mode 100644
index 0000000000..b2181752fc
--- /dev/null
+++
b/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/PipelineRecoveryBackoff.java
@@ -0,0 +1,107 @@
+/*
+ * 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 java.time.Clock;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+final class PipelineRecoveryBackoff {
+
+ static final Duration DEFAULT_INITIAL_DELAY = Duration.ofSeconds(30);
+ static final Duration DEFAULT_MAX_DELAY = Duration.ofMinutes(10);
+
+ private final Clock clock;
+ private final Duration initialDelay;
+ private final Duration maxDelay;
+ private final Map<RecoveryKey, RecoveryState> recoveryStates = new
ConcurrentHashMap<>();
+
+ PipelineRecoveryBackoff() {
+ this(Clock.systemUTC(), DEFAULT_INITIAL_DELAY, DEFAULT_MAX_DELAY);
+ }
+
+ PipelineRecoveryBackoff(Duration healthCheckInterval) {
+ this(
+ Clock.systemUTC(),
+ healthCheckInterval,
+ max(DEFAULT_MAX_DELAY, healthCheckInterval)
+ );
+ }
+
+ PipelineRecoveryBackoff(Clock clock,
+ Duration initialDelay,
+ Duration maxDelay) {
+ this.clock = clock;
+ this.initialDelay = initialDelay;
+ this.maxDelay = maxDelay;
+ }
+
+ boolean isAttemptDue(String pipelineId,
+ String instanceId) {
+ var state = recoveryStates.get(new RecoveryKey(pipelineId, instanceId));
+ return state == null || !clock.instant().isBefore(state.nextAttemptAt());
+ }
+
+ RecoveryState recordFailure(String pipelineId,
+ String instanceId) {
+ var key = new RecoveryKey(pipelineId, instanceId);
+ return recoveryStates.compute(key, (ignored, previousState) -> {
+ int failedAttempts = previousState == null ? 1 :
previousState.failedAttempts() + 1;
+ Duration delay = calculateDelay(failedAttempts);
+ return new RecoveryState(failedAttempts, delay,
clock.instant().plus(delay));
+ });
+ }
+
+ int reset(String pipelineId,
+ String instanceId) {
+ var previousState = recoveryStates.remove(new RecoveryKey(pipelineId,
instanceId));
+ return previousState == null ? 0 : previousState.failedAttempts();
+ }
+
+ void retainOnly(Set<RecoveryKey> activeInstances) {
+ recoveryStates.keySet().retainAll(activeInstances);
+ }
+
+ RecoveryState getState(String pipelineId,
+ String instanceId) {
+ return recoveryStates.get(new RecoveryKey(pipelineId, instanceId));
+ }
+
+ private Duration calculateDelay(int failedAttempts) {
+ int exponent = Math.min(failedAttempts - 1, 30);
+ long multiplier = 1L << exponent;
+ Duration calculatedDelay = initialDelay.multipliedBy(multiplier);
+ return calculatedDelay.compareTo(maxDelay) > 0 ? maxDelay :
calculatedDelay;
+ }
+
+ private static Duration max(Duration first,
+ Duration second) {
+ return first.compareTo(second) >= 0 ? first : second;
+ }
+
+ record RecoveryKey(String pipelineId, String instanceId) {
+ }
+
+ record RecoveryState(int failedAttempts,
+ Duration delay,
+ Instant nextAttemptAt) {
+ }
+}
diff --git
a/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/utils/HealthCheckUtils.java
b/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/utils/HealthCheckUtils.java
index fdec9191bc..ee7e79043c 100644
---
a/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/utils/HealthCheckUtils.java
+++
b/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/utils/HealthCheckUtils.java
@@ -39,6 +39,12 @@ public class HealthCheckUtils {
+ "' was not available and could not be restored.");
}
+ public static void addPendingRecoveryNotification(List<String>
pipelineNotifications,
+ InvocableStreamPipesEntity
pipelineElement) {
+ pipelineNotifications.add(getCurrentDatetime() + "Pipeline element '" +
pipelineElement.getName()
+ + "' is not available. The next automatic recovery attempt is
pending.");
+ }
+
private static String getCurrentDatetime() {
DateTimeFormatter dtf = DateTimeFormatter.ofPattern("uuuu/MM/dd HH:mm:ss");
LocalDateTime now = LocalDateTime.now();
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..8c83031102
--- /dev/null
+++
b/streampipes-health-monitoring/src/test/java/org/apache/streampipes/health/monitoring/PipelineHealthCheckTest.java
@@ -0,0 +1,256 @@
+/*
+ * 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.health.ExtensionInstanceHealth;
+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.time.Clock;
+import java.time.Duration;
+import java.time.Instant;
+import java.time.ZoneId;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+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 clock = new MutableClock();
+ var recoveryBackoff = new PipelineRecoveryBackoff(
+ clock,
+ PipelineRecoveryBackoff.DEFAULT_INITIAL_DELAY,
+ PipelineRecoveryBackoff.DEFAULT_MAX_DELAY
+ );
+ var healthCheck = healthCheck(pipeline, Map.of("missing", false),
recoveryBackoff);
+
+ for (int i = 0; i < 11; i++) {
+ healthCheck.runCheck();
+ clock.advance(PipelineRecoveryBackoff.DEFAULT_MAX_DELAY);
+ }
+
+ assertEquals(11, healthCheck.restoreAttempts("missing"));
+ assertEquals(PipelineHealthStatus.FAILURE, pipeline.getHealthStatus());
+ }
+
+ @Test
+ public void waitsUntilBackoffExpiresBeforeRetrying() {
+ var pipeline = pipeline(processor("missing", "Missing processor"));
+ var clock = new MutableClock();
+ var recoveryBackoff = new PipelineRecoveryBackoff(
+ clock,
+ PipelineRecoveryBackoff.DEFAULT_INITIAL_DELAY,
+ PipelineRecoveryBackoff.DEFAULT_MAX_DELAY
+ );
+ var healthCheck = healthCheck(pipeline, Map.of("missing", false),
recoveryBackoff);
+
+ healthCheck.runCheck();
+ healthCheck.runCheck();
+
+ assertEquals(1, healthCheck.restoreAttempts("missing"));
+ assertEquals(PipelineHealthStatus.FAILURE, pipeline.getHealthStatus());
+ assertEquals(1, pipeline.getPipelineNotifications().size());
+ assertTrue(pipeline.getPipelineNotifications().get(0)
+ .contains("The next automatic recovery attempt is pending."));
+
+ clock.advance(PipelineRecoveryBackoff.DEFAULT_INITIAL_DELAY);
+ healthCheck.runCheck();
+
+ assertEquals(2, healthCheck.restoreAttempts("missing"));
+ }
+
+ @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());
+ }
+
+ @Test
+ public void observedRunningElementClearsFailureAndBackoffState() {
+ var pipeline = pipeline(processor("running", "Running processor"));
+ pipeline.setHealthStatus(PipelineHealthStatus.FAILURE);
+ var recoveryBackoff = new PipelineRecoveryBackoff();
+ recoveryBackoff.recordFailure(pipeline.getPipelineId(), "running");
+ var healthCheck = healthCheck(
+ pipeline,
+ Map.of(),
+ recoveryBackoff,
+ Set.of("running")
+ );
+
+ healthCheck.runCheck();
+
+ assertEquals(PipelineHealthStatus.OK, pipeline.getHealthStatus());
+ assertEquals(0, recoveryBackoff.reset(pipeline.getPipelineId(),
"running"));
+ assertEquals(0, healthCheck.restoreAttempts("running"));
+ }
+
+ private TestPipelineHealthCheck healthCheck(Pipeline pipeline,
+ Map<String, Boolean>
restorationResults) {
+ return healthCheck(pipeline, restorationResults, new
PipelineRecoveryBackoff());
+ }
+
+ private TestPipelineHealthCheck healthCheck(Pipeline pipeline,
+ Map<String, Boolean>
restorationResults,
+ PipelineRecoveryBackoff
recoveryBackoff) {
+ return healthCheck(pipeline, restorationResults, recoveryBackoff,
Set.of());
+ }
+
+ private TestPipelineHealthCheck healthCheck(Pipeline pipeline,
+ Map<String, Boolean>
restorationResults,
+ PipelineRecoveryBackoff
recoveryBackoff,
+ Set<String> runningInstanceIds) {
+ 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("service-id", new ExtensionInstanceHealth(Map.of(),
runningInstanceIds))
+ );
+
+ return new TestPipelineHealthCheck(
+ healthCheckData,
+ resourceProvider,
+ restorationResults,
+ recoveryBackoff
+ );
+ }
+
+ 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,
+ PipelineRecoveryBackoff recoveryBackoff) {
+ super(healthCheckData, null, resourceProvider, null, recoveryBackoff);
+ 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);
+ }
+ }
+
+ private static class MutableClock extends Clock {
+
+ private Instant instant = Instant.EPOCH;
+
+ @Override
+ public ZoneId getZone() {
+ return ZoneId.of("UTC");
+ }
+
+ @Override
+ public Clock withZone(ZoneId zone) {
+ return this;
+ }
+
+ @Override
+ public Instant instant() {
+ return instant;
+ }
+
+ void advance(Duration duration) {
+ instant = instant.plus(duration);
+ }
+ }
+}
diff --git
a/streampipes-health-monitoring/src/test/java/org/apache/streampipes/health/monitoring/PipelineRecoveryBackoffTest.java
b/streampipes-health-monitoring/src/test/java/org/apache/streampipes/health/monitoring/PipelineRecoveryBackoffTest.java
new file mode 100644
index 0000000000..8bbb32f2be
--- /dev/null
+++
b/streampipes-health-monitoring/src/test/java/org/apache/streampipes/health/monitoring/PipelineRecoveryBackoffTest.java
@@ -0,0 +1,100 @@
+/*
+ * 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.junit.jupiter.api.Test;
+
+import java.time.Clock;
+import java.time.Duration;
+import java.time.Instant;
+import java.time.ZoneOffset;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class PipelineRecoveryBackoffTest {
+
+ @Test
+ public void delayIncreasesExponentiallyAndIsCapped() {
+ var recoveryBackoff = recoveryBackoff();
+ var expectedDelays = new long[]{30, 60, 120, 240, 480, 600, 600};
+
+ for (long expectedDelay : expectedDelays) {
+ var state = recoveryBackoff.recordFailure("pipeline", "instance");
+ assertEquals(Duration.ofSeconds(expectedDelay), state.delay());
+ }
+ }
+
+ @Test
+ public void configuredHealthCheckIntervalIsUsedAsInitialDelay() {
+ var recoveryBackoff = new PipelineRecoveryBackoff(Duration.ofMinutes(2));
+ var expectedDelays = new long[]{2, 4, 8, 10, 10};
+
+ for (long expectedDelay : expectedDelays) {
+ var state = recoveryBackoff.recordFailure("pipeline", "instance");
+ assertEquals(Duration.ofMinutes(expectedDelay), state.delay());
+ }
+ }
+
+ @Test
+ public void healthCheckIntervalAboveDefaultMaximumBecomesMaximumDelay() {
+ var healthCheckInterval = Duration.ofMinutes(15);
+ var recoveryBackoff = new PipelineRecoveryBackoff(healthCheckInterval);
+
+ var firstFailure = recoveryBackoff.recordFailure("pipeline", "instance");
+ var secondFailure = recoveryBackoff.recordFailure("pipeline", "instance");
+
+ assertEquals(healthCheckInterval, firstFailure.delay());
+ assertEquals(healthCheckInterval, secondFailure.delay());
+ }
+
+ @Test
+ public void resetAllowsImmediateRecoveryAttempt() {
+ var recoveryBackoff = recoveryBackoff();
+ recoveryBackoff.recordFailure("pipeline", "instance");
+
+ assertFalse(recoveryBackoff.isAttemptDue("pipeline", "instance"));
+ assertEquals(1, recoveryBackoff.reset("pipeline", "instance"));
+ assertTrue(recoveryBackoff.isAttemptDue("pipeline", "instance"));
+ }
+
+ @Test
+ public void inactiveInstancesAreRemoved() {
+ var recoveryBackoff = recoveryBackoff();
+ recoveryBackoff.recordFailure("pipeline", "active");
+ recoveryBackoff.recordFailure("pipeline", "inactive");
+
+ recoveryBackoff.retainOnly(Set.of(
+ new PipelineRecoveryBackoff.RecoveryKey("pipeline", "active")
+ ));
+
+ assertEquals(1, recoveryBackoff.getState("pipeline",
"active").failedAttempts());
+ assertNull(recoveryBackoff.getState("pipeline", "inactive"));
+ }
+
+ private PipelineRecoveryBackoff recoveryBackoff() {
+ return new PipelineRecoveryBackoff(
+ Clock.fixed(Instant.EPOCH, ZoneOffset.UTC),
+ PipelineRecoveryBackoff.DEFAULT_INITIAL_DELAY,
+ PipelineRecoveryBackoff.DEFAULT_MAX_DELAY
+ );
+ }
+}
diff --git
a/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/PostStartupTask.java
b/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/PostStartupTask.java
index ccd7ee61a2..95d1211a9b 100644
---
a/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/PostStartupTask.java
+++
b/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/PostStartupTask.java
@@ -40,6 +40,7 @@ import
org.apache.streampipes.storage.management.StorageDispatcher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.time.Duration;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -68,7 +69,8 @@ public class PostStartupTask implements Runnable {
ExtensionServiceRequestManager
extensionServiceRequestManager,
WorkerRestClient workerRestClient,
SpResourceManager resourceManager,
- List<HealthCheck> registeredHealthChecks) {
+ List<HealthCheck> registeredHealthChecks,
+ Duration healthCheckInterval) {
this.pipelineStorage = pipelineStorage;
this.extensionServiceRequestManager = extensionServiceRequestManager;
this.executorService = Executors.newSingleThreadScheduledExecutor();
@@ -93,7 +95,8 @@ public class PostStartupTask implements Runnable {
StorageDispatcher.INSTANCE.getNoSqlStore().getExtensionsServiceStorage(),
extensionServiceRequestManager,
resourceManager,
- registeredHealthChecks
+ registeredHealthChecks,
+ healthCheckInterval
)
);
}
diff --git
a/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/StreamPipesCoreApplication.java
b/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/StreamPipesCoreApplication.java
index 42bd160b7c..eaf7fff5bc 100644
---
a/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/StreamPipesCoreApplication.java
+++
b/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/StreamPipesCoreApplication.java
@@ -82,6 +82,7 @@ import jakarta.annotation.PreDestroy;
import java.io.IOException;
import java.net.UnknownHostException;
+import java.time.Duration;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
@@ -196,16 +197,21 @@ public class StreamPipesCoreApplication extends
StreamPipesServiceBase {
new ApplyDefaultRolesAndPrivilegesTask(roleStorage,
privilegeStorage).execute();
coreStatusManager.updateCoreStatus(SpCoreConfigurationStatus.READY);
+ var healthCheckInterval = Duration.ofMillis(
+ env.getHealthCheckIntervalInMillis().getValueOrDefault()
+ );
+
executorService.schedule(new PostStartupTask(
getPipelineStorage(),
extensionServiceRequestManager,
workerRestClient,
resourceManager,
- getRegisteredExtensionHealthChecks()),
+ getRegisteredExtensionHealthChecks(),
+ healthCheckInterval),
env.getInitialHealthCheckDelayInMillis().getValueOrDefault(),
TimeUnit.MILLISECONDS);
-
scheduleHealthChecks(env.getHealthCheckIntervalInMillis().getValueOrDefault(),
List
+ scheduleHealthChecks(healthCheckInterval, List
.of(new ServiceHealthCheck(
extensionsServiceStorage,
extensionServiceRequestManager,
@@ -224,7 +230,8 @@ public class StreamPipesCoreApplication extends
StreamPipesServiceBase {
StorageDispatcher.INSTANCE.getNoSqlStore().getExtensionsServiceStorage(),
extensionServiceRequestManager,
resourceManager,
- getRegisteredExtensionHealthChecks()
+ getRegisteredExtensionHealthChecks(),
+ healthCheckInterval
)));
var logFetchInterval =
env.getLogFetchIntervalInMillis().getValueOrDefault();
@@ -236,13 +243,13 @@ public class StreamPipesCoreApplication extends
StreamPipesServiceBase {
TimeUnit.MILLISECONDS);
}
- private void scheduleHealthChecks(int healthCheckIntervalInMillis,
List<Runnable> checks) {
+ private void scheduleHealthChecks(Duration healthCheckInterval,
List<Runnable> checks) {
var healthCheckExecutorService =
Executors.newSingleThreadScheduledExecutor();
checks.forEach(check -> {
- LOG.info("Health check {} configured to run every {} seconds",
check.getClass().getSimpleName(),
- TimeUnit.MILLISECONDS.toSeconds(healthCheckIntervalInMillis));
- healthCheckExecutorService.scheduleAtFixedRate(check,
healthCheckIntervalInMillis,
- healthCheckIntervalInMillis,
+ LOG.info("Health check {} configured with a delay of {} seconds between
runs", check.getClass().getSimpleName(),
+ healthCheckInterval.toSeconds());
+ healthCheckExecutorService.scheduleWithFixedDelay(check,
healthCheckInterval.toMillis(),
+ healthCheckInterval.toMillis(),
TimeUnit.MILLISECONDS);
});
}
diff --git
a/ui/src/app/core-ui/pipeline/pipeline-operation-status/pipeline-operation-status.component.html
b/ui/src/app/core-ui/pipeline/pipeline-operation-status/pipeline-operation-status.component.html
index 88dde64932..59b4670f55 100644
---
a/ui/src/app/core-ui/pipeline/pipeline-operation-status/pipeline-operation-status.component.html
+++
b/ui/src/app/core-ui/pipeline/pipeline-operation-status/pipeline-operation-status.component.html
@@ -16,39 +16,66 @@
~
-->
-@for (msg of pipelineOperationStatus()?.elementStatus ?? []; track msg) {
- <div fxFlex="100" fxLayout="column" class="status-outer mt-10">
- <div fxFlex="100" fxLayout="column">
- <div
- fxFlex="100"
- fxLayout="row"
- fxLayoutAlign="start center"
- class="p-15"
- >
- @if (msg.success) {
- <mat-icon color="accent">done</mat-icon>
- }
- @if (!msg.success) {
- <mat-icon style="color: red">warning</mat-icon>
- }
- <div fxFlex="100" fxLayout="column" class="ml-5">
- <span
- ><b>{{ msg.elementName }}</b></span
- >
- <small>{{
- msg.elementId.substr(0, msg.elementId.lastIndexOf('/'))
- }}</small>
- </div>
+<div class="status-list" role="list">
+ @for (msg of elementStatuses(); track msg.elementId + $index) {
+ @if (!msg.success) {
+ <div class="status-entry status-error" role="listitem">
+ <sp-alert-banner
+ type="error"
+ [title]="msg.elementName || ('Error' | translate)"
+ >
+ <div class="status-meta">
+ <sp-label
+ tone="error"
+ size="small"
+ variant="soft"
+ shape="badge"
+ textCase="uppercase"
+ [labelText]="'Error' | translate"
+ ></sp-label>
+ @if (msg.elementId) {
+ <code class="status-id" [title]="msg.elementId">{{
+ getDisplayElementId(msg)
+ }}</code>
+ }
+ </div>
+ @if (msg.optionalMessage) {
+ <pre
+ class="error-detail"
+ [textContent]="msg.optionalMessage"
+ ></pre>
+ }
+ </sp-alert-banner>
</div>
- <div>
- @if (msg.optionalMessage) {
- <div fxFlex="100" fxLayout="column" class="mt-10">
- <div class="error-message">
- <div class="p-10">{{ msg.optionalMessage }}</div>
- </div>
+ } @else {
+ <div class="status-entry status-success" role="listitem">
+ <mat-icon class="status-icon" aria-hidden="true"
+ >check_circle</mat-icon
+ >
+ <div class="status-content">
+ <div class="status-heading">
+ <span class="status-name">{{ msg.elementName }}</span>
+ <sp-label
+ tone="success"
+ size="small"
+ variant="soft"
+ shape="badge"
+ textCase="uppercase"
+ [labelText]="'Success' | translate"
+ ></sp-label>
</div>
- }
+ @if (msg.elementId) {
+ <code class="status-id" [title]="msg.elementId">{{
+ getDisplayElementId(msg)
+ }}</code>
+ }
+ @if (msg.optionalMessage) {
+ <div class="success-detail">
+ {{ msg.optionalMessage }}
+ </div>
+ }
+ </div>
</div>
- </div>
- </div>
-}
+ }
+ }
+</div>
diff --git
a/ui/src/app/core-ui/pipeline/pipeline-operation-status/pipeline-operation-status.component.scss
b/ui/src/app/core-ui/pipeline/pipeline-operation-status/pipeline-operation-status.component.scss
index 16a5951fa8..db67c36927 100644
---
a/ui/src/app/core-ui/pipeline/pipeline-operation-status/pipeline-operation-status.component.scss
+++
b/ui/src/app/core-ui/pipeline/pipeline-operation-status/pipeline-operation-status.component.scss
@@ -16,6 +16,121 @@
*
*/
-.status-outer {
- border: 1px solid var(--color-bg-3);
+.status-list {
+ display: flex;
+ flex-direction: column;
+ gap: var(--space-sm);
+ width: 100%;
+ margin-top: var(--space-md);
+}
+
+.status-entry {
+ min-width: 0;
+ animation: status-entry-in var(--motion-duration-standard)
+ var(--motion-easing-standard) both;
+}
+
+.status-error {
+ display: block;
+}
+
+.status-error sp-alert-banner {
+ display: block;
+}
+
+.status-meta,
+.status-heading {
+ display: flex;
+ align-items: center;
+ gap: var(--space-sm);
+ min-width: 0;
+}
+
+.status-meta {
+ margin-top: var(--space-2xs);
+}
+
+.status-id {
+ min-width: 0;
+ color: var(--fg-muted);
+ font-size: var(--font-size-xs);
+ line-height: var(--line-height-normal);
+ overflow-wrap: anywhere;
+}
+
+.error-detail {
+ max-height: 14rem;
+ margin: var(--space-sm) 0 0;
+ padding: var(--space-sm) var(--space-md);
+ overflow: auto;
+ border-radius: var(--radius-xs);
+ background: color-mix(in srgb, var(--color-bg-0) 72%, transparent);
+ color: var(--mat-sys-on-surface);
+ font-size: var(--font-size-xs);
+ line-height: var(--line-height-relaxed);
+ white-space: pre-wrap;
+ overflow-wrap: anywhere;
+}
+
+.status-success {
+ display: flex;
+ align-items: flex-start;
+ gap: var(--space-sm);
+ padding: var(--space-sm) var(--space-md);
+ border-bottom: 1px solid var(--color-border-subtle);
+ transition: background-color var(--motion-duration-fast)
+ var(--motion-easing-standard);
+}
+
+.status-success:hover {
+ background-color: var(--color-surface-interactive-hover);
+}
+
+.status-icon {
+ flex: 0 0 auto;
+ color: var(--color-success);
+}
+
+.status-content {
+ flex: 1 1 auto;
+ min-width: 0;
+}
+
+.status-heading {
+ justify-content: space-between;
+}
+
+.status-name {
+ min-width: 0;
+ font-weight: var(--font-weight-medium);
+ overflow-wrap: anywhere;
+}
+
+.success-detail {
+ margin-top: var(--space-2xs);
+ color: var(--fg-muted);
+ font-size: var(--font-size-sm);
+ line-height: var(--line-height-normal);
+ overflow-wrap: anywhere;
+}
+
+@keyframes status-entry-in {
+ from {
+ opacity: 0;
+ transform: translateY(var(--space-xs));
+ }
+ to {
+ opacity: 1;
+ transform: translateY(0);
+ }
+}
+
+@media (prefers-reduced-motion: reduce) {
+ .status-entry {
+ animation: none;
+ }
+
+ .status-success {
+ transition: none;
+ }
}
diff --git
a/ui/src/app/core-ui/pipeline/pipeline-operation-status/pipeline-operation-status.component.ts
b/ui/src/app/core-ui/pipeline/pipeline-operation-status/pipeline-operation-status.component.ts
index cbfd29fe76..666ef51c17 100644
---
a/ui/src/app/core-ui/pipeline/pipeline-operation-status/pipeline-operation-status.component.ts
+++
b/ui/src/app/core-ui/pipeline/pipeline-operation-status/pipeline-operation-status.component.ts
@@ -16,23 +16,40 @@
*
*/
-import { Component, input } from '@angular/core';
-import { PipelineOperationStatus } from '@streampipes/platform-services';
+import { Component, computed, input } from '@angular/core';
import {
- FlexDirective,
- LayoutAlignDirective,
- LayoutDirective,
-} from '@ngbracket/ngx-layout/flex';
+ PipelineElementStatus,
+ PipelineOperationStatus,
+} from '@streampipes/platform-services';
import { MatIcon } from '@angular/material/icon';
+import {
+ SpAlertBannerComponent,
+ SpLabelComponent,
+} from '@streampipes/shared-ui';
+import { TranslatePipe } from '@ngx-translate/core';
@Component({
selector: 'sp-pipeline-operation-status',
templateUrl: './pipeline-operation-status.component.html',
styleUrls: ['./pipeline-operation-status.component.scss'],
- imports: [FlexDirective, LayoutDirective, LayoutAlignDirective, MatIcon],
+ imports: [MatIcon, SpAlertBannerComponent, SpLabelComponent,
TranslatePipe],
})
export class PipelineOperationStatusComponent {
readonly pipelineOperationStatus = input<
PipelineOperationStatus | undefined
>(undefined);
+
+ readonly elementStatuses = computed(() =>
+ [...(this.pipelineOperationStatus()?.elementStatus ?? [])].sort(
+ (first, second) => Number(first.success) - Number(second.success),
+ ),
+ );
+
+ getDisplayElementId(status: PipelineElementStatus): string {
+ const elementId = status.elementId ?? '';
+ const lastSeparatorIndex = elementId.lastIndexOf('/');
+ return lastSeparatorIndex > 0
+ ? elementId.substring(0, lastSeparatorIndex)
+ : elementId;
+ }
}
diff --git a/ui/src/scss/sp/_variables.scss b/ui/src/scss/sp/_variables.scss
index 100b4a7888..33b9ad2ed0 100644
--- a/ui/src/scss/sp/_variables.scss
+++ b/ui/src/scss/sp/_variables.scss
@@ -54,6 +54,7 @@
--color-bg-3: var(--mat-sys-surface-container-highest);
--color-bg-4: var(--mat-sys-surface-container-highest);
--color-code-bg: #282a36;
+ --color-code-fg: #f8f8f2;
// for alert banners and labels
--color-info: #2563eb;
diff --git a/ui/src/scss/sp/main.scss b/ui/src/scss/sp/main.scss
index e5d0bb9f30..23c690d9cc 100644
--- a/ui/src/scss/sp/main.scss
+++ b/ui/src/scss/sp/main.scss
@@ -715,7 +715,7 @@ label {
font:
0.7rem Inconsolata,
monospace;
- color: var(--mat-sys-on-surface);
+ color: var(--color-code-fg);
width: 100%;
max-width: 100%;
}
