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 a340cd0c569b6dd869a78543a7aab2fe64935b4b Author: Daniel Gomm <[email protected]> AuthorDate: Thu Feb 18 15:34:53 2021 +0100 [WIP] Migration functionality: Add rollback capabilities and possibility for simultaneous migration of multiple processors --- .../manager/execution/http/MigrationHelpers.java | 75 ++++++++ .../manager/execution/http/PipelineExecutor.java | 196 +++++++++++---------- .../manager/matching/InvocationGraphBuilder.java | 131 ++++++-------- .../manager/matching/ProtocolSelector.java | 4 +- .../streampipes/manager/operations/Operations.java | 40 ++++- .../rest/impl/PipelineWithUserResource.java | 5 +- .../migrate-pipeline-processors.component.html | 2 +- .../migrate-pipeline-processors.component.ts | 33 +++- .../app/platform-services/apis/pipeline.service.ts | 6 +- 9 files changed, 297 insertions(+), 195 deletions(-) diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/MigrationHelpers.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/MigrationHelpers.java new file mode 100644 index 0000000..43aa454 --- /dev/null +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/MigrationHelpers.java @@ -0,0 +1,75 @@ +/* + * 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.http; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.streampipes.model.base.NamedStreamPipesEntity; +import org.apache.streampipes.model.graph.DataProcessorInvocation; +import org.apache.streampipes.model.pipeline.Pipeline; +import org.apache.streampipes.model.pipeline.PipelineElementStatus; +import org.apache.streampipes.model.pipeline.PipelineOperationStatus; +import org.apache.streampipes.sdk.helpers.Tuple2; + +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; + +public class MigrationHelpers { + + static public List<Tuple2<DataProcessorInvocation, DataProcessorInvocation>> getDelta(Pipeline pipelineX, Pipeline pipelineY){ + List<Tuple2<DataProcessorInvocation, DataProcessorInvocation>> delta = new ArrayList<>(); + pipelineX.getSepas().forEach(iX -> { + if (pipelineY.getSepas().stream().filter(iY -> iY.getElementId().equals(iX.getElementId())) + .noneMatch(iY -> iY.getDeploymentTargetNodeId().equals(iX.getDeploymentTargetNodeId()))){ + Optional<DataProcessorInvocation> invocationY = pipelineY.getSepas().stream(). + filter(iY -> iY.getDeploymentRunningInstanceId().equals(iX.getDeploymentRunningInstanceId())).findFirst(); + invocationY.ifPresent(dataProcessorInvocation -> delta.add(new Tuple2<>(iX, dataProcessorInvocation))); + } + }); + return delta; + } + + static public PipelineOperationStatus verifyPipelineOperationStatus(PipelineOperationStatus status, String successMessage, + String errorMessage) { + //Duplicate from method in GraphSubmitter + status.setSuccess(status.getElementStatus().stream().allMatch(PipelineElementStatus::isSuccess)); + if (status.isSuccess()) { + status.setTitle(successMessage); + } else { + status.setTitle(errorMessage); + } + return status; + } + + public static void exchangePipelineElement(Pipeline exchangePipeline, Tuple2<? extends NamedStreamPipesEntity, ? extends NamedStreamPipesEntity> entity){ + + if (entity.a instanceof DataProcessorInvocation){ + int index = exchangePipeline.getSepas().indexOf(entity.b); + exchangePipeline.getSepas().remove(index); + exchangePipeline.getSepas().add(index, (DataProcessorInvocation) entity.a); + } + } + + public static Pipeline deepCopyPipeline(Pipeline object) throws JsonProcessingException { + ObjectMapper objectMapper = new ObjectMapper(); + return objectMapper.readValue(objectMapper.writeValueAsString(object), Pipeline.class); + } + +} diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/PipelineExecutor.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/PipelineExecutor.java index 122156d..8068cbe 100644 --- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/PipelineExecutor.java +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/PipelineExecutor.java @@ -154,55 +154,101 @@ public class PipelineExecutor { return status; } - public PipelineOperationStatus updatePipelineDeploymentPartial(Pipeline pipelineOld){ - PipelineOperationStatus status = initPipelineOperationStatus(pipeline.getPipelineId(), pipeline.getName()); - //Adjust Relays in Description using existing configure Method in InvocationGraphBuilder + public PipelineOperationStatus migratePipelineElement(Pipeline currentPipeline, + Tuple2<DataProcessorInvocation, DataProcessorInvocation> targetElement){ + PipelineOperationStatus status = initPipelineOperationStatus(); + //Purge existing relays pipeline.getSepas().forEach(s -> s.setOutputStreamRelays(new ArrayList<>())); - - PipelineGraph oldPipelineGraph = new PipelineGraphBuilder(pipelineOld).buildGraph(); + //Generate new relays PipelineGraph newPipelineGraph = new PipelineGraphBuilder(pipeline).buildGraph(); new InvocationGraphBuilder(newPipelineGraph, pipeline.getPipelineId()).buildGraphs(); + //Get predecessors + List<NamedStreamPipesEntity> predecessors = new ArrayList<>(); + PipelineGraph currentPipelineGraph = new PipelineGraphBuilder(currentPipeline).buildGraph(); + PipelineGraphHelpers + .findStreams(newPipelineGraph) + .forEach(source -> predecessors.addAll(getPredecessors(source, targetElement.a, newPipelineGraph, new ArrayList<>()))); + //Find counterpart for predecessors in currentPipeline + List<Tuple2<NamedStreamPipesEntity, NamedStreamPipesEntity>> matchingPredecessors = new ArrayList<>(); + for(NamedStreamPipesEntity e : predecessors){ + matchingPredecessors.add(new Tuple2<>(e, findMatching(e, currentPipelineGraph))); + } + List<NamedStreamPipesEntity> currentPredecessors = matchingPredecessors.stream().map(t -> t.b).collect(Collectors.toList()); - getDelta(this.pipeline, pipelineOld).forEach(t -> { - List<NamedStreamPipesEntity> predecessors = new ArrayList<>(); - PipelineGraphHelpers - .findStreams(newPipelineGraph) - .forEach(source -> predecessors.addAll(getPredecessors(source, t.a, newPipelineGraph, new ArrayList<>()))); - List<Tuple2<NamedStreamPipesEntity, NamedStreamPipesEntity>> matchingPredecessors = new ArrayList<>(); - for(NamedStreamPipesEntity e : predecessors){ - matchingPredecessors.add(new Tuple2<>(e, findMatching(e, oldPipelineGraph))); - } - List<NamedStreamPipesEntity> predecessorsOld = new ArrayList<>(); - for(Tuple2<NamedStreamPipesEntity, NamedStreamPipesEntity> pred : matchingPredecessors){ - predecessorsOld.add(pred.b); - } + PipelineOperationStatus statusStartTarget = startPipelineElement(Collections.singletonList(targetElement.a)); + statusStartTarget.getElementStatus().forEach(status::addPipelineElementStatus); + if(!statusStartTarget.isSuccess()){ + //Target PE could not be started; nothing to roll back + return status; + } + //Stop relay to origin + PipelineOperationStatus statusStopRelays = stopRelays(currentPredecessors, targetElement.b); + statusStopRelays.getElementStatus().forEach(status::addPipelineElementStatus); + if(!statusStopRelays.isSuccess()){ + //Rollback of target PE and all successfully stopped Relays + stopPipelineElements(Collections.singletonList(targetElement.a)) + .getElementStatus().forEach(status::addPipelineElementStatus); + + Set<String> relaysToRollBack = statusStopRelays.getElementStatus().stream().filter(PipelineElementStatus::isSuccess) + .map(PipelineElementStatus::getElementId).collect(Collectors.toSet()); + Map<NamedStreamPipesEntity, SpDataStreamRelayContainer> rollbackRelays = findRelays(currentPredecessors, targetElement.b) + .entrySet() + .stream() + .filter(entry -> relaysToRollBack.contains(entry.getValue().getRunningStreamRelayInstanceId())) + .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)); + new GraphSubmitter(pipeline.getPipelineId(), + pipeline.getName(), new ArrayList<>(), new ArrayList<>(), new ArrayList<>()).invokeRelays(rollbackRelays) + .getElementStatus().forEach(status::addPipelineElementStatus); + return status; + } + //start Relay to target + PipelineOperationStatus statusStartRelays = startRelays(predecessors, targetElement.a); + statusStartRelays.getElementStatus().forEach(status::addPipelineElementStatus); + if(!statusStartRelays.isSuccess()){ + //Rollback target PE, stopped relays and all successfully started Relays + stopPipelineElements(Collections.singletonList(targetElement.a)) + .getElementStatus().forEach(status::addPipelineElementStatus); + + startRelays(currentPredecessors, targetElement.b) + .getElementStatus().forEach(status::addPipelineElementStatus); + + Set<String> relaysToRollBack = statusStartRelays.getElementStatus().stream().filter(PipelineElementStatus::isSuccess) + .map(PipelineElementStatus::getElementId).collect(Collectors.toSet()); + Map<NamedStreamPipesEntity, SpDataStreamRelayContainer> rollbackRelays = findRelays(predecessors, targetElement.a) + .entrySet() + .stream() + .filter(entry -> relaysToRollBack.contains(entry.getValue().getRunningStreamRelayInstanceId())) + .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)); + new GraphSubmitter(pipeline.getPipelineId(), + pipeline.getName(), new ArrayList<>(), new ArrayList<>(), new ArrayList<>()).detachRelays(rollbackRelays) + .getElementStatus().forEach(status::addPipelineElementStatus); + return status; + } + //Stop origin and associated relay + PipelineOperationStatus statusStopOrigin = stopPipelineElements(Collections.singletonList(targetElement.b)); + statusStopOrigin.getElementStatus().forEach(status::addPipelineElementStatus); + + if(!statusStopOrigin.isSuccess()){ + //Rollback everything + stopRelays(predecessors, targetElement.a) + .getElementStatus().forEach(status::addPipelineElementStatus); + startRelays(currentPredecessors, targetElement.b) + .getElementStatus().forEach(status::addPipelineElementStatus); + stopPipelineElements(Collections.singletonList(targetElement.a)) + .getElementStatus().forEach(status::addPipelineElementStatus); + return status; + } - PipelineOperationStatus statusStartTarget = startPipelineElement(Collections.singletonList(t.a)); - statusStartTarget.getElementStatus().forEach(status::addPipelineElementStatus); - - //Stop relay to origin - PipelineOperationStatus statusStopRelays = stopRelays(predecessorsOld, t.b); - statusStopRelays.getElementStatus().forEach(status::addPipelineElementStatus); - //start Relay to target - PipelineOperationStatus statusStartRelays = startRelays(predecessors, t.a); - statusStartRelays.getElementStatus().forEach(status::addPipelineElementStatus); - //Stop origin and associated relay - PipelineOperationStatus stopOrigin = stopPipelineElements(Collections.singletonList(t.b)); - stopOrigin.getElementStatus().forEach(status::addPipelineElementStatus); - - //Remove if unnecessary - List<InvocableStreamPipesEntity> graphs = new ArrayList<>(); - graphs.addAll(pipeline.getActions()); - graphs.addAll(pipeline.getSepas()); - List<SpDataSet> dataSets = pipeline.getStreams().stream().filter(s -> s instanceof SpDataSet).map(s -> new - SpDataSet((SpDataSet) s)).collect(Collectors.toList()); - storeInvocationGraphs(pipeline.getPipelineId(), graphs, dataSets); + List<InvocableStreamPipesEntity> graphs = new ArrayList<>(); + graphs.addAll(pipeline.getActions()); + graphs.addAll(pipeline.getSepas()); + List<SpDataSet> dataSets = pipeline.getStreams().stream().filter(s -> s instanceof SpDataSet).map(s -> new + SpDataSet((SpDataSet) s)).collect(Collectors.toList()); + storeInvocationGraphs(pipeline.getPipelineId(), graphs, dataSets); - }); - return verifyPipelineOperationStatus(status, - "Successfully migrated Pipeline Elements in Pipeline " + pipeline.getName(), - "Could not migrate all Pipeline Elements in Pipeline " + pipeline.getName()); + status.setSuccess(status.getElementStatus().stream().allMatch(PipelineElementStatus::isSuccess)); + return status; } private PipelineOperationStatus stopPipelineElements(List<NamedStreamPipesEntity> graphs){ @@ -210,13 +256,12 @@ public class PipelineExecutor { graphs.stream().filter(i -> i instanceof InvocableStreamPipesEntity).forEach(i -> invocations.add((InvocableStreamPipesEntity) i)); if (invocations.isEmpty()) { - PipelineOperationStatus ret = initPipelineOperationStatus(pipeline.getPipelineId(), pipeline.getName()); + PipelineOperationStatus ret = initPipelineOperationStatus(); ret.setSuccess(true); return ret; } - List<SpDataStreamRelayContainer> dataStreamRelayContainers = generateDataStreamRelays(invocations); return new GraphSubmitter(pipeline.getPipelineId(), - pipeline.getName(), invocations, new ArrayList<>(), dataStreamRelayContainers) + pipeline.getName(), invocations, new ArrayList<>(), new ArrayList<>()) .detachGraphs(); } @@ -225,46 +270,32 @@ public class PipelineExecutor { graphs.stream().filter(i -> i instanceof InvocableStreamPipesEntity).forEach(i -> invocations.add((InvocableStreamPipesEntity) i)); if (invocations.isEmpty()) { - PipelineOperationStatus ret = initPipelineOperationStatus(pipeline.getPipelineId(), pipeline.getName()); + PipelineOperationStatus ret = initPipelineOperationStatus(); ret.setSuccess(true); return ret; } List<InvocableStreamPipesEntity> decryptedGraphs = decryptSecrets(invocations); - List<SpDataStreamRelayContainer> dataStreamRelayContainers = generateDataStreamRelays(invocations); return new GraphSubmitter(pipeline.getPipelineId(), - pipeline.getName(), decryptedGraphs, new ArrayList<>(), dataStreamRelayContainers) + pipeline.getName(), decryptedGraphs, new ArrayList<>(), new ArrayList<>()) .invokeGraphs(); } private PipelineOperationStatus stopRelays(List<NamedStreamPipesEntity> predecessors, InvocableStreamPipesEntity target){ - Map<NamedStreamPipesEntity, SpDataStreamRelayContainer> relays = new HashMap<>(); - - findRelays(relays, predecessors, target); - - if (relays.isEmpty()) { - PipelineOperationStatus ret = initPipelineOperationStatus(pipeline.getPipelineId(), pipeline.getName()); - ret.setSuccess(true); - return ret; - } + Map<NamedStreamPipesEntity, SpDataStreamRelayContainer> relays = findRelays(predecessors, target); return new GraphSubmitter(pipeline.getPipelineId(), pipeline.getName(), new ArrayList<>(), new ArrayList<>(), new ArrayList<>()).detachRelays(relays); } private PipelineOperationStatus startRelays(List<NamedStreamPipesEntity> predecessors, InvocableStreamPipesEntity target){ - Map<NamedStreamPipesEntity, SpDataStreamRelayContainer> relays = new HashMap<>(); - findRelays(relays, predecessors, target); - - if (relays.isEmpty()) { - PipelineOperationStatus ret = initPipelineOperationStatus(pipeline.getPipelineId(), pipeline.getName()); - ret.setSuccess(true); - return ret; - } + Map<NamedStreamPipesEntity, SpDataStreamRelayContainer> relays = findRelays(predecessors, target); return new GraphSubmitter(pipeline.getPipelineId(), pipeline.getName(), new ArrayList<>(), new ArrayList<>(), new ArrayList<>()).invokeRelays(relays); } - private void findRelays(Map<NamedStreamPipesEntity, SpDataStreamRelayContainer> relays, - List<NamedStreamPipesEntity> predecessors, InvocableStreamPipesEntity target){ + private Map<NamedStreamPipesEntity, SpDataStreamRelayContainer> findRelays(List<NamedStreamPipesEntity> predecessors, + InvocableStreamPipesEntity target){ + + Map<NamedStreamPipesEntity, SpDataStreamRelayContainer> relays = new HashMap<>(); predecessors.forEach(pred -> { List<SpDataStreamRelay> dataStreamRelays = new ArrayList<>(); @@ -310,21 +341,9 @@ public class PipelineExecutor { } } }); - + return relays; } - private List<Tuple2<DataProcessorInvocation, DataProcessorInvocation>> getDelta(Pipeline pipelineX, Pipeline pipelineY){ - List<Tuple2<DataProcessorInvocation, DataProcessorInvocation>> delta = new ArrayList<>(); - pipelineX.getSepas().forEach(iX -> { - if (pipelineY.getSepas().stream().filter(iY -> iY.getElementId().equals(iX.getElementId())). - filter(iY -> iY.getDeploymentTargetNodeId().equals(iX.getDeploymentTargetNodeId())).count() == 0){ - Optional<DataProcessorInvocation> invocationY = pipelineY.getSepas().stream(). - filter(iY -> iY.getDeploymentRunningInstanceId().equals(iX.getDeploymentRunningInstanceId())).findFirst(); - invocationY.ifPresent(dataProcessorInvocation -> delta.add(new Tuple2<>(iX, dataProcessorInvocation))); - } - }); - return delta; - } private List<NamedStreamPipesEntity> getPredecessors(NamedStreamPipesEntity source, InvocableStreamPipesEntity target, PipelineGraph pipelineGraph, @@ -373,25 +392,14 @@ public class PipelineExecutor { return null; } - private PipelineOperationStatus initPipelineOperationStatus(String pipelineId, String pipelineName) { + private PipelineOperationStatus initPipelineOperationStatus() { //Duplicate from method in GraphSubmitter PipelineOperationStatus status = new PipelineOperationStatus(); - status.setPipelineId(pipelineId); - status.setPipelineName(pipelineName); + status.setPipelineId(pipeline.getPipelineId()); + status.setPipelineName(pipeline.getName()); return status; } - private PipelineOperationStatus verifyPipelineOperationStatus(PipelineOperationStatus status, String successMessage, - String errorMessage) { - //Duplicate from method in GraphSubmitter - status.setSuccess(status.getElementStatus().stream().allMatch(PipelineElementStatus::isSuccess)); - if (status.isSuccess()) { - status.setTitle(successMessage); - } else { - status.setTitle(errorMessage); - } - return status; - } private String extractTopic(EventGrounding eg) { return eg.getTransportProtocol().getTopicDefinition().getActualTopicName(); diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/InvocationGraphBuilder.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/InvocationGraphBuilder.java index d838b80..e5d9ae5 100644 --- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/InvocationGraphBuilder.java +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/InvocationGraphBuilder.java @@ -52,7 +52,6 @@ public class InvocationGraphBuilder { private static final String DEFAULT_TAG = "default"; private final PipelineGraph pipelineGraph; private final String pipelineId; - private Integer uniquePeIndex = 0; private final List<InvocableStreamPipesEntity> graphs; public InvocationGraphBuilder(PipelineGraph pipelineGraph, String pipelineId) { @@ -102,63 +101,15 @@ public class InvocationGraphBuilder { } else if (defaultDeploymentTarget(t) && source instanceof DataProcessorInvocation) { // target runs on cloud node: use central cloud broker, e.g. kafka - - if (!eventRelayExists(source, t)) { - - if(!graphExists(t.getDOM())) { - // add initial relay grounding to source processor - ((DataProcessorInvocation) source).addOutputStreamRelay( - new SpDataStreamRelay(generateRelayGrounding(inputGrounding,false))); - } else { - // modify relay topic of existing relay grounding - modifyTopicForDataStreamRelay(source, t, extractTopic(inputGrounding)); - } - modifyTopicForTargetInputStream(source, t, extractTopic(inputGrounding)); - } else { - - EventGrounding updatedRelayGrounding = generateRelayGrounding(inputGrounding, false); - removeExistingStreamRelay(source, t); - ((DataProcessorInvocation) source).addOutputStreamRelay(new SpDataStreamRelay(updatedRelayGrounding)); - modifyTopicForTargetInputStream(source, t, extractTopic(inputGrounding)); - } + connectSourceToTarget((DataProcessorInvocation) source, t, inputGrounding, false); } else { // target runs on other edge node: use target edge node broker - if (!eventRelayExists(source ,t)) { - - EventGrounding relayGrounding = generateRelayGrounding(inputGrounding, t,true); - if(!graphExists(t.getDOM())) { - ((DataProcessorInvocation) source).addOutputStreamRelay(new SpDataStreamRelay(relayGrounding)); - } else { - t.getInputStreams() - .get(getIndex(source.getDOM(), t)) - .setEventGrounding(relayGrounding); - } - - t.getInputStreams() - .get(getIndex(source.getDOM(), t)) - .setEventGrounding(((DataProcessorInvocation) source) - .getOutputStreamRelays() - .get(getIndex(source.getDOM(), t)) - .getEventGrounding()); - } else { - EventGrounding updatedRelayGrounding = generateRelayGrounding(inputGrounding, t,true); - removeExistingStreamRelay(source, t); - ((DataProcessorInvocation) source).addOutputStreamRelay(new SpDataStreamRelay(updatedRelayGrounding)); - - t.getInputStreams() - .get(getIndex(source.getDOM(), t)) - .setEventGrounding(updatedRelayGrounding); - } + connectSourceToTarget((DataProcessorInvocation) source, t, inputGrounding, true); } } else { - // TODO: Handle following edge situation: - // data stream -> invocable (processor, sink) in edge deployments that do not reside on same node - // idea: trigger corresponding node controller to relay topic to adjecent broker (either node broker or - // global cloud broker) - if (matchingDeploymentTargets(source, t)) { t.getInputStreams() .get(getIndex(source.getDOM(), t)) @@ -197,15 +148,12 @@ public class InvocationGraphBuilder { .setEventSchema(getInputSchema(source)); String elementIdentifier = makeElementIdentifier(pipelineId, - inputGrounding.getTransportProtocol().getTopicDefinition().getActualTopicName(), t.getName()); + inputGrounding.getTransportProtocol().getTopicDefinition().getActualTopicName(), t.getName(), t.getDOM()); t.setElementId(t.getBelongsTo() + "/" + elementIdentifier); t.setDeploymentRunningInstanceId(elementIdentifier); t.setCorrespondingPipeline(pipelineId); t.setStatusInfoSettings(makeStatusInfoSettings(elementIdentifier)); - - uniquePeIndex++; - configure(t, getConnections(t)); }); @@ -215,7 +163,7 @@ public class InvocationGraphBuilder { ((DataProcessorInvocation) source).removeOutputStreamRelay( ((DataProcessorInvocation) source) .getOutputStreamRelays() - .get(getIndex(source.getDOM(), t))); + .get(getRelayIndex((DataProcessorInvocation) source, t))); } private boolean targetInvocableOnEdgeNode(InvocableStreamPipesEntity t) { @@ -226,27 +174,36 @@ public class InvocationGraphBuilder { return t.getDeploymentTargetNodeId() != null && t.getDeploymentTargetNodeId().equals(DEFAULT_TAG); } - private void modifyTopicForTargetInputStream(NamedStreamPipesEntity s, InvocableStreamPipesEntity t, - String topic) { - t.getInputStreams() - .get(getIndex(s.getDOM(), t)) - .getEventGrounding() - .getTransportProtocol() - .getTopicDefinition() - .setActualTopicName(topic); + private void connectSourceToTarget(DataProcessorInvocation source, InvocableStreamPipesEntity target, EventGrounding inputGrounding, boolean edgeToEdgeRelay){ + EventGrounding relayGrounding = edgeToEdgeRelay ? + generateRelayGrounding(inputGrounding, target, true) : generateRelayGrounding(inputGrounding, false); + if (!eventRelayExists(source, target)) { + modifyTargetInputStream(source, target, relayGrounding, edgeToEdgeRelay); + // add initial relay grounding to source processor + source.addOutputStreamRelay(new SpDataStreamRelay(relayGrounding)); + } else { + removeExistingStreamRelay(source, target); + modifyTargetInputStream(source, target, relayGrounding, edgeToEdgeRelay); + source.addOutputStreamRelay(new SpDataStreamRelay(relayGrounding)); + } } - private void modifyTopicForDataStreamRelay(NamedStreamPipesEntity s, InvocableStreamPipesEntity t, - String topic) { - ((DataProcessorInvocation) s) - .getOutputStreamRelays() - .get(getIndex(s.getDOM(), t)) - .getEventGrounding() - .getTransportProtocol() - .getTopicDefinition() - .setActualTopicName(topic); + private void modifyTargetInputStream(NamedStreamPipesEntity s, InvocableStreamPipesEntity t, + EventGrounding grounding, boolean edgeToEdgeRelay) { + SpDataStream inputStream = t.getInputStreams() + .get(getIndex(s.getDOM(),t)); //Is this correct? + if(edgeToEdgeRelay){ + inputStream.setEventGrounding(grounding); + }else{ + inputStream.getEventGrounding() + .getTransportProtocol() + .getTopicDefinition() + .setActualTopicName(extractTopic(grounding)); + inputStream.getEventGrounding().setElementId(grounding.getElementId()); + } } + private boolean deploymentTargetNotNull(NamedStreamPipesEntity s, InvocableStreamPipesEntity t) { if (s instanceof SpDataStream) { return ((SpDataStream) s).getDeploymentTargetNodeId() != null && t.getDeploymentTargetNodeId() != null; @@ -261,11 +218,11 @@ public class InvocationGraphBuilder { } private boolean eventRelayExists(NamedStreamPipesEntity s, InvocableStreamPipesEntity t) { - int idx = getIndex(s.getDOM(), t); - return ((DataProcessorInvocation) s).getOutputStreamRelays() + SpDataStream targetInput = t.getInputStreams().get(getIndex(s.getDOM(), t)); + return ((DataProcessorInvocation)s).getOutputStreamRelays() .stream() - .anyMatch(i -> extractTopic(i.getEventGrounding()).equals( - extractTopic(t.getInputStreams().get(idx).getEventGrounding()))); + .anyMatch(i -> i.getEventGrounding().getElementId().equals( + targetInput.getEventGrounding().getElementId())); } private Tuple2<EventSchema,? extends OutputStrategy> getOutputSettings(DataProcessorInvocation dataProcessorInvocation) { @@ -344,14 +301,14 @@ public class InvocationGraphBuilder { return statusSettings; } - private String makeElementIdentifier(String pipelineId, String topic, String elementName) { + private String makeElementIdentifier(String pipelineId, String topic, String elementName, String dom) { return pipelineId + "-" + topic + "-" + elementName.replaceAll(" ", "").toLowerCase() + "-" - + uniquePeIndex; + + dom.split("_")[dom.split("_").length-1]; } private EventSchema getInputSchema(NamedStreamPipesEntity source) { @@ -378,6 +335,21 @@ public class InvocationGraphBuilder { return targetElement.getConnectedTo().indexOf(sourceDomId); } + private Integer getRelayIndex(DataProcessorInvocation source, InvocableStreamPipesEntity target) throws SpRuntimeException { + Optional<SpDataStreamRelay> relay = source.getOutputStreamRelays().stream().filter(r -> + target.getInputStreams().stream().map(s -> + s.getEventGrounding() + .getElementId()) + .collect(Collectors.toList()) + .contains(r.getEventGrounding() + .getElementId())).findFirst(); + if(relay.isPresent()){ + return source.getOutputStreamRelays().indexOf(relay.get()); + }else { + throw new SpRuntimeException("Index for relay not available"); + } + } + private boolean graphExists(String domId) { return graphs .stream() @@ -409,6 +381,7 @@ public class InvocationGraphBuilder { BackendConfig.INSTANCE.getMessagingSettings().getPrioritizedProtocols().get(0), topic)); } eg.setTransportFormats(sourceInvocableOutputGrounding.getTransportFormats()); + eg.setElementId(sourceInvocableOutputGrounding.getElementId()); return eg; } diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/ProtocolSelector.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/ProtocolSelector.java index 9967fe5..117fbb8 100644 --- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/ProtocolSelector.java +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/ProtocolSelector.java @@ -79,7 +79,9 @@ public class ProtocolSelector extends GroundingSelector { outputTopic); } else { MqttTransportProtocol tp = (MqttTransportProtocol) - ((InvocableStreamPipesEntity) source).getInputStreams().get(0).getEventGrounding().getTransportProtocol(); + ((InvocableStreamPipesEntity) source).getInputStreams().stream() + .filter(s -> (s.getEventGrounding().getTransportProtocol() instanceof MqttTransportProtocol)) + .findFirst().get().getEventGrounding().getTransportProtocol(); return new MqttTransportProtocol(tp.getBrokerHostname(), tp.getPort(), outputTopic); } } 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 6648499..7861a6f 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 @@ -18,10 +18,12 @@ package org.apache.streampipes.manager.operations; +import com.fasterxml.jackson.core.JsonProcessingException; import org.apache.streampipes.commons.exceptions.NoSuitableSepasAvailableException; 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.http.MigrationHelpers; import org.apache.streampipes.manager.execution.http.PipelineExecutor; import org.apache.streampipes.manager.execution.http.PipelineStorageService; import org.apache.streampipes.manager.matching.DataSetGroundingSelector; @@ -36,6 +38,7 @@ import org.apache.streampipes.manager.topic.WildcardTopicGenerator; import org.apache.streampipes.manager.verification.extractor.TypeExtractor; import org.apache.streampipes.model.SpDataSet; import org.apache.streampipes.model.SpDataStream; +import org.apache.streampipes.model.base.InvocableStreamPipesEntity; import org.apache.streampipes.model.client.endpoint.RdfEndpoint; import org.apache.streampipes.model.client.endpoint.RdfEndpointItem; import org.apache.streampipes.model.graph.DataProcessorInvocation; @@ -44,6 +47,7 @@ import org.apache.streampipes.model.message.Message; import org.apache.streampipes.model.message.PipelineModificationMessage; import org.apache.streampipes.model.pipeline.Pipeline; import org.apache.streampipes.model.pipeline.PipelineElementRecommendationMessage; +import org.apache.streampipes.model.pipeline.PipelineElementStatus; import org.apache.streampipes.model.pipeline.PipelineOperationStatus; import org.apache.streampipes.model.runtime.RuntimeOptionsRequest; import org.apache.streampipes.model.runtime.RuntimeOptionsResponse; @@ -54,6 +58,8 @@ import org.apache.streampipes.storage.management.StorageDispatcher; import java.util.ArrayList; import java.util.List; +import java.util.Optional; +import java.util.stream.Collectors; /** @@ -190,9 +196,35 @@ public class Operations { return PipelineElementRuntimeInfoFetcher.INSTANCE.getCurrentData(spDataStream); } - public static PipelineOperationStatus updatePipelineDeploymentPartial(Pipeline pipelineOld, Pipeline pipelineNew, boolean visualize, boolean storeStatus, - boolean monitor){ - return new PipelineExecutor(pipelineNew, visualize, storeStatus, monitor) - .updatePipelineDeploymentPartial(pipelineOld); + public static PipelineOperationStatus migratePipelineProcessors(Pipeline newPipeline, boolean visualize, boolean storeStatus, + boolean monitor) throws SpRuntimeException{ + + + Pipeline currentPipeline = StorageDispatcher.INSTANCE.getNoSqlStore().getPipelineStorageAPI().getPipeline(newPipeline.getPipelineId()); + Pipeline migrationPipeline = StorageDispatcher.INSTANCE.getNoSqlStore().getPipelineStorageAPI().getPipeline(newPipeline.getPipelineId()); + PipelineOperationStatus status = new PipelineOperationStatus(); + + for(Tuple2<DataProcessorInvocation, DataProcessorInvocation> t : MigrationHelpers.getDelta(newPipeline, migrationPipeline)){ + MigrationHelpers.exchangePipelineElement(migrationPipeline, t); + + PipelineOperationStatus migrationStatus = new PipelineExecutor(migrationPipeline, visualize, storeStatus, monitor) + .migratePipelineElement(currentPipeline, t); + migrationStatus.getElementStatus().forEach(status::addPipelineElementStatus); + if (migrationStatus.isSuccess()) { + try { + currentPipeline = MigrationHelpers.deepCopyPipeline(migrationPipeline); + } catch (JsonProcessingException e) { + throw new SpRuntimeException(e); + } + } else { + Tuple2<DataProcessorInvocation, DataProcessorInvocation> failedMigration = new Tuple2<>(t.b, t.a); + MigrationHelpers.exchangePipelineElement(migrationPipeline, failedMigration); + } + } + + overwritePipeline(migrationPipeline); + return MigrationHelpers.verifyPipelineOperationStatus(status, + "Successfully migrated Pipeline Elements in Pipeline " + newPipeline.getName(), + "Could not migrate all Pipeline Elements in Pipeline " + newPipeline.getName()); } } diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineWithUserResource.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineWithUserResource.java index 4b10996..0ccd0f3 100644 --- a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineWithUserResource.java +++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineWithUserResource.java @@ -258,10 +258,7 @@ public class PipelineWithUserResource extends AbstractRestInterface implements I @PathParam("pipelineId") String pipelineId, Pipeline pipelineNew) { try { - Pipeline pipelineOld = getPipelineStorage() - .getPipeline(pipelineId); - PipelineOperationStatus status = Operations.updatePipelineDeploymentPartial(pipelineOld, pipelineNew, true, true, true); - Operations.overwritePipeline(pipelineNew); + PipelineOperationStatus status = Operations.migratePipelineProcessors(pipelineNew, true, true, true); return ok(status); } catch (Exception e) { e.printStackTrace(); diff --git a/ui/src/app/editor/dialog/migrate-pipeline-processors/migrate-pipeline-processors.component.html b/ui/src/app/editor/dialog/migrate-pipeline-processors/migrate-pipeline-processors.component.html index 88cf5bd..d1a7c83 100644 --- a/ui/src/app/editor/dialog/migrate-pipeline-processors/migrate-pipeline-processors.component.html +++ b/ui/src/app/editor/dialog/migrate-pipeline-processors/migrate-pipeline-processors.component.html @@ -162,7 +162,7 @@ <mat-divider></mat-divider> <div class="sp-dialog-actions" fxLayoutAlign="left center"> <button [disabled]="!submitPipelineForm.valid || saving || saved" mat-button mat-raised-button - color="primary" (click)="migratePipelineProccesors()" style="margin-right:10px;"> + color="primary" (click)="migratePipelineProcessors()" style="margin-right:10px;"> Start Migration </button> <button mat-button mat-raised-button class="mat-basic" (click)="hide()"> diff --git a/ui/src/app/editor/dialog/migrate-pipeline-processors/migrate-pipeline-processors.component.ts b/ui/src/app/editor/dialog/migrate-pipeline-processors/migrate-pipeline-processors.component.ts index b555991..8ea0f81 100644 --- a/ui/src/app/editor/dialog/migrate-pipeline-processors/migrate-pipeline-processors.component.ts +++ b/ui/src/app/editor/dialog/migrate-pipeline-processors/migrate-pipeline-processors.component.ts @@ -2,7 +2,7 @@ import {Component, Input, OnInit} from '@angular/core'; import { DataProcessorInvocation, Message, NodeInfoDescription, - Pipeline, + Pipeline, PipelineOperationStatus, StaticNodeMetadata } from "../../../core-model/gen/streampipes-model"; import {FormControl, FormGroup, Validators} from "@angular/forms"; @@ -11,6 +11,10 @@ import {DialogRef} from "../../../core-ui/dialog/base-dialog/dialog-ref"; import {ObjectProvider} from "../../services/object-provider.service"; import {PipelineService} from "../../../platform-services/apis/pipeline.service"; import {NodeService} from "../../../platform-services/apis/node.service"; +import {DialogService} from "../../../core-ui/dialog/base-dialog/base-dialog.service"; +import {PipelineStatusDialogComponent} from "../../../pipelines/dialog/pipeline-status/pipeline-status-dialog.component"; +import {PanelType} from "../../../core-ui/dialog/base-dialog/base-dialog.model"; +import {map} from "rxjs/operators"; @Component({ selector: 'migrate-pipeline-processors', @@ -42,7 +46,8 @@ export class MigratePipelineProcessorsComponent implements OnInit { private dialogRef: DialogRef<MigratePipelineProcessorsComponent>, private objectProvider: ObjectProvider, private pipelineService: PipelineService, - private nodeService: NodeService) { + private nodeService: NodeService, + private dialogService: DialogService) { this.advancedSettings = true; this.panelOpenState = true; @@ -86,9 +91,10 @@ export class MigratePipelineProcessorsComponent implements OnInit { : source as T; } - displayErrors(data?: string) { + displayErrors(statusMessage?: PipelineOperationStatus) { this.storageError = true; - this.errorMessage = data; + this.errorMessage = statusMessage.title; + this.showDialog(statusMessage); } loadAndPrepareEdgeNodes() { @@ -154,7 +160,7 @@ export class MigratePipelineProcessorsComponent implements OnInit { }) } - migratePipelineProccesors() { + migratePipelineProcessors() { if (this.tmpPipeline.name == "") { //this.showToast("error", "Please enter a name for your pipeline"); return false; @@ -174,19 +180,30 @@ export class MigratePipelineProcessorsComponent implements OnInit { if (statusMessage.success) { this.afterMigration(statusMessage, this.pipeline._id); } else { - this.displayErrors(statusMessage.notifications[0]); + this.displayErrors(statusMessage); } }, data => { this.displayErrors(); }); }; - afterMigration(statusMessage: Message, pipelineId?: string) { + afterMigration(statusMessage: PipelineOperationStatus, pipelineId?: string) { this.hide(); - // TODO: show dialog with statusMessagge + this.showDialog(statusMessage); // this.editorService.removePipelineFromCache().subscribe(); } + showDialog(data: PipelineOperationStatus) { + this.dialogService.open(PipelineStatusDialogComponent, { + panelType: PanelType.STANDARD_PANEL, + title: "Pipeline Status", + width: "70vw", + data: { + "pipelineOperationStatus": data + } + }); + }; + hide() { this.dialogRef.close(); }; diff --git a/ui/src/app/platform-services/apis/pipeline.service.ts b/ui/src/app/platform-services/apis/pipeline.service.ts index 3c1c312..0abfc8a 100644 --- a/ui/src/app/platform-services/apis/pipeline.service.ts +++ b/ui/src/app/platform-services/apis/pipeline.service.ts @@ -81,12 +81,10 @@ export class PipelineService { })); } - migratePipeline(pipeline: Pipeline): Observable<Message> { + migratePipeline(pipeline: Pipeline): Observable<PipelineOperationStatus> { var pipelineId = pipeline._id; return this.http.post(this.platformServicesCommons.authUserBasePath() + "/pipelines/migrate/" + pipelineId, pipeline) - .pipe(map(response => { - return Message.fromData(response as Message); - })); + .pipe(map(result => PipelineOperationStatus.fromData(result as PipelineOperationStatus))); } getOwnPipelines(): Observable<Pipeline[]> {
