You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampipes.apache.org by go...@apache.org on 2021/12/31 16:07:15 UTC
[incubator-streampipes] branch edge-extensions updated (d304587 -> 5aff17f)
This is an automated email from the ASF dual-hosted git repository.
gomm pushed a change to branch edge-extensions
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git.
from d304587 preliminary stateful migration implementation
new 4820306 [WIP] refactor pipeline executor
new 47c6a80 [WIP] refactor pipeline executor
new 5aff17f [WIP] improve pipeline executor
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../api/InvocablePipelineElementResource.java | 7 +
.../model/base/InvocableStreamPipesEntity.java | 10 +
.../PipelineElementReconfigurationEntity.java | 10 +
.../model/pipeline/PipelineElementStatus.java | 10 +
.../manager/data/PipelineGraphBuilder.java | 6 +-
.../manager/execution/http/HttpRequestBuilder.java | 2 +
.../pipeline/AbstractPipelineExecutor.java | 628 ---------------------
.../PipelineElementReconfigurationExecutor.java | 41 --
.../execution/pipeline/PipelineExecutor.java | 110 ----
.../pipeline/PipelineMigrationExecutor.java | 461 ---------------
.../pipeline/PipelineMigrationHelpers.java | 21 -
.../pipeline/executor/PipelineExecutor.java | 182 ++++++
.../pipeline/executor/PipelineExecutorBuilder.java | 117 ++++
.../pipeline/executor/PipelineExecutorFactory.java | 66 +++
.../executor/steps/EntitiesLifecycleObject.java | 37 +-
.../pipeline/executor/steps/GetStateStep.java | 84 +++
.../executor/steps/PipelineExecutionStep.java | 56 ++
.../executor/steps/PrepareMigrationStep.java | 136 +++++
.../executor/steps/PrepareStartPipelineStep.java | 103 ++++
.../executor/steps/ReconfigureElementStep.java | 86 +++
.../steps/StartGraphsAndAssociatedRelaysStep.java | 81 +++
.../pipeline/executor/steps/StartPipelineStep.java | 84 +++
.../pipeline/executor/steps/StartRelaysStep.java | 66 +++
.../steps/StopGraphsAndAssociatedRelaysStep.java | 78 +++
.../pipeline/executor/steps/StopPipelineStep.java | 94 +++
.../pipeline/executor/steps/StopRelaysStep.java | 66 +++
.../executor/steps/StoreMigratedPipelineStep.java | 56 ++
.../pipeline/executor/steps/StorePipelineStep.java | 69 +++
.../executor/utils/CommunicationUtils.java | 71 +++
.../pipeline/executor/utils/DataSetUtils.java | 20 +
.../executor/utils/PipelineElementUtils.java | 203 +++++++
.../pipeline/executor/utils/PipelineUtils.java | 129 +++++
.../pipeline/executor/utils/RelayUtils.java | 243 ++++++++
.../pipeline/executor/utils/StatusUtils.java | 61 ++
.../pipeline/executor/utils/StorageUtils.java | 94 +++
.../migration/PipelineElementMigrationHandler.java | 9 +-
.../streampipes/manager/operations/Operations.java | 8 +-
.../PipelineElementReconfigurationHandler.java | 20 +-
.../migrate-pipeline-processors.component.ts | 23 +-
39 files changed, 2337 insertions(+), 1311 deletions(-)
delete mode 100644 streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/AbstractPipelineExecutor.java
delete mode 100644 streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/PipelineElementReconfigurationExecutor.java
delete mode 100644 streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/PipelineExecutor.java
delete mode 100644 streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/PipelineMigrationExecutor.java
delete mode 100644 streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/PipelineMigrationHelpers.java
create mode 100644 streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/PipelineExecutor.java
create mode 100644 streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/PipelineExecutorBuilder.java
create mode 100644 streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/PipelineExecutorFactory.java
copy streampipes-model/src/main/java/org/apache/streampipes/model/node/container/ContainerEnvBuilder.java => streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/EntitiesLifecycleObject.java (52%)
create mode 100644 streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/GetStateStep.java
create mode 100644 streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/PipelineExecutionStep.java
create mode 100644 streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/PrepareMigrationStep.java
create mode 100644 streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/PrepareStartPipelineStep.java
create mode 100644 streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/ReconfigureElementStep.java
create mode 100644 streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/StartGraphsAndAssociatedRelaysStep.java
create mode 100644 streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/StartPipelineStep.java
create mode 100644 streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/StartRelaysStep.java
create mode 100644 streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/StopGraphsAndAssociatedRelaysStep.java
create mode 100644 streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/StopPipelineStep.java
create mode 100644 streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/StopRelaysStep.java
create mode 100644 streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/StoreMigratedPipelineStep.java
create mode 100644 streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/StorePipelineStep.java
create mode 100644 streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/CommunicationUtils.java
create mode 100644 streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/DataSetUtils.java
create mode 100644 streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/PipelineElementUtils.java
create mode 100644 streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/PipelineUtils.java
create mode 100644 streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/RelayUtils.java
create mode 100644 streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/StatusUtils.java
create mode 100644 streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/StorageUtils.java
[incubator-streampipes] 03/03: [WIP] improve pipeline executor
Posted by go...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
gomm pushed a commit to branch edge-extensions
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit 5aff17f9c09c3c194444e0e661d932daf47cdce7
Author: daniel-gomm <da...@outlook.de>
AuthorDate: Fri Dec 31 17:06:41 2021 +0100
[WIP] improve pipeline executor
---
.../model/pipeline/PipelineElementStatus.java | 10 +
.../manager/data/PipelineGraphBuilder.java | 6 +-
.../manager/execution/http/HttpRequestBuilder.java | 2 +
.../pipeline/AbstractPipelineExecutor.java | 628 ---------------------
.../pipeline/PipelineMigrationHelpers.java | 21 -
.../pipeline/executor/PipelineExecutor.java | 50 +-
.../pipeline/executor/PipelineExecutorBuilder.java | 66 +--
.../pipeline/executor/PipelineExecutorFactory.java | 24 +-
.../operations/types/MigrationOperation.java | 21 -
.../operations/types/ReconfigurationOperation.java | 21 -
.../EntitiesLifecycleObject.java} | 6 +-
.../GetStateStep.java} | 7 +-
.../PipelineExecutionStep.java} | 7 +-
.../PrepareMigrationStep.java} | 7 +-
.../PrepareStartPipelineStep.java} | 6 +-
.../ReconfigureElementStep.java} | 7 +-
.../StartGraphsAndAssociatedRelaysStep.java} | 7 +-
.../StartPipelineStep.java} | 6 +-
.../StartRelaysStep.java} | 7 +-
.../StopGraphsAndAssociatedRelaysStep.java} | 7 +-
.../StopPipelineStep.java} | 25 +-
.../StopRelaysStep.java} | 7 +-
.../StoreMigratedPipelineStep.java} | 6 +-
.../StorePipelineStep.java} | 6 +-
.../pipeline/executor/utils/RelayUtils.java | 2 +-
.../pipeline/executor/utils/StatusUtils.java | 2 +-
.../migrate-pipeline-processors.component.ts | 23 +-
27 files changed, 140 insertions(+), 847 deletions(-)
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/PipelineElementStatus.java b/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/PipelineElementStatus.java
index 7e7274b..54f34e8 100644
--- a/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/PipelineElementStatus.java
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/PipelineElementStatus.java
@@ -25,6 +25,8 @@ public class PipelineElementStatus {
private String optionalMessage;
private String operation;
private String elementNode;
+ //TODO: Assess if runningInstanceId is needed separately or if it can be combined with the elementId
+ private String runningInstanceId;
private boolean success;
@@ -85,4 +87,12 @@ public class PipelineElementStatus {
public void setElementNode(String elementNode) {
this.elementNode = elementNode;
}
+
+ public String getRunningInstanceId() {
+ return runningInstanceId;
+ }
+
+ public void setRunningInstanceId(String runningInstanceId) {
+ this.runningInstanceId = runningInstanceId;
+ }
}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/data/PipelineGraphBuilder.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/data/PipelineGraphBuilder.java
index a83c125..bbcadfb 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/data/PipelineGraphBuilder.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/data/PipelineGraphBuilder.java
@@ -28,9 +28,9 @@ import java.util.stream.Collectors;
public class PipelineGraphBuilder {
- private Pipeline pipeline;
- private List<NamedStreamPipesEntity> allPipelineElements;
- private List<InvocableStreamPipesEntity> invocableElements;
+ private final Pipeline pipeline;
+ private final List<NamedStreamPipesEntity> allPipelineElements;
+ private final List<InvocableStreamPipesEntity> invocableElements;
public PipelineGraphBuilder(Pipeline pipeline) {
this.pipeline = pipeline;
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/HttpRequestBuilder.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/HttpRequestBuilder.java
index dc0fdd9..82272c8 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/HttpRequestBuilder.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/HttpRequestBuilder.java
@@ -99,10 +99,12 @@ public class HttpRequestBuilder {
PipelineElementStatus status = new PipelineElementStatus(endpointUrl, payload.getName(), response.isSuccess(),
response.getOptionalMessage());
if(payload instanceof InvocableStreamPipesEntity){
+ status.setRunningInstanceId(((InvocableStreamPipesEntity) payload).getDeploymentRunningInstanceId());
status.setElementNode(((InvocableStreamPipesEntity)payload).getDeploymentTargetNodeId());
status.setOperation(action);
}
else if(payload instanceof SpDataStreamRelayContainer){
+ status.setRunningInstanceId(((SpDataStreamRelayContainer) payload).getRunningStreamRelayInstanceId() + " relay");
status.setElementNode(((SpDataStreamRelayContainer)payload).getDeploymentTargetNodeId());
status.setOperation(action + " relay");
}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/AbstractPipelineExecutor.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/AbstractPipelineExecutor.java
deleted file mode 100644
index 01c4f2c..0000000
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/AbstractPipelineExecutor.java
+++ /dev/null
@@ -1,628 +0,0 @@
-/*
- * 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.execution.pipeline;
-
-import org.apache.streampipes.commons.exceptions.SpRuntimeException;
-import org.apache.streampipes.manager.data.PipelineGraph;
-import org.apache.streampipes.manager.data.PipelineGraphHelpers;
-import org.apache.streampipes.manager.execution.http.GraphSubmitter;
-
-import org.apache.streampipes.manager.execution.http.StateSubmitter;
-import org.apache.streampipes.manager.util.TemporaryGraphStorage;
-import org.apache.streampipes.model.SpDataSet;
-import org.apache.streampipes.model.SpDataStream;
-import org.apache.streampipes.model.eventrelay.SpDataStreamRelay;
-import org.apache.streampipes.model.eventrelay.SpDataStreamRelayContainer;
-import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
-import org.apache.streampipes.model.base.NamedStreamPipesEntity;
-import org.apache.streampipes.model.graph.DataProcessorInvocation;
-import org.apache.streampipes.model.graph.DataSinkInvocation;
-import org.apache.streampipes.model.grounding.EventGrounding;
-import org.apache.streampipes.model.grounding.KafkaTransportProtocol;
-import org.apache.streampipes.model.grounding.TransportProtocol;
-import org.apache.streampipes.model.pipeline.Pipeline;
-import org.apache.streampipes.model.pipeline.PipelineElementStatus;
-import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
-import org.apache.streampipes.model.staticproperty.SecretStaticProperty;
-import org.apache.streampipes.storage.api.INodeDataStreamRelay;
-import org.apache.streampipes.storage.api.IPipelineStorage;
-import org.apache.streampipes.storage.management.StorageDispatcher;
-import org.apache.streampipes.user.management.encryption.CredentialsManager;
-
-import java.security.GeneralSecurityException;
-import java.util.*;
-import java.util.concurrent.atomic.AtomicReference;
-import java.util.stream.Collectors;
-
-public abstract class AbstractPipelineExecutor {
-
- protected Pipeline pipeline;
- protected boolean visualize;
- protected boolean storeStatus;
- protected boolean monitor;
-
- public AbstractPipelineExecutor(Pipeline pipeline, boolean visualize, boolean storeStatus, boolean monitor) {
- this.pipeline = pipeline;
- this.visualize = visualize;
- this.storeStatus = storeStatus;
- this.monitor = monitor;
- }
-
- // standard methods
- protected void setPipelineStarted(Pipeline pipeline) {
- pipeline.setRunning(true);
- pipeline.setStartedAt(new Date().getTime());
- getPipelineStorageApi().updatePipeline(pipeline);
- }
-
- protected void setPipelineStopped(Pipeline pipeline) {
- pipeline.setRunning(false);
- getPipelineStorageApi().updatePipeline(pipeline);
- }
-
- protected void deleteVisualization(String pipelineId) {
- StorageDispatcher.INSTANCE
- .getNoSqlStore()
- .getVisualizationStorageApi()
- .deleteVisualization(pipelineId);
- }
-
- protected void storeInvocationGraphs(String pipelineId, List<InvocableStreamPipesEntity> graphs,
- List<SpDataSet> dataSets) {
- TemporaryGraphStorage.graphStorage.put(pipelineId, graphs);
- TemporaryGraphStorage.datasetStorage.put(pipelineId, dataSets);
- }
-
- protected void storeDataStreamRelayContainer(List<SpDataStreamRelayContainer> relays) {
- //relays.forEach(StreamPipesClusterManager::persistDataStreamRelay);
- relays.forEach(relay -> getDataStreamRelayApi().addRelayContainer(relay));
- }
-
- protected void deleteDataStreamRelayContainer(List<SpDataStreamRelayContainer> relays) {
- //relays.forEach(StreamPipesClusterManager::deleteDataStreamRelay);
- relays.forEach(relay -> getDataStreamRelayApi().deleteRelayContainer(relay));
- }
-
- protected void updateDataStreamRelayContainer(List<SpDataStreamRelayContainer> relays) {
- //relays.forEach(StreamPipesClusterManager::updateDataStreamRelay);
- relays.forEach(relay -> getDataStreamRelayApi().updateRelayContainer(relay));
- }
-
-
- protected PipelineOperationStatus startPipelineElementsAndRelays(List<InvocableStreamPipesEntity> graphs,
- List<SpDataStreamRelayContainer> relays){
- if (graphs.isEmpty()) {
- return initPipelineOperationStatus();
- }
- return new GraphSubmitter(pipeline.getPipelineId(), pipeline.getName(),
- graphs, new ArrayList<>(), relays).invokePipelineElementsAndRelays();
- }
-
- protected PipelineOperationStatus stopPipelineElementsAndRelays(List<InvocableStreamPipesEntity> graphs,
- List<SpDataStreamRelayContainer> relays){
- if (graphs.isEmpty()) {
- return initPipelineOperationStatus();
- }
- return new GraphSubmitter(pipeline.getPipelineId(), pipeline.getName(),
- graphs, new ArrayList<>(),relays).detachPipelineElementsAndRelays();
- }
-
- protected PipelineOperationStatus startRelays(List<SpDataStreamRelayContainer> relays){
- return new GraphSubmitter(pipeline.getPipelineId(), pipeline.getName(), new ArrayList<>(), new ArrayList<>(),
- relays).invokeRelaysOnMigration();
- }
-
- protected PipelineOperationStatus stopRelays(List<SpDataStreamRelayContainer> relays){
- return new GraphSubmitter(pipeline.getPipelineId(), pipeline.getName(),new ArrayList<>(), new ArrayList<>(),
- relays).detachRelaysOnMigration();
- }
-
- protected PipelineElementStatus getState(InvocableStreamPipesEntity graph){
- return new StateSubmitter(pipeline.getPipelineId(), pipeline.getName(), graph, null).getElementState();
- }
-
- protected PipelineElementStatus setState(InvocableStreamPipesEntity graph, String state){
- return new StateSubmitter(pipeline.getPipelineId(), pipeline.getName(), graph, state).setElementState();
- }
-
- protected List<SpDataStreamRelayContainer> findRelaysWhenStopping(List<NamedStreamPipesEntity> predecessors,
- InvocableStreamPipesEntity target){
-
- List<SpDataStreamRelayContainer> relays = new ArrayList<>();
-
- predecessors.forEach(pred -> {
- List<SpDataStreamRelay> dataStreamRelays = new ArrayList<>();
- SpDataStreamRelayContainer relayContainer = new SpDataStreamRelayContainer();
-
- if (pred instanceof DataProcessorInvocation){
- //Data Processor
- DataProcessorInvocation graph = (DataProcessorInvocation) pred;
- if (differentDeploymentTargets(pred, target)) {
-
- // TODO only add if no other processor or sink depends on relay
- String predDOMId = pred.getDOM();
- String targetRunningInstanceId = target.getDeploymentRunningInstanceId();
- Optional<DataProcessorInvocation> foundProcessor = pipeline.getSepas().stream()
- .filter(processor -> processor.getConnectedTo().contains(predDOMId))
- .filter(processor -> !processor.getDeploymentRunningInstanceId().equals(targetRunningInstanceId))
- .findAny();
-
- Optional<DataSinkInvocation> foundSink = pipeline.getActions().stream()
- .filter(action -> action.getConnectedTo().contains(predDOMId))
- .findAny();
-
- boolean foundDependencyOnDifferentTarget = false;
- if (foundProcessor.isPresent()) {
- foundDependencyOnDifferentTarget = differentDeploymentTargets(foundProcessor.get(), target);
- }
-
- if (foundSink.isPresent()) {
- foundDependencyOnDifferentTarget = differentDeploymentTargets(foundSink.get(), target);
- }
-
- if (foundDependencyOnDifferentTarget) {
- dataStreamRelays.addAll(findRelaysWithMatchingTopic(graph, target));
-
- relayContainer = new SpDataStreamRelayContainer(graph);
- relayContainer.setOutputStreamRelays(dataStreamRelays);
-
- relays.add(relayContainer);
- }
-
- }
- } else if (pred instanceof SpDataStream){
- //DataStream
- SpDataStream stream = (SpDataStream) pred;
- if (differentDeploymentTargets(stream, target)){
-
- String id = extractUniqueAdpaterId(stream.getElementId());
- //There is a relay that needs to be removed
- dataStreamRelays.add(new SpDataStreamRelay(new EventGrounding(target.getInputStreams()
- .get(getIndex(pred.getDOM(), target))
- .getEventGrounding())));
- String relayStrategy = pipeline.getEventRelayStrategy();
- relays.add(new SpDataStreamRelayContainer(id, relayStrategy, stream, dataStreamRelays));
- }
- }
- });
- return relays;
- }
-
-
- protected List<SpDataStreamRelayContainer> findRelays(List<NamedStreamPipesEntity> predecessors,
- InvocableStreamPipesEntity target){
-
- List<SpDataStreamRelayContainer> relays = new ArrayList<>();
-
- predecessors.forEach(pred -> {
- List<SpDataStreamRelay> dataStreamRelays = new ArrayList<>();
- SpDataStreamRelayContainer relayContainer = new SpDataStreamRelayContainer();
-
- if (pred instanceof DataProcessorInvocation){
- //Data Processor
- DataProcessorInvocation graph = (DataProcessorInvocation) pred;
- if (differentDeploymentTargets(pred, target)) {
-
- String runningRelayId = ((DataProcessorInvocation) pred).getDeploymentRunningInstanceId();
- Optional<SpDataStreamRelayContainer> existingRelay = getRelayContainerById(runningRelayId);
-
- // only add relay if not existing - prevent from duplicate relays with same topic to same target
- Collection<? extends SpDataStreamRelay> foundRelays = findRelaysWithMatchingTopic(graph, target);
-
- if (!existingRelay.isPresent() || missingRelayToTarget(existingRelay.get(), foundRelays)) {
- dataStreamRelays.addAll(findRelaysWithMatchingTopic(graph, target));
-
- //dsRelayContainer.setRunningStreamRelayInstanceId(pipeline.getPipelineId());
- relayContainer = new SpDataStreamRelayContainer(graph);
- relayContainer.setOutputStreamRelays(dataStreamRelays);
-
- relays.add(relayContainer);
- }
-
- }
- } else if (pred instanceof SpDataStream){
- //DataStream
- SpDataStream stream = (SpDataStream) pred;
- if (differentDeploymentTargets(stream, target)){
-
- String id = extractUniqueAdpaterId(stream.getElementId());
- Optional<SpDataStreamRelayContainer> existingRelay = getRelayContainerById(id);
-
- // only add relay if not existing - prevent from duplicate relays with same topic
- if(!existingRelay.isPresent()) {
- //There is a relay that needs to be removed
- dataStreamRelays.add(new SpDataStreamRelay(new EventGrounding(target.getInputStreams()
- .get(getIndex(pred.getDOM(), target))
- .getEventGrounding())));
- String relayStrategy = pipeline.getEventRelayStrategy();
- relays.add(new SpDataStreamRelayContainer(id, relayStrategy, stream, dataStreamRelays));
- } else {
- // generate relays for adapter streams to remote processors
- List<SpDataStreamRelayContainer> generatedRelays =
- generateDataStreamRelays(Collections.singletonList(target));
-
- relays.addAll(generatedRelays);
- }
- }
- }
- });
- return relays;
- }
-
- private boolean missingRelayToTarget(SpDataStreamRelayContainer existingRelayContainer,
- Collection<? extends SpDataStreamRelay> foundRelays) {
-
- List<TransportProtocol> set = foundRelays.stream()
- .map(SpDataStreamRelay::getEventGrounding)
- .map(EventGrounding::getTransportProtocol)
- .collect(Collectors.toList());
-
- List<TransportProtocol> relay = existingRelayContainer.getOutputStreamRelays().stream()
- .map(SpDataStreamRelay::getEventGrounding)
- .map(EventGrounding::getTransportProtocol)
- .collect(Collectors.toList());
-
- boolean missing = true;
- for (TransportProtocol tp: set) {
- for (TransportProtocol r: relay) {
- String targetTopic = tp.getTopicDefinition().getActualTopicName();
- String rTopic = r.getTopicDefinition().getActualTopicName();
- String targetHost = tp.getBrokerHostname();
- String rHost = r.getBrokerHostname();
-
- if (targetHost.equals(rHost) && targetTopic.equals(rTopic)) {
- missing = false;
- }
- }
- }
-
- return missing;
- }
-
- protected List<NamedStreamPipesEntity> getPredecessors(NamedStreamPipesEntity source,
- InvocableStreamPipesEntity target,
- PipelineGraph pipelineGraph,
- List<NamedStreamPipesEntity> foundPredecessors){
-
- Set<InvocableStreamPipesEntity> targets = getTargetsAsSet(source, pipelineGraph,
- InvocableStreamPipesEntity.class);
-
- //TODO: Check if this works for all graph topologies
- if (targets.contains(target)){
- foundPredecessors.add(source);
- } else {
- List<NamedStreamPipesEntity> successors = getTargetsAsList(source, pipelineGraph,
- NamedStreamPipesEntity.class);
-
- if (successors.isEmpty()) return foundPredecessors;
- successors.forEach(successor -> getPredecessors(successor, target, pipelineGraph, foundPredecessors));
- }
- return foundPredecessors;
- }
-
- protected NamedStreamPipesEntity findMatching(NamedStreamPipesEntity entity, PipelineGraph pipelineGraph){
- AtomicReference<NamedStreamPipesEntity> match = new AtomicReference<>();
- List<SpDataStream> dataStreams = PipelineGraphHelpers.findStreams(pipelineGraph);
-
- for (SpDataStream stream : dataStreams) {
- NamedStreamPipesEntity foundEntity = compareGraphs(stream, entity, pipelineGraph, new ArrayList<>());
- if (foundEntity != null) {
- match.set(foundEntity);
- }
- }
- return match.get();
- }
-
- private NamedStreamPipesEntity compareGraphs(NamedStreamPipesEntity source,
- NamedStreamPipesEntity searchedEntity,
- PipelineGraph pipelineGraph,
- List<NamedStreamPipesEntity> successors){
- if(matchingDOM(source, searchedEntity)) {
- return source;
- } else if (successors.isEmpty()) {
- successors = getTargetsAsList(source, pipelineGraph, NamedStreamPipesEntity.class);
- Optional<NamedStreamPipesEntity> successor = successors.stream().findFirst();
- if (successor.isPresent()) {
- successors.remove(successor.get());
- return compareGraphs(successor.get(), searchedEntity, pipelineGraph, successors);
- }
- }
- return null;
- }
-
- protected List<SpDataStreamRelayContainer> generateRelays(List<InvocableStreamPipesEntity> graphs) {
- return generateDataStreamRelays(graphs).stream()
- .filter(r -> r.getOutputStreamRelays().size() > 0)
- .collect(Collectors.toList());
- }
-
- // TODO: when using kafka as edge protocol it generates duplicate event relays -> check with mqtt as edge
- // protocol and fix
- private List<SpDataStreamRelayContainer> generateDataStreamRelays(List<InvocableStreamPipesEntity> graphs) {
- List<SpDataStreamRelayContainer> relays = new ArrayList<>();
-
- for (InvocableStreamPipesEntity graph : graphs) {
- for (SpDataStream stream: pipeline.getStreams()) {
- if (differentDeploymentTargets(stream, graph) && connected(stream, graph)) {
-
- List<SpDataStreamRelay> dataStreamRelays = new ArrayList<>();
- dataStreamRelays.add(new SpDataStreamRelay(new EventGrounding(graph.getInputStreams()
- .get(getIndex(stream.getDOM(), graph))
- .getEventGrounding())));
-
- String id = extractUniqueAdpaterId(stream.getElementId());
- String relayStrategy = pipeline.getEventRelayStrategy();
-
- if (!relayExists(relays, id)) {
- relays.add(new SpDataStreamRelayContainer(id, relayStrategy, stream, dataStreamRelays));
- }
- }
- }
- for (DataProcessorInvocation processor : pipeline.getSepas()) {
- if (differentDeploymentTargets(processor, graph) && connected(processor, graph)) {
- if (!relayExists(relays, processor.getDeploymentRunningInstanceId())) {
-// String previousId = processor.getDeploymentRunningInstanceId();
-// String modifiedId = previousId + "-" + processor.getDeploymentTargetNodeId();
-// processor.setDeploymentRunningInstanceId(modifiedId);
- SpDataStreamRelayContainer processorRelay = new SpDataStreamRelayContainer(processor);
- relays.add(processorRelay);
- }
- }
- }
- }
- return relays;
- }
-
-
- // Helpers
-
- /**
- *
- * @param id
- * @return
- */
- private Optional<SpDataStreamRelayContainer> getRelayContainerById(String id) {
- return StorageDispatcher.INSTANCE.getNoSqlStore().getNodeDataStreamRelayStorage().getRelayContainerById(id);
- }
-
- /**
- * Check if relay with deploymentRunningInstanceId of predecessor already exists
- *
- * @param relays List of existing relays
- * @param deploymentRunningInstanceId Id to check
- * @return boolean
- */
- private boolean relayExists(List<SpDataStreamRelayContainer> relays,
- String deploymentRunningInstanceId) {
- return relays.stream().anyMatch(r -> r.getRunningStreamRelayInstanceId().equals(deploymentRunningInstanceId));
- }
-
- /**
- * Updates group.id for data processor/sink. Note: KafkaTransportProtocol only!!
- *
- * @param entity data processor/sink
- */
- protected void updateKafkaGroupIds(InvocableStreamPipesEntity entity) {
- entity.getInputStreams()
- .stream()
- .filter(is -> is.getEventGrounding().getTransportProtocol() instanceof KafkaTransportProtocol)
- .map(is -> is.getEventGrounding().getTransportProtocol())
- .map(KafkaTransportProtocol.class::cast)
- .forEach(tp -> tp.setGroupId(UUID.randomUUID().toString()));
- }
-
- /**
- * Decrypt potential secrets contained in static properties, e.g., passwords
- *
- * @param graphs List of graphs
- * @return List of decrypted graphs
- */
- protected List<InvocableStreamPipesEntity> decryptSecrets(List<InvocableStreamPipesEntity> graphs) {
- List<InvocableStreamPipesEntity> decryptedGraphs = new ArrayList<>();
- graphs.stream().map(g -> {
- if (g instanceof DataProcessorInvocation) {
- return new DataProcessorInvocation((DataProcessorInvocation) g);
- } else {
- return new DataSinkInvocation((DataSinkInvocation) g);
- }
- }).forEach(g -> {
- g.getStaticProperties()
- .stream()
- .filter(SecretStaticProperty.class::isInstance)
- .forEach(sp -> {
- try {
- String decrypted = CredentialsManager.decrypt(pipeline.getCreatedByUser(),
- ((SecretStaticProperty) sp).getValue());
- ((SecretStaticProperty) sp).setValue(decrypted);
- ((SecretStaticProperty) sp).setEncrypted(false);
- } catch (GeneralSecurityException e) {
- e.printStackTrace();
- }
- });
- decryptedGraphs.add(g);
- });
- return decryptedGraphs;
- }
-
- /**
- * Get pipeline storage dispatcher API
- *
- * @return IPipelineStorage NoSQL storage interface for pipelines
- */
- private IPipelineStorage getPipelineStorageApi() {
- return StorageDispatcher.INSTANCE.getNoSqlStore().getPipelineStorageAPI();
- }
-
- /**
- * Get data stream relay storage dispatcher API
- *
- * @return INodeDataStreamRelay NoSQL storage interface for data stream relays
- */
- private INodeDataStreamRelay getDataStreamRelayApi() {
- return StorageDispatcher.INSTANCE.getNoSqlStore().getNodeDataStreamRelayStorage();
- }
-
- /**
- * Extract topic name
- *
- * @param entity
- * @return
- */
- private String extractActualTopic(NamedStreamPipesEntity entity) {
- if (entity instanceof SpDataStream) {
- return ((SpDataStream) entity)
- .getEventGrounding().getTransportProtocol().getTopicDefinition().getActualTopicName();
- } else if (entity instanceof SpDataStreamRelay) {
- return ((SpDataStreamRelay) entity)
- .getEventGrounding().getTransportProtocol().getTopicDefinition().getActualTopicName();
- }
- throw new SpRuntimeException("Could not extract actual topic name from entity");
- }
-
- // Edge / Migration Helpers
-
- /**
- * Compare deployment targets of two pipeline elements, namely data stream/processor (source) and data
- * processor/sink (target)
- *
- * @param e1
- * @param e2
- * @return boolean value that returns true if source and target share the same deployment target, else false
- */
- private boolean differentDeploymentTargets(NamedStreamPipesEntity e1, InvocableStreamPipesEntity e2) {
- if (e1 instanceof SpDataStream) {
- return !((SpDataStream) e1).getDeploymentTargetNodeId().equals(e2.getDeploymentTargetNodeId());
- } else if (e1 instanceof DataProcessorInvocation) {
- return !((DataProcessorInvocation) e1).getDeploymentTargetNodeId().equals(e2.getDeploymentTargetNodeId());
- } else if (e1 instanceof DataSinkInvocation) {
- return !((DataSinkInvocation) e1).getDeploymentTargetNodeId().equals(e2.getDeploymentTargetNodeId());
- }
- throw new SpRuntimeException("Matching deployment targets check failed");
- }
-
- /**
- * Find relays with matching topics
- *
- * @param graph data processor
- * @param target data processor/sink
- * @return collection of data stream relays
- */
- private Collection<? extends SpDataStreamRelay> findRelaysWithMatchingTopic(DataProcessorInvocation graph,
- InvocableStreamPipesEntity target) {
- return graph.getOutputStreamRelays().stream().
- filter(relay ->
- target.getInputStreams().stream()
- .map(this::extractActualTopic)
- .collect(Collectors.toSet())
- .contains(extractActualTopic(relay)))
- .collect(Collectors.toList());
- }
-
-
- private <T> Set<T> getTargetsAsSet(NamedStreamPipesEntity source, PipelineGraph pipelineGraph,
- Class<T> clazz){
- return pipelineGraph.outgoingEdgesOf(source)
- .stream()
- .map(pipelineGraph::getEdgeTarget)
- .map(clazz::cast)
- .collect(Collectors.toSet());
- }
-
- private <T> List<T> getTargetsAsList(NamedStreamPipesEntity source, PipelineGraph pipelineGraph,
- Class<T> clazz){
- return new ArrayList<>(getTargetsAsSet(source, pipelineGraph, clazz));
- }
-
- /**
- * Compare connection of two pipeline elements, namely data stream/processor (source) and data processor/sink
- * (target) by DOM identifier.
- *
- * @param source data stream or data processor
- * @param target data processor/sink
- * @return boolean value that returns true if source and target are connected, else false
- */
- private boolean connected(NamedStreamPipesEntity source, InvocableStreamPipesEntity target) {
- int index = getIndex(source.getDOM(), target);
- if (index != -1) {
- return target.getConnectedTo().get(index).equals(source.getDOM());
- }
- return false;
- }
-
- /**
- * Get index of data processor/sink connection based on source DOM identifier
- *
- * @param sourceDomId source DOM identifier
- * @param target data processor/sink
- * @return Integer with index of connection, if invalid returns -1.
- */
- private Integer getIndex(String sourceDomId, InvocableStreamPipesEntity target) {
- return target.getConnectedTo().indexOf(sourceDomId);
- }
-
- /**
- * Checks if DOM are equal
- *
- * @param source pipeline element
- * @param target pipeline element
- * @return true if DOM is the same, else false
- */
- private boolean matchingDOM(NamedStreamPipesEntity source, NamedStreamPipesEntity target) {
- return source.getDOM().equals(target.getDOM());
- }
-
- /**
- * Get List of InvocableStreamPipes entities, i.e., data processors/sinks from list of NameStreamPipesEntity
- *
- * @param graphs List<NamedStreamPipesEntity> graphs
- * @return
- */
- private List<InvocableStreamPipesEntity> getListOfInvocableStreamPipesEntity(List<NamedStreamPipesEntity> graphs) {
- List<InvocableStreamPipesEntity> invocableEntities = new ArrayList<>();
- graphs.stream()
- .filter(i -> i instanceof InvocableStreamPipesEntity)
- .forEach(i -> invocableEntities.add((InvocableStreamPipesEntity) i));
- return invocableEntities;
- }
-
- /**
- * Create pipeline operation status with pipeline id and name and set success to true
- *
- * @return PipelineOperationStatus
- */
- protected PipelineOperationStatus initPipelineOperationStatus() {
- PipelineOperationStatus status = new PipelineOperationStatus();
- status.setPipelineId(pipeline.getPipelineId());
- status.setPipelineName(pipeline.getName());
- status.setSuccess(true);
- return status;
- }
-
- private <T> List<T> filter(List<InvocableStreamPipesEntity> graphs, Class<T> clazz) {
- return graphs
- .stream()
- .filter(clazz::isInstance)
- .map(clazz::cast)
- .collect(Collectors.toList());
- }
-
- private String extractUniqueAdpaterId(String s) {
- return s.substring(s.lastIndexOf("/") + 1);
- }
-
-}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/PipelineMigrationHelpers.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/PipelineMigrationHelpers.java
deleted file mode 100644
index 2258cb9..0000000
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/PipelineMigrationHelpers.java
+++ /dev/null
@@ -1,21 +0,0 @@
-/*
- * 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.execution.pipeline;
-
-public class PipelineMigrationHelpers {
-}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/PipelineExecutor.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/PipelineExecutor.java
index 3e50d91..948ae4b 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/PipelineExecutor.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/PipelineExecutor.java
@@ -17,10 +17,8 @@
*/
package org.apache.streampipes.manager.execution.pipeline.executor;
-import org.apache.streampipes.manager.execution.pipeline.executor.operations.LifecycleEntity;
-import org.apache.streampipes.manager.execution.pipeline.executor.operations.PipelineExecutionOperation;
-import org.apache.streampipes.manager.execution.pipeline.executor.operations.types.MigrationOperation;
-import org.apache.streampipes.manager.execution.pipeline.executor.operations.types.ReconfigurationOperation;
+import org.apache.streampipes.manager.execution.pipeline.executor.steps.EntitiesLifecycleObject;
+import org.apache.streampipes.manager.execution.pipeline.executor.steps.PipelineExecutionStep;
import org.apache.streampipes.manager.execution.pipeline.executor.utils.StatusUtils;
import org.apache.streampipes.model.SpDataSet;
import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
@@ -48,16 +46,16 @@ public class PipelineExecutor {
private PipelineElementReconfigurationEntity reconfigurationEntity;
- private final LifecycleEntity<InvocableStreamPipesEntity> graphs;
+ private final EntitiesLifecycleObject<InvocableStreamPipesEntity> graphs;
- private final LifecycleEntity<SpDataStreamRelayContainer> relays;
+ private final EntitiesLifecycleObject<SpDataStreamRelayContainer> relays;
- private final LifecycleEntity<SpDataSet> dataSets;
+ private final EntitiesLifecycleObject<SpDataSet> dataSets;
private PipelineOperationStatus status;
- private final LinkedList<PipelineExecutionOperation> operations = new LinkedList<>();
+ private final LinkedList<PipelineExecutionStep> operations = new LinkedList<>();
public PipelineExecutor(Pipeline pipeline, boolean visualize, boolean storeStatus, boolean monitor){
this.pipeline = pipeline;
@@ -66,19 +64,19 @@ public class PipelineExecutor {
this.monitor = monitor;
this.status = StatusUtils.initPipelineOperationStatus(pipeline);
- this.graphs = new LifecycleEntity<>();
- this.relays = new LifecycleEntity<>();
- this.dataSets = new LifecycleEntity<>();
+ this.graphs = new EntitiesLifecycleObject<>();
+ this.relays = new EntitiesLifecycleObject<>();
+ this.dataSets = new EntitiesLifecycleObject<>();
}
public PipelineOperationStatus execute(){
- for(PipelineExecutionOperation pipelineExecutionOperation: this.operations){
- PipelineOperationStatus operationStatus = pipelineExecutionOperation.executeOperation();
+ for(PipelineExecutionStep pipelineExecutionStep : this.operations){
+ PipelineOperationStatus operationStatus = pipelineExecutionStep.executeOperation();
StatusUtils.checkSuccess(operationStatus);
- pipelineExecutionOperation.setStatus(operationStatus);
+ pipelineExecutionStep.setStatus(operationStatus);
StatusUtils.updateStatus(operationStatus, this.status);
if(!operationStatus.isSuccess()){
- rollback(pipelineExecutionOperation);
+ rollback(pipelineExecutionStep);
break;
}
}
@@ -86,11 +84,11 @@ public class PipelineExecutor {
return this.status;
}
- private void rollback(PipelineExecutionOperation failedOperation){
+ private void rollback(PipelineExecutionStep failedOperation){
PipelineOperationStatus rollbackStatus = StatusUtils.initPipelineOperationStatus(pipeline);
for(int currentOperationIndex = this.operations.indexOf(failedOperation);
- currentOperationIndex<=0; currentOperationIndex--){
- PipelineExecutionOperation currentOperation = this.operations.get(currentOperationIndex);
+ currentOperationIndex>=0; currentOperationIndex--){
+ PipelineExecutionStep currentOperation = this.operations.get(currentOperationIndex);
PipelineOperationStatus rollbackOperationStatus = currentOperation.rollbackOperation();
StatusUtils.checkSuccess(rollbackOperationStatus);
StatusUtils.updateStatus(rollbackOperationStatus, rollbackStatus);
@@ -99,18 +97,10 @@ public class PipelineExecutor {
StatusUtils.updateStatus(rollbackStatus, this.status);
}
- public void addOperation(PipelineExecutionOperation operation){
+ public void addStep(PipelineExecutionStep operation){
this.operations.add(operation);
}
- public boolean containsReconfigurationOperation(){
- return this.operations.stream().anyMatch(operation -> operation instanceof ReconfigurationOperation);
- }
-
- public boolean containsMigrationOperation(){
- return this.operations.stream().anyMatch(operation -> operation instanceof MigrationOperation);
- }
-
//Getter and Setter
@@ -178,15 +168,15 @@ public class PipelineExecutor {
this.reconfigurationEntity = reconfigurationEntity;
}
- public LifecycleEntity<InvocableStreamPipesEntity> getGraphs() {
+ public EntitiesLifecycleObject<InvocableStreamPipesEntity> getGraphs() {
return graphs;
}
- public LifecycleEntity<SpDataStreamRelayContainer> getRelays() {
+ public EntitiesLifecycleObject<SpDataStreamRelayContainer> getRelays() {
return relays;
}
- public LifecycleEntity<SpDataSet> getDataSets() {
+ public EntitiesLifecycleObject<SpDataSet> getDataSets() {
return dataSets;
}
}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/PipelineExecutorBuilder.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/PipelineExecutorBuilder.java
index d423cf1..250cde7 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/PipelineExecutorBuilder.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/PipelineExecutorBuilder.java
@@ -17,8 +17,7 @@
*/
package org.apache.streampipes.manager.execution.pipeline.executor;
-import org.apache.streampipes.commons.exceptions.SpRuntimeException;
-import org.apache.streampipes.manager.execution.pipeline.executor.operations.*;
+import org.apache.streampipes.manager.execution.pipeline.executor.steps.*;
import org.apache.streampipes.model.pipeline.Pipeline;
import org.apache.streampipes.model.pipeline.PipelineElementMigrationEntity;
import org.apache.streampipes.model.pipeline.PipelineElementReconfigurationEntity;
@@ -27,10 +26,6 @@ public class PipelineExecutorBuilder {
private PipelineExecutor pipelineExecutor;
- private boolean reconfigurationParametersSet = false;
-
- private boolean migrationParametersSet = false;
-
public static PipelineExecutorBuilder getBuilder(){
return new PipelineExecutorBuilder();
}
@@ -46,7 +41,6 @@ public class PipelineExecutorBuilder {
PipelineElementMigrationEntity migrationEntity){
pipelineExecutor.setSecondaryPipeline(pipelineBeforeMigration);
pipelineExecutor.setMigrationEntity(migrationEntity);
- this.migrationParametersSet = true;
return this;
}
@@ -54,76 +48,70 @@ public class PipelineExecutorBuilder {
PipelineElementReconfigurationEntity reconfigurationEntity){
pipelineExecutor.setSecondaryPipeline(reconfiguredPipeline);
pipelineExecutor.setReconfigurationEntity(reconfigurationEntity);
- this.reconfigurationParametersSet = true;
return this;
}
- public PipelineExecutorBuilder addPrepareMigrationOperation(){
- pipelineExecutor.addOperation(new PrepareMigrationOperation(pipelineExecutor));
+ public PipelineExecutorBuilder addPrepareMigrationStep(){
+ pipelineExecutor.addStep(new PrepareMigrationStep(pipelineExecutor));
return this;
}
- public PipelineExecutorBuilder addGetStateOperation(){
- pipelineExecutor.addOperation(new GetStateOperation(pipelineExecutor));
+ public PipelineExecutorBuilder addGetStateStep(){
+ pipelineExecutor.addStep(new GetStateStep(pipelineExecutor));
return this;
}
- public PipelineExecutorBuilder addStartRelaysOperation(){
- pipelineExecutor.addOperation(new StartRelaysOperation(pipelineExecutor));
+ public PipelineExecutorBuilder addStartRelaysStep(){
+ pipelineExecutor.addStep(new StartRelaysStep(pipelineExecutor));
return this;
}
- public PipelineExecutorBuilder addStartGraphsAndAssociatedRelaysOperation(){
- pipelineExecutor.addOperation(new StartGraphsAndAssociatedRelaysOperation(pipelineExecutor));
+ public PipelineExecutorBuilder addStartGraphsAndAssociatedRelaysStep(){
+ pipelineExecutor.addStep(new StartGraphsAndAssociatedRelaysStep(pipelineExecutor));
return this;
}
- public PipelineExecutorBuilder addStopGraphsAndAssociatedRelaysOperation(){
- pipelineExecutor.addOperation(new StopGraphsAndAssociatedRelaysOperation(pipelineExecutor));
+ public PipelineExecutorBuilder addStopGraphsAndAssociatedRelaysStep(){
+ pipelineExecutor.addStep(new StopGraphsAndAssociatedRelaysStep(pipelineExecutor));
return this;
}
- public PipelineExecutorBuilder addStopRelaysOperation(){
- pipelineExecutor.addOperation(new StopRelaysOperation(pipelineExecutor));
+ public PipelineExecutorBuilder addStopRelaysStep(){
+ pipelineExecutor.addStep(new StopRelaysStep(pipelineExecutor));
return this;
}
- public PipelineExecutorBuilder addStoreMigratedPipelineOperation(){
- pipelineExecutor.addOperation(new StoreMigratedPipelineOperation(pipelineExecutor));
+ public PipelineExecutorBuilder addStoreMigratedPipelineStep(){
+ pipelineExecutor.addStep(new StoreMigratedPipelineStep(pipelineExecutor));
return this;
}
- public PipelineExecutorBuilder addReconfigureElementOperation(){
- pipelineExecutor.addOperation(new ReconfigureElementOperation(pipelineExecutor));
+ public PipelineExecutorBuilder addReconfigureElementStep(){
+ pipelineExecutor.addStep(new ReconfigureElementStep(pipelineExecutor));
return this;
}
- public PipelineExecutorBuilder addPreparePipelineStartOperation(){
- pipelineExecutor.addOperation(new PrepareStartPipelineOperation(pipelineExecutor));
+ public PipelineExecutorBuilder addPreparePipelineStartStep(){
+ pipelineExecutor.addStep(new PrepareStartPipelineStep(pipelineExecutor));
return this;
}
- public PipelineExecutorBuilder addStartPipelineOperation(){
- pipelineExecutor.addOperation(new StartPipelineOperation(pipelineExecutor));
+ public PipelineExecutorBuilder addStartPipelineStep(){
+ pipelineExecutor.addStep(new StartPipelineStep(pipelineExecutor));
return this;
}
- public PipelineExecutorBuilder addStorePipelineOperation(){
- pipelineExecutor.addOperation(new StorePipelineOperation(pipelineExecutor));
+ public PipelineExecutorBuilder addStorePipelineStep(){
+ pipelineExecutor.addStep(new StorePipelineStep(pipelineExecutor));
return this;
}
- public PipelineExecutorBuilder addStopPipelineOperation(){
- pipelineExecutor.addOperation(new StopPipelineOperation(pipelineExecutor));
+ public PipelineExecutorBuilder addStopPipelineStep(){
+ pipelineExecutor.addStep(new StopPipelineStep(pipelineExecutor));
return this;
}
public PipelineExecutor buildPipelineExecutor(){
- //Is this check needed? Only relevant for core development not for users but gives a little more clarity at the
- //cost of introducing some new boolean flags and marker interfaces
- if((this.pipelineExecutor.containsMigrationOperation() && !this.migrationParametersSet)
- || (this.pipelineExecutor.containsReconfigurationOperation() && !this.reconfigurationParametersSet))
- throw new SpRuntimeException("PipelineExecutor can't be build since the required parameters have not been set");
- return this.pipelineExecutor; }
-
+ return this.pipelineExecutor;
+ }
}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/PipelineExecutorFactory.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/PipelineExecutorFactory.java
index 617ff01..e4b042b 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/PipelineExecutorFactory.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/PipelineExecutorFactory.java
@@ -29,14 +29,14 @@ public class PipelineExecutorFactory {
PipelineExecutorBuilder builder = PipelineExecutorBuilder.getBuilder()
.initializePipelineExecutor(pipeline, visualize, storeStatus, monitor)
.setMigrationParameters(pipelineBeforeMigration, migrationEntity)
- .addPrepareMigrationOperation();
+ .addPrepareMigrationStep();
if(migrationEntity.getTargetElement().isStateful())
- builder.addGetStateOperation();
- builder.addStartGraphsAndAssociatedRelaysOperation()
- .addStopRelaysOperation()
- .addStartRelaysOperation()
- .addStopGraphsAndAssociatedRelaysOperation()
- .addStoreMigratedPipelineOperation();
+ builder.addGetStateStep();
+ builder.addStartGraphsAndAssociatedRelaysStep()
+ .addStopRelaysStep()
+ .addStartRelaysStep()
+ .addStopGraphsAndAssociatedRelaysStep()
+ .addStoreMigratedPipelineStep();
return builder.buildPipelineExecutor();
}
@@ -44,23 +44,23 @@ public class PipelineExecutorFactory {
boolean monitor, Pipeline reconfiguredPipeline,
PipelineElementReconfigurationEntity reconfigurationEntity){
return PipelineExecutorBuilder.getBuilder().initializePipelineExecutor(pipeline, visualize, storeStatus, monitor)
- .setReconfigurationParameters(reconfiguredPipeline, reconfigurationEntity).addReconfigureElementOperation()
+ .setReconfigurationParameters(reconfiguredPipeline, reconfigurationEntity).addReconfigureElementStep()
.buildPipelineExecutor();
}
public static PipelineExecutor createInvocationExecutor(Pipeline pipeline, boolean visualize, boolean storeStatus,
boolean monitor){
return PipelineExecutorBuilder.getBuilder().initializePipelineExecutor(pipeline, visualize, storeStatus, monitor)
- .addPreparePipelineStartOperation()
- .addStartPipelineOperation()
- .addStorePipelineOperation()
+ .addPreparePipelineStartStep()
+ .addStartPipelineStep()
+ .addStorePipelineStep()
.buildPipelineExecutor();
}
public static PipelineExecutor createDetachExecutor(Pipeline pipeline, boolean visualize, boolean storeStatus,
boolean monitor){
return PipelineExecutorBuilder.getBuilder().initializePipelineExecutor(pipeline, visualize, storeStatus, monitor)
- .addStopPipelineOperation().buildPipelineExecutor();
+ .addStopPipelineStep().buildPipelineExecutor();
}
}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/types/MigrationOperation.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/types/MigrationOperation.java
deleted file mode 100644
index 22c21d1..0000000
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/types/MigrationOperation.java
+++ /dev/null
@@ -1,21 +0,0 @@
-/*
- * 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.execution.pipeline.executor.operations.types;
-
-public interface MigrationOperation {
-}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/types/ReconfigurationOperation.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/types/ReconfigurationOperation.java
deleted file mode 100644
index 13c67a0..0000000
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/types/ReconfigurationOperation.java
+++ /dev/null
@@ -1,21 +0,0 @@
-/*
- * 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.execution.pipeline.executor.operations.types;
-
-public interface ReconfigurationOperation {
-}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/LifecycleEntity.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/EntitiesLifecycleObject.java
similarity index 94%
rename from streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/LifecycleEntity.java
rename to streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/EntitiesLifecycleObject.java
index d2c6d8c..5a2ca91 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/LifecycleEntity.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/EntitiesLifecycleObject.java
@@ -15,12 +15,12 @@
* limitations under the License.
*
*/
-package org.apache.streampipes.manager.execution.pipeline.executor.operations;
+package org.apache.streampipes.manager.execution.pipeline.executor.steps;
import java.util.ArrayList;
import java.util.List;
-public class LifecycleEntity<T> {
+public class EntitiesLifecycleObject<T> {
private final List<T> entitiesToStart;
@@ -30,7 +30,7 @@ public class LifecycleEntity<T> {
private final List<T> entitiesToDelete;
- public LifecycleEntity(){
+ public EntitiesLifecycleObject(){
entitiesToStart = new ArrayList<>();
entitiesToStop = new ArrayList<>();
entitiesToStore = new ArrayList<>();
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/GetStateOperation.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/GetStateStep.java
similarity index 93%
rename from streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/GetStateOperation.java
rename to streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/GetStateStep.java
index e11609b..1fc19f8 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/GetStateOperation.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/GetStateStep.java
@@ -15,10 +15,9 @@
* limitations under the License.
*
*/
-package org.apache.streampipes.manager.execution.pipeline.executor.operations;
+package org.apache.streampipes.manager.execution.pipeline.executor.steps;
import org.apache.streampipes.logging.evaluation.EvaluationLogger;
-import org.apache.streampipes.manager.execution.pipeline.executor.operations.types.MigrationOperation;
import org.apache.streampipes.manager.execution.pipeline.executor.utils.CommunicationUtils;
import org.apache.streampipes.manager.execution.pipeline.executor.PipelineExecutor;
import org.apache.streampipes.manager.execution.pipeline.executor.utils.StatusUtils;
@@ -29,9 +28,9 @@ import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectOutputStream;
-public class GetStateOperation extends PipelineExecutionOperation implements MigrationOperation {
+public class GetStateStep extends PipelineExecutionStep {
- public GetStateOperation(PipelineExecutor pipelineExecutor) {
+ public GetStateStep(PipelineExecutor pipelineExecutor) {
super(pipelineExecutor);
}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/PipelineExecutionOperation.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/PipelineExecutionStep.java
similarity index 90%
rename from streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/PipelineExecutionOperation.java
rename to streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/PipelineExecutionStep.java
index e822ec4..098de93 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/PipelineExecutionOperation.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/PipelineExecutionStep.java
@@ -15,25 +15,26 @@
* limitations under the License.
*
*/
-package org.apache.streampipes.manager.execution.pipeline.executor.operations;
+package org.apache.streampipes.manager.execution.pipeline.executor.steps;
import org.apache.streampipes.manager.execution.pipeline.executor.PipelineExecutor;
import org.apache.streampipes.manager.execution.pipeline.executor.utils.StatusUtils;
import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
-public abstract class PipelineExecutionOperation {
+public abstract class PipelineExecutionStep {
protected final PipelineExecutor pipelineExecutor;
private PipelineOperationStatus status;
- public PipelineExecutionOperation(PipelineExecutor pipelineExecutor){
+ public PipelineExecutionStep(PipelineExecutor pipelineExecutor){
this.pipelineExecutor = pipelineExecutor;
this.status = StatusUtils.initPipelineOperationStatus(this.pipelineExecutor.getPipeline());
}
public abstract PipelineOperationStatus executeOperation();
+ //TODO: Check if partial and full rollback can be unified to single method
public abstract PipelineOperationStatus rollbackOperationPartially();
public abstract PipelineOperationStatus rollbackOperationFully();
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/PrepareMigrationOperation.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/PrepareMigrationStep.java
similarity index 95%
rename from streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/PrepareMigrationOperation.java
rename to streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/PrepareMigrationStep.java
index 4c9304d..4c45b15 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/PrepareMigrationOperation.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/PrepareMigrationStep.java
@@ -15,13 +15,12 @@
* limitations under the License.
*
*/
-package org.apache.streampipes.manager.execution.pipeline.executor.operations;
+package org.apache.streampipes.manager.execution.pipeline.executor.steps;
import org.apache.streampipes.manager.data.PipelineGraph;
import org.apache.streampipes.manager.data.PipelineGraphBuilder;
import org.apache.streampipes.manager.data.PipelineGraphHelpers;
import org.apache.streampipes.manager.execution.pipeline.executor.PipelineExecutor;
-import org.apache.streampipes.manager.execution.pipeline.executor.operations.types.MigrationOperation;
import org.apache.streampipes.manager.execution.pipeline.executor.utils.PipelineElementUtils;
import org.apache.streampipes.manager.execution.pipeline.executor.utils.PipelineUtils;
import org.apache.streampipes.manager.execution.pipeline.executor.utils.RelayUtils;
@@ -37,13 +36,13 @@ import java.util.Collections;
import java.util.List;
import java.util.stream.Collectors;
-public class PrepareMigrationOperation extends PipelineExecutionOperation implements MigrationOperation {
+public class PrepareMigrationStep extends PipelineExecutionStep {
private List<NamedStreamPipesEntity> predecessorsAfterMigration;
private final List<NamedStreamPipesEntity> predecessorsBeforeMigration = new ArrayList<>();
private Pipeline pipeline;
- public PrepareMigrationOperation(PipelineExecutor pipelineExecutor) {
+ public PrepareMigrationStep(PipelineExecutor pipelineExecutor) {
super(pipelineExecutor);
}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/PrepareStartPipelineOperation.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/PrepareStartPipelineStep.java
similarity index 96%
rename from streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/PrepareStartPipelineOperation.java
rename to streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/PrepareStartPipelineStep.java
index 264ac19..ca9dc66 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/PrepareStartPipelineOperation.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/PrepareStartPipelineStep.java
@@ -15,7 +15,7 @@
* limitations under the License.
*
*/
-package org.apache.streampipes.manager.execution.pipeline.executor.operations;
+package org.apache.streampipes.manager.execution.pipeline.executor.steps;
import org.apache.streampipes.manager.execution.pipeline.executor.PipelineExecutor;
import org.apache.streampipes.manager.execution.pipeline.executor.utils.PipelineElementUtils;
@@ -35,10 +35,10 @@ import java.util.List;
import java.util.UUID;
import java.util.stream.Collectors;
-public class PrepareStartPipelineOperation extends PipelineExecutionOperation {
+public class PrepareStartPipelineStep extends PipelineExecutionStep {
- public PrepareStartPipelineOperation(PipelineExecutor pipelineExecutor) {
+ public PrepareStartPipelineStep(PipelineExecutor pipelineExecutor) {
super(pipelineExecutor);
}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/ReconfigureElementOperation.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/ReconfigureElementStep.java
similarity index 92%
rename from streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/ReconfigureElementOperation.java
rename to streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/ReconfigureElementStep.java
index 2e66272..2dddf83 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/ReconfigureElementOperation.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/ReconfigureElementStep.java
@@ -15,11 +15,10 @@
* limitations under the License.
*
*/
-package org.apache.streampipes.manager.execution.pipeline.executor.operations;
+package org.apache.streampipes.manager.execution.pipeline.executor.steps;
import org.apache.streampipes.manager.execution.http.ReconfigurationSubmitter;
import org.apache.streampipes.manager.execution.pipeline.executor.PipelineExecutor;
-import org.apache.streampipes.manager.execution.pipeline.executor.operations.types.ReconfigurationOperation;
import org.apache.streampipes.manager.execution.pipeline.executor.utils.StatusUtils;
import org.apache.streampipes.model.graph.DataProcessorInvocation;
import org.apache.streampipes.model.pipeline.Pipeline;
@@ -31,9 +30,9 @@ import java.util.List;
import java.util.Optional;
import java.util.stream.Collectors;
-public class ReconfigureElementOperation extends PipelineExecutionOperation implements ReconfigurationOperation {
+public class ReconfigureElementStep extends PipelineExecutionStep {
- public ReconfigureElementOperation(PipelineExecutor pipelineExecutor) {
+ public ReconfigureElementStep(PipelineExecutor pipelineExecutor) {
super(pipelineExecutor);
}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StartGraphsAndAssociatedRelaysOperation.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/StartGraphsAndAssociatedRelaysStep.java
similarity index 91%
rename from streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StartGraphsAndAssociatedRelaysOperation.java
rename to streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/StartGraphsAndAssociatedRelaysStep.java
index 72d3802..ae976e3 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StartGraphsAndAssociatedRelaysOperation.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/StartGraphsAndAssociatedRelaysStep.java
@@ -15,11 +15,10 @@
* limitations under the License.
*
*/
-package org.apache.streampipes.manager.execution.pipeline.executor.operations;
+package org.apache.streampipes.manager.execution.pipeline.executor.steps;
import org.apache.streampipes.logging.evaluation.EvaluationLogger;
import org.apache.streampipes.manager.execution.pipeline.executor.*;
-import org.apache.streampipes.manager.execution.pipeline.executor.operations.types.MigrationOperation;
import org.apache.streampipes.manager.execution.pipeline.executor.utils.*;
import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
import org.apache.streampipes.model.eventrelay.SpDataStreamRelayContainer;
@@ -28,9 +27,9 @@ import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
import java.util.List;
import java.util.Set;
-public class StartGraphsAndAssociatedRelaysOperation extends PipelineExecutionOperation implements MigrationOperation {
+public class StartGraphsAndAssociatedRelaysStep extends PipelineExecutionStep {
- public StartGraphsAndAssociatedRelaysOperation(PipelineExecutor pipelineExecutor){
+ public StartGraphsAndAssociatedRelaysStep(PipelineExecutor pipelineExecutor){
super(pipelineExecutor);
}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StartPipelineOperation.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/StartPipelineStep.java
similarity index 96%
rename from streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StartPipelineOperation.java
rename to streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/StartPipelineStep.java
index 720d8c5..19d465d 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StartPipelineOperation.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/StartPipelineStep.java
@@ -15,7 +15,7 @@
* limitations under the License.
*
*/
-package org.apache.streampipes.manager.execution.pipeline.executor.operations;
+package org.apache.streampipes.manager.execution.pipeline.executor.steps;
import org.apache.streampipes.manager.execution.http.GraphSubmitter;
import org.apache.streampipes.manager.execution.pipeline.executor.PipelineExecutor;
@@ -33,9 +33,9 @@ import java.util.List;
import java.util.Set;
-public class StartPipelineOperation extends PipelineExecutionOperation{
+public class StartPipelineStep extends PipelineExecutionStep {
- public StartPipelineOperation(PipelineExecutor pipelineExecutor) {
+ public StartPipelineStep(PipelineExecutor pipelineExecutor) {
super(pipelineExecutor);
}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StartRelaysOperation.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/StartRelaysStep.java
similarity index 90%
rename from streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StartRelaysOperation.java
rename to streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/StartRelaysStep.java
index 07c6480..07dc88f 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StartRelaysOperation.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/StartRelaysStep.java
@@ -15,10 +15,9 @@
* limitations under the License.
*
*/
-package org.apache.streampipes.manager.execution.pipeline.executor.operations;
+package org.apache.streampipes.manager.execution.pipeline.executor.steps;
import org.apache.streampipes.logging.evaluation.EvaluationLogger;
-import org.apache.streampipes.manager.execution.pipeline.executor.operations.types.MigrationOperation;
import org.apache.streampipes.manager.execution.pipeline.executor.utils.CommunicationUtils;
import org.apache.streampipes.manager.execution.pipeline.executor.PipelineExecutor;
import org.apache.streampipes.manager.execution.pipeline.executor.utils.RelayUtils;
@@ -29,10 +28,10 @@ import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
import java.util.List;
import java.util.Set;
-public class StartRelaysOperation extends PipelineExecutionOperation implements MigrationOperation {
+public class StartRelaysStep extends PipelineExecutionStep {
- public StartRelaysOperation(PipelineExecutor pipelineExecutor) {
+ public StartRelaysStep(PipelineExecutor pipelineExecutor) {
super(pipelineExecutor);
}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StopGraphsAndAssociatedRelaysOperation.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/StopGraphsAndAssociatedRelaysStep.java
similarity index 92%
rename from streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StopGraphsAndAssociatedRelaysOperation.java
rename to streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/StopGraphsAndAssociatedRelaysStep.java
index 2e1a0d0..810f374 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StopGraphsAndAssociatedRelaysOperation.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/StopGraphsAndAssociatedRelaysStep.java
@@ -15,10 +15,9 @@
* limitations under the License.
*
*/
-package org.apache.streampipes.manager.execution.pipeline.executor.operations;
+package org.apache.streampipes.manager.execution.pipeline.executor.steps;
import org.apache.streampipes.logging.evaluation.EvaluationLogger;
-import org.apache.streampipes.manager.execution.pipeline.executor.operations.types.MigrationOperation;
import org.apache.streampipes.manager.execution.pipeline.executor.utils.CommunicationUtils;
import org.apache.streampipes.manager.execution.pipeline.executor.utils.PipelineElementUtils;
import org.apache.streampipes.manager.execution.pipeline.executor.PipelineExecutor;
@@ -31,9 +30,9 @@ import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
import java.util.List;
import java.util.Set;
-public class StopGraphsAndAssociatedRelaysOperation extends PipelineExecutionOperation implements MigrationOperation {
+public class StopGraphsAndAssociatedRelaysStep extends PipelineExecutionStep {
- public StopGraphsAndAssociatedRelaysOperation(PipelineExecutor pipelineExecutor) {
+ public StopGraphsAndAssociatedRelaysStep(PipelineExecutor pipelineExecutor) {
super(pipelineExecutor);
}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StopPipelineOperation.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/StopPipelineStep.java
similarity index 84%
rename from streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StopPipelineOperation.java
rename to streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/StopPipelineStep.java
index 1f28247..473b039 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StopPipelineOperation.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/StopPipelineStep.java
@@ -15,7 +15,7 @@
* limitations under the License.
*
*/
-package org.apache.streampipes.manager.execution.pipeline.executor.operations;
+package org.apache.streampipes.manager.execution.pipeline.executor.steps;
import org.apache.streampipes.manager.execution.http.GraphSubmitter;
import org.apache.streampipes.manager.execution.pipeline.executor.PipelineExecutor;
@@ -33,9 +33,9 @@ import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
import java.util.List;
import java.util.Set;
-public class StopPipelineOperation extends PipelineExecutionOperation {
+public class StopPipelineStep extends PipelineExecutionStep {
- public StopPipelineOperation(PipelineExecutor pipelineExecutor) {
+ public StopPipelineStep(PipelineExecutor pipelineExecutor) {
super(pipelineExecutor);
}
@@ -70,18 +70,16 @@ public class StopPipelineOperation extends PipelineExecutionOperation {
Set<String> idsToRollback = StatusUtils.extractUniqueSuccessfulIds(this.getStatus());
+ List<InvocableStreamPipesEntity> graphs = TemporaryGraphStorage.graphStorage.get(pipeline.getPipelineId());
+ List<SpDataSet> dataSets = TemporaryGraphStorage.datasetStorage.get(pipeline.getPipelineId());
+ List<SpDataStreamRelayContainer> relays = PipelineElementUtils.generateRelays(graphs, pipeline);
+
List<InvocableStreamPipesEntity> graphsToRollBack =
- PipelineElementUtils.filterPipelineElementsById(
- pipelineExecutor.getGraphs().getEntitiesToStop(),
- idsToRollback);
+ PipelineElementUtils.filterPipelineElementsById(graphs, idsToRollback);
List<SpDataSet> dataSetsToRollBack =
- DataSetUtils.filterDataSetsById(
- pipelineExecutor.getDataSets().getEntitiesToStart(),
- idsToRollback);
+ DataSetUtils.filterDataSetsById(dataSets, idsToRollback);
List<SpDataStreamRelayContainer> relaysToRollBack =
- RelayUtils.filterRelaysById(
- pipelineExecutor.getRelays().getEntitiesToStart(),
- idsToRollback);
+ RelayUtils.filterRelaysById(relays, idsToRollback);
return new GraphSubmitter(pipeline.getPipelineId(), pipeline.getName(),
graphsToRollBack,
@@ -91,7 +89,6 @@ public class StopPipelineOperation extends PipelineExecutionOperation {
@Override
public PipelineOperationStatus rollbackOperationFully() {
- //TODO: Implement sth?
- return StatusUtils.initPipelineOperationStatus(pipelineExecutor.getPipeline());
+ return rollbackOperationPartially();
}
}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StopRelaysOperation.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/StopRelaysStep.java
similarity index 90%
rename from streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StopRelaysOperation.java
rename to streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/StopRelaysStep.java
index dee1933..cc34c52 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StopRelaysOperation.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/StopRelaysStep.java
@@ -15,11 +15,10 @@
* limitations under the License.
*
*/
-package org.apache.streampipes.manager.execution.pipeline.executor.operations;
+package org.apache.streampipes.manager.execution.pipeline.executor.steps;
import org.apache.streampipes.logging.evaluation.EvaluationLogger;
import org.apache.streampipes.manager.execution.pipeline.executor.*;
-import org.apache.streampipes.manager.execution.pipeline.executor.operations.types.MigrationOperation;
import org.apache.streampipes.manager.execution.pipeline.executor.utils.CommunicationUtils;
import org.apache.streampipes.manager.execution.pipeline.executor.utils.RelayUtils;
import org.apache.streampipes.manager.execution.pipeline.executor.utils.StatusUtils;
@@ -29,9 +28,9 @@ import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
import java.util.List;
import java.util.Set;
-public class StopRelaysOperation extends PipelineExecutionOperation implements MigrationOperation {
+public class StopRelaysStep extends PipelineExecutionStep {
- public StopRelaysOperation(PipelineExecutor pipelineExecutor){
+ public StopRelaysStep(PipelineExecutor pipelineExecutor){
super(pipelineExecutor);
}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StoreMigratedPipelineOperation.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/StoreMigratedPipelineStep.java
similarity index 93%
rename from streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StoreMigratedPipelineOperation.java
rename to streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/StoreMigratedPipelineStep.java
index 895e38d..a300683 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StoreMigratedPipelineOperation.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/StoreMigratedPipelineStep.java
@@ -15,7 +15,7 @@
* limitations under the License.
*
*/
-package org.apache.streampipes.manager.execution.pipeline.executor.operations;
+package org.apache.streampipes.manager.execution.pipeline.executor.steps;
import org.apache.streampipes.manager.execution.pipeline.executor.PipelineExecutor;
import org.apache.streampipes.manager.execution.pipeline.executor.utils.PipelineUtils;
@@ -26,9 +26,9 @@ import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
import java.util.List;
-public class StoreMigratedPipelineOperation extends PipelineExecutionOperation{
+public class StoreMigratedPipelineStep extends PipelineExecutionStep {
- public StoreMigratedPipelineOperation(PipelineExecutor pipelineExecutor) {
+ public StoreMigratedPipelineStep(PipelineExecutor pipelineExecutor) {
super(pipelineExecutor);
}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StorePipelineOperation.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/StorePipelineStep.java
similarity index 95%
rename from streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StorePipelineOperation.java
rename to streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/StorePipelineStep.java
index 960aec0..98def31 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StorePipelineOperation.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/steps/StorePipelineStep.java
@@ -15,7 +15,7 @@
* limitations under the License.
*
*/
-package org.apache.streampipes.manager.execution.pipeline.executor.operations;
+package org.apache.streampipes.manager.execution.pipeline.executor.steps;
import org.apache.streampipes.manager.execution.pipeline.executor.PipelineExecutor;
import org.apache.streampipes.manager.execution.pipeline.executor.utils.StatusUtils;
@@ -26,9 +26,9 @@ import org.apache.streampipes.model.message.PipelineStatusMessageType;
import org.apache.streampipes.model.pipeline.Pipeline;
import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
-public class StorePipelineOperation extends PipelineExecutionOperation{
+public class StorePipelineStep extends PipelineExecutionStep {
- public StorePipelineOperation(PipelineExecutor pipelineExecutor) {
+ public StorePipelineStep(PipelineExecutor pipelineExecutor) {
super(pipelineExecutor);
}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/RelayUtils.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/RelayUtils.java
index ac0b210..7d8fae8 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/RelayUtils.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/RelayUtils.java
@@ -237,7 +237,7 @@ public class RelayUtils {
public static List<SpDataStreamRelayContainer> filterRelaysById(List<SpDataStreamRelayContainer> relays,
Set<String> relayIds) {
return relays.stream().
- filter(relay -> relayIds.contains(relay.getRunningStreamRelayInstanceId()))
+ filter(relay -> relayIds.contains(relay.getRunningStreamRelayInstanceId() + " relay"))
.collect(Collectors.toList());
}
}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/StatusUtils.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/StatusUtils.java
index c4d8a4f..381bebf 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/StatusUtils.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/StatusUtils.java
@@ -50,7 +50,7 @@ public class StatusUtils {
public static Set<String> extractUniqueSuccessfulIds(PipelineOperationStatus status) {
return status.getElementStatus().stream()
.filter(PipelineElementStatus::isSuccess)
- .map(PipelineElementStatus::getElementId)
+ .map(PipelineElementStatus::getRunningInstanceId)
.collect(Collectors.toSet());
}
diff --git a/ui/src/app/editor/dialog/migrate-pipeline-processors/migrate-pipeline-processors.component.ts b/ui/src/app/editor/dialog/migrate-pipeline-processors/migrate-pipeline-processors.component.ts
index eb34d39..46af116 100644
--- a/ui/src/app/editor/dialog/migrate-pipeline-processors/migrate-pipeline-processors.component.ts
+++ b/ui/src/app/editor/dialog/migrate-pipeline-processors/migrate-pipeline-processors.component.ts
@@ -170,26 +170,29 @@ export class MigratePipelineProcessorsComponent implements OnInit {
modifyPipelineElementsDeployments(pipelineElements) {
pipelineElements.forEach(p => {
- let selectedTargetNodeId = p.deploymentTargetNodeId
+ const selectedTargetNodeId = p.deploymentTargetNodeId
// Currently relay only for data processors
if (p instanceof DataProcessorInvocation) {
p.eventRelayStrategy = this.selectedRelayStrategyVal;
}
- if(selectedTargetNodeId != "default") {
- let selectedNode = this.edgeNodes
- .filter(node => node.nodeControllerId === selectedTargetNodeId)
+ if (selectedTargetNodeId !== 'default') {
+ const selectedNode = this.edgeNodes
+ .filter(node => node.nodeControllerId === selectedTargetNodeId);
p.deploymentTargetNodeHostname = selectedNode
- .map(node => node.hostname)[0]
+ .map(node => node.hostname)[0];
p.deploymentTargetNodePort = selectedNode
- .map(node => node.port)[0]
- }
- else {
- p.deploymentTargetNodeHostname = null
- p.deploymentTargetNodePort = null
+ .map(node => node.port)[0];
+
+ p.elementEndpointHostname = selectedNode
+ .map(node => node.hostname)[0];
+ } else {
+ p.deploymentTargetNodeHostname = null;
+ p.deploymentTargetNodePort = null;
+ p.elementEndpointHostname = null;
}
})
}
[incubator-streampipes] 02/03: [WIP] refactor pipeline executor
Posted by go...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
gomm pushed a commit to branch edge-extensions
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit 47c6a8006f23a1a9b3091dacad743dd832e5b205
Author: daniel-gomm <da...@outlook.de>
AuthorDate: Mon Dec 13 19:42:45 2021 +0100
[WIP] refactor pipeline executor
---
.../PipelineElementReconfigurationEntity.java | 10 +
.../PipelineElementReconfigurationExecutor.java | 41 --
.../execution/pipeline/PipelineExecutor.java | 110 -----
.../pipeline/PipelineMigrationExecutor.java | 461 ---------------------
.../pipeline/executor/PipelineExecutor.java | 139 ++-----
.../pipeline/executor/PipelineExecutorBuilder.java | 21 +-
.../pipeline/executor/PipelineExecutorFactory.java | 13 +-
.../executor/operations/GetStateOperation.java | 18 +-
.../executor/operations/LifecycleEntity.java | 17 +
.../operations/PipelineExecutionOperation.java | 6 +-
.../operations/PrepareMigrationOperation.java | 73 +++-
...ion.java => PrepareStartPipelineOperation.java} | 27 +-
.../operations/ReconfigureElementOperation.java | 46 +-
...> StartGraphsAndAssociatedRelaysOperation.java} | 52 +--
.../operations/StartPipelineOperation.java | 88 ++--
...orsOperation.java => StartRelaysOperation.java} | 36 +-
...=> StopGraphsAndAssociatedRelaysOperation.java} | 38 +-
.../executor/operations/StopPipelineOperation.java | 36 +-
...ssorOperation.java => StopRelaysOperation.java} | 32 +-
.../operations/StoreMigratedPipelineOperation.java | 21 +-
.../operations/StorePipelineOperation.java | 16 +-
.../pipeline/executor/utils/DataSetUtils.java | 20 +
.../executor/utils/PipelineElementUtils.java | 7 +
.../pipeline/executor/utils/RelayUtils.java | 7 +
.../pipeline/executor/utils/StatusUtils.java | 2 +-
.../migration/PipelineElementMigrationHandler.java | 3 -
.../PipelineElementReconfigurationHandler.java | 10 +-
27 files changed, 384 insertions(+), 966 deletions(-)
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/PipelineElementReconfigurationEntity.java b/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/PipelineElementReconfigurationEntity.java
index e6efba8..ba621a3 100644
--- a/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/PipelineElementReconfigurationEntity.java
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/PipelineElementReconfigurationEntity.java
@@ -17,6 +17,7 @@
*/
package org.apache.streampipes.model.pipeline;
+import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
import org.apache.streampipes.model.staticproperty.StaticProperty;
import java.util.ArrayList;
@@ -47,6 +48,15 @@ public class PipelineElementReconfigurationEntity {
this.reconfiguredStaticProperties = reconfiguredStaticProperties;
}
+ public PipelineElementReconfigurationEntity(InvocableStreamPipesEntity entity){
+ this.deploymentRunningInstanceId = entity.getDeploymentRunningInstanceId();
+ this.pipelineElementName = entity.getName();
+ this.deploymentTargetNodeId = entity.getDeploymentTargetNodeId();
+ this.deploymentTargetNodeHostname = entity.getDeploymentTargetNodeHostname();
+ this.deploymentTargetNodePort = entity.getDeploymentTargetNodePort();
+ this.reconfiguredStaticProperties = new ArrayList<>();
+ }
+
public String getDeploymentRunningInstanceId() {
return deploymentRunningInstanceId;
}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/PipelineElementReconfigurationExecutor.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/PipelineElementReconfigurationExecutor.java
deleted file mode 100644
index dc61206..0000000
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/PipelineElementReconfigurationExecutor.java
+++ /dev/null
@@ -1,41 +0,0 @@
-/*
- * 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.execution.pipeline;
-
-import org.apache.streampipes.manager.execution.http.ReconfigurationSubmitter;
-import org.apache.streampipes.model.pipeline.Pipeline;
-import org.apache.streampipes.model.pipeline.PipelineElementReconfigurationEntity;
-import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
-
-public class PipelineElementReconfigurationExecutor {
-
- private final Pipeline reconfiguredPipeline;
- private final PipelineElementReconfigurationEntity reconfigurationEntity;
-
- public PipelineElementReconfigurationExecutor(Pipeline reconfiguredPipeline,
- PipelineElementReconfigurationEntity reconfigurationEntity) {
- this.reconfiguredPipeline = reconfiguredPipeline;
- this.reconfigurationEntity = reconfigurationEntity;
- }
-
- public PipelineOperationStatus reconfigurePipelineElement() {
- return new ReconfigurationSubmitter(reconfiguredPipeline.getPipelineId(), reconfiguredPipeline.getName(),
- reconfigurationEntity).reconfigure();
- }
-
-}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/PipelineExecutor.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/PipelineExecutor.java
deleted file mode 100644
index 17aa1ce..0000000
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/PipelineExecutor.java
+++ /dev/null
@@ -1,110 +0,0 @@
-/*
- * 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.execution.pipeline;
-
-import org.apache.streampipes.manager.execution.http.GraphSubmitter;
-import org.apache.streampipes.model.eventrelay.SpDataStreamRelayContainer;
-import org.apache.streampipes.manager.execution.status.PipelineStatusManager;
-import org.apache.streampipes.manager.util.TemporaryGraphStorage;
-import org.apache.streampipes.model.SpDataSet;
-import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
-import org.apache.streampipes.model.graph.DataProcessorInvocation;
-import org.apache.streampipes.model.graph.DataSinkInvocation;
-import org.apache.streampipes.model.message.PipelineStatusMessage;
-import org.apache.streampipes.model.message.PipelineStatusMessageType;
-import org.apache.streampipes.model.pipeline.Pipeline;
-import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
-
-import java.util.*;
-import java.util.stream.Collectors;
-
-public class PipelineExecutor extends AbstractPipelineExecutor {
-
- public PipelineExecutor(Pipeline pipeline, boolean visualize, boolean storeStatus, boolean monitor) {
- super(pipeline, visualize, storeStatus, monitor);
- }
-
- public PipelineOperationStatus startPipeline() {
-
- pipeline.getSepas().forEach(this::updateKafkaGroupIds);
- pipeline.getActions().forEach(this::updateKafkaGroupIds);
-
- List<DataProcessorInvocation> sepas = pipeline.getSepas();
- List<DataSinkInvocation> secs = pipeline.getActions();
-
- List<SpDataSet> dataSets = pipeline.getStreams().stream().filter(s -> s instanceof SpDataSet).map(s -> new
- SpDataSet((SpDataSet) s)).collect(Collectors.toList());
-
- for (SpDataSet ds : dataSets) {
- ds.setCorrespondingPipeline(pipeline.getPipelineId());
- }
-
- List<InvocableStreamPipesEntity> graphs = new ArrayList<>();
- graphs.addAll(sepas);
- graphs.addAll(secs);
-
- List<InvocableStreamPipesEntity> decryptedGraphs = decryptSecrets(graphs);
-
- graphs.forEach(g -> g.setStreamRequirements(Collections.emptyList()));
-
- List<SpDataStreamRelayContainer> relays = generateRelays(decryptedGraphs);
-
- PipelineOperationStatus status = new GraphSubmitter(pipeline.getPipelineId(), pipeline.getName(),
- decryptedGraphs, dataSets, relays).invokePipelineElementsAndRelays();
-
- if (status.isSuccess()) {
- storeInvocationGraphs(pipeline.getPipelineId(), graphs, dataSets);
- storeDataStreamRelayContainer(relays);
-
- PipelineStatusManager.addPipelineStatus(
- pipeline.getPipelineId(),
- new PipelineStatusMessage(pipeline.getPipelineId(),
- System.currentTimeMillis(),
- PipelineStatusMessageType.PIPELINE_STARTED.title(),
- PipelineStatusMessageType.PIPELINE_STARTED.description()));
-
- if (storeStatus) setPipelineStarted(pipeline);
- }
- return status;
- }
-
- public PipelineOperationStatus stopPipeline() {
- List<InvocableStreamPipesEntity> graphs = TemporaryGraphStorage.graphStorage.get(pipeline.getPipelineId());
- List<SpDataSet> dataSets = TemporaryGraphStorage.datasetStorage.get(pipeline.getPipelineId());
- List<SpDataStreamRelayContainer> relays = generateRelays(graphs);
-
- PipelineOperationStatus status = new GraphSubmitter(pipeline.getPipelineId(), pipeline.getName(), graphs,
- dataSets, relays).detachPipelineElementsAndRelays();
-
- if (status.isSuccess()) {
- if (visualize) deleteVisualization(pipeline.getPipelineId());
- if (storeStatus) setPipelineStopped(pipeline);
-
- deleteDataStreamRelayContainer(relays);
-
- PipelineStatusManager.addPipelineStatus(pipeline.getPipelineId(),
- new PipelineStatusMessage(pipeline.getPipelineId(),
- System.currentTimeMillis(),
- PipelineStatusMessageType.PIPELINE_STOPPED.title(),
- PipelineStatusMessageType.PIPELINE_STOPPED.description()));
- }
- return status;
- }
-
-}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/PipelineMigrationExecutor.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/PipelineMigrationExecutor.java
deleted file mode 100644
index 7115d40..0000000
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/PipelineMigrationExecutor.java
+++ /dev/null
@@ -1,461 +0,0 @@
-/*
- * 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.execution.pipeline;
-
-import org.apache.streampipes.logging.evaluation.EvaluationLogger;
-import org.apache.streampipes.manager.data.PipelineGraph;
-import org.apache.streampipes.manager.data.PipelineGraphBuilder;
-import org.apache.streampipes.manager.data.PipelineGraphHelpers;
-import org.apache.streampipes.manager.execution.http.GraphSubmitter;
-import org.apache.streampipes.manager.matching.InvocationGraphBuilder;
-import org.apache.streampipes.model.SpDataSet;
-import org.apache.streampipes.model.eventrelay.SpDataStreamRelayContainer;
-import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
-import org.apache.streampipes.model.base.NamedStreamPipesEntity;
-import org.apache.streampipes.model.graph.DataProcessorInvocation;
-import org.apache.streampipes.model.pipeline.Pipeline;
-import org.apache.streampipes.model.pipeline.PipelineElementMigrationEntity;
-import org.apache.streampipes.model.pipeline.PipelineElementStatus;
-import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
-
-import java.util.*;
-import java.util.stream.Collectors;
-
-public class PipelineMigrationExecutor extends AbstractPipelineExecutor {
-
- /**
- * Old pipeline before migration
- */
- private final Pipeline pipelineBeforeMigration;
-
- /**
- * Pipeline element to be migrated. Contains pair of source and target element description
- */
- private final PipelineElementMigrationEntity migrationEntity;
-
- /**
- * Predecessors of the pipeline element to be migrated in the migration pipeline
- */
- private final List<NamedStreamPipesEntity> predecessorsAfterMigration;
-
- /**
- * Predecessors of the pipeline element to be migrated in the old pipeline before migration
- */
- private final List<NamedStreamPipesEntity> predecessorsBeforeMigration;
-
- /**
- * Collection of relays that were created in the migration process needing to be stored
- */
- private final List<SpDataStreamRelayContainer> relaysToBePersisted;
-
- /**
- * Collection of relays that were removed in the migration process needing to be deleted from persistent storage.
- */
- private final List<SpDataStreamRelayContainer> relaysToBeDeleted;
-
-
- public PipelineMigrationExecutor(Pipeline pipeline, Pipeline pipelineBeforeMigration,
- PipelineElementMigrationEntity migrationEntity,
- boolean visualize, boolean storeStatus, boolean monitor) {
- super(pipeline, visualize, storeStatus, monitor);
- this.pipelineBeforeMigration = pipelineBeforeMigration;
- this.migrationEntity = migrationEntity;
- this.predecessorsAfterMigration = new ArrayList<>();
- this.predecessorsBeforeMigration = new ArrayList<>();
- this.relaysToBePersisted = new ArrayList<>();
- this.relaysToBeDeleted = new ArrayList<>();
- }
-
- public PipelineOperationStatus migratePipelineElement() {
- if(this.migrationEntity.getTargetElement().isStateful())
- return migrateStatefulPipelineElement();
- return migrateStatelessPipelineElement();
- }
-
- public PipelineOperationStatus migrateStatelessPipelineElement() {
-
- PipelineOperationStatus status = initPipelineOperationStatus();
-
- // 1. start new element
- // 2. stop relay to origin element
- // 3. start relay to new element
- // 4. stop origin element
- // 5. stop origin relay
-
- prepareMigration();
- //TODO: Remove logging after evaluation
- EvaluationLogger logger = EvaluationLogger.getInstance();
-
- // Start target pipeline elements and relays on new target node
- long before_start_target = System.nanoTime();
- PipelineOperationStatus statusStartTarget = startTargetPipelineElementsAndRelays(status);
- if(!statusStartTarget.isSuccess()){
- //Target PE could not be started; nothing to roll back
- return status;
- }
- long start_target_duration = System.nanoTime() - before_start_target;
- logger.logMQTT("Migration", "start target element","",start_target_duration,start_target_duration/1000000000.0);
-
- // Stop relays from origin predecessor
- long downtime_beginning = System.nanoTime();
- PipelineOperationStatus statusStopRelays = stopRelaysFromPredecessorsBeforeMigration(status);
- if(!statusStopRelays.isSuccess()){
- rollbackToPreMigrationStepOne(statusStopRelays, status);
- return status;
- }
- long stop_relays_origin_duration = System.nanoTime() - downtime_beginning;
- logger.logMQTT("Migration", "stop relay from origin","",stop_relays_origin_duration,stop_relays_origin_duration/1000000000.0);
-
- // Start relays to target after migration
- long before_start_relay_target = System.nanoTime();
- PipelineOperationStatus statusStartRelays = startRelaysFromPredecessorsAfterMigration(status);
- if(!statusStartRelays.isSuccess()){
- rollbackToPreMigrationStepTwo(statusStartRelays, status);
- return status;
- }
- long start_relay_target_duration = System.nanoTime() - before_start_relay_target;
- logger.logMQTT("Migration", "start relay to target","",start_relay_target_duration,start_relay_target_duration/1000000000.0);
-
- long downtime = System.nanoTime() - downtime_beginning;
- logger.logMQTT("Migration", "downtime", "", downtime, downtime/1000000000.0);
-
- //Stop origin and associated relay
- long before_stop_origin = System.nanoTime();
- PipelineOperationStatus statusStopOrigin = stopOriginPipelineElementsAndRelays(status);
- if(!statusStopOrigin.isSuccess()){
- rollbackToPreMigrationStepThree(status);
- return status;
- }
- long stop_origin_duration = System.nanoTime() - before_stop_origin;
- logger.logMQTT("Migration", "stop origin element","",stop_origin_duration,stop_origin_duration/1000000000.0);
-
- List<InvocableStreamPipesEntity> graphs = new ArrayList<>();
- graphs.addAll(pipeline.getActions());
- graphs.addAll(pipeline.getSepas());
-
- List<SpDataSet> dataSets = findDataSets();
-
- // store new pipeline and relays
- storeInvocationGraphs(pipeline.getPipelineId(), graphs, dataSets);
- deleteDataStreamRelayContainer(relaysToBeDeleted);
- storeDataStreamRelayContainer(relaysToBePersisted);
-
- // set global status
- status.setSuccess(status.getElementStatus().stream().allMatch(PipelineElementStatus::isSuccess));
-
- return status;
- }
-
- public PipelineOperationStatus migrateStatefulPipelineElement() {
- //TODO: Assess how this 'dirty' implementation/procedure can be improved
-
- PipelineOperationStatus status = initPipelineOperationStatus();
-
- // 1. start new element
- // 2. stop relay to origin element
- // 3. start relay to new element
- // 4. stop origin element
- // 5. stop origin relay
-
- prepareMigration();
-
- //Try to get state of origin element (success not checked)
- PipelineElementStatus statusGettingState = getState(this.migrationEntity.getSourceElement());
- if(!statusGettingState.isSuccess()){
- //status.addPipelineElementStatus(statusGettingState);
- status.setSuccess(false);
- return status;
- }
- String currentState = statusGettingState.getOptionalMessage();
- // Start target pipeline elements and relays on new target node
- PipelineOperationStatus statusStartTarget = startTargetPipelineElementsAndRelays(status);
- if(!statusStartTarget.isSuccess()){
- //Target PE could not be started; nothing to roll back
- return status;
- }
-
- //Set state of the newly invoked Pipeline Element (success not checked)
- PipelineElementStatus statusSettingState = setState(this.migrationEntity.getTargetElement(), currentState);
-
- if(!statusSettingState.isSuccess()){
- //status.addPipelineElementStatus(statusSettingState);
- status.setSuccess(false);
- rollbackToPreMigrationStepOne(new PipelineOperationStatus(), status);
- return status;
- }
-
- // Stop relays from origin predecessor
- PipelineOperationStatus statusStopRelays = stopRelaysFromPredecessorsBeforeMigration(status);
- if(!statusStopRelays.isSuccess()){
- rollbackToPreMigrationStepOne(statusStopRelays, status);
- return status;
- }
-
- // Start relays to target after migration
- PipelineOperationStatus statusStartRelays = startRelaysFromPredecessorsAfterMigration(status);
- if(!statusStartRelays.isSuccess()){
- rollbackToPreMigrationStepTwo(statusStartRelays, status);
- return status;
- }
-
- //Stop origin and associated relay
- PipelineOperationStatus statusStopOrigin = stopOriginPipelineElementsAndRelays(status);
- if(!statusStopOrigin.isSuccess()){
- rollbackToPreMigrationStepThree(status);
- return status;
- }
-
- List<InvocableStreamPipesEntity> graphs = new ArrayList<>();
- graphs.addAll(pipeline.getActions());
- graphs.addAll(pipeline.getSepas());
-
- List<SpDataSet> dataSets = findDataSets();
-
- // store new pipeline and relays
- storeInvocationGraphs(pipeline.getPipelineId(), graphs, dataSets);
- deleteDataStreamRelayContainer(relaysToBeDeleted);
- storeDataStreamRelayContainer(relaysToBePersisted);
-
- // set global status
- status.setSuccess(status.getElementStatus().stream().allMatch(PipelineElementStatus::isSuccess));
-
- return status;
- }
-
- private void prepareMigration() {
- //Purge existing relays
- purgeExistingRelays();
-
- //Generate new relays
- PipelineGraph pipelineGraphAfterMigration = new PipelineGraphBuilder(pipeline).buildGraph();
- buildGraphWithRelays(pipelineGraphAfterMigration);
-
- //Get predecessors
- PipelineGraph pipelineGraphBeforeMigration = new PipelineGraphBuilder(pipelineBeforeMigration).buildGraph();
- findPredecessorsInMigrationPipeline(pipelineGraphAfterMigration);
-
- //Find counterpart for predecessors in currentPipeline
- findAndComparePredecessorsInCurrentPipeline(pipelineGraphBeforeMigration);
- }
-
- private void rollbackToPreMigrationStepOne(PipelineOperationStatus statusStopRelays,
- PipelineOperationStatus status) {
- // Stop target pipeline element and relays on new target node
- PipelineOperationStatus statusTargetRollback = rollbackTargetPipelineElementsAndRelays();
-
- // Restart relays before migration attempt
- // extract unique running instance ids of original relays
- Set<String> relayIdsToRollback = extractUniqueRelayIds(statusStopRelays);
- PipelineOperationStatus statusRelaysRollback = rollbackToOriginRelaysByInvoke(relayIdsToRollback);
-
- // Add status to global migration status
- updateMigrationStatus(statusTargetRollback, status);
- updateMigrationStatus(statusRelaysRollback, status);
- }
-
- private void rollbackToPreMigrationStepTwo(PipelineOperationStatus statusStartRelays,
- PipelineOperationStatus status) {
- //Rollback target PE, stopped relays and all successfully started relays
- PipelineOperationStatus statusTargetRollback = rollbackTargetPipelineElementsAndRelays();
-
- // Restart relays to origin from predecessors before migration
- PipelineOperationStatus statusRelaysInvokeRollback = startRelays(findRelays(predecessorsBeforeMigration,
- migrationEntity.getSourceElement()));
-
- // Stop relays that were started due to migration attempt
- // extract unique running instance ids of original relays
- Set<String> relayIdsToRollback = extractUniqueRelayIds(statusStartRelays);
- PipelineOperationStatus statusRelaysDetachRollback = rollbackTargetRelaysByDetach(relayIdsToRollback);
-
- // Add status to global migration status
- updateMigrationStatus(statusTargetRollback, status);
- updateMigrationStatus(statusRelaysInvokeRollback, status);
- updateMigrationStatus(statusRelaysDetachRollback, status);
- }
-
- private void rollbackToPreMigrationStepThree(PipelineOperationStatus status) {
- //Rollback everything
- PipelineOperationStatus statusStopRelays = stopRelays(findRelays(predecessorsAfterMigration,
- migrationEntity.getTargetElement()));
-
- PipelineOperationStatus statusStartRelays = startRelays(findRelays(predecessorsBeforeMigration,
- migrationEntity.getSourceElement()));
-
- List<InvocableStreamPipesEntity> graphs = Collections.singletonList(migrationEntity.getTargetElement());
- List<SpDataStreamRelayContainer> relays = extractRelaysFromDataProcessor(graphs);
-
- PipelineOperationStatus statusStopTargetAndRelays = stopPipelineElementsAndRelays(graphs, relays);
-
- // Add status to global migration status
- updateMigrationStatus(statusStopRelays, status);
- updateMigrationStatus(statusStartRelays, status);
- updateMigrationStatus(statusStopTargetAndRelays, status);
- }
-
-
- private PipelineOperationStatus rollbackToOriginRelaysByInvoke(Set<String> relayIdsToRollback) {
- List<SpDataStreamRelayContainer> rollbackRelays = findRelaysAndFilterById(relayIdsToRollback,
- predecessorsBeforeMigration, migrationEntity.getSourceElement());
-
- return startRelays(rollbackRelays);
- }
-
- private PipelineOperationStatus rollbackTargetRelaysByDetach(Set<String> relayIdsToRollback) {
- List<SpDataStreamRelayContainer> rollbackRelays = findRelaysAndFilterById(relayIdsToRollback,
- predecessorsAfterMigration, migrationEntity.getTargetElement());
-
- return stopRelays(rollbackRelays);
- }
-
- private PipelineOperationStatus rollbackTargetPipelineElementsAndRelays() {
- List<InvocableStreamPipesEntity> graphs = Collections.singletonList(migrationEntity.getTargetElement());
- List<SpDataStreamRelayContainer> relays = extractRelaysFromDataProcessor(graphs);
-
- return new GraphSubmitter(pipeline.getPipelineId(),
- pipeline.getName(), graphs, new ArrayList<>(), relays).detachPipelineElementsAndRelays();
- }
-
- private PipelineOperationStatus startRelaysFromPredecessorsAfterMigration(PipelineOperationStatus status) {
- List<SpDataStreamRelayContainer> relays = findRelays(predecessorsAfterMigration,
- migrationEntity.getTargetElement());
-
- updateRelaysToBePersisted(relays);
-
- PipelineOperationStatus statusStartRelays = startRelays(relays);
- updateMigrationStatus(statusStartRelays, status);
-
- return statusStartRelays;
- }
-
- private PipelineOperationStatus stopRelaysFromPredecessorsBeforeMigration(PipelineOperationStatus status) {
- List<SpDataStreamRelayContainer> relays = findRelaysWhenStopping(predecessorsBeforeMigration,
- migrationEntity.getSourceElement());
-
- updateRelaysToBeDeleted(relays);
-
- PipelineOperationStatus statusStopRelays = stopRelays(relays);
- updateMigrationStatus(statusStopRelays, status);
-
- return statusStopRelays;
- }
-
- private PipelineOperationStatus startTargetPipelineElementsAndRelays(PipelineOperationStatus status) {
- List<InvocableStreamPipesEntity> decryptedGraphs =
- decryptSecrets(Collections.singletonList(migrationEntity.getTargetElement()));
- List<SpDataStreamRelayContainer> relays = extractRelaysFromDataProcessor(decryptedGraphs);
-
- updateRelaysToBePersisted(relays);
-
- PipelineOperationStatus statusStartTarget = startPipelineElementsAndRelays(decryptedGraphs, relays);
- updateMigrationStatus(statusStartTarget, status);
-
- return statusStartTarget;
- }
-
- private PipelineOperationStatus stopOriginPipelineElementsAndRelays(PipelineOperationStatus status) {
- List<InvocableStreamPipesEntity> graphs = Collections.singletonList(migrationEntity.getSourceElement());
- List<SpDataStreamRelayContainer> relays = extractRelaysFromDataProcessor(graphs);
-
- updateRelaysToBeDeleted(relays);
-
- PipelineOperationStatus statusStopOrigin = stopPipelineElementsAndRelays(graphs, relays);
- updateMigrationStatus(statusStopOrigin, status);
-
- return statusStopOrigin;
- }
-
- // Helpers
-
- private void updateRelaysToBePersisted(List<SpDataStreamRelayContainer> relays) {
- relays.stream()
- .filter(r -> r.getOutputStreamRelays().size() > 0)
- .forEach(relaysToBePersisted::add);
- }
-
- private void updateRelaysToBeDeleted(List<SpDataStreamRelayContainer> relays) {
- relays.stream()
- .filter(r -> r.getOutputStreamRelays().size() > 0)
- .forEach(relaysToBeDeleted::add);
- }
-
- private List<SpDataStreamRelayContainer> findRelaysAndFilterById(Set<String> relayIdsToRollback,
- List<NamedStreamPipesEntity> predecessor,
- InvocableStreamPipesEntity target) {
- return findRelays(predecessor, target).stream()
- .filter(relay -> relayIdsToRollback.contains(relay.getRunningStreamRelayInstanceId()))
- .collect(Collectors.toList());
- }
-
- private void findAndComparePredecessorsInCurrentPipeline(PipelineGraph pipelineGraphBeforeMigration) {
- predecessorsAfterMigration.forEach(migrationPredecessor ->
- predecessorsBeforeMigration.add(findMatching(migrationPredecessor, pipelineGraphBeforeMigration)));
- }
-
- private void findPredecessorsInMigrationPipeline(PipelineGraph pipelineGraphAfterMigration) {
- // get unique list of predecessors
- List<NamedStreamPipesEntity> predecessors = PipelineGraphHelpers.findStreams(pipelineGraphAfterMigration).stream()
- .map(stream -> getPredecessors(stream, migrationEntity.getTargetElement(),
- pipelineGraphAfterMigration, new ArrayList<>()))
- .flatMap(List::stream)
- .collect(Collectors.toList())
- .stream()
- .distinct()
- .collect(Collectors.toList());
-
- predecessorsAfterMigration.addAll(predecessors);
- }
-
- private List<SpDataSet> findDataSets() {
- return pipeline.getStreams().stream()
- .filter(s -> s instanceof SpDataSet)
- .map(s -> new SpDataSet((SpDataSet) s))
- .collect(Collectors.toList());
- }
-
- private void buildGraphWithRelays(PipelineGraph pipelineGraphAfterMigration) {
- new InvocationGraphBuilder(pipelineGraphAfterMigration, pipeline.getPipelineId()).buildGraphs();
- }
-
- private void purgeExistingRelays() {
- pipeline.getSepas().forEach(s -> s.setOutputStreamRelays(new ArrayList<>()));
- }
-
- private void updateMigrationStatus(PipelineOperationStatus partialStatus, PipelineOperationStatus status) {
- // Add status to global migration status
- partialStatus.getElementStatus().forEach(status::addPipelineElementStatus);
- }
-
- private List<SpDataStreamRelayContainer> extractRelaysFromDataProcessor(List<InvocableStreamPipesEntity> graphs) {
- return graphs.stream()
- .map(DataProcessorInvocation.class::cast)
-// .map(p -> {
-// String modifiedId = p.getDeploymentRunningInstanceId() + "-" + p.getDeploymentTargetNodeId();
-// p.setDeploymentRunningInstanceId(modifiedId);
-// return p;
-// })
- .map(SpDataStreamRelayContainer::new)
- .collect(Collectors.toList());
- }
-
- private Set<String> extractUniqueRelayIds(PipelineOperationStatus status) {
- return status.getElementStatus().stream()
- .filter(PipelineElementStatus::isSuccess)
- .map(PipelineElementStatus::getElementId)
- .collect(Collectors.toSet());
- }
-}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/PipelineExecutor.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/PipelineExecutor.java
index 0bd26ec..3e50d91 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/PipelineExecutor.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/PipelineExecutor.java
@@ -24,16 +24,13 @@ import org.apache.streampipes.manager.execution.pipeline.executor.operations.typ
import org.apache.streampipes.manager.execution.pipeline.executor.utils.StatusUtils;
import org.apache.streampipes.model.SpDataSet;
import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
-import org.apache.streampipes.model.base.NamedStreamPipesEntity;
import org.apache.streampipes.model.eventrelay.SpDataStreamRelayContainer;
import org.apache.streampipes.model.pipeline.Pipeline;
import org.apache.streampipes.model.pipeline.PipelineElementMigrationEntity;
import org.apache.streampipes.model.pipeline.PipelineElementReconfigurationEntity;
import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
-import java.util.ArrayList;
import java.util.LinkedList;
-import java.util.List;
public class PipelineExecutor {
@@ -42,9 +39,6 @@ public class PipelineExecutor {
private boolean storeStatus;
private boolean monitor;
- /**
- * Old pipeline before migration
- */
private Pipeline secondaryPipeline;
/**
@@ -52,58 +46,29 @@ public class PipelineExecutor {
*/
private PipelineElementMigrationEntity migrationEntity;
- /**
- * Predecessors of the pipeline element to be migrated in the migration pipeline
- */
- private List<NamedStreamPipesEntity> predecessorsAfterMigration;
-
- /**
- * Predecessors of the pipeline element to be migrated in the old pipeline before migration
- */
- private List<NamedStreamPipesEntity> predecessorsBeforeMigration;
-
- /**
- * Collection of relays that were created in the migration process needing to be stored
- */
- private List<SpDataStreamRelayContainer> relaysToBePersisted;
-
- /**
- * Collection of relays that were removed in the migration process needing to be deleted from persistent storage.
- */
- private List<SpDataStreamRelayContainer> relaysToBeDeleted;
-
- private PipelineOperationStatus status;
-
private PipelineElementReconfigurationEntity reconfigurationEntity;
- private final LinkedList<PipelineExecutionOperation> operations = new LinkedList<>();
-
- private List<SpDataSet> dataSets;
+ private final LifecycleEntity<InvocableStreamPipesEntity> graphs;
- private List<SpDataStreamRelayContainer> relays;
+ private final LifecycleEntity<SpDataStreamRelayContainer> relays;
- private List<InvocableStreamPipesEntity> graphs;
+ private final LifecycleEntity<SpDataSet> dataSets;
- private final LifecycleEntity<InvocableStreamPipesEntity> grafs;
- private final LifecycleEntity<SpDataStreamRelayContainer> relais;
+ private PipelineOperationStatus status;
- private final LifecycleEntity<SpDataSet> dataZeds;
+ private final LinkedList<PipelineExecutionOperation> operations = new LinkedList<>();
public PipelineExecutor(Pipeline pipeline, boolean visualize, boolean storeStatus, boolean monitor){
this.pipeline = pipeline;
this.visualize = visualize;
this.storeStatus = storeStatus;
this.monitor = monitor;
- this.predecessorsAfterMigration = new ArrayList<>();
- this.predecessorsBeforeMigration = new ArrayList<>();
- this.relaysToBePersisted = new ArrayList<>();
- this.relaysToBeDeleted = new ArrayList<>();
this.status = StatusUtils.initPipelineOperationStatus(pipeline);
- this.grafs = new LifecycleEntity<>();
- this.relais = new LifecycleEntity<>();
- this.dataZeds = new LifecycleEntity<>();
+ this.graphs = new LifecycleEntity<>();
+ this.relays = new LifecycleEntity<>();
+ this.dataSets = new LifecycleEntity<>();
}
public PipelineOperationStatus execute(){
@@ -134,12 +99,21 @@ public class PipelineExecutor {
StatusUtils.updateStatus(rollbackStatus, this.status);
}
- //Getter and Setter
-
public void addOperation(PipelineExecutionOperation operation){
this.operations.add(operation);
}
+ public boolean containsReconfigurationOperation(){
+ return this.operations.stream().anyMatch(operation -> operation instanceof ReconfigurationOperation);
+ }
+
+ public boolean containsMigrationOperation(){
+ return this.operations.stream().anyMatch(operation -> operation instanceof MigrationOperation);
+ }
+
+
+ //Getter and Setter
+
public Pipeline getPipeline() {
return pipeline;
}
@@ -188,38 +162,6 @@ public class PipelineExecutor {
this.migrationEntity = migrationEntity;
}
- public List<NamedStreamPipesEntity> getPredecessorsAfterMigration() {
- return predecessorsAfterMigration;
- }
-
- public void setPredecessorsAfterMigration(List<NamedStreamPipesEntity> predecessorsAfterMigration) {
- this.predecessorsAfterMigration = predecessorsAfterMigration;
- }
-
- public List<NamedStreamPipesEntity> getPredecessorsBeforeMigration() {
- return predecessorsBeforeMigration;
- }
-
- public void setPredecessorsBeforeMigration(List<NamedStreamPipesEntity> predecessorsBeforeMigration) {
- this.predecessorsBeforeMigration = predecessorsBeforeMigration;
- }
-
- public List<SpDataStreamRelayContainer> getRelaysToBePersisted() {
- return relaysToBePersisted;
- }
-
- public void setRelaysToBePersisted(List<SpDataStreamRelayContainer> relaysToBePersisted) {
- this.relaysToBePersisted = relaysToBePersisted;
- }
-
- public List<SpDataStreamRelayContainer> getRelaysToBeDeleted() {
- return relaysToBeDeleted;
- }
-
- public void setRelaysToBeDeleted(List<SpDataStreamRelayContainer> relaysToBeDeleted) {
- this.relaysToBeDeleted = relaysToBeDeleted;
- }
-
public PipelineOperationStatus getStatus() {
return status;
}
@@ -236,48 +178,15 @@ public class PipelineExecutor {
this.reconfigurationEntity = reconfigurationEntity;
}
- public List<SpDataSet> getDataSets() {
- return dataSets;
- }
-
- public void setDataSets(List<SpDataSet> dataSets) {
- this.dataSets = dataSets;
- }
-
- public List<SpDataStreamRelayContainer> getRelays() {
- return relays;
- }
-
- public void setRelays(List<SpDataStreamRelayContainer> relays) {
- this.relays = relays;
- }
-
- public List<InvocableStreamPipesEntity> getGraphs() {
+ public LifecycleEntity<InvocableStreamPipesEntity> getGraphs() {
return graphs;
}
- public void setGraphs(List<InvocableStreamPipesEntity> graphs) {
- this.graphs = graphs;
- }
-
-
- public LifecycleEntity<InvocableStreamPipesEntity> getGrafs() {
- return grafs;
- }
-
- public LifecycleEntity<SpDataStreamRelayContainer> getRelais() {
- return relais;
- }
-
- public LifecycleEntity<SpDataSet> getDataZeds() {
- return dataZeds;
- }
-
- public boolean containsReconfigurationOperation(){
- return this.operations.stream().anyMatch(operation -> operation instanceof ReconfigurationOperation);
+ public LifecycleEntity<SpDataStreamRelayContainer> getRelays() {
+ return relays;
}
- public boolean containsMigrationOperation(){
- return this.operations.stream().anyMatch(operation -> operation instanceof MigrationOperation);
+ public LifecycleEntity<SpDataSet> getDataSets() {
+ return dataSets;
}
}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/PipelineExecutorBuilder.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/PipelineExecutorBuilder.java
index 52fc870..d423cf1 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/PipelineExecutorBuilder.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/PipelineExecutorBuilder.java
@@ -68,23 +68,23 @@ public class PipelineExecutorBuilder {
return this;
}
- public PipelineExecutorBuilder addStartRelaysFromPredecessorsOperation(){
- pipelineExecutor.addOperation(new StartRelaysFromPredecessorsOperation(pipelineExecutor));
+ public PipelineExecutorBuilder addStartRelaysOperation(){
+ pipelineExecutor.addOperation(new StartRelaysOperation(pipelineExecutor));
return this;
}
- public PipelineExecutorBuilder addStartTargetPipelineElementsOperation(){
- pipelineExecutor.addOperation(new StartTargetPipelineElementsOperation(pipelineExecutor));
+ public PipelineExecutorBuilder addStartGraphsAndAssociatedRelaysOperation(){
+ pipelineExecutor.addOperation(new StartGraphsAndAssociatedRelaysOperation(pipelineExecutor));
return this;
}
- public PipelineExecutorBuilder addStopOriginPipelineElementAndRelaysOperation(){
- pipelineExecutor.addOperation(new StopOriginPipelineElementAndRelaysOperation(pipelineExecutor));
+ public PipelineExecutorBuilder addStopGraphsAndAssociatedRelaysOperation(){
+ pipelineExecutor.addOperation(new StopGraphsAndAssociatedRelaysOperation(pipelineExecutor));
return this;
}
- public PipelineExecutorBuilder addStopRelaysFromPredecessorOperation(){
- pipelineExecutor.addOperation(new StopRelaysFromPredecessorOperation(pipelineExecutor));
+ public PipelineExecutorBuilder addStopRelaysOperation(){
+ pipelineExecutor.addOperation(new StopRelaysOperation(pipelineExecutor));
return this;
}
@@ -98,6 +98,11 @@ public class PipelineExecutorBuilder {
return this;
}
+ public PipelineExecutorBuilder addPreparePipelineStartOperation(){
+ pipelineExecutor.addOperation(new PrepareStartPipelineOperation(pipelineExecutor));
+ return this;
+ }
+
public PipelineExecutorBuilder addStartPipelineOperation(){
pipelineExecutor.addOperation(new StartPipelineOperation(pipelineExecutor));
return this;
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/PipelineExecutorFactory.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/PipelineExecutorFactory.java
index 413edd3..617ff01 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/PipelineExecutorFactory.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/PipelineExecutorFactory.java
@@ -32,10 +32,10 @@ public class PipelineExecutorFactory {
.addPrepareMigrationOperation();
if(migrationEntity.getTargetElement().isStateful())
builder.addGetStateOperation();
- builder.addStartTargetPipelineElementsOperation()
- .addStopRelaysFromPredecessorOperation()
- .addStartRelaysFromPredecessorsOperation()
- .addStopOriginPipelineElementAndRelaysOperation()
+ builder.addStartGraphsAndAssociatedRelaysOperation()
+ .addStopRelaysOperation()
+ .addStartRelaysOperation()
+ .addStopGraphsAndAssociatedRelaysOperation()
.addStoreMigratedPipelineOperation();
return builder.buildPipelineExecutor();
}
@@ -51,7 +51,10 @@ public class PipelineExecutorFactory {
public static PipelineExecutor createInvocationExecutor(Pipeline pipeline, boolean visualize, boolean storeStatus,
boolean monitor){
return PipelineExecutorBuilder.getBuilder().initializePipelineExecutor(pipeline, visualize, storeStatus, monitor)
- .addStartPipelineOperation().addStorePipelineOperation().buildPipelineExecutor();
+ .addPreparePipelineStartOperation()
+ .addStartPipelineOperation()
+ .addStorePipelineOperation()
+ .buildPipelineExecutor();
}
public static PipelineExecutor createDetachExecutor(Pipeline pipeline, boolean visualize, boolean storeStatus,
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/GetStateOperation.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/GetStateOperation.java
index c0fe26c..e11609b 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/GetStateOperation.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/GetStateOperation.java
@@ -39,20 +39,20 @@ public class GetStateOperation extends PipelineExecutionOperation implements Mig
public PipelineOperationStatus executeOperation() {
long nanoTimeBeforeOperation = System.nanoTime();
PipelineElementStatus statusGettingState = CommunicationUtils.getState(
- associatedPipelineExecutor.getMigrationEntity().getSourceElement(),
- associatedPipelineExecutor.getPipeline());
+ pipelineExecutor.getMigrationEntity().getSourceElement(),
+ pipelineExecutor.getPipeline());
if(statusGettingState.isSuccess()) {
- associatedPipelineExecutor.getMigrationEntity().getTargetElement()
+ pipelineExecutor.getMigrationEntity().getTargetElement()
.setState(statusGettingState.getOptionalMessage());
statusGettingState.setOptionalMessage("Successfully retrieved state");
}
- PipelineOperationStatus getStateStatus = StatusUtils.initPipelineOperationStatus(associatedPipelineExecutor.getPipeline());
+ PipelineOperationStatus getStateStatus = StatusUtils.initPipelineOperationStatus(pipelineExecutor.getPipeline());
getStateStatus.addPipelineElementStatus(statusGettingState);
StatusUtils.checkSuccess(getStateStatus);
long duration = System.nanoTime() - nanoTimeBeforeOperation;
EvaluationLogger.getInstance().logMQTT("Migration", "get state", "", duration, duration/1000000000.0);
try {
- int stateSize = getSizeInBytes(associatedPipelineExecutor.getMigrationEntity().getTargetElement().getState());
+ int stateSize = getSizeInBytes(pipelineExecutor.getMigrationEntity().getTargetElement().getState());
EvaluationLogger.getInstance().logMQTT("Migration", "state size", stateSize/1024.0, stateSize/1048576.0);
} catch (IOException e) {
e.printStackTrace();
@@ -73,13 +73,13 @@ public class GetStateOperation extends PipelineExecutionOperation implements Mig
@Override
public PipelineOperationStatus rollbackOperationPartially() {
- associatedPipelineExecutor.getMigrationEntity().getTargetElement().setState(null);
- return StatusUtils.initPipelineOperationStatus(associatedPipelineExecutor.getPipeline());
+ pipelineExecutor.getMigrationEntity().getTargetElement().setState(null);
+ return StatusUtils.initPipelineOperationStatus(pipelineExecutor.getPipeline());
}
@Override
public PipelineOperationStatus rollbackOperationFully() {
- associatedPipelineExecutor.getMigrationEntity().getTargetElement().setState(null);
- return StatusUtils.initPipelineOperationStatus(associatedPipelineExecutor.getPipeline());
+ pipelineExecutor.getMigrationEntity().getTargetElement().setState(null);
+ return StatusUtils.initPipelineOperationStatus(pipelineExecutor.getPipeline());
}
}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/LifecycleEntity.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/LifecycleEntity.java
index 09777ba..d2c6d8c 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/LifecycleEntity.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/LifecycleEntity.java
@@ -1,3 +1,20 @@
+/*
+ * 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.execution.pipeline.executor.operations;
import java.util.ArrayList;
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/PipelineExecutionOperation.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/PipelineExecutionOperation.java
index 1d6d76a..e822ec4 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/PipelineExecutionOperation.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/PipelineExecutionOperation.java
@@ -23,13 +23,13 @@ import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
public abstract class PipelineExecutionOperation {
- protected final PipelineExecutor associatedPipelineExecutor;
+ protected final PipelineExecutor pipelineExecutor;
private PipelineOperationStatus status;
public PipelineExecutionOperation(PipelineExecutor pipelineExecutor){
- this.associatedPipelineExecutor = pipelineExecutor;
- this.status = StatusUtils.initPipelineOperationStatus(associatedPipelineExecutor.getPipeline());
+ this.pipelineExecutor = pipelineExecutor;
+ this.status = StatusUtils.initPipelineOperationStatus(this.pipelineExecutor.getPipeline());
}
public abstract PipelineOperationStatus executeOperation();
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/PrepareMigrationOperation.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/PrepareMigrationOperation.java
index 9b668b7..4c9304d 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/PrepareMigrationOperation.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/PrepareMigrationOperation.java
@@ -22,25 +22,34 @@ import org.apache.streampipes.manager.data.PipelineGraphBuilder;
import org.apache.streampipes.manager.data.PipelineGraphHelpers;
import org.apache.streampipes.manager.execution.pipeline.executor.PipelineExecutor;
import org.apache.streampipes.manager.execution.pipeline.executor.operations.types.MigrationOperation;
+import org.apache.streampipes.manager.execution.pipeline.executor.utils.PipelineElementUtils;
import org.apache.streampipes.manager.execution.pipeline.executor.utils.PipelineUtils;
+import org.apache.streampipes.manager.execution.pipeline.executor.utils.RelayUtils;
import org.apache.streampipes.manager.execution.pipeline.executor.utils.StatusUtils;
+import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
import org.apache.streampipes.model.base.NamedStreamPipesEntity;
+import org.apache.streampipes.model.eventrelay.SpDataStreamRelayContainer;
import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
import org.apache.streampipes.model.pipeline.Pipeline;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
import java.util.stream.Collectors;
public class PrepareMigrationOperation extends PipelineExecutionOperation implements MigrationOperation {
+ private List<NamedStreamPipesEntity> predecessorsAfterMigration;
+ private final List<NamedStreamPipesEntity> predecessorsBeforeMigration = new ArrayList<>();
+ private Pipeline pipeline;
+
public PrepareMigrationOperation(PipelineExecutor pipelineExecutor) {
super(pipelineExecutor);
}
@Override
public PipelineOperationStatus executeOperation() {
- Pipeline pipeline = associatedPipelineExecutor.getPipeline();
+ this.pipeline = pipelineExecutor.getPipeline();
//Purge existing relays
PipelineUtils.purgeExistingRelays(pipeline);
@@ -51,42 +60,78 @@ public class PrepareMigrationOperation extends PipelineExecutionOperation implem
//Get predecessors
PipelineGraph pipelineGraphBeforeMigration =
- new PipelineGraphBuilder(associatedPipelineExecutor.getSecondaryPipeline()).buildGraph();
- findPredecessorsInMigrationPipeline(pipelineGraphAfterMigration);
+ new PipelineGraphBuilder(pipelineExecutor.getSecondaryPipeline()).buildGraph();
+ findPredecessorsAfterMigration(pipelineGraphAfterMigration);
//Find counterpart for predecessors in currentPipeline
- findAndComparePredecessorsInCurrentPipeline(pipelineGraphBeforeMigration);
+ findPredecessorsBeforeMigration(pipelineGraphBeforeMigration);
+
+ initializeGraphs();
+ initializeRelays();
+
return StatusUtils.initPipelineOperationStatus(pipeline);
}
@Override
public PipelineOperationStatus rollbackOperationPartially() {
- return null;
+ return StatusUtils.initPipelineOperationStatus(pipelineExecutor.getPipeline());
}
@Override
public PipelineOperationStatus rollbackOperationFully() {
- return null;
+ return StatusUtils.initPipelineOperationStatus(pipelineExecutor.getPipeline());
}
- private void findPredecessorsInMigrationPipeline(PipelineGraph pipelineGraphAfterMigration) {
+ private void findPredecessorsAfterMigration(PipelineGraph pipelineGraphAfterMigration) {
// get unique list of predecessors
- List<NamedStreamPipesEntity> predecessors = PipelineGraphHelpers.findStreams(pipelineGraphAfterMigration).stream()
+ this.predecessorsAfterMigration = PipelineGraphHelpers.findStreams(pipelineGraphAfterMigration).stream()
.map(stream -> PipelineUtils.getPredecessors(stream,
- associatedPipelineExecutor.getMigrationEntity().getTargetElement(),
+ pipelineExecutor.getMigrationEntity().getTargetElement(),
pipelineGraphAfterMigration, new ArrayList<>()))
.flatMap(List::stream)
.collect(Collectors.toList())
.stream()
.distinct()
.collect(Collectors.toList());
+ }
- associatedPipelineExecutor.getPredecessorsAfterMigration().addAll(predecessors);
+ private void findPredecessorsBeforeMigration(PipelineGraph pipelineGraphBeforeMigration) {
+ this.predecessorsAfterMigration.forEach(migrationPredecessor ->
+ {
+ NamedStreamPipesEntity predecessor = PipelineUtils.findMatching(migrationPredecessor, pipelineGraphBeforeMigration);
+ this.predecessorsBeforeMigration
+ .add(predecessor);
+ });
}
- private void findAndComparePredecessorsInCurrentPipeline(PipelineGraph pipelineGraphBeforeMigration) {
- associatedPipelineExecutor.getPredecessorsAfterMigration().forEach(migrationPredecessor ->
- associatedPipelineExecutor.getPredecessorsBeforeMigration()
- .add(PipelineUtils.findMatching(migrationPredecessor, pipelineGraphBeforeMigration)));
+ private void initializeGraphs(){
+ List<InvocableStreamPipesEntity> decryptedTargetGraphs = PipelineElementUtils.decryptSecrets(
+ Collections.singletonList(pipelineExecutor.getMigrationEntity().getTargetElement()),
+ pipelineExecutor.getPipeline());
+ pipelineExecutor.getGraphs().getEntitiesToStart().addAll(decryptedTargetGraphs);
+
+ List<InvocableStreamPipesEntity> originGraphs = Collections.singletonList(
+ pipelineExecutor.getMigrationEntity().getSourceElement());
+ pipelineExecutor.getGraphs().getEntitiesToStop().addAll(originGraphs);
+
+ pipelineExecutor.getGraphs().getEntitiesToStore().addAll(pipeline.getActions());
+ pipelineExecutor.getGraphs().getEntitiesToStore().addAll(pipeline.getSepas());
+ }
+
+ private void initializeRelays(){
+ List<SpDataStreamRelayContainer> relaysToTarget = RelayUtils.findRelays(
+ predecessorsAfterMigration,
+ pipelineExecutor.getMigrationEntity().getTargetElement(),
+ pipelineExecutor.getPipeline());
+ pipelineExecutor.getRelays().getEntitiesToStart().addAll(relaysToTarget);
+ pipelineExecutor.getRelays().getEntitiesToStore().addAll(relaysToTarget);
+
+ List<SpDataStreamRelayContainer> relaysToOrigin = RelayUtils
+ .findRelaysWhenStopping(
+ predecessorsBeforeMigration,
+ pipelineExecutor.getMigrationEntity().getSourceElement(),
+ pipelineExecutor.getPipeline());
+ pipelineExecutor.getRelays().getEntitiesToStop().addAll(relaysToOrigin);
+ pipelineExecutor.getRelays().getEntitiesToDelete().addAll(relaysToOrigin);
}
}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StartPipelineOperation.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/PrepareStartPipelineOperation.java
similarity index 80%
copy from streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StartPipelineOperation.java
copy to streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/PrepareStartPipelineOperation.java
index 29bc7e6..264ac19 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StartPipelineOperation.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/PrepareStartPipelineOperation.java
@@ -17,7 +17,6 @@
*/
package org.apache.streampipes.manager.execution.pipeline.executor.operations;
-import org.apache.streampipes.manager.execution.http.GraphSubmitter;
import org.apache.streampipes.manager.execution.pipeline.executor.PipelineExecutor;
import org.apache.streampipes.manager.execution.pipeline.executor.utils.PipelineElementUtils;
import org.apache.streampipes.manager.execution.pipeline.executor.utils.StatusUtils;
@@ -36,15 +35,16 @@ import java.util.List;
import java.util.UUID;
import java.util.stream.Collectors;
-public class StartPipelineOperation extends PipelineExecutionOperation{
+public class PrepareStartPipelineOperation extends PipelineExecutionOperation {
- public StartPipelineOperation(PipelineExecutor pipelineExecutor) {
+
+ public PrepareStartPipelineOperation(PipelineExecutor pipelineExecutor) {
super(pipelineExecutor);
}
@Override
public PipelineOperationStatus executeOperation() {
- Pipeline pipeline = associatedPipelineExecutor.getPipeline();
+ Pipeline pipeline = pipelineExecutor.getPipeline();
pipeline.getSepas().forEach(this::updateKafkaGroupIds);
pipeline.getActions().forEach(this::updateKafkaGroupIds);
@@ -68,24 +68,23 @@ public class StartPipelineOperation extends PipelineExecutionOperation{
List<SpDataStreamRelayContainer> relays = PipelineElementUtils.generateRelays(decryptedGraphs, pipeline);
- associatedPipelineExecutor.setGraphs(graphs);
- associatedPipelineExecutor.setDataSets(dataSets);
- associatedPipelineExecutor.setRelays(relays);
-
- return new GraphSubmitter(pipeline.getPipelineId(), pipeline.getName(),
- decryptedGraphs, dataSets, relays).invokePipelineElementsAndRelays();
+ pipelineExecutor.getGraphs().getEntitiesToStart().addAll(graphs);
+ pipelineExecutor.getDataSets().getEntitiesToStart().addAll(dataSets);
+ pipelineExecutor.getRelays().getEntitiesToStart().addAll(relays);
+ pipelineExecutor.getGraphs().getEntitiesToStore().addAll(graphs);
+ pipelineExecutor.getDataSets().getEntitiesToStore().addAll(dataSets);
+ pipelineExecutor.getRelays().getEntitiesToStore().addAll(relays);
+ return StatusUtils.initPipelineOperationStatus(pipeline);
}
@Override
public PipelineOperationStatus rollbackOperationPartially() {
- //TODO: Check if there is a possible realization of partial rollback
- return StatusUtils.initPipelineOperationStatus(associatedPipelineExecutor.getPipeline());
+ return StatusUtils.initPipelineOperationStatus(pipelineExecutor.getPipeline());
}
@Override
public PipelineOperationStatus rollbackOperationFully() {
- //TODO: Implement full rollback
- return StatusUtils.initPipelineOperationStatus(associatedPipelineExecutor.getPipeline());
+ return StatusUtils.initPipelineOperationStatus(pipelineExecutor.getPipeline());
}
/**
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/ReconfigureElementOperation.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/ReconfigureElementOperation.java
index 8e0e8c2..2e66272 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/ReconfigureElementOperation.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/ReconfigureElementOperation.java
@@ -21,8 +21,15 @@ import org.apache.streampipes.manager.execution.http.ReconfigurationSubmitter;
import org.apache.streampipes.manager.execution.pipeline.executor.PipelineExecutor;
import org.apache.streampipes.manager.execution.pipeline.executor.operations.types.ReconfigurationOperation;
import org.apache.streampipes.manager.execution.pipeline.executor.utils.StatusUtils;
+import org.apache.streampipes.model.graph.DataProcessorInvocation;
import org.apache.streampipes.model.pipeline.Pipeline;
+import org.apache.streampipes.model.pipeline.PipelineElementReconfigurationEntity;
import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
+import org.apache.streampipes.model.staticproperty.StaticProperty;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
public class ReconfigureElementOperation extends PipelineExecutionOperation implements ReconfigurationOperation {
@@ -32,18 +39,49 @@ public class ReconfigureElementOperation extends PipelineExecutionOperation impl
@Override
public PipelineOperationStatus executeOperation() {
- Pipeline reconfiguredPipeline = associatedPipelineExecutor.getSecondaryPipeline();
+ Pipeline reconfiguredPipeline = pipelineExecutor.getSecondaryPipeline();
return new ReconfigurationSubmitter(reconfiguredPipeline.getPipelineId(), reconfiguredPipeline.getName(),
- associatedPipelineExecutor.getReconfigurationEntity()).reconfigure();
+ pipelineExecutor.getReconfigurationEntity()).reconfigure();
}
@Override
public PipelineOperationStatus rollbackOperationPartially() {
- return StatusUtils.initPipelineOperationStatus(associatedPipelineExecutor.getPipeline());
+ return rollbackOperationFully();
}
@Override
public PipelineOperationStatus rollbackOperationFully() {
- return StatusUtils.initPipelineOperationStatus(associatedPipelineExecutor.getPipeline());
+ Pipeline originalPipeline = pipelineExecutor.getPipeline();
+ PipelineElementReconfigurationEntity originalReconfigurationEntity = pipelineExecutor.getReconfigurationEntity();
+
+ Optional<DataProcessorInvocation> reconfiguredEntity =
+ findReconfiguredEntity(originalPipeline, originalReconfigurationEntity);
+
+ if(!reconfiguredEntity.isPresent())
+ return StatusUtils.initPipelineOperationStatus(pipelineExecutor.getPipeline());
+
+ PipelineElementReconfigurationEntity rollbackReconfigurationEntity =
+ new PipelineElementReconfigurationEntity(reconfiguredEntity.get());
+
+ rollbackReconfigurationEntity.setReconfiguredStaticProperties(
+ findMatchingStaticProperty(reconfiguredEntity.get().getStaticProperties(),
+ originalReconfigurationEntity.getReconfiguredStaticProperties()));
+
+ return new ReconfigurationSubmitter(originalPipeline.getPipelineId(), originalPipeline.getName(),
+ rollbackReconfigurationEntity).reconfigure();
+ }
+
+ private Optional<DataProcessorInvocation> findReconfiguredEntity(Pipeline originalPipeline,
+ PipelineElementReconfigurationEntity reconfigurationEntity){
+ return originalPipeline.getSepas().stream()
+ .filter(invocation -> invocation.getDeploymentRunningInstanceId()
+ .equals(reconfigurationEntity.getDeploymentRunningInstanceId())).findFirst();
+ }
+
+ private List<StaticProperty> findMatchingStaticProperty(List<StaticProperty> listToSearchIn,
+ List<StaticProperty> listToCompareTo){
+ return listToSearchIn.stream().filter(searchProperty ->
+ listToCompareTo.stream().anyMatch(compareProperty ->
+ compareProperty.getInternalName().equals(searchProperty.getInternalName()))).collect(Collectors.toList());
}
}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StartTargetPipelineElementsOperation.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StartGraphsAndAssociatedRelaysOperation.java
similarity index 54%
rename from streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StartTargetPipelineElementsOperation.java
rename to streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StartGraphsAndAssociatedRelaysOperation.java
index 9e79ab8..72d3802 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StartTargetPipelineElementsOperation.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StartGraphsAndAssociatedRelaysOperation.java
@@ -18,43 +18,35 @@
package org.apache.streampipes.manager.execution.pipeline.executor.operations;
import org.apache.streampipes.logging.evaluation.EvaluationLogger;
-import org.apache.streampipes.manager.execution.http.GraphSubmitter;
import org.apache.streampipes.manager.execution.pipeline.executor.*;
import org.apache.streampipes.manager.execution.pipeline.executor.operations.types.MigrationOperation;
-import org.apache.streampipes.manager.execution.pipeline.executor.utils.CommunicationUtils;
-import org.apache.streampipes.manager.execution.pipeline.executor.utils.PipelineElementUtils;
-import org.apache.streampipes.manager.execution.pipeline.executor.utils.RelayUtils;
-import org.apache.streampipes.manager.execution.pipeline.executor.utils.StatusUtils;
+import org.apache.streampipes.manager.execution.pipeline.executor.utils.*;
import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
import org.apache.streampipes.model.eventrelay.SpDataStreamRelayContainer;
import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
-import java.util.ArrayList;
-import java.util.Collections;
import java.util.List;
+import java.util.Set;
-public class StartTargetPipelineElementsOperation extends PipelineExecutionOperation implements MigrationOperation {
+public class StartGraphsAndAssociatedRelaysOperation extends PipelineExecutionOperation implements MigrationOperation {
- public StartTargetPipelineElementsOperation(PipelineExecutor pipelineExecutor){
+ public StartGraphsAndAssociatedRelaysOperation(PipelineExecutor pipelineExecutor){
super(pipelineExecutor);
}
@Override
public PipelineOperationStatus executeOperation() {
long nanoTimeBeforeOperation = System.nanoTime();
- List<InvocableStreamPipesEntity> decryptedGraphs =
- PipelineElementUtils.decryptSecrets(
- Collections.singletonList(associatedPipelineExecutor.getMigrationEntity().getTargetElement()),
- associatedPipelineExecutor.getPipeline());
- List<SpDataStreamRelayContainer> relays = PipelineElementUtils.extractRelaysFromDataProcessor(decryptedGraphs);
+ List<InvocableStreamPipesEntity> graphs = pipelineExecutor.getGraphs().getEntitiesToStart();
+ List<SpDataStreamRelayContainer> relays = PipelineElementUtils.extractRelaysFromDataProcessor(graphs);
- RelayUtils.updateRelays(relays, associatedPipelineExecutor.getRelaysToBePersisted());
+ RelayUtils.updateRelays(relays, pipelineExecutor.getRelays().getEntitiesToStore());
- //State needs to be deleted to not store it
- associatedPipelineExecutor.getMigrationEntity().getTargetElement().setState(null);
+ //State needs to be deleted to not store it in database
+ pipelineExecutor.getGraphs().getEntitiesToStart().forEach(entity -> entity.setState(null));
PipelineOperationStatus status = CommunicationUtils.
- startPipelineElementsAndRelays(decryptedGraphs, relays, associatedPipelineExecutor.getPipeline());
+ startPipelineElementsAndRelays(graphs, relays, pipelineExecutor.getPipeline());
long duration = System.nanoTime() - nanoTimeBeforeOperation;
EvaluationLogger.getInstance().logMQTT("Migration", "start target element", "", duration, duration/1000000000.0);
@@ -64,17 +56,27 @@ public class StartTargetPipelineElementsOperation extends PipelineExecutionOpera
@Override
public PipelineOperationStatus rollbackOperationFully() {
- List<InvocableStreamPipesEntity> graphs =
- Collections.singletonList(associatedPipelineExecutor.getMigrationEntity().getTargetElement());
- List<SpDataStreamRelayContainer> relays = PipelineElementUtils.extractRelaysFromDataProcessor(graphs);
+ List<InvocableStreamPipesEntity> graphsToRollBack = pipelineExecutor.getGraphs().getEntitiesToStart();
+ List<SpDataStreamRelayContainer> relaysToRollBack = PipelineElementUtils.extractRelaysFromDataProcessor(graphsToRollBack);
- return new GraphSubmitter(associatedPipelineExecutor.getPipeline().getPipelineId(),
- associatedPipelineExecutor.getPipeline().getName(), graphs, new ArrayList<>(), relays)
- .detachPipelineElementsAndRelays();
+ return CommunicationUtils.stopPipelineElementsAndRelays(graphsToRollBack,
+ relaysToRollBack, pipelineExecutor.getPipeline());
}
@Override
public PipelineOperationStatus rollbackOperationPartially() {
- return StatusUtils.initPipelineOperationStatus(associatedPipelineExecutor.getPipeline());
+ Set<String> idsToRollback = StatusUtils.extractUniqueSuccessfulIds(this.getStatus());
+
+ List<InvocableStreamPipesEntity> graphs = pipelineExecutor.getGraphs().getEntitiesToStart();
+ List<SpDataStreamRelayContainer> relays = PipelineElementUtils.extractRelaysFromDataProcessor(graphs);
+
+ List<InvocableStreamPipesEntity> graphsToRollBack =
+ PipelineElementUtils.filterPipelineElementsById(graphs, idsToRollback);
+
+ List<SpDataStreamRelayContainer> relaysToRollBack =
+ RelayUtils.filterRelaysById(relays, idsToRollback);
+
+ return CommunicationUtils.stopPipelineElementsAndRelays(graphsToRollBack,
+ relaysToRollBack, pipelineExecutor.getPipeline());
}
}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StartPipelineOperation.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StartPipelineOperation.java
index 29bc7e6..720d8c5 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StartPipelineOperation.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StartPipelineOperation.java
@@ -19,22 +19,19 @@ package org.apache.streampipes.manager.execution.pipeline.executor.operations;
import org.apache.streampipes.manager.execution.http.GraphSubmitter;
import org.apache.streampipes.manager.execution.pipeline.executor.PipelineExecutor;
+import org.apache.streampipes.manager.execution.pipeline.executor.utils.DataSetUtils;
import org.apache.streampipes.manager.execution.pipeline.executor.utils.PipelineElementUtils;
+import org.apache.streampipes.manager.execution.pipeline.executor.utils.RelayUtils;
import org.apache.streampipes.manager.execution.pipeline.executor.utils.StatusUtils;
import org.apache.streampipes.model.SpDataSet;
import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
import org.apache.streampipes.model.eventrelay.SpDataStreamRelayContainer;
-import org.apache.streampipes.model.graph.DataProcessorInvocation;
-import org.apache.streampipes.model.graph.DataSinkInvocation;
-import org.apache.streampipes.model.grounding.KafkaTransportProtocol;
import org.apache.streampipes.model.pipeline.Pipeline;
import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
-import java.util.ArrayList;
-import java.util.Collections;
import java.util.List;
-import java.util.UUID;
-import java.util.stream.Collectors;
+import java.util.Set;
+
public class StartPipelineOperation extends PipelineExecutionOperation{
@@ -44,61 +41,44 @@ public class StartPipelineOperation extends PipelineExecutionOperation{
@Override
public PipelineOperationStatus executeOperation() {
- Pipeline pipeline = associatedPipelineExecutor.getPipeline();
- pipeline.getSepas().forEach(this::updateKafkaGroupIds);
- pipeline.getActions().forEach(this::updateKafkaGroupIds);
-
- List<DataProcessorInvocation> sepas = pipeline.getSepas();
- List<DataSinkInvocation> secs = pipeline.getActions();
-
- List<SpDataSet> dataSets = pipeline.getStreams().stream().filter(s -> s instanceof SpDataSet).map(s -> new
- SpDataSet((SpDataSet) s)).collect(Collectors.toList());
-
- for (SpDataSet ds : dataSets) {
- ds.setCorrespondingPipeline(pipeline.getPipelineId());
- }
-
- List<InvocableStreamPipesEntity> graphs = new ArrayList<>();
- graphs.addAll(sepas);
- graphs.addAll(secs);
-
- List<InvocableStreamPipesEntity> decryptedGraphs = PipelineElementUtils.decryptSecrets(graphs, pipeline);
-
- graphs.forEach(g -> g.setStreamRequirements(Collections.emptyList()));
-
- List<SpDataStreamRelayContainer> relays = PipelineElementUtils.generateRelays(decryptedGraphs, pipeline);
-
- associatedPipelineExecutor.setGraphs(graphs);
- associatedPipelineExecutor.setDataSets(dataSets);
- associatedPipelineExecutor.setRelays(relays);
-
+ Pipeline pipeline = pipelineExecutor.getPipeline();
return new GraphSubmitter(pipeline.getPipelineId(), pipeline.getName(),
- decryptedGraphs, dataSets, relays).invokePipelineElementsAndRelays();
+ pipelineExecutor.getGraphs().getEntitiesToStart(),
+ pipelineExecutor.getDataSets().getEntitiesToStart(),
+ pipelineExecutor.getRelays().getEntitiesToStart()).invokePipelineElementsAndRelays();
}
@Override
public PipelineOperationStatus rollbackOperationPartially() {
- //TODO: Check if there is a possible realization of partial rollback
- return StatusUtils.initPipelineOperationStatus(associatedPipelineExecutor.getPipeline());
+ Pipeline pipeline = pipelineExecutor.getPipeline();
+
+ Set<String> idsToRollback = StatusUtils.extractUniqueSuccessfulIds(this.getStatus());
+
+ List<InvocableStreamPipesEntity> graphsToRollBack =
+ PipelineElementUtils.filterPipelineElementsById(
+ pipelineExecutor.getGraphs().getEntitiesToStart(),
+ idsToRollback);
+ List<SpDataSet> dataSetsToRollBack =
+ DataSetUtils.filterDataSetsById(
+ pipelineExecutor.getDataSets().getEntitiesToStart(),
+ idsToRollback);
+ List<SpDataStreamRelayContainer> relaysToRollBack =
+ RelayUtils.filterRelaysById(
+ pipelineExecutor.getRelays().getEntitiesToStart(),
+ idsToRollback);
+
+ return new GraphSubmitter(pipeline.getPipelineId(), pipeline.getName(),
+ graphsToRollBack,
+ dataSetsToRollBack,
+ relaysToRollBack).detachPipelineElementsAndRelays();
}
@Override
public PipelineOperationStatus rollbackOperationFully() {
- //TODO: Implement full rollback
- return StatusUtils.initPipelineOperationStatus(associatedPipelineExecutor.getPipeline());
- }
-
- /**
- * Updates group.id for data processor/sink. Note: KafkaTransportProtocol only!!
- *
- * @param entity data processor/sink
- */
- private void updateKafkaGroupIds(InvocableStreamPipesEntity entity) {
- entity.getInputStreams()
- .stream()
- .filter(is -> is.getEventGrounding().getTransportProtocol() instanceof KafkaTransportProtocol)
- .map(is -> is.getEventGrounding().getTransportProtocol())
- .map(KafkaTransportProtocol.class::cast)
- .forEach(tp -> tp.setGroupId(UUID.randomUUID().toString()));
+ Pipeline pipeline = pipelineExecutor.getPipeline();
+ return new GraphSubmitter(pipeline.getPipelineId(), pipeline.getName(),
+ pipelineExecutor.getGraphs().getEntitiesToStart(),
+ pipelineExecutor.getDataSets().getEntitiesToStart(),
+ pipelineExecutor.getRelays().getEntitiesToStart()).detachPipelineElementsAndRelays();
}
}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StartRelaysFromPredecessorsOperation.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StartRelaysOperation.java
similarity index 56%
rename from streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StartRelaysFromPredecessorsOperation.java
rename to streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StartRelaysOperation.java
index 90bbf96..07c6480 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StartRelaysFromPredecessorsOperation.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StartRelaysOperation.java
@@ -28,26 +28,20 @@ import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
import java.util.List;
import java.util.Set;
-import java.util.stream.Collectors;
-public class StartRelaysFromPredecessorsOperation extends PipelineExecutionOperation implements MigrationOperation {
+public class StartRelaysOperation extends PipelineExecutionOperation implements MigrationOperation {
- public StartRelaysFromPredecessorsOperation(PipelineExecutor pipelineExecutor) {
+ public StartRelaysOperation(PipelineExecutor pipelineExecutor) {
super(pipelineExecutor);
}
@Override
public PipelineOperationStatus executeOperation() {
long nanoTimeBeforeOperation = System.nanoTime();
- List<SpDataStreamRelayContainer> relays = RelayUtils.findRelays(
- associatedPipelineExecutor.getPredecessorsAfterMigration(),
- associatedPipelineExecutor.getMigrationEntity().getTargetElement(),
- associatedPipelineExecutor.getPipeline());
+ List<SpDataStreamRelayContainer> relays = pipelineExecutor.getRelays().getEntitiesToStart();
- RelayUtils.updateRelays(relays, associatedPipelineExecutor.getRelaysToBePersisted());
-
- PipelineOperationStatus status= CommunicationUtils.startRelays(relays, associatedPipelineExecutor.getPipeline());
+ PipelineOperationStatus status= CommunicationUtils.startRelays(relays, pipelineExecutor.getPipeline());
long duration = System.nanoTime() - nanoTimeBeforeOperation;
EvaluationLogger.getInstance().logMQTT("Migration", "start relay to target", "", duration, duration/1000000000.0);
@@ -57,23 +51,17 @@ public class StartRelaysFromPredecessorsOperation extends PipelineExecutionOpera
@Override
public PipelineOperationStatus rollbackOperationPartially() {
- Set<String> relayIdsToRollback = StatusUtils.extractUniqueRelayIds(this.getStatus());
- List<SpDataStreamRelayContainer> relaysFromPredecessors = RelayUtils.findRelays(
- associatedPipelineExecutor.getPredecessorsAfterMigration(),
- associatedPipelineExecutor.getMigrationEntity().getTargetElement(),
- associatedPipelineExecutor.getPipeline());
- List<SpDataStreamRelayContainer> relaysToRollBack = relaysFromPredecessors.stream().
- filter(relay -> relayIdsToRollback.contains(relay.getRunningStreamRelayInstanceId()))
- .collect(Collectors.toList());
- return CommunicationUtils.stopRelays(relaysToRollBack, associatedPipelineExecutor.getPipeline());
+ Set<String> relayIdsToRollback = StatusUtils.extractUniqueSuccessfulIds(this.getStatus());
+ List<SpDataStreamRelayContainer> relaysToRollBack =
+ RelayUtils.filterRelaysById(pipelineExecutor.getRelays().getEntitiesToStart(),
+ relayIdsToRollback);
+
+ return CommunicationUtils.stopRelays(relaysToRollBack, pipelineExecutor.getPipeline());
}
@Override
public PipelineOperationStatus rollbackOperationFully() {
- List<SpDataStreamRelayContainer> relaysToRollBack = RelayUtils.findRelays(
- associatedPipelineExecutor.getPredecessorsAfterMigration(),
- associatedPipelineExecutor.getMigrationEntity().getTargetElement(),
- associatedPipelineExecutor.getPipeline());
- return CommunicationUtils.stopRelays(relaysToRollBack, associatedPipelineExecutor.getPipeline());
+ List<SpDataStreamRelayContainer> relaysToRollBack = pipelineExecutor.getRelays().getEntitiesToStart();
+ return CommunicationUtils.stopRelays(relaysToRollBack, pipelineExecutor.getPipeline());
}
}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StopOriginPipelineElementAndRelaysOperation.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StopGraphsAndAssociatedRelaysOperation.java
similarity index 62%
rename from streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StopOriginPipelineElementAndRelaysOperation.java
rename to streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StopGraphsAndAssociatedRelaysOperation.java
index 42f7c3d..2e1a0d0 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StopOriginPipelineElementAndRelaysOperation.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StopGraphsAndAssociatedRelaysOperation.java
@@ -23,29 +23,29 @@ import org.apache.streampipes.manager.execution.pipeline.executor.utils.Communic
import org.apache.streampipes.manager.execution.pipeline.executor.utils.PipelineElementUtils;
import org.apache.streampipes.manager.execution.pipeline.executor.PipelineExecutor;
import org.apache.streampipes.manager.execution.pipeline.executor.utils.RelayUtils;
+import org.apache.streampipes.manager.execution.pipeline.executor.utils.StatusUtils;
import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
import org.apache.streampipes.model.eventrelay.SpDataStreamRelayContainer;
import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
-import java.util.Collections;
import java.util.List;
+import java.util.Set;
-public class StopOriginPipelineElementAndRelaysOperation extends PipelineExecutionOperation implements MigrationOperation {
+public class StopGraphsAndAssociatedRelaysOperation extends PipelineExecutionOperation implements MigrationOperation {
- public StopOriginPipelineElementAndRelaysOperation(PipelineExecutor pipelineExecutor) {
+ public StopGraphsAndAssociatedRelaysOperation(PipelineExecutor pipelineExecutor) {
super(pipelineExecutor);
}
@Override
public PipelineOperationStatus executeOperation() {
long nanoTimeBeforeOperation = System.nanoTime();
- List<InvocableStreamPipesEntity> graphs = Collections.singletonList(
- associatedPipelineExecutor.getMigrationEntity().getSourceElement());
+ List<InvocableStreamPipesEntity> graphs = pipelineExecutor.getGraphs().getEntitiesToStop();
List<SpDataStreamRelayContainer> relays = PipelineElementUtils.extractRelaysFromDataProcessor(graphs);
- RelayUtils.updateRelays(relays, associatedPipelineExecutor.getRelaysToBeDeleted());
+ RelayUtils.updateRelays(relays, pipelineExecutor.getRelays().getEntitiesToDelete());
- PipelineOperationStatus status = CommunicationUtils.stopPipelineElementsAndRelays(graphs, relays, associatedPipelineExecutor.getPipeline());
+ PipelineOperationStatus status = CommunicationUtils.stopPipelineElementsAndRelays(graphs, relays, pipelineExecutor.getPipeline());
long duration = System.nanoTime() - nanoTimeBeforeOperation;
EvaluationLogger.getInstance().logMQTT("Migration", "stop origin element", "", duration, duration/1000000000.0);
@@ -55,17 +55,25 @@ public class StopOriginPipelineElementAndRelaysOperation extends PipelineExecuti
@Override
public PipelineOperationStatus rollbackOperationPartially() {
- //TODO: Implement
- return null;
+ Set<String> idsToRollback = StatusUtils.extractUniqueSuccessfulIds(this.getStatus());
+
+ List<InvocableStreamPipesEntity> graphs = pipelineExecutor.getGraphs().getEntitiesToStop();
+ List<SpDataStreamRelayContainer> relays = PipelineElementUtils.extractRelaysFromDataProcessor(graphs);
+
+ List<InvocableStreamPipesEntity> graphsToRollBack =
+ PipelineElementUtils.filterPipelineElementsById(graphs, idsToRollback);
+
+ List<SpDataStreamRelayContainer> relaysToRollBack =
+ RelayUtils.filterRelaysById(relays, idsToRollback);
+
+ return CommunicationUtils.startPipelineElementsAndRelays(graphsToRollBack, relaysToRollBack,
+ pipelineExecutor.getPipeline());
}
@Override
public PipelineOperationStatus rollbackOperationFully() {
- List<InvocableStreamPipesEntity> graphs = Collections.singletonList(
- associatedPipelineExecutor.getMigrationEntity().getSourceElement());
- List<SpDataStreamRelayContainer> relays = PipelineElementUtils.extractRelaysFromDataProcessor(graphs);
-
- RelayUtils.updateRelays(relays, associatedPipelineExecutor.getRelaysToBeDeleted());
- return CommunicationUtils.startPipelineElementsAndRelays(graphs, relays, associatedPipelineExecutor.getPipeline());
+ List<InvocableStreamPipesEntity> graphsToRollBack = pipelineExecutor.getGraphs().getEntitiesToStop();
+ List<SpDataStreamRelayContainer> relaysToRollBack = PipelineElementUtils.extractRelaysFromDataProcessor(graphsToRollBack);
+ return CommunicationUtils.startPipelineElementsAndRelays(graphsToRollBack, relaysToRollBack, pipelineExecutor.getPipeline());
}
}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StopPipelineOperation.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StopPipelineOperation.java
index 2e8f367..1f28247 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StopPipelineOperation.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StopPipelineOperation.java
@@ -19,9 +19,7 @@ package org.apache.streampipes.manager.execution.pipeline.executor.operations;
import org.apache.streampipes.manager.execution.http.GraphSubmitter;
import org.apache.streampipes.manager.execution.pipeline.executor.PipelineExecutor;
-import org.apache.streampipes.manager.execution.pipeline.executor.utils.PipelineElementUtils;
-import org.apache.streampipes.manager.execution.pipeline.executor.utils.StatusUtils;
-import org.apache.streampipes.manager.execution.pipeline.executor.utils.StorageUtils;
+import org.apache.streampipes.manager.execution.pipeline.executor.utils.*;
import org.apache.streampipes.manager.execution.status.PipelineStatusManager;
import org.apache.streampipes.manager.util.TemporaryGraphStorage;
import org.apache.streampipes.model.SpDataSet;
@@ -33,6 +31,7 @@ import org.apache.streampipes.model.pipeline.Pipeline;
import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
import java.util.List;
+import java.util.Set;
public class StopPipelineOperation extends PipelineExecutionOperation {
@@ -42,7 +41,7 @@ public class StopPipelineOperation extends PipelineExecutionOperation {
@Override
public PipelineOperationStatus executeOperation() {
- Pipeline pipeline = associatedPipelineExecutor.getPipeline();
+ Pipeline pipeline = pipelineExecutor.getPipeline();
List<InvocableStreamPipesEntity> graphs = TemporaryGraphStorage.graphStorage.get(pipeline.getPipelineId());
List<SpDataSet> dataSets = TemporaryGraphStorage.datasetStorage.get(pipeline.getPipelineId());
List<SpDataStreamRelayContainer> relays = PipelineElementUtils.generateRelays(graphs, pipeline);
@@ -51,8 +50,8 @@ public class StopPipelineOperation extends PipelineExecutionOperation {
dataSets, relays).detachPipelineElementsAndRelays();
if (status.isSuccess()) {
- if (associatedPipelineExecutor.isMonitor()) StorageUtils.deleteVisualization(pipeline.getPipelineId());
- if (associatedPipelineExecutor.isStoreStatus()) StorageUtils.setPipelineStopped(pipeline);
+ if (pipelineExecutor.isMonitor()) StorageUtils.deleteVisualization(pipeline.getPipelineId());
+ if (pipelineExecutor.isStoreStatus()) StorageUtils.setPipelineStopped(pipeline);
StorageUtils.deleteDataStreamRelayContainer(relays);
@@ -67,13 +66,32 @@ public class StopPipelineOperation extends PipelineExecutionOperation {
@Override
public PipelineOperationStatus rollbackOperationPartially() {
- //TODO: Implement sth?
- return StatusUtils.initPipelineOperationStatus(associatedPipelineExecutor.getPipeline());
+ Pipeline pipeline = pipelineExecutor.getPipeline();
+
+ Set<String> idsToRollback = StatusUtils.extractUniqueSuccessfulIds(this.getStatus());
+
+ List<InvocableStreamPipesEntity> graphsToRollBack =
+ PipelineElementUtils.filterPipelineElementsById(
+ pipelineExecutor.getGraphs().getEntitiesToStop(),
+ idsToRollback);
+ List<SpDataSet> dataSetsToRollBack =
+ DataSetUtils.filterDataSetsById(
+ pipelineExecutor.getDataSets().getEntitiesToStart(),
+ idsToRollback);
+ List<SpDataStreamRelayContainer> relaysToRollBack =
+ RelayUtils.filterRelaysById(
+ pipelineExecutor.getRelays().getEntitiesToStart(),
+ idsToRollback);
+
+ return new GraphSubmitter(pipeline.getPipelineId(), pipeline.getName(),
+ graphsToRollBack,
+ dataSetsToRollBack,
+ relaysToRollBack).invokePipelineElementsAndRelays();
}
@Override
public PipelineOperationStatus rollbackOperationFully() {
//TODO: Implement sth?
- return StatusUtils.initPipelineOperationStatus(associatedPipelineExecutor.getPipeline());
+ return StatusUtils.initPipelineOperationStatus(pipelineExecutor.getPipeline());
}
}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StopRelaysFromPredecessorOperation.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StopRelaysOperation.java
similarity index 57%
rename from streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StopRelaysFromPredecessorOperation.java
rename to streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StopRelaysOperation.java
index c1ee4ad..dee1933 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StopRelaysFromPredecessorOperation.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StopRelaysOperation.java
@@ -21,7 +21,6 @@ import org.apache.streampipes.logging.evaluation.EvaluationLogger;
import org.apache.streampipes.manager.execution.pipeline.executor.*;
import org.apache.streampipes.manager.execution.pipeline.executor.operations.types.MigrationOperation;
import org.apache.streampipes.manager.execution.pipeline.executor.utils.CommunicationUtils;
-import org.apache.streampipes.manager.execution.pipeline.executor.utils.PipelineElementUtils;
import org.apache.streampipes.manager.execution.pipeline.executor.utils.RelayUtils;
import org.apache.streampipes.manager.execution.pipeline.executor.utils.StatusUtils;
import org.apache.streampipes.model.eventrelay.SpDataStreamRelayContainer;
@@ -30,24 +29,19 @@ import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
import java.util.List;
import java.util.Set;
-public class StopRelaysFromPredecessorOperation extends PipelineExecutionOperation implements MigrationOperation {
+public class StopRelaysOperation extends PipelineExecutionOperation implements MigrationOperation {
- public StopRelaysFromPredecessorOperation(PipelineExecutor pipelineExecutor){
+ public StopRelaysOperation(PipelineExecutor pipelineExecutor){
super(pipelineExecutor);
}
@Override
public PipelineOperationStatus executeOperation() {
long nanoTimeBeforeOperation = System.nanoTime();
- List<SpDataStreamRelayContainer> relays = RelayUtils
- .findRelaysWhenStopping(associatedPipelineExecutor.getPredecessorsBeforeMigration(),
- associatedPipelineExecutor.getMigrationEntity().getSourceElement(),
- associatedPipelineExecutor.getPipeline());
-
- RelayUtils.updateRelays(relays, associatedPipelineExecutor.getRelaysToBeDeleted());
+ List<SpDataStreamRelayContainer> relays = pipelineExecutor.getRelays().getEntitiesToStop();
PipelineOperationStatus statusStopRelays =
- CommunicationUtils.stopRelays(relays, associatedPipelineExecutor.getPipeline());
+ CommunicationUtils.stopRelays(relays, pipelineExecutor.getPipeline());
long duration = System.nanoTime() - nanoTimeBeforeOperation;
EvaluationLogger.getInstance().logMQTT("Migration", "stop relay from origin", "", duration, duration/1000000000.0);
@@ -57,21 +51,17 @@ public class StopRelaysFromPredecessorOperation extends PipelineExecutionOperati
@Override
public PipelineOperationStatus rollbackOperationPartially() {
- Set<String> relayIdsToRollback = StatusUtils.extractUniqueRelayIds(this.getStatus());
- List<SpDataStreamRelayContainer> rollbackRelays = PipelineElementUtils.findRelaysAndFilterById(relayIdsToRollback,
- associatedPipelineExecutor.getPredecessorsBeforeMigration(),
- associatedPipelineExecutor.getMigrationEntity().getSourceElement(),
- associatedPipelineExecutor.getPipeline());
+ Set<String> relayIdsToRollback = StatusUtils.extractUniqueSuccessfulIds(this.getStatus());
+ List<SpDataStreamRelayContainer> rollbackRelays =
+ RelayUtils.filterRelaysById(pipelineExecutor.getRelays().getEntitiesToStop(),
+ relayIdsToRollback);
- return CommunicationUtils.startRelays(rollbackRelays, associatedPipelineExecutor.getPipeline());
+ return CommunicationUtils.startRelays(rollbackRelays, pipelineExecutor.getPipeline());
}
@Override
public PipelineOperationStatus rollbackOperationFully() {
- List<SpDataStreamRelayContainer> rollbackRelays = RelayUtils.findRelays(
- associatedPipelineExecutor.getPredecessorsBeforeMigration(),
- associatedPipelineExecutor.getMigrationEntity().getSourceElement(),
- associatedPipelineExecutor.getPipeline());
- return CommunicationUtils.startRelays(rollbackRelays, associatedPipelineExecutor.getPipeline());
+ List<SpDataStreamRelayContainer> rollbackRelays = pipelineExecutor.getRelays().getEntitiesToStop();
+ return CommunicationUtils.startRelays(rollbackRelays, pipelineExecutor.getPipeline());
}
}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StoreMigratedPipelineOperation.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StoreMigratedPipelineOperation.java
index e691e9d..895e38d 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StoreMigratedPipelineOperation.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StoreMigratedPipelineOperation.java
@@ -22,10 +22,8 @@ import org.apache.streampipes.manager.execution.pipeline.executor.utils.Pipeline
import org.apache.streampipes.manager.execution.pipeline.executor.utils.StatusUtils;
import org.apache.streampipes.manager.execution.pipeline.executor.utils.StorageUtils;
import org.apache.streampipes.model.SpDataSet;
-import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
-import java.util.ArrayList;
import java.util.List;
public class StoreMigratedPipelineOperation extends PipelineExecutionOperation{
@@ -36,26 +34,23 @@ public class StoreMigratedPipelineOperation extends PipelineExecutionOperation{
@Override
public PipelineOperationStatus executeOperation() {
- List<InvocableStreamPipesEntity> graphs = new ArrayList<>();
- graphs.addAll(associatedPipelineExecutor.getPipeline().getActions());
- graphs.addAll(associatedPipelineExecutor.getPipeline().getSepas());
- List<SpDataSet> dataSets = PipelineUtils.findDataSets(associatedPipelineExecutor.getPipeline());
+ List<SpDataSet> dataSets = PipelineUtils.findDataSets(pipelineExecutor.getPipeline());
- // store new pipeline and relays
- StorageUtils.storeInvocationGraphs(associatedPipelineExecutor.getPipeline().getPipelineId(), graphs, dataSets);
- StorageUtils.deleteDataStreamRelayContainer(associatedPipelineExecutor.getRelaysToBeDeleted());
- StorageUtils.storeDataStreamRelayContainer(associatedPipelineExecutor.getRelaysToBePersisted());
- return StatusUtils.initPipelineOperationStatus(associatedPipelineExecutor.getPipeline());
+ StorageUtils.storeInvocationGraphs(pipelineExecutor.getPipeline().getPipelineId(),
+ pipelineExecutor.getGraphs().getEntitiesToStore(), dataSets);
+ StorageUtils.deleteDataStreamRelayContainer(pipelineExecutor.getRelays().getEntitiesToDelete());
+ StorageUtils.storeDataStreamRelayContainer(pipelineExecutor.getRelays().getEntitiesToStore());
+ return StatusUtils.initPipelineOperationStatus(pipelineExecutor.getPipeline());
}
@Override
public PipelineOperationStatus rollbackOperationPartially() {
- return null;
+ return StatusUtils.initPipelineOperationStatus(pipelineExecutor.getPipeline());
}
@Override
public PipelineOperationStatus rollbackOperationFully() {
- return null;
+ return StatusUtils.initPipelineOperationStatus(pipelineExecutor.getPipeline());
}
}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StorePipelineOperation.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StorePipelineOperation.java
index b0f47bb..960aec0 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StorePipelineOperation.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StorePipelineOperation.java
@@ -34,12 +34,12 @@ public class StorePipelineOperation extends PipelineExecutionOperation{
@Override
public PipelineOperationStatus executeOperation() {
- PipelineOperationStatus status = StatusUtils.initPipelineOperationStatus(associatedPipelineExecutor.getPipeline());
- Pipeline pipeline = associatedPipelineExecutor.getPipeline();
+ PipelineOperationStatus status = StatusUtils.initPipelineOperationStatus(pipelineExecutor.getPipeline());
+ Pipeline pipeline = pipelineExecutor.getPipeline();
try{
- StorageUtils.storeInvocationGraphs(pipeline.getPipelineId(), associatedPipelineExecutor.getGraphs(),
- associatedPipelineExecutor.getDataSets());
- StorageUtils.storeDataStreamRelayContainer(associatedPipelineExecutor.getRelays());
+ StorageUtils.storeInvocationGraphs(pipeline.getPipelineId(), pipelineExecutor.getGraphs().getEntitiesToStore(),
+ pipelineExecutor.getDataSets().getEntitiesToStore());
+ StorageUtils.storeDataStreamRelayContainer(pipelineExecutor.getRelays().getEntitiesToStore());
PipelineStatusManager.addPipelineStatus(
pipeline.getPipelineId(),
@@ -48,7 +48,7 @@ public class StorePipelineOperation extends PipelineExecutionOperation{
PipelineStatusMessageType.PIPELINE_STARTED.title(),
PipelineStatusMessageType.PIPELINE_STARTED.description()));
- if (associatedPipelineExecutor.isStoreStatus()) StorageUtils.setPipelineStarted(pipeline);
+ if (pipelineExecutor.isStoreStatus()) StorageUtils.setPipelineStarted(pipeline);
}catch (Exception e){
status.setSuccess(false);
status.setTitle(e.getMessage());
@@ -59,11 +59,11 @@ public class StorePipelineOperation extends PipelineExecutionOperation{
@Override
public PipelineOperationStatus rollbackOperationPartially() {
- return null;
+ return StatusUtils.initPipelineOperationStatus(pipelineExecutor.getPipeline());
}
@Override
public PipelineOperationStatus rollbackOperationFully() {
- return null;
+ return StatusUtils.initPipelineOperationStatus(pipelineExecutor.getPipeline());
}
}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/DataSetUtils.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/DataSetUtils.java
new file mode 100644
index 0000000..0f8deee
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/DataSetUtils.java
@@ -0,0 +1,20 @@
+package org.apache.streampipes.manager.execution.pipeline.executor.utils;
+
+import org.apache.streampipes.model.SpDataSet;
+import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
+
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class DataSetUtils {
+
+ public static List<SpDataSet> filterDataSetsById(List<SpDataSet> dataSets,
+ Set<String> dataSetIds) {
+ //TODO: Check if DatasetInvocationId is the correct id to check for
+ return dataSets.stream().
+ filter(dataSet -> dataSetIds.contains(dataSet.getDatasetInvocationId()))
+ .collect(Collectors.toList());
+ }
+
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/PipelineElementUtils.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/PipelineElementUtils.java
index 58f731c..83e4712 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/PipelineElementUtils.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/PipelineElementUtils.java
@@ -193,4 +193,11 @@ public class PipelineElementUtils {
.filter(r -> r.getOutputStreamRelays().size() > 0)
.collect(Collectors.toList());
}
+
+ public static List<InvocableStreamPipesEntity> filterPipelineElementsById(List<InvocableStreamPipesEntity> pipelineElements,
+ Set<String> pipelineElementIds) {
+ return pipelineElements.stream().
+ filter(pipelineElement -> pipelineElementIds.contains(pipelineElement.getDeploymentRunningInstanceId()))
+ .collect(Collectors.toList());
+ }
}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/RelayUtils.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/RelayUtils.java
index e88707e..ac0b210 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/RelayUtils.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/RelayUtils.java
@@ -233,4 +233,11 @@ public class RelayUtils {
String deploymentRunningInstanceId) {
return relays.stream().anyMatch(r -> r.getRunningStreamRelayInstanceId().equals(deploymentRunningInstanceId));
}
+
+ public static List<SpDataStreamRelayContainer> filterRelaysById(List<SpDataStreamRelayContainer> relays,
+ Set<String> relayIds) {
+ return relays.stream().
+ filter(relay -> relayIds.contains(relay.getRunningStreamRelayInstanceId()))
+ .collect(Collectors.toList());
+ }
}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/StatusUtils.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/StatusUtils.java
index 713045b..c4d8a4f 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/StatusUtils.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/StatusUtils.java
@@ -47,7 +47,7 @@ public class StatusUtils {
status.setTitle(partialStatus.getTitle());
}
- public static Set<String> extractUniqueRelayIds(PipelineOperationStatus status) {
+ public static Set<String> extractUniqueSuccessfulIds(PipelineOperationStatus status) {
return status.getElementStatus().stream()
.filter(PipelineElementStatus::isSuccess)
.map(PipelineElementStatus::getElementId)
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/PipelineElementMigrationHandler.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/PipelineElementMigrationHandler.java
index bf12bb9..e2dd478 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/PipelineElementMigrationHandler.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/PipelineElementMigrationHandler.java
@@ -87,7 +87,6 @@ public class PipelineElementMigrationHandler {
swapPipelineElement(migrationPipeline, failedEntity);
}
});
- //Why overwrite when changes are rolled back? TODO: Check if needed when pipeline is rolled back (else return)
Operations.overwritePipeline(migrationPipeline);
}
@@ -95,8 +94,6 @@ public class PipelineElementMigrationHandler {
PipelineExecutor migrationExecutor = PipelineExecutorFactory.
createMigrationExecutor(migrationPipeline, visualize, storeStatus, monitor, currentPipeline, migrationEntity);
return migrationExecutor.execute();
- //return new PipelineMigrationExecutor(migrationPipeline, currentPipeline, migrationEntity, visualize,
- // storeStatus, monitor).migratePipelineElement();
}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/reconfiguration/PipelineElementReconfigurationHandler.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/reconfiguration/PipelineElementReconfigurationHandler.java
index 959530d..ba5b7cb 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/reconfiguration/PipelineElementReconfigurationHandler.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/reconfiguration/PipelineElementReconfigurationHandler.java
@@ -60,8 +60,6 @@ public class PipelineElementReconfigurationHandler {
if (entityStatus.isSuccess()) {
Operations.overwritePipeline(reconfiguredPipeline);
- } else {
- //TODO: Send old/existing configuration
}
});
}
@@ -70,7 +68,6 @@ public class PipelineElementReconfigurationHandler {
return PipelineExecutorFactory
.createReconfigurationExecutor(this.storedPipeline, false, false, false, this.reconfiguredPipeline, entity)
.execute();
- //return new PipelineElementReconfigurationExecutor(reconfiguredPipeline, entity).reconfigurePipelineElement();
}
private List<PipelineElementReconfigurationEntity> comparePipelinesAndGetReconfiguration() {
@@ -103,12 +100,7 @@ public class PipelineElementReconfigurationHandler {
private PipelineElementReconfigurationEntity reconfigurationEntity(DataProcessorInvocation graph,
List<StaticProperty> adaptedStaticProperty) {
- PipelineElementReconfigurationEntity entity = new PipelineElementReconfigurationEntity();
- entity.setDeploymentRunningInstanceId(graph.getDeploymentRunningInstanceId());
- entity.setPipelineElementName(graph.getName());
- entity.setDeploymentTargetNodeId(graph.getDeploymentTargetNodeId());
- entity.setDeploymentTargetNodeHostname(graph.getDeploymentTargetNodeHostname());
- entity.setDeploymentTargetNodePort(graph.getDeploymentTargetNodePort());
+ PipelineElementReconfigurationEntity entity = new PipelineElementReconfigurationEntity(graph);
entity.setReconfiguredStaticProperties(adaptedStaticProperty);
return entity;
}
[incubator-streampipes] 01/03: [WIP] refactor pipeline executor
Posted by go...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
gomm pushed a commit to branch edge-extensions
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit 4820306ae6acde7a18b339e8f31720165bac1265
Author: daniel-gomm <da...@outlook.de>
AuthorDate: Wed Dec 8 13:28:52 2021 +0100
[WIP] refactor pipeline executor
---
.../api/InvocablePipelineElementResource.java | 7 +
.../model/base/InvocableStreamPipesEntity.java | 10 +
.../pipeline/executor/PipelineExecutor.java | 283 +++++++++++++++++++++
.../pipeline/executor/PipelineExecutorBuilder.java | 124 +++++++++
.../pipeline/executor/PipelineExecutorFactory.java | 63 +++++
.../executor/operations/GetStateOperation.java | 85 +++++++
.../executor/operations/LifecycleEntity.java | 38 +++
.../operations/PipelineExecutionOperation.java | 55 ++++
.../operations/PrepareMigrationOperation.java | 92 +++++++
.../operations/ReconfigureElementOperation.java | 49 ++++
.../operations/StartPipelineOperation.java | 104 ++++++++
.../StartRelaysFromPredecessorsOperation.java | 79 ++++++
.../StartTargetPipelineElementsOperation.java | 80 ++++++
...topOriginPipelineElementAndRelaysOperation.java | 71 ++++++
.../executor/operations/StopPipelineOperation.java | 79 ++++++
.../StopRelaysFromPredecessorOperation.java | 77 ++++++
.../operations/StoreMigratedPipelineOperation.java | 61 +++++
.../operations/StorePipelineOperation.java | 69 +++++
.../operations/types/MigrationOperation.java | 21 ++
.../operations/types/ReconfigurationOperation.java | 21 ++
.../executor/utils/CommunicationUtils.java | 71 ++++++
.../executor/utils/PipelineElementUtils.java | 196 ++++++++++++++
.../pipeline/executor/utils/PipelineUtils.java | 129 ++++++++++
.../pipeline/executor/utils/RelayUtils.java | 236 +++++++++++++++++
.../pipeline/executor/utils/StatusUtils.java | 61 +++++
.../pipeline/executor/utils/StorageUtils.java | 94 +++++++
.../migration/PipelineElementMigrationHandler.java | 12 +-
.../streampipes/manager/operations/Operations.java | 8 +-
.../PipelineElementReconfigurationHandler.java | 12 +-
29 files changed, 2273 insertions(+), 14 deletions(-)
diff --git a/streampipes-container/src/main/java/org/apache/streampipes/container/api/InvocablePipelineElementResource.java b/streampipes-container/src/main/java/org/apache/streampipes/container/api/InvocablePipelineElementResource.java
index 578de6d..3622e01 100644
--- a/streampipes-container/src/main/java/org/apache/streampipes/container/api/InvocablePipelineElementResource.java
+++ b/streampipes-container/src/main/java/org/apache/streampipes/container/api/InvocablePipelineElementResource.java
@@ -36,6 +36,9 @@ import org.slf4j.LoggerFactory;
import javax.ws.rs.*;
import javax.ws.rs.core.MediaType;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.ObjectOutputStream;
import java.util.List;
import java.util.Map;
@@ -72,6 +75,10 @@ public abstract class InvocablePipelineElementResource<I extends InvocableStream
if (!RunningInstances.INSTANCE.exists(runningInstanceId)) {
RunningInstances.INSTANCE.add(runningInstanceId, graph, declarer.getClass().newInstance());
Response resp = RunningInstances.INSTANCE.getInvocation(runningInstanceId).invokeRuntime(graph);
+ if(resp.isSuccess() && graph.getState() != null && graph.isStateful()){
+ //TODO: Handle fail while setting state
+ setState(elementId, runningInstanceId, graph.getState());
+ }
return ok(resp);
} else {
LOG.info("Pipeline element {} with id {} seems to be already running, skipping invocation request.", graph.getName(), runningInstanceId);
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/base/InvocableStreamPipesEntity.java b/streampipes-model/src/main/java/org/apache/streampipes/model/base/InvocableStreamPipesEntity.java
index de46346..a00b45a 100644
--- a/streampipes-model/src/main/java/org/apache/streampipes/model/base/InvocableStreamPipesEntity.java
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/base/InvocableStreamPipesEntity.java
@@ -113,6 +113,7 @@ public abstract class InvocableStreamPipesEntity extends NamedStreamPipesEntity
@RdfProperty(StreamPipes.IS_STATEFUL)
private boolean isStateful;
+ private String state;
public boolean isStateful() {
return isStateful;
@@ -122,6 +123,14 @@ public abstract class InvocableStreamPipesEntity extends NamedStreamPipesEntity
isStateful = stateful;
}
+ public String getState() {
+ return state;
+ }
+
+ public void setState(String state) {
+ this.state = state;
+ }
+
public InvocableStreamPipesEntity() {
super();
}
@@ -129,6 +138,7 @@ public abstract class InvocableStreamPipesEntity extends NamedStreamPipesEntity
public InvocableStreamPipesEntity(InvocableStreamPipesEntity other) {
super(other);
this.setStateful(other.isStateful());
+ this.state = other.getState();
this.belongsTo = other.getBelongsTo();
this.correspondingPipeline = other.getCorrespondingPipeline();
this.inputStreams = new Cloner().streams(other.getInputStreams());
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/PipelineExecutor.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/PipelineExecutor.java
new file mode 100644
index 0000000..0bd26ec
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/PipelineExecutor.java
@@ -0,0 +1,283 @@
+/*
+ * 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.execution.pipeline.executor;
+
+import org.apache.streampipes.manager.execution.pipeline.executor.operations.LifecycleEntity;
+import org.apache.streampipes.manager.execution.pipeline.executor.operations.PipelineExecutionOperation;
+import org.apache.streampipes.manager.execution.pipeline.executor.operations.types.MigrationOperation;
+import org.apache.streampipes.manager.execution.pipeline.executor.operations.types.ReconfigurationOperation;
+import org.apache.streampipes.manager.execution.pipeline.executor.utils.StatusUtils;
+import org.apache.streampipes.model.SpDataSet;
+import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
+import org.apache.streampipes.model.base.NamedStreamPipesEntity;
+import org.apache.streampipes.model.eventrelay.SpDataStreamRelayContainer;
+import org.apache.streampipes.model.pipeline.Pipeline;
+import org.apache.streampipes.model.pipeline.PipelineElementMigrationEntity;
+import org.apache.streampipes.model.pipeline.PipelineElementReconfigurationEntity;
+import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
+
+import java.util.ArrayList;
+import java.util.LinkedList;
+import java.util.List;
+
+public class PipelineExecutor {
+
+ private Pipeline pipeline;
+ private boolean visualize;
+ private boolean storeStatus;
+ private boolean monitor;
+
+ /**
+ * Old pipeline before migration
+ */
+ private Pipeline secondaryPipeline;
+
+ /**
+ * Pipeline element to be migrated. Contains pair of source and target element description
+ */
+ private PipelineElementMigrationEntity migrationEntity;
+
+ /**
+ * Predecessors of the pipeline element to be migrated in the migration pipeline
+ */
+ private List<NamedStreamPipesEntity> predecessorsAfterMigration;
+
+ /**
+ * Predecessors of the pipeline element to be migrated in the old pipeline before migration
+ */
+ private List<NamedStreamPipesEntity> predecessorsBeforeMigration;
+
+ /**
+ * Collection of relays that were created in the migration process needing to be stored
+ */
+ private List<SpDataStreamRelayContainer> relaysToBePersisted;
+
+ /**
+ * Collection of relays that were removed in the migration process needing to be deleted from persistent storage.
+ */
+ private List<SpDataStreamRelayContainer> relaysToBeDeleted;
+
+ private PipelineOperationStatus status;
+
+ private PipelineElementReconfigurationEntity reconfigurationEntity;
+
+ private final LinkedList<PipelineExecutionOperation> operations = new LinkedList<>();
+
+ private List<SpDataSet> dataSets;
+
+ private List<SpDataStreamRelayContainer> relays;
+
+ private List<InvocableStreamPipesEntity> graphs;
+
+ private final LifecycleEntity<InvocableStreamPipesEntity> grafs;
+
+ private final LifecycleEntity<SpDataStreamRelayContainer> relais;
+
+ private final LifecycleEntity<SpDataSet> dataZeds;
+
+ public PipelineExecutor(Pipeline pipeline, boolean visualize, boolean storeStatus, boolean monitor){
+ this.pipeline = pipeline;
+ this.visualize = visualize;
+ this.storeStatus = storeStatus;
+ this.monitor = monitor;
+ this.predecessorsAfterMigration = new ArrayList<>();
+ this.predecessorsBeforeMigration = new ArrayList<>();
+ this.relaysToBePersisted = new ArrayList<>();
+ this.relaysToBeDeleted = new ArrayList<>();
+ this.status = StatusUtils.initPipelineOperationStatus(pipeline);
+
+ this.grafs = new LifecycleEntity<>();
+ this.relais = new LifecycleEntity<>();
+ this.dataZeds = new LifecycleEntity<>();
+ }
+
+ public PipelineOperationStatus execute(){
+ for(PipelineExecutionOperation pipelineExecutionOperation: this.operations){
+ PipelineOperationStatus operationStatus = pipelineExecutionOperation.executeOperation();
+ StatusUtils.checkSuccess(operationStatus);
+ pipelineExecutionOperation.setStatus(operationStatus);
+ StatusUtils.updateStatus(operationStatus, this.status);
+ if(!operationStatus.isSuccess()){
+ rollback(pipelineExecutionOperation);
+ break;
+ }
+ }
+ StatusUtils.checkSuccess(this.status);
+ return this.status;
+ }
+
+ private void rollback(PipelineExecutionOperation failedOperation){
+ PipelineOperationStatus rollbackStatus = StatusUtils.initPipelineOperationStatus(pipeline);
+ for(int currentOperationIndex = this.operations.indexOf(failedOperation);
+ currentOperationIndex<=0; currentOperationIndex--){
+ PipelineExecutionOperation currentOperation = this.operations.get(currentOperationIndex);
+ PipelineOperationStatus rollbackOperationStatus = currentOperation.rollbackOperation();
+ StatusUtils.checkSuccess(rollbackOperationStatus);
+ StatusUtils.updateStatus(rollbackOperationStatus, rollbackStatus);
+ }
+ StatusUtils.checkSuccess(rollbackStatus);
+ StatusUtils.updateStatus(rollbackStatus, this.status);
+ }
+
+ //Getter and Setter
+
+ public void addOperation(PipelineExecutionOperation operation){
+ this.operations.add(operation);
+ }
+
+ public Pipeline getPipeline() {
+ return pipeline;
+ }
+
+ public void setPipeline(Pipeline pipeline) {
+ this.pipeline = pipeline;
+ }
+
+ public boolean isVisualize() {
+ return visualize;
+ }
+
+ public void setVisualize(boolean visualize) {
+ this.visualize = visualize;
+ }
+
+ public boolean isStoreStatus() {
+ return storeStatus;
+ }
+
+ public void setStoreStatus(boolean storeStatus) {
+ this.storeStatus = storeStatus;
+ }
+
+ public boolean isMonitor() {
+ return monitor;
+ }
+
+ public void setMonitor(boolean monitor) {
+ this.monitor = monitor;
+ }
+
+ public Pipeline getSecondaryPipeline() {
+ return secondaryPipeline;
+ }
+
+ public void setSecondaryPipeline(Pipeline secondaryPipeline) {
+ this.secondaryPipeline = secondaryPipeline;
+ }
+
+ public PipelineElementMigrationEntity getMigrationEntity() {
+ return migrationEntity;
+ }
+
+ public void setMigrationEntity(PipelineElementMigrationEntity migrationEntity) {
+ this.migrationEntity = migrationEntity;
+ }
+
+ public List<NamedStreamPipesEntity> getPredecessorsAfterMigration() {
+ return predecessorsAfterMigration;
+ }
+
+ public void setPredecessorsAfterMigration(List<NamedStreamPipesEntity> predecessorsAfterMigration) {
+ this.predecessorsAfterMigration = predecessorsAfterMigration;
+ }
+
+ public List<NamedStreamPipesEntity> getPredecessorsBeforeMigration() {
+ return predecessorsBeforeMigration;
+ }
+
+ public void setPredecessorsBeforeMigration(List<NamedStreamPipesEntity> predecessorsBeforeMigration) {
+ this.predecessorsBeforeMigration = predecessorsBeforeMigration;
+ }
+
+ public List<SpDataStreamRelayContainer> getRelaysToBePersisted() {
+ return relaysToBePersisted;
+ }
+
+ public void setRelaysToBePersisted(List<SpDataStreamRelayContainer> relaysToBePersisted) {
+ this.relaysToBePersisted = relaysToBePersisted;
+ }
+
+ public List<SpDataStreamRelayContainer> getRelaysToBeDeleted() {
+ return relaysToBeDeleted;
+ }
+
+ public void setRelaysToBeDeleted(List<SpDataStreamRelayContainer> relaysToBeDeleted) {
+ this.relaysToBeDeleted = relaysToBeDeleted;
+ }
+
+ public PipelineOperationStatus getStatus() {
+ return status;
+ }
+
+ public void setStatus(PipelineOperationStatus status) {
+ this.status = status;
+ }
+
+ public PipelineElementReconfigurationEntity getReconfigurationEntity() {
+ return reconfigurationEntity;
+ }
+
+ public void setReconfigurationEntity(PipelineElementReconfigurationEntity reconfigurationEntity) {
+ this.reconfigurationEntity = reconfigurationEntity;
+ }
+
+ public List<SpDataSet> getDataSets() {
+ return dataSets;
+ }
+
+ public void setDataSets(List<SpDataSet> dataSets) {
+ this.dataSets = dataSets;
+ }
+
+ public List<SpDataStreamRelayContainer> getRelays() {
+ return relays;
+ }
+
+ public void setRelays(List<SpDataStreamRelayContainer> relays) {
+ this.relays = relays;
+ }
+
+ public List<InvocableStreamPipesEntity> getGraphs() {
+ return graphs;
+ }
+
+ public void setGraphs(List<InvocableStreamPipesEntity> graphs) {
+ this.graphs = graphs;
+ }
+
+
+ public LifecycleEntity<InvocableStreamPipesEntity> getGrafs() {
+ return grafs;
+ }
+
+ public LifecycleEntity<SpDataStreamRelayContainer> getRelais() {
+ return relais;
+ }
+
+ public LifecycleEntity<SpDataSet> getDataZeds() {
+ return dataZeds;
+ }
+
+ public boolean containsReconfigurationOperation(){
+ return this.operations.stream().anyMatch(operation -> operation instanceof ReconfigurationOperation);
+ }
+
+ public boolean containsMigrationOperation(){
+ return this.operations.stream().anyMatch(operation -> operation instanceof MigrationOperation);
+ }
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/PipelineExecutorBuilder.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/PipelineExecutorBuilder.java
new file mode 100644
index 0000000..52fc870
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/PipelineExecutorBuilder.java
@@ -0,0 +1,124 @@
+/*
+ * 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.execution.pipeline.executor;
+
+import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import org.apache.streampipes.manager.execution.pipeline.executor.operations.*;
+import org.apache.streampipes.model.pipeline.Pipeline;
+import org.apache.streampipes.model.pipeline.PipelineElementMigrationEntity;
+import org.apache.streampipes.model.pipeline.PipelineElementReconfigurationEntity;
+
+public class PipelineExecutorBuilder {
+
+ private PipelineExecutor pipelineExecutor;
+
+ private boolean reconfigurationParametersSet = false;
+
+ private boolean migrationParametersSet = false;
+
+ public static PipelineExecutorBuilder getBuilder(){
+ return new PipelineExecutorBuilder();
+ }
+
+ private PipelineExecutorBuilder(){}
+
+ public PipelineExecutorBuilder initializePipelineExecutor(Pipeline pipeline, boolean visualize, boolean storeStatus, boolean monitor){
+ this.pipelineExecutor = new PipelineExecutor(pipeline, visualize, storeStatus, monitor);
+ return this;
+ }
+
+ public PipelineExecutorBuilder setMigrationParameters(Pipeline pipelineBeforeMigration,
+ PipelineElementMigrationEntity migrationEntity){
+ pipelineExecutor.setSecondaryPipeline(pipelineBeforeMigration);
+ pipelineExecutor.setMigrationEntity(migrationEntity);
+ this.migrationParametersSet = true;
+ return this;
+ }
+
+ public PipelineExecutorBuilder setReconfigurationParameters(Pipeline reconfiguredPipeline,
+ PipelineElementReconfigurationEntity reconfigurationEntity){
+ pipelineExecutor.setSecondaryPipeline(reconfiguredPipeline);
+ pipelineExecutor.setReconfigurationEntity(reconfigurationEntity);
+ this.reconfigurationParametersSet = true;
+ return this;
+ }
+
+ public PipelineExecutorBuilder addPrepareMigrationOperation(){
+ pipelineExecutor.addOperation(new PrepareMigrationOperation(pipelineExecutor));
+ return this;
+ }
+
+ public PipelineExecutorBuilder addGetStateOperation(){
+ pipelineExecutor.addOperation(new GetStateOperation(pipelineExecutor));
+ return this;
+ }
+
+ public PipelineExecutorBuilder addStartRelaysFromPredecessorsOperation(){
+ pipelineExecutor.addOperation(new StartRelaysFromPredecessorsOperation(pipelineExecutor));
+ return this;
+ }
+
+ public PipelineExecutorBuilder addStartTargetPipelineElementsOperation(){
+ pipelineExecutor.addOperation(new StartTargetPipelineElementsOperation(pipelineExecutor));
+ return this;
+ }
+
+ public PipelineExecutorBuilder addStopOriginPipelineElementAndRelaysOperation(){
+ pipelineExecutor.addOperation(new StopOriginPipelineElementAndRelaysOperation(pipelineExecutor));
+ return this;
+ }
+
+ public PipelineExecutorBuilder addStopRelaysFromPredecessorOperation(){
+ pipelineExecutor.addOperation(new StopRelaysFromPredecessorOperation(pipelineExecutor));
+ return this;
+ }
+
+ public PipelineExecutorBuilder addStoreMigratedPipelineOperation(){
+ pipelineExecutor.addOperation(new StoreMigratedPipelineOperation(pipelineExecutor));
+ return this;
+ }
+
+ public PipelineExecutorBuilder addReconfigureElementOperation(){
+ pipelineExecutor.addOperation(new ReconfigureElementOperation(pipelineExecutor));
+ return this;
+ }
+
+ public PipelineExecutorBuilder addStartPipelineOperation(){
+ pipelineExecutor.addOperation(new StartPipelineOperation(pipelineExecutor));
+ return this;
+ }
+
+ public PipelineExecutorBuilder addStorePipelineOperation(){
+ pipelineExecutor.addOperation(new StorePipelineOperation(pipelineExecutor));
+ return this;
+ }
+
+ public PipelineExecutorBuilder addStopPipelineOperation(){
+ pipelineExecutor.addOperation(new StopPipelineOperation(pipelineExecutor));
+ return this;
+ }
+
+ public PipelineExecutor buildPipelineExecutor(){
+ //Is this check needed? Only relevant for core development not for users but gives a little more clarity at the
+ //cost of introducing some new boolean flags and marker interfaces
+ if((this.pipelineExecutor.containsMigrationOperation() && !this.migrationParametersSet)
+ || (this.pipelineExecutor.containsReconfigurationOperation() && !this.reconfigurationParametersSet))
+ throw new SpRuntimeException("PipelineExecutor can't be build since the required parameters have not been set");
+ return this.pipelineExecutor; }
+
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/PipelineExecutorFactory.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/PipelineExecutorFactory.java
new file mode 100644
index 0000000..413edd3
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/PipelineExecutorFactory.java
@@ -0,0 +1,63 @@
+/*
+ * 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.execution.pipeline.executor;
+
+import org.apache.streampipes.model.pipeline.Pipeline;
+import org.apache.streampipes.model.pipeline.PipelineElementMigrationEntity;
+import org.apache.streampipes.model.pipeline.PipelineElementReconfigurationEntity;
+
+public class PipelineExecutorFactory {
+
+ public static PipelineExecutor createMigrationExecutor(Pipeline pipeline, boolean visualize, boolean storeStatus,
+ boolean monitor, Pipeline pipelineBeforeMigration,
+ PipelineElementMigrationEntity migrationEntity){
+ PipelineExecutorBuilder builder = PipelineExecutorBuilder.getBuilder()
+ .initializePipelineExecutor(pipeline, visualize, storeStatus, monitor)
+ .setMigrationParameters(pipelineBeforeMigration, migrationEntity)
+ .addPrepareMigrationOperation();
+ if(migrationEntity.getTargetElement().isStateful())
+ builder.addGetStateOperation();
+ builder.addStartTargetPipelineElementsOperation()
+ .addStopRelaysFromPredecessorOperation()
+ .addStartRelaysFromPredecessorsOperation()
+ .addStopOriginPipelineElementAndRelaysOperation()
+ .addStoreMigratedPipelineOperation();
+ return builder.buildPipelineExecutor();
+ }
+
+ public static PipelineExecutor createReconfigurationExecutor(Pipeline pipeline, boolean visualize, boolean storeStatus,
+ boolean monitor, Pipeline reconfiguredPipeline,
+ PipelineElementReconfigurationEntity reconfigurationEntity){
+ return PipelineExecutorBuilder.getBuilder().initializePipelineExecutor(pipeline, visualize, storeStatus, monitor)
+ .setReconfigurationParameters(reconfiguredPipeline, reconfigurationEntity).addReconfigureElementOperation()
+ .buildPipelineExecutor();
+ }
+
+ public static PipelineExecutor createInvocationExecutor(Pipeline pipeline, boolean visualize, boolean storeStatus,
+ boolean monitor){
+ return PipelineExecutorBuilder.getBuilder().initializePipelineExecutor(pipeline, visualize, storeStatus, monitor)
+ .addStartPipelineOperation().addStorePipelineOperation().buildPipelineExecutor();
+ }
+
+ public static PipelineExecutor createDetachExecutor(Pipeline pipeline, boolean visualize, boolean storeStatus,
+ boolean monitor){
+ return PipelineExecutorBuilder.getBuilder().initializePipelineExecutor(pipeline, visualize, storeStatus, monitor)
+ .addStopPipelineOperation().buildPipelineExecutor();
+ }
+
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/GetStateOperation.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/GetStateOperation.java
new file mode 100644
index 0000000..c0fe26c
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/GetStateOperation.java
@@ -0,0 +1,85 @@
+/*
+ * 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.execution.pipeline.executor.operations;
+
+import org.apache.streampipes.logging.evaluation.EvaluationLogger;
+import org.apache.streampipes.manager.execution.pipeline.executor.operations.types.MigrationOperation;
+import org.apache.streampipes.manager.execution.pipeline.executor.utils.CommunicationUtils;
+import org.apache.streampipes.manager.execution.pipeline.executor.PipelineExecutor;
+import org.apache.streampipes.manager.execution.pipeline.executor.utils.StatusUtils;
+import org.apache.streampipes.model.pipeline.PipelineElementStatus;
+import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.ObjectOutputStream;
+
+public class GetStateOperation extends PipelineExecutionOperation implements MigrationOperation {
+
+ public GetStateOperation(PipelineExecutor pipelineExecutor) {
+ super(pipelineExecutor);
+ }
+
+ @Override
+ public PipelineOperationStatus executeOperation() {
+ long nanoTimeBeforeOperation = System.nanoTime();
+ PipelineElementStatus statusGettingState = CommunicationUtils.getState(
+ associatedPipelineExecutor.getMigrationEntity().getSourceElement(),
+ associatedPipelineExecutor.getPipeline());
+ if(statusGettingState.isSuccess()) {
+ associatedPipelineExecutor.getMigrationEntity().getTargetElement()
+ .setState(statusGettingState.getOptionalMessage());
+ statusGettingState.setOptionalMessage("Successfully retrieved state");
+ }
+ PipelineOperationStatus getStateStatus = StatusUtils.initPipelineOperationStatus(associatedPipelineExecutor.getPipeline());
+ getStateStatus.addPipelineElementStatus(statusGettingState);
+ StatusUtils.checkSuccess(getStateStatus);
+ long duration = System.nanoTime() - nanoTimeBeforeOperation;
+ EvaluationLogger.getInstance().logMQTT("Migration", "get state", "", duration, duration/1000000000.0);
+ try {
+ int stateSize = getSizeInBytes(associatedPipelineExecutor.getMigrationEntity().getTargetElement().getState());
+ EvaluationLogger.getInstance().logMQTT("Migration", "state size", stateSize/1024.0, stateSize/1048576.0);
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ return getStateStatus;
+ }
+
+ private int getSizeInBytes(Object map) throws IOException {
+ // Measuring the size by serializing it and then measuring the bytes
+ ByteArrayOutputStream byteStream = new ByteArrayOutputStream();
+ ObjectOutputStream out = new ObjectOutputStream(byteStream);
+
+ out.writeObject(map);
+ out.close();
+
+ return byteStream.toByteArray().length;
+ }
+
+ @Override
+ public PipelineOperationStatus rollbackOperationPartially() {
+ associatedPipelineExecutor.getMigrationEntity().getTargetElement().setState(null);
+ return StatusUtils.initPipelineOperationStatus(associatedPipelineExecutor.getPipeline());
+ }
+
+ @Override
+ public PipelineOperationStatus rollbackOperationFully() {
+ associatedPipelineExecutor.getMigrationEntity().getTargetElement().setState(null);
+ return StatusUtils.initPipelineOperationStatus(associatedPipelineExecutor.getPipeline());
+ }
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/LifecycleEntity.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/LifecycleEntity.java
new file mode 100644
index 0000000..09777ba
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/LifecycleEntity.java
@@ -0,0 +1,38 @@
+package org.apache.streampipes.manager.execution.pipeline.executor.operations;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class LifecycleEntity<T> {
+
+ private final List<T> entitiesToStart;
+
+ private final List<T> entitiesToStop;
+
+ private final List<T> entitiesToStore;
+
+ private final List<T> entitiesToDelete;
+
+ public LifecycleEntity(){
+ entitiesToStart = new ArrayList<>();
+ entitiesToStop = new ArrayList<>();
+ entitiesToStore = new ArrayList<>();
+ entitiesToDelete = new ArrayList<>();
+ }
+
+ public List<T> getEntitiesToStart() {
+ return entitiesToStart;
+ }
+
+ public List<T> getEntitiesToStop() {
+ return entitiesToStop;
+ }
+
+ public List<T> getEntitiesToStore() {
+ return entitiesToStore;
+ }
+
+ public List<T> getEntitiesToDelete() {
+ return entitiesToDelete;
+ }
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/PipelineExecutionOperation.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/PipelineExecutionOperation.java
new file mode 100644
index 0000000..1d6d76a
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/PipelineExecutionOperation.java
@@ -0,0 +1,55 @@
+/*
+ * 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.execution.pipeline.executor.operations;
+
+import org.apache.streampipes.manager.execution.pipeline.executor.PipelineExecutor;
+import org.apache.streampipes.manager.execution.pipeline.executor.utils.StatusUtils;
+import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
+
+public abstract class PipelineExecutionOperation {
+
+ protected final PipelineExecutor associatedPipelineExecutor;
+
+ private PipelineOperationStatus status;
+
+ public PipelineExecutionOperation(PipelineExecutor pipelineExecutor){
+ this.associatedPipelineExecutor = pipelineExecutor;
+ this.status = StatusUtils.initPipelineOperationStatus(associatedPipelineExecutor.getPipeline());
+ }
+
+ public abstract PipelineOperationStatus executeOperation();
+
+ public abstract PipelineOperationStatus rollbackOperationPartially();
+
+ public abstract PipelineOperationStatus rollbackOperationFully();
+
+ public PipelineOperationStatus rollbackOperation(){
+ if(status.isSuccess()){
+ return rollbackOperationFully();
+ }
+ return rollbackOperationPartially();
+ }
+
+ public PipelineOperationStatus getStatus() {
+ return status;
+ }
+
+ public void setStatus(PipelineOperationStatus status){
+ this.status = status;
+ }
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/PrepareMigrationOperation.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/PrepareMigrationOperation.java
new file mode 100644
index 0000000..9b668b7
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/PrepareMigrationOperation.java
@@ -0,0 +1,92 @@
+/*
+ * 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.execution.pipeline.executor.operations;
+
+import org.apache.streampipes.manager.data.PipelineGraph;
+import org.apache.streampipes.manager.data.PipelineGraphBuilder;
+import org.apache.streampipes.manager.data.PipelineGraphHelpers;
+import org.apache.streampipes.manager.execution.pipeline.executor.PipelineExecutor;
+import org.apache.streampipes.manager.execution.pipeline.executor.operations.types.MigrationOperation;
+import org.apache.streampipes.manager.execution.pipeline.executor.utils.PipelineUtils;
+import org.apache.streampipes.manager.execution.pipeline.executor.utils.StatusUtils;
+import org.apache.streampipes.model.base.NamedStreamPipesEntity;
+import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
+import org.apache.streampipes.model.pipeline.Pipeline;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.Collectors;
+
+public class PrepareMigrationOperation extends PipelineExecutionOperation implements MigrationOperation {
+
+ public PrepareMigrationOperation(PipelineExecutor pipelineExecutor) {
+ super(pipelineExecutor);
+ }
+
+ @Override
+ public PipelineOperationStatus executeOperation() {
+ Pipeline pipeline = associatedPipelineExecutor.getPipeline();
+ //Purge existing relays
+ PipelineUtils.purgeExistingRelays(pipeline);
+
+ //Generate new relays
+ PipelineGraph pipelineGraphAfterMigration =
+ new PipelineGraphBuilder(pipeline).buildGraph();
+ PipelineUtils.buildPipelineGraph(pipelineGraphAfterMigration, pipeline.getPipelineId());
+
+ //Get predecessors
+ PipelineGraph pipelineGraphBeforeMigration =
+ new PipelineGraphBuilder(associatedPipelineExecutor.getSecondaryPipeline()).buildGraph();
+ findPredecessorsInMigrationPipeline(pipelineGraphAfterMigration);
+
+ //Find counterpart for predecessors in currentPipeline
+ findAndComparePredecessorsInCurrentPipeline(pipelineGraphBeforeMigration);
+ return StatusUtils.initPipelineOperationStatus(pipeline);
+ }
+
+ @Override
+ public PipelineOperationStatus rollbackOperationPartially() {
+ return null;
+ }
+
+ @Override
+ public PipelineOperationStatus rollbackOperationFully() {
+ return null;
+ }
+
+ private void findPredecessorsInMigrationPipeline(PipelineGraph pipelineGraphAfterMigration) {
+ // get unique list of predecessors
+ List<NamedStreamPipesEntity> predecessors = PipelineGraphHelpers.findStreams(pipelineGraphAfterMigration).stream()
+ .map(stream -> PipelineUtils.getPredecessors(stream,
+ associatedPipelineExecutor.getMigrationEntity().getTargetElement(),
+ pipelineGraphAfterMigration, new ArrayList<>()))
+ .flatMap(List::stream)
+ .collect(Collectors.toList())
+ .stream()
+ .distinct()
+ .collect(Collectors.toList());
+
+ associatedPipelineExecutor.getPredecessorsAfterMigration().addAll(predecessors);
+ }
+
+ private void findAndComparePredecessorsInCurrentPipeline(PipelineGraph pipelineGraphBeforeMigration) {
+ associatedPipelineExecutor.getPredecessorsAfterMigration().forEach(migrationPredecessor ->
+ associatedPipelineExecutor.getPredecessorsBeforeMigration()
+ .add(PipelineUtils.findMatching(migrationPredecessor, pipelineGraphBeforeMigration)));
+ }
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/ReconfigureElementOperation.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/ReconfigureElementOperation.java
new file mode 100644
index 0000000..8e0e8c2
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/ReconfigureElementOperation.java
@@ -0,0 +1,49 @@
+/*
+ * 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.execution.pipeline.executor.operations;
+
+import org.apache.streampipes.manager.execution.http.ReconfigurationSubmitter;
+import org.apache.streampipes.manager.execution.pipeline.executor.PipelineExecutor;
+import org.apache.streampipes.manager.execution.pipeline.executor.operations.types.ReconfigurationOperation;
+import org.apache.streampipes.manager.execution.pipeline.executor.utils.StatusUtils;
+import org.apache.streampipes.model.pipeline.Pipeline;
+import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
+
+public class ReconfigureElementOperation extends PipelineExecutionOperation implements ReconfigurationOperation {
+
+ public ReconfigureElementOperation(PipelineExecutor pipelineExecutor) {
+ super(pipelineExecutor);
+ }
+
+ @Override
+ public PipelineOperationStatus executeOperation() {
+ Pipeline reconfiguredPipeline = associatedPipelineExecutor.getSecondaryPipeline();
+ return new ReconfigurationSubmitter(reconfiguredPipeline.getPipelineId(), reconfiguredPipeline.getName(),
+ associatedPipelineExecutor.getReconfigurationEntity()).reconfigure();
+ }
+
+ @Override
+ public PipelineOperationStatus rollbackOperationPartially() {
+ return StatusUtils.initPipelineOperationStatus(associatedPipelineExecutor.getPipeline());
+ }
+
+ @Override
+ public PipelineOperationStatus rollbackOperationFully() {
+ return StatusUtils.initPipelineOperationStatus(associatedPipelineExecutor.getPipeline());
+ }
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StartPipelineOperation.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StartPipelineOperation.java
new file mode 100644
index 0000000..29bc7e6
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StartPipelineOperation.java
@@ -0,0 +1,104 @@
+/*
+ * 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.execution.pipeline.executor.operations;
+
+import org.apache.streampipes.manager.execution.http.GraphSubmitter;
+import org.apache.streampipes.manager.execution.pipeline.executor.PipelineExecutor;
+import org.apache.streampipes.manager.execution.pipeline.executor.utils.PipelineElementUtils;
+import org.apache.streampipes.manager.execution.pipeline.executor.utils.StatusUtils;
+import org.apache.streampipes.model.SpDataSet;
+import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
+import org.apache.streampipes.model.eventrelay.SpDataStreamRelayContainer;
+import org.apache.streampipes.model.graph.DataProcessorInvocation;
+import org.apache.streampipes.model.graph.DataSinkInvocation;
+import org.apache.streampipes.model.grounding.KafkaTransportProtocol;
+import org.apache.streampipes.model.pipeline.Pipeline;
+import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+import java.util.stream.Collectors;
+
+public class StartPipelineOperation extends PipelineExecutionOperation{
+
+ public StartPipelineOperation(PipelineExecutor pipelineExecutor) {
+ super(pipelineExecutor);
+ }
+
+ @Override
+ public PipelineOperationStatus executeOperation() {
+ Pipeline pipeline = associatedPipelineExecutor.getPipeline();
+ pipeline.getSepas().forEach(this::updateKafkaGroupIds);
+ pipeline.getActions().forEach(this::updateKafkaGroupIds);
+
+ List<DataProcessorInvocation> sepas = pipeline.getSepas();
+ List<DataSinkInvocation> secs = pipeline.getActions();
+
+ List<SpDataSet> dataSets = pipeline.getStreams().stream().filter(s -> s instanceof SpDataSet).map(s -> new
+ SpDataSet((SpDataSet) s)).collect(Collectors.toList());
+
+ for (SpDataSet ds : dataSets) {
+ ds.setCorrespondingPipeline(pipeline.getPipelineId());
+ }
+
+ List<InvocableStreamPipesEntity> graphs = new ArrayList<>();
+ graphs.addAll(sepas);
+ graphs.addAll(secs);
+
+ List<InvocableStreamPipesEntity> decryptedGraphs = PipelineElementUtils.decryptSecrets(graphs, pipeline);
+
+ graphs.forEach(g -> g.setStreamRequirements(Collections.emptyList()));
+
+ List<SpDataStreamRelayContainer> relays = PipelineElementUtils.generateRelays(decryptedGraphs, pipeline);
+
+ associatedPipelineExecutor.setGraphs(graphs);
+ associatedPipelineExecutor.setDataSets(dataSets);
+ associatedPipelineExecutor.setRelays(relays);
+
+ return new GraphSubmitter(pipeline.getPipelineId(), pipeline.getName(),
+ decryptedGraphs, dataSets, relays).invokePipelineElementsAndRelays();
+ }
+
+ @Override
+ public PipelineOperationStatus rollbackOperationPartially() {
+ //TODO: Check if there is a possible realization of partial rollback
+ return StatusUtils.initPipelineOperationStatus(associatedPipelineExecutor.getPipeline());
+ }
+
+ @Override
+ public PipelineOperationStatus rollbackOperationFully() {
+ //TODO: Implement full rollback
+ return StatusUtils.initPipelineOperationStatus(associatedPipelineExecutor.getPipeline());
+ }
+
+ /**
+ * Updates group.id for data processor/sink. Note: KafkaTransportProtocol only!!
+ *
+ * @param entity data processor/sink
+ */
+ private void updateKafkaGroupIds(InvocableStreamPipesEntity entity) {
+ entity.getInputStreams()
+ .stream()
+ .filter(is -> is.getEventGrounding().getTransportProtocol() instanceof KafkaTransportProtocol)
+ .map(is -> is.getEventGrounding().getTransportProtocol())
+ .map(KafkaTransportProtocol.class::cast)
+ .forEach(tp -> tp.setGroupId(UUID.randomUUID().toString()));
+ }
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StartRelaysFromPredecessorsOperation.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StartRelaysFromPredecessorsOperation.java
new file mode 100644
index 0000000..90bbf96
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StartRelaysFromPredecessorsOperation.java
@@ -0,0 +1,79 @@
+/*
+ * 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.execution.pipeline.executor.operations;
+
+import org.apache.streampipes.logging.evaluation.EvaluationLogger;
+import org.apache.streampipes.manager.execution.pipeline.executor.operations.types.MigrationOperation;
+import org.apache.streampipes.manager.execution.pipeline.executor.utils.CommunicationUtils;
+import org.apache.streampipes.manager.execution.pipeline.executor.PipelineExecutor;
+import org.apache.streampipes.manager.execution.pipeline.executor.utils.RelayUtils;
+import org.apache.streampipes.manager.execution.pipeline.executor.utils.StatusUtils;
+import org.apache.streampipes.model.eventrelay.SpDataStreamRelayContainer;
+import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
+
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class StartRelaysFromPredecessorsOperation extends PipelineExecutionOperation implements MigrationOperation {
+
+
+ public StartRelaysFromPredecessorsOperation(PipelineExecutor pipelineExecutor) {
+ super(pipelineExecutor);
+ }
+
+ @Override
+ public PipelineOperationStatus executeOperation() {
+ long nanoTimeBeforeOperation = System.nanoTime();
+ List<SpDataStreamRelayContainer> relays = RelayUtils.findRelays(
+ associatedPipelineExecutor.getPredecessorsAfterMigration(),
+ associatedPipelineExecutor.getMigrationEntity().getTargetElement(),
+ associatedPipelineExecutor.getPipeline());
+
+ RelayUtils.updateRelays(relays, associatedPipelineExecutor.getRelaysToBePersisted());
+
+ PipelineOperationStatus status= CommunicationUtils.startRelays(relays, associatedPipelineExecutor.getPipeline());
+
+ long duration = System.nanoTime() - nanoTimeBeforeOperation;
+ EvaluationLogger.getInstance().logMQTT("Migration", "start relay to target", "", duration, duration/1000000000.0);
+
+ return status;
+ }
+
+ @Override
+ public PipelineOperationStatus rollbackOperationPartially() {
+ Set<String> relayIdsToRollback = StatusUtils.extractUniqueRelayIds(this.getStatus());
+ List<SpDataStreamRelayContainer> relaysFromPredecessors = RelayUtils.findRelays(
+ associatedPipelineExecutor.getPredecessorsAfterMigration(),
+ associatedPipelineExecutor.getMigrationEntity().getTargetElement(),
+ associatedPipelineExecutor.getPipeline());
+ List<SpDataStreamRelayContainer> relaysToRollBack = relaysFromPredecessors.stream().
+ filter(relay -> relayIdsToRollback.contains(relay.getRunningStreamRelayInstanceId()))
+ .collect(Collectors.toList());
+ return CommunicationUtils.stopRelays(relaysToRollBack, associatedPipelineExecutor.getPipeline());
+ }
+
+ @Override
+ public PipelineOperationStatus rollbackOperationFully() {
+ List<SpDataStreamRelayContainer> relaysToRollBack = RelayUtils.findRelays(
+ associatedPipelineExecutor.getPredecessorsAfterMigration(),
+ associatedPipelineExecutor.getMigrationEntity().getTargetElement(),
+ associatedPipelineExecutor.getPipeline());
+ return CommunicationUtils.stopRelays(relaysToRollBack, associatedPipelineExecutor.getPipeline());
+ }
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StartTargetPipelineElementsOperation.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StartTargetPipelineElementsOperation.java
new file mode 100644
index 0000000..9e79ab8
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StartTargetPipelineElementsOperation.java
@@ -0,0 +1,80 @@
+/*
+ * 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.execution.pipeline.executor.operations;
+
+import org.apache.streampipes.logging.evaluation.EvaluationLogger;
+import org.apache.streampipes.manager.execution.http.GraphSubmitter;
+import org.apache.streampipes.manager.execution.pipeline.executor.*;
+import org.apache.streampipes.manager.execution.pipeline.executor.operations.types.MigrationOperation;
+import org.apache.streampipes.manager.execution.pipeline.executor.utils.CommunicationUtils;
+import org.apache.streampipes.manager.execution.pipeline.executor.utils.PipelineElementUtils;
+import org.apache.streampipes.manager.execution.pipeline.executor.utils.RelayUtils;
+import org.apache.streampipes.manager.execution.pipeline.executor.utils.StatusUtils;
+import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
+import org.apache.streampipes.model.eventrelay.SpDataStreamRelayContainer;
+import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+public class StartTargetPipelineElementsOperation extends PipelineExecutionOperation implements MigrationOperation {
+
+ public StartTargetPipelineElementsOperation(PipelineExecutor pipelineExecutor){
+ super(pipelineExecutor);
+ }
+
+ @Override
+ public PipelineOperationStatus executeOperation() {
+ long nanoTimeBeforeOperation = System.nanoTime();
+ List<InvocableStreamPipesEntity> decryptedGraphs =
+ PipelineElementUtils.decryptSecrets(
+ Collections.singletonList(associatedPipelineExecutor.getMigrationEntity().getTargetElement()),
+ associatedPipelineExecutor.getPipeline());
+ List<SpDataStreamRelayContainer> relays = PipelineElementUtils.extractRelaysFromDataProcessor(decryptedGraphs);
+
+ RelayUtils.updateRelays(relays, associatedPipelineExecutor.getRelaysToBePersisted());
+
+ //State needs to be deleted to not store it
+ associatedPipelineExecutor.getMigrationEntity().getTargetElement().setState(null);
+
+ PipelineOperationStatus status = CommunicationUtils.
+ startPipelineElementsAndRelays(decryptedGraphs, relays, associatedPipelineExecutor.getPipeline());
+
+ long duration = System.nanoTime() - nanoTimeBeforeOperation;
+ EvaluationLogger.getInstance().logMQTT("Migration", "start target element", "", duration, duration/1000000000.0);
+
+ return status;
+ }
+
+ @Override
+ public PipelineOperationStatus rollbackOperationFully() {
+ List<InvocableStreamPipesEntity> graphs =
+ Collections.singletonList(associatedPipelineExecutor.getMigrationEntity().getTargetElement());
+ List<SpDataStreamRelayContainer> relays = PipelineElementUtils.extractRelaysFromDataProcessor(graphs);
+
+ return new GraphSubmitter(associatedPipelineExecutor.getPipeline().getPipelineId(),
+ associatedPipelineExecutor.getPipeline().getName(), graphs, new ArrayList<>(), relays)
+ .detachPipelineElementsAndRelays();
+ }
+
+ @Override
+ public PipelineOperationStatus rollbackOperationPartially() {
+ return StatusUtils.initPipelineOperationStatus(associatedPipelineExecutor.getPipeline());
+ }
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StopOriginPipelineElementAndRelaysOperation.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StopOriginPipelineElementAndRelaysOperation.java
new file mode 100644
index 0000000..42f7c3d
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StopOriginPipelineElementAndRelaysOperation.java
@@ -0,0 +1,71 @@
+/*
+ * 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.execution.pipeline.executor.operations;
+
+import org.apache.streampipes.logging.evaluation.EvaluationLogger;
+import org.apache.streampipes.manager.execution.pipeline.executor.operations.types.MigrationOperation;
+import org.apache.streampipes.manager.execution.pipeline.executor.utils.CommunicationUtils;
+import org.apache.streampipes.manager.execution.pipeline.executor.utils.PipelineElementUtils;
+import org.apache.streampipes.manager.execution.pipeline.executor.PipelineExecutor;
+import org.apache.streampipes.manager.execution.pipeline.executor.utils.RelayUtils;
+import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
+import org.apache.streampipes.model.eventrelay.SpDataStreamRelayContainer;
+import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
+
+import java.util.Collections;
+import java.util.List;
+
+public class StopOriginPipelineElementAndRelaysOperation extends PipelineExecutionOperation implements MigrationOperation {
+
+ public StopOriginPipelineElementAndRelaysOperation(PipelineExecutor pipelineExecutor) {
+ super(pipelineExecutor);
+ }
+
+ @Override
+ public PipelineOperationStatus executeOperation() {
+ long nanoTimeBeforeOperation = System.nanoTime();
+ List<InvocableStreamPipesEntity> graphs = Collections.singletonList(
+ associatedPipelineExecutor.getMigrationEntity().getSourceElement());
+ List<SpDataStreamRelayContainer> relays = PipelineElementUtils.extractRelaysFromDataProcessor(graphs);
+
+ RelayUtils.updateRelays(relays, associatedPipelineExecutor.getRelaysToBeDeleted());
+
+ PipelineOperationStatus status = CommunicationUtils.stopPipelineElementsAndRelays(graphs, relays, associatedPipelineExecutor.getPipeline());
+
+ long duration = System.nanoTime() - nanoTimeBeforeOperation;
+ EvaluationLogger.getInstance().logMQTT("Migration", "stop origin element", "", duration, duration/1000000000.0);
+
+ return status;
+ }
+
+ @Override
+ public PipelineOperationStatus rollbackOperationPartially() {
+ //TODO: Implement
+ return null;
+ }
+
+ @Override
+ public PipelineOperationStatus rollbackOperationFully() {
+ List<InvocableStreamPipesEntity> graphs = Collections.singletonList(
+ associatedPipelineExecutor.getMigrationEntity().getSourceElement());
+ List<SpDataStreamRelayContainer> relays = PipelineElementUtils.extractRelaysFromDataProcessor(graphs);
+
+ RelayUtils.updateRelays(relays, associatedPipelineExecutor.getRelaysToBeDeleted());
+ return CommunicationUtils.startPipelineElementsAndRelays(graphs, relays, associatedPipelineExecutor.getPipeline());
+ }
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StopPipelineOperation.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StopPipelineOperation.java
new file mode 100644
index 0000000..2e8f367
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StopPipelineOperation.java
@@ -0,0 +1,79 @@
+/*
+ * 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.execution.pipeline.executor.operations;
+
+import org.apache.streampipes.manager.execution.http.GraphSubmitter;
+import org.apache.streampipes.manager.execution.pipeline.executor.PipelineExecutor;
+import org.apache.streampipes.manager.execution.pipeline.executor.utils.PipelineElementUtils;
+import org.apache.streampipes.manager.execution.pipeline.executor.utils.StatusUtils;
+import org.apache.streampipes.manager.execution.pipeline.executor.utils.StorageUtils;
+import org.apache.streampipes.manager.execution.status.PipelineStatusManager;
+import org.apache.streampipes.manager.util.TemporaryGraphStorage;
+import org.apache.streampipes.model.SpDataSet;
+import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
+import org.apache.streampipes.model.eventrelay.SpDataStreamRelayContainer;
+import org.apache.streampipes.model.message.PipelineStatusMessage;
+import org.apache.streampipes.model.message.PipelineStatusMessageType;
+import org.apache.streampipes.model.pipeline.Pipeline;
+import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
+
+import java.util.List;
+
+public class StopPipelineOperation extends PipelineExecutionOperation {
+
+ public StopPipelineOperation(PipelineExecutor pipelineExecutor) {
+ super(pipelineExecutor);
+ }
+
+ @Override
+ public PipelineOperationStatus executeOperation() {
+ Pipeline pipeline = associatedPipelineExecutor.getPipeline();
+ List<InvocableStreamPipesEntity> graphs = TemporaryGraphStorage.graphStorage.get(pipeline.getPipelineId());
+ List<SpDataSet> dataSets = TemporaryGraphStorage.datasetStorage.get(pipeline.getPipelineId());
+ List<SpDataStreamRelayContainer> relays = PipelineElementUtils.generateRelays(graphs, pipeline);
+
+ PipelineOperationStatus status = new GraphSubmitter(pipeline.getPipelineId(), pipeline.getName(), graphs,
+ dataSets, relays).detachPipelineElementsAndRelays();
+
+ if (status.isSuccess()) {
+ if (associatedPipelineExecutor.isMonitor()) StorageUtils.deleteVisualization(pipeline.getPipelineId());
+ if (associatedPipelineExecutor.isStoreStatus()) StorageUtils.setPipelineStopped(pipeline);
+
+ StorageUtils.deleteDataStreamRelayContainer(relays);
+
+ PipelineStatusManager.addPipelineStatus(pipeline.getPipelineId(),
+ new PipelineStatusMessage(pipeline.getPipelineId(),
+ System.currentTimeMillis(),
+ PipelineStatusMessageType.PIPELINE_STOPPED.title(),
+ PipelineStatusMessageType.PIPELINE_STOPPED.description()));
+ }
+ return status;
+ }
+
+ @Override
+ public PipelineOperationStatus rollbackOperationPartially() {
+ //TODO: Implement sth?
+ return StatusUtils.initPipelineOperationStatus(associatedPipelineExecutor.getPipeline());
+ }
+
+ @Override
+ public PipelineOperationStatus rollbackOperationFully() {
+ //TODO: Implement sth?
+ return StatusUtils.initPipelineOperationStatus(associatedPipelineExecutor.getPipeline());
+ }
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StopRelaysFromPredecessorOperation.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StopRelaysFromPredecessorOperation.java
new file mode 100644
index 0000000..c1ee4ad
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StopRelaysFromPredecessorOperation.java
@@ -0,0 +1,77 @@
+/*
+ * 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.execution.pipeline.executor.operations;
+
+import org.apache.streampipes.logging.evaluation.EvaluationLogger;
+import org.apache.streampipes.manager.execution.pipeline.executor.*;
+import org.apache.streampipes.manager.execution.pipeline.executor.operations.types.MigrationOperation;
+import org.apache.streampipes.manager.execution.pipeline.executor.utils.CommunicationUtils;
+import org.apache.streampipes.manager.execution.pipeline.executor.utils.PipelineElementUtils;
+import org.apache.streampipes.manager.execution.pipeline.executor.utils.RelayUtils;
+import org.apache.streampipes.manager.execution.pipeline.executor.utils.StatusUtils;
+import org.apache.streampipes.model.eventrelay.SpDataStreamRelayContainer;
+import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
+
+import java.util.List;
+import java.util.Set;
+
+public class StopRelaysFromPredecessorOperation extends PipelineExecutionOperation implements MigrationOperation {
+
+ public StopRelaysFromPredecessorOperation(PipelineExecutor pipelineExecutor){
+ super(pipelineExecutor);
+ }
+
+ @Override
+ public PipelineOperationStatus executeOperation() {
+ long nanoTimeBeforeOperation = System.nanoTime();
+ List<SpDataStreamRelayContainer> relays = RelayUtils
+ .findRelaysWhenStopping(associatedPipelineExecutor.getPredecessorsBeforeMigration(),
+ associatedPipelineExecutor.getMigrationEntity().getSourceElement(),
+ associatedPipelineExecutor.getPipeline());
+
+ RelayUtils.updateRelays(relays, associatedPipelineExecutor.getRelaysToBeDeleted());
+
+ PipelineOperationStatus statusStopRelays =
+ CommunicationUtils.stopRelays(relays, associatedPipelineExecutor.getPipeline());
+
+ long duration = System.nanoTime() - nanoTimeBeforeOperation;
+ EvaluationLogger.getInstance().logMQTT("Migration", "stop relay from origin", "", duration, duration/1000000000.0);
+
+ return statusStopRelays;
+ }
+
+ @Override
+ public PipelineOperationStatus rollbackOperationPartially() {
+ Set<String> relayIdsToRollback = StatusUtils.extractUniqueRelayIds(this.getStatus());
+ List<SpDataStreamRelayContainer> rollbackRelays = PipelineElementUtils.findRelaysAndFilterById(relayIdsToRollback,
+ associatedPipelineExecutor.getPredecessorsBeforeMigration(),
+ associatedPipelineExecutor.getMigrationEntity().getSourceElement(),
+ associatedPipelineExecutor.getPipeline());
+
+ return CommunicationUtils.startRelays(rollbackRelays, associatedPipelineExecutor.getPipeline());
+ }
+
+ @Override
+ public PipelineOperationStatus rollbackOperationFully() {
+ List<SpDataStreamRelayContainer> rollbackRelays = RelayUtils.findRelays(
+ associatedPipelineExecutor.getPredecessorsBeforeMigration(),
+ associatedPipelineExecutor.getMigrationEntity().getSourceElement(),
+ associatedPipelineExecutor.getPipeline());
+ return CommunicationUtils.startRelays(rollbackRelays, associatedPipelineExecutor.getPipeline());
+ }
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StoreMigratedPipelineOperation.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StoreMigratedPipelineOperation.java
new file mode 100644
index 0000000..e691e9d
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StoreMigratedPipelineOperation.java
@@ -0,0 +1,61 @@
+/*
+ * 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.execution.pipeline.executor.operations;
+
+import org.apache.streampipes.manager.execution.pipeline.executor.PipelineExecutor;
+import org.apache.streampipes.manager.execution.pipeline.executor.utils.PipelineUtils;
+import org.apache.streampipes.manager.execution.pipeline.executor.utils.StatusUtils;
+import org.apache.streampipes.manager.execution.pipeline.executor.utils.StorageUtils;
+import org.apache.streampipes.model.SpDataSet;
+import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
+import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class StoreMigratedPipelineOperation extends PipelineExecutionOperation{
+
+ public StoreMigratedPipelineOperation(PipelineExecutor pipelineExecutor) {
+ super(pipelineExecutor);
+ }
+
+ @Override
+ public PipelineOperationStatus executeOperation() {
+ List<InvocableStreamPipesEntity> graphs = new ArrayList<>();
+ graphs.addAll(associatedPipelineExecutor.getPipeline().getActions());
+ graphs.addAll(associatedPipelineExecutor.getPipeline().getSepas());
+
+ List<SpDataSet> dataSets = PipelineUtils.findDataSets(associatedPipelineExecutor.getPipeline());
+
+ // store new pipeline and relays
+ StorageUtils.storeInvocationGraphs(associatedPipelineExecutor.getPipeline().getPipelineId(), graphs, dataSets);
+ StorageUtils.deleteDataStreamRelayContainer(associatedPipelineExecutor.getRelaysToBeDeleted());
+ StorageUtils.storeDataStreamRelayContainer(associatedPipelineExecutor.getRelaysToBePersisted());
+ return StatusUtils.initPipelineOperationStatus(associatedPipelineExecutor.getPipeline());
+ }
+
+ @Override
+ public PipelineOperationStatus rollbackOperationPartially() {
+ return null;
+ }
+
+ @Override
+ public PipelineOperationStatus rollbackOperationFully() {
+ return null;
+ }
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StorePipelineOperation.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StorePipelineOperation.java
new file mode 100644
index 0000000..b0f47bb
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/StorePipelineOperation.java
@@ -0,0 +1,69 @@
+/*
+ * 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.execution.pipeline.executor.operations;
+
+import org.apache.streampipes.manager.execution.pipeline.executor.PipelineExecutor;
+import org.apache.streampipes.manager.execution.pipeline.executor.utils.StatusUtils;
+import org.apache.streampipes.manager.execution.pipeline.executor.utils.StorageUtils;
+import org.apache.streampipes.manager.execution.status.PipelineStatusManager;
+import org.apache.streampipes.model.message.PipelineStatusMessage;
+import org.apache.streampipes.model.message.PipelineStatusMessageType;
+import org.apache.streampipes.model.pipeline.Pipeline;
+import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
+
+public class StorePipelineOperation extends PipelineExecutionOperation{
+
+ public StorePipelineOperation(PipelineExecutor pipelineExecutor) {
+ super(pipelineExecutor);
+ }
+
+ @Override
+ public PipelineOperationStatus executeOperation() {
+ PipelineOperationStatus status = StatusUtils.initPipelineOperationStatus(associatedPipelineExecutor.getPipeline());
+ Pipeline pipeline = associatedPipelineExecutor.getPipeline();
+ try{
+ StorageUtils.storeInvocationGraphs(pipeline.getPipelineId(), associatedPipelineExecutor.getGraphs(),
+ associatedPipelineExecutor.getDataSets());
+ StorageUtils.storeDataStreamRelayContainer(associatedPipelineExecutor.getRelays());
+
+ PipelineStatusManager.addPipelineStatus(
+ pipeline.getPipelineId(),
+ new PipelineStatusMessage(pipeline.getPipelineId(),
+ System.currentTimeMillis(),
+ PipelineStatusMessageType.PIPELINE_STARTED.title(),
+ PipelineStatusMessageType.PIPELINE_STARTED.description()));
+
+ if (associatedPipelineExecutor.isStoreStatus()) StorageUtils.setPipelineStarted(pipeline);
+ }catch (Exception e){
+ status.setSuccess(false);
+ status.setTitle(e.getMessage());
+ }
+
+ return status;
+ }
+
+ @Override
+ public PipelineOperationStatus rollbackOperationPartially() {
+ return null;
+ }
+
+ @Override
+ public PipelineOperationStatus rollbackOperationFully() {
+ return null;
+ }
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/types/MigrationOperation.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/types/MigrationOperation.java
new file mode 100644
index 0000000..22c21d1
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/types/MigrationOperation.java
@@ -0,0 +1,21 @@
+/*
+ * 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.execution.pipeline.executor.operations.types;
+
+public interface MigrationOperation {
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/types/ReconfigurationOperation.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/types/ReconfigurationOperation.java
new file mode 100644
index 0000000..13c67a0
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/operations/types/ReconfigurationOperation.java
@@ -0,0 +1,21 @@
+/*
+ * 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.execution.pipeline.executor.operations.types;
+
+public interface ReconfigurationOperation {
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/CommunicationUtils.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/CommunicationUtils.java
new file mode 100644
index 0000000..a066b22
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/CommunicationUtils.java
@@ -0,0 +1,71 @@
+/*
+ * 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.execution.pipeline.executor.utils;
+
+import org.apache.streampipes.manager.execution.http.GraphSubmitter;
+import org.apache.streampipes.manager.execution.http.StateSubmitter;
+import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
+import org.apache.streampipes.model.eventrelay.SpDataStreamRelayContainer;
+import org.apache.streampipes.model.pipeline.Pipeline;
+import org.apache.streampipes.model.pipeline.PipelineElementStatus;
+import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class CommunicationUtils {
+
+ public static PipelineOperationStatus startPipelineElementsAndRelays(List<InvocableStreamPipesEntity> graphs,
+ List<SpDataStreamRelayContainer> relays,
+ Pipeline pipeline){
+ if (graphs.isEmpty()) {
+ return StatusUtils.initPipelineOperationStatus(pipeline);
+ }
+ return new GraphSubmitter(pipeline.getPipelineId(), pipeline.getName(),
+ graphs, new ArrayList<>(), relays).invokePipelineElementsAndRelays();
+ }
+
+ public static PipelineOperationStatus stopPipelineElementsAndRelays(List<InvocableStreamPipesEntity> graphs,
+ List<SpDataStreamRelayContainer> relays,
+ Pipeline pipeline){
+ if (graphs.isEmpty()) {
+ return StatusUtils.initPipelineOperationStatus(pipeline);
+ }
+ return new GraphSubmitter(pipeline.getPipelineId(), pipeline.getName(),
+ graphs, new ArrayList<>(),relays).detachPipelineElementsAndRelays();
+ }
+
+ public static PipelineOperationStatus startRelays(List<SpDataStreamRelayContainer> relays, Pipeline pipeline){
+ return new GraphSubmitter(pipeline.getPipelineId(), pipeline.getName(), new ArrayList<>(), new ArrayList<>(),
+ relays).invokeRelaysOnMigration();
+ }
+
+ public static PipelineOperationStatus stopRelays(List<SpDataStreamRelayContainer> relays, Pipeline pipeline){
+ return new GraphSubmitter(pipeline.getPipelineId(), pipeline.getName(),new ArrayList<>(), new ArrayList<>(),
+ relays).detachRelaysOnMigration();
+ }
+
+ public static PipelineElementStatus getState(InvocableStreamPipesEntity graph, Pipeline pipeline){
+ return new StateSubmitter(pipeline.getPipelineId(), pipeline.getName(), graph, null).getElementState();
+ }
+
+ public static PipelineElementStatus setState(InvocableStreamPipesEntity graph, String state, Pipeline pipeline){
+ return new StateSubmitter(pipeline.getPipelineId(), pipeline.getName(), graph, state).setElementState();
+ }
+
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/PipelineElementUtils.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/PipelineElementUtils.java
new file mode 100644
index 0000000..58f731c
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/PipelineElementUtils.java
@@ -0,0 +1,196 @@
+/*
+ * 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.execution.pipeline.executor.utils;
+
+import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import org.apache.streampipes.model.SpDataStream;
+import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
+import org.apache.streampipes.model.base.NamedStreamPipesEntity;
+import org.apache.streampipes.model.eventrelay.SpDataStreamRelay;
+import org.apache.streampipes.model.eventrelay.SpDataStreamRelayContainer;
+import org.apache.streampipes.model.graph.DataProcessorInvocation;
+import org.apache.streampipes.model.graph.DataSinkInvocation;
+import org.apache.streampipes.model.grounding.EventGrounding;
+import org.apache.streampipes.model.pipeline.Pipeline;
+import org.apache.streampipes.model.staticproperty.SecretStaticProperty;
+import org.apache.streampipes.user.management.encryption.CredentialsManager;
+
+import java.security.GeneralSecurityException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class PipelineElementUtils {
+
+ public static List<SpDataStreamRelayContainer> extractRelaysFromDataProcessor(List<InvocableStreamPipesEntity> graphs) {
+ return graphs.stream()
+ .map(DataProcessorInvocation.class::cast)
+ .map(SpDataStreamRelayContainer::new)
+ .collect(Collectors.toList());
+ }
+
+ /**
+ * Compare deployment targets of two pipeline elements, namely data stream/processor (source) and data
+ * processor/sink (target)
+ *
+ * @param e1
+ * @param e2
+ * @return boolean value that returns true if source and target share the same deployment target, else false
+ */
+ public static boolean differentDeploymentTargets(NamedStreamPipesEntity e1, InvocableStreamPipesEntity e2) {
+ if (e1 instanceof SpDataStream) {
+ return !((SpDataStream) e1).getDeploymentTargetNodeId().equals(e2.getDeploymentTargetNodeId());
+ } else if (e1 instanceof DataProcessorInvocation) {
+ return !((DataProcessorInvocation) e1).getDeploymentTargetNodeId().equals(e2.getDeploymentTargetNodeId());
+ } else if (e1 instanceof DataSinkInvocation) {
+ return !((DataSinkInvocation) e1).getDeploymentTargetNodeId().equals(e2.getDeploymentTargetNodeId());
+ }
+ throw new SpRuntimeException("Matching deployment targets check failed");
+ }
+
+ /**
+ * Extract topic name
+ *
+ * @param entity
+ * @return
+ */
+ public static String extractActualTopic(NamedStreamPipesEntity entity) {
+ if (entity instanceof SpDataStream) {
+ return ((SpDataStream) entity)
+ .getEventGrounding().getTransportProtocol().getTopicDefinition().getActualTopicName();
+ } else if (entity instanceof SpDataStreamRelay) {
+ return ((SpDataStreamRelay) entity)
+ .getEventGrounding().getTransportProtocol().getTopicDefinition().getActualTopicName();
+ }
+ throw new SpRuntimeException("Could not extract actual topic name from entity");
+ }
+
+ /**
+ * Get index of data processor/sink connection based on source DOM identifier
+ *
+ * @param sourceDomId source DOM identifier
+ * @param target data processor/sink
+ * @return Integer with index of connection, if invalid returns -1.
+ */
+ public static Integer getIndex(String sourceDomId, InvocableStreamPipesEntity target) {
+ return target.getConnectedTo().indexOf(sourceDomId);
+ }
+
+ public static List<SpDataStreamRelayContainer> findRelaysAndFilterById(Set<String> relayIdsToRollback,
+ List<NamedStreamPipesEntity> predecessor,
+ InvocableStreamPipesEntity target, Pipeline pipeline) {
+ return RelayUtils.findRelays(predecessor, target, pipeline).stream()
+ .filter(relay -> relayIdsToRollback.contains(relay.getRunningStreamRelayInstanceId()))
+ .collect(Collectors.toList());
+ }
+
+ // TODO: when using kafka as edge protocol it generates duplicate event relays -> check with mqtt as edge
+ // protocol and fix
+ public static List<SpDataStreamRelayContainer> generateDataStreamRelays(List<InvocableStreamPipesEntity> graphs,
+ Pipeline pipeline) {
+ List<SpDataStreamRelayContainer> relays = new ArrayList<>();
+
+ for (InvocableStreamPipesEntity graph : graphs) {
+ for (SpDataStream stream: pipeline.getStreams()) {
+ if (differentDeploymentTargets(stream, graph) && connected(stream, graph)) {
+
+ List<SpDataStreamRelay> dataStreamRelays = new ArrayList<>();
+ dataStreamRelays.add(new SpDataStreamRelay(new EventGrounding(graph.getInputStreams()
+ .get(getIndex(stream.getDOM(), graph))
+ .getEventGrounding())));
+
+ String id = RelayUtils.extractUniqueAdpaterId(stream.getElementId());
+ String relayStrategy = pipeline.getEventRelayStrategy();
+
+ if (!RelayUtils.relayExists(relays, id)) {
+ relays.add(new SpDataStreamRelayContainer(id, relayStrategy, stream, dataStreamRelays));
+ }
+ }
+ }
+ for (DataProcessorInvocation processor : pipeline.getSepas()) {
+ if (differentDeploymentTargets(processor, graph) && connected(processor, graph)) {
+ if (!RelayUtils.relayExists(relays, processor.getDeploymentRunningInstanceId())) {
+// String previousId = processor.getDeploymentRunningInstanceId();
+// String modifiedId = previousId + "-" + processor.getDeploymentTargetNodeId();
+// processor.setDeploymentRunningInstanceId(modifiedId);
+ SpDataStreamRelayContainer processorRelay = new SpDataStreamRelayContainer(processor);
+ relays.add(processorRelay);
+ }
+ }
+ }
+ }
+ return relays;
+ }
+
+ /**
+ * Compare connection of two pipeline elements, namely data stream/processor (source) and data processor/sink
+ * (target) by DOM identifier.
+ *
+ * @param source data stream or data processor
+ * @param target data processor/sink
+ * @return boolean value that returns true if source and target are connected, else false
+ */
+ private static boolean connected(NamedStreamPipesEntity source, InvocableStreamPipesEntity target) {
+ int index = getIndex(source.getDOM(), target);
+ if (index != -1) {
+ return target.getConnectedTo().get(index).equals(source.getDOM());
+ }
+ return false;
+ }
+
+ /**
+ * Decrypt potential secrets contained in static properties, e.g., passwords
+ *
+ * @param graphs List of graphs
+ * @return List of decrypted graphs
+ */
+ public static List<InvocableStreamPipesEntity> decryptSecrets(List<InvocableStreamPipesEntity> graphs, Pipeline pipeline) {
+ List<InvocableStreamPipesEntity> decryptedGraphs = new ArrayList<>();
+ graphs.stream().map(g -> {
+ if (g instanceof DataProcessorInvocation) {
+ return new DataProcessorInvocation((DataProcessorInvocation) g);
+ } else {
+ return new DataSinkInvocation((DataSinkInvocation) g);
+ }
+ }).forEach(g -> {
+ g.getStaticProperties()
+ .stream()
+ .filter(SecretStaticProperty.class::isInstance)
+ .forEach(sp -> {
+ try {
+ String decrypted = CredentialsManager.decrypt(pipeline.getCreatedByUser(),
+ ((SecretStaticProperty) sp).getValue());
+ ((SecretStaticProperty) sp).setValue(decrypted);
+ ((SecretStaticProperty) sp).setEncrypted(false);
+ } catch (GeneralSecurityException e) {
+ e.printStackTrace();
+ }
+ });
+ decryptedGraphs.add(g);
+ });
+ return decryptedGraphs;
+ }
+
+ public static List<SpDataStreamRelayContainer> generateRelays(List<InvocableStreamPipesEntity> graphs,
+ Pipeline pipeline) {
+ return generateDataStreamRelays(graphs, pipeline).stream()
+ .filter(r -> r.getOutputStreamRelays().size() > 0)
+ .collect(Collectors.toList());
+ }
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/PipelineUtils.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/PipelineUtils.java
new file mode 100644
index 0000000..47d41b2
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/PipelineUtils.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.manager.execution.pipeline.executor.utils;
+
+import org.apache.streampipes.manager.data.PipelineGraph;
+import org.apache.streampipes.manager.data.PipelineGraphHelpers;
+import org.apache.streampipes.manager.matching.InvocationGraphBuilder;
+import org.apache.streampipes.model.SpDataSet;
+import org.apache.streampipes.model.SpDataStream;
+import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
+import org.apache.streampipes.model.base.NamedStreamPipesEntity;
+import org.apache.streampipes.model.pipeline.Pipeline;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
+
+public class PipelineUtils {
+
+ public static List<SpDataSet> findDataSets(Pipeline pipeline) {
+ return pipeline.getStreams().stream()
+ .filter(s -> s instanceof SpDataSet)
+ .map(s -> new SpDataSet((SpDataSet) s))
+ .collect(Collectors.toList());
+ }
+
+ public static List<NamedStreamPipesEntity> getPredecessors(NamedStreamPipesEntity source,
+ InvocableStreamPipesEntity target,
+ PipelineGraph pipelineGraph,
+ List<NamedStreamPipesEntity> foundPredecessors){
+
+ Set<InvocableStreamPipesEntity> targets = getTargetsAsSet(source, pipelineGraph,
+ InvocableStreamPipesEntity.class);
+
+ //TODO: Check if this works for all graph topologies
+ if (targets.contains(target)){
+ foundPredecessors.add(source);
+ } else {
+ List<NamedStreamPipesEntity> successors = getTargetsAsList(source, pipelineGraph,
+ NamedStreamPipesEntity.class);
+
+ if (successors.isEmpty()) return foundPredecessors;
+ successors.forEach(successor -> getPredecessors(successor, target, pipelineGraph, foundPredecessors));
+ }
+ return foundPredecessors;
+ }
+
+ private static <T> List<T> getTargetsAsList(NamedStreamPipesEntity source, PipelineGraph pipelineGraph,
+ Class<T> clazz){
+ return new ArrayList<>(getTargetsAsSet(source, pipelineGraph, clazz));
+ }
+
+ private static <T> Set<T> getTargetsAsSet(NamedStreamPipesEntity source, PipelineGraph pipelineGraph,
+ Class<T> clazz){
+ return pipelineGraph.outgoingEdgesOf(source)
+ .stream()
+ .map(pipelineGraph::getEdgeTarget)
+ .map(clazz::cast)
+ .collect(Collectors.toSet());
+ }
+
+ public static NamedStreamPipesEntity findMatching(NamedStreamPipesEntity entity, PipelineGraph pipelineGraph){
+ AtomicReference<NamedStreamPipesEntity> match = new AtomicReference<>();
+ List<SpDataStream> dataStreams = PipelineGraphHelpers.findStreams(pipelineGraph);
+
+ for (SpDataStream stream : dataStreams) {
+ NamedStreamPipesEntity foundEntity = compareGraphs(stream, entity, pipelineGraph, new ArrayList<>());
+ if (foundEntity != null) {
+ match.set(foundEntity);
+ }
+ }
+ return match.get();
+ }
+
+ private static NamedStreamPipesEntity compareGraphs(NamedStreamPipesEntity source,
+ NamedStreamPipesEntity searchedEntity,
+ PipelineGraph pipelineGraph,
+ List<NamedStreamPipesEntity> successors){
+ if(matchingDOM(source, searchedEntity)) {
+ return source;
+ } else if (successors.isEmpty()) {
+ successors = getTargetsAsList(source, pipelineGraph, NamedStreamPipesEntity.class);
+ Optional<NamedStreamPipesEntity> successor = successors.stream().findFirst();
+ if (successor.isPresent()) {
+ successors.remove(successor.get());
+ return compareGraphs(successor.get(), searchedEntity, pipelineGraph, successors);
+ }
+ }
+ return null;
+ }
+
+ /**
+ * Checks if DOM are equal
+ *
+ * @param source pipeline element
+ * @param target pipeline element
+ * @return true if DOM is the same, else false
+ */
+ private static boolean matchingDOM(NamedStreamPipesEntity source, NamedStreamPipesEntity target) {
+ return source.getDOM().equals(target.getDOM());
+ }
+
+ public static void purgeExistingRelays(Pipeline pipeline) {
+ pipeline.getSepas().forEach(s -> s.setOutputStreamRelays(new ArrayList<>()));
+ }
+
+ public static void buildPipelineGraph(PipelineGraph pipelineGraphAfterMigration, String pipelineId) {
+ new InvocationGraphBuilder(pipelineGraphAfterMigration,
+ pipelineId).buildGraphs();
+ }
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/RelayUtils.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/RelayUtils.java
new file mode 100644
index 0000000..e88707e
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/RelayUtils.java
@@ -0,0 +1,236 @@
+/*
+ * 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.execution.pipeline.executor.utils;
+
+import org.apache.streampipes.model.SpDataStream;
+import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
+import org.apache.streampipes.model.base.NamedStreamPipesEntity;
+import org.apache.streampipes.model.eventrelay.SpDataStreamRelay;
+import org.apache.streampipes.model.eventrelay.SpDataStreamRelayContainer;
+import org.apache.streampipes.model.graph.DataProcessorInvocation;
+import org.apache.streampipes.model.graph.DataSinkInvocation;
+import org.apache.streampipes.model.grounding.EventGrounding;
+import org.apache.streampipes.model.grounding.TransportProtocol;
+import org.apache.streampipes.model.pipeline.Pipeline;
+import org.apache.streampipes.storage.management.StorageDispatcher;
+
+import java.util.*;
+import java.util.stream.Collectors;
+
+public class RelayUtils {
+
+ public static void updateRelays(List<SpDataStreamRelayContainer> relays,
+ List<SpDataStreamRelayContainer> relaysToBeUpdated) {
+ relays.stream()
+ .filter(r -> r.getOutputStreamRelays().size() > 0)
+ .forEach(relaysToBeUpdated::add);
+ }
+
+ public static List<SpDataStreamRelayContainer> findRelays(List<NamedStreamPipesEntity> predecessors,
+ InvocableStreamPipesEntity target, Pipeline pipeline){
+
+ List<SpDataStreamRelayContainer> relays = new ArrayList<>();
+
+ predecessors.forEach(pred -> {
+ List<SpDataStreamRelay> dataStreamRelays = new ArrayList<>();
+
+ if (pred instanceof DataProcessorInvocation){
+ //Data Processor
+ DataProcessorInvocation graph = (DataProcessorInvocation) pred;
+ if (PipelineElementUtils.differentDeploymentTargets(pred, target)) {
+
+ String runningRelayId = ((DataProcessorInvocation) pred).getDeploymentRunningInstanceId();
+ Optional<SpDataStreamRelayContainer> existingRelay = getRelayContainerById(runningRelayId);
+
+ // only add relay if not existing - prevent from duplicate relays with same topic to same target
+ Collection<? extends SpDataStreamRelay> foundRelays = findRelaysWithMatchingTopic(graph, target);
+
+ if (!existingRelay.isPresent() || missingRelayToTarget(existingRelay.get(), foundRelays)) {
+ dataStreamRelays.addAll(findRelaysWithMatchingTopic(graph, target));
+
+ //dsRelayContainer.setRunningStreamRelayInstanceId(pipeline.getPipelineId());
+ SpDataStreamRelayContainer relayContainer = new SpDataStreamRelayContainer(graph);
+ relayContainer.setOutputStreamRelays(dataStreamRelays);
+
+ relays.add(relayContainer);
+ }
+
+ }
+ } else if (pred instanceof SpDataStream){
+ //DataStream
+ SpDataStream stream = (SpDataStream) pred;
+ if (PipelineElementUtils.differentDeploymentTargets(stream, target)){
+
+ String id = extractUniqueAdpaterId(stream.getElementId());
+ Optional<SpDataStreamRelayContainer> existingRelay = getRelayContainerById(id);
+
+ // only add relay if not existing - prevent from duplicate relays with same topic
+ if(!existingRelay.isPresent()) {
+ //There is a relay that needs to be removed
+ dataStreamRelays.add(new SpDataStreamRelay(new EventGrounding(target.getInputStreams()
+ .get(PipelineElementUtils.getIndex(pred.getDOM(), target))
+ .getEventGrounding())));
+ String relayStrategy = pipeline.getEventRelayStrategy();
+ relays.add(new SpDataStreamRelayContainer(id, relayStrategy, stream, dataStreamRelays));
+ } else {
+ // generate relays for adapter streams to remote processors
+ List<SpDataStreamRelayContainer> generatedRelays =
+ PipelineElementUtils.generateDataStreamRelays(Collections.singletonList(target), pipeline);
+
+ relays.addAll(generatedRelays);
+ }
+ }
+ }
+ });
+ return relays;
+ }
+
+ public static List<SpDataStreamRelayContainer> findRelaysWhenStopping(List<NamedStreamPipesEntity> predecessors,
+ InvocableStreamPipesEntity target, Pipeline pipeline){
+
+ List<SpDataStreamRelayContainer> relays = new ArrayList<>();
+
+ predecessors.forEach(pred -> {
+ List<SpDataStreamRelay> dataStreamRelays = new ArrayList<>();
+
+ if (pred instanceof DataProcessorInvocation){
+ //Data Processor
+ DataProcessorInvocation graph = (DataProcessorInvocation) pred;
+ if (PipelineElementUtils.differentDeploymentTargets(pred, target)) {
+
+ // TODO only add if no other processor or sink depends on relay
+ String predDOMId = pred.getDOM();
+ String targetRunningInstanceId = target.getDeploymentRunningInstanceId();
+ Optional<DataProcessorInvocation> foundProcessor = pipeline.getSepas().stream()
+ .filter(processor -> processor.getConnectedTo().contains(predDOMId))
+ .filter(processor -> !processor.getDeploymentRunningInstanceId().equals(targetRunningInstanceId))
+ .findAny();
+
+ Optional<DataSinkInvocation> foundSink = pipeline.getActions().stream()
+ .filter(action -> action.getConnectedTo().contains(predDOMId))
+ .findAny();
+
+ boolean foundDependencyOnDifferentTarget = false;
+ if (foundProcessor.isPresent()) {
+ foundDependencyOnDifferentTarget = PipelineElementUtils.differentDeploymentTargets(foundProcessor.get(), target);
+ }
+
+ if (foundSink.isPresent()) {
+ foundDependencyOnDifferentTarget = PipelineElementUtils.differentDeploymentTargets(foundSink.get(), target);
+ }
+
+ if (foundDependencyOnDifferentTarget) {
+ dataStreamRelays.addAll(findRelaysWithMatchingTopic(graph, target));
+
+ SpDataStreamRelayContainer relayContainer = new SpDataStreamRelayContainer(graph);
+ relayContainer.setOutputStreamRelays(dataStreamRelays);
+
+ relays.add(relayContainer);
+ }
+
+ }
+ } else if (pred instanceof SpDataStream){
+ //DataStream
+ SpDataStream stream = (SpDataStream) pred;
+ if (PipelineElementUtils.differentDeploymentTargets(stream, target)){
+
+ String id = extractUniqueAdpaterId(stream.getElementId());
+ //There is a relay that needs to be removed
+ dataStreamRelays.add(new SpDataStreamRelay(new EventGrounding(target.getInputStreams()
+ .get(PipelineElementUtils.getIndex(pred.getDOM(), target))
+ .getEventGrounding())));
+ String relayStrategy = pipeline.getEventRelayStrategy();
+ relays.add(new SpDataStreamRelayContainer(id, relayStrategy, stream, dataStreamRelays));
+ }
+ }
+ });
+ return relays;
+ }
+
+ /**
+ * Find relays with matching topics
+ *
+ * @param graph data processor
+ * @param target data processor/sink
+ * @return collection of data stream relays
+ */
+ public static Collection<? extends SpDataStreamRelay> findRelaysWithMatchingTopic(DataProcessorInvocation graph,
+ InvocableStreamPipesEntity target) {
+ return graph.getOutputStreamRelays().stream().
+ filter(relay ->
+ target.getInputStreams().stream()
+ .map(PipelineElementUtils::extractActualTopic)
+ .collect(Collectors.toSet())
+ .contains(PipelineElementUtils.extractActualTopic(relay)))
+ .collect(Collectors.toList());
+ }
+
+ public static String extractUniqueAdpaterId(String s) {
+ return s.substring(s.lastIndexOf("/") + 1);
+ }
+
+ /**
+ *
+ * @param id
+ * @return
+ */
+ private static Optional<SpDataStreamRelayContainer> getRelayContainerById(String id) {
+ return StorageDispatcher.INSTANCE.getNoSqlStore().getNodeDataStreamRelayStorage().getRelayContainerById(id);
+ }
+
+ private static boolean missingRelayToTarget(SpDataStreamRelayContainer existingRelayContainer,
+ Collection<? extends SpDataStreamRelay> foundRelays) {
+
+ List<TransportProtocol> set = foundRelays.stream()
+ .map(SpDataStreamRelay::getEventGrounding)
+ .map(EventGrounding::getTransportProtocol)
+ .collect(Collectors.toList());
+
+ List<TransportProtocol> relay = existingRelayContainer.getOutputStreamRelays().stream()
+ .map(SpDataStreamRelay::getEventGrounding)
+ .map(EventGrounding::getTransportProtocol)
+ .collect(Collectors.toList());
+
+ for (TransportProtocol tp: set) {
+ for (TransportProtocol r: relay) {
+ String targetTopic = tp.getTopicDefinition().getActualTopicName();
+ String rTopic = r.getTopicDefinition().getActualTopicName();
+ String targetHost = tp.getBrokerHostname();
+ String rHost = r.getBrokerHostname();
+
+ if (targetHost.equals(rHost) && targetTopic.equals(rTopic)) {
+ return false;
+ }
+ }
+ }
+
+ return true;
+ }
+
+ /**
+ * Check if relay with deploymentRunningInstanceId of predecessor already exists
+ *
+ * @param relays List of existing relays
+ * @param deploymentRunningInstanceId Id to check
+ * @return boolean
+ */
+ public static boolean relayExists(List<SpDataStreamRelayContainer> relays,
+ String deploymentRunningInstanceId) {
+ return relays.stream().anyMatch(r -> r.getRunningStreamRelayInstanceId().equals(deploymentRunningInstanceId));
+ }
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/StatusUtils.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/StatusUtils.java
new file mode 100644
index 0000000..713045b
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/StatusUtils.java
@@ -0,0 +1,61 @@
+/*
+ * 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.execution.pipeline.executor.utils;
+
+import org.apache.streampipes.model.pipeline.Pipeline;
+import org.apache.streampipes.model.pipeline.PipelineElementStatus;
+import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
+
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class StatusUtils {
+
+ /**
+ * Create pipeline operation status with pipeline id and name and set success to true
+ *
+ * @return PipelineOperationStatus
+ */
+ public static PipelineOperationStatus initPipelineOperationStatus(Pipeline pipeline) {
+ PipelineOperationStatus status = new PipelineOperationStatus();
+ status.setPipelineId(pipeline.getPipelineId());
+ status.setPipelineName(pipeline.getName());
+ status.setSuccess(true);
+ return status;
+ }
+
+ public static void updateStatus(PipelineOperationStatus partialStatus, PipelineOperationStatus status) {
+ // Add status to global migration status
+ partialStatus.getElementStatus().forEach(status::addPipelineElementStatus);
+ String title = partialStatus.getTitle();
+ if(title != null)
+ status.setTitle(partialStatus.getTitle());
+ }
+
+ public static Set<String> extractUniqueRelayIds(PipelineOperationStatus status) {
+ return status.getElementStatus().stream()
+ .filter(PipelineElementStatus::isSuccess)
+ .map(PipelineElementStatus::getElementId)
+ .collect(Collectors.toSet());
+ }
+
+ public static void checkSuccess(PipelineOperationStatus status){
+ if(status.isSuccess())
+ status.setSuccess(status.getElementStatus().stream().allMatch(PipelineElementStatus::isSuccess));
+ }
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/StorageUtils.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/StorageUtils.java
new file mode 100644
index 0000000..ec74238
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/executor/utils/StorageUtils.java
@@ -0,0 +1,94 @@
+/*
+ * 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.execution.pipeline.executor.utils;
+
+import org.apache.streampipes.manager.util.TemporaryGraphStorage;
+import org.apache.streampipes.model.SpDataSet;
+import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
+import org.apache.streampipes.model.eventrelay.SpDataStreamRelayContainer;
+import org.apache.streampipes.model.pipeline.Pipeline;
+import org.apache.streampipes.storage.api.INodeDataStreamRelay;
+import org.apache.streampipes.storage.api.IPipelineStorage;
+import org.apache.streampipes.storage.management.StorageDispatcher;
+
+import java.util.Date;
+import java.util.List;
+
+public class StorageUtils {
+
+ public static void setPipelineStarted(Pipeline pipeline) {
+ pipeline.setRunning(true);
+ pipeline.setStartedAt(new Date().getTime());
+ getPipelineStorageApi().updatePipeline(pipeline);
+ }
+
+ public static void setPipelineStopped(Pipeline pipeline) {
+ pipeline.setRunning(false);
+ getPipelineStorageApi().updatePipeline(pipeline);
+ }
+
+ public static void deleteVisualization(String pipelineId) {
+ StorageDispatcher.INSTANCE
+ .getNoSqlStore()
+ .getVisualizationStorageApi()
+ .deleteVisualization(pipelineId);
+ }
+
+ public static void storeInvocationGraphs(String pipelineId, List<InvocableStreamPipesEntity> graphs,
+ List<SpDataSet> dataSets) {
+ TemporaryGraphStorage.graphStorage.put(pipelineId, graphs);
+ TemporaryGraphStorage.datasetStorage.put(pipelineId, dataSets);
+ }
+
+ public static void storeDataStreamRelayContainer(List<SpDataStreamRelayContainer> relays) {
+ //relays.forEach(StreamPipesClusterManager::persistDataStreamRelay);
+ relays.forEach(relay -> getDataStreamRelayApi().addRelayContainer(relay));
+ }
+
+ public static void deleteDataStreamRelayContainer(List<SpDataStreamRelayContainer> relays) {
+ //relays.forEach(StreamPipesClusterManager::deleteDataStreamRelay);
+ relays.forEach(relay -> getDataStreamRelayApi().deleteRelayContainer(relay));
+ }
+
+ public static void updateDataStreamRelayContainer(List<SpDataStreamRelayContainer> relays) {
+ //relays.forEach(StreamPipesClusterManager::updateDataStreamRelay);
+ relays.forEach(relay -> getDataStreamRelayApi().updateRelayContainer(relay));
+ }
+
+
+
+ /**
+ * Get data stream relay storage dispatcher API
+ *
+ * @return INodeDataStreamRelay NoSQL storage interface for data stream relays
+ */
+ private static INodeDataStreamRelay getDataStreamRelayApi() {
+ return StorageDispatcher.INSTANCE.getNoSqlStore().getNodeDataStreamRelayStorage();
+ }
+
+ /**
+ * Get pipeline storage dispatcher API
+ *
+ * @return IPipelineStorage NoSQL storage interface for pipelines
+ */
+ private static IPipelineStorage getPipelineStorageApi() {
+ return StorageDispatcher.INSTANCE.getNoSqlStore().getPipelineStorageAPI();
+ }
+
+
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/PipelineElementMigrationHandler.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/PipelineElementMigrationHandler.java
index e16e2fd..bf12bb9 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/PipelineElementMigrationHandler.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/PipelineElementMigrationHandler.java
@@ -20,7 +20,8 @@ package org.apache.streampipes.manager.migration;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.streampipes.commons.exceptions.SpRuntimeException;
-import org.apache.streampipes.manager.execution.pipeline.PipelineMigrationExecutor;
+import org.apache.streampipes.manager.execution.pipeline.executor.PipelineExecutor;
+import org.apache.streampipes.manager.execution.pipeline.executor.PipelineExecutorFactory;
import org.apache.streampipes.manager.operations.Operations;
import org.apache.streampipes.model.graph.DataProcessorInvocation;
import org.apache.streampipes.model.pipeline.Pipeline;
@@ -86,13 +87,16 @@ public class PipelineElementMigrationHandler {
swapPipelineElement(migrationPipeline, failedEntity);
}
});
-
+ //Why overwrite when changes are rolled back? TODO: Check if needed when pipeline is rolled back (else return)
Operations.overwritePipeline(migrationPipeline);
}
private PipelineOperationStatus migratePipelineElement(PipelineElementMigrationEntity migrationEntity) {
- return new PipelineMigrationExecutor(migrationPipeline, currentPipeline, migrationEntity, visualize,
- storeStatus, monitor).migratePipelineElement();
+ PipelineExecutor migrationExecutor = PipelineExecutorFactory.
+ createMigrationExecutor(migrationPipeline, visualize, storeStatus, monitor, currentPipeline, migrationEntity);
+ return migrationExecutor.execute();
+ //return new PipelineMigrationExecutor(migrationPipeline, currentPipeline, migrationEntity, visualize,
+ // storeStatus, monitor).migratePipelineElement();
}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/operations/Operations.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/operations/Operations.java
index ec8e337..91162c5 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/operations/Operations.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/operations/Operations.java
@@ -22,11 +22,10 @@ import org.apache.streampipes.commons.exceptions.NoSuitableSepasAvailableExcepti
import org.apache.streampipes.commons.exceptions.SepaParseException;
import org.apache.streampipes.commons.exceptions.SpRuntimeException;
import org.apache.streampipes.manager.endpoint.EndpointItemFetcher;
-import org.apache.streampipes.manager.execution.pipeline.PipelineExecutor;
import org.apache.streampipes.manager.execution.pipeline.PipelineStorageService;
+import org.apache.streampipes.manager.execution.pipeline.executor.PipelineExecutorFactory;
import org.apache.streampipes.manager.matching.DataSetGroundingSelector;
import org.apache.streampipes.manager.matching.PipelineVerificationHandler;
-import org.apache.streampipes.manager.migration.MigrationPipelineGenerator;
import org.apache.streampipes.manager.migration.PipelineElementMigrationHandler;
import org.apache.streampipes.manager.migration.PipelineElementOffloadHandler;
import org.apache.streampipes.manager.recommender.ElementRecommender;
@@ -126,7 +125,8 @@ public class Operations {
public static PipelineOperationStatus startPipeline(
Pipeline pipeline, boolean visualize, boolean storeStatus,
boolean monitor) {
- return new PipelineExecutor(pipeline, visualize, storeStatus, monitor).startPipeline();
+ return PipelineExecutorFactory.createInvocationExecutor(pipeline, visualize, storeStatus, monitor).execute();
+ //return new PipelineExecutor(pipeline, visualize, storeStatus, monitor).startPipeline();
}
public static PipelineOperationStatus stopPipeline(
@@ -150,7 +150,7 @@ public class Operations {
public static PipelineOperationStatus stopPipeline(
Pipeline pipeline, boolean visualize, boolean storeStatus,
boolean monitor) {
- return new PipelineExecutor(pipeline, visualize, storeStatus, monitor).stopPipeline();
+ return PipelineExecutorFactory.createDetachExecutor(pipeline, visualize, storeStatus, monitor).execute();
}
public static List<RdfEndpointItem> getEndpointUriContents(List<RdfEndpoint> endpoints) {
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/reconfiguration/PipelineElementReconfigurationHandler.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/reconfiguration/PipelineElementReconfigurationHandler.java
index 9a4ff83..959530d 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/reconfiguration/PipelineElementReconfigurationHandler.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/reconfiguration/PipelineElementReconfigurationHandler.java
@@ -17,10 +17,7 @@
*/
package org.apache.streampipes.manager.reconfiguration;
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import org.apache.streampipes.commons.exceptions.SpRuntimeException;
-import org.apache.streampipes.manager.execution.pipeline.PipelineElementReconfigurationExecutor;
+import org.apache.streampipes.manager.execution.pipeline.executor.PipelineExecutorFactory;
import org.apache.streampipes.manager.operations.Operations;
import org.apache.streampipes.model.graph.DataProcessorInvocation;
import org.apache.streampipes.model.pipeline.*;
@@ -38,7 +35,7 @@ public class PipelineElementReconfigurationHandler {
private final PipelineOperationStatus pipelineReconfigurationStatus;
private final Pipeline reconfiguredPipeline;
- private Pipeline storedPipeline;
+ private final Pipeline storedPipeline;
public PipelineElementReconfigurationHandler(Pipeline reconfiguredPipeline) {
this.reconfiguredPipeline = reconfiguredPipeline;
@@ -70,7 +67,10 @@ public class PipelineElementReconfigurationHandler {
}
private PipelineOperationStatus reconfigurePipelineElement(PipelineElementReconfigurationEntity entity) {
- return new PipelineElementReconfigurationExecutor(reconfiguredPipeline, entity).reconfigurePipelineElement();
+ return PipelineExecutorFactory
+ .createReconfigurationExecutor(this.storedPipeline, false, false, false, this.reconfiguredPipeline, entity)
+ .execute();
+ //return new PipelineElementReconfigurationExecutor(reconfiguredPipeline, entity).reconfigurePipelineElement();
}
private List<PipelineElementReconfigurationEntity> comparePipelinesAndGetReconfiguration() {