This is an automated email from the ASF dual-hosted git repository. riemer pushed a commit to branch fix-startup in repository https://gitbox.apache.org/repos/asf/streampipes.git
commit 40085c70ff5f4da3b4e164f87e1394a372bc03c9 Author: Dominik Riemer <[email protected]> AuthorDate: Fri Nov 3 18:13:43 2023 +0100 improvement: Optimize startup behaviour --- .../streampipes/config/backend/BackendConfig.java | 6 ---- .../health/CoreInitialInstallationProgress.java | 36 ++++++---------------- .../manager/health/ServiceHealthCheck.java | 8 +++-- .../manager/setup/AutoInstallation.java | 2 -- .../org/apache/streampipes/rest/impl/Setup.java | 15 ++++++--- .../rest/impl/admin/MigrationResource.java | 14 ++++++--- .../service/core/StreamPipesCoreApplication.java | 9 +++--- .../svcdiscovery/SpServiceDiscoveryCore.java | 2 +- .../streampipes/storage/api/IUserStorage.java | 2 ++ .../storage/couchdb/impl/UserStorage.java | 13 ++++++++ .../streampipes/storage/couchdb/utils/Utils.java | 17 ++++++++-- 11 files changed, 70 insertions(+), 54 deletions(-) diff --git a/streampipes-config/src/main/java/org/apache/streampipes/config/backend/BackendConfig.java b/streampipes-config/src/main/java/org/apache/streampipes/config/backend/BackendConfig.java index f07366814..7937866ca 100644 --- a/streampipes-config/src/main/java/org/apache/streampipes/config/backend/BackendConfig.java +++ b/streampipes-config/src/main/java/org/apache/streampipes/config/backend/BackendConfig.java @@ -38,8 +38,6 @@ public enum BackendConfig { config = SpServiceDiscovery.getSpConfig("backend"); config.register(BackendConfigKeys.IS_CONFIGURED, false, "Boolean that indicates whether streampipes is " + "already configured or not"); - config.register(BackendConfigKeys.IS_SETUP_RUNNING, false, - "Boolean that indicates whether the initial setup " + "is currently running"); } public String getJmsHost() { @@ -128,8 +126,4 @@ public enum BackendConfig { private String makeDefaultJwtSecret() { return TokenGenerator.generateNewToken(); } - - public void updateSetupStatus(boolean status) { - config.setBoolean(BackendConfigKeys.IS_SETUP_RUNNING, status); - } } diff --git a/streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/IUserStorage.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/health/CoreInitialInstallationProgress.java similarity index 52% copy from streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/IUserStorage.java copy to streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/health/CoreInitialInstallationProgress.java index 1757e2468..d71acc728 100644 --- a/streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/IUserStorage.java +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/health/CoreInitialInstallationProgress.java @@ -15,36 +15,20 @@ * limitations under the License. * */ -package org.apache.streampipes.storage.api; -import org.apache.streampipes.model.client.user.Principal; -import org.apache.streampipes.model.client.user.ServiceAccount; -import org.apache.streampipes.model.client.user.UserAccount; +package org.apache.streampipes.manager.health; -import java.util.List; +public enum CoreInitialInstallationProgress { -public interface IUserStorage { + INSTANCE; - List<Principal> getAllUsers(); + private boolean initiallyInstalling = false; - List<UserAccount> getAllUserAccounts(); - - List<ServiceAccount> getAllServiceAccounts(); - - Principal getUser(String username); - - UserAccount getUserAccount(String username); - - ServiceAccount getServiceAccount(String username); - - void storeUser(Principal user); - - void updateUser(Principal user); - - boolean checkUser(String username); - - void deleteUser(String principalId); - - Principal getUserById(String principalId); + public boolean isInitiallyInstalling() { + return initiallyInstalling; + } + public void triggerInitiallyInstallingMode() { + this.initiallyInstalling = true; + } } 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 46d2792ea..88c9292ae 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 @@ -35,7 +35,7 @@ public class ServiceHealthCheck implements Runnable { private static final Logger LOG = LoggerFactory.getLogger(ServiceHealthCheck.class); - private static final int MAX_UNHEALTHY_DURATION_BEFORE_REMOVAL_MS = 20000; + private static final int MAX_UNHEALTHY_DURATION_BEFORE_REMOVAL_MS = 60000; private final ServiceRegistrationManager serviceRegistrationManager; @@ -56,7 +56,7 @@ public class ServiceHealthCheck implements Runnable { try { var request = ExtensionServiceExecutions.extServiceGetRequest(healthCheckUrl); var response = request.execute(); - if (response.returnResponse().getStatusLine().getStatusCode() != HttpStatus.SC_OK) { + if (response.returnResponse().getStatusLine().getStatusCode() != HttpStatus.SC_OK && !isStarting(service)) { processUnhealthyService(service); } else { if (service.getStatus() == SpServiceStatus.UNHEALTHY) { @@ -68,6 +68,10 @@ public class ServiceHealthCheck implements Runnable { } } + private boolean isStarting(SpServiceRegistration service) { + return service.getStatus() == SpServiceStatus.REGISTERED || service.getStatus() == SpServiceStatus.MIGRATING; + } + private void processUnhealthyService(SpServiceRegistration service) { if (service.getStatus() == SpServiceStatus.HEALTHY) { serviceRegistrationManager.applyServiceStatus( diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/setup/AutoInstallation.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/setup/AutoInstallation.java index 578231701..f7e14125a 100644 --- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/setup/AutoInstallation.java +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/setup/AutoInstallation.java @@ -20,7 +20,6 @@ package org.apache.streampipes.manager.setup; import org.apache.streampipes.commons.environment.Environment; import org.apache.streampipes.commons.environment.Environments; import org.apache.streampipes.commons.environment.variable.StringEnvironmentVariable; -import org.apache.streampipes.config.backend.BackendConfig; import org.apache.streampipes.model.client.setup.InitialSettings; import org.slf4j.Logger; @@ -121,7 +120,6 @@ public class AutoInstallation implements BackgroundTaskNotifier { if (errorCount.get() > 0) { LOG.error("{} errors occurred during the setup process", errorCount); } else { - BackendConfig.INSTANCE.setIsConfigured(true); LOG.info("Initial setup completed successfully - you can now open the login page in the browser."); } LOG.info("\n\n**********\n\nAuto-Setup finished\n\n**********\n\n"); diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/Setup.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/Setup.java index 0874680c8..ea249e6f2 100644 --- a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/Setup.java +++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/Setup.java @@ -19,8 +19,10 @@ package org.apache.streampipes.rest.impl; -import org.apache.streampipes.config.backend.BackendConfig; +import org.apache.streampipes.manager.health.CoreServiceStatusManager; import org.apache.streampipes.rest.core.base.impl.AbstractRestResource; +import org.apache.streampipes.storage.api.ISpCoreConfigurationStorage; +import org.apache.streampipes.storage.management.StorageDispatcher; import com.google.gson.JsonObject; import io.swagger.v3.oas.annotations.Operation; @@ -34,6 +36,9 @@ import jakarta.ws.rs.core.Response; @Path("/v2/setup") public class Setup extends AbstractRestResource { + private final ISpCoreConfigurationStorage storage = StorageDispatcher + .INSTANCE.getNoSqlStore().getSpCoreConfigurationStorage(); + @GET @Path("/configured") @Produces(MediaType.APPLICATION_JSON) @@ -41,13 +46,13 @@ public class Setup extends AbstractRestResource { tags = {"Configurated"}) public Response isConfigured() { JsonObject obj = new JsonObject(); - if (BackendConfig.INSTANCE.isConfigured()) { + var statusManager = new CoreServiceStatusManager(storage); + if (statusManager.isCoreReady()) { obj.addProperty("configured", true); - return ok(obj.toString()); } else { obj.addProperty("configured", false); - obj.addProperty("setupRunning", BackendConfig.INSTANCE.isConfigured()); - return ok(obj.toString()); + obj.addProperty("setupRunning", false); } + return ok(obj.toString()); } } diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/admin/MigrationResource.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/admin/MigrationResource.java index 6b28afe0a..4d2d66202 100644 --- a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/admin/MigrationResource.java +++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/admin/MigrationResource.java @@ -18,8 +18,8 @@ package org.apache.streampipes.rest.impl.admin; -import org.apache.streampipes.config.backend.BackendConfig; import org.apache.streampipes.connect.management.management.AdapterMigrationManager; +import org.apache.streampipes.manager.health.CoreInitialInstallationProgress; import org.apache.streampipes.manager.health.CoreServiceStatusManager; import org.apache.streampipes.manager.health.ServiceRegistrationManager; import org.apache.streampipes.manager.migration.AdapterDescriptionMigration093; @@ -104,11 +104,17 @@ public class MigrationResource extends AbstractAuthGuardedRestResource { var serviceManager = new ServiceRegistrationManager(extensionsServiceStorage); var extensionsServiceConfig = serviceManager.getService(serviceId); - if (BackendConfig.INSTANCE.isConfigured()) { + if (!CoreInitialInstallationProgress.INSTANCE.isInitiallyInstalling()) { new AdapterDescriptionMigration093(adapterDescriptionStorage).reinstallAdapters(extensionsServiceConfig); if (!migrationConfigs.isEmpty()) { - if (serviceManager.isAnyServiceMigrating() || !isCoreReady()) { - LOG.info("Refusing migration request since precondition is not met."); + var anyServiceMigrating = serviceManager.isAnyServiceMigrating(); + var coreReady = isCoreReady(); + if (anyServiceMigrating || !coreReady) { + LOG.info( + "Refusing migration request since precondition is not met (anyServiceMigratione={}, coreReady={}.", + anyServiceMigrating, + coreReady + ); return Response.status(HttpStatus.SC_CONFLICT).build(); } else { serviceManager.applyServiceStatus(serviceId, SpServiceStatus.MIGRATING); 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 49b218f73..9038afa4c 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,8 +17,8 @@ */ package org.apache.streampipes.service.core; -import org.apache.streampipes.config.backend.BackendConfig; import org.apache.streampipes.connect.management.health.AdapterHealthCheck; +import org.apache.streampipes.manager.health.CoreInitialInstallationProgress; import org.apache.streampipes.manager.health.CoreServiceStatusManager; import org.apache.streampipes.manager.health.PipelineHealthCheck; import org.apache.streampipes.manager.health.ServiceHealthCheck; @@ -41,6 +41,7 @@ import org.apache.streampipes.service.base.StreamPipesServiceBase; import org.apache.streampipes.service.core.migrations.MigrationsHandler; import org.apache.streampipes.storage.api.IPipelineStorage; import org.apache.streampipes.storage.api.ISpCoreConfigurationStorage; +import org.apache.streampipes.storage.couchdb.impl.UserStorage; import org.apache.streampipes.storage.couchdb.utils.CouchDbViewGenerator; import org.apache.streampipes.storage.management.StorageDispatcher; @@ -121,6 +122,7 @@ public class StreamPipesCoreApplication extends StreamPipesServiceBase { new CouchDbViewGenerator().createGenericDatabaseIfNotExists(); if (!isConfigured()) { + CoreInitialInstallationProgress.INSTANCE.triggerInitiallyInstallingMode(); doInitialSetup(); } else { // Check needs to be present since core configuration is part of migration @@ -161,20 +163,17 @@ public class StreamPipesCoreApplication extends StreamPipesServiceBase { } private boolean isConfigured() { - return BackendConfig.INSTANCE.isConfigured(); + return new UserStorage().existsDatabase(); } private void doInitialSetup() { 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 ;-)..."); - - BackendConfig.INSTANCE.updateSetupStatus(true); LOG.info("Auto-setup will start in 5 seconds to make sure all services are running..."); try { TimeUnit.SECONDS.sleep(5); LOG.info("Starting installation procedure"); new AutoInstallation().startAutoInstallation(); - BackendConfig.INSTANCE.updateSetupStatus(false); } catch (InterruptedException e) { LOG.error("Ooops, something went wrong during the installation", e); } diff --git a/streampipes-service-discovery/src/main/java/org/apache/streampipes/svcdiscovery/SpServiceDiscoveryCore.java b/streampipes-service-discovery/src/main/java/org/apache/streampipes/svcdiscovery/SpServiceDiscoveryCore.java index 0455dc1d8..9d6851289 100644 --- a/streampipes-service-discovery/src/main/java/org/apache/streampipes/svcdiscovery/SpServiceDiscoveryCore.java +++ b/streampipes-service-discovery/src/main/java/org/apache/streampipes/svcdiscovery/SpServiceDiscoveryCore.java @@ -89,7 +89,7 @@ public class SpServiceDiscoveryCore implements ISpServiceDiscovery { TimeUnit.MILLISECONDS.sleep(1000); return findService(retryCount); } catch (InterruptedException e) { - e.printStackTrace(); + LOG.warn("Could not find a service currently due to exception {}", e.getMessage()); return Collections.emptyList(); } } else { diff --git a/streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/IUserStorage.java b/streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/IUserStorage.java index 1757e2468..cd4e0581f 100644 --- a/streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/IUserStorage.java +++ b/streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/IUserStorage.java @@ -47,4 +47,6 @@ public interface IUserStorage { Principal getUserById(String principalId); + boolean existsDatabase(); + } diff --git a/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/UserStorage.java b/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/UserStorage.java index b2d3c75c2..677aa2186 100644 --- a/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/UserStorage.java +++ b/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/UserStorage.java @@ -25,9 +25,12 @@ import org.apache.streampipes.storage.api.IUserStorage; import org.apache.streampipes.storage.couchdb.dao.CrudViewDao; import org.apache.streampipes.storage.couchdb.utils.Utils; +import org.apache.http.client.methods.HttpGet; +import org.lightcouch.NoDocumentException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.net.URI; import java.util.List; import java.util.stream.Collectors; @@ -117,4 +120,14 @@ public class UserStorage extends CrudViewDao implements IUserStorage { return findWithNullIfEmpty(principalId, Principal.class); } + @Override + public boolean existsDatabase() { + var client = Utils.getCouchDbClient(Utils.USER_DB_NAME, false); + try { + client.executeRequest(new HttpGet(URI.create(client.getBaseUri() + "/" + Utils.USER_DB_NAME))); + return true; + } catch (NoDocumentException e) { + return false; + } + } } diff --git a/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/utils/Utils.java b/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/utils/Utils.java index b9b48c303..401001d9d 100644 --- a/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/utils/Utils.java +++ b/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/utils/Utils.java @@ -32,6 +32,8 @@ import org.lightcouch.CouchDbProperties; public class Utils { + public static final String USER_DB_NAME = "users"; + public static CouchDbClient getCouchDbDataProcessorDescriptionClient() { return getCouchDbGsonClient("data-processor"); } @@ -135,7 +137,7 @@ public class Utils { } public static CouchDbClient getCouchDbUserClient() { - return getCouchDbPrincipalClient("users"); + return getCouchDbPrincipalClient(USER_DB_NAME); } public static CouchDbClient getCouchDbInternalUsersClient() { @@ -188,15 +190,20 @@ public class Utils { return new CouchDbClient(props(dbname)); } + public static CouchDbClient getCouchDbClient(String database, boolean createIfNotExists) { + return new CouchDbClient(props(database, createIfNotExists)); + } + public static CouchDbClient getCouchDbClient(String database) { return new CouchDbClient(props(database)); } - private static CouchDbProperties props(String dbname) { + private static CouchDbProperties props(String dbname, + boolean createDbIfNotExists) { var env = getEnvironment(); return new CouchDbProperties( dbname, - true, + createDbIfNotExists, env.getCouchDbProtocol().getValueOrDefault(), env.getCouchDbHost().getValueOrDefault(), env.getCouchDbPort().getValueOrDefault(), @@ -204,6 +211,10 @@ public class Utils { env.getCouchDbPassword().getValueOrDefault()); } + private static CouchDbProperties props(String dbname) { + return props(dbname, true); + } + public static String getDatabaseRoute(String databaseName) { return toUrl() + "/" + databaseName; }
