This is an automated email from the ASF dual-hosted git repository.
bossenti pushed a commit to branch
2045-implement-migration-for-adapter-and-pipeline-element-configurations
in repository https://gitbox.apache.org/repos/asf/streampipes.git
The following commit(s) were added to
refs/heads/2045-implement-migration-for-adapter-and-pipeline-element-configurations
by this push:
new 10375f3bb Improvements to extension migration (#2101)
10375f3bb is described below
commit 10375f3bb029e75091b4f034b5bebf647fcd48cb
Author: Dominik Riemer <[email protected]>
AuthorDate: Mon Oct 30 07:50:38 2023 +0100
Improvements to extension migration (#2101)
* Extract MigrationResource logic into smaller units
* Use single request for submitting migrations from extensions to core
* refactor: remove dead code
* style: change formatting
---------
Co-authored-by: bossenti <[email protected]>
---
.../apache/streampipes/client/api/IAdminApi.java | 4 +-
.../apache/streampipes/client/api/AdminApi.java | 18 +-
.../management/AdapterMigrationManager.java | 113 +++++
.../migration/AbstractMigrationManager.java | 155 +++++++
.../manager/migration/IMigrationHandler.java | 26 +-
.../manager/migration/MigrationUtils.java | 73 +++
.../migration/PipelineElementMigrationManager.java | 249 ++++++++++
.../manager/migration/MigrationUtilsTest.java | 12 +-
.../rest/impl/admin/MigrationResource.java | 510 ++-------------------
.../extensions/ExtensionsModelSubmitter.java | 26 +-
10 files changed, 645 insertions(+), 541 deletions(-)
diff --git
a/streampipes-client-api/src/main/java/org/apache/streampipes/client/api/IAdminApi.java
b/streampipes-client-api/src/main/java/org/apache/streampipes/client/api/IAdminApi.java
index 35047ca9e..ba776ce51 100644
---
a/streampipes-client-api/src/main/java/org/apache/streampipes/client/api/IAdminApi.java
+++
b/streampipes-client-api/src/main/java/org/apache/streampipes/client/api/IAdminApi.java
@@ -40,9 +40,7 @@ public interface IAdminApi {
void deregisterFunction(String functionId);
- void registerAdapterMigrations(List<ModelMigratorConfig> migrationConfigs,
String serviceId);
-
- void registerPipelineElementMigrations(List<ModelMigratorConfig>
migratorConfigs, String serviceId);
+ void registerMigrations(List<ModelMigratorConfig> migrationConfigs, String
serviceId);
MessagingSettings getMessagingSettings();
}
diff --git
a/streampipes-client/src/main/java/org/apache/streampipes/client/api/AdminApi.java
b/streampipes-client/src/main/java/org/apache/streampipes/client/api/AdminApi.java
index 0670d9448..970eb6224 100644
---
a/streampipes-client/src/main/java/org/apache/streampipes/client/api/AdminApi.java
+++
b/streampipes-client/src/main/java/org/apache/streampipes/client/api/AdminApi.java
@@ -72,12 +72,8 @@ public class AdminApi extends AbstractClientApi implements
IAdminApi {
* @param migrationConfigs list of migration configs to be registered
*/
@Override
- public void registerAdapterMigrations(List<ModelMigratorConfig>
migrationConfigs, String serviceId) {
- post(getAdapterMigrationPath().addToPath(serviceId), migrationConfigs);
- }
-
- public void registerPipelineElementMigrations(List<ModelMigratorConfig>
migratorConfigs, String serviceId) {
- post(getPipelineElementMigrationPath().addToPath(serviceId),
migratorConfigs);
+ public void registerMigrations(List<ModelMigratorConfig> migrationConfigs,
String serviceId) {
+ post(getMigrationPath().addToPath(serviceId), migrationConfigs);
}
@Override
@@ -113,15 +109,9 @@ public class AdminApi extends AbstractClientApi implements
IAdminApi {
return getFunctionsPath().addToPath(functionId);
}
- private StreamPipesApiPath getAdapterMigrationPath() {
- return StreamPipesApiPath
- .fromBaseApiPath()
- .addToPath("migrations/adapter");
- }
-
- private StreamPipesApiPath getPipelineElementMigrationPath() {
+ private StreamPipesApiPath getMigrationPath() {
return StreamPipesApiPath
.fromBaseApiPath()
- .addToPath("migrations/pipeline-element");
+ .addToPath("migrations");
}
}
diff --git
a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/AdapterMigrationManager.java
b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/AdapterMigrationManager.java
new file mode 100644
index 000000000..26f158a5b
--- /dev/null
+++
b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/AdapterMigrationManager.java
@@ -0,0 +1,113 @@
+/*
+ * 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.connect.management.management;
+
+import org.apache.streampipes.commons.exceptions.connect.AdapterException;
+import org.apache.streampipes.manager.migration.AbstractMigrationManager;
+import org.apache.streampipes.manager.migration.IMigrationHandler;
+import
org.apache.streampipes.model.extensions.svcdiscovery.SpServiceRegistration;
+import org.apache.streampipes.model.migration.ModelMigratorConfig;
+import org.apache.streampipes.storage.api.IAdapterStorage;
+
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+
+public class AdapterMigrationManager extends AbstractMigrationManager
implements IMigrationHandler {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(AdapterMigrationManager.class);
+
+ private final IAdapterStorage adapterStorage;
+
+ public AdapterMigrationManager(IAdapterStorage adapterStorage) {
+ this.adapterStorage = adapterStorage;
+ }
+
+ @Override
+ public void handleMigrations(SpServiceRegistration extensionsServiceConfig,
+ List<ModelMigratorConfig> migrationConfigs) {
+
+ LOG.info("Received {} migrations from extension service {}.",
+ migrationConfigs.size(),
+ extensionsServiceConfig.getServiceUrl());
+ LOG.info("Updating adapter descriptions by replacement...");
+ updateDescriptions(migrationConfigs,
extensionsServiceConfig.getServiceUrl());
+ LOG.info("Adapter descriptions are up to date.");
+
+ LOG.info("Checking migrations for existing adapters in StreamPipes Core
...");
+ for (var migrationConfig : migrationConfigs) {
+ LOG.info("Searching for assets of '{}'", migrationConfig.targetAppId());
+ LOG.debug("Searching for assets of '{}' with config {}",
migrationConfig.targetAppId(), migrationConfig);
+ var adapterDescriptions =
adapterStorage.getAdaptersByAppId(migrationConfig.targetAppId());
+ LOG.info("Found {} instances for appId '{}'",
adapterDescriptions.size(), migrationConfig.targetAppId());
+ for (var adapterDescription : adapterDescriptions) {
+
+ var adapterVersion = adapterDescription.getVersion();
+
+ if (adapterVersion == migrationConfig.fromVersion()) {
+ LOG.info("Migration is required for adapter '{}'. Migrating from
version '{}' to '{}' ...",
+ adapterDescription.getElementId(),
+ adapterVersion, migrationConfig.toVersion()
+ );
+
+ var migrationResult = performMigration(
+ adapterDescription,
+ migrationConfig,
+ String.format("%s/%s/adapter",
+ extensionsServiceConfig.getServiceUrl(),
+ MIGRATION_ENDPOINT
+ )
+ );
+
+ if (migrationResult.success()) {
+ LOG.info("Migration successfully performed by extensions service.
Updating adapter description ...");
+ LOG.debug(
+ "Migration was performed by extensions service '{}'",
+ extensionsServiceConfig.getServiceUrl());
+
+ adapterStorage.updateAdapter(migrationResult.element());
+ LOG.info("Adapter description is updated - Migration successfully
completed at Core.");
+ } else {
+ LOG.error("Migration failed with the following reason: {}",
migrationResult.message());
+ LOG.error(
+ "Migration for adapter '{}' failed - Stopping adapter ...",
+ migrationResult.element().getElementId()
+ );
+ try {
+
WorkerRestClient.stopStreamAdapter(extensionsServiceConfig.getServiceUrl(),
adapterDescription);
+ } catch (AdapterException e) {
+ LOG.error("Stopping adapter failed: {}",
StringUtils.join(e.getStackTrace(), "\n"));
+ }
+ LOG.info("Adapter successfully stopped.");
+ }
+ } else {
+ LOG.info(
+ "Migration is not applicable for adapter '{}' because of a
version mismatch - "
+ + "adapter version: '{}', migration starts at: '{}'",
+ adapterDescription.getElementId(),
+ adapterVersion,
+ migrationConfig.fromVersion()
+ );
+ }
+ }
+ }
+ }
+}
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/AbstractMigrationManager.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/AbstractMigrationManager.java
new file mode 100644
index 000000000..3641f0b7c
--- /dev/null
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/AbstractMigrationManager.java
@@ -0,0 +1,155 @@
+/*
+ * 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.manager.migration;
+
+import org.apache.streampipes.commons.exceptions.SepaParseException;
+import org.apache.streampipes.manager.endpoint.HttpJsonParser;
+import org.apache.streampipes.manager.execution.ExtensionServiceExecutions;
+import org.apache.streampipes.manager.operations.Operations;
+import org.apache.streampipes.model.base.VersionedNamedStreamPipesEntity;
+import org.apache.streampipes.model.extensions.migration.MigrationRequest;
+import org.apache.streampipes.model.message.Notification;
+import org.apache.streampipes.model.migration.MigrationResult;
+import org.apache.streampipes.model.migration.ModelMigratorConfig;
+import org.apache.streampipes.serializers.json.JacksonSerializer;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.URI;
+import java.util.List;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import static
org.apache.streampipes.manager.migration.MigrationUtils.getRequestUrl;
+
+public abstract class AbstractMigrationManager {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(AbstractMigrationManager.class);
+
+ protected static final String MIGRATION_ENDPOINT = "api/v1/migrations";
+
+ /**
+ * Performs the actual migration of a pipeline element.
+ * This includes the communication with the extensions service which runs
the migration.
+ *
+ * @param pipelineElement pipeline element to be migrated
+ * @param migrationConfig config of the migration to be performed
+ * @param url url of the migration endpoint at the extensions
service
+ * where the migration should be performed
+ * @param <T> type of the processing element
+ * @return result of the migration
+ */
+ protected <T extends VersionedNamedStreamPipesEntity> MigrationResult<T>
performMigration(
+ T pipelineElement,
+ ModelMigratorConfig migrationConfig,
+ String url
+ ) {
+
+ try {
+
+ var migrationRequest = new MigrationRequest<>(pipelineElement,
migrationConfig);
+
+ String serializedRequest =
JacksonSerializer.getObjectMapper().writeValueAsString(migrationRequest);
+
+ var migrationResponse = ExtensionServiceExecutions.extServicePostRequest(
+ url,
+ serializedRequest
+ ).execute();
+
+ TypeReference<MigrationResult<T>> typeReference = new TypeReference<>() {
+ };
+
+ return JacksonSerializer
+ .getObjectMapper()
+ .readValue(migrationResponse.returnContent().asString(),
typeReference);
+ } catch (JsonProcessingException e) {
+ LOG.error(
+ "Migration of pipeline element failed before sending to the
extensions service, "
+ + "pipeline element is not migrated. Serialization of migration
request failed: {}",
+ StringUtils.join(e.getStackTrace(), "\n")
+ );
+ } catch (IOException e) {
+ LOG.error("Migration of pipeline element failed at the extensions
service, pipeline element is not migrated: {}.",
+ StringUtils.join(e.getStackTrace(), "\n")
+ );
+ }
+ return MigrationResult.failure(pipelineElement, "Internal error during
migration at StreamPipes Core");
+ }
+
+ /**
+ * Update all descriptions of entities in the Core that are affected by
migrations.
+ *
+ * @param migrationConfigs List of migrations to take in account
+ * @param serviceUrl Url of the extension service that provides the
migrations.
+ */
+ protected void updateDescriptions(List<ModelMigratorConfig>
migrationConfigs, String serviceUrl) {
+ migrationConfigs
+ .stream()
+ .collect(
+ // We only need to update the description once per appId,
+ // because this is directly done with the newest version of the
description and
+ // there is iterative migration required.
+ // To avoid unnecessary, multiple updates,
+ // we filter the migration configs such that every appId is unique.
+ // This ensures that every description is only updated once.
+ Collectors.toMap(
+ ModelMigratorConfig::targetAppId,
+ Function.identity(),
+ (existing, replacement) -> existing
+ )
+ )
+ .values()
+ .stream()
+ .peek(config -> {
+ var requestUrl = getRequestUrl(config.modelType(),
config.targetAppId(), serviceUrl);
+ performUpdate(requestUrl);
+ })
+ .toList();
+ }
+
+ /**
+ * Perform the update of the description based on the given requestUrl
+ *
+ * @param requestUrl URl that references the description to be updated at
the extensions service.
+ */
+ protected void performUpdate(String requestUrl) {
+
+ try {
+ var entityPayload =
HttpJsonParser.getContentFromUrl(URI.create(requestUrl));
+ var updateResult = Operations.verifyAndUpdateElement(entityPayload);
+ if (!updateResult.isSuccess()) {
+ LOG.error(
+ "Updating the pipeline element description failed: {}",
+ StringUtils.join(
+
updateResult.getNotifications().stream().map(Notification::toString).toList(),
+ "\n")
+ );
+ }
+ } catch (IOException | SepaParseException e) {
+ LOG.error("Updating the pipeline element description failed due to the
following exception:\n{}",
+ StringUtils.join(e.getStackTrace(), "\n")
+ );
+ }
+ }
+}
diff --git
a/streampipes-client-api/src/main/java/org/apache/streampipes/client/api/IAdminApi.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/IMigrationHandler.java
similarity index 51%
copy from
streampipes-client-api/src/main/java/org/apache/streampipes/client/api/IAdminApi.java
copy to
streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/IMigrationHandler.java
index 35047ca9e..7847e2432 100644
---
a/streampipes-client-api/src/main/java/org/apache/streampipes/client/api/IAdminApi.java
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/IMigrationHandler.java
@@ -16,33 +16,15 @@
*
*/
-package org.apache.streampipes.client.api;
+package org.apache.streampipes.manager.migration;
-import org.apache.streampipes.model.configuration.MessagingSettings;
-import
org.apache.streampipes.model.extensions.configuration.SpServiceConfiguration;
import
org.apache.streampipes.model.extensions.svcdiscovery.SpServiceRegistration;
-import org.apache.streampipes.model.function.FunctionDefinition;
import org.apache.streampipes.model.migration.ModelMigratorConfig;
import java.util.List;
-public interface IAdminApi {
+public interface IMigrationHandler {
- void registerService(SpServiceRegistration serviceRegistration);
-
- void deregisterService(String serviceId);
-
- void registerServiceConfiguration(SpServiceConfiguration
serviceConfiguration);
-
- SpServiceConfiguration getServiceConfiguration(String serviceGroup);
-
- void registerFunctions(List<FunctionDefinition> functions);
-
- void deregisterFunction(String functionId);
-
- void registerAdapterMigrations(List<ModelMigratorConfig> migrationConfigs,
String serviceId);
-
- void registerPipelineElementMigrations(List<ModelMigratorConfig>
migratorConfigs, String serviceId);
-
- MessagingSettings getMessagingSettings();
+ void handleMigrations(SpServiceRegistration serviceRegistration,
+ List<ModelMigratorConfig> migrationConfigs);
}
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/MigrationUtils.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/MigrationUtils.java
new file mode 100644
index 000000000..8a96c3c1f
--- /dev/null
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/MigrationUtils.java
@@ -0,0 +1,73 @@
+/*
+ * 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.manager.migration;
+
+import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
+import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceTagPrefix;
+import org.apache.streampipes.model.migration.ModelMigratorConfig;
+import org.apache.streampipes.svcdiscovery.api.model.SpServiceUrlProvider;
+
+import java.util.List;
+import java.util.Optional;
+
+public class MigrationUtils {
+
+ /**
+ * Filter the application migration for a pipeline definition.
+ * By definition, there is only one migration config that fulfills the
requirements.
+ * Otherwise, it should have been detected as duplicate by the extensions
service.
+ *
+ * @param pipelineElement pipeline element that should be migrated
+ * @param migrationConfigs available migration configs to pick the
applicable from
+ * @return config that is applicable for the given pipeline element
+ */
+ public static Optional<ModelMigratorConfig> getApplicableMigration(
+ InvocableStreamPipesEntity pipelineElement,
+ List<ModelMigratorConfig> migrationConfigs
+ ) {
+ return migrationConfigs
+ .stream()
+ .filter(
+ config ->
config.modelType().equals(pipelineElement.getServiceTagPrefix())
+ && config.targetAppId().equals(pipelineElement.getAppId())
+ && config.fromVersion() == pipelineElement.getVersion()
+ )
+ .findFirst();
+ }
+
+ /**
+ * Get the URL that provides the description for an entity.
+ *
+ * @param entityType Type of the entity to be updated.
+ * @param appId AppId of the entity to be updated
+ * @param serviceUrl URL of the extensions service to which the entity
belongs
+ * @return URL of the endpoint that provides the description for the given
entity
+ */
+ public static String getRequestUrl(SpServiceTagPrefix entityType, String
appId, String serviceUrl) {
+
+ SpServiceUrlProvider urlProvider;
+ switch (entityType) {
+ case ADAPTER -> urlProvider = SpServiceUrlProvider.ADAPTER;
+ case DATA_PROCESSOR -> urlProvider = SpServiceUrlProvider.DATA_PROCESSOR;
+ case DATA_SINK -> urlProvider = SpServiceUrlProvider.DATA_SINK;
+ default -> throw new RuntimeException("Unexpected instance type.");
+ }
+ return urlProvider.getInvocationUrl(serviceUrl, appId);
+ }
+}
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/PipelineElementMigrationManager.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/PipelineElementMigrationManager.java
new file mode 100644
index 000000000..28338e1ac
--- /dev/null
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/PipelineElementMigrationManager.java
@@ -0,0 +1,249 @@
+/*
+ * 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.manager.migration;
+
+import org.apache.streampipes.manager.execution.PipelineExecutor;
+import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
+import
org.apache.streampipes.model.extensions.svcdiscovery.SpServiceRegistration;
+import org.apache.streampipes.model.graph.DataProcessorInvocation;
+import org.apache.streampipes.model.graph.DataSinkInvocation;
+import org.apache.streampipes.model.migration.MigrationResult;
+import org.apache.streampipes.model.migration.ModelMigratorConfig;
+import org.apache.streampipes.model.pipeline.Pipeline;
+import org.apache.streampipes.model.pipeline.PipelineHealthStatus;
+import org.apache.streampipes.model.staticproperty.StaticProperty;
+import org.apache.streampipes.storage.api.IDataProcessorStorage;
+import org.apache.streampipes.storage.api.IDataSinkStorage;
+import org.apache.streampipes.storage.api.IPipelineStorage;
+
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static
org.apache.streampipes.manager.migration.MigrationUtils.getApplicableMigration;
+
+public class PipelineElementMigrationManager extends AbstractMigrationManager
implements IMigrationHandler {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(PipelineElementMigrationManager.class);
+
+ private final IPipelineStorage pipelineStorage;
+ private final IDataProcessorStorage dataProcessorStorage;
+ private final IDataSinkStorage dataSinkStorage;
+
+ public PipelineElementMigrationManager(IPipelineStorage pipelineStorage,
+ IDataProcessorStorage
dataProcessorStorage,
+ IDataSinkStorage dataSinkStorage) {
+ this.pipelineStorage = pipelineStorage;
+ this.dataProcessorStorage = dataProcessorStorage;
+ this.dataSinkStorage = dataSinkStorage;
+ }
+
+ @Override
+ public void handleMigrations(SpServiceRegistration extensionsServiceConfig,
+ List<ModelMigratorConfig> migrationConfigs) {
+
+ LOG.info("Updating pipeline element descriptions by replacement...");
+ updateDescriptions(migrationConfigs,
extensionsServiceConfig.getServiceUrl());
+ LOG.info("Pipeline element descriptions are up to date.");
+
+ LOG.info("Received {} pipeline element migrations from extension service
{}.",
+ migrationConfigs.size(),
+ extensionsServiceConfig.getServiceUrl());
+ var availablePipelines = pipelineStorage.getAllPipelines();
+ if (!availablePipelines.isEmpty()) {
+ LOG.info("Found {} available pipelines. Checking pipelines for
applicable migrations...",
+ availablePipelines.size()
+ );
+ }
+
+ for (var pipeline : availablePipelines) {
+ List<MigrationResult<?>> failedMigrations = new ArrayList<>();
+
+ var migratedDataProcessors = pipeline.getSepas()
+ .stream()
+ .map(processor -> {
+ if (getApplicableMigration(processor,
migrationConfigs).isPresent()) {
+ return migratePipelineElement(
+ processor,
+ migrationConfigs,
+ String.format("%s/%s/processor",
+ extensionsServiceConfig.getServiceUrl(),
+ MIGRATION_ENDPOINT
+ ),
+ failedMigrations
+ );
+ } else {
+ LOG.info("No migration applicable for data processor '{}'.",
processor.getElementId());
+ return processor;
+ }
+ })
+ .toList();
+ pipeline.setSepas(migratedDataProcessors);
+
+ var migratedDataSinks = pipeline.getActions()
+ .stream()
+ .map(sink -> {
+ if (getApplicableMigration(sink, migrationConfigs).isPresent()) {
+ return migratePipelineElement(
+ sink,
+ migrationConfigs,
+ String.format("%s/%s/sink",
+ extensionsServiceConfig.getServiceUrl(),
+ MIGRATION_ENDPOINT
+ ),
+ failedMigrations
+ );
+ } else {
+ LOG.info("No migration applicable for data sink '{}'.",
sink.getElementId());
+ return sink;
+ }
+ })
+ .toList();
+ pipeline.setActions(migratedDataSinks);
+
+ pipelineStorage.updatePipeline(pipeline);
+
+ if (failedMigrations.isEmpty()) {
+ LOG.info("Migration for pipeline successfully completed.");
+ } else {
+ // pass most recent version of pipeline
+
handleFailedMigrations(pipelineStorage.getPipeline(pipeline.getPipelineId()),
failedMigrations);
+ }
+ }
+ }
+
+ /**
+ * Takes care about the failed migrations of pipeline elements.
+ * This includes the following steps:
+ * <ul>
+ * <li> logging of failed pipeline elements
+ * <li> setting migration results as pipeline notifications
+ * <li> updating pipeline health status
+ * <li> stopping the pipeline
+ * </ul>
+ *
+ * @param pipeline the pipeline affected by failed migrations
+ * @param failedMigrations the list of failed migrations
+ */
+ protected void handleFailedMigrations(Pipeline pipeline,
List<MigrationResult<?>> failedMigrations) {
+ LOG.error("Failures in migration detected - The following pipeline
elements could to be migrated:\n"
+ +
StringUtils.join(failedMigrations.stream().map(Record::toString).toList()),
"\n");
+
+ pipeline.setPipelineNotifications(failedMigrations.stream().map(
+ failedMigration -> "Failed migration of pipeline element:
%s".formatted(failedMigration.message())
+ ).toList());
+ pipeline.setHealthStatus(PipelineHealthStatus.REQUIRES_ATTENTION);
+
+ pipelineStorage.updatePipeline(pipeline);
+
+ // get updated version of pipeline after modification
+ pipeline = pipelineStorage.getPipeline(pipeline.getPipelineId());
+
+ stopPipeline(pipeline);
+ }
+
+
+ public void stopPipeline(Pipeline pipeline) {
+ var pipelineExecutor = new PipelineExecutor(pipeline, true);
+ var pipelineStopResult = pipelineExecutor.stopPipeline();
+
+ if (pipelineStopResult.isSuccess()) {
+ LOG.info("Pipeline successfully stopped.");
+ } else {
+ LOG.error("Pipeline stop failed.");
+ }
+ }
+
+ /**
+ * Handle the migration of a pipeline element with respect to the given
model migration configs.
+ * All applicable migrations found in the provided configs are executed for
the given pipeline element.
+ * In case a migration fails, the related pipeline element receives the
latest definition of its static properties,
+ * so that the pipeline element can be adapted by the user to resolve the
failed migration.
+ *
+ * @param pipelineElement pipeline element to be migrated
+ * @param modelMigrations list of model migrations that might be applicable
for this pipeline element
+ * @param url url of the extensions service endpoint that
handles the migration
+ * @param failedMigrations collection of failed migrations which is extended
by occurring migration failures
+ * @param <T> type of the pipeline element (e.g.,
DataProcessorInvocation)
+ * @return the migrated (or - in case of a failure - updated) pipeline
element
+ */
+ protected <T extends InvocableStreamPipesEntity> T migratePipelineElement(
+ T pipelineElement,
+ List<ModelMigratorConfig> modelMigrations,
+ String url,
+ List<MigrationResult<?>> failedMigrations
+ ) {
+
+ // loop until no migrations are available anymore
+ // this allows to apply multiple migrations for a pipeline element
sequentially
+ // For example, first migration from 0 to 1 and the second migration from
1 to 2
+ while (getApplicableMigration(pipelineElement,
modelMigrations).isPresent() && failedMigrations.isEmpty()) {
+
+ var migrationConfig = getApplicableMigration(pipelineElement,
modelMigrations).get();
+ LOG.info(
+ "Found applicable migration for pipeline element '{}': {}",
+ pipelineElement.getElementId(),
+ migrationConfig
+ );
+
+ var migrationResult = performMigration(
+ pipelineElement,
+ migrationConfig,
+ url
+ );
+
+ if (migrationResult.success()) {
+ LOG.info("Migration successfully performed by extensions service.
Updating pipeline element invocation ...");
+ LOG.debug("Migration was performed at extensions service endpoint
'{}'", url);
+ pipelineElement = migrationResult.element();
+ } else {
+ LOG.error("Migration failed with the following reason: {}",
migrationResult.message());
+ failedMigrations.add(migrationResult);
+ }
+ }
+ if (!failedMigrations.isEmpty()) {
+ updateFailedPipelineElement(pipelineElement);
+ LOG.info("Updated pipeline elements with new description where automatic
migration failed.");
+ }
+ return pipelineElement;
+ }
+
+ /**
+ * Update the static properties of the failed pipeline element with its
description.
+ * This allows to adapt the failed pipeline element in the UI to overcome
the failed migration.
+ *
+ * @param pipelineElement pipeline element with failed migration
+ */
+ protected void updateFailedPipelineElement(InvocableStreamPipesEntity
pipelineElement) {
+ List<StaticProperty> updatedStaticProperties = new ArrayList<>();
+ if (pipelineElement instanceof DataProcessorInvocation) {
+ updatedStaticProperties = dataProcessorStorage
+ .getFirstDataProcessorByAppId(pipelineElement.getAppId())
+ .getStaticProperties();
+ } else if (pipelineElement instanceof DataSinkInvocation) {
+ updatedStaticProperties = dataSinkStorage
+ .getFirstDataSinkByAppId(pipelineElement.getAppId())
+ .getStaticProperties();
+ }
+ pipelineElement.setStaticProperties(updatedStaticProperties);
+ }
+}
diff --git
a/streampipes-rest/src/test/java/org/apache/streampipes/rest/impl/admin/MigrationResourceTest.java
b/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/migration/MigrationUtilsTest.java
similarity index 83%
rename from
streampipes-rest/src/test/java/org/apache/streampipes/rest/impl/admin/MigrationResourceTest.java
rename to
streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/migration/MigrationUtilsTest.java
index 63f0b92a7..05545db19 100644
---
a/streampipes-rest/src/test/java/org/apache/streampipes/rest/impl/admin/MigrationResourceTest.java
+++
b/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/migration/MigrationUtilsTest.java
@@ -16,7 +16,7 @@
*
*/
-package org.apache.streampipes.rest.impl.admin;
+package org.apache.streampipes.manager.migration;
import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceTagPrefix;
import org.apache.streampipes.model.graph.DataProcessorInvocation;
@@ -30,7 +30,7 @@ import java.util.List;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
-public class MigrationResourceTest {
+public class MigrationUtilsTest {
List<ModelMigratorConfig> migrationConfigs = List.of(
new ModelMigratorConfig("app-id",
SpServiceTagPrefix.DATA_PROCESSOR, 0, 1),
@@ -60,18 +60,18 @@ public class MigrationResourceTest {
assertEquals(
migrationConfigs.get(0),
- MigrationResource.getApplicableMigration(pipelineElement1,
migrationConfigs).get()
+ MigrationUtils.getApplicableMigration(pipelineElement1,
migrationConfigs).get()
);
assertEquals(
migrationConfigs.get(1),
- MigrationResource.getApplicableMigration(pipelineElement2,
migrationConfigs).get()
+ MigrationUtils.getApplicableMigration(pipelineElement2,
migrationConfigs).get()
);
assertEquals(
migrationConfigs.get(2),
- MigrationResource.getApplicableMigration(pipelineElement3,
migrationConfigs).get()
+ MigrationUtils.getApplicableMigration(pipelineElement3,
migrationConfigs).get()
);
assertTrue(
- MigrationResource.getApplicableMigration(pipelineElement4,
migrationConfigs).isEmpty()
+ MigrationUtils.getApplicableMigration(pipelineElement4,
migrationConfigs).isEmpty()
);
}
}
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 7342bcc21..13c87875f 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,45 +18,24 @@
package org.apache.streampipes.rest.impl.admin;
-import org.apache.streampipes.commons.exceptions.SepaParseException;
-import org.apache.streampipes.commons.exceptions.connect.AdapterException;
-import org.apache.streampipes.connect.management.management.WorkerRestClient;
-import org.apache.streampipes.manager.execution.ExtensionServiceExecutions;
-import org.apache.streampipes.manager.execution.PipelineExecutor;
-import org.apache.streampipes.manager.operations.Operations;
-import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
-import org.apache.streampipes.model.base.VersionedNamedStreamPipesEntity;
-import org.apache.streampipes.model.extensions.migration.MigrationRequest;
+import
org.apache.streampipes.connect.management.management.AdapterMigrationManager;
+import
org.apache.streampipes.manager.migration.PipelineElementMigrationManager;
import
org.apache.streampipes.model.extensions.svcdiscovery.SpServiceRegistration;
import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceTagPrefix;
-import org.apache.streampipes.model.graph.DataProcessorInvocation;
-import org.apache.streampipes.model.graph.DataSinkInvocation;
-import org.apache.streampipes.model.message.Notification;
-import org.apache.streampipes.model.migration.MigrationResult;
import org.apache.streampipes.model.migration.ModelMigratorConfig;
-import org.apache.streampipes.model.pipeline.Pipeline;
-import org.apache.streampipes.model.pipeline.PipelineHealthStatus;
-import org.apache.streampipes.model.staticproperty.StaticProperty;
import
org.apache.streampipes.rest.core.base.impl.AbstractAuthGuardedRestResource;
import org.apache.streampipes.rest.security.AuthConstants;
-import org.apache.streampipes.serializers.json.JacksonSerializer;
import org.apache.streampipes.storage.api.CRUDStorage;
import org.apache.streampipes.storage.api.IAdapterStorage;
import org.apache.streampipes.storage.api.IDataProcessorStorage;
import org.apache.streampipes.storage.api.IDataSinkStorage;
import org.apache.streampipes.storage.api.IPipelineStorage;
-import org.apache.streampipes.svcdiscovery.api.model.SpServiceUrlProvider;
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.core.type.TypeReference;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.Parameter;
import io.swagger.v3.oas.annotations.enums.ParameterIn;
import io.swagger.v3.oas.annotations.responses.ApiResponse;
-import org.apache.commons.lang3.StringUtils;
import org.apache.http.HttpStatus;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import org.springframework.security.access.prepost.PreAuthorize;
import org.springframework.stereotype.Component;
@@ -67,24 +46,15 @@ import jakarta.ws.rs.PathParam;
import jakarta.ws.rs.core.MediaType;
import jakarta.ws.rs.core.Response;
-import java.io.IOException;
-import java.net.URISyntaxException;
-import java.util.ArrayList;
import java.util.List;
-import java.util.Optional;
-import java.util.function.Function;
-import java.util.stream.Collectors;
@Path("v2/migrations")
@Component
@PreAuthorize(AuthConstants.IS_ADMIN_ROLE)
public class MigrationResource extends AbstractAuthGuardedRestResource {
- private static final Logger LOG =
LoggerFactory.getLogger(MigrationResource.class);
- private static final String MIGRATION_ENDPOINT = "api/v1/migrations";
-
private final CRUDStorage<String, SpServiceRegistration>
extensionsServiceStorage =
- getNoSqlStorage().getExtensionsServiceStorage();
+ getNoSqlStorage().getExtensionsServiceStorage();
private final IAdapterStorage adapterStorage =
getNoSqlStorage().getAdapterInstanceStorage();
@@ -94,457 +64,51 @@ public class MigrationResource extends
AbstractAuthGuardedRestResource {
private final IPipelineStorage pipelineStorage =
getNoSqlStorage().getPipelineStorageAPI();
@POST
- @Path("adapter/{serviceId}")
+ @Path("{serviceId}")
@Consumes(MediaType.APPLICATION_JSON)
@Operation(
- summary = "Migrate adapters based on migration configs", tags =
{"Core", "Migration"},
- responses = {
- @ApiResponse(
- responseCode = "" + HttpStatus.SC_OK,
- description = "All provided migrations are handled. If
an error appeared, "
- + "the corresponding actions are taken.")
- }
- )
- public Response registerAdapterMigrations(
- @Parameter(
- in = ParameterIn.PATH,
- description = "the id of the extensions service that
requests migrations",
- required = true
- )
- @PathParam("serviceId") String serviceId,
- @Parameter(
- description = "list of configs (ModelMigratorConfig) that
describe the requested migrations",
- required = true
- )
- List<ModelMigratorConfig> migrationConfigs) {
-
- var extensionsServiceConfig =
extensionsServiceStorage.getElementById(serviceId);
-
- LOG.info("Received {} migrations from extension service {}.",
- migrationConfigs.size(),
- extensionsServiceConfig.getServiceUrl());
- LOG.info("Updating adapter descriptions by replacement...");
- updateDescriptions(migrationConfigs,
extensionsServiceConfig.getServiceUrl());
- LOG.info("Adapter descriptions are up to date.");
-
- LOG.info("Checking migrations for existing adapters in StreamPipes Core
...");
- for (var migrationConfig : migrationConfigs) {
- LOG.info("Searching for assets of '{}'", migrationConfig.targetAppId());
- LOG.debug("Searching for assets of '{}' with config {}",
migrationConfig.targetAppId(), migrationConfig);
- var adapterDescriptions =
adapterStorage.getAdaptersByAppId(migrationConfig.targetAppId());
- LOG.info("Found {} instances for appId '{}'",
adapterDescriptions.size(), migrationConfig.targetAppId());
- for (var adapterDescription : adapterDescriptions) {
-
- var adapterVersion = adapterDescription.getVersion();
-
- if (adapterVersion == migrationConfig.fromVersion()) {
- LOG.info("Migration is required for adapter '{}'. Migrating from
version '{}' to '{}' ...",
- adapterDescription.getElementId(),
- adapterVersion, migrationConfig.toVersion()
- );
-
- var migrationResult = performMigration(
- adapterDescription,
- migrationConfig,
- String.format("%s/%s/adapter",
- extensionsServiceConfig.getServiceUrl(),
- MIGRATION_ENDPOINT
- )
- );
-
- if (migrationResult.success()) {
- LOG.info("Migration successfully performed by extensions service.
Updating adapter description ...");
- LOG.debug(
- "Migration was performed by extensions service '{}'",
- extensionsServiceConfig.getServiceUrl());
-
- adapterStorage.updateAdapter(migrationResult.element());
- LOG.info("Adapter description is updated - Migration successfully
completed at Core.");
- } else {
- LOG.error("Migration failed with the following reason: {}",
migrationResult.message());
- LOG.error(
- "Migration for adapter '{}' failed - Stopping adapter ...",
- migrationResult.element().getElementId()
- );
- try {
-
WorkerRestClient.stopStreamAdapter(extensionsServiceConfig.getServiceUrl(),
adapterDescription);
- } catch (AdapterException e) {
- LOG.error("Stopping adapter failed: {}",
StringUtils.join(e.getStackTrace(), "\n"));
- }
- LOG.info("Adapter successfully stopped.");
- }
- } else {
- LOG.info(
- "Migration is not applicable for adapter '{}' because of a
version mismatch - "
- + "adapter version: '{}', migration starts at:
'{}'",
- adapterDescription.getElementId(),
- adapterVersion,
- migrationConfig.fromVersion()
- );
- }
+ summary = "Migrate adapters and pipeline elements based on migration
configs", tags = {"Core", "Migration"},
+ responses = {
+ @ApiResponse(
+ responseCode = "" + HttpStatus.SC_OK,
+ description = "All provided migrations are handled. If an error
appeared, "
+ + "the corresponding actions are taken.")
}
- }
- return ok();
- }
-
- @POST
- @Path("pipeline-element/{serviceId}")
- @Consumes(MediaType.APPLICATION_JSON)
- @Operation(
- summary = "Migrate pipeline elements based on migration configs",
tags = {"Core", "Migration"},
- responses = {
- @ApiResponse(
- responseCode = "200" + HttpStatus.SC_OK,
- description = "All provided migrations are handled. "
- + "If an error appeared, the corresponding
actions are taken."
- )
- }
)
- public Response registerPipelineElementMigrations(
- @Parameter(
- in = ParameterIn.PATH,
- description = "the id of the extensions service that
requests migrations",
- required = true
- )
- @PathParam("serviceId") String serviceId,
- @Parameter(
- description = "list of config that describe the requested
migrations"
- )
- List<ModelMigratorConfig> migrationConfigs) {
+ public Response performMigrations(
+ @Parameter(
+ in = ParameterIn.PATH,
+ description = "the id of the extensions service that requests
migrations",
+ required = true
+ )
+ @PathParam("serviceId") String serviceId,
+ @Parameter(
+ description = "list of configs (ModelMigratorConfig) that describe
the requested migrations",
+ required = true
+ )
+ List<ModelMigratorConfig> migrationConfigs) {
var extensionsServiceConfig =
extensionsServiceStorage.getElementById(serviceId);
- LOG.info("Updating pipeline element descriptions by replacement...");
- updateDescriptions(migrationConfigs,
extensionsServiceConfig.getServiceUrl());
- LOG.info("Pipeline element descriptions are up to date.");
-
- LOG.info("Received {} pipeline element migrations from extension service
{}.",
- migrationConfigs.size(),
- extensionsServiceConfig.getServiceUrl());
- var availablePipelines = pipelineStorage.getAllPipelines();
- if (!availablePipelines.isEmpty()) {
- LOG.info("Found {} available pipelines. Checking pipelines for
applicable migrations...",
- availablePipelines.size()
- );
- }
-
- for (var pipeline : availablePipelines) {
- List<MigrationResult<?>> failedMigrations = new ArrayList<>();
-
- var migratedDataProcessors = pipeline.getSepas()
- .stream()
- .map(processor -> {
- if (getApplicableMigration(processor,
migrationConfigs).isPresent()) {
- return migratePipelineElement(
- processor,
- migrationConfigs,
- String.format("%s/%s/processor",
-
extensionsServiceConfig.getServiceUrl(),
- MIGRATION_ENDPOINT
- ),
- failedMigrations
- );
- } else {
- LOG.info("No migration applicable for data processor
'{}'.", processor.getElementId());
- return processor;
- }
- }
- ).toList();
- pipeline.setSepas(migratedDataProcessors);
-
- var migratedDataSinks = pipeline.getActions()
- .stream()
- .map(sink -> {
- if (getApplicableMigration(sink,
migrationConfigs).isPresent()) {
- return migratePipelineElement(
- sink,
- migrationConfigs,
- String.format("%s/%s/sink",
-
extensionsServiceConfig.getServiceUrl(),
- MIGRATION_ENDPOINT
- ),
- failedMigrations
- );
- } else {
- LOG.info("No migration applicable for data sink
'{}'.", sink.getElementId());
- return sink;
- }
- }
- ).toList();
- pipeline.setActions(migratedDataSinks);
-
- pipelineStorage.updatePipeline(pipeline);
-
- if (failedMigrations.isEmpty()) {
- LOG.info("Migration for pipeline successfully completed.");
- } else {
- // pass most recent version of pipeline
-
handleFailedMigrations(pipelineStorage.getPipeline(pipeline.getPipelineId()),
failedMigrations);
- }
- }
+ var adapterMigrations = filterConfigs(migrationConfigs,
List.of(SpServiceTagPrefix.ADAPTER));
+ var pipelineElementMigrations = filterConfigs(
+ migrationConfigs,
+ List.of(SpServiceTagPrefix.DATA_PROCESSOR,
SpServiceTagPrefix.DATA_SINK)
+ );
+
+ new
AdapterMigrationManager(adapterStorage).handleMigrations(extensionsServiceConfig,
adapterMigrations);
+ new PipelineElementMigrationManager(
+ pipelineStorage,
+ dataProcessorStorage,
+ dataSinkStorage)
+ .handleMigrations(extensionsServiceConfig, pipelineElementMigrations);
return ok();
}
- /**
- * Takes care about the failed migrations of pipeline elements.
- * This includes the following steps:
- * <ul>
- * <li> logging of failed pipeline elements
- * <li> setting migration results as pipeline notifications
- * <li> updating pipeline health status
- * <li> stopping the pipeline
- * </ul>
- *
- * @param pipeline the pipeline affected by failed migrations
- * @param failedMigrations the list of failed migrations
- */
- protected void handleFailedMigrations(Pipeline pipeline,
List<MigrationResult<?>> failedMigrations) {
- LOG.error("Failures in migration detected - The following pipeline
elements could to be migrated:\n"
- +
StringUtils.join(failedMigrations.stream().map(Record::toString).toList()),
"\n");
-
- pipeline.setPipelineNotifications(failedMigrations.stream().map(
- failedMigration -> "Failed migration of pipeline element:
%s".formatted(failedMigration.message())
- ).toList());
- pipeline.setHealthStatus(PipelineHealthStatus.REQUIRES_ATTENTION);
-
- pipelineStorage.updatePipeline(pipeline);
-
- // get updated version of pipeline after modification
- pipeline = pipelineStorage.getPipeline(pipeline.getPipelineId());
-
- stopPipeline(pipeline);
- }
-
-
- public void stopPipeline(Pipeline pipeline) {
- var pipelineExecutor = new PipelineExecutor(pipeline, true);
- var pipelineStopResult = pipelineExecutor.stopPipeline();
-
- if (pipelineStopResult.isSuccess()) {
- LOG.info("Pipeline successfully stopped.");
- } else {
- LOG.error("Pipeline stop failed.");
- }
- }
-
- /**
- * Filter the application migration for a pipeline definition.
- * By definition, there is only one migration config that fulfills the
requirements.
- * Otherwise, it should have been detected as duplicate by the extensions
service.
- *
- * @param pipelineElement pipeline element that should be migrated
- * @param migrationConfigs available migration configs to pick the
applicable from
- * @return config that is applicable for the given pipeline element
- */
- protected static Optional<ModelMigratorConfig> getApplicableMigration(
- InvocableStreamPipesEntity pipelineElement,
- List<ModelMigratorConfig> migrationConfigs
- ) {
+ private List<ModelMigratorConfig> filterConfigs(List<ModelMigratorConfig>
migrationConfigs,
+
List<SpServiceTagPrefix> modelTypes) {
return migrationConfigs
- .stream()
- .filter(
- config ->
config.modelType().equals(pipelineElement.getServiceTagPrefix())
- &&
config.targetAppId().equals(pipelineElement.getAppId())
- && config.fromVersion() ==
pipelineElement.getVersion()
- )
- .findFirst();
- }
-
- /**
- * Handle the migration of a pipeline element with respect to the given
model migration configs.
- * All applicable migrations found in the provided configs are executed for
the given pipeline element.
- * In case a migration fails, the related pipeline element receives the
latest definition of its static properties,
- * so that the pipeline element can be adapted by the user to resolve the
failed migration.
- *
- * @param pipelineElement pipeline element to be migrated
- * @param modelMigrations list of model migrations that might be applicable
for this pipeline element
- * @param url url of the extensions service endpoint that
handles the migration
- * @param failedMigrations collection of failed migrations which is extended
by occurring migration failures
- * @param <T> type of the pipeline element (e.g.,
DataProcessorInvocation)
- * @return the migrated (or - in case of a failure - updated) pipeline
element
- */
- protected <T extends InvocableStreamPipesEntity> T migratePipelineElement(
- T pipelineElement,
- List<ModelMigratorConfig> modelMigrations,
- String url,
- List<MigrationResult<?>> failedMigrations
- ) {
-
- // loop until no migrations are available anymore
- // this allows to apply multiple migrations for a pipeline element
sequentially
- // For example, first migration from 0 to 1 and the second migration from
1 to 2
- while (getApplicableMigration(pipelineElement,
modelMigrations).isPresent() && failedMigrations.isEmpty()) {
-
- var migrationConfig = getApplicableMigration(pipelineElement,
modelMigrations).get();
- LOG.info(
- "Found applicable migration for pipeline element '{}': {}",
- pipelineElement.getElementId(),
- migrationConfig
- );
-
- var migrationResult = performMigration(
- pipelineElement,
- migrationConfig,
- url
- );
-
- if (migrationResult.success()) {
- LOG.info("Migration successfully performed by extensions service.
Updating pipeline element invocation ...");
- LOG.debug("Migration was performed at extensions service endpoint
'{}'", url);
- pipelineElement = migrationResult.element();
- } else {
- LOG.error("Migration failed with the following reason: {}",
migrationResult.message());
- failedMigrations.add(migrationResult);
- }
- }
- if (!failedMigrations.isEmpty()) {
- updateFailedPipelineElement(pipelineElement);
- LOG.info("Updated pipeline elements with new description where automatic
migration failed.");
- }
- return pipelineElement;
- }
-
- /**
- * Performs the actual migration of a pipeline element.
- * This includes the communication with the extensions service which runs
the migration.
- *
- * @param pipelineElement pipeline element to be migrated
- * @param migrationConfig config of the migration to be performed
- * @param url url of the migration endpoint at the extensions
service
- * where the migration should be performed
- * @param <T> type of the processing element
- * @return result of the migration
- */
- protected <T extends VersionedNamedStreamPipesEntity> MigrationResult<T>
performMigration(
- T pipelineElement,
- ModelMigratorConfig migrationConfig,
- String url
- ) {
-
- try {
-
- var migrationRequest = new MigrationRequest<>(pipelineElement,
migrationConfig);
-
- String serializedRequest =
JacksonSerializer.getObjectMapper().writeValueAsString(migrationRequest);
-
- var migrationResponse = ExtensionServiceExecutions.extServicePostRequest(
- url,
- serializedRequest
- ).execute();
-
- TypeReference<MigrationResult<T>> typeReference = new TypeReference<>() {
- };
-
- return JacksonSerializer
- .getObjectMapper()
- .readValue(migrationResponse.returnContent().asString(),
typeReference);
- } catch (JsonProcessingException e) {
- LOG.error(
- "Migration of pipeline element failed before sending to the
extensions service, "
- + "pipeline element is not migrated. Serialization of
migration request failed: {}",
- StringUtils.join(e.getStackTrace(), "\n")
- );
- } catch (IOException e) {
- LOG.error("Migration of pipeline element failed at the extensions
service, pipeline element is not migrated: {}.",
- StringUtils.join(e.getStackTrace(), "\n")
- );
- }
- return MigrationResult.failure(pipelineElement, "Internal error during
migration at StreamPipes Core");
- }
-
- /**
- * Update all descriptions of entities in the Core that are affected by
migrations.
- *
- * @param migrationConfigs List of migrations to take in account
- * @param serviceUrl Url of the extension service that provides the
migrations.
- */
- protected void updateDescriptions(List<ModelMigratorConfig>
migrationConfigs, String serviceUrl) {
- migrationConfigs
- .stream()
- .collect(
- // We only need to update the description once per appId,
- // because this is directly done with the newest version of
the description and
- // there is iterative migration required.
- // To avoid unnecessary, multiple updates,
- // we filter the migration configs such that every appId is
unique.
- // This ensures that every description is only updated once.
- Collectors.toMap(
- ModelMigratorConfig::targetAppId,
- Function.identity(),
- (existing, replacement) -> existing
- )
- )
- .values()
.stream()
- .peek(config -> {
- var requestUrl = getRequestUrl(config.modelType(),
config.targetAppId(), serviceUrl);
- performUpdate(requestUrl);
- }
- )
+ .filter(config -> modelTypes.stream().anyMatch(modelType -> modelType
== config.modelType()))
.toList();
}
-
- /**
- * Get the URL that provides the description for an entity.
- *
- * @param entityType Type of the entity to be updated.
- * @param appId AppId of the entity to be updated
- * @param serviceUrl URL of the extensions service to which the entity
belongs
- * @return URL of the endpoint that provides the description for the given
entity
- */
- protected String getRequestUrl(SpServiceTagPrefix entityType, String appId,
String serviceUrl) {
-
- SpServiceUrlProvider urlProvider;
- switch (entityType) {
- case ADAPTER -> urlProvider = SpServiceUrlProvider.ADAPTER;
- case DATA_PROCESSOR -> urlProvider = SpServiceUrlProvider.DATA_PROCESSOR;
- case DATA_SINK -> urlProvider = SpServiceUrlProvider.DATA_SINK;
- default -> throw new RuntimeException("Unexpected instance type.");
- }
- return urlProvider.getInvocationUrl(serviceUrl, appId);
- }
-
- /**
- * Perform the update of the description based on the given requestUrl
- *
- * @param requestUrl URl that references the description to be updated at
the extensions service.
- */
- protected void performUpdate(String requestUrl) {
-
- try {
- var entityPayload = parseURIContent(requestUrl);
- var updateResult = Operations.verifyAndUpdateElement(entityPayload);
- if (!updateResult.isSuccess()) {
- LOG.error(
- "Updating the pipeline element description failed: {}",
- StringUtils.join(
-
updateResult.getNotifications().stream().map(Notification::toString).toList(),
- "\n")
- );
- }
- } catch (IOException | URISyntaxException | SepaParseException e) {
- LOG.error("Updating the pipeline element description failed due to the
following exception:\n{}",
- StringUtils.join(e.getStackTrace(), "\n")
- );
- }
- }
-
- /**
- * Update the static properties of the failed pipeline element with its
description.
- * This allows to adapt the failed pipeline element in the UI to overcome
the failed migration.
- *
- * @param pipelineElement pipeline element with failed migration
- */
- protected void updateFailedPipelineElement(InvocableStreamPipesEntity
pipelineElement) {
- List<StaticProperty> updatedStaticProperties = new ArrayList<>();
- if (pipelineElement instanceof DataProcessorInvocation) {
- updatedStaticProperties = dataProcessorStorage
- .getFirstDataProcessorByAppId(pipelineElement.getAppId())
- .getStaticProperties();
- } else if (pipelineElement instanceof DataSinkInvocation) {
- updatedStaticProperties = dataSinkStorage
- .getFirstDataSinkByAppId(pipelineElement.getAppId())
- .getStaticProperties();
- }
- pipelineElement.setStaticProperties(updatedStaticProperties);
- }
}
diff --git
a/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/ExtensionsModelSubmitter.java
b/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/ExtensionsModelSubmitter.java
index c562ae02f..d15efbc68 100644
---
a/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/ExtensionsModelSubmitter.java
+++
b/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/ExtensionsModelSubmitter.java
@@ -23,7 +23,6 @@ import
org.apache.streampipes.extensions.management.client.StreamPipesClientReso
import org.apache.streampipes.extensions.management.init.DeclarersSingleton;
import org.apache.streampipes.extensions.management.model.SpServiceDefinition;
import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceTag;
-import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceTagPrefix;
import
org.apache.streampipes.service.extensions.function.StreamPipesFunctionHandler;
import org.apache.streampipes.service.extensions.security.WebSecurityConfig;
@@ -51,28 +50,9 @@ public abstract class ExtensionsModelSubmitter extends
StreamPipesExtensionsServ
public void afterServiceRegistered(SpServiceDefinition serviceDef) {
StreamPipesClient client = new
StreamPipesClientResolver().makeStreamPipesClientInstance();
- // register all adapter migrations at StreamPipes Core
- var adapterMigrations = serviceDef.getMigrators()
- .stream()
- .filter(modelMigrator -> modelMigrator.config().modelType() ==
SpServiceTagPrefix.ADAPTER)
- .toList();
- client.adminApi().registerAdapterMigrations(
- adapterMigrations.stream().map(ModelMigrator::config).toList(),
- serviceId()
- );
-
- // register all pipeline element migrations at StreamPipes Core
- var pipelineElementMigrations = serviceDef.getMigrators()
- .stream()
- .filter(modelMigrator ->
- modelMigrator.config().modelType() ==
SpServiceTagPrefix.DATA_PROCESSOR
- || modelMigrator.config().modelType() ==
SpServiceTagPrefix.DATA_SINK
- )
- .toList();
- client.adminApi().registerPipelineElementMigrations(
-
pipelineElementMigrations.stream().map(ModelMigrator::config).toList(),
- serviceId()
- );
+ // register all migrations at StreamPipes Core
+ var migrationConfigs =
serviceDef.getMigrators().stream().map(ModelMigrator::config).toList();
+ client.adminApi().registerMigrations(migrationConfigs, serviceId());
// initialize all function instances
StreamPipesFunctionHandler.INSTANCE.initializeFunctions(serviceDef.getServiceGroup());
