This is an automated email from the ASF dual-hosted git repository.
riemer pushed a commit to branch 2135-max-health-check-intervals-configurable
in repository https://gitbox.apache.org/repos/asf/streampipes.git
The following commit(s) were added to
refs/heads/2135-max-health-check-intervals-configurable by this push:
new 11585ca20 feat(#2135): Max health check intervals configurable
11585ca20 is described below
commit 11585ca20cc500e7bcfb04a146ba1249362eb8ce
Author: Dominik Riemer <[email protected]>
AuthorDate: Fri Nov 3 22:48:58 2023 +0100
feat(#2135): Max health check intervals configurable
---
.../apache/streampipes/commons/constants/Envs.java | 12 ++++-
.../commons/environment/DefaultEnvironment.java | 25 ++++++++++
.../commons/environment/Environment.java | 12 +++++
.../manager/health/ServiceHealthCheck.java | 13 +++--
.../service/core/StreamPipesCoreApplication.java | 57 ++++++++++++----------
5 files changed, 87 insertions(+), 32 deletions(-)
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 077347861..bdbcb4105 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
@@ -85,7 +85,17 @@ public enum Envs {
SP_PROMETHEUS_ENDPOINT_INCLUDE("SP_PROMETHEUS_ENDPOINT_INCLUDE",
"health,prometheus"),
- SP_SETUP_PROMETHEUS_ENDPOINT("SP_SETUP_PROMETHEUS_ENDPOINT", "false");
+ SP_SETUP_PROMETHEUS_ENDPOINT("SP_SETUP_PROMETHEUS_ENDPOINT", "false"),
+
+ SP_HEALTH_CHECK_INTERVAL_MS("SP_HEALTH_CHECK_INTERVAL_MS", "30000"),
+
+ SP_HEALTH_CHECK_INITIAL_DELAY_MS("SP_HEALTH_CHECK_INITIAL_DELAY", "10000"),
+
+ SP_LOG_FETCH_INTERVAL_MS("SP_LOG_FETCH_INTERVAL_MS", "60000"),
+
+
SP_HEALTH_SERVICE_MAX_UNHEALTHY_TIME_MS("SP_HEALTH_SERVICE_MAX_UNHEALTHY_TIME_MS",
"60000"),
+
+
SP_INITIAL_WAIT_BEFORE_INSTALLATION_MS("SP_INITIAL_WAIT_BEFORE_INSTALLATION_MS",
"5000");
private final String envVariableName;
private String defaultValue;
diff --git
a/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/DefaultEnvironment.java
b/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/DefaultEnvironment.java
index c6ac2a3ed..3a5831ae8 100644
---
a/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/DefaultEnvironment.java
+++
b/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/DefaultEnvironment.java
@@ -224,6 +224,31 @@ public class DefaultEnvironment implements Environment {
return new BooleanEnvironmentVariable(Envs.SP_SETUP_PROMETHEUS_ENDPOINT);
}
+ @Override
+ public IntEnvironmentVariable getHealthCheckIntervalInMillis() {
+ return new IntEnvironmentVariable(Envs.SP_HEALTH_CHECK_INTERVAL_MS);
+ }
+
+ @Override
+ public IntEnvironmentVariable getInitialHealthCheckDelayInMillis() {
+ return new IntEnvironmentVariable(Envs.SP_HEALTH_CHECK_INITIAL_DELAY_MS);
+ }
+
+ @Override
+ public IntEnvironmentVariable getLogFetchIntervalInMillis() {
+ return new IntEnvironmentVariable(Envs.SP_LOG_FETCH_INTERVAL_MS);
+ }
+
+ @Override
+ public IntEnvironmentVariable
getUnhealthyTimeBeforeServiceDeletionInMillis() {
+ return new
IntEnvironmentVariable(Envs.SP_HEALTH_SERVICE_MAX_UNHEALTHY_TIME_MS);
+ }
+
+ @Override
+ public IntEnvironmentVariable getInitialWaitTimeBeforeInstallationInMillis()
{
+ return new
IntEnvironmentVariable(Envs.SP_INITIAL_WAIT_BEFORE_INSTALLATION_MS);
+ }
+
@Override
public StringEnvironmentVariable getConsulLocation() {
return new StringEnvironmentVariable(Envs.SP_CONSUL_LOCATION);
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 98662652a..30c6749c2 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
@@ -119,4 +119,16 @@ public interface Environment {
StringEnvironmentVariable getPrometheusEndpointInclude();
BooleanEnvironmentVariable getSetupPrometheusEndpoint();
+
+ // Health checks and logging
+ IntEnvironmentVariable getHealthCheckIntervalInMillis();
+
+ IntEnvironmentVariable getInitialHealthCheckDelayInMillis();
+
+ IntEnvironmentVariable getLogFetchIntervalInMillis();
+
+ IntEnvironmentVariable getUnhealthyTimeBeforeServiceDeletionInMillis();
+
+ IntEnvironmentVariable getInitialWaitTimeBeforeInstallationInMillis();
+
}
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/health/ServiceHealthCheck.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/health/ServiceHealthCheck.java
index 88c9292ae..ce434908d 100644
---
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/health/ServiceHealthCheck.java
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/health/ServiceHealthCheck.java
@@ -19,6 +19,7 @@
package org.apache.streampipes.manager.health;
+import org.apache.streampipes.commons.environment.Environments;
import org.apache.streampipes.manager.execution.ExtensionServiceExecutions;
import
org.apache.streampipes.model.extensions.svcdiscovery.SpServiceRegistration;
import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceStatus;
@@ -35,13 +36,15 @@ public class ServiceHealthCheck implements Runnable {
private static final Logger LOG =
LoggerFactory.getLogger(ServiceHealthCheck.class);
- private static final int MAX_UNHEALTHY_DURATION_BEFORE_REMOVAL_MS = 60000;
-
private final ServiceRegistrationManager serviceRegistrationManager;
+ private final int maxUnhealthyDurationBeforeRemovalMs;
public ServiceHealthCheck() {
var storage =
StorageDispatcher.INSTANCE.getNoSqlStore().getExtensionsServiceStorage();
this.serviceRegistrationManager = new ServiceRegistrationManager(storage);
+ this.maxUnhealthyDurationBeforeRemovalMs = Environments
+ .getEnvironment()
+ .getUnhealthyTimeBeforeServiceDeletionInMillis().getValueOrDefault();
}
@Override
@@ -80,15 +83,15 @@ public class ServiceHealthCheck implements Runnable {
System.currentTimeMillis());
}
if (shouldDeleteService(service)) {
- LOG.info("Removing service {} which has been unhealthy for more than {}
seconds.",
- service.getSvcId(), MAX_UNHEALTHY_DURATION_BEFORE_REMOVAL_MS / 1000);
+ LOG.info("Removing service {} which has been unhealthy for more than {}
milliseconds.",
+ service.getSvcId(), maxUnhealthyDurationBeforeRemovalMs);
serviceRegistrationManager.removeService(service.getSvcId());
}
}
private boolean shouldDeleteService(SpServiceRegistration service) {
var currentTimeMillis = System.currentTimeMillis();
- return (currentTimeMillis - service.getFirstTimeSeenUnhealthy() >
MAX_UNHEALTHY_DURATION_BEFORE_REMOVAL_MS);
+ return (currentTimeMillis - service.getFirstTimeSeenUnhealthy() >
maxUnhealthyDurationBeforeRemovalMs);
}
private String makeHealthCheckUrl(SpServiceRegistration service) {
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 9038afa4c..aeaaa7524 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
@@ -17,6 +17,7 @@
*/
package org.apache.streampipes.service.core;
+import org.apache.streampipes.commons.environment.Environments;
import org.apache.streampipes.connect.management.health.AdapterHealthCheck;
import org.apache.streampipes.manager.health.CoreInitialInstallationProgress;
import org.apache.streampipes.manager.health.CoreServiceStatusManager;
@@ -74,12 +75,6 @@ public class StreamPipesCoreApplication extends
StreamPipesServiceBase {
private static final Logger LOG =
LoggerFactory.getLogger(StreamPipesCoreApplication.class.getCanonicalName());
- private static final int LOG_FETCH_INTERVAL = 60;
- private static final TimeUnit LOG_FETCH_UNIT = TimeUnit.SECONDS;
-
- private static final int HEALTH_CHECK_INTERVAL = 30;
- private static final TimeUnit HEALTH_CHECK_UNIT = TimeUnit.SECONDS;
-
private final ISpCoreConfigurationStorage coreConfigStorage =
StorageDispatcher.INSTANCE
.getNoSqlStore().getSpCoreConfigurationStorage();
@@ -120,10 +115,11 @@ public class StreamPipesCoreApplication extends
StreamPipesServiceBase {
new StreamPipesEnvChecker().updateEnvironmentVariables();
new CouchDbViewGenerator().createGenericDatabaseIfNotExists();
+ var env = Environments.getEnvironment();
if (!isConfigured()) {
CoreInitialInstallationProgress.INSTANCE.triggerInitiallyInstallingMode();
- doInitialSetup();
+
doInitialSetup(env.getInitialWaitTimeBeforeInstallationInMillis().getValueOrDefault());
} else {
// Check needs to be present since core configuration is part of
migration
if (coreConfigStorage.exists()) {
@@ -133,32 +129,41 @@ public class StreamPipesCoreApplication extends
StreamPipesServiceBase {
}
coreStatusManager.updateCoreStatus(SpCoreConfigurationStatus.READY);
- executorService.schedule(new PostStartupTask(getPipelineStorage()), 10,
TimeUnit.SECONDS);
-
- scheduleHealthChecks(List.of(
- new ServiceHealthCheck(),
- new PipelineHealthCheck(),
- new AdapterHealthCheck()));
-
- LOG.info("Extensions logs will be fetched every {} seconds",
LOG_FETCH_INTERVAL);
+ executorService.schedule(
+ new PostStartupTask(getPipelineStorage()),
+ env.getInitialHealthCheckDelayInMillis().getValueOrDefault(),
+ TimeUnit.MILLISECONDS);
+
+ scheduleHealthChecks(
+ env.getHealthCheckIntervalInMillis().getValueOrDefault(),
+ List.of(
+ new ServiceHealthCheck(),
+ new PipelineHealthCheck(),
+ new AdapterHealthCheck())
+ );
+
+ var logFetchInterval =
env.getLogFetchIntervalInMillis().getValueOrDefault();
+ LOG.info("Extensions logs will be fetched every {} milliseconds",
logFetchInterval);
logCheckExecutorService.scheduleAtFixedRate(new
ExtensionsServiceLogExecutor(),
- LOG_FETCH_INTERVAL,
- LOG_FETCH_INTERVAL,
- LOG_FETCH_UNIT);
+ logFetchInterval,
+ logFetchInterval,
+ TimeUnit.MILLISECONDS);
}
- private void scheduleHealthChecks(List<Runnable> checks) {
+ private void scheduleHealthChecks(
+ int healthCheckIntervalInMillis,
+ List<Runnable> checks) {
var healthCheckExecutorService =
Executors.newSingleThreadScheduledExecutor();
checks.forEach(check -> {
LOG.info(
"Health check {} configured to run every {} {}",
check.getClass().getCanonicalName(),
- HEALTH_CHECK_INTERVAL,
- HEALTH_CHECK_UNIT);
+ healthCheckIntervalInMillis,
+ TimeUnit.MILLISECONDS);
healthCheckExecutorService.scheduleAtFixedRate(check,
- HEALTH_CHECK_INTERVAL,
- HEALTH_CHECK_INTERVAL,
- HEALTH_CHECK_UNIT);
+ healthCheckIntervalInMillis,
+ healthCheckIntervalInMillis,
+ TimeUnit.MILLISECONDS);
});
}
@@ -166,12 +171,12 @@ public class StreamPipesCoreApplication extends
StreamPipesServiceBase {
return new UserStorage().existsDatabase();
}
- private void doInitialSetup() {
+ private void doInitialSetup(int initialSleepBeforeInstallation) {
LOG.info("\n\n**********\n\nWelcome to Apache
StreamPipes!\n\n**********\n\n");
LOG.info("We will perform the initial setup, grab some coffee and cross
your fingers ;-)...");
LOG.info("Auto-setup will start in 5 seconds to make sure all services are
running...");
try {
- TimeUnit.SECONDS.sleep(5);
+ TimeUnit.MILLISECONDS.sleep(initialSleepBeforeInstallation);
LOG.info("Starting installation procedure");
new AutoInstallation().startAutoInstallation();
} catch (InterruptedException e) {
