You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampipes.apache.org by wi...@apache.org on 2020/11/25 22:21:38 UTC
[incubator-streampipes] 01/05: Merge branch 'dev' into
edge-extensions
This is an automated email from the ASF dual-hosted git repository.
wiener pushed a commit to branch edge-extensions
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit d16ecea65e1df5bbc092ddd243eb2a290f0527f3
Merge: 84d72fa 427a7ef
Author: Patrick Wiener <wi...@fzi.de>
AuthorDate: Tue Nov 24 10:42:20 2020 +0100
Merge branch 'dev' into edge-extensions
.github/workflows/build.yml | 21 +---
.idea/runConfigurations/backend.xml | 2 +
.idea/runConfigurations/connect_master.xml | 16 ---
streampipes-backend/pom.xml | 7 +-
.../backend/StreamPipesResourceConfig.java | 14 +++
streampipes-backend/src/main/resources/shiro.ini | 4 +
streampipes-connect-container-master/Dockerfile | 25 -----
.../aarch64.Dockerfile | 26 -----
.../arm.Dockerfile | 26 -----
.../development/env | 21 ----
streampipes-connect-container-master/pom.xml | 25 -----
.../master/init/AdapterMasterContainer.java | 48 ---------
.../init/AdapterMasterContainerResourceConfig.java | 46 ---------
.../master/management/AdapterMasterManagement.java | 7 +-
.../container/master/rest/AdapterResource.java | 2 +-
.../master/rest/AdapterTemplateResource.java | 2 +-
.../container/master/rest/DescriptionResource.java | 2 +-
.../container/master/rest/GuessResource.java | 2 +-
.../master/rest/RuntimeResolvableResource.java | 2 +-
.../container/master/rest/SourcesResource.java | 14 +--
.../container/master/rest/UnitResource.java | 2 +-
.../container/master/rest/WelcomePageMaster.java | 2 +-
.../master/rest/WorkerAdministrationResource.java | 2 +-
.../src/main/resources/banner.txt | 6 --
.../worker/management/MasterRestClient.java | 2 +-
.../streampipes/connect/config/ConfigKeys.java | 4 +-
.../connect/config/ConnectContainerConfig.java | 19 +---
.../apache/streampipes/connect/init/Config.java | 25 -----
.../execution/http/DataSetEntityUrlGenerator.java | 8 +-
.../execution/http/EndpointUrlGenerator.java | 4 +-
.../manager/execution/http/GraphSubmitter.java | 112 ++++++++++++++-------
.../http/InvocableEntityUrlGenerator.java | 26 ++---
ui/nginx_config/default.conf | 6 +-
ui/nginx_config/ssl.conf | 6 +-
.../data-marketplace/data-marketplace.component.ts | 2 +-
.../connect/services/data-marketplace.service.ts | 11 +-
ui/src/app/connect/services/rest.service.ts | 28 +++---
ui/webpack.partial.dev.js | 4 +-
38 files changed, 164 insertions(+), 417 deletions(-)
diff --cc .github/workflows/build.yml
index e91b6be,199c39d..29380a3
--- a/.github/workflows/build.yml
+++ b/.github/workflows/build.yml
@@@ -71,44 -71,6 +71,25 @@@ jobs
docker build --pull --build-arg BASE_IMAGE=$BASE_IMG_JRE_ARM64V8 -t $IMG_NAME_ARM64V8 -f aarch64.Dockerfile .
docker push $IMG_NAME_DEFAULT
docker push $IMG_NAME_AMD64
+ docker push $IMG_NAME_ARM32V7
+ docker push $IMG_NAME_ARM64V8
+ docker manifest create $IMG_NAME_DEFAULT $IMG_NAME_AMD64 $IMG_NAME_ARM32V7 $IMG_NAME_ARM64V8
+ docker manifest annotate $IMG_NAME_DEFAULT $IMG_NAME_ARM32V7 --os linux --arch arm
+ docker manifest annotate $IMG_NAME_DEFAULT $IMG_NAME_ARM64V8 --os linux --arch arm64
+ docker manifest push $IMG_NAME_DEFAULT
- - name: Build and Push Docker Connect Image
- working-directory: ./streampipes-connect-container-master
- env:
- IMG_NAME_DEFAULT: ${{ env.DOCKERHUB_APACHE_REPO }}/streampipes-connect-container:${{ env.MVN_VERSION }}
- IMG_NAME_AMD64: ${{ env.DOCKERHUB_APACHE_REPO }}/streampipes-connect-container:amd64-${{ env.MVN_VERSION }}
- IMG_NAME_ARM32V7: ${{ env.DOCKERHUB_APACHE_REPO }}/streampipes-connect-container:arm32v7-${{ env.MVN_VERSION }}
- IMG_NAME_ARM64V8: ${{ env.DOCKERHUB_APACHE_REPO }}/streampipes-connect-container:arm64v8-${{ env.MVN_VERSION }}
- run: |
- docker build --pull --build-arg BASE_IMAGE=$BASE_IMG_JRE_DEFAULT -t $IMG_NAME_DEFAULT -t $IMG_NAME_AMD64 -f Dockerfile .
- docker build --pull --build-arg BASE_IMAGE=$BASE_IMG_JRE_ARM32V7 -t $IMG_NAME_ARM32V7 -f arm.Dockerfile .
- docker build --pull --build-arg BASE_IMAGE=$BASE_IMG_JRE_ARM64V8 -t $IMG_NAME_ARM64V8 -f aarch64.Dockerfile .
- docker push $IMG_NAME_DEFAULT
- docker push $IMG_NAME_AMD64
- docker push $IMG_NAME_ARM32V7
- docker push $IMG_NAME_ARM64V8
- docker manifest create $IMG_NAME_DEFAULT $IMG_NAME_AMD64 $IMG_NAME_ARM32V7 $IMG_NAME_ARM64V8
- docker manifest annotate $IMG_NAME_DEFAULT $IMG_NAME_ARM32V7 --os linux --arch arm
- docker manifest annotate $IMG_NAME_DEFAULT $IMG_NAME_ARM64V8 --os linux --arch arm64
- docker manifest push $IMG_NAME_DEFAULT
+ - name: Build and Push Docker Node Controller Image
+ working-directory: ./streampipes-node-controller-container
+ env:
+ IMG_NAME_DEFAULT: ${{ env.DOCKERHUB_APACHE_REPO }}/node-controller:${{ env.MVN_VERSION }}
+ IMG_NAME_AMD64: ${{ env.DOCKERHUB_APACHE_REPO }}/node-controller:amd64-${{ env.MVN_VERSION }}
+ IMG_NAME_ARM32V7: ${{ env.DOCKERHUB_APACHE_REPO }}/node-controller:arm32v7-${{ env.MVN_VERSION }}
+ IMG_NAME_ARM64V8: ${{ env.DOCKERHUB_APACHE_REPO }}/node-controller:arm64v8-${{ env.MVN_VERSION }}
+ run: |
+ docker build --pull --build-arg BASE_IMG=$BASE_IMG_JRE_DEFAULT -t $IMG_NAME_DEFAULT -t $IMG_NAME_AMD64 -f Dockerfile .
+ docker build --pull --build-arg BASE_IMG=$BASE_IMG_JRE_ARM32V7 -t $IMG_NAME_ARM32V7 -f arm.Dockerfile .
+ docker build --pull --build-arg BASE_IMG=$BASE_IMG_JRE_ARM64V8 -t $IMG_NAME_ARM64V8 -f aarch64.Dockerfile .
+ docker push $IMG_NAME_DEFAULT
+ docker push $IMG_NAME_AMD64
docker push $IMG_NAME_ARM32V7
docker push $IMG_NAME_ARM64V8
docker manifest create $IMG_NAME_DEFAULT $IMG_NAME_AMD64 $IMG_NAME_ARM32V7 $IMG_NAME_ARM64V8
diff --cc streampipes-connect-container-master/development/env
index cca9a9e,0000000..e69de29
mode 100644,000000..100644
--- a/streampipes-connect-container-master/development/env
+++ b/streampipes-connect-container-master/development/env
diff --cc streampipes-connect-container-master/src/main/java/org/apache/streampipes/connect/container/master/init/AdapterMasterContainer.java
index 2df6871,0000000..e69de29
mode 100644,000000..100644
--- a/streampipes-connect-container-master/src/main/java/org/apache/streampipes/connect/container/master/init/AdapterMasterContainer.java
+++ b/streampipes-connect-container-master/src/main/java/org/apache/streampipes/connect/container/master/init/AdapterMasterContainer.java
diff --cc streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/DataSetEntityUrlGenerator.java
index bcacf31,630dcb5..0e9a3eb
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/DataSetEntityUrlGenerator.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/DataSetEntityUrlGenerator.java
@@@ -15,25 -15,20 +15,25 @@@
* limitations under the License.
*
*/
+package org.apache.streampipes.manager.execution.http;
-import {Component, Input} from '@angular/core';
-import {StreampipesPeContainerConifgs} from "../shared/streampipes-pe-container-configs";
-import {ConfigurationService} from '../shared/configuration.service'
+import org.apache.streampipes.model.SpDataSet;
-@Component({
- selector: 'consul-configs-boolean',
- templateUrl: './consul-configs-boolean.component.html',
- styleUrls: ['./consul-configs-boolean.component.css'],
- providers: [ConfigurationService]
-})
-export class ConsulConfigsBooleanComponent {
- @Input() configuration: StreampipesPeContainerConifgs
- constructor(public configService:ConfigurationService) {
+public class DataSetEntityUrlGenerator extends EndpointUrlGenerator<SpDataSet> {
+
+ public DataSetEntityUrlGenerator(SpDataSet pipelineElement) {
+ super(pipelineElement);
+ }
+
+ @Override
- public String generateStartPipelineEndpointUrl() {
- return pipelineElement.getUri();
++ public String generateInvokeEndpoint() {
++ return pipelineElement.getElementId();
}
-}
+ @Override
- public String generateStopPipelineEndpointUrl() {
- return pipelineElement.getUri()
++ public String generateDetachEndpoint() {
++ return pipelineElement.getElementId()
+ + SLASH
+ + pipelineElement.getDatasetInvocationId() ;
+ }
+}
diff --cc streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/EndpointUrlGenerator.java
index 9cca57b,c5dce8d..4222c77
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/EndpointUrlGenerator.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/EndpointUrlGenerator.java
@@@ -15,23 -15,32 +15,23 @@@
* limitations under the License.
*
*/
+package org.apache.streampipes.manager.execution.http;
-@import '../../../../scss/sp/sp-dialog.scss';
+import org.apache.streampipes.model.base.NamedStreamPipesEntity;
-.customize-section {
- display:flex;
- flex: 1 1 auto;
- padding: 20px;
-}
+public abstract class EndpointUrlGenerator<T extends NamedStreamPipesEntity> {
-.padding-20 {
- padding: 20px;
-}
+ protected static final String SLASH = "/";
+ protected static final String URLPREFIX = "http://";
-.mb-10 {
- margin-bottom: 10px;
-}
+ protected T pipelineElement;
-::ng-deep .pipeline-radio-group .mat-radio-label {
- padding: 0;
-}
+ public EndpointUrlGenerator(T pipelineElement) {
+ this.pipelineElement = pipelineElement;
+ }
- public abstract String generateStartPipelineEndpointUrl();
-.status-text {
- font-size: 14pt;
- margin-top:10px;
-}
++ public abstract String generateInvokeEndpoint();
+
- public abstract String generateStopPipelineEndpointUrl();
++ public abstract String generateDetachEndpoint();
-.status-subtext {
- font-size: 12pt;
-}
+}
diff --cc streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/GraphSubmitter.java
index e29de56,1899ee1..173e0b0
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/GraphSubmitter.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/GraphSubmitter.java
@@@ -18,6 -18,6 +18,7 @@@
package org.apache.streampipes.manager.execution.http;
++import org.apache.streampipes.model.base.NamedStreamPipesEntity;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.streampipes.model.SpDataSet;
@@@ -30,14 -30,14 +31,13 @@@ import java.util.Optional
public class GraphSubmitter {
-- private List<InvocableStreamPipesEntity> graphs;
-- private List<SpDataSet> dataSets;
--
-- private String pipelineId;
-- private String pipelineName;
--
private final static Logger LOG = LoggerFactory.getLogger(GraphSubmitter.class);
++ private final List<InvocableStreamPipesEntity> graphs;
++ private final List<SpDataSet> dataSets;
++ private final String pipelineId;
++ private final String pipelineName;
++
public GraphSubmitter(String pipelineId, String pipelineName, List<InvocableStreamPipesEntity> graphs,
List<SpDataSet> dataSets) {
this.graphs = graphs;
@@@ -47,62 -47,62 +47,98 @@@
}
public PipelineOperationStatus invokeGraphs() {
-- PipelineOperationStatus status = new PipelineOperationStatus();
-- status.setPipelineId(pipelineId);
-- status.setPipelineName(pipelineName);
++ PipelineOperationStatus status = initPipelineOperationStatus();
--
-- graphs.forEach(g -> status.addPipelineElementStatus(new HttpRequestBuilder(g, g.getBelongsTo()).invoke()));
-- if (status.getElementStatus().stream().allMatch(PipelineElementStatus::isSuccess)) {
- dataSets.forEach(dataSet ->
- dataSets.forEach(dataSet ->
-- status.addPipelineElementStatus
- (new HttpRequestBuilder(dataSet, dataSet.getElementId()).invoke()));
- (new HttpRequestBuilder(dataSet, dataSet.getUri()).invoke()));
++ graphs.forEach(graph -> invoke(new InvocableEntityUrlGenerator(graph), graph, status));
++ // only invoke datasets when following pipeline elements are started
++ if (allInvocableEntitiesRunning(status)) {
++ dataSets.forEach(dataset -> invoke(new DataSetEntityUrlGenerator(dataset), dataset, status));
}
++
++ return verifyPipelineOperationStatus(
++ status,
++ "Successfully started pipeline " + pipelineName,
++ "Could not start pipeline" + pipelineName,
++ true);
++ }
++
++ public PipelineOperationStatus detachGraphs() {
++ PipelineOperationStatus status = initPipelineOperationStatus();
++
++ graphs.forEach(graph -> detach(new InvocableEntityUrlGenerator(graph), graph, status));
++ dataSets.forEach(dataset -> detach(new DataSetEntityUrlGenerator(dataset), dataset, status));
++
++ return verifyPipelineOperationStatus(
++ status,
++ "Successfully stopped pipeline " + pipelineName,
++ "Could not stop all pipeline elements of pipeline " + pipelineName,
++ false);
++ }
++
++ private PipelineOperationStatus verifyPipelineOperationStatus(PipelineOperationStatus status, String successMessage,
++ String errorMessage, boolean rollbackIfFailed) {
status.setSuccess(status.getElementStatus().stream().allMatch(PipelineElementStatus::isSuccess));
if (status.isSuccess()) {
-- status.setTitle("Pipeline " + pipelineName + " successfully started");
++ status.setTitle(successMessage);
} else {
-- LOG.info("Could not start pipeline, initializing rollback...");
-- rollbackInvokedPipelineElements(status);
-- status.setTitle("Could not start pipeline " + pipelineName + ".");
++ if (rollbackIfFailed) {
++ LOG.info("Could not start pipeline, initializing rollback...");
++ rollbackInvokedEntities(status);
++ }
++ status.setTitle(errorMessage);
}
return status;
}
-- private void rollbackInvokedPipelineElements(PipelineOperationStatus status) {
++ private void rollbackInvokedEntities(PipelineOperationStatus status) {
for (PipelineElementStatus s : status.getElementStatus()) {
if (s.isSuccess()) {
-- Optional<InvocableStreamPipesEntity> graph = findGraph(s.getElementId());
-- graph.ifPresent(g -> {
-- LOG.info("Rolling back element " + g.getElementId());
-- new HttpRequestBuilder(g, g.getBelongsTo()).detach();
++ Optional<InvocableStreamPipesEntity> graphs = findGraphs(s.getElementId());
++ graphs.ifPresent(graph -> {
++ LOG.info("Rolling back element " + graph.getElementId());
++ makeHttpRequest(new InvocableEntityUrlGenerator(graph), graph, "detach");
});
}
}
}
-- private Optional<InvocableStreamPipesEntity> findGraph(String elementId) {
-- return graphs.stream().filter(g -> g.getBelongsTo().equals(elementId)).findFirst();
++ private void invoke(EndpointUrlGenerator<?> urlGenerator,
++ NamedStreamPipesEntity namedEntity, PipelineOperationStatus status) {
++ status.addPipelineElementStatus(makeHttpRequest(urlGenerator, namedEntity, "invoke"));
}
-- public PipelineOperationStatus detachGraphs() {
++ private void detach(EndpointUrlGenerator<?> urlGenerator,
++ NamedStreamPipesEntity namedEntity, PipelineOperationStatus status) {
++ status.addPipelineElementStatus(makeHttpRequest(urlGenerator, namedEntity, "detach"));
++ }
++
++ // Helper methods
++
++ private PipelineOperationStatus initPipelineOperationStatus() {
PipelineOperationStatus status = new PipelineOperationStatus();
status.setPipelineId(pipelineId);
status.setPipelineName(pipelineName);
++ return status;
++ }
- graphs.forEach(g -> status.addPipelineElementStatus(new HttpRequestBuilder(g, g.getElementId()).detach()));
- dataSets.forEach(dataSet -> status.addPipelineElementStatus(new HttpRequestBuilder(dataSet, dataSet.getElementId() +
- graphs.forEach(g -> status.addPipelineElementStatus(new HttpRequestBuilder(g, g.getUri()).detach()));
- dataSets.forEach(dataSet -> status.addPipelineElementStatus(new HttpRequestBuilder(dataSet, dataSet.getUri() +
-- "/" +dataSet.getDatasetInvocationId())
-- .detach()));
-- status.setSuccess(status.getElementStatus().stream().allMatch(PipelineElementStatus::isSuccess));
--
-- if (status.isSuccess()) {
-- status.setTitle("Pipeline " + pipelineName + " successfully stopped");
-- } else {
-- status.setTitle("Could not stop all pipeline elements of pipeline " + pipelineName + ".");
++ private PipelineElementStatus makeHttpRequest(EndpointUrlGenerator<?> urlGenerator,
++ NamedStreamPipesEntity namedEntity, String type) {
++ switch (type) {
++ case "invoke":
++ return new HttpRequestBuilder(namedEntity, urlGenerator.generateInvokeEndpoint()).invoke();
++ case "detach":
++ return new HttpRequestBuilder(namedEntity, urlGenerator.generateDetachEndpoint()).detach();
++ default:
++ throw new IllegalArgumentException("Type not known: " + type);
}
++ }
-- return status;
++ private Optional<InvocableStreamPipesEntity> findGraphs(String elementId) {
++ return graphs.stream().filter(i -> i.getBelongsTo().equals(elementId)).findFirst();
++ }
++
++ private boolean allInvocableEntitiesRunning(PipelineOperationStatus status) {
++ return status.getElementStatus().stream().allMatch(PipelineElementStatus::isSuccess);
}
}
diff --cc streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/InvocableEntityUrlGenerator.java
index 94bd5cb,0000000..9fec987
mode 100644,000000..100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/InvocableEntityUrlGenerator.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/InvocableEntityUrlGenerator.java
@@@ -1,142 -1,0 +1,144 @@@
+/*
+ * 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 org.apache.streampipes.config.consul.ConsulSpConfig;
+import org.apache.streampipes.container.util.ConsulUtil;
+import org.apache.streampipes.manager.node.AvailableNodesFetcher;
+import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
+import org.apache.streampipes.model.graph.DataProcessorInvocation;
+import org.apache.streampipes.model.node.NodeInfo;
+
+import java.util.Optional;
+
+public class InvocableEntityUrlGenerator extends EndpointUrlGenerator<InvocableStreamPipesEntity> {
+
+ private static final String COLON = ":";
+ private static final String SLASH = "/";
+ protected static final String SINK_IDENTIFIER = "sec";
+ protected static final String PROCESSOR_IDENTIFIER = "sepa";
+ private static final String DEFAULT_NODE_ID = "default";
+ private static final String PE_PORT_KEY = "SP_PORT";
+ private static final String PE_HOST_KEY = "SP_HOST";
+
+ private static final String NODE_CONTROLLER_ROUTE = "node/container";
+
+ public InvocableEntityUrlGenerator(InvocableStreamPipesEntity pipelineElement) {
+ super(pipelineElement);
+ }
+
+ @Override
- public String generateStartPipelineEndpointUrl() {
++ public String generateInvokeEndpoint() {
+ if (pipelineElement.getDeploymentTargetNodeId() == null ||
+ pipelineElement.getDeploymentTargetNodeId().equals(DEFAULT_NODE_ID)) {
+ // default deployments to primary pipeline element
- return URLPREFIX
- + getHost()
- + SLASH
- + getIdentifier()
- + SLASH
- + pipelineElement.getAppId();
++// return URLPREFIX
++// + getHost()
++// + SLASH
++// + getIdentifier()
++// + SLASH
++// + pipelineElement.getAppId();
++ return defaultHost();
+ } else {
+ // edge deployments to secondary pipeline element
+ return URLPREFIX
+ + getHost()
+ + SLASH
+ + NODE_CONTROLLER_ROUTE
+ + "invoke"
+ + SLASH
+ + getIdentifier()
+ + SLASH
+ + pipelineElement.getAppId();
+ }
+ }
+
+ @Override
- public String generateStopPipelineEndpointUrl() {
++ public String generateDetachEndpoint() {
+ // TODO: handle stop requests
- return generateStartPipelineEndpointUrl()
++ return generateInvokeEndpoint()
+ + SLASH
+ + pipelineElement.getDeploymentRunningInstanceId();
+ }
+
+ private String getHost() {
+ if (pipelineElement.getDeploymentTargetNodeId() == null ||
+ pipelineElement.getDeploymentTargetNodeId().equals(DEFAULT_NODE_ID)) {
+ return defaultHost();
+ }
+ else {
+ if (deploymentTargetNodeRunning()) {
+
+ String route = ConsulSpConfig.SERVICE_ROUTE_PREFIX
+ + pipelineElement.getElementEndpointServiceName()
+ + SLASH
+ + ConsulSpConfig.BASE_PREFIX
+ + SLASH
+ + ConsulSpConfig.SECONDARY_NODE_KEY
+ + SLASH
+ + pipelineElement.getDeploymentTargetNodeId()
+ + SLASH;
+
+ String host = ConsulUtil.getStringValue(route + PE_HOST_KEY);
+ int port = ConsulUtil.getIntValue(route + PE_PORT_KEY);
+
+ // Necessary because secondary pipeline element description is not stored in backend
+ // It uses information from primary pipeline element. Node controller will locally forward
+ // request accordingly, thus fields must be correct.
+ pipelineElement.setElementEndpointHostname(host);
+ pipelineElement.setElementEndpointPort(port);
+ pipelineElement.setBelongsTo(URLPREFIX + host + COLON + port + SLASH + getIdentifier() + SLASH + pipelineElement.getAppId());
+ pipelineElement.setElementId(pipelineElement.getBelongsTo() + SLASH + pipelineElement.getDeploymentRunningInstanceId());
+
+ return pipelineElement.getDeploymentTargetNodeHostname()
+ + COLON
+ + pipelineElement.getDeploymentTargetNodePort();
+ }
+ else {
+ return defaultHost();
+ }
+ }
+ }
+
+ private String defaultHost() {
- return pipelineElement.getElementEndpointHostname()
- + COLON
- + pipelineElement.getElementEndpointPort();
++// return pipelineElement.getElementEndpointHostname()
++// + COLON
++// + pipelineElement.getElementEndpointPort();
++ return pipelineElement.getBelongsTo();
+ }
+
+ private String getIdentifier() {
+ return pipelineElement instanceof DataProcessorInvocation ? PROCESSOR_IDENTIFIER : SINK_IDENTIFIER;
+ }
+
+ private Optional<NodeInfo> getNodeInfo() {
+ return new AvailableNodesFetcher()
+ .fetchNodes()
+ .stream()
+ .filter(node -> node.getNodeControllerId().equals(pipelineElement.getDeploymentTargetNodeId()))
+ .findFirst();
+ }
+
+ private boolean deploymentTargetNodeRunning() {
+ return new AvailableNodesFetcher()
+ .fetchNodes()
+ .stream()
+ .anyMatch(node -> node.getNodeControllerId().equals(pipelineElement.getDeploymentTargetNodeId()));
+ }
+
+}