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
The following commit(s) were added to
refs/heads/remove-max-retries-health-check by this push:
new 4105d0a0cc Add exponential backoff strategy
4105d0a0cc is described below
commit 4105d0a0cc53951fa1a39965e2340cf576c6970b
Author: Dominik Riemer <[email protected]>
AuthorDate: Mon Jul 20 18:36:12 2026 +0200
Add exponential backoff strategy
---
.../pe/InvocablePipelineElementManagement.java | 2 +-
.../health/monitoring/ExtensionHealthCheck.java | 10 +-
.../health/monitoring/PipelineHealthCheck.java | 92 +++++++++++++---
.../health/monitoring/PipelineRecoveryBackoff.java | 94 ++++++++++++++++
.../health/monitoring/PipelineHealthCheckTest.java | 109 ++++++++++++++++++-
.../monitoring/PipelineRecoveryBackoffTest.java | 77 +++++++++++++
.../pipeline-operation-status.component.html | 93 ++++++++++------
.../pipeline-operation-status.component.scss | 119 ++++++++++++++++++++-
.../pipeline-operation-status.component.ts | 31 ++++--
ui/src/scss/sp/main.scss | 2 +-
10 files changed, 566 insertions(+), 63 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..5b92be9ff5 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
@@ -40,6 +40,7 @@ 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,
@@ -51,6 +52,7 @@ public class ExtensionHealthCheck implements Runnable {
this.extensionRequestManager = extensionRequestManager;
this.resourceManager = resourceManager;
this.registeredHealthChecks = registeredHealthChecks;
+ this.pipelineRecoveryBackoff = new PipelineRecoveryBackoff();
}
@Override
@@ -100,7 +102,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 5d6b5263a4..1158e962af 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
@@ -43,6 +43,8 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
+import java.util.Set;
+import java.util.stream.Collectors;
import java.util.stream.Stream;
public class PipelineHealthCheck implements HealthCheck {
@@ -54,15 +56,25 @@ public class PipelineHealthCheck implements HealthCheck {
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
@@ -102,8 +114,9 @@ public class PipelineHealthCheck implements HealthCheck {
}
private void checkAndRestorePipelineElements() {
+ recoveryBackoff.retainOnly(getActiveRecoveryKeys());
healthCheckData.activeResources().runningPipelines().forEach(pipeline -> {
- 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(
@@ -114,25 +127,37 @@ public class PipelineHealthCheck implements HealthCheck {
runningPipelineElements.forEach(pipelineElement -> {
String instanceId =
HealthCheckUtils.extractInstanceId(pipelineElement);
if (isNowhereRunning(instanceId)) {
- 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 {
+ missingInstances.add(instanceId);
+ if (recoveryBackoff.isAttemptDue(pipeline.getPipelineId(),
instanceId)) {
+ boolean success = restorePipelineElement(pipelineElement,
pipeline.getPipelineId());
+ if (!success) {
+ var state =
recoveryBackoff.recordFailure(pipeline.getPipelineId(), instanceId);
+
HealthCheckUtils.addFailedAttemptNotification(pipelineNotifications,
pipelineElement);
+ logFailedRecovery(pipeline, pipelineElement, state);
+ } else {
+ missingInstances.remove(instanceId);
+ recoveredInstances.add(instanceId);
+ int previousFailures =
recoveryBackoff.reset(pipeline.getPipelineId(), instanceId);
+ logSuccessfulRecovery(pipeline, pipelineElement,
previousFailures);
+ }
+ }
+ } else {
+ int previousFailures =
recoveryBackoff.reset(pipeline.getPipelineId(), instanceId);
+ if (previousFailures > 0) {
recoveredInstances.add(instanceId);
- LOG.info("Successfully restored pipeline element {} of pipeline
{}",
- pipelineElement.getName(), pipeline.getName());
+ LOG.info("Pipeline element {} of pipeline {} is running again
after {} failed recovery attempts",
+ pipelineElement.getName(), pipeline.getName(),
previousFailures);
}
}
});
- if (!failedInstances.isEmpty() || !recoveredInstances.isEmpty()) {
+ 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();
}
@@ -154,6 +179,47 @@ public class PipelineHealthCheck implements HealthCheck {
pipelinesStats.setElementCount(getElementsCount(healthCheckData.activeResources().allPipelines()));
}
+ 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 void logFailedRecovery(Pipeline pipeline,
+ InvocableStreamPipesEntity pipelineElement,
+ PipelineRecoveryBackoff.RecoveryState state) {
+ var logMessage = "Could not restore pipeline element {} of pipeline {} on
attempt {}; "
+ + "next attempt 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 {
+ LOG.debug(logMessage, pipelineElement.getName(), pipeline.getName(),
state.failedAttempts(), delaySeconds);
+ }
+ }
+
+ 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 boolean isPowerOfTwo(int value) {
+ return (value & (value - 1)) == 0;
+ }
+
protected boolean restorePipelineElement(InvocableStreamPipesEntity
pipelineElement,
String pipelineId) {
try {
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..82a9bcbeee
--- /dev/null
+++
b/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/PipelineRecoveryBackoff.java
@@ -0,0 +1,94 @@
+/*
+ * 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(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;
+ }
+
+ record RecoveryKey(String pipelineId, String instanceId) {
+ }
+
+ record RecoveryState(int failedAttempts,
+ Duration delay,
+ Instant nextAttemptAt) {
+ }
+}
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
index 6f5dc89493..de5278042b 100644
---
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
@@ -22,15 +22,21 @@ 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.mockito.ArgumentMatchers.any;
@@ -42,16 +48,46 @@ public class PipelineHealthCheckTest {
@Test
public void retriesRestorationBeyondPreviousLimit() {
var pipeline = pipeline(processor("missing", "Missing processor"));
- var healthCheck = healthCheck(pipeline, Map.of("missing", false));
+ 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());
+
+ clock.advance(PipelineRecoveryBackoff.DEFAULT_INITIAL_DELAY);
+ healthCheck.runCheck();
+
+ assertEquals(2, healthCheck.restoreAttempts("missing"));
+ }
+
@Test
public void recoveredElementDoesNotMaskFailedElement() {
var pipeline = pipeline(
@@ -85,8 +121,41 @@ public class PipelineHealthCheckTest {
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));
@@ -102,10 +171,15 @@ public class PipelineHealthCheckTest {
resourceProvider,
activeResources,
Map.of(),
- Map.of()
+ Map.of("service-id", new ExtensionInstanceHealth(Map.of(),
runningInstanceIds))
);
- return new TestPipelineHealthCheck(healthCheckData, resourceProvider,
restorationResults);
+ return new TestPipelineHealthCheck(
+ healthCheckData,
+ resourceProvider,
+ restorationResults,
+ recoveryBackoff
+ );
}
private Pipeline pipeline(DataProcessorInvocation... processors) {
@@ -133,8 +207,9 @@ public class PipelineHealthCheckTest {
TestPipelineHealthCheck(HealthCheckData healthCheckData,
ResourceProvider resourceProvider,
- Map<String, Boolean> restorationResults) {
- super(healthCheckData, null, resourceProvider, null);
+ Map<String, Boolean> restorationResults,
+ PipelineRecoveryBackoff recoveryBackoff) {
+ super(healthCheckData, null, resourceProvider, null, recoveryBackoff);
this.restorationResults = restorationResults;
}
@@ -150,4 +225,28 @@ public class PipelineHealthCheckTest {
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..802d21691e
--- /dev/null
+++
b/streampipes-health-monitoring/src/test/java/org/apache/streampipes/health/monitoring/PipelineRecoveryBackoffTest.java
@@ -0,0 +1,77 @@
+/*
+ * 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 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/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/main.scss b/ui/src/scss/sp/main.scss
index e5d0bb9f30..9aabe57d2d 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-bg-0);
width: 100%;
max-width: 100%;
}
