This is an automated email from the ASF dual-hosted git repository. dominikriemer pushed a commit to branch fix-startup-recovery in repository https://gitbox.apache.org/repos/asf/streampipes.git
commit 7cbda6e3b8b66acf44a51cc6245aee7eac9e6203 Author: Dominik Riemer <[email protected]> AuthorDate: Sat Oct 31 14:32:48 2026 +0100 feat: Make startup recovery more robust --- 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); + } +}
