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 <[email protected]> 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() {
