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 e88e420a66 feat: Make startup recovery more robust (#5168)
e88e420a66 is described below
commit e88e420a6662b84e34c31f2b24b65604095d4b31
Author: Dominik Riemer <[email protected]>
AuthorDate: Sun Nov 1 22:23:04 2026 +0100
feat: Make startup recovery more robust (#5168)
---
docker-compose.yml | 2 +
installer/compose/docker-compose.kafka.yml | 2 +
installer/compose/docker-compose.minimal.yml | 2 +
installer/compose/docker-compose.yml | 2 +
.../streampipes/client/http/HttpRequest.java | 9 +-
.../apache/streampipes/commons/constants/Envs.java | 2 +
.../commons/environment/Environment.java | 10 +
.../management/init/RunningInstances.java | 10 +
.../function/StreamPipesFunctionHandler.java | 22 +-
.../health/monitoring/AdapterHealthCheck.java | 18 ++
.../health/monitoring/ExtensionHealthCheck.java | 55 ++++-
.../ExtensionInstanceAvailabilityCheck.java | 66 ++++--
.../health/monitoring/PipelineHealthCheck.java | 20 +-
.../health/monitoring/PipelineRecoveryBackoff.java | 71 ++++---
...ata.java => ServiceAvailabilityGeneration.java} | 27 ++-
.../health/monitoring/ServiceHealthCheck.java | 4 +
.../monitoring/ServiceRegistrationManager.java | 8 +
.../HealthCheckData.java => ShutdownSignal.java} | 31 ++-
.../monitoring/StaleRegistrationCleanup.java | 105 ++++++++++
.../health/monitoring/model/HealthCheckData.java | 16 +-
.../health/monitoring/utils/HealthCheckUtils.java | 7 +
.../health/monitoring/AdapterHealthCheckTest.java | 45 +++++
.../monitoring/ExtensionHealthCheckTest.java | 129 ++++++++++++
.../ExtensionInstanceAvailabilityCheckTest.java | 123 ++++++++++++
.../health/monitoring/PipelineHealthCheckTest.java | 44 +++-
.../monitoring/PipelineRecoveryBackoffTest.java | 24 ++-
.../monitoring/ServiceRegistrationManagerTest.java | 74 +++++++
.../monitoring/StaleRegistrationCleanupTest.java | 128 ++++++++++++
.../impl/admin/ServiceRegistrationResource.java | 6 +
.../streampipes/service/core/PostStartupTask.java | 66 ++++--
.../service/core/StreamPipesCoreApplication.java | 146 ++++++++++----
.../extensions/ExtensionServiceShutdownHook.java | 22 +-
.../ExtensionsServiceShutdownHandler.java | 173 +++++++++++++++-
.../StreamPipesExtensionsServiceBase.java | 30 ++-
.../ExtensionsServiceShutdownHandlerTest.java | 221 +++++++++++++++++++++
35 files changed, 1580 insertions(+), 140 deletions(-)
diff --git a/docker-compose.yml b/docker-compose.yml
index 0ada05cfe9..d8aec467e5 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -34,6 +34,7 @@ services:
context: ./streampipes-service-core
dockerfile: Dockerfile
image: streampipes_backend:release-validation
+ stop_grace_period: 60s
environment:
- SP_INITIAL_SERVICE_USER_SECRET=${SP_SERVICE_SECRET:?Run the Compose
configure helper first}
- SP_CLIENT_SECRET=${SP_SERVICE_SECRET:?Run the Compose configure helper
first}
@@ -75,6 +76,7 @@ services:
context: ./streampipes-extensions/streampipes-extensions-all-iiot
dockerfile: Dockerfile
image: streampipes_pipeline-elements-all-iiot:release-validation
+ stop_grace_period: 60s
environment:
- SP_CLIENT_SECRET=${SP_SERVICE_SECRET:?Run the Compose configure helper
first}
logging: *default-logging
diff --git a/installer/compose/docker-compose.kafka.yml
b/installer/compose/docker-compose.kafka.yml
index 446ead976e..8cb6bfc0dd 100644
--- a/installer/compose/docker-compose.kafka.yml
+++ b/installer/compose/docker-compose.kafka.yml
@@ -41,6 +41,7 @@ services:
- backend:/root/.streampipes
logging: *default-logging
restart: unless-stopped
+ stop_grace_period: 60s
networks:
spnet:
@@ -128,6 +129,7 @@ services:
- SP_ENCRYPTION_PASSCODE=${SP_ENCRYPTION_PASSCODE:?Run the Compose
configure helper first}
logging: *default-logging
restart: unless-stopped
+ stop_grace_period: 60s
networks:
spnet:
diff --git a/installer/compose/docker-compose.minimal.yml
b/installer/compose/docker-compose.minimal.yml
index 4ae3260fbe..89d106f822 100644
--- a/installer/compose/docker-compose.minimal.yml
+++ b/installer/compose/docker-compose.minimal.yml
@@ -41,6 +41,7 @@ services:
- backend:/root/.streampipes
logging: *default-logging
restart: unless-stopped
+ stop_grace_period: 60s
networks:
spnet:
@@ -107,6 +108,7 @@ services:
- SP_ENCRYPTION_PASSCODE=${SP_ENCRYPTION_PASSCODE:?Run the Compose
configure helper first}
logging: *default-logging
restart: unless-stopped
+ stop_grace_period: 60s
networks:
spnet:
diff --git a/installer/compose/docker-compose.yml
b/installer/compose/docker-compose.yml
index 3bcd728f85..12d68a2b30 100644
--- a/installer/compose/docker-compose.yml
+++ b/installer/compose/docker-compose.yml
@@ -41,6 +41,7 @@ services:
- backend:/root/.streampipes
logging: *default-logging
restart: unless-stopped
+ stop_grace_period: 60s
networks:
spnet:
@@ -108,6 +109,7 @@ services:
- SP_ENCRYPTION_PASSCODE=${SP_ENCRYPTION_PASSCODE:?Run the Compose
configure helper first}
logging: *default-logging
restart: unless-stopped
+ stop_grace_period: 60s
networks:
spnet:
diff --git
a/streampipes-client/src/main/java/org/apache/streampipes/client/http/HttpRequest.java
b/streampipes-client/src/main/java/org/apache/streampipes/client/http/HttpRequest.java
index 69e324c5b6..1ebdcce84f 100644
---
a/streampipes-client/src/main/java/org/apache/streampipes/client/http/HttpRequest.java
+++
b/streampipes-client/src/main/java/org/apache/streampipes/client/http/HttpRequest.java
@@ -49,6 +49,13 @@ import java.util.List;
public abstract class HttpRequest<K, V, T> {
+ /**
+ * Bounds the connection attempt so that a core whose address no longer
refuses connections
+ * cannot block the caller indefinitely, for example during an extension
service's shutdown.
+ * Responses are not bounded, since some requests legitimately take long.
+ */
+ private static final int CONNECT_TIMEOUT_MS = 10_000;
+
private final StreamPipesClientConfig clientConfig;
private final ClientConnectionUrlResolver connectionConfig;
private final StreamPipesApiPath apiPath;
@@ -98,7 +105,7 @@ public abstract class HttpRequest<K, V, T> {
}
public T executeRequest() {
- Request request = makeRequest(serializer);
+ Request request =
makeRequest(serializer).connectTimeout(CONNECT_TIMEOUT_MS);
try {
HttpResponse response = request.execute().returnResponse();
StatusLine status = response.getStatusLine();
diff --git
a/streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/Envs.java
b/streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/Envs.java
index 491a2146f5..fa214d0cda 100644
---
a/streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/Envs.java
+++
b/streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/Envs.java
@@ -103,6 +103,8 @@ public enum Envs {
SP_INITIAL_WAIT_BEFORE_INSTALLATION_MS("SP_INITIAL_WAIT_BEFORE_INSTALLATION_MS",
"5000"),
+ SP_SHUTDOWN_WAIT_FOR_EXTENSIONS_MS("SP_SHUTDOWN_WAIT_FOR_EXTENSIONS_MS",
"5000"),
+
// Broker defaults
SP_KAFKA_HOST("SP_KAFKA_HOST", "kafka"),
diff --git
a/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/Environment.java
b/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/Environment.java
index a49129b62d..4fc1b5ead0 100644
---
a/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/Environment.java
+++
b/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/Environment.java
@@ -150,6 +150,16 @@ public interface Environment {
IntEnvironmentVariable getInitialWaitTimeBeforeInstallationInMillis();
+ /**
+ * How long the core keeps serving requests during its shutdown so that
extension
+ * services stopping at the same time can persist their function state. The
wait ends
+ * early once every extension service has deregistered. Keep it below the
container
+ * stop grace period.
+ */
+ default IntEnvironmentVariable getShutdownWaitForExtensionsInMillis() {
+ return new IntEnvironmentVariable(Envs.SP_SHUTDOWN_WAIT_FOR_EXTENSIONS_MS);
+ }
+
// Broker defaults
StringEnvironmentVariable getKafkaHost();
diff --git
a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/init/RunningInstances.java
b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/init/RunningInstances.java
index 712d1de23e..ae4e85e59f 100644
---
a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/init/RunningInstances.java
+++
b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/init/RunningInstances.java
@@ -67,6 +67,16 @@ public enum RunningInstances {
return runningInstances.size();
}
+ /**
+ * Returns a snapshot of the keys under which runtimes are registered. These
are the
+ * ids accepted by {@link #getInvocation(String)} and {@link
#remove(String)}, as
+ * opposed to {@link #getRunningInstanceIds()} which returns the element ids
of the
+ * descriptions.
+ */
+ public Set<String> getRunningInstanceKeys() {
+ return Set.copyOf(runningInstances.keySet());
+ }
+
public Set<String> getRunningInstanceIds() {
return runningInstances
.entrySet()
diff --git
a/streampipes-extensions-management/src/main/java/org/apache/streampipes/service/extensions/function/StreamPipesFunctionHandler.java
b/streampipes-extensions-management/src/main/java/org/apache/streampipes/service/extensions/function/StreamPipesFunctionHandler.java
index 8dd336d00e..4a0745363c 100644
---
a/streampipes-extensions-management/src/main/java/org/apache/streampipes/service/extensions/function/StreamPipesFunctionHandler.java
+++
b/streampipes-extensions-management/src/main/java/org/apache/streampipes/service/extensions/function/StreamPipesFunctionHandler.java
@@ -28,6 +28,9 @@ import
org.apache.streampipes.model.function.FunctionShutdownResult;
import org.apache.streampipes.model.function.FunctionsShutdownResponse;
import org.apache.streampipes.sdk.builder.stream.DataStreamConfiguration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
@@ -39,6 +42,8 @@ public enum StreamPipesFunctionHandler {
INSTANCE;
+ private static final Logger LOG =
LoggerFactory.getLogger(StreamPipesFunctionHandler.class);
+
private final Map<String, IStreamPipesFunctionDeclarer> runningInstances;
StreamPipesFunctionHandler() {
@@ -95,11 +100,24 @@ public enum StreamPipesFunctionHandler {
});
}
+ /**
+ * Discards every function and collects its registered state. A function
whose stop hook
+ * fails is logged and skipped; the other functions are still discarded and
their state is
+ * still returned, since this is the only path that persists it on shutdown.
+ */
public FunctionsShutdownResponse shutdownFunctionsAndGetState() {
var shutdownResults = new ArrayList<FunctionShutdownResult>();
this.runningInstances.forEach((key, value) -> {
- value.discardRuntime();
- shutdownResults.add(new FunctionShutdownResult(key,
value.getRegisteredStatePayload().orElse(null)));
+ try {
+ value.discardRuntime();
+ } catch (RuntimeException e) {
+ LOG.warn("Function {} threw an exception while stopping, its state is
persisted anyway", key, e);
+ }
+ try {
+ shutdownResults.add(new FunctionShutdownResult(key,
value.getRegisteredStatePayload().orElse(null)));
+ } catch (RuntimeException e) {
+ LOG.warn("Could not read the state of function {}", key, e);
+ }
});
this.runningInstances.clear();
diff --git
a/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/AdapterHealthCheck.java
b/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/AdapterHealthCheck.java
index aedb4b34be..fba77973c4 100644
---
a/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/AdapterHealthCheck.java
+++
b/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/AdapterHealthCheck.java
@@ -173,6 +173,7 @@ public class AdapterHealthCheck implements HealthCheck {
.stream()
.filter(Objects::nonNull)
.filter(a -> a.getElementId() != null)
+ .filter(a -> !isOnUnreachableService(a))
.filter(a -> !adapterInstanceStates.containsKey(a.getElementId()))
.toList();
}
@@ -182,8 +183,25 @@ public class AdapterHealthCheck implements HealthCheck {
return adapterInstanceStates == null ? Map.of() : adapterInstanceStates;
}
+ /**
+ * The adapter's service is registered but did not answer. The adapter may
still run
+ * there, so recovering it on another service could start a duplicate.
+ */
+ private boolean isOnUnreachableService(AdapterDescription adapter) {
+ var serviceId = adapter.getSelectedServiceId();
+ var unreachable = serviceId != null &&
healthCheckData.unreachableServiceIds().contains(serviceId);
+ if (unreachable) {
+ LOG.debug("Adapter {} runs on unreachable service {}, deferring
recovery", adapter.getElementId(), serviceId);
+ }
+ return unreachable;
+ }
+
public void recoverAdapters(List<AdapterDescription> adaptersToRecover) {
for (AdapterDescription adapterDescription : adaptersToRecover) {
+ if (ShutdownSignal.INSTANCE.isShutdownRequested()) {
+ LOG.debug("Shutdown in progress, skipping adapter recovery");
+ return;
+ }
// Invoke all adapters that were running when the adapter container was
stopped
try {
if (adapterDescription.isRunning()) {
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 02a8724f2c..28b8a592f0 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
@@ -18,6 +18,7 @@
package org.apache.streampipes.health.monitoring;
+import org.apache.streampipes.commons.environment.Environments;
import org.apache.streampipes.health.monitoring.model.HealthCheckData;
import
org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestManager;
import org.apache.streampipes.model.health.ExtensionInstanceHealth;
@@ -30,7 +31,11 @@ import org.slf4j.LoggerFactory;
import java.time.Duration;
import java.util.ArrayList;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
public class ExtensionHealthCheck implements Runnable {
@@ -42,6 +47,7 @@ public class ExtensionHealthCheck implements Runnable {
private final SpResourceManager resourceManager;
private final List<HealthCheck> registeredHealthChecks;
private final PipelineRecoveryBackoff pipelineRecoveryBackoff;
+ private final Map<String, Integer> consecutiveUnknownTicks = new HashMap<>();
public ExtensionHealthCheck(ResourceProvider resourceProvider,
IExtensionsServiceStorage
extensionsServiceStorage,
@@ -74,24 +80,42 @@ public class ExtensionHealthCheck implements Runnable {
@Override
public void run() {
+ if (ShutdownSignal.INSTANCE.isShutdownRequested()) {
+ LOG.debug("Shutdown in progress, skipping extension health check");
+ return;
+ }
try {
var activeResources = resourceProvider.loadActiveResources();
var activeCoreInstances =
resourceProvider.loadActiveInstances(activeResources);
var activeExtensionInstances = new HashMap<String,
ExtensionInstanceHealth>();
+ var unreachableServiceIds = new HashSet<String>();
+ consecutiveUnknownTicks.keySet().retainAll(activeCoreInstances.keySet());
activeCoreInstances.keySet().forEach(serviceId -> {
- activeExtensionInstances.put(
- serviceId,
- new ExtensionInstanceAvailabilityCheck(
- extensionsServiceStorage, serviceId, extensionRequestManager,
resourceManager
- ).checkRunningInstances());
+ var health = checkInstances(serviceId);
+ if (health.isPresent()) {
+ consecutiveUnknownTicks.remove(serviceId);
+ activeExtensionInstances.put(serviceId, health.get());
+ } else {
+ int unknownTicks = consecutiveUnknownTicks.merge(serviceId, 1,
Integer::sum);
+ if (unknownTicks >= maxUnknownTicks()) {
+ // The service answers its health probe but not the instance
request, or the
+ // service health check lags behind. Do not defer recovery forever.
+ LOG.warn("Extension service {} did not answer {} instance health
requests in a row, "
+ + "treating its instances as not running", serviceId,
unknownTicks);
+ activeExtensionInstances.put(serviceId, new
ExtensionInstanceHealth(Map.of(), Set.of()));
+ } else {
+ unreachableServiceIds.add(serviceId);
+ }
+ }
});
var healthCheckData = new HealthCheckData(
resourceProvider,
activeResources,
activeCoreInstances,
- activeExtensionInstances
+ activeExtensionInstances,
+ unreachableServiceIds
);
var healthChecks = getHealthChecks(healthCheckData);
runHealthChecks(healthChecks);
@@ -100,8 +124,27 @@ public class ExtensionHealthCheck implements Runnable {
}
}
+ protected Optional<ExtensionInstanceHealth> checkInstances(String serviceId)
{
+ return new ExtensionInstanceAvailabilityCheck(
+ extensionsServiceStorage, serviceId, extensionRequestManager,
resourceManager
+ ).checkReachableRunningInstances();
+ }
+
+ /**
+ * How many consecutive ticks a registered service may leave its instance
state unknown
+ * before it is treated as not running anything. Aligned with the number of
failed probes
+ * after which the service health check declares a service unhealthy.
+ */
+ protected int maxUnknownTicks() {
+ return Math.max(1,
Environments.getEnvironment().getServiceHealthFailureThreshold().getValueOrDefault());
+ }
+
private void runHealthChecks(List<HealthCheck> healthChecks) {
healthChecks.forEach(healthCheck -> {
+ if (ShutdownSignal.INSTANCE.isShutdownRequested()) {
+ LOG.debug("Shutdown in progress, skipping health check {}",
healthCheck.getClass().getSimpleName());
+ return;
+ }
try {
healthCheck.runCheck();
} catch (Exception e) {
diff --git
a/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/ExtensionInstanceAvailabilityCheck.java
b/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/ExtensionInstanceAvailabilityCheck.java
index 69124efd72..c446b19d6a 100644
---
a/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/ExtensionInstanceAvailabilityCheck.java
+++
b/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/ExtensionInstanceAvailabilityCheck.java
@@ -22,6 +22,7 @@ import
org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestTarg
import
org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestTargets;
import org.apache.streampipes.manager.api.extensions.ExtensionServiceRequests;
import
org.apache.streampipes.model.extensions.svcdiscovery.SpServiceRegistration;
+import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceStatus;
import org.apache.streampipes.model.health.ExtensionInstanceHealth;
import org.apache.streampipes.resource.management.SpResourceManager;
import org.apache.streampipes.serializers.json.JacksonSerializer;
@@ -33,6 +34,7 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.Map;
+import java.util.Optional;
import java.util.Set;
public class ExtensionInstanceAvailabilityCheck {
@@ -54,31 +56,61 @@ public class ExtensionInstanceAvailabilityCheck {
this.resourceManager = resourceManager;
}
+ /**
+ * Returns the instances the service reports, or an empty health if the
service is no
+ * longer registered. An unreachable service is reported as empty too;
callers that need
+ * to distinguish the two use {@link #checkReachableRunningInstances()}.
+ */
public ExtensionInstanceHealth checkRunningInstances() {
- try {
- var service = extensionsServiceStorage.findAll().stream()
- .filter(svc -> svc.getSvcId().equals(serviceId))
- .findFirst();
+ return
checkReachableRunningInstances().orElseGet(ExtensionInstanceAvailabilityCheck::noInstances);
+ }
- if (service.isEmpty()) {
- return new ExtensionInstanceHealth(Map.of(), Set.of());
- } else {
- var response = extensionRequestManager.request(
- ExtensionServiceRequests
- .extensionInstanceHealth(makeRequestTarget(service.get()),
resourceManager)
- );
- if (response.statusCode() != 200) {
- return new ExtensionInstanceHealth(Map.of(), Set.of());
- }
- return deserialize(response.responseBody());
- }
+ /**
+ * Returns the instances the service reports, an empty health if the service
is no longer
+ * registered or already marked {@link SpServiceStatus#UNHEALTHY}, and
{@link Optional#empty()}
+ * if a service that is not known to be unhealthy did not answer. In the
last case the state
+ * of its instances is unknown and must not be treated as "not running",
otherwise instances
+ * still running there would be restored a second time elsewhere. Once the
service health
+ * check has declared the service unhealthy after repeated failed probes,
its instances are
+ * considered gone so that recovery does not wait for the registration to be
removed.
+ */
+ public Optional<ExtensionInstanceHealth> checkReachableRunningInstances() {
+ var service = extensionsServiceStorage.findAll().stream()
+ .filter(svc -> svc.getSvcId().equals(serviceId))
+ .findFirst();
+ if (service.isEmpty()) {
+ return Optional.of(noInstances());
+ }
+ try {
+ var response = extensionRequestManager.request(
+ ExtensionServiceRequests
+ .extensionInstanceHealth(makeRequestTarget(service.get()),
resourceManager)
+ );
+ if (response.statusCode() != 200) {
+ LOG.warn("Extension service {} answered the instance health request
with status {}",
+ serviceId, response.statusCode());
+ return unanswered(service.get());
+ }
+ return Optional.of(deserialize(response.responseBody()));
} catch (IOException e) {
LOG.error("Extension service {} is unavailable", serviceId);
- return new ExtensionInstanceHealth(Map.of(), Set.of());
+ return unanswered(service.get());
}
}
+ private Optional<ExtensionInstanceHealth> unanswered(SpServiceRegistration
service) {
+ if (service.getStatus() == SpServiceStatus.UNHEALTHY) {
+ LOG.debug("Extension service {} is marked unhealthy, treating its
instances as not running", serviceId);
+ return Optional.of(noInstances());
+ }
+ return Optional.empty();
+ }
+
+ private static ExtensionInstanceHealth noInstances() {
+ return new ExtensionInstanceHealth(Map.of(), Set.of());
+ }
+
private ExtensionInstanceHealth deserialize(String json) throws
JsonProcessingException {
return JacksonSerializer.getObjectMapper().readValue(json,
ExtensionInstanceHealth.class);
}
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 18ae3ada08..579661c7d5 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
@@ -115,7 +115,11 @@ public class PipelineHealthCheck implements HealthCheck {
private void checkAndRestorePipelineElements() {
recoveryBackoff.retainOnly(getActiveRecoveryKeys());
- healthCheckData.activeResources().runningPipelines().forEach(pipeline -> {
+ for (Pipeline pipeline :
healthCheckData.activeResources().runningPipelines()) {
+ if (ShutdownSignal.INSTANCE.isShutdownRequested()) {
+ LOG.debug("Shutdown in progress, skipping pipeline recovery");
+ return;
+ }
List<String> missingInstances = new ArrayList<>();
List<String> recoveredInstances = new ArrayList<>();
List<String> pipelineNotifications = new ArrayList<>();
@@ -128,7 +132,12 @@ public class PipelineHealthCheck implements HealthCheck {
String instanceId =
HealthCheckUtils.extractInstanceId(pipelineElement);
if (isNowhereRunning(instanceId)) {
missingInstances.add(instanceId);
- if (recoveryBackoff.isAttemptDue(pipeline.getPipelineId(),
instanceId)) {
+ if (isOnUnreachableService(pipelineElement)) {
+ // The service is registered but did not answer: the element may
well still be
+ // running there, so restoring it elsewhere could start a
duplicate.
+
HealthCheckUtils.addUnreachableServiceNotification(pipelineNotifications,
pipelineElement);
+ } else if (!ShutdownSignal.INSTANCE.isShutdownRequested()
+ && recoveryBackoff.isAttemptDue(pipeline.getPipelineId(),
instanceId)) {
boolean success = restorePipelineElement(pipelineElement,
pipeline.getPipelineId());
if (!success) {
var state =
recoveryBackoff.recordFailure(pipeline.getPipelineId(), instanceId);
@@ -174,7 +183,7 @@ public class PipelineHealthCheck implements HealthCheck {
pipelinesStats.updatePipelineHealthState(currentPipeline.getElementId(),
currentPipeline.getName(),
currentPipeline.getHealthStatus().toString());
}
- });
+ }
int healthNum = pipelinesStats.getRunningPipelines() -
pipelinesStats.getFailedPipelines()
- pipelinesStats.getAttentionRequiredPipelines();
pipelinesStats.setHealthyPipelines(healthNum);
@@ -242,6 +251,11 @@ public class PipelineHealthCheck implements HealthCheck {
}
}
+ private boolean isOnUnreachableService(InvocableStreamPipesEntity
pipelineElement) {
+ var serviceId = pipelineElement.getSelectedServiceId();
+ return serviceId != null &&
healthCheckData.unreachableServiceIds().contains(serviceId);
+ }
+
private boolean isNowhereRunning(String instanceId) {
return (healthCheckData.activeExtensionInstances().entrySet().stream()
.noneMatch(entry ->
entry.getValue().runningPipelineElementInstanceIds().contains(instanceId)));
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
index b2181752fc..1559fb3d54 100644
---
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
@@ -23,22 +23,31 @@ import java.time.Instant;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.LongSupplier;
+
+/**
+ * Exponential backoff for recovery attempts of individual instances, keyed by
pipeline and
+ * instance id. The delay starts at the health check interval and doubles up
to ten minutes.
+ * An attempt becomes due early when an extension service becomes available,
see
+ * {@link ServiceAvailabilityGeneration}. Public so that registered {@link
HealthCheck}
+ * implementations can apply the same policy to the resources they recover.
+ */
+public final class PipelineRecoveryBackoff {
-final class PipelineRecoveryBackoff {
-
- static final Duration DEFAULT_INITIAL_DELAY = Duration.ofSeconds(30);
- static final Duration DEFAULT_MAX_DELAY = Duration.ofMinutes(10);
+ public static final Duration DEFAULT_INITIAL_DELAY = Duration.ofSeconds(30);
+ public static final Duration DEFAULT_MAX_DELAY = Duration.ofMinutes(10);
private final Clock clock;
private final Duration initialDelay;
private final Duration maxDelay;
+ private final LongSupplier availabilityGeneration;
private final Map<RecoveryKey, RecoveryState> recoveryStates = new
ConcurrentHashMap<>();
- PipelineRecoveryBackoff() {
+ public PipelineRecoveryBackoff() {
this(Clock.systemUTC(), DEFAULT_INITIAL_DELAY, DEFAULT_MAX_DELAY);
}
- PipelineRecoveryBackoff(Duration healthCheckInterval) {
+ public PipelineRecoveryBackoff(Duration healthCheckInterval) {
this(
Clock.systemUTC(),
healthCheckInterval,
@@ -46,42 +55,57 @@ final class PipelineRecoveryBackoff {
);
}
- PipelineRecoveryBackoff(Clock clock,
- Duration initialDelay,
- Duration maxDelay) {
+ public PipelineRecoveryBackoff(Clock clock,
+ Duration initialDelay,
+ Duration maxDelay) {
+ this(clock, initialDelay, maxDelay,
ServiceAvailabilityGeneration.INSTANCE::current);
+ }
+
+ public PipelineRecoveryBackoff(Clock clock,
+ Duration initialDelay,
+ Duration maxDelay,
+ LongSupplier availabilityGeneration) {
this.clock = clock;
this.initialDelay = initialDelay;
this.maxDelay = maxDelay;
+ this.availabilityGeneration = availabilityGeneration;
}
- boolean isAttemptDue(String pipelineId,
- String instanceId) {
+ /**
+ * An attempt is due when no failure is recorded, when the backoff delay has
elapsed,
+ * or when an extension service became available since the last failure.
+ */
+ public boolean isAttemptDue(String pipelineId,
+ String instanceId) {
var state = recoveryStates.get(new RecoveryKey(pipelineId, instanceId));
- return state == null || !clock.instant().isBefore(state.nextAttemptAt());
+ return state == null
+ || state.availabilityGeneration() != availabilityGeneration.getAsLong()
+ || !clock.instant().isBefore(state.nextAttemptAt());
}
- RecoveryState recordFailure(String pipelineId,
- String instanceId) {
+ public 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));
+ return new RecoveryState(failedAttempts, delay,
clock.instant().plus(delay),
+ availabilityGeneration.getAsLong());
});
}
- int reset(String pipelineId,
- String instanceId) {
+ public 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) {
+ public void retainOnly(Set<RecoveryKey> activeInstances) {
recoveryStates.keySet().retainAll(activeInstances);
}
- RecoveryState getState(String pipelineId,
- String instanceId) {
+ public RecoveryState getState(String pipelineId,
+ String instanceId) {
return recoveryStates.get(new RecoveryKey(pipelineId, instanceId));
}
@@ -97,11 +121,12 @@ final class PipelineRecoveryBackoff {
return first.compareTo(second) >= 0 ? first : second;
}
- record RecoveryKey(String pipelineId, String instanceId) {
+ public record RecoveryKey(String pipelineId, String instanceId) {
}
- record RecoveryState(int failedAttempts,
+ public record RecoveryState(int failedAttempts,
Duration delay,
- Instant nextAttemptAt) {
+ Instant nextAttemptAt,
+ long availabilityGeneration) {
}
}
diff --git
a/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/model/HealthCheckData.java
b/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/ServiceAvailabilityGeneration.java
similarity index 53%
copy from
streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/model/HealthCheckData.java
copy to
streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/ServiceAvailabilityGeneration.java
index 6a0744d10f..3b965f2a9a 100644
---
a/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/model/HealthCheckData.java
+++
b/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/ServiceAvailabilityGeneration.java
@@ -16,15 +16,26 @@
*
*/
-package org.apache.streampipes.health.monitoring.model;
+package org.apache.streampipes.health.monitoring;
-import org.apache.streampipes.health.monitoring.ResourceProvider;
-import org.apache.streampipes.model.health.ExtensionInstanceHealth;
+import java.util.concurrent.atomic.AtomicLong;
-import java.util.Map;
+/**
+ * Counts the times an extension service became available (registered and
healthy).
+ * {@link PipelineRecoveryBackoff} compares the generation recorded at the
last failed
+ * recovery attempt with the current one, so a service that comes back makes
pending
+ * recoveries eligible immediately instead of waiting out the exponential
backoff.
+ */
+public enum ServiceAvailabilityGeneration {
+ INSTANCE;
+
+ private final AtomicLong generation = new AtomicLong();
+
+ public void serviceBecameAvailable() {
+ generation.incrementAndGet();
+ }
-public record HealthCheckData(ResourceProvider resourceProvider,
- ActiveResources activeResources,
- Map<String, ActiveCoreInstances>
activeCoreInstances,
- Map<String, ExtensionInstanceHealth>
activeExtensionInstances) {
+ public long current() {
+ return generation.get();
+ }
}
diff --git
a/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/ServiceHealthCheck.java
b/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/ServiceHealthCheck.java
index aba373ec3c..7542489f8b 100644
---
a/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/ServiceHealthCheck.java
+++
b/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/ServiceHealthCheck.java
@@ -85,6 +85,10 @@ public class ServiceHealthCheck implements Runnable {
@Override
public void run() {
+ if (ShutdownSignal.INSTANCE.isShutdownRequested()) {
+ LOG.debug("Shutdown in progress, skipping service health check");
+ return;
+ }
try {
Environment env = Environments.getEnvironment();
diff --git
a/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/ServiceRegistrationManager.java
b/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/ServiceRegistrationManager.java
index 43c1d571fe..de69749fd0 100644
---
a/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/ServiceRegistrationManager.java
+++
b/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/ServiceRegistrationManager.java
@@ -60,6 +60,7 @@ public class ServiceRegistrationManager {
serviceRegistration.setStatus(status);
storage.updateElement(serviceRegistration);
logService(serviceRegistration);
+ notifyIfAvailable(status);
}
public void addService(SpServiceRegistration serviceRegistration,
SpServiceStatus status) {
@@ -72,6 +73,13 @@ public class ServiceRegistrationManager {
storage.persist(serviceRegistration);
}
logService(serviceRegistration);
+ notifyIfAvailable(status);
+ }
+
+ private void notifyIfAvailable(SpServiceStatus status) {
+ if (status == SpServiceStatus.HEALTHY) {
+ ServiceAvailabilityGeneration.INSTANCE.serviceBecameAvailable();
+ }
}
public List<SpServiceRegistration> getAllServices() {
diff --git
a/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/model/HealthCheckData.java
b/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/ShutdownSignal.java
similarity index 55%
copy from
streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/model/HealthCheckData.java
copy to
streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/ShutdownSignal.java
index 6a0744d10f..3c3979e74a 100644
---
a/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/model/HealthCheckData.java
+++
b/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/ShutdownSignal.java
@@ -16,15 +16,30 @@
*
*/
-package org.apache.streampipes.health.monitoring.model;
+package org.apache.streampipes.health.monitoring;
-import org.apache.streampipes.health.monitoring.ResourceProvider;
-import org.apache.streampipes.model.health.ExtensionInstanceHealth;
+/**
+ * Process-wide flag raised by the core when it starts shutting down. Health
checks
+ * and recovery tasks consult it so that they do not restore or re-invoke
pipeline
+ * elements while the service is going away.
+ */
+public enum ShutdownSignal {
+ INSTANCE;
+
+ private volatile boolean shutdownRequested = false;
+
+ public void requestShutdown() {
+ this.shutdownRequested = true;
+ }
-import java.util.Map;
+ public boolean isShutdownRequested() {
+ return shutdownRequested;
+ }
-public record HealthCheckData(ResourceProvider resourceProvider,
- ActiveResources activeResources,
- Map<String, ActiveCoreInstances>
activeCoreInstances,
- Map<String, ExtensionInstanceHealth>
activeExtensionInstances) {
+ /**
+ * Clears the signal. Intended for tests that share the singleton.
+ */
+ public void reset() {
+ this.shutdownRequested = false;
+ }
}
diff --git
a/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/StaleRegistrationCleanup.java
b/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/StaleRegistrationCleanup.java
new file mode 100644
index 0000000000..f23cf0bc5a
--- /dev/null
+++
b/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/StaleRegistrationCleanup.java
@@ -0,0 +1,105 @@
+/*
+ * 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.manager.api.extensions.ExtensionServiceRequestManager;
+import
org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestTargets;
+import org.apache.streampipes.manager.api.extensions.ExtensionServiceRequests;
+import
org.apache.streampipes.model.extensions.svcdiscovery.SpServiceRegistration;
+import org.apache.streampipes.resource.management.SpResourceManager;
+import org.apache.streampipes.storage.api.system.IExtensionsServiceStorage;
+
+import org.apache.http.HttpStatus;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * Removes registrations left behind by a predecessor of a newly registered
extension service.
+ *
+ * <p>A restarted extension service registers under a new id but keeps its
group, host and
+ * port. A registration of the same group at the same address that was never
deregistered
+ * usually belongs to a predecessor that is gone. Left in place, it counts as
an unreachable
+ * service whose instances are unknown, and recovery of the pipeline elements
that ran there
+ * waits until the service health check removes it. Several live replicas may
share one
+ * address as well, for example behind a Kubernetes service name, so every
candidate is asked
+ * for its own health first: the per-service health endpoint answers only for
its own id, and
+ * the NATS request topic is per id, so a live replica keeps its registration
and only a
+ * registration nobody answers for is removed.
+ */
+public class StaleRegistrationCleanup {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(StaleRegistrationCleanup.class);
+
+ private final IExtensionsServiceStorage storage;
+ private final ExtensionServiceRequestManager requestManager;
+ private final SpResourceManager resourceManager;
+
+ public StaleRegistrationCleanup(IExtensionsServiceStorage storage,
+ ExtensionServiceRequestManager
requestManager,
+ SpResourceManager resourceManager) {
+ this.storage = storage;
+ this.requestManager = requestManager;
+ this.resourceManager = resourceManager;
+ }
+
+ /**
+ * Probes every other registration of the same group at the same address and
removes the
+ * ones that do not answer for their own id. Never throws; failures are
logged.
+ */
+ public void removePredecessorsOf(SpServiceRegistration registration) {
+ try {
+ findCandidates(registration).forEach(candidate -> {
+ if (isAlive(candidate)) {
+ LOG.info("Service {} at {}:{} is alive next to new registration {},
keeping it",
+ candidate.getSvcId(), candidate.getHost(), candidate.getPort(),
registration.getSvcId());
+ } else {
+ LOG.info("Removing stale registration {} of service {} at {}:{},
replaced by {}",
+ candidate.getSvcId(), candidate.getSvcGroup(),
candidate.getHost(), candidate.getPort(),
+ registration.getSvcId());
+ storage.deleteElement(candidate);
+ }
+ });
+ } catch (RuntimeException e) {
+ LOG.warn("Could not clean up stale registrations for service {}",
registration.getSvcId(), e);
+ }
+ }
+
+ List<SpServiceRegistration> findCandidates(SpServiceRegistration
registration) {
+ return storage.findAll().stream()
+ .filter(other -> !Objects.equals(other.getSvcId(),
registration.getSvcId()))
+ .filter(other -> Objects.equals(other.getSvcGroup(),
registration.getSvcGroup()))
+ .filter(other -> Objects.equals(other.getHost(),
registration.getHost()))
+ .filter(other -> other.getPort() == registration.getPort())
+ .toList();
+ }
+
+ private boolean isAlive(SpServiceRegistration candidate) {
+ try {
+ var target = ExtensionServiceRequestTargets.serviceHealth(candidate,
candidate.getHealthCheckPath());
+ var response =
requestManager.request(ExtensionServiceRequests.serviceHealth(target,
resourceManager));
+ return response.statusCode() == HttpStatus.SC_OK;
+ } catch (IOException | RuntimeException e) {
+ return false;
+ }
+ }
+}
diff --git
a/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/model/HealthCheckData.java
b/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/model/HealthCheckData.java
index 6a0744d10f..54c6ce00a6 100644
---
a/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/model/HealthCheckData.java
+++
b/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/model/HealthCheckData.java
@@ -22,9 +22,23 @@ import
org.apache.streampipes.health.monitoring.ResourceProvider;
import org.apache.streampipes.model.health.ExtensionInstanceHealth;
import java.util.Map;
+import java.util.Set;
+/**
+ * @param activeExtensionInstances instance health per reachable extension
service
+ * @param unreachableServiceIds registered services that did not answer the
health request;
+ * the state of their instances is unknown, not
empty
+ */
public record HealthCheckData(ResourceProvider resourceProvider,
ActiveResources activeResources,
Map<String, ActiveCoreInstances>
activeCoreInstances,
- Map<String, ExtensionInstanceHealth>
activeExtensionInstances) {
+ Map<String, ExtensionInstanceHealth>
activeExtensionInstances,
+ Set<String> unreachableServiceIds) {
+
+ public HealthCheckData(ResourceProvider resourceProvider,
+ ActiveResources activeResources,
+ Map<String, ActiveCoreInstances> activeCoreInstances,
+ Map<String, ExtensionInstanceHealth>
activeExtensionInstances) {
+ this(resourceProvider, activeResources, activeCoreInstances,
activeExtensionInstances, Set.of());
+ }
}
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 ee7e79043c..0160c4f466 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
@@ -45,6 +45,13 @@ public class HealthCheckUtils {
+ "' is not available. The next automatic recovery attempt is
pending.");
}
+ public static void addUnreachableServiceNotification(List<String>
pipelineNotifications,
+
InvocableStreamPipesEntity pipelineElement) {
+ pipelineNotifications.add(getCurrentDatetime() + "Pipeline element '" +
pipelineElement.getName()
+ + "' runs on extension service '" +
pipelineElement.getSelectedServiceId()
+ + "' which is currently unreachable. Recovery waits until the service
answers again or is considered gone.");
+ }
+
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/AdapterHealthCheckTest.java
b/streampipes-health-monitoring/src/test/java/org/apache/streampipes/health/monitoring/AdapterHealthCheckTest.java
index 47207d027d..40628749b8 100644
---
a/streampipes-health-monitoring/src/test/java/org/apache/streampipes/health/monitoring/AdapterHealthCheckTest.java
+++
b/streampipes-health-monitoring/src/test/java/org/apache/streampipes/health/monitoring/AdapterHealthCheckTest.java
@@ -30,6 +30,7 @@ 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;
public class AdapterHealthCheckTest {
@@ -62,6 +63,50 @@ public class AdapterHealthCheckTest {
assertTrue(adaptersToRecover.isEmpty());
}
+ @Test
+ public void adapterOnUnreachableServiceIsNotRecovered() {
+ var adapterDescription = new AdapterDescription();
+ adapterDescription.setElementId("adapter-id");
+ adapterDescription.setRunning(true);
+ adapterDescription.setSelectedServiceId("unreachable-service");
+ var activeResources = new ActiveResources(
+ List.of(),
+ List.of(),
+ List.of(adapterDescription),
+ List.of(adapterDescription)
+ );
+ var healthCheckData = new HealthCheckData(
+ null,
+ activeResources,
+ Map.of(),
+ Map.of(),
+ Set.of("unreachable-service")
+ );
+
+ var adaptersToRecover = new
AdapterHealthCheck(healthCheckData).getAdaptersToRecover();
+
+ assertTrue(adaptersToRecover.isEmpty());
+ }
+
+ @Test
+ public void adapterOnRemovedServiceIsRecovered() {
+ var adapterDescription = new AdapterDescription();
+ adapterDescription.setElementId("adapter-id");
+ adapterDescription.setRunning(true);
+ adapterDescription.setSelectedServiceId("removed-service");
+ var activeResources = new ActiveResources(
+ List.of(),
+ List.of(),
+ List.of(adapterDescription),
+ List.of(adapterDescription)
+ );
+ var healthCheckData = new HealthCheckData(null, activeResources, Map.of(),
Map.of(), Set.of());
+
+ var adaptersToRecover = new
AdapterHealthCheck(healthCheckData).getAdaptersToRecover();
+
+ assertEquals(1, adaptersToRecover.size());
+ }
+
@Test
public void runCheckDoesNotRestartStartingAdapters() {
var adapterHealthCheck = new TestAdapterHealthCheck(
diff --git
a/streampipes-health-monitoring/src/test/java/org/apache/streampipes/health/monitoring/ExtensionHealthCheckTest.java
b/streampipes-health-monitoring/src/test/java/org/apache/streampipes/health/monitoring/ExtensionHealthCheckTest.java
new file mode 100644
index 0000000000..dc0b359e6b
--- /dev/null
+++
b/streampipes-health-monitoring/src/test/java/org/apache/streampipes/health/monitoring/ExtensionHealthCheckTest.java
@@ -0,0 +1,129 @@
+/*
+ * 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.ActiveCoreInstances;
+import org.apache.streampipes.health.monitoring.model.ActiveResources;
+import org.apache.streampipes.health.monitoring.model.HealthCheckData;
+import
org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestManager;
+import org.apache.streampipes.resource.management.SpResourceManager;
+import org.apache.streampipes.storage.api.system.IExtensionsServiceStorage;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+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.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class ExtensionHealthCheckTest {
+
+ private ResourceProvider resourceProvider;
+ private ExtensionHealthCheck check;
+
+ @BeforeEach
+ void setUp() {
+ ShutdownSignal.INSTANCE.reset();
+ resourceProvider = mock(ResourceProvider.class);
+ when(resourceProvider.loadActiveResources())
+ .thenReturn(new ActiveResources(List.of(), List.of(), List.of(),
List.of()));
+ when(resourceProvider.loadActiveInstances(any())).thenReturn(Map.of());
+ check = new ExtensionHealthCheck(
+ resourceProvider,
+ mock(IExtensionsServiceStorage.class),
+ mock(ExtensionServiceRequestManager.class),
+ mock(SpResourceManager.class),
+ List.of(),
+ Duration.ofSeconds(30)
+ );
+ }
+
+ @AfterEach
+ void tearDown() {
+ ShutdownSignal.INSTANCE.reset();
+ }
+
+ @Test
+ void runsWhileServiceIsUp() {
+ check.run();
+
+ verify(resourceProvider).loadActiveResources();
+ }
+
+ @Test
+ void serviceStaysUnknownForLimitedTicksThenCountsAsDown() {
+ when(resourceProvider.loadActiveInstances(any()))
+ .thenReturn(Map.of("svc", new ActiveCoreInstances(List.of(),
Map.of())));
+ var observed = new AtomicReference<HealthCheckData>();
+ var neverAnswering = new ExtensionHealthCheck(
+ resourceProvider,
+ mock(IExtensionsServiceStorage.class),
+ mock(ExtensionServiceRequestManager.class),
+ mock(SpResourceManager.class),
+ List.of(),
+ Duration.ofSeconds(30)
+ ) {
+ @Override
+ protected
Optional<org.apache.streampipes.model.health.ExtensionInstanceHealth>
checkInstances(String id) {
+ return Optional.empty();
+ }
+
+ @Override
+ protected int maxUnknownTicks() {
+ return 2;
+ }
+
+ @Override
+ protected List<HealthCheck> getBuiltInHealthChecks(HealthCheckData
healthCheckData) {
+ observed.set(healthCheckData);
+ return List.of();
+ }
+ };
+
+ neverAnswering.run();
+ assertTrue(observed.get().unreachableServiceIds().contains("svc"));
+ assertFalse(observed.get().activeExtensionInstances().containsKey("svc"));
+
+ neverAnswering.run();
+ assertFalse(observed.get().unreachableServiceIds().contains("svc"));
+ assertTrue(observed.get().activeExtensionInstances().containsKey("svc"));
+ assertEquals(0,
observed.get().activeExtensionInstances().get("svc").runningPipelineElementInstanceIds().size());
+ }
+
+ @Test
+ void skipsCheckOnceShutdownIsRequested() {
+ ShutdownSignal.INSTANCE.requestShutdown();
+
+ check.run();
+
+ verify(resourceProvider, never()).loadActiveResources();
+ }
+}
diff --git
a/streampipes-health-monitoring/src/test/java/org/apache/streampipes/health/monitoring/ExtensionInstanceAvailabilityCheckTest.java
b/streampipes-health-monitoring/src/test/java/org/apache/streampipes/health/monitoring/ExtensionInstanceAvailabilityCheckTest.java
new file mode 100644
index 0000000000..0edae65cf4
--- /dev/null
+++
b/streampipes-health-monitoring/src/test/java/org/apache/streampipes/health/monitoring/ExtensionInstanceAvailabilityCheckTest.java
@@ -0,0 +1,123 @@
+/*
+ * 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.manager.api.extensions.ExtensionServiceOperationResult;
+import org.apache.streampipes.manager.api.extensions.ExtensionServiceRequest;
+import
org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestManager;
+import org.apache.streampipes.manager.api.extensions.ExtensionServiceRequests;
+import
org.apache.streampipes.model.extensions.svcdiscovery.SpServiceRegistration;
+import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceStatus;
+import org.apache.streampipes.resource.management.SpResourceManager;
+import org.apache.streampipes.storage.api.system.IExtensionsServiceStorage;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+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.mockStatic;
+import static org.mockito.Mockito.when;
+
+class ExtensionInstanceAvailabilityCheckTest {
+
+ private static final String SERVICE_ID = "svc-1";
+
+ private IExtensionsServiceStorage storage;
+ private ExtensionServiceRequestManager requests;
+ private MockedStatic<ExtensionServiceRequests> requestFactory;
+ private SpServiceRegistration service;
+ private ExtensionInstanceAvailabilityCheck check;
+
+ @BeforeEach
+ void setUp() {
+ storage = mock(IExtensionsServiceStorage.class);
+ requests = mock(ExtensionServiceRequestManager.class);
+ service = new SpServiceRegistration();
+ service.setSvcId(SERVICE_ID);
+ service.setStatus(SpServiceStatus.HEALTHY);
+ when(storage.findAll()).thenReturn(List.of(service));
+ requestFactory = mockStatic(ExtensionServiceRequests.class);
+ requestFactory.when(() ->
ExtensionServiceRequests.extensionInstanceHealth(any(), any()))
+ .thenReturn(mock(ExtensionServiceRequest.class));
+ check = new ExtensionInstanceAvailabilityCheck(storage, SERVICE_ID,
requests, mock(SpResourceManager.class));
+ }
+
+ @AfterEach
+ void tearDown() {
+ requestFactory.close();
+ }
+
+ @Test
+ void reachableServiceReportsItsInstances() throws IOException {
+ var body =
"{\"adapterInstanceStates\":{},\"runningPipelineElementInstanceIds\":[\"instance-a\"]}";
+ when(requests.request(any()))
+ .thenReturn(new ExtensionServiceOperationResult(200,
body.getBytes(StandardCharsets.UTF_8)));
+
+ var health = check.checkReachableRunningInstances();
+
+ assertTrue(health.isPresent());
+ assertEquals(Set.of("instance-a"),
health.get().runningPipelineElementInstanceIds());
+ }
+
+ @Test
+ void healthyServiceThatDoesNotAnswerIsUnknown() throws IOException {
+ when(requests.request(any())).thenThrow(new IOException("no responders"));
+
+ assertTrue(check.checkReachableRunningInstances().isEmpty());
+
assertTrue(check.checkRunningInstances().runningPipelineElementInstanceIds().isEmpty());
+ }
+
+ @Test
+ void healthyServiceAnsweringWithErrorIsUnknown() throws IOException {
+ when(requests.request(any())).thenReturn(new
ExtensionServiceOperationResult(500, null));
+
+ assertTrue(check.checkReachableRunningInstances().isEmpty());
+ }
+
+ @Test
+ void unhealthyServiceThatDoesNotAnswerHasNoInstances() throws IOException {
+ service.setStatus(SpServiceStatus.UNHEALTHY);
+ when(requests.request(any())).thenThrow(new IOException("no responders"));
+
+ var health = check.checkReachableRunningInstances();
+
+ assertTrue(health.isPresent());
+ assertTrue(health.get().runningPipelineElementInstanceIds().isEmpty());
+ }
+
+ @Test
+ void unregisteredServiceHasNoInstances() throws IOException {
+ when(storage.findAll()).thenReturn(List.of());
+
+ var health = check.checkReachableRunningInstances();
+
+ assertTrue(health.isPresent());
+ assertTrue(health.get().runningPipelineElementInstanceIds().isEmpty());
+ }
+}
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 8c83031102..04c8f5a10f 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
@@ -27,6 +27,7 @@ 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.AfterEach;
import org.junit.jupiter.api.Test;
import java.time.Clock;
@@ -46,6 +47,38 @@ import static org.mockito.Mockito.when;
public class PipelineHealthCheckTest {
+ @AfterEach
+ public void tearDown() {
+ ShutdownSignal.INSTANCE.reset();
+ }
+
+ @Test
+ public void elementOnUnreachableServiceIsNotRestoredElsewhere() {
+ var processor = processor("missing", "Missing processor");
+ processor.setSelectedServiceId("unreachable-service");
+ var pipeline = pipeline(processor);
+ var healthCheck = healthCheck(pipeline, Map.of("missing", true), new
PipelineRecoveryBackoff(), Set.of(),
+ Set.of("unreachable-service"));
+
+ healthCheck.runCheck();
+
+ assertEquals(0, healthCheck.restoreAttempts("missing"));
+ assertEquals(PipelineHealthStatus.FAILURE, pipeline.getHealthStatus());
+ assertTrue(pipeline.getPipelineNotifications().get(0).contains("currently
unreachable"));
+ }
+
+ @Test
+ public void noRecoveryOnceShutdownIsRequested() {
+ var pipeline = pipeline(processor("missing", "Missing processor"));
+ var healthCheck = healthCheck(pipeline, Map.of("missing", true));
+ ShutdownSignal.INSTANCE.requestShutdown();
+
+ healthCheck.runCheck();
+
+ assertEquals(0, healthCheck.restoreAttempts("missing"));
+ assertEquals(PipelineHealthStatus.OK, pipeline.getHealthStatus());
+ }
+
@Test
public void retriesRestorationBeyondPreviousLimit() {
var pipeline = pipeline(processor("missing", "Missing processor"));
@@ -160,6 +193,14 @@ public class PipelineHealthCheckTest {
Map<String, Boolean>
restorationResults,
PipelineRecoveryBackoff
recoveryBackoff,
Set<String> runningInstanceIds) {
+ return healthCheck(pipeline, restorationResults, recoveryBackoff,
runningInstanceIds, Set.of());
+ }
+
+ private TestPipelineHealthCheck healthCheck(Pipeline pipeline,
+ Map<String, Boolean>
restorationResults,
+ PipelineRecoveryBackoff
recoveryBackoff,
+ Set<String> runningInstanceIds,
+ Set<String>
unreachableServiceIds) {
IPipelineStorage pipelineStorage = mock(IPipelineStorage.class);
when(pipelineStorage.getElementById(pipeline.getPipelineId())).thenReturn(pipeline);
when(pipelineStorage.updateElement(any())).thenAnswer(invocation ->
invocation.getArgument(0));
@@ -175,7 +216,8 @@ public class PipelineHealthCheckTest {
resourceProvider,
activeResources,
Map.of(),
- Map.of("service-id", new ExtensionInstanceHealth(Map.of(),
runningInstanceIds))
+ Map.of("service-id", new ExtensionInstanceHealth(Map.of(),
runningInstanceIds)),
+ unreachableServiceIds
);
return new TestPipelineHealthCheck(
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
index 8bbb32f2be..bfdd118158 100644
---
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
@@ -24,6 +24,7 @@ import java.time.Duration;
import java.time.Instant;
import java.time.ZoneOffset;
import java.util.Set;
+import java.util.concurrent.atomic.AtomicLong;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -76,6 +77,26 @@ public class PipelineRecoveryBackoffTest {
assertTrue(recoveryBackoff.isAttemptDue("pipeline", "instance"));
}
+ @Test
+ public void serviceBecomingAvailableAllowsImmediateRecoveryAttempt() {
+ var generation = new AtomicLong();
+ var recoveryBackoff = new PipelineRecoveryBackoff(
+ Clock.fixed(Instant.EPOCH, ZoneOffset.UTC),
+ PipelineRecoveryBackoff.DEFAULT_INITIAL_DELAY,
+ PipelineRecoveryBackoff.DEFAULT_MAX_DELAY,
+ generation::get
+ );
+ recoveryBackoff.recordFailure("pipeline", "instance");
+ assertFalse(recoveryBackoff.isAttemptDue("pipeline", "instance"));
+
+ generation.incrementAndGet();
+
+ assertTrue(recoveryBackoff.isAttemptDue("pipeline", "instance"));
+ // A further failure after the service came back is subject to the backoff
again.
+ recoveryBackoff.recordFailure("pipeline", "instance");
+ assertFalse(recoveryBackoff.isAttemptDue("pipeline", "instance"));
+ }
+
@Test
public void inactiveInstancesAreRemoved() {
var recoveryBackoff = recoveryBackoff();
@@ -94,7 +115,8 @@ public class PipelineRecoveryBackoffTest {
return new PipelineRecoveryBackoff(
Clock.fixed(Instant.EPOCH, ZoneOffset.UTC),
PipelineRecoveryBackoff.DEFAULT_INITIAL_DELAY,
- PipelineRecoveryBackoff.DEFAULT_MAX_DELAY
+ PipelineRecoveryBackoff.DEFAULT_MAX_DELAY,
+ () -> 0L
);
}
}
diff --git
a/streampipes-health-monitoring/src/test/java/org/apache/streampipes/health/monitoring/ServiceRegistrationManagerTest.java
b/streampipes-health-monitoring/src/test/java/org/apache/streampipes/health/monitoring/ServiceRegistrationManagerTest.java
new file mode 100644
index 0000000000..0c05636289
--- /dev/null
+++
b/streampipes-health-monitoring/src/test/java/org/apache/streampipes/health/monitoring/ServiceRegistrationManagerTest.java
@@ -0,0 +1,74 @@
+/*
+ * 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.model.extensions.svcdiscovery.SpServiceRegistration;
+import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceStatus;
+import org.apache.streampipes.storage.api.system.IExtensionsServiceStorage;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class ServiceRegistrationManagerTest {
+
+ @Test
+ void newRegistrationIsPersistedWithoutTouchingOtherRegistrations() {
+ var storage = mock(IExtensionsServiceStorage.class);
+ var sameAddress = registration("iiot-old", "iiot", "extensions", 8090);
+ when(storage.findAll()).thenReturn(List.of(sameAddress));
+ var fresh = registration("iiot-new", "iiot", "extensions", 8090);
+
+ new ServiceRegistrationManager(storage).addService(fresh,
SpServiceStatus.REGISTERED);
+
+ // Stale predecessors are removed by StaleRegistrationCleanup after
probing them, not here.
+ verify(storage, never()).deleteElement(any());
+ verify(storage).persist(fresh);
+ }
+
+ @Test
+ void reRegistrationOfKnownServiceDoesNotRemoveAnything() {
+ var storage = mock(IExtensionsServiceStorage.class);
+ var existing = registration("iiot-1", "iiot", "extensions", 8090);
+ existing.setRev("3-abc");
+ when(storage.getElementById("iiot-1")).thenReturn(existing);
+ var refreshed = registration("iiot-1", "iiot", "extensions", 8090);
+
+ new ServiceRegistrationManager(storage).addService(refreshed,
SpServiceStatus.REGISTERED);
+
+ verify(storage, never()).deleteElement(any());
+ verify(storage).updateElement(refreshed);
+ }
+
+ private SpServiceRegistration registration(String svcId, String group,
String host, int port) {
+ var registration = new SpServiceRegistration();
+ registration.setSvcId(svcId);
+ registration.setSvcGroup(group);
+ registration.setHost(host);
+ registration.setPort(port);
+ registration.setStatus(SpServiceStatus.HEALTHY);
+ return registration;
+ }
+}
diff --git
a/streampipes-health-monitoring/src/test/java/org/apache/streampipes/health/monitoring/StaleRegistrationCleanupTest.java
b/streampipes-health-monitoring/src/test/java/org/apache/streampipes/health/monitoring/StaleRegistrationCleanupTest.java
new file mode 100644
index 0000000000..4405f33ad5
--- /dev/null
+++
b/streampipes-health-monitoring/src/test/java/org/apache/streampipes/health/monitoring/StaleRegistrationCleanupTest.java
@@ -0,0 +1,128 @@
+/*
+ * 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.manager.api.extensions.ExtensionServiceOperationResult;
+import org.apache.streampipes.manager.api.extensions.ExtensionServiceRequest;
+import
org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestManager;
+import org.apache.streampipes.manager.api.extensions.ExtensionServiceRequests;
+import
org.apache.streampipes.model.extensions.svcdiscovery.SpServiceRegistration;
+import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceStatus;
+import org.apache.streampipes.resource.management.SpResourceManager;
+import org.apache.streampipes.storage.api.system.IExtensionsServiceStorage;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+
+import java.io.IOException;
+import java.util.List;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class StaleRegistrationCleanupTest {
+
+ private IExtensionsServiceStorage storage;
+ private ExtensionServiceRequestManager requests;
+ private MockedStatic<ExtensionServiceRequests> requestFactory;
+ private StaleRegistrationCleanup cleanup;
+
+ @BeforeEach
+ void setUp() {
+ storage = mock(IExtensionsServiceStorage.class);
+ requests = mock(ExtensionServiceRequestManager.class);
+ requestFactory = mockStatic(ExtensionServiceRequests.class);
+ requestFactory.when(() -> ExtensionServiceRequests.serviceHealth(any(),
any()))
+ .thenReturn(mock(ExtensionServiceRequest.class));
+ cleanup = new StaleRegistrationCleanup(storage, requests,
mock(SpResourceManager.class));
+ }
+
+ @AfterEach
+ void tearDown() {
+ requestFactory.close();
+ }
+
+ @Test
+ void removesPredecessorThatNobodyAnswersFor() throws IOException {
+ var stale = registration("iiot-old", "iiot", "extensions", 8090);
+ var fresh = registration("iiot-new", "iiot", "extensions", 8090);
+ when(storage.findAll()).thenReturn(List.of(stale, fresh));
+ when(requests.request(any())).thenThrow(new IOException("connection
refused"));
+
+ cleanup.removePredecessorsOf(fresh);
+
+ verify(storage).deleteElement(stale);
+ }
+
+ @Test
+ void removesPredecessorWhoseAddressIsNowServedByAnotherId() throws
IOException {
+ var stale = registration("iiot-old", "iiot", "extensions", 8090);
+ var fresh = registration("iiot-new", "iiot", "extensions", 8090);
+ when(storage.findAll()).thenReturn(List.of(stale, fresh));
+ when(requests.request(any())).thenReturn(new
ExtensionServiceOperationResult(404, null));
+
+ cleanup.removePredecessorsOf(fresh);
+
+ verify(storage).deleteElement(stale);
+ }
+
+ @Test
+ void keepsLiveReplicaAtSameAddress() throws IOException {
+ var replica = registration("iiot-a", "iiot", "extensions", 8090);
+ var fresh = registration("iiot-b", "iiot", "extensions", 8090);
+ when(storage.findAll()).thenReturn(List.of(replica, fresh));
+ when(requests.request(any())).thenReturn(new
ExtensionServiceOperationResult(200, null));
+
+ cleanup.removePredecessorsOf(fresh);
+
+ verify(storage, never()).deleteElement(any());
+ }
+
+ @Test
+ void onlyProbesRegistrationsOfSameGroupAndAddress() throws IOException {
+ var otherPort = registration("iiot-other", "iiot", "extensions", 8091);
+ var otherGroup = registration("ai-old", "ai", "extensions", 8090);
+ var otherHost = registration("iiot-remote", "iiot", "node-2", 8090);
+ var fresh = registration("iiot-new", "iiot", "extensions", 8090);
+ when(storage.findAll()).thenReturn(List.of(otherPort, otherGroup,
otherHost, fresh));
+
+ cleanup.removePredecessorsOf(fresh);
+
+ verify(requests, times(0)).request(any());
+ verify(storage, never()).deleteElement(any());
+ }
+
+ private SpServiceRegistration registration(String svcId, String group,
String host, int port) {
+ var registration = new SpServiceRegistration();
+ registration.setSvcId(svcId);
+ registration.setSvcGroup(group);
+ registration.setHost(host);
+ registration.setPort(port);
+ registration.setHealthCheckPath("svchealth/" + svcId);
+ registration.setStatus(SpServiceStatus.HEALTHY);
+ return registration;
+ }
+}
diff --git
a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/admin/ServiceRegistrationResource.java
b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/admin/ServiceRegistrationResource.java
index 5de6bd0f41..b1d35f244c 100644
---
a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/admin/ServiceRegistrationResource.java
+++
b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/admin/ServiceRegistrationResource.java
@@ -21,6 +21,7 @@ import
org.apache.streampipes.health.monitoring.ExtensionInstanceDetailsProvider
import
org.apache.streampipes.health.monitoring.ExtensionInstanceRemovalService;
import org.apache.streampipes.health.monitoring.ResourceProvider;
import org.apache.streampipes.health.monitoring.ServiceRegistrationManager;
+import org.apache.streampipes.health.monitoring.StaleRegistrationCleanup;
import
org.apache.streampipes.health.monitoring.model.RunningExtensionInstances;
import
org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestManager;
import
org.apache.streampipes.model.extensions.svcdiscovery.SpServiceRegistration;
@@ -46,6 +47,7 @@ import org.springframework.web.bind.annotation.RestController;
import java.io.IOException;
import java.util.List;
+import java.util.concurrent.CompletableFuture;
@RestController
@RequestMapping("/api/v2/extensions-services")
@@ -117,6 +119,10 @@ public class ServiceRegistrationResource extends
AbstractAuthGuardedRestResource
public ResponseEntity<Void> registerService(@RequestBody
SpServiceRegistration serviceRegistration) {
new
ServiceRegistrationManager(extensionsServiceStorage).addService(serviceRegistration,
SpServiceStatus.REGISTERED);
+ // Probing predecessors may wait for connection timeouts; do not delay the
registration.
+ CompletableFuture.runAsync(() -> new StaleRegistrationCleanup(
+ extensionsServiceStorage, extensionServiceRequestManager,
resourceManager
+ ).removePredecessorsOf(serviceRegistration));
return ok();
}
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 ecddb6f161..2fff6ac78f 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
@@ -19,6 +19,7 @@
package org.apache.streampipes.service.core;
import org.apache.streampipes.audit.events.adapter.AdapterAuditRecorder;
+import org.apache.streampipes.commons.environment.Environments;
import org.apache.streampipes.commons.prometheus.adapter.AdapterMetricsManager;
import
org.apache.streampipes.connect.management.management.AdapterMasterManagement;
import
org.apache.streampipes.connect.management.management.WorkerAdministrationManagement;
@@ -28,6 +29,7 @@ import org.apache.streampipes.health.monitoring.HealthCheck;
import org.apache.streampipes.health.monitoring.PostStartupRecovery;
import org.apache.streampipes.health.monitoring.ResourceProvider;
import org.apache.streampipes.health.monitoring.ServiceHealthCheck;
+import org.apache.streampipes.health.monitoring.ShutdownSignal;
import
org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestManager;
import org.apache.streampipes.manager.execution.PipelineExecutor;
import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceTagPrefix;
@@ -108,12 +110,42 @@ public class PostStartupTask implements Runnable {
@Override
public void run() {
- new ServiceHealthCheck(extensionsServiceStorage,
extensionServiceRequestManager, resourceManager).run();
+ if (isShuttingDown()) {
+ return;
+ }
+ probeRegisteredServices();
performAdapterAssetUpdate();
startAllPreviouslyStoppedPipelines();
runHealthCheckOnce();
}
+ /**
+ * Probes the registered services as often as the service health check needs
consecutive
+ * failures before it declares a service unhealthy. Registrations left
behind by a crashed
+ * run are therefore unhealthy when the health check below runs, and the
pipeline elements
+ * that ran there are restored right away instead of after several scheduled
ticks. Live
+ * services answer every probe, so for them this is cheap.
+ */
+ private void probeRegisteredServices() {
+ var serviceHealthCheck = new ServiceHealthCheck(
+ extensionsServiceStorage, extensionServiceRequestManager,
resourceManager);
+ int probes = Math.max(1,
Environments.getEnvironment().getServiceHealthFailureThreshold().getValueOrDefault());
+ for (int i = 0; i < probes; i++) {
+ if (isShuttingDown()) {
+ return;
+ }
+ serviceHealthCheck.run();
+ if (i < probes - 1) {
+ try {
+ TimeUnit.SECONDS.sleep(1);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return;
+ }
+ }
+ }
+ }
+
private void performAdapterAssetUpdate() {
var installedAppIds = extensionsServiceStorage
.findAll()
@@ -129,20 +161,20 @@ public class PostStartupTask implements Runnable {
postStartupRecovery.checkAndRestore(0);
}
+ /**
+ * Pipelines that are marked as running are deliberately not started here.
They usually
+ * still run on the extension services, which the core did not stop when it
shut down. A
+ * fresh start would allocate endpoints again and, with several services
offering the same
+ * element, could start a second instance next to the running one. The
health check that
+ * follows asks every service which instances it runs and restores only the
missing ones.
+ *
+ * <p>Pipelines flagged with {@code restartOnSystemReboot} were stopped by
an older core
+ * during its shutdown and are not running anywhere, so starting them is
safe.
+ */
private void startAllPreviouslyStoppedPipelines() {
var allPipelines = pipelineStorage.findAll();
- LOG.info("Checking for orphaned pipelines...");
- List<Pipeline> orphanedPipelines = allPipelines
- .stream()
- .filter(Pipeline::isRunning)
- .toList();
-
- LOG.info("Found {} orphaned pipelines", orphanedPipelines.size());
-
- orphanedPipelines.forEach(pipeline -> {
- LOG.info("Restoring orphaned pipeline {}", pipeline.getName());
- startPipeline(pipeline, false);
- });
+ LOG.info("Found {} running pipelines which will be reconciled by the
health check",
+ allPipelines.stream().filter(Pipeline::isRunning).count());
List<Pipeline> pipelinesToRestart = allPipelines
.stream()
@@ -160,6 +192,10 @@ public class PostStartupTask implements Runnable {
}
private void startPipeline(Pipeline pipeline, boolean restartOnReboot) {
+ if (isShuttingDown()) {
+ LOG.info("Shutdown in progress, not restoring pipeline {}",
pipeline.getName());
+ return;
+ }
PipelineOperationStatus status = new PipelineExecutor(pipeline,
extensionServiceRequestManager, resourceManager)
.startPipeline();
if (status.isSuccess()) {
@@ -205,6 +241,10 @@ public class PostStartupTask implements Runnable {
}
}
+ private boolean isShuttingDown() {
+ return ShutdownSignal.INSTANCE.isShutdownRequested();
+ }
+
private IPipelineStorage getPipelineStorage() {
return resourceManager.managePipelines().getDb();
}
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 7ce3b9658c..6edc4ab024 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
@@ -29,16 +29,15 @@ import
org.apache.streampipes.health.monitoring.ExtensionHealthCheck;
import org.apache.streampipes.health.monitoring.HealthCheck;
import org.apache.streampipes.health.monitoring.ResourceProvider;
import org.apache.streampipes.health.monitoring.ServiceHealthCheck;
+import org.apache.streampipes.health.monitoring.ShutdownSignal;
import org.apache.streampipes.loadbalance.LoadManager;
import
org.apache.streampipes.loadbalance.service.ExtensionsServiceReportExecutor;
import
org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestManager;
import
org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestTargets;
import org.apache.streampipes.manager.api.extensions.ExtensionServiceRequests;
-import org.apache.streampipes.manager.function.FunctionManager;
import org.apache.streampipes.manager.health.CoreInitialInstallationProgress;
import org.apache.streampipes.manager.health.CoreServiceStatusManager;
import org.apache.streampipes.manager.pipeline.ExtensionsServiceLogExecutor;
-import org.apache.streampipes.manager.pipeline.PipelineManager;
import org.apache.streampipes.manager.setup.AutoInstallation;
import org.apache.streampipes.manager.setup.InstallationConfiguration;
import org.apache.streampipes.manager.setup.StreamPipesEnvChecker;
@@ -49,8 +48,6 @@ import
org.apache.streampipes.messaging.mqtt.SpMqttProtocolFactory;
import org.apache.streampipes.messaging.nats.SpNatsProtocolFactory;
import org.apache.streampipes.messaging.pulsar.SpPulsarProtocolFactory;
import org.apache.streampipes.model.configuration.SpCoreConfigurationStatus;
-import org.apache.streampipes.model.pipeline.Pipeline;
-import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
import org.apache.streampipes.resource.management.SpResourceManager;
import
org.apache.streampipes.resource.management.permission.SpPermissionEvaluator;
import org.apache.streampipes.service.base.BaseNetworkingConfig;
@@ -62,7 +59,6 @@ import
org.apache.streampipes.service.core.migrations.Migration;
import org.apache.streampipes.service.core.migrations.MigrationsHandler;
import org.apache.streampipes.service.core.storage.StorageApiConfiguration;
import org.apache.streampipes.storage.api.connect.IAdapterStorage;
-import org.apache.streampipes.storage.api.function.IFunctionStateStorage;
import
org.apache.streampipes.storage.api.pipeline.ICompactPipelineTemplateStorage;
import org.apache.streampipes.storage.api.pipeline.IPipelineStorage;
import org.apache.streampipes.storage.api.system.ICertificateStorage;
@@ -80,6 +76,8 @@ import
org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
+import org.springframework.context.event.ContextClosedEvent;
+import org.springframework.context.event.EventListener;
import org.springframework.scheduling.annotation.EnableScheduling;
import jakarta.annotation.PostConstruct;
@@ -89,9 +87,12 @@ import java.io.IOException;
import java.net.UnknownHostException;
import java.time.Duration;
import java.util.List;
+import java.util.Objects;
import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
+import java.util.stream.Stream;
@Configuration
@EnableAutoConfiguration
@@ -108,8 +109,8 @@ public class StreamPipesCoreApplication extends
StreamPipesServiceBase {
private static final Logger LOG =
LoggerFactory.getLogger(StreamPipesCoreApplication.class.getCanonicalName());
- @Autowired
- private IFunctionStateStorage functionStateStorage;
+ private static final Duration EXECUTOR_SHUTDOWN_TIMEOUT =
Duration.ofSeconds(5);
+ private static final Duration EXTENSION_DEREGISTRATION_POLL_INTERVAL =
Duration.ofMillis(250);
@Autowired
protected ICertificateStorage certificateStorage;
@@ -142,6 +143,10 @@ public class StreamPipesCoreApplication extends
StreamPipesServiceBase {
@Qualifier("adapterDescriptionStorage")
private IAdapterStorage adapterDescriptionStorage;
+ private ScheduledExecutorService postStartupExecutor;
+ private ScheduledExecutorService healthCheckExecutor;
+ private ScheduledExecutorService logCheckExecutor;
+
public static void main(String[] args) {
StreamPipesCoreApplication application = new StreamPipesCoreApplication();
application.initialize(() -> List.of(
@@ -178,8 +183,9 @@ public class StreamPipesCoreApplication extends
StreamPipesServiceBase {
@PostConstruct
public void init() {
- var executorService = Executors.newSingleThreadScheduledExecutor();
- var logCheckExecutorService = Executors.newSingleThreadScheduledExecutor();
+ ShutdownSignal.INSTANCE.reset();
+ this.postStartupExecutor = Executors.newSingleThreadScheduledExecutor();
+ this.logCheckExecutor = Executors.newSingleThreadScheduledExecutor();
new
StreamPipesEnvChecker(resourceManager.getCoreConfigurationStorage()).updateEnvironmentVariables();
new CouchDbViewGenerator().createGenericDatabaseIfNotExists();
@@ -221,7 +227,7 @@ public class StreamPipesCoreApplication extends
StreamPipesServiceBase {
env.getHealthCheckIntervalInMillis().getValueOrDefault()
);
- executorService.schedule(new PostStartupTask(
+ postStartupExecutor.schedule(new PostStartupTask(
getPipelineStorage(),
adapterDescriptionStorage,
extensionsServiceStorage,
@@ -259,7 +265,7 @@ public class StreamPipesCoreApplication extends
StreamPipesServiceBase {
var logFetchInterval =
env.getLogFetchIntervalInMillis().getValueOrDefault();
LOG.info("Extensions logs will be fetched every {} seconds",
TimeUnit.MILLISECONDS.toSeconds(logFetchInterval));
- logCheckExecutorService.scheduleAtFixedRate(new
ExtensionsServiceLogExecutor(
+ logCheckExecutor.scheduleAtFixedRate(new ExtensionsServiceLogExecutor(
extensionServiceRequestManager, resourceManager
),
logFetchInterval, logFetchInterval,
@@ -267,11 +273,11 @@ public class StreamPipesCoreApplication extends
StreamPipesServiceBase {
}
private void scheduleHealthChecks(Duration healthCheckInterval,
List<Runnable> checks) {
- var healthCheckExecutorService =
Executors.newSingleThreadScheduledExecutor();
+ this.healthCheckExecutor = Executors.newSingleThreadScheduledExecutor();
checks.forEach(check -> {
LOG.info("Health check {} configured with a delay of {} seconds between
runs", check.getClass().getSimpleName(),
healthCheckInterval.toSeconds());
- healthCheckExecutorService.scheduleWithFixedDelay(check,
healthCheckInterval.toMillis(),
+ healthCheckExecutor.scheduleWithFixedDelay(check,
healthCheckInterval.toMillis(),
healthCheckInterval.toMillis(),
TimeUnit.MILLISECONDS);
});
@@ -307,39 +313,103 @@ public class StreamPipesCoreApplication extends
StreamPipesServiceBase {
}
}
- @PreDestroy
- public void onExit() {
+ /**
+ * Runs at the very beginning of the context shutdown, before the web server
and the
+ * beans are torn down. Pipelines and adapters are intentionally left
running on the
+ * extension services: their state in storage stays "running", so a
restarted core
+ * re-attaches to them through {@link PostStartupTask} and the health
checks, and a
+ * core-only restart causes no processing gap. Extension services clean up
their own
+ * instances when they shut down.
+ *
+ * <p>While this listener runs, the web server still serves requests. The
core uses that
+ * window to let extension services that are stopping at the same time
persist their
+ * function state, see {@link #waitForExtensionServicesToDeregister()}.
+ */
+ @EventListener(ContextClosedEvent.class)
+ public void onShutdownRequested() {
LOG.info("Shutting down StreamPipes...");
- LOG.info("Flagging currently running pipelines for restart...");
- List<Pipeline> pipelinesToStop =
- getAllPipelines().stream().filter(Pipeline::isRunning).toList();
-
- LOG.info("Found {} running pipelines which will be stopped...",
pipelinesToStop.size());
-
- pipelinesToStop.forEach(pipeline -> {
- pipeline.setRestartOnSystemReboot(true);
- resourceManager.managePipelines().getDb().updateElement(pipeline);
- });
+ ShutdownSignal.INSTANCE.requestShutdown();
+ LOG.info("Health checks and recovery tasks are disabled, running pipelines
are left untouched");
+ stopExecutors(Stream.of(postStartupExecutor, healthCheckExecutor,
logCheckExecutor)
+ .filter(Objects::nonNull)
+ .toList());
+ waitForExtensionServicesToDeregister();
+ }
- LOG.info("Gracefully stopping all running pipelines...");
- var pipelineManager = new PipelineManager(resourceManager);
- List<PipelineOperationStatus> status =
pipelineManager.stopAllPipelines(true, extensionServiceRequestManager);
- status.forEach(s -> {
- if (s.isSuccess()) {
- LOG.info("Pipeline {} successfully stopped", s.getPipelineName());
- } else {
- LOG.error("Pipeline {} could not be stopped", s.getPipelineName());
+ /**
+ * Extension services persist their function state through the core's REST
API and
+ * deregister afterwards. Waiting until all of them are gone, bounded by
+ * {@code SP_SHUTDOWN_WAIT_FOR_EXTENSIONS_MS}, keeps that state from being
lost when the
+ * whole stack stops at once. A core-only restart pays the full timeout,
since the
+ * extension services stay registered. This is a stopgap until functions
checkpoint
+ * their state periodically.
+ */
+ private void waitForExtensionServicesToDeregister() {
+ var timeout = Duration.ofMillis(
+
Environments.getEnvironment().getShutdownWaitForExtensionsInMillis().getValueOrDefault());
+ if (timeout.isZero() || timeout.isNegative()) {
+ return;
+ }
+ var deadline = System.nanoTime() + timeout.toNanos();
+ var registered = countRegisteredExtensionServices();
+ if (registered == 0) {
+ return;
+ }
+ LOG.info("Waiting up to {} ms for {} extension services to persist their
state and deregister...",
+ timeout.toMillis(), registered);
+ while (registered > 0 && System.nanoTime() < deadline) {
+ try {
+
TimeUnit.MILLISECONDS.sleep(EXTENSION_DEREGISTRATION_POLL_INTERVAL.toMillis());
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return;
}
- });
+ registered = countRegisteredExtensionServices();
+ }
+ if (registered > 0) {
+ LOG.info("{} extension services are still registered, continuing
shutdown", registered);
+ } else {
+ LOG.info("All extension services deregistered");
+ }
+ }
- new FunctionManager(extensionServiceRequestManager, resourceManager,
extensionsServiceStorage)
- .stopAllFunctionsAndPersistState(functionStateStorage);
+ private long countRegisteredExtensionServices() {
+ try {
+ return extensionsServiceStorage.findAll().size();
+ } catch (RuntimeException e) {
+ LOG.warn("Could not read registered extension services during shutdown:
{}", e.getMessage());
+ return 0;
+ }
+ }
+ @PreDestroy
+ public void onExit() {
LOG.info("Thanks for using Apache StreamPipes - see you next time!");
}
- private List<Pipeline> getAllPipelines() {
- return getPipelineStorage().findAll();
+ /**
+ * Interrupts the executors and waits for running tasks to finish, within
one shared budget
+ * of {@link #EXECUTOR_SHUTDOWN_TIMEOUT}. A health check that is already
past its shutdown
+ * guard may be in the middle of an HTTP request, which does not react to
the interrupt, so
+ * the wait keeps it from touching beans that are being destroyed. The
checks consult
+ * {@link ShutdownSignal} before every recovery side effect. Idle executors
terminate at once.
+ */
+ private void stopExecutors(List<ScheduledExecutorService> executors) {
+ var deadline = System.nanoTime() + EXECUTOR_SHUTDOWN_TIMEOUT.toNanos();
+ executors.forEach(ScheduledExecutorService::shutdownNow);
+ for (var executor : executors) {
+ var remaining = deadline - System.nanoTime();
+ try {
+ if (remaining <= 0 || !executor.awaitTermination(remaining,
TimeUnit.NANOSECONDS)) {
+ LOG.warn("A scheduled task did not finish within {} seconds,
continuing shutdown",
+ EXECUTOR_SHUTDOWN_TIMEOUT.toSeconds());
+ return;
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return;
+ }
+ }
}
private IPipelineStorage getPipelineStorage() {
diff --git
a/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/model/HealthCheckData.java
b/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/ExtensionServiceShutdownHook.java
similarity index 57%
copy from
streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/model/HealthCheckData.java
copy to
streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/ExtensionServiceShutdownHook.java
index 6a0744d10f..794865eb7e 100644
---
a/streampipes-health-monitoring/src/main/java/org/apache/streampipes/health/monitoring/model/HealthCheckData.java
+++
b/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/ExtensionServiceShutdownHook.java
@@ -16,15 +16,19 @@
*
*/
-package org.apache.streampipes.health.monitoring.model;
+package org.apache.streampipes.service.extensions;
-import org.apache.streampipes.health.monitoring.ResourceProvider;
-import org.apache.streampipes.model.health.ExtensionInstanceHealth;
-
-import java.util.Map;
+/**
+ * Cleanup that an extension service runs when it shuts down, in addition to
stopping
+ * the pipeline elements, adapters and functions it knows about. Register
hooks by
+ * overriding {@link
StreamPipesExtensionsServiceBase#getAdditionalShutdownHooks()}.
+ *
+ * <p>Hooks run after the web server and the broker receiver are closed, so
they must
+ * not rely on requests from the core. Exceptions are logged and do not
prevent the
+ * remaining hooks from running.
+ */
+@FunctionalInterface
+public interface ExtensionServiceShutdownHook {
-public record HealthCheckData(ResourceProvider resourceProvider,
- ActiveResources activeResources,
- Map<String, ActiveCoreInstances>
activeCoreInstances,
- Map<String, ExtensionInstanceHealth>
activeExtensionInstances) {
+ void onShutdown();
}
diff --git
a/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/ExtensionsServiceShutdownHandler.java
b/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/ExtensionsServiceShutdownHandler.java
index 8a294525ef..e96f3a9b32 100644
---
a/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/ExtensionsServiceShutdownHandler.java
+++
b/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/ExtensionsServiceShutdownHandler.java
@@ -15,30 +15,189 @@
* limitations under the License.
*
*/
+
package org.apache.streampipes.service.extensions;
+import org.apache.streampipes.client.StreamPipesClient;
+import org.apache.streampipes.commons.exceptions.SpHttpErrorStatusCode;
+import org.apache.streampipes.commons.exceptions.connect.AdapterException;
+import org.apache.streampipes.extensions.api.pe.runtime.IStreamPipesRuntime;
+import
org.apache.streampipes.extensions.management.client.StreamPipesClientResolver;
+import
org.apache.streampipes.extensions.management.connect.AdapterTransitionRegistry;
+import
org.apache.streampipes.extensions.management.connect.AdapterWorkerManagement;
+import org.apache.streampipes.extensions.management.init.DeclarersSingleton;
+import
org.apache.streampipes.extensions.management.init.RunningAdapterInstances;
import org.apache.streampipes.extensions.management.init.RunningInstances;
+import org.apache.streampipes.model.connect.adapter.AdapterDescription;
+import org.apache.streampipes.model.function.FunctionShutdownResult;
+import org.apache.streampipes.model.function.FunctionsShutdownResponse;
+import
org.apache.streampipes.service.extensions.function.StreamPipesFunctionHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.List;
+import java.util.function.Supplier;
+
+/**
+ * Stops everything this extension service is running when the service shuts
down.
+ *
+ * <p>By the time the shutdown handler runs, the service no longer accepts
requests from
+ * the core, so it cannot wait for detach requests. Instead, pipeline
elements, adapters
+ * and functions are stopped locally. Nothing is reported back to the core on
purpose:
+ * pipelines and adapters remain marked as running in storage, and the core's
health
+ * checks re-attach them as soon as this service is registered again. Function
state is
+ * the one thing that is persisted, since it would otherwise be lost. It is
persisted first,
+ * because the core only keeps serving requests for a short while after it
received its own
+ * shutdown signal, and stopping adapters and pipeline elements can take
seconds. Persisting
+ * requires the core's function state endpoints ({@code
SP_FUNCTION_STATE_ENDPOINTS_ENABLED}).
+ *
+ * <p>Services that run instances outside the registries known here, register
an
+ * {@link ExtensionServiceShutdownHook} to stop them.
+ */
public class ExtensionsServiceShutdownHandler {
private static final Logger LOG =
LoggerFactory.getLogger(ExtensionsServiceShutdownHandler.class);
+ private final RunningInstances runningInstances;
+ private final AdapterWorkerManagement adapterManagement;
+ private final Supplier<FunctionsShutdownResponse> functionShutdown;
+ private final Supplier<StreamPipesClient> clientSupplier;
+ private final List<ExtensionServiceShutdownHook> additionalHooks;
+
+ public ExtensionsServiceShutdownHandler() {
+ this(List.of());
+ }
+
+ public ExtensionsServiceShutdownHandler(List<ExtensionServiceShutdownHook>
additionalHooks) {
+ this(
+ RunningInstances.INSTANCE,
+ new AdapterWorkerManagement(
+ RunningAdapterInstances.INSTANCE,
+ DeclarersSingleton.getInstance(),
+ AdapterTransitionRegistry.INSTANCE
+ ),
+ StreamPipesFunctionHandler.INSTANCE::shutdownFunctionsAndGetState,
+ () -> new StreamPipesClientResolver().makeStreamPipesClientInstance(),
+ additionalHooks
+ );
+ }
+
+ ExtensionsServiceShutdownHandler(RunningInstances runningInstances,
+ AdapterWorkerManagement adapterManagement,
+ Supplier<FunctionsShutdownResponse>
functionShutdown,
+ Supplier<StreamPipesClient> clientSupplier,
+ List<ExtensionServiceShutdownHook>
additionalHooks) {
+ this.runningInstances = runningInstances;
+ this.adapterManagement = adapterManagement;
+ this.functionShutdown = functionShutdown;
+ this.clientSupplier = clientSupplier;
+ this.additionalHooks = List.copyOf(additionalHooks);
+ }
public void onShutdown() {
LOG.info("Shutting down StreamPipes extensions service...");
- int runningInstancesCount =
RunningInstances.INSTANCE.getRunningInstancesCount();
+ stopFunctions();
+ stopPipelineElements();
+ stopAdapters();
+ runAdditionalHooks();
+ }
+
+ private void runAdditionalHooks() {
+ additionalHooks.forEach(hook -> {
+ try {
+ hook.onShutdown();
+ } catch (RuntimeException e) {
+ LOG.warn("Shutdown hook {} threw an exception",
hook.getClass().getName(), e);
+ }
+ });
+ }
- while (runningInstancesCount > 0) {
- LOG.info("Waiting for {} running pipeline elements to be stopped...",
runningInstancesCount);
+ private void stopPipelineElements() {
+ var instanceKeys = runningInstances.getRunningInstanceKeys();
+ LOG.info("Stopping {} running pipeline elements...", instanceKeys.size());
+ instanceKeys.forEach(this::stopPipelineElement);
+ }
+
+ private void stopPipelineElement(String instanceKey) {
+ IStreamPipesRuntime<?, ?> runtime =
runningInstances.getInvocation(instanceKey);
+ try {
+ if (runtime != null) {
+ var response = runtime.onRuntimeDetached(instanceKey);
+ if (response != null && !response.isSuccess()) {
+ LOG.warn("Pipeline element {} did not stop cleanly: {}",
instanceKey, response.getOptionalMessage());
+ }
+ }
+ } catch (RuntimeException e) {
+ LOG.warn("Pipeline element {} threw an exception while stopping",
instanceKey, e);
+ } finally {
+ runningInstances.remove(instanceKey);
+ }
+ }
+
+ private void stopAdapters() {
+ List<AdapterDescription> adapters =
List.copyOf(adapterManagement.getAllRunningAdapterInstances());
+ LOG.info("Stopping {} running adapters...", adapters.size());
+ adapters.forEach(adapter -> {
try {
- Thread.sleep(1000);
- } catch (InterruptedException e) {
- LOG.error("Could not pause current thread...");
+ adapterManagement.stopAdapter(adapter);
+ } catch (AdapterException | RuntimeException e) {
+ LOG.warn("Adapter {} threw an exception while stopping",
adapter.getElementId(), e);
+ }
+ });
+ }
+
+ private void stopFunctions() {
+ FunctionsShutdownResponse response;
+ try {
+ response = functionShutdown.get();
+ } catch (RuntimeException e) {
+ LOG.warn("Functions threw an exception while stopping", e);
+ return;
+ }
+ var statefulFunctions = response.getFunctions().stream()
+ .filter(result -> result.getState() != null)
+ .toList();
+ if (statefulFunctions.isEmpty()) {
+ return;
+ }
+ LOG.info("Persisting state of {} functions...", statefulFunctions.size());
+ StreamPipesClient client;
+ try {
+ client = clientSupplier.get();
+ } catch (RuntimeException e) {
+ LOG.warn("Could not persist function state, the core might already be
down: {}", e.getMessage());
+ return;
+ }
+ for (int i = 0; i < statefulFunctions.size(); i++) {
+ if (!persistFunctionState(client, statefulFunctions.get(i))) {
+ LOG.warn("The core is unreachable, skipping state persistence of {}
remaining functions",
+ statefulFunctions.size() - i - 1);
+ return;
+ }
+ }
+ }
+
+ /**
+ * @return false if the core could not be reached at all, so that the caller
does not wait
+ * for a connection timeout per remaining function
+ */
+ private boolean persistFunctionState(StreamPipesClient client,
FunctionShutdownResult result) {
+ try {
+ client.adminApi().persistFunctionState(result.getFunctionId(),
result.getState());
+ return true;
+ } catch (SpHttpErrorStatusCode e) {
+ if (e.getHttpStatusCode() != null && e.getHttpStatusCode() == 404) {
+ LOG.warn("Could not persist state of function {}: the core's function
state endpoints are disabled "
+ + "(set SP_FUNCTION_STATE_ENDPOINTS_ENABLED=true)",
result.getFunctionId());
+ } else {
+ LOG.warn("Could not persist state of function {}: core answered with
status {}",
+ result.getFunctionId(), e.getHttpStatusCode());
}
- runningInstancesCount =
RunningInstances.INSTANCE.getRunningInstancesCount();
+ return true;
+ } catch (RuntimeException e) {
+ LOG.warn("Could not persist state of function {}: {}",
result.getFunctionId(), e.getMessage());
+ return false;
}
}
}
diff --git
a/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/StreamPipesExtensionsServiceBase.java
b/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/StreamPipesExtensionsServiceBase.java
index 049b0a639c..43b89eeaf5 100644
---
a/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/StreamPipesExtensionsServiceBase.java
+++
b/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/StreamPipesExtensionsServiceBase.java
@@ -253,13 +253,31 @@ public abstract class StreamPipesExtensionsServiceBase
extends StreamPipesServic
client.adminApi().registerServiceConfiguration(serviceConfiguration);
}
+ /**
+ * The web server is already stopped when this runs, and the broker receiver
is closed
+ * first, so no further requests from the core can arrive. Running pipeline
elements,
+ * adapters and functions are therefore stopped locally. The core keeps
their state as
+ * running and re-attaches once this service registers again.
+ */
@PreDestroy
public void onExit() {
if (extensionBrokerRequestReceiver != null) {
extensionBrokerRequestReceiver.stop();
}
- new ExtensionsServiceShutdownHandler().onShutdown();
- deregisterService(DeclarersSingleton.getInstance().getServiceId());
+ List<ExtensionServiceShutdownHook> additionalHooks;
+ try {
+ additionalHooks = getAdditionalShutdownHooks();
+ } catch (RuntimeException e) {
+ LOG.warn("Could not collect additional shutdown hooks, continuing
without them", e);
+ additionalHooks = List.of();
+ }
+ new ExtensionsServiceShutdownHandler(additionalHooks).onShutdown();
+ ServiceLoadDataReportGenerator.getInstance().shutdown();
+ try {
+ deregisterService(DeclarersSingleton.getInstance().getServiceId());
+ } catch (RuntimeException e) {
+ LOG.warn("Could not deregister service at core (core might already be
down): {}", e.getMessage());
+ }
}
public String serviceId() {
@@ -270,6 +288,14 @@ public abstract class StreamPipesExtensionsServiceBase
extends StreamPipesServic
return List.of();
}
+ /**
+ * Override to stop instances this service runs outside the standard
pipeline element,
+ * adapter and function registries. Hooks run after those registries are
cleaned up.
+ */
+ protected List<ExtensionServiceShutdownHook> getAdditionalShutdownHooks() {
+ return List.of();
+ }
+
public abstract SpServiceDefinition provideServiceDefinition();
}
diff --git
a/streampipes-service-extensions/src/test/java/org/apache/streampipes/service/extensions/ExtensionsServiceShutdownHandlerTest.java
b/streampipes-service-extensions/src/test/java/org/apache/streampipes/service/extensions/ExtensionsServiceShutdownHandlerTest.java
new file mode 100644
index 0000000000..6aa950f02a
--- /dev/null
+++
b/streampipes-service-extensions/src/test/java/org/apache/streampipes/service/extensions/ExtensionsServiceShutdownHandlerTest.java
@@ -0,0 +1,221 @@
+/*
+ * 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.service.extensions;
+
+import org.apache.streampipes.client.StreamPipesClient;
+import org.apache.streampipes.commons.exceptions.SpHttpErrorStatusCode;
+import org.apache.streampipes.commons.exceptions.connect.AdapterException;
+import org.apache.streampipes.extensions.api.pe.runtime.IStreamPipesRuntime;
+import
org.apache.streampipes.extensions.management.connect.AdapterWorkerManagement;
+import org.apache.streampipes.extensions.management.init.RunningInstances;
+import org.apache.streampipes.model.Response;
+import org.apache.streampipes.model.connect.adapter.AdapterDescription;
+import org.apache.streampipes.model.function.FunctionShutdownResult;
+import org.apache.streampipes.model.function.FunctionsShutdownResponse;
+import org.apache.streampipes.model.graph.DataProcessorInvocation;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.Map;
+import java.util.function.Supplier;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.inOrder;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class ExtensionsServiceShutdownHandlerTest {
+
+ private static final String HEALTHY_INSTANCE = "healthy-instance";
+ private static final String FAILING_INSTANCE = "failing-instance";
+
+ private final RunningInstances runningInstances = RunningInstances.INSTANCE;
+ private AdapterWorkerManagement adapterManagement;
+ private StreamPipesClient client;
+ private Supplier<FunctionsShutdownResponse> functionShutdown;
+
+ @BeforeEach
+ void setUp() {
+ clearRunningInstances();
+ adapterManagement = mock(AdapterWorkerManagement.class);
+
when(adapterManagement.getAllRunningAdapterInstances()).thenReturn(List.of());
+ client = mock(StreamPipesClient.class, RETURNS_DEEP_STUBS);
+ functionShutdown = FunctionsShutdownResponse::new;
+ }
+
+ @AfterEach
+ void tearDown() {
+ clearRunningInstances();
+ }
+
+ @Test
+ void detachesEveryPipelineElementEvenIfOneFails() {
+ var healthy = mock(IStreamPipesRuntime.class);
+ when(healthy.onRuntimeDetached(HEALTHY_INSTANCE)).thenReturn(new
Response(HEALTHY_INSTANCE, true));
+ var failing = mock(IStreamPipesRuntime.class);
+ when(failing.onRuntimeDetached(FAILING_INSTANCE)).thenThrow(new
IllegalStateException("boom"));
+ runningInstances.add(FAILING_INSTANCE, invocation(FAILING_INSTANCE),
failing);
+ runningInstances.add(HEALTHY_INSTANCE, invocation(HEALTHY_INSTANCE),
healthy);
+
+ handler().onShutdown();
+
+ verify(healthy).onRuntimeDetached(HEALTHY_INSTANCE);
+ verify(failing).onRuntimeDetached(FAILING_INSTANCE);
+ assertEquals(0, runningInstances.getRunningInstancesCount());
+ }
+
+ @Test
+ void stopsEveryAdapterEvenIfOneFails() throws AdapterException {
+ var failingAdapter = adapter("failing-adapter");
+ var healthyAdapter = adapter("healthy-adapter");
+
when(adapterManagement.getAllRunningAdapterInstances()).thenReturn(List.of(failingAdapter,
healthyAdapter));
+ doThrow(new
AdapterException("boom")).when(adapterManagement).stopAdapter(failingAdapter);
+
+ handler().onShutdown();
+
+ verify(adapterManagement).stopAdapter(failingAdapter);
+ verify(adapterManagement).stopAdapter(healthyAdapter);
+ }
+
+ @Test
+ void persistsStateOfStatefulFunctionsOnly() {
+ var state = Map.<String, Object>of("counter", 42);
+ functionShutdown = () -> new FunctionsShutdownResponse(List.of(
+ new FunctionShutdownResult("stateful", state),
+ new FunctionShutdownResult("stateless", null)
+ ));
+
+ handler().onShutdown();
+
+ verify(client.adminApi()).persistFunctionState("stateful", state);
+ verify(client.adminApi(), never()).persistFunctionState("stateless", null);
+ }
+
+ @Test
+ void unreachableCoreStopsPersistenceEarlyButNotLocalCleanup() {
+ var runtime = mock(IStreamPipesRuntime.class);
+ when(runtime.onRuntimeDetached(HEALTHY_INSTANCE)).thenReturn(new
Response(HEALTHY_INSTANCE, true));
+ runningInstances.add(HEALTHY_INSTANCE, invocation(HEALTHY_INSTANCE),
runtime);
+ var adapter = adapter("adapter");
+
when(adapterManagement.getAllRunningAdapterInstances()).thenReturn(List.of(adapter));
+ functionShutdown = () -> new FunctionsShutdownResponse(List.of(
+ new FunctionShutdownResult("first", Map.of("counter", 1)),
+ new FunctionShutdownResult("second", Map.of("counter", 2))
+ ));
+ var admin = client.adminApi();
+ doThrow(new IllegalStateException("connection
refused")).when(admin).persistFunctionState(any(), any());
+
+ assertDoesNotThrow(() -> handler().onShutdown());
+
+ verify(admin, times(1)).persistFunctionState(any(), any());
+ verify(runtime).onRuntimeDetached(HEALTHY_INSTANCE);
+ assertEquals(0, runningInstances.getRunningInstancesCount());
+ }
+
+ @Test
+ void endpointErrorsDoNotStopPersistenceOfOtherFunctions() {
+ functionShutdown = () -> new FunctionsShutdownResponse(List.of(
+ new FunctionShutdownResult("first", Map.of("counter", 1)),
+ new FunctionShutdownResult("second", Map.of("counter", 2))
+ ));
+ var admin = client.adminApi();
+ doThrow(new SpHttpErrorStatusCode("404",
404)).when(admin).persistFunctionState(any(), any());
+
+ handler().onShutdown();
+
+ verify(admin, times(2)).persistFunctionState(any(), any());
+ }
+
+ @Test
+ void toleratesMissingClient() {
+ functionShutdown = () -> new FunctionsShutdownResponse(List.of(
+ new FunctionShutdownResult("stateful", Map.of("counter", 1))
+ ));
+ var withoutClient = new ExtensionsServiceShutdownHandler(runningInstances,
adapterManagement, functionShutdown,
+ () -> {
+ throw new IllegalStateException("no client");
+ }, List.of());
+
+ assertDoesNotThrow(withoutClient::onShutdown);
+ }
+
+ @Test
+ void persistsFunctionStateBeforeStoppingPipelineElements() {
+ var runtime = mock(IStreamPipesRuntime.class);
+ when(runtime.onRuntimeDetached(HEALTHY_INSTANCE)).thenReturn(new
Response(HEALTHY_INSTANCE, true));
+ runningInstances.add(HEALTHY_INSTANCE, invocation(HEALTHY_INSTANCE),
runtime);
+ var state = Map.<String, Object>of("counter", 7);
+ functionShutdown = () -> new FunctionsShutdownResponse(List.of(new
FunctionShutdownResult("stateful", state)));
+ var admin = client.adminApi();
+
+ handler().onShutdown();
+
+ var order = inOrder(admin, runtime);
+ order.verify(admin).persistFunctionState("stateful", state);
+ order.verify(runtime).onRuntimeDetached(HEALTHY_INSTANCE);
+ }
+
+ @Test
+ void runsEveryAdditionalHookEvenIfOneFails() {
+ var failing = mock(ExtensionServiceShutdownHook.class);
+ doThrow(new IllegalStateException("boom")).when(failing).onShutdown();
+ var healthy = mock(ExtensionServiceShutdownHook.class);
+
+ handler(List.of(failing, healthy)).onShutdown();
+
+ verify(failing).onShutdown();
+ verify(healthy).onShutdown();
+ }
+
+ private ExtensionsServiceShutdownHandler handler() {
+ return handler(List.of());
+ }
+
+ private ExtensionsServiceShutdownHandler
handler(List<ExtensionServiceShutdownHook> hooks) {
+ return new ExtensionsServiceShutdownHandler(runningInstances,
adapterManagement, functionShutdown, () -> client,
+ hooks);
+ }
+
+ private DataProcessorInvocation invocation(String elementId) {
+ var invocation = new DataProcessorInvocation();
+ invocation.setElementId(elementId);
+ invocation.setAppId("app-" + elementId);
+ return invocation;
+ }
+
+ private AdapterDescription adapter(String elementId) {
+ var description = new AdapterDescription();
+ description.setElementId(elementId);
+ return description;
+ }
+
+ private void clearRunningInstances() {
+
runningInstances.getRunningInstanceKeys().forEach(runningInstances::remove);
+ }
+}
