This is an automated email from the ASF dual-hosted git repository.
riemer pushed a commit to branch STREAMPIPES-319
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
The following commit(s) were added to refs/heads/STREAMPIPES-319 by this push:
new a8143fe [STREAMPIPES-319] Refactor data set invocation to support
service discovery
a8143fe is described below
commit a8143fe36df28fdb0674d813ddf5dd66f9c13a65
Author: Dominik Riemer <[email protected]>
AuthorDate: Fri Jul 16 11:48:50 2021 +0200
[STREAMPIPES-319] Refactor data set invocation to support service discovery
---
streampipes-backend/src/main/resources/shiro.ini | 2 +-
.../constants/GlobalStreamPipesConstants.java | 2 +
.../master/management/SourcesManagement.java | 14 +++---
.../container/master/rest/SourcesResource.java | 9 ++--
.../connect/container/master/util/Utils.java | 4 +-
.../master/management/SourcesManagementTest.java | 2 +-
.../org/apache/streampipes/model/SpDataSet.java | 9 ++++
.../ExtensionsServiceEndpointGenerator.java | 2 +-
.../manager/execution/http/GraphSubmitter.java | 16 +++++--
.../manager/execution/http/PipelineExecutor.java | 50 ++++++++++++++++++----
.../manager/setup/CouchDbInstallationStep.java | 2 +-
.../serializers/json/GsonSerializer.java | 2 +
12 files changed, 83 insertions(+), 31 deletions(-)
diff --git a/streampipes-backend/src/main/resources/shiro.ini
b/streampipes-backend/src/main/resources/shiro.ini
index 3d348a1..8851e0b 100644
--- a/streampipes-backend/src/main/resources/shiro.ini
+++ b/streampipes-backend/src/main/resources/shiro.ini
@@ -68,7 +68,7 @@ securityManager.rememberMeManager.cookie.maxAge = 1000000000
/api/v2/users/*/labeling/* = anon
/api/v2/users/*/labeling = anon
/api/v2/connect/*/master/administration = anon
-/api/v2/connect/*/master/sources/** = anon
+/api/v2/connect/master/sources/** = anon
/api/v2/connect/*/master/resolvable/*/configurations = anon
/api/openapi.json = anon
/api/** = authcBearer, customFilter
diff --git
a/streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/GlobalStreamPipesConstants.java
b/streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/GlobalStreamPipesConstants.java
index b1e2230..c706245 100644
---
a/streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/GlobalStreamPipesConstants.java
+++
b/streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/GlobalStreamPipesConstants.java
@@ -22,4 +22,6 @@ public class GlobalStreamPipesConstants {
public static final String STD_ICON_NAME = "icon.png";
public static final String STD_DOCUMENTATION_NAME = "documentation.md";
+ public static final String CONNECT_MASTER_SOURCES_ENDPOINT =
"/streampipes-backend/api/v2/connect/master/sources/";
+
}
diff --git
a/streampipes-connect-container-master/src/main/java/org/apache/streampipes/connect/container/master/management/SourcesManagement.java
b/streampipes-connect-container-master/src/main/java/org/apache/streampipes/connect/container/master/management/SourcesManagement.java
index bba0366..34f2457 100644
---
a/streampipes-connect-container-master/src/main/java/org/apache/streampipes/connect/container/master/management/SourcesManagement.java
+++
b/streampipes-connect-container-master/src/main/java/org/apache/streampipes/connect/container/master/management/SourcesManagement.java
@@ -19,10 +19,9 @@
package org.apache.streampipes.connect.container.master.management;
import
org.apache.streampipes.commons.exceptions.NoServiceEndpointsAvailableException;
-import org.apache.streampipes.connect.api.exception.AdapterException;
import org.apache.streampipes.connect.adapter.util.TransportFormatGenerator;
+import org.apache.streampipes.connect.api.exception.AdapterException;
import
org.apache.streampipes.connect.container.master.util.AdapterEncryptionService;
-import org.apache.streampipes.connect.container.master.util.Utils;
import org.apache.streampipes.container.html.JSONGenerator;
import org.apache.streampipes.container.html.model.DataSourceDescriptionHtml;
import org.apache.streampipes.container.html.model.Description;
@@ -61,10 +60,10 @@ public class SourcesManagement {
this.workerUrlProvider = new WorkerUrlProvider();
}
- public void addAdapter(String streamId, SpDataSet dataSet, String
username) throws AdapterException, NoServiceEndpointsAvailableException {
+ public void addAdapter(String streamId, SpDataSet dataSet) throws
AdapterException, NoServiceEndpointsAvailableException {
- String newUrl = getAdapterUrl(streamId, username);
+ String newUrl = getAdapterUrl(streamId);
AdapterSetDescription adapterDescription = (AdapterSetDescription)
getAdapterDescriptionById(streamId);
adapterDescription.setDataSet(dataSet);
@@ -85,11 +84,11 @@ public class SourcesManagement {
adapterDescription.setUri(newId);
adapterDescription.setId(newId);
- String newUrl = getAdapterUrl(streamId, username);
+ String newUrl = getAdapterUrl(streamId);
WorkerRestClient.stopSetAdapter(newUrl, adapterDescription);
}
- private String getAdapterUrl(String streamId, String username) throws
NoServiceEndpointsAvailableException {
+ private String getAdapterUrl(String streamId) throws
NoServiceEndpointsAvailableException {
String appId = "";
List<AdapterDescription> adapterDescriptions =
this.adapterStorage.getAllAdapters();
for (AdapterDescription ad : adapterDescriptions) {
@@ -97,9 +96,8 @@ public class SourcesManagement {
appId = ad.getAppId();
}
}
- String workerUrl = workerUrlProvider.getWorkerBaseUrl(appId);
- return Utils.addUserNameToApi(workerUrl, username);
+ return workerUrlProvider.getWorkerBaseUrl(appId);
}
diff --git
a/streampipes-connect-container-master/src/main/java/org/apache/streampipes/connect/container/master/rest/SourcesResource.java
b/streampipes-connect-container-master/src/main/java/org/apache/streampipes/connect/container/master/rest/SourcesResource.java
index 7d5e678..3d6a67a 100644
---
a/streampipes-connect-container-master/src/main/java/org/apache/streampipes/connect/container/master/rest/SourcesResource.java
+++
b/streampipes-connect-container-master/src/main/java/org/apache/streampipes/connect/container/master/rest/SourcesResource.java
@@ -32,7 +32,7 @@ import javax.ws.rs.*;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
-@Path("/v2/connect/{username}/master/sources")
+@Path("/v2/connect/master/sources")
public class SourcesResource extends
AbstractAdapterResource<SourcesManagement> {
private static final Logger LOG =
LoggerFactory.getLogger(SourcesResource.class);
@@ -73,13 +73,12 @@ public class SourcesResource extends
AbstractAdapterResource<SourcesManagement>
@Path("/{streamId}")
@Produces(MediaType.APPLICATION_JSON)
public Response addAdapter(@PathParam("streamId") String elementId,
- @PathParam("username") String username,
SpDataSet dataSet) {
String responseMessage = "Instance of data set " + dataSet.getUri() +
" successfully started";
try {
- managementService.addAdapter(elementId, dataSet, username);
+ managementService.addAdapter(elementId, dataSet);
} catch (AdapterException | NoServiceEndpointsAvailableException e) {
LOG.error("Could not set data set instance: " + dataSet.getUri(),
e);
return ok(Notifications.error("Could not set data set instance: "
+ dataSet.getUri()));
@@ -91,7 +90,9 @@ public class SourcesResource extends
AbstractAdapterResource<SourcesManagement>
@DELETE
@Path("/{streamId}/{runningInstanceId}")
@Produces(MediaType.APPLICATION_JSON)
- public Response detach(@PathParam("streamId") String elementId,
@PathParam("runningInstanceId") String runningInstanceId,
@PathParam("username") String username) {
+ public Response detach(@PathParam("streamId") String elementId,
+ @PathParam("runningInstanceId") String
runningInstanceId,
+ @PathParam("username") String username) {
String responseMessage = "Instance of set id: " + elementId + " with
instance id: "+ runningInstanceId + " successfully started";
// String workerUrl = new Utils().getWorkerUrlById(elementId);
diff --git
a/streampipes-connect-container-master/src/main/java/org/apache/streampipes/connect/container/master/util/Utils.java
b/streampipes-connect-container-master/src/main/java/org/apache/streampipes/connect/container/master/util/Utils.java
index 846944f..3ba2b94 100644
---
a/streampipes-connect-container-master/src/main/java/org/apache/streampipes/connect/container/master/util/Utils.java
+++
b/streampipes-connect-container-master/src/main/java/org/apache/streampipes/connect/container/master/util/Utils.java
@@ -21,8 +21,8 @@ package org.apache.streampipes.connect.container.master.util;
public class Utils {
public static String addUserNameToApi(String url, String userName) {
- //return url;
- return url + "/api/v1/" + userName + "/";
+ return url;
+ //return url + "/api/v1/" + userName + "/";
}
diff --git
a/streampipes-connect-container-master/src/test/java/org/apache/streampipes/connect/container/master/management/SourcesManagementTest.java
b/streampipes-connect-container-master/src/test/java/org/apache/streampipes/connect/container/master/management/SourcesManagementTest.java
index a11fa93..d06ab34 100644
---
a/streampipes-connect-container-master/src/test/java/org/apache/streampipes/connect/container/master/management/SourcesManagementTest.java
+++
b/streampipes-connect-container-master/src/test/java/org/apache/streampipes/connect/container/master/management/SourcesManagementTest.java
@@ -61,7 +61,7 @@ public class SourcesManagementTest {
SourcesManagement sourcesManagement = new
SourcesManagement(adapterStorage);
doNothing().when(WorkerRestClient.class, "invokeSetAdapter",
anyString(), any());
- sourcesManagement.addAdapter(ID, new SpDataSet(), "user");
+ sourcesManagement.addAdapter(ID, new SpDataSet());
verify(adapterStorage, times(1)).getAllAdapters();
verifyStatic(WorkerRestClient.class, times(1));
diff --git
a/streampipes-model/src/main/java/org/apache/streampipes/model/SpDataSet.java
b/streampipes-model/src/main/java/org/apache/streampipes/model/SpDataSet.java
index 66bf4d2..13abd82 100644
---
a/streampipes-model/src/main/java/org/apache/streampipes/model/SpDataSet.java
+++
b/streampipes-model/src/main/java/org/apache/streampipes/model/SpDataSet.java
@@ -32,6 +32,7 @@ public class SpDataSet extends SpDataStream {
private String datasetInvocationId;
private String correspondingPipeline;
+ private String selectedEndpointUrl;
public SpDataSet(String uri, String name, String description, String
iconUrl, List<EventStreamQualityDefinition>
hasEventStreamQualities,
@@ -98,4 +99,12 @@ public class SpDataSet extends SpDataStream {
public void setCorrespondingPipeline(String correspondingPipeline) {
this.correspondingPipeline = correspondingPipeline;
}
+
+ public String getSelectedEndpointUrl() {
+ return selectedEndpointUrl;
+ }
+
+ public void setSelectedEndpointUrl(String selectedEndpointUrl) {
+ this.selectedEndpointUrl = selectedEndpointUrl;
+ }
}
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/endpoint/ExtensionsServiceEndpointGenerator.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/endpoint/ExtensionsServiceEndpointGenerator.java
index 507226d..b3feb3b 100644
---
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/endpoint/ExtensionsServiceEndpointGenerator.java
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/endpoint/ExtensionsServiceEndpointGenerator.java
@@ -64,7 +64,7 @@ public class ExtensionsServiceEndpointGenerator {
if (serviceEndpoints.size() > 0) {
return getServiceEndpoints().get(0);
} else {
- LOG.error("Could not find any service endpoints for appId {}, serviceTag
{}", appId, this.spServiceUrlProvider.getServiceTag(appId));
+ LOG.error("Could not find any service endpoints for appId {}, serviceTag
{}", appId, this.spServiceUrlProvider.getServiceTag(appId).asString());
throw new NoServiceEndpointsAvailableException("Could not find any
matching service endpoints");
}
}
diff --git
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
index 5b3f34c..8074b4d 100644
---
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
@@ -60,7 +60,7 @@ public class GraphSubmitter {
if
(status.getElementStatus().stream().allMatch(PipelineElementStatus::isSuccess))
{
dataSets.forEach(dataSet ->
status.addPipelineElementStatus
- (new HttpRequestBuilder(dataSet,
dataSet.getUri()).invoke()));
+ (performInvocation(dataSet)));
}
status.setSuccess(status.getElementStatus().stream().allMatch(PipelineElementStatus::isSuccess));
@@ -96,9 +96,7 @@ public class GraphSubmitter {
status.setPipelineName(pipelineName);
graphs.forEach(g -> status.addPipelineElementStatus(performDetach(g)));
- dataSets.forEach(dataSet -> status.addPipelineElementStatus(new
HttpRequestBuilder(dataSet, dataSet.getUri() +
- "/" +dataSet.getDatasetInvocationId())
- .detach()));
+ dataSets.forEach(dataSet ->
status.addPipelineElementStatus(performDetach(dataSet)));
status.setSuccess(status.getElementStatus().stream().allMatch(PipelineElementStatus::isSuccess));
if (status.isSuccess()) {
@@ -115,8 +113,18 @@ public class GraphSubmitter {
return new HttpRequestBuilder(entity, endpointUrl).invoke();
}
+ private PipelineElementStatus performInvocation(SpDataSet dataset) {
+ String endpointUrl = dataset.getSelectedEndpointUrl();
+ return new HttpRequestBuilder(dataset, endpointUrl).invoke();
+ }
+
private PipelineElementStatus performDetach(InvocableStreamPipesEntity
entity) {
String endpointUrl = entity.getSelectedEndpointUrl() + "/" +
InstanceIdExtractor.extractId(entity.getElementId());
return new HttpRequestBuilder(entity, endpointUrl).detach();
}
+
+ private PipelineElementStatus performDetach(SpDataSet dataset) {
+ String endpointUrl = dataset.getSelectedEndpointUrl() + "/" +
dataset.getDatasetInvocationId();
+ return new HttpRequestBuilder(dataset, endpointUrl).detach();
+ }
}
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 ad4d430..85553b2 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
@@ -18,6 +18,7 @@
package org.apache.streampipes.manager.execution.http;
+import org.apache.streampipes.commons.constants.GlobalStreamPipesConstants;
import
org.apache.streampipes.commons.exceptions.NoServiceEndpointsAvailableException;
import
org.apache.streampipes.manager.execution.endpoint.ExtensionsServiceEndpointGenerator;
import
org.apache.streampipes.manager.execution.endpoint.ExtensionsServiceEndpointUtils;
@@ -26,6 +27,7 @@ import org.apache.streampipes.manager.secret.SecretProvider;
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.base.NamedStreamPipesEntity;
import org.apache.streampipes.model.graph.DataProcessorInvocation;
import org.apache.streampipes.model.graph.DataSinkInvocation;
import org.apache.streampipes.model.grounding.KafkaTransportProtocol;
@@ -37,12 +39,13 @@ import
org.apache.streampipes.model.pipeline.PipelineHealthStatus;
import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
import org.apache.streampipes.storage.api.IPipelineStorage;
import org.apache.streampipes.storage.management.StorageDispatcher;
+import org.apache.streampipes.svcdiscovery.SpServiceDiscovery;
+import org.apache.streampipes.svcdiscovery.api.model.DefaultSpServiceGroups;
+import org.apache.streampipes.svcdiscovery.api.model.DefaultSpServiceTags;
+import org.apache.streampipes.svcdiscovery.api.model.SpServiceUrlProvider;
import org.lightcouch.DocumentConflictException;
-import java.util.ArrayList;
-import java.util.Date;
-import java.util.List;
-import java.util.UUID;
+import java.util.*;
import java.util.stream.Collectors;
public class PipelineExecutor {
@@ -73,12 +76,23 @@ public class PipelineExecutor {
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());
+ List<SpDataSet> dataSets = pipeline
+ .getStreams()
+ .stream()
+ .filter(s -> s instanceof SpDataSet)
+ .map(s -> new SpDataSet((SpDataSet) s))
+ .collect(Collectors.toList());
+
+ List<NamedStreamPipesEntity> failedServices = new ArrayList<>();
- for (SpDataSet ds : dataSets) {
+ dataSets.forEach(ds -> {
ds.setCorrespondingPipeline(pipeline.getPipelineId());
- }
+ try {
+ ds.setSelectedEndpointUrl(findSelectedEndpoint(ds));
+ } catch (NoServiceEndpointsAvailableException e) {
+ failedServices.add(ds);
+ }
+ });
List<InvocableStreamPipesEntity> graphs = new ArrayList<>();
graphs.addAll(sepas);
@@ -86,7 +100,6 @@ public class PipelineExecutor {
decryptSecrets(graphs);
- List<InvocableStreamPipesEntity> failedServices = new ArrayList<>();
graphs.forEach(g -> {
try {
g.setSelectedEndpointUrl(findSelectedEndpoint(g));
@@ -136,6 +149,25 @@ public class PipelineExecutor {
.getEndpointResourceUrl();
}
+ private String findSelectedEndpoint(SpDataSet ds) throws
NoServiceEndpointsAvailableException {
+ String appId = ds.getAppId() != null ? ds.getAppId() :
ds.getCorrespondingAdapterId();
+ if (ds.isInternallyManaged()) {
+ return getConnectMasterSourcesUrl(ds.getElementId());
+ } else {
+ return new ExtensionsServiceEndpointGenerator(appId,
SpServiceUrlProvider.DATA_SET)
+ .getEndpointResourceUrl();
+ }
+ }
+
+ private String getConnectMasterSourcesUrl(String elementId) throws
NoServiceEndpointsAvailableException {
+ List<String> connectMasterEndpoints =
SpServiceDiscovery.getServiceDiscovery().getServiceEndpoints(DefaultSpServiceGroups.CORE,
true,
Collections.singletonList(DefaultSpServiceTags.CONNECT_MASTER.asString()));
+ if (connectMasterEndpoints.size() > 0) {
+ return connectMasterEndpoints.get(0) +
GlobalStreamPipesConstants.CONNECT_MASTER_SOURCES_ENDPOINT + elementId;
+ } else {
+ throw new NoServiceEndpointsAvailableException("Could not find any
available connect master service endpoint");
+ }
+ }
+
private void updateGroupIds(InvocableStreamPipesEntity entity) {
entity.getInputStreams()
.stream()
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/setup/CouchDbInstallationStep.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/setup/CouchDbInstallationStep.java
index 7c29181..e481c14 100644
---
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/setup/CouchDbInstallationStep.java
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/setup/CouchDbInstallationStep.java
@@ -32,7 +32,7 @@ import java.util.*;
public class CouchDbInstallationStep implements InstallationStep {
private static List<String> initRdfEndpointPorts =
-
Collections.singletonList("8099/api/v1/[email protected]/master/sources/");
+ Collections.singletonList("8099/api/v1/master/sources/");
private static final String initRdfEndpointHost = "http://localhost:";;
private static final String PREPARING_NOTIFICATIONS_TEXT = "Preparing
database " +
diff --git
a/streampipes-serializers-json/src/main/java/org/apache/streampipes/serializers/json/GsonSerializer.java
b/streampipes-serializers-json/src/main/java/org/apache/streampipes/serializers/json/GsonSerializer.java
index 45114a9..bde286b 100644
---
a/streampipes-serializers-json/src/main/java/org/apache/streampipes/serializers/json/GsonSerializer.java
+++
b/streampipes-serializers-json/src/main/java/org/apache/streampipes/serializers/json/GsonSerializer.java
@@ -86,6 +86,8 @@ public class GsonSerializer {
builder.registerTypeAdapterFactory(RuntimeTypeAdapterFactory.of(SpDataStream.class,
"sourceType")
.registerSubtype(SpDataSet.class,
"org.apache.streampipes.model.SpDataSet")
.registerSubtype(SpDataStream.class,
"org.apache.streampipes.model.SpDataStream"));
+
builder.registerTypeAdapterFactory(RuntimeTypeAdapterFactory.of(SpDataSet.class,
"sourceType")
+ .registerSubtype(SpDataSet.class,
"org.apache.streampipes.model.SpDataSet"));
builder.registerTypeAdapterFactory(RuntimeTypeAdapterFactory.of(TransformationRuleDescription.class,
"sourceType")
.registerSubtype(RenameRuleDescription.class,
"org.apache.streampipes.model.RenameRuleDescription")
