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 fc272ecd0577731cdd296e2d1065b5e3f514883c Author: Patrick Wiener <[email protected]> AuthorDate: Mon Apr 12 09:47:31 2021 +0200 add kafka as optional edge node broker --- .../config/backend/BackendConfigKeys.java | 1 + .../config/backend/MessagingSettings.java | 32 ++++++--- .../config/backend/SpEdgeNodeProtocol.java | 3 +- .../streampipes/connect/adapter/Adapter.java | 15 ++-- .../connect/adapter/GroundingService.java | 54 ++++++++------ .../api/DataProcessorPipelineElementResource.java | 18 +++-- .../api/DataSinkPipelineElementResource.java | 11 ++- .../model/node/NodeInfoDescriptionBuilder.java | 16 +++++ .../node/container/ContainerVolumesBuilder.java | 6 +- .../model/node/container/DeploymentContainer.java | 19 ++++- .../model/node/container/DockerContainer.java | 5 +- .../node/container/DockerContainerBuilder.java | 7 +- .../node/controller/config/ConfigKeys.java | 1 + .../controller/config/NodeControllerConfig.java | 9 +++ .../container/DockerExtensionsContainer.java | 20 +++--- .../controller/container/DockerKafkaContainer.java | 40 +++++------ .../container/DockerMosquittoContainer.java | 14 ++-- .../container/DockerZookeeperContainer.java | 14 ++-- .../container/StreamPipesDockerServiceID.java | 18 ++--- .../management/NodeControllerSubmitter.java | 6 +- .../controller/management/node/NodeConstants.java | 1 + .../controller/management/node/NodeManager.java | 39 +++++++--- .../orchestrator/DockerOrchestratorManager.java | 4 +- .../docker/DockerContainerDeclarerSingleton.java | 76 ++++++++++++++++++-- .../orchestrator/docker/utils/DockerUtils.java | 66 ++++++++++++++--- .../pipeline/AbstractPipelineExecutor.java | 20 +++++- .../manager/matching/InvocationGraphBuilder.java | 37 ++++++++-- .../manager/matching/ProtocolSelector.java | 82 ++++++++++++++++------ .../apache/streampipes/vocabulary/StreamPipes.java | 1 + .../messaging-configuration.component.html | 11 +-- .../messaging-configuration.component.ts | 4 ++ .../shared/messaging-settings.model.ts | 2 +- ui/src/app/core-model/gen/streampipes-model.ts | 6 +- 33 files changed, 485 insertions(+), 173 deletions(-) diff --git a/streampipes-config/src/main/java/org/apache/streampipes/config/backend/BackendConfigKeys.java b/streampipes-config/src/main/java/org/apache/streampipes/config/backend/BackendConfigKeys.java index 257427a..80033c1 100644 --- a/streampipes-config/src/main/java/org/apache/streampipes/config/backend/BackendConfigKeys.java +++ b/streampipes-config/src/main/java/org/apache/streampipes/config/backend/BackendConfigKeys.java @@ -51,4 +51,5 @@ public class BackendConfigKeys { public static final String SERVICE_NAME = "SP_SERVICE_NAME"; public static final String PRIORITIZED_PROTOCOL = "SP_PRIORITIZED_PROTOCOL"; + public static final String PRIORITIZED_EDGE_PROTOCOL = "SP_PRIORITIZED_EDGE_PROTOCOL"; } diff --git a/streampipes-config/src/main/java/org/apache/streampipes/config/backend/MessagingSettings.java b/streampipes-config/src/main/java/org/apache/streampipes/config/backend/MessagingSettings.java index 216fec6..c59d31e 100644 --- a/streampipes-config/src/main/java/org/apache/streampipes/config/backend/MessagingSettings.java +++ b/streampipes-config/src/main/java/org/apache/streampipes/config/backend/MessagingSettings.java @@ -29,7 +29,7 @@ public class MessagingSettings { private List<SpDataFormat> prioritizedFormats; private List<SpProtocol> prioritizedProtocols; - private SpEdgeNodeProtocol edgeNodeProtocol; + private List<SpEdgeNodeProtocol> prioritizedEdgeProtocols; public static MessagingSettings fromDefault() { List<SpProtocol> prioritzedProtocolList; @@ -51,22 +51,38 @@ public class MessagingSettings { prioritzedProtocolList = Arrays.asList(SpProtocol.KAFKA, SpProtocol.MQTT, SpProtocol.JMS); } + List<SpEdgeNodeProtocol> prioritizedEdgeProtocolList; + if (System.getenv(BackendConfigKeys.PRIORITIZED_EDGE_PROTOCOL) != null) { + switch (System.getenv(BackendConfigKeys.PRIORITIZED_EDGE_PROTOCOL).toLowerCase()) { + case "mqtt": + prioritizedEdgeProtocolList = Arrays.asList(SpEdgeNodeProtocol.MQTT, SpEdgeNodeProtocol.KAFKA); + break; + case "kafka": + prioritizedEdgeProtocolList = Arrays.asList(SpEdgeNodeProtocol.KAFKA, SpEdgeNodeProtocol.MQTT); + break; + default: + prioritizedEdgeProtocolList = Arrays.asList(SpEdgeNodeProtocol.MQTT, SpEdgeNodeProtocol.KAFKA); + } + } else { + prioritizedEdgeProtocolList = Arrays.asList(SpEdgeNodeProtocol.MQTT, SpEdgeNodeProtocol.KAFKA); + } + return new MessagingSettings( 1638400, 5000012, 20, 2, Arrays.asList(SpDataFormat.JSON, SpDataFormat.CBOR, SpDataFormat.FST, SpDataFormat.SMILE), - prioritzedProtocolList, SpEdgeNodeProtocol.MQTT); + prioritzedProtocolList, prioritizedEdgeProtocolList); } public MessagingSettings(Integer batchSize, Integer messageMaxBytes, Integer lingerMs, Integer acks, List<SpDataFormat> prioritizedFormats, - List<SpProtocol> prioritizedProtocols, SpEdgeNodeProtocol edgeNodeProtocol) { + List<SpProtocol> prioritizedProtocols, List<SpEdgeNodeProtocol> prioritizedEdgeProtocols) { this.batchSize = batchSize; this.messageMaxBytes = messageMaxBytes; this.lingerMs = lingerMs; this.acks = acks; this.prioritizedFormats = prioritizedFormats; this.prioritizedProtocols = prioritizedProtocols; - this.edgeNodeProtocol = edgeNodeProtocol; + this.prioritizedEdgeProtocols = prioritizedEdgeProtocols; } public MessagingSettings() { @@ -121,11 +137,11 @@ public class MessagingSettings { this.prioritizedProtocols = prioritizedProtocols; } - public SpEdgeNodeProtocol getEdgeNodeProtocol() { - return edgeNodeProtocol; + public List<SpEdgeNodeProtocol> getPrioritizedEdgeProtocols() { + return prioritizedEdgeProtocols; } - public void setEdgeNodeProtocol(SpEdgeNodeProtocol edgeNodeProtocol) { - this.edgeNodeProtocol = edgeNodeProtocol; + public void setPrioritizedEdgeProtocols(List<SpEdgeNodeProtocol> prioritizedEdgeProtocols) { + this.prioritizedEdgeProtocols = prioritizedEdgeProtocols; } } diff --git a/streampipes-config/src/main/java/org/apache/streampipes/config/backend/SpEdgeNodeProtocol.java b/streampipes-config/src/main/java/org/apache/streampipes/config/backend/SpEdgeNodeProtocol.java index a6033e0..576675d 100644 --- a/streampipes-config/src/main/java/org/apache/streampipes/config/backend/SpEdgeNodeProtocol.java +++ b/streampipes-config/src/main/java/org/apache/streampipes/config/backend/SpEdgeNodeProtocol.java @@ -19,7 +19,8 @@ package org.apache.streampipes.config.backend; public enum SpEdgeNodeProtocol { - MQTT("MQTT", "org.apache.streampipes.model.grounding.MqttTransportProtocol"); + MQTT("MQTT", "org.apache.streampipes.model.grounding.MqttTransportProtocol"), + KAFKA("KAFKA","org.apache.streampipes.model.grounding.KafkaTransportProtocol"); private final String name; private final String protocolClass; diff --git a/streampipes-connect/src/main/java/org/apache/streampipes/connect/adapter/Adapter.java b/streampipes-connect/src/main/java/org/apache/streampipes/connect/adapter/Adapter.java index a563966..183018e 100644 --- a/streampipes-connect/src/main/java/org/apache/streampipes/connect/adapter/Adapter.java +++ b/streampipes-connect/src/main/java/org/apache/streampipes/connect/adapter/Adapter.java @@ -169,17 +169,22 @@ public abstract class Adapter<T extends AdapterDescription> implements Connector // use edge protocol SpEdgeNodeProtocol edgeNodeProtocol = BackendConfig.INSTANCE .getMessagingSettings() - .getEdgeNodeProtocol(); + .getPrioritizedEdgeProtocols() + .get(0); if (GroundingService.isEdgeProtocol(edgeNodeProtocol, MqttTransportProtocol.class)) { return new SendToMqttAdapterSink(adapterDescription); + } else if (GroundingService.isEdgeProtocol(edgeNodeProtocol, KafkaTransportProtocol.class)) { + return new SendToKafkaAdapterSink(adapterDescription); + } else { + throw new AdapterException("Edge node protocol not supported. " + edgeNodeProtocol); } - throw new AdapterException("Edge node protocol not supported. " + edgeNodeProtocol); - } else { - SpProtocol prioritizedProtocol = - BackendConfig.INSTANCE.getMessagingSettings().getPrioritizedProtocols().get(0); + SpProtocol prioritizedProtocol = BackendConfig.INSTANCE + .getMessagingSettings() + .getPrioritizedProtocols() + .get(0); if (GroundingService.isPrioritized(prioritizedProtocol, JmsTransportProtocol.class)) { return new SendToJmsAdapterSink(adapterDescription); diff --git a/streampipes-connect/src/main/java/org/apache/streampipes/connect/adapter/GroundingService.java b/streampipes-connect/src/main/java/org/apache/streampipes/connect/adapter/GroundingService.java index bc574e2..954cf59 100644 --- a/streampipes-connect/src/main/java/org/apache/streampipes/connect/adapter/GroundingService.java +++ b/streampipes-connect/src/main/java/org/apache/streampipes/connect/adapter/GroundingService.java @@ -22,11 +22,11 @@ import org.apache.streampipes.commons.exceptions.SpRuntimeException; import org.apache.streampipes.config.backend.BackendConfig; import org.apache.streampipes.config.backend.SpEdgeNodeProtocol; import org.apache.streampipes.config.backend.SpProtocol; +import org.apache.streampipes.connect.adapter.exception.AdapterException; +import org.apache.streampipes.connect.adapter.preprocessing.elements.SendToKafkaAdapterSink; +import org.apache.streampipes.connect.adapter.preprocessing.elements.SendToMqttAdapterSink; import org.apache.streampipes.connect.adapter.util.TransportFormatGenerator; -import org.apache.streampipes.container.util.ConsulUtil; -import org.apache.streampipes.model.base.InvocableStreamPipesEntity; import org.apache.streampipes.model.connect.adapter.*; -import org.apache.streampipes.model.connect.worker.ConnectWorkerContainer; import org.apache.streampipes.model.grounding.EventGrounding; import org.apache.streampipes.model.grounding.JmsTransportProtocol; import org.apache.streampipes.model.grounding.KafkaTransportProtocol; @@ -85,7 +85,7 @@ public class GroundingService { return eventGrounding; } - public static EventGrounding createEventGrounding(AdapterDescription adapterDescription) { + public static EventGrounding createEventGrounding(AdapterDescription adapterDescription) throws AdapterException { EventGrounding eventGrounding = new EventGrounding(); @@ -93,30 +93,28 @@ public class GroundingService { TopicDefinition topicDefinition = new SimpleTopicDefinition(topic); if (adapterDescription.getDeploymentTargetNodeId() != null) { - SpEdgeNodeProtocol edgeNodeProtocol = BackendConfig.INSTANCE.getMessagingSettings().getEdgeNodeProtocol(); - if (isEdgeProtocol(edgeNodeProtocol, MqttTransportProtocol.class)) { - - if (adapterDescription instanceof GenericAdapterDescription) { + String nodeControllerId = extractNodeControllerId(adapterDescription); + SpEdgeNodeProtocol edgeNodeProtocol = BackendConfig.INSTANCE + .getMessagingSettings() + .getPrioritizedEdgeProtocols() + .get(0); - String nodeControllerId = ((GenericAdapterDescription) adapterDescription) - .getProtocolDescription() - .getDeploymentTargetNodeId(); - - MqttTransportProtocol brokerTransportProtocol = - (MqttTransportProtocol) getNodeBrokerTransportProtocol(nodeControllerId); - brokerTransportProtocol.setTopicDefinition(topicDefinition); + if (isEdgeProtocol(edgeNodeProtocol, MqttTransportProtocol.class)) { + MqttTransportProtocol brokerTransportProtocol = + (MqttTransportProtocol) getNodeBrokerTransportProtocol(nodeControllerId); + brokerTransportProtocol.setTopicDefinition(topicDefinition); - eventGrounding.setTransportProtocol(brokerTransportProtocol); - } else { + eventGrounding.setTransportProtocol(brokerTransportProtocol); - String nodeControllerId = adapterDescription.getDeploymentTargetNodeId(); - MqttTransportProtocol brokerTransportProtocol = - (MqttTransportProtocol) getNodeBrokerTransportProtocol(nodeControllerId); - brokerTransportProtocol.setTopicDefinition(topicDefinition); + } else if (isEdgeProtocol(edgeNodeProtocol, KafkaTransportProtocol.class)) { + KafkaTransportProtocol brokerTransportProtocol = + (KafkaTransportProtocol) getNodeBrokerTransportProtocol(nodeControllerId); + brokerTransportProtocol.setTopicDefinition(topicDefinition); - eventGrounding.setTransportProtocol(brokerTransportProtocol); - } + eventGrounding.setTransportProtocol(brokerTransportProtocol); + } else { + throw new AdapterException("Edge node protocol not supported. " + edgeNodeProtocol); } } else { @@ -150,6 +148,16 @@ public class GroundingService { return eventGrounding; } + private static String extractNodeControllerId(AdapterDescription adapterDescription) { + if (adapterDescription instanceof GenericAdapterDescription) { + return ((GenericAdapterDescription) adapterDescription) + .getProtocolDescription() + .getDeploymentTargetNodeId(); + } else { + return adapterDescription.getDeploymentTargetNodeId(); + } + } + public static Boolean isPrioritized(SpProtocol prioritizedProtocol, Class<?> protocolClass) { return prioritizedProtocol.getProtocolClass().equals(protocolClass.getCanonicalName()); diff --git a/streampipes-container/src/main/java/org/apache/streampipes/container/api/DataProcessorPipelineElementResource.java b/streampipes-container/src/main/java/org/apache/streampipes/container/api/DataProcessorPipelineElementResource.java index cf65524..d1e4ea6 100644 --- a/streampipes-container/src/main/java/org/apache/streampipes/container/api/DataProcessorPipelineElementResource.java +++ b/streampipes-container/src/main/java/org/apache/streampipes/container/api/DataProcessorPipelineElementResource.java @@ -56,19 +56,27 @@ public class DataProcessorPipelineElementResource extends InvocablePipelineEleme @Override protected DataProcessorInvocation createGroundingDebugInformation(DataProcessorInvocation graph) { + boolean isDefaultGraph = graph.getDeploymentTargetNodeId().equals("default"); graph.getInputStreams().forEach(is -> { - modifyGrounding(is.getEventGrounding()); + modifyGrounding(is.getEventGrounding(), isDefaultGraph); }); - modifyGrounding(graph.getOutputStream().getEventGrounding()); + modifyGrounding(graph.getOutputStream().getEventGrounding(), isDefaultGraph); return graph; } - private void modifyGrounding(EventGrounding grounding) { + private void modifyGrounding(EventGrounding grounding, boolean isCloudBroker) { TransportProtocol protocol = grounding.getTransportProtocol(); protocol.setBrokerHostname("localhost"); - if (protocol instanceof KafkaTransportProtocol) { - ((KafkaTransportProtocol) protocol).setKafkaPort(9094); + if (isCloudBroker) { + if (protocol instanceof KafkaTransportProtocol) { + ((KafkaTransportProtocol) protocol).setKafkaPort(9095); + } + } else { + if (protocol instanceof KafkaTransportProtocol) { + ((KafkaTransportProtocol) protocol).setKafkaPort(9094); + } } + } } diff --git a/streampipes-container/src/main/java/org/apache/streampipes/container/api/DataSinkPipelineElementResource.java b/streampipes-container/src/main/java/org/apache/streampipes/container/api/DataSinkPipelineElementResource.java index 19ae988..72ef4eb 100644 --- a/streampipes-container/src/main/java/org/apache/streampipes/container/api/DataSinkPipelineElementResource.java +++ b/streampipes-container/src/main/java/org/apache/streampipes/container/api/DataSinkPipelineElementResource.java @@ -57,10 +57,15 @@ public class DataSinkPipelineElementResource extends InvocablePipelineElementRes graph.getInputStreams().forEach(is -> { TransportProtocol protocol = is.getEventGrounding().getTransportProtocol(); protocol.setBrokerHostname("localhost"); - if (protocol instanceof KafkaTransportProtocol) { - ((KafkaTransportProtocol) protocol).setKafkaPort(9094); + if (graph.getDeploymentTargetNodeId().equals("default")) { + if (protocol instanceof KafkaTransportProtocol) { + ((KafkaTransportProtocol) protocol).setKafkaPort(9095); + } + } else { + if (protocol instanceof KafkaTransportProtocol) { + ((KafkaTransportProtocol) protocol).setKafkaPort(9094); + } } - }); return graph; diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/node/NodeInfoDescriptionBuilder.java b/streampipes-model/src/main/java/org/apache/streampipes/model/node/NodeInfoDescriptionBuilder.java index e8d0729..d1ba465 100644 --- a/streampipes-model/src/main/java/org/apache/streampipes/model/node/NodeInfoDescriptionBuilder.java +++ b/streampipes-model/src/main/java/org/apache/streampipes/model/node/NodeInfoDescriptionBuilder.java @@ -17,6 +17,7 @@ */ package org.apache.streampipes.model.node; +import org.apache.streampipes.model.grounding.KafkaTransportProtocol; import org.apache.streampipes.model.grounding.MqttTransportProtocol; import org.apache.streampipes.model.grounding.TransportProtocol; import org.apache.streampipes.model.node.container.DeploymentContainer; @@ -87,6 +88,21 @@ public class NodeInfoDescriptionBuilder { return this; } + public NodeInfoDescriptionBuilder withNodeBroker(String host, int port, String protocol) { + if (protocol.toLowerCase().equals("mqtt")) { + MqttTransportProtocol tp = new MqttTransportProtocol(); + tp.setBrokerHostname(host); + tp.setPort(port); + this.nodeBroker.setNodeTransportProtocol(tp); + } else if (protocol.toLowerCase().equals("kafka")) { + KafkaTransportProtocol tp = new KafkaTransportProtocol(); + tp.setBrokerHostname(host); + tp.setKafkaPort(port); + this.nodeBroker.setNodeTransportProtocol(tp); + } + return this; + } + public NodeInfoDescriptionBuilder withNodeBroker(TransportProtocol tp) { this.nodeBroker.setNodeTransportProtocol(tp); return this; diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/node/container/ContainerVolumesBuilder.java b/streampipes-model/src/main/java/org/apache/streampipes/model/node/container/ContainerVolumesBuilder.java index 8733b3e..26a9675 100644 --- a/streampipes-model/src/main/java/org/apache/streampipes/model/node/container/ContainerVolumesBuilder.java +++ b/streampipes-model/src/main/java/org/apache/streampipes/model/node/container/ContainerVolumesBuilder.java @@ -36,11 +36,11 @@ public class ContainerVolumesBuilder { return this; } - public ContainerVolumesBuilder add(String key, String value, boolean readOnly) { + public ContainerVolumesBuilder add(String namedVolume, String containerPath, boolean readOnly) { if (readOnly) { - this.volumes.add(String.format("%s:%s:%s", key, value, "ro")); + this.volumes.add(String.format("%s:%s:%s", namedVolume, containerPath, "ro")); } else { - this.volumes.add(String.format("%s:%s", key, value)); + this.volumes.add(String.format("%s:%s", namedVolume, containerPath)); } return this; } diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/node/container/DeploymentContainer.java b/streampipes-model/src/main/java/org/apache/streampipes/model/node/container/DeploymentContainer.java index c4f1cf4..dad53ad 100644 --- a/streampipes-model/src/main/java/org/apache/streampipes/model/node/container/DeploymentContainer.java +++ b/streampipes-model/src/main/java/org/apache/streampipes/model/node/container/DeploymentContainer.java @@ -65,10 +65,16 @@ public abstract class DeploymentContainer extends UnnamedStreamPipesEntity { @RdfProperty(StreamPipes.DEPLOYMENT_CONTAINER_VOLUMES) private List<String> volumes; + @OneToMany(fetch = FetchType.EAGER, + cascade = {CascadeType.ALL}) + @RdfProperty(StreamPipes.DEPLOYMENT_CONTAINER_DEPENDENCIES) + private List<String> dependsOnContainers; + public DeploymentContainer() { this.envVars = new ArrayList<>(); this.labels = new HashMap<>(); this.volumes = new ArrayList<>(); + this.dependsOnContainers = new ArrayList<>(); } public DeploymentContainer(String elementId) { @@ -76,6 +82,7 @@ public abstract class DeploymentContainer extends UnnamedStreamPipesEntity { this.envVars = new ArrayList<>(); this.labels = new HashMap<>(); this.volumes = new ArrayList<>(); + this.dependsOnContainers = new ArrayList<>(); } public DeploymentContainer(DeploymentContainer other) { @@ -83,7 +90,8 @@ public abstract class DeploymentContainer extends UnnamedStreamPipesEntity { } public DeploymentContainer(String imageUri, String containerName, String serviceId, String[] containerPorts, - List<String> envVars, Map<String, String> labels, List<String> volumes) { + List<String> envVars, Map<String, String> labels, List<String> volumes, + List<String> dependsOnContainers) { this.imageUri = imageUri; this.containerName = containerName; this.serviceId = serviceId; @@ -91,6 +99,7 @@ public abstract class DeploymentContainer extends UnnamedStreamPipesEntity { this.envVars = envVars; this.labels = labels; this.volumes = volumes; + this.dependsOnContainers = dependsOnContainers; } public String getImageUri() { @@ -147,4 +156,12 @@ public abstract class DeploymentContainer extends UnnamedStreamPipesEntity { public void setVolumes(List<String> volumes) { this.volumes = volumes; } + + public List<String> getDependsOnContainers() { + return dependsOnContainers; + } + + public void setDependsOnContainers(List<String> dependsOnContainers) { + this.dependsOnContainers = dependsOnContainers; + } } diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/node/container/DockerContainer.java b/streampipes-model/src/main/java/org/apache/streampipes/model/node/container/DockerContainer.java index b2e6bb7..f66e67e 100644 --- a/streampipes-model/src/main/java/org/apache/streampipes/model/node/container/DockerContainer.java +++ b/streampipes-model/src/main/java/org/apache/streampipes/model/node/container/DockerContainer.java @@ -39,7 +39,8 @@ public class DockerContainer extends DeploymentContainer { } public DockerContainer(String imageURI, String containerName, String serviceId, String[] containerPorts, - List<String> envVars, Map<String, String> labels, List<String> volumes) { - super(imageURI, containerName, serviceId, containerPorts, envVars, labels, volumes); + List<String> envVars, Map<String, String> labels, List<String> volumes, + List<String> dependsOnContainers) { + super(imageURI, containerName, serviceId, containerPorts, envVars, labels, volumes, dependsOnContainers); } } diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/node/container/DockerContainerBuilder.java b/streampipes-model/src/main/java/org/apache/streampipes/model/node/container/DockerContainerBuilder.java index c7d33a7..349df93 100644 --- a/streampipes-model/src/main/java/org/apache/streampipes/model/node/container/DockerContainerBuilder.java +++ b/streampipes-model/src/main/java/org/apache/streampipes/model/node/container/DockerContainerBuilder.java @@ -35,11 +35,16 @@ public class DockerContainerBuilder { return this; } - public DockerContainerBuilder withName(String name) { + public DockerContainerBuilder withContainerName(String name) { this.dockerContainer.setContainerName(name); return this; } + public DockerContainerBuilder dependsOn(String... dependsOnContainers) { + this.dockerContainer.setDependsOnContainers(Arrays.asList(dependsOnContainers)); + return this; + } + public DockerContainerBuilder withExposedPorts(String[] ports) { this.dockerContainer.setContainerPorts(ports); return this; diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/config/ConfigKeys.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/config/ConfigKeys.java index 0dccd71..1375e08 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/config/ConfigKeys.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/config/ConfigKeys.java @@ -47,4 +47,5 @@ public class ConfigKeys { public static final String RESOURCE_UPDATE_FREQ_SECS = "SP_NODE_RESOURCE_UPDATE_FREQ_SECS"; public static final String EVENT_BUFFER_SIZE = "SP_NODE_EVENT_BUFFER_SIZE"; public static final String SP_VERSION = "SP_VERSION"; + public static final String NODE_BROKER_PROTOCOL = "SP_NODE_BROKER_PROTOCOL"; } diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/config/NodeControllerConfig.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/config/NodeControllerConfig.java index 0659259..723e838 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/config/NodeControllerConfig.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/config/NodeControllerConfig.java @@ -40,6 +40,7 @@ public enum NodeControllerConfig { private static final String DEFAULT_NODE_HOST = "host.docker.internal"; private static final String DEFAULT_NODE_CONTROLLER_HOST = "streampipes-node-controller"; private static final String DEFAULT_NODE_BROKER_HOST = "streampipes-node-broker"; + private static final String DEFAULT_NODE_BROKER_PROTOCOL = "mqtt"; private static final String DEFAULT_BACKEND_HOST = "host.docker.internal"; // Default ports: service endpoint ports @@ -211,6 +212,14 @@ public enum NodeControllerConfig { String.class); } + public String getNodeBrokerProtocol() { + return getEnvOrDefault( + ConfigKeys.NODE_BROKER_PROTOCOL, + DEFAULT_NODE_BROKER_PROTOCOL, + String.class + ); + } + public String getApiKey() { String apiKey = getEnvOrDefault( ConfigKeys.SP_API_KEY, diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/DockerExtensionsContainer.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/DockerExtensionsContainer.java index f527307..84a9831 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/DockerExtensionsContainer.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/DockerExtensionsContainer.java @@ -23,22 +23,20 @@ import org.apache.streampipes.node.controller.management.orchestrator.docker.Abs public class DockerExtensionsContainer extends AbstractStreamPipesDockerContainer { - private static final String SP_CONTAINER_ID = "svc/org.apache.streampipes.pe.extensions"; - @Override public DockerContainer declareDockerContainer() { - return DockerContainerBuilder.create(SP_CONTAINER_ID) + return DockerContainerBuilder.create(StreamPipesDockerServiceID.SP_SVC_EXTENSIONS_ID) .withImage("apachestreampipes/extensions-all-jvm:" + getStreamPipesVersion()) - .withName("streampipes-extensions") + .withContainerName("streampipes-extensions") .withExposedPorts(Ports.withMapping("8090")) - .withEnvironmentVariables( - ContainerEnvBuilder.create() - .addNodeEnvs(generateStreamPipesNodeEnvs()) - .add("CONSUL_LOCATION", NodeControllerConfig.INSTANCE.consulLocation()) - .add("SP_HOST", NodeControllerConfig.INSTANCE.getNodeHost()) - .add("SP_PORT", "8090") + .withEnvironmentVariables(ContainerEnvBuilder.create() + .addNodeEnvs(generateStreamPipesNodeEnvs()) + .add("CONSUL_LOCATION", NodeControllerConfig.INSTANCE.consulLocation()) + .add("SP_HOST", NodeControllerConfig.INSTANCE.getNodeHost()) + .add("SP_PORT", "8090") .build()) - .withLabels(ContainerLabels.with(SP_CONTAINER_ID, retrieveNodeType(), ContainerType.EXTENSIONS)) + .withLabels(ContainerLabels.with(StreamPipesDockerServiceID.SP_SVC_EXTENSIONS_ID, retrieveNodeType(), + ContainerType.EXTENSIONS)) .build(); } } diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/DockerKafkaContainer.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/DockerKafkaContainer.java index 781d13c..b52f5e1 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/DockerKafkaContainer.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/DockerKafkaContainer.java @@ -24,32 +24,30 @@ import org.apache.streampipes.node.controller.management.orchestrator.docker.Abs public class DockerKafkaContainer extends AbstractStreamPipesDockerContainer { - private static final String SP_CONTAINER_ID = "svc/org.apache.streampipes.node.broker.kafka"; - @Override public DockerContainer declareDockerContainer() { - return DockerContainerBuilder.create(SP_CONTAINER_ID) + return DockerContainerBuilder.create(StreamPipesDockerServiceID.SP_SVC_KAFKA_ID) .withImage("fogsyio/kafka:2.2.0") - .withName("streampipes-node-broker") + .withContainerName("streampipes-node-broker") + .dependsOn("svc/org.apache.streampipes.node.broker.zookeeper") .withExposedPorts(Ports.withMapping("9094")) - .withEnvironmentVariables( - ContainerEnvBuilder.create() - .addNodeEnvs(generateStreamPipesNodeEnvs()) - .add("KAFKA_LISTENER_SECURITY_PROTOCOL_MAP", "PLAINTEXT:PLAINTEXT,OUTSIDE:PLAINTEXT") - .add("KAFKA_ADVERTISED_LISTENERS", - "PLAINTEXT://:9092,OUTSIDE://" + NodeControllerConfig.INSTANCE.getNodeHost() + ":9094") - .add("KAFKA_LISTENERS", "PLAINTEXT://:9092,OUTSIDE://:9094") - .add("KAFKA_INTER_BROKER_LISTENER_NAME", "PLAINTEXT") - .add("KAFKA_ADVERTISED_HOST_NAME", "kafka") - .add("KAFKA_ZOOKEEPER_CONNECT", "zookeeper:2181") - .add("KAFKA_MESSAGE_MAX_BYTES", "5000012") - .add("KAFKA_FETCH_MESSAGE_MAX_BYTES", "5000012") - .add("KAFKA_REPLICA_FETCH_MAX_BYTES", "10000000") + .withEnvironmentVariables(ContainerEnvBuilder.create() + .addNodeEnvs(generateStreamPipesNodeEnvs()) + .add("KAFKA_LISTENER_SECURITY_PROTOCOL_MAP", "PLAINTEXT:PLAINTEXT,OUTSIDE:PLAINTEXT") + .add("KAFKA_ADVERTISED_LISTENERS", + "PLAINTEXT://:9092,OUTSIDE://" + NodeControllerConfig.INSTANCE.getNodeHost() + ":9094") + .add("KAFKA_LISTENERS", "PLAINTEXT://:9092,OUTSIDE://:9094") + .add("KAFKA_INTER_BROKER_LISTENER_NAME", "PLAINTEXT") + .add("KAFKA_ADVERTISED_HOST_NAME", "streampipes-node-broker") + .add("KAFKA_ZOOKEEPER_CONNECT", "streampipes-node-zookeeper:2181") + .add("KAFKA_MESSAGE_MAX_BYTES", "5000012") + .add("KAFKA_FETCH_MESSAGE_MAX_BYTES", "5000012") + .add("KAFKA_REPLICA_FETCH_MAX_BYTES", "10000000") .build()) - .withLabels(ContainerLabels.with(SP_CONTAINER_ID, retrieveNodeType(), ContainerType.BROKER)) - .withVolumes( - ContainerVolumesBuilder.create() - .add("/var/lib/streampipes/kafka", "/kafka", false) + .withLabels(ContainerLabels.with(StreamPipesDockerServiceID.SP_SVC_KAFKA_ID, retrieveNodeType(), + ContainerType.BROKER)) + .withVolumes(ContainerVolumesBuilder.create() + .add("streampipes-kafka-vol", "/kafka", false) .build()) .build(); } diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/DockerMosquittoContainer.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/DockerMosquittoContainer.java index e869a9f..b00eb30 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/DockerMosquittoContainer.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/DockerMosquittoContainer.java @@ -22,19 +22,17 @@ import org.apache.streampipes.node.controller.management.orchestrator.docker.Abs public class DockerMosquittoContainer extends AbstractStreamPipesDockerContainer { - private static final String SP_CONTAINER_ID = "svc/org.apache.streampipes.node.broker.mosquitto"; - @Override public DockerContainer declareDockerContainer() { - return DockerContainerBuilder.create(SP_CONTAINER_ID) + return DockerContainerBuilder.create(StreamPipesDockerServiceID.SP_SVC_MOSQUITTO_ID) .withImage("eclipse-mosquitto:1.6.12") - .withName("streampipes-node-broker") + .withContainerName("streampipes-node-broker") .withExposedPorts(Ports.withMapping("1883")) - .withEnvironmentVariables( - ContainerEnvBuilder.create() - .addNodeEnvs(generateStreamPipesNodeEnvs()) + .withEnvironmentVariables(ContainerEnvBuilder.create() + .addNodeEnvs(generateStreamPipesNodeEnvs()) .build()) - .withLabels(ContainerLabels.with(SP_CONTAINER_ID, retrieveNodeType(), ContainerType.BROKER)) + .withLabels(ContainerLabels.with(StreamPipesDockerServiceID.SP_SVC_MOSQUITTO_ID, retrieveNodeType(), + ContainerType.BROKER)) .build(); } diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/DockerZookeeperContainer.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/DockerZookeeperContainer.java index 76382a4..b83407a 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/DockerZookeeperContainer.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/DockerZookeeperContainer.java @@ -22,19 +22,17 @@ import org.apache.streampipes.node.controller.management.orchestrator.docker.Abs public class DockerZookeeperContainer extends AbstractStreamPipesDockerContainer { - private static final String SP_CONTAINER_ID = "svc/org.apache.streampipes.node.broker.kafka.zookeeper"; - @Override public DockerContainer declareDockerContainer() { - return DockerContainerBuilder.create(SP_CONTAINER_ID) + return DockerContainerBuilder.create(StreamPipesDockerServiceID.SP_SVC_ZOOKEEPER_ID) .withImage("fogsyio/zookeeper:3.4.13") - .withName("streampipes-node-zookeeper") + .withContainerName("streampipes-node-zookeeper") .withExposedPorts(Ports.withMapping("2181")) - .withEnvironmentVariables( - ContainerEnvBuilder.create() - .addNodeEnvs(generateStreamPipesNodeEnvs()) + .withEnvironmentVariables(ContainerEnvBuilder.create() + .addNodeEnvs(generateStreamPipesNodeEnvs()) .build()) - .withLabels(ContainerLabels.with(SP_CONTAINER_ID, retrieveNodeType(), ContainerType.BROKER)) + .withLabels(ContainerLabels.with(StreamPipesDockerServiceID.SP_SVC_ZOOKEEPER_ID, retrieveNodeType(), + ContainerType.BROKER)) .build(); } } diff --git a/ui/src/app/configuration/shared/messaging-settings.model.ts b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/StreamPipesDockerServiceID.java similarity index 62% copy from ui/src/app/configuration/shared/messaging-settings.model.ts copy to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/StreamPipesDockerServiceID.java index a33ce26..a6bc540 100644 --- a/ui/src/app/configuration/shared/messaging-settings.model.ts +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/StreamPipesDockerServiceID.java @@ -1,4 +1,3 @@ - /* * Licensed to the Apache Software Foundation (ASF) under one or more * contributor license agreements. See the NOTICE file distributed with @@ -16,14 +15,11 @@ * limitations under the License. * */ +package org.apache.streampipes.node.controller.container; -export interface MessagingSettings { - batchSize: number; - messageMaxBytes: number; - lingerMs: number; - acks: number; - - prioritizedFormats: [string]; - prioritizedProtocols: [string]; - edgeNodeProtocol: string; -} \ No newline at end of file +public class StreamPipesDockerServiceID { + public static String SP_SVC_EXTENSIONS_ID = "svc/org.apache.streampipes.pe.extensions"; + public static String SP_SVC_MOSQUITTO_ID = "svc/org.apache.streampipes.node.broker.mosquitto"; + public static String SP_SVC_KAFKA_ID = "svc/org.apache.streampipes.node.broker.kafka"; + public static String SP_SVC_ZOOKEEPER_ID = "svc/org.apache.streampipes.node.broker.zookeeper"; +} diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/NodeControllerSubmitter.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/NodeControllerSubmitter.java index 5c2ff0c..7399ab9 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/NodeControllerSubmitter.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/NodeControllerSubmitter.java @@ -63,7 +63,7 @@ public abstract class NodeControllerSubmitter { LOG.info("Load node info description"); NodeManager.getInstance().init(); - LOG.info("Register node controller at node management"); + LOG.info("Register node controller at StreamPipes cluster management"); boolean success = NodeManager.getInstance().register(); if (success) { @@ -72,14 +72,14 @@ public abstract class NodeControllerSubmitter { if (!"true".equals(System.getenv("SP_DEBUG"))) { - LOG.info("Register auto-deploy containers"); + LOG.info("Register container descriptions"); DockerContainerDeclarerSingleton.getInstance() .register(new DockerExtensionsContainer()) .register(new DockerMosquittoContainer()) .register(new DockerKafkaContainer()) .register(new DockerZookeeperContainer()); - LOG.info("Auto-deploy extensions container"); + LOG.info("Auto-deploy extensions and selected broker container"); DockerOrchestratorManager.getInstance().init(); LOG.info("Start janitor manager"); diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/node/NodeConstants.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/node/NodeConstants.java index 19b5985..3a1d00b 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/node/NodeConstants.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/node/NodeConstants.java @@ -58,6 +58,7 @@ public class NodeConstants { public static final int NODE_PORT = NodeControllerConfig.INSTANCE.getNodeControllerPort(); public static final String NODE_BROKER_HOST = NodeControllerConfig.INSTANCE.getNodeBrokerHost(); public static final int NODE_BROKER_PORT = NodeControllerConfig.INSTANCE.getNodeBrokerPort(); + public static final String NODE_BROKER_PROTOCOL = NodeControllerConfig.INSTANCE.getNodeBrokerProtocol(); public static final String NODE_TYPE = NodeControllerConfig.INSTANCE.getNodeType(); public static final List<String> NODE_LOCATION_TAGS = NodeControllerConfig.INSTANCE.getNodeLocations(); public static final List<String> SUPPORTED_PIPELINE_ELEMENTS = diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/node/NodeManager.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/node/NodeManager.java index 4025d0c..e8f39f9 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/node/NodeManager.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/node/NodeManager.java @@ -35,6 +35,8 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.IOException; +import java.net.InetSocketAddress; +import java.net.Socket; import java.util.List; public class NodeManager { @@ -75,7 +77,10 @@ public class NodeManager { NodeInfoDescriptionBuilder.create(NodeConstants.NODE_CONTROLLER_ID) .withHostname(NodeConstants.NODE_HOSTNAME) .withPort(NodeConstants.NODE_PORT) - .withNodeBroker(NodeConstants.NODE_BROKER_HOST, NodeConstants.NODE_BROKER_PORT) + .withNodeBroker( + NodeConstants.NODE_BROKER_HOST, + NodeConstants.NODE_BROKER_PORT, + NodeConstants.NODE_BROKER_PROTOCOL) .staticNodeMetadata( NodeConstants.NODE_TYPE, NodeConstants.NODE_MODEL, @@ -140,14 +145,15 @@ public class NodeManager { public boolean register() { boolean connected = false; + boolean registered = false; try { String body = JacksonSerializer.getObjectMapper().writeValueAsString(this.nodeInfo); String endpoint = generateRegistrationEndpoint(); - LOG.info("Trying to register node controller at node management: " + endpoint); - while (!connected) { - connected = post(endpoint, body); + LOG.info("Trying to register node controller at StreamPipes cluster management: " + endpoint); + connected = isReady(NodeControllerConfig.INSTANCE.backendLocation(), + NodeControllerConfig.INSTANCE.backendPort()); if (!connected) { LOG.info("Retrying in {} seconds", (RETRY_INTERVAL_MS / 1000)); try { @@ -157,11 +163,16 @@ public class NodeManager { } } } - LOG.info("Successfully registered node controller at node management"); + + registered = post(endpoint, body); + if (registered) { + LOG.info("Successfully registered node controller at StreamPipes cluster management"); + } + } catch (IOException e) { e.printStackTrace(); } - return connected; + return registered; } @@ -171,7 +182,7 @@ public class NodeManager { String body = JacksonSerializer.getObjectMapper().writeValueAsString(this.nodeInfo); String endpoint = generateSyncronizationEndpoint(); - LOG.info("Trying to sync node updates with node management: " + endpoint); + LOG.info("Trying to sync node updates with StreamPipes cluster management: " + endpoint); while (!connected) { connected = post(endpoint, body); @@ -184,7 +195,7 @@ public class NodeManager { } } } - LOG.info("Successfully synced node controller with node management"); + LOG.info("Successfully synced node controller with StreamPipes cluster management"); } catch (IOException e) { e.printStackTrace(); } @@ -241,6 +252,18 @@ public class NodeManager { // Helpers + private static boolean isReady(String host, int port) { + try { + InetSocketAddress sa = new InetSocketAddress(host, port); + Socket ss = new Socket(); + ss.connect(sa, 1000); + ss.close(); + } catch(Exception e) { + return false; + } + return true; + } + private boolean registered(DockerContainer container) { return this.nodeInfo.getRegisteredContainers().contains(container); } diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/orchestrator/DockerOrchestratorManager.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/orchestrator/DockerOrchestratorManager.java index 2dec572..e0e8b8d 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/orchestrator/DockerOrchestratorManager.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/orchestrator/DockerOrchestratorManager.java @@ -24,6 +24,8 @@ import com.spotify.docker.client.messages.Container; import org.apache.commons.lang.StringUtils; import org.apache.streampipes.container.util.ConsulUtil; import org.apache.streampipes.model.node.container.DockerContainer; +import org.apache.streampipes.node.controller.config.ConfigKeys; +import org.apache.streampipes.node.controller.config.NodeControllerConfig; import org.apache.streampipes.node.controller.management.node.NodeManager; import org.apache.streampipes.node.controller.management.orchestrator.docker.DockerContainerDeclarerSingleton; import org.apache.streampipes.node.controller.management.orchestrator.docker.model.ContainerStatus; @@ -57,7 +59,7 @@ public class DockerOrchestratorManager implements IContainerOrchestrator { @Override public void init() { - DockerContainerDeclarerSingleton.getInstance().getAllDockerContainerAsList().forEach(container -> { + DockerContainerDeclarerSingleton.getInstance().getAutoDeploymentDockerContainers().forEach(container -> { ContainerDeploymentStatus status = deploy(container); if (status.getStatus() == ContainerStatus.DEPLOYED) { diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/orchestrator/docker/DockerContainerDeclarerSingleton.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/orchestrator/docker/DockerContainerDeclarerSingleton.java index ad17d07..7338d50 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/orchestrator/docker/DockerContainerDeclarerSingleton.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/orchestrator/docker/DockerContainerDeclarerSingleton.java @@ -19,13 +19,12 @@ package org.apache.streampipes.node.controller.management.orchestrator.docker; import org.apache.streampipes.model.node.container.DockerContainer; +import org.apache.streampipes.node.controller.config.ConfigKeys; +import org.apache.streampipes.node.controller.container.StreamPipesDockerServiceID; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; -import java.util.Map; +import java.util.*; public class DockerContainerDeclarerSingleton { private static final Logger LOG = @@ -60,4 +59,73 @@ public class DockerContainerDeclarerSingleton { public List<DockerContainer> getAllDockerContainerAsList() { return new ArrayList<>(dockerContainers.values()); } + + public List<DockerContainer> getAutoDeploymentDockerContainers() { + if ("kafka".equals(System.getenv(ConfigKeys.NODE_BROKER_PROTOCOL))) { + remove(StreamPipesDockerServiceID.SP_SVC_MOSQUITTO_ID); + } else { + remove(StreamPipesDockerServiceID.SP_SVC_KAFKA_ID, StreamPipesDockerServiceID.SP_SVC_ZOOKEEPER_ID); + } + LinkedHashMap<String, DockerContainer> sorted = sort(); + return new ArrayList<>(sorted.values()); + } + + // Helpers + + private LinkedHashMap<String, DockerContainer> sort() { + Map<String, List<String>> dependencyGraph = new HashMap<>(); + + // initialize + dockerContainers.keySet().forEach(k -> dependencyGraph.put(k, new ArrayList<>())); + + // add dependencies + dockerContainers.values().forEach(container -> { + if (container.getDependsOnContainers().size() > 0) { + container.getDependsOnContainers().forEach(d -> { + dependencyGraph.get(d).add(container.getServiceId()); + }); + } + }); + + LinkedHashMap<String, DockerContainer> sortedContainerList = new LinkedHashMap<>(); + for (String container: topologicalSortBFS(dependencyGraph)) { + sortedContainerList.put(container, dockerContainers.get(container)); + } + + return sortedContainerList; + } + + private List<String> topologicalSortBFS(Map<String, List<String>> graph){ + Map<String, Integer> indegree = new HashMap<>(); + for(String k : graph.keySet()){ + if(!indegree.containsKey(k)) + indegree.put(k, 0); + for(String d : graph.get(k)) + indegree.put(d, 1 + indegree.getOrDefault(d, 0)); + } + + Queue<String> q = new LinkedList<>(); + + for(String v : indegree.keySet()) { + if(indegree.get(v)==0) + q.add(v); + } + + List<String> topologicalSort = new ArrayList<>(); + while(!q.isEmpty()){ + String vertex = q.remove(); + topologicalSort.add(vertex); + for(String nextVertex : graph.get(vertex)){ + indegree.put(nextVertex, indegree.get(nextVertex)-1); + if(indegree.get(nextVertex)==0) + q.add(nextVertex); + } + } + return topologicalSort; + } + + private void remove(String... keys) { + Set<String> filterPredicate = new HashSet<>(Arrays.asList(keys)); + dockerContainers.keySet().removeAll(filterPredicate); + } } diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/orchestrator/docker/utils/DockerUtils.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/orchestrator/docker/utils/DockerUtils.java index f57dd98..86ca42c 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/orchestrator/docker/utils/DockerUtils.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/orchestrator/docker/utils/DockerUtils.java @@ -25,6 +25,9 @@ import com.spotify.docker.client.exceptions.DockerCertificateException; import com.spotify.docker.client.exceptions.DockerException; import com.spotify.docker.client.messages.*; import com.spotify.docker.client.shaded.com.google.common.collect.ImmutableList; +import com.spotify.docker.client.shaded.com.google.common.collect.ImmutableMap; +import com.spotify.docker.client.shaded.com.google.common.collect.Lists; +import org.apache.streampipes.commons.exceptions.SpRuntimeException; import org.apache.streampipes.model.node.container.DockerContainer; import org.apache.streampipes.node.controller.management.orchestrator.docker.model.DockerInfo; import org.slf4j.Logger; @@ -112,26 +115,54 @@ public class DockerUtils { .get() .id(); } catch (DockerException | InterruptedException e) { - LOG.error("Unable to list containers {}", e.toString()); + throw new SpRuntimeException("Unable to list containers. " + e.getMessage(), e); } - return ""; } - public String createContainer(DockerContainer p) { LOG.info("Create pipeline element container {}", p.getContainerName()); try { return docker.createContainer(getContainerConfig(p), verifyContainerName(p.getContainerName())).id(); } catch (DockerException | InterruptedException e) { - LOG.error("Pipeline element container could not be created. {}", e.toString()); + throw new SpRuntimeException("Pipeline element container could not be created. " + e.getMessage(), e); } - return ""; } private String verifyContainerName(String containerName) { return containerName.startsWith(SP_CONTAINER_PREFIX) ? containerName : SP_CONTAINER_PREFIX + containerName; } + public void createVolume() throws DockerException, InterruptedException { + + docker.pull("busybox:latest"); + LOG.info("Create Docker volume"); + Volume myVolume = Volume.builder() + .name("my-first-vol") + .build(); + + HostConfig.Bind bindUsingVolume = HostConfig.Bind.from(myVolume) + .to("/app") + .build(); + + HostConfig hostConfig = HostConfig.builder() + .appendBinds(bindUsingVolume) + .build(); + + ContainerConfig volumeConfig = ContainerConfig.builder() + .image("busybox:latest") + .hostConfig(hostConfig) + .build(); + + //docker.createVolume(myVolume); + String id = docker.createContainer(volumeConfig, "my-busybox-container").id(); + LOG.info("containerId: " + id); + } + + public void removeVolume() throws DockerException, InterruptedException { + LOG.info("Remove Docker volume"); + docker.removeVolume("my-first-vol"); + } + private ContainerConfig getContainerConfig(DockerContainer p) { return ContainerConfig.builder() .hostname(p.getContainerName()) @@ -140,6 +171,7 @@ public class DockerUtils { .labels(p.getLabels()) .env(p.getEnvVars()) .hostConfig(getHostConfig(SP_CONTAINER_NETWORK, p.getContainerPorts(), p.getVolumes())) + .exposedPorts(modifyExposedPorts(p.getContainerPorts())) .networkingConfig(getNetworkingConfig(SP_CONTAINER_NETWORK, p.getContainerName())) .build(); } @@ -185,16 +217,25 @@ public class DockerUtils { Map<String, List<PortBinding>> portBindings = new HashMap<>(); if (ports != null) { for (String port : ports) { - portBindings.put(port + "/tcp", Collections.singletonList(PortBinding.of("0.0.0.0", port))); + portBindings.put(port + "/tcp", Lists.newArrayList(PortBinding.of("0.0.0.0", port))); } } - if (volumes != null) { + if (Objects.requireNonNull(volumes).size() > 0) { + List<HostConfig.Bind> allVolumeBinds = new ArrayList<>(); + for (String volume: volumes) { + String hostVol = volume.split(":")[0]; + String toVol = volume.split(":")[1]; + + Volume vol = Volume.builder().name(hostVol).build(); + allVolumeBinds.add(HostConfig.Bind.from(vol).to(toVol).build()); + } + return HostConfig.builder() .portBindings(portBindings) .networkMode(network) .restartPolicy(HostConfig.RestartPolicy.unlessStopped()) - .appendBinds(volumes) + .appendBinds(allVolumeBinds.toArray(new HostConfig.Bind[0])) .build(); } else { return HostConfig.builder() @@ -210,7 +251,6 @@ public class DockerUtils { .create(new HashMap<String, EndpointConfig>() {{ put(network, getEndpointConfig(containerName));}} ); - } private static EndpointConfig getEndpointConfig(String containerName) { @@ -334,4 +374,12 @@ public class DockerUtils { return hasNvidiaRuntime; } + + private String[] modifyExposedPorts(String[] containerPorts) { + List<String> modifyPorts = new ArrayList<>(); + for (String port: containerPorts) { + modifyPorts.add(port + "/tcp"); + } + return modifyPorts.toArray(new String[0]); + } } diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/AbstractPipelineExecutor.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/AbstractPipelineExecutor.java index e471c04..2bf4f00 100644 --- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/AbstractPipelineExecutor.java +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/AbstractPipelineExecutor.java @@ -222,6 +222,8 @@ public abstract class AbstractPipelineExecutor { .collect(Collectors.toList()); } + // TODO: when using kafka as edge protocol it generates duplicate event relays -> check with mqtt as edge + // protocol and fix private List<SpDataStreamRelayContainer> generateDataStreamRelays(List<InvocableStreamPipesEntity> graphs) { List<SpDataStreamRelayContainer> relays = new ArrayList<>(); @@ -242,8 +244,10 @@ public abstract class AbstractPipelineExecutor { } for (DataProcessorInvocation processor : pipeline.getSepas()) { if (differentDeploymentTargets(processor, graph) && connected(processor, graph)) { - SpDataStreamRelayContainer processorRelay = new SpDataStreamRelayContainer(processor); - relays.add(processorRelay); + if (!relayExists(relays, processor.getDeploymentRunningInstanceId())) { + SpDataStreamRelayContainer processorRelay = new SpDataStreamRelayContainer(processor); + relays.add(processorRelay); + } } } } @@ -254,6 +258,18 @@ public abstract class AbstractPipelineExecutor { // Helpers /** + * Check if relay with deploymentRunningInstanceId of predecessor already exists + * + * @param relays List of existing relays + * @param deploymentRunningInstanceId Id to check + * @return boolean + */ + private boolean relayExists(List<SpDataStreamRelayContainer> relays, + String deploymentRunningInstanceId) { + return relays.stream().anyMatch(r -> r.getRunningStreamRelayInstanceId().equals(deploymentRunningInstanceId)); + } + + /** * Updates group.id for data processor/sink. Note: KafkaTransportProtocol only!! * * @param entity data processor/sink 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 533b8a2..8c73766 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 @@ -174,7 +174,8 @@ public class InvocationGraphBuilder { return t.getDeploymentTargetNodeId() != null && t.getDeploymentTargetNodeId().equals(DEFAULT_TAG); } - private void connectSourceToTarget(DataProcessorInvocation source, InvocableStreamPipesEntity target, EventGrounding inputGrounding, boolean edgeToEdgeRelay){ + private void connectSourceToTarget(DataProcessorInvocation source, InvocableStreamPipesEntity target, + EventGrounding inputGrounding, boolean edgeToEdgeRelay){ //connect a source entity with a target entity through an appropriate relay EventGrounding relayGrounding = edgeToEdgeRelay ? generateRelayGrounding(inputGrounding, target, true) : generateRelayGrounding(inputGrounding, false); @@ -182,7 +183,16 @@ public class InvocationGraphBuilder { removeExistingStreamRelay(source, target); } modifyTargetInputStream(source, target, relayGrounding, edgeToEdgeRelay); - source.addOutputStreamRelay(new SpDataStreamRelay(relayGrounding)); + addOutputStreamRelayIfNoDuplicate(source, relayGrounding); + } + + private void addOutputStreamRelayIfNoDuplicate(DataProcessorInvocation source, EventGrounding relayGrounding) { + boolean isDuplicate = source.getOutputStreamRelays().stream() + .anyMatch(r -> r.getEventGrounding().getElementId().equals(relayGrounding.getElementId())); + + if (!isDuplicate) { + source.addOutputStreamRelay(new SpDataStreamRelay(relayGrounding)); + } } private void modifyTargetInputStream(NamedStreamPipesEntity s, InvocableStreamPipesEntity t, @@ -201,7 +211,6 @@ public class InvocationGraphBuilder { } } - private boolean deploymentTargetNotNull(NamedStreamPipesEntity s, InvocableStreamPipesEntity t) { if (s instanceof SpDataStream) { return ((SpDataStream) s).getDeploymentTargetNodeId() != null && t.getDeploymentTargetNodeId() != null; @@ -372,8 +381,8 @@ public class InvocationGraphBuilder { EventGrounding eg = new EventGrounding(); String topic = extractTopic(sourceInvocableOutputGrounding); if (edgeToEdgeRelay) { - eg.setTransportProtocol(getTargetNodeProtocol( - BackendConfig.INSTANCE.getMessagingSettings().getEdgeNodeProtocol(), topic, target)); + eg.setTransportProtocol(getPrioritizedEdgeProtocol( + BackendConfig.INSTANCE.getMessagingSettings().getPrioritizedEdgeProtocols().get(0), topic, target)); } else { eg.setTransportProtocol(getPrioritizedGlobalProtocol( BackendConfig.INSTANCE.getMessagingSettings().getPrioritizedProtocols().get(0), topic)); @@ -394,10 +403,12 @@ public class InvocationGraphBuilder { throw new SpRuntimeException("Could not retrieve prioritized transport protocol"); } - private TransportProtocol getTargetNodeProtocol(SpEdgeNodeProtocol p, String topic, - InvocableStreamPipesEntity target) { + private TransportProtocol getPrioritizedEdgeProtocol(SpEdgeNodeProtocol p, String topic, + InvocableStreamPipesEntity target) { if (matches(p, MqttTransportProtocol.class)){ return mqttTransportProtocol(topic, target); + } else if (matches(p, KafkaTransportProtocol.class)){ + return kafkaTransportProtocol(topic, target); } throw new SpRuntimeException("Could not retrieve prioritized transport protocol"); } @@ -410,6 +421,18 @@ public class InvocationGraphBuilder { } private KafkaTransportProtocol kafkaTransportProtocol(String topic) { + return kafkaTransportProtocol(topic, null); + } + + private KafkaTransportProtocol kafkaTransportProtocol(String topic, InvocableStreamPipesEntity target) { + if (target != null) { + return new KafkaTransportProtocol( + getTargetNodeBrokerHost(target), + getTargetNodeBrokerPort(target), + topic); +// getTargetNodeZookeeperHost(target), +// getTargetNodeZookeeperPort(target)); + } return new KafkaTransportProtocol( BackendConfig.INSTANCE.getKafkaHost(), BackendConfig.INSTANCE.getKafkaPort(), 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 a869e30..ee90a00 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 @@ -22,7 +22,6 @@ import org.apache.streampipes.commons.exceptions.SpRuntimeException; import org.apache.streampipes.config.backend.BackendConfig; import org.apache.streampipes.config.backend.SpEdgeNodeProtocol; import org.apache.streampipes.config.backend.SpProtocol; -import org.apache.streampipes.container.util.ConsulUtil; import org.apache.streampipes.manager.util.TopicGenerator; import org.apache.streampipes.model.base.InvocableStreamPipesEntity; import org.apache.streampipes.model.base.NamedStreamPipesEntity; @@ -40,7 +39,7 @@ public class ProtocolSelector extends GroundingSelector { private final String outputTopic; private final List<SpProtocol> prioritizedProtocols; - private final SpEdgeNodeProtocol edgeNodeProtocol; + private final List<SpEdgeNodeProtocol> prioritizedEdgeProtocols; public ProtocolSelector(NamedStreamPipesEntity source, Set<InvocableStreamPipesEntity> targets) { super(source, targets); @@ -61,9 +60,9 @@ public class ProtocolSelector extends GroundingSelector { this.prioritizedProtocols = BackendConfig.INSTANCE .getMessagingSettings() .getPrioritizedProtocols(); - this.edgeNodeProtocol = BackendConfig.INSTANCE + this.prioritizedEdgeProtocols = BackendConfig.INSTANCE .getMessagingSettings() - .getEdgeNodeProtocol(); + .getPrioritizedEdgeProtocols(); } public TransportProtocol getPreferredProtocol() { @@ -73,23 +72,12 @@ public class ProtocolSelector extends GroundingSelector { .getTransportProtocol(); } else { if(sourceInvocableOnEdgeNode()) { - // use edge node protocol - if (matches(edgeNodeProtocol, MqttTransportProtocol.class) && supportsProtocol(MqttTransportProtocol.class)) { - if(matchesDeploymentTargets()) { - // get node transport protocol - String nodeControllerId = ((InvocableStreamPipesEntity) source).getDeploymentTargetNodeId(); - - MqttTransportProtocol nodeTransportProtocol = (MqttTransportProtocol) - getNodeBrokerTransportProtocol(nodeControllerId); - nodeTransportProtocol.setTopicDefinition(new SimpleTopicDefinition(outputTopic)); - - return nodeTransportProtocol; - } else { - MqttTransportProtocol tp = (MqttTransportProtocol) - ((InvocableStreamPipesEntity) source).getInputStreams().stream() - .filter(s -> (s.getEventGrounding().getTransportProtocol() instanceof MqttTransportProtocol)) - .findFirst().get().getEventGrounding().getTransportProtocol(); - return new MqttTransportProtocol(tp.getBrokerHostname(), tp.getPort(), outputTopic); + for (SpEdgeNodeProtocol p: prioritizedEdgeProtocols) { + // use edge node protocol + if (matches(p, MqttTransportProtocol.class) && supportsProtocol(MqttTransportProtocol.class)) { + return mqttTransportProtocolForEdge(); + } else if (matches(p, KafkaTransportProtocol.class) && supportsProtocol(KafkaTransportProtocol.class)) { + return kafkaTransportProtocolForEdge(); } } } else { @@ -109,6 +97,58 @@ public class ProtocolSelector extends GroundingSelector { } } + private TransportProtocol mqttTransportProtocolForEdge() { + if(matchesDeploymentTargets()) { + return mqttTransportProtocolForEdge(true); + } else { + return mqttTransportProtocolForEdge(false); + } + } + + private TransportProtocol mqttTransportProtocolForEdge(boolean matchingDeploymentTarget) { + if (matchingDeploymentTarget) { + // get node transport protocol + String nodeControllerId = ((InvocableStreamPipesEntity) source).getDeploymentTargetNodeId(); + MqttTransportProtocol nodeTransportProtocol = (MqttTransportProtocol) + getNodeBrokerTransportProtocol(nodeControllerId); + nodeTransportProtocol.setTopicDefinition(new SimpleTopicDefinition(outputTopic)); + + return nodeTransportProtocol; + } else { + MqttTransportProtocol tp = (MqttTransportProtocol) + ((InvocableStreamPipesEntity) source).getInputStreams().stream() + .filter(s -> (s.getEventGrounding().getTransportProtocol() instanceof MqttTransportProtocol)) + .findFirst().get().getEventGrounding().getTransportProtocol(); + return new MqttTransportProtocol(tp.getBrokerHostname(), tp.getPort(), outputTopic); + } + } + + private TransportProtocol kafkaTransportProtocolForEdge() { + if(matchesDeploymentTargets()) { + return kafkaTransportProtocolForEdge(true); + } else { + return kafkaTransportProtocolForEdge(false); + } + } + + private TransportProtocol kafkaTransportProtocolForEdge(boolean matchingDeploymentTarget) { + if (matchingDeploymentTarget) { + // get node transport protocol + String nodeControllerId = ((InvocableStreamPipesEntity) source).getDeploymentTargetNodeId(); + KafkaTransportProtocol nodeTransportProtocol = (KafkaTransportProtocol) + getNodeBrokerTransportProtocol(nodeControllerId); + nodeTransportProtocol.setTopicDefinition(new SimpleTopicDefinition(outputTopic)); + + return nodeTransportProtocol; + } else { + KafkaTransportProtocol tp = (KafkaTransportProtocol) + ((InvocableStreamPipesEntity) source).getInputStreams().stream() + .filter(s -> (s.getEventGrounding().getTransportProtocol() instanceof KafkaTransportProtocol)) + .findFirst().get().getEventGrounding().getTransportProtocol(); + return new KafkaTransportProtocol(tp.getBrokerHostname(), tp.getKafkaPort(), outputTopic); + } + } + private boolean sourceInvocableOnEdgeNode() { return ((InvocableStreamPipesEntity) source).getDeploymentTargetNodeId() != null && !((InvocableStreamPipesEntity) source).getDeploymentTargetNodeId().equals("default"); diff --git a/streampipes-vocabulary/src/main/java/org/apache/streampipes/vocabulary/StreamPipes.java b/streampipes-vocabulary/src/main/java/org/apache/streampipes/vocabulary/StreamPipes.java index 4ab49c0..0c09714 100644 --- a/streampipes-vocabulary/src/main/java/org/apache/streampipes/vocabulary/StreamPipes.java +++ b/streampipes-vocabulary/src/main/java/org/apache/streampipes/vocabulary/StreamPipes.java @@ -369,6 +369,7 @@ public class StreamPipes { public static final String DEPLOYMENT_CONTAINER_ENV_VARS = NS + "dockerContainerEnvVars"; public static final String DEPLOYMENT_CONTAINER_LABELS = NS + "dockerContainerLabels"; public static final String DEPLOYMENT_CONTAINER_VOLUMES = NS + "dockerContainerVolumes"; + public static final String DEPLOYMENT_CONTAINER_DEPENDENCIES = NS + "dockerContainerDependencies"; // UI Rendering diff --git a/ui/src/app/configuration/messaging-configuration/messaging-configuration.component.html b/ui/src/app/configuration/messaging-configuration/messaging-configuration.component.html index ea9a876..25a760e 100644 --- a/ui/src/app/configuration/messaging-configuration/messaging-configuration.component.html +++ b/ui/src/app/configuration/messaging-configuration/messaging-configuration.component.html @@ -102,15 +102,16 @@ </div> </div> <div fxFlex="100" fxLayout="column" fxLayoutAlign="start start" class="sp-blue-border page-container-padding-inner"> - <div cdkDropList class="data-format-list" (cdkDropListDropped)="dropProtocol($event)" + <div cdkDropList class="data-format-list" (cdkDropListDropped)="dropEdgeProtocol($event)" *ngIf="loadingCompleted"> - <div class="data-format-box"cdkDrag> - {{messagingSettings.edgeNodeProtocol}} + <div class="data-format-box" *ngFor="let edgeProtocol of messagingSettings.prioritizedEdgeProtocols" + cdkDrag> + {{edgeProtocol}} </div> </div> <div fxLayoutAlign="end center"> -<!-- <button mat-raised-button color="primary" type="submit" class="md-raised md-primary submit-button" (click)="updateMessagingSettings()">Update--> -<!-- </button>--> + <button mat-raised-button color="primary" type="submit" class="md-raised md-primary submit-button" (click)="updateMessagingSettings()">Update + </button> </div> </div> diff --git a/ui/src/app/configuration/messaging-configuration/messaging-configuration.component.ts b/ui/src/app/configuration/messaging-configuration/messaging-configuration.component.ts index 99617d7..cb1d53c 100644 --- a/ui/src/app/configuration/messaging-configuration/messaging-configuration.component.ts +++ b/ui/src/app/configuration/messaging-configuration/messaging-configuration.component.ts @@ -57,4 +57,8 @@ export class MessagingConfigurationComponent { dropProtocol(event: CdkDragDrop<string[]>) { moveItemInArray(this.messagingSettings.prioritizedProtocols, event.previousIndex, event.currentIndex); } + + dropEdgeProtocol(event: CdkDragDrop<string[]>) { + moveItemInArray(this.messagingSettings.prioritizedEdgeProtocols, event.previousIndex, event.currentIndex); + } } \ No newline at end of file diff --git a/ui/src/app/configuration/shared/messaging-settings.model.ts b/ui/src/app/configuration/shared/messaging-settings.model.ts index a33ce26..f90c7e1 100644 --- a/ui/src/app/configuration/shared/messaging-settings.model.ts +++ b/ui/src/app/configuration/shared/messaging-settings.model.ts @@ -25,5 +25,5 @@ export interface MessagingSettings { prioritizedFormats: [string]; prioritizedProtocols: [string]; - edgeNodeProtocol: string; + prioritizedEdgeProtocols: [string]; } \ No newline at end of file diff --git a/ui/src/app/core-model/gen/streampipes-model.ts b/ui/src/app/core-model/gen/streampipes-model.ts index 437271e..5039b62 100644 --- a/ui/src/app/core-model/gen/streampipes-model.ts +++ b/ui/src/app/core-model/gen/streampipes-model.ts @@ -19,7 +19,7 @@ /* tslint:disable */ /* eslint-disable */ // @ts-nocheck -// Generated using typescript-generator version 2.27.744 on 2021-03-13 13:56:44. +// Generated using typescript-generator version 2.27.744 on 2021-04-12 09:37:20. export class AbstractStreamPipesEntity { "@class": "org.apache.streampipes.model.base.AbstractStreamPipesEntity" | "org.apache.streampipes.model.base.NamedStreamPipesEntity" | "org.apache.streampipes.model.connect.adapter.AdapterDescription" | "org.apache.streampipes.model.connect.adapter.AdapterSetDescription" | "org.apache.streampipes.model.connect.adapter.GenericAdapterSetDescription" | "org.apache.streampipes.model.connect.adapter.SpecificAdapterSetDescription" | "org.apache.streampipes.model.connect.adapter.AdapterStre [...] @@ -1222,10 +1222,12 @@ export class DeploymentContainer extends UnnamedStreamPipesEntity { "@class": "org.apache.streampipes.model.node.container.DeploymentContainer" | "org.apache.streampipes.model.node.container.DockerContainer"; containerName: string; containerPorts: string[]; + dependsOnContainers: string[]; envVars: string[]; imageUri: string; labels: { [index: string]: string }; serviceId: string; + volumes: string[]; static fromData(data: DeploymentContainer, target?: DeploymentContainer): DeploymentContainer { if (!data) { @@ -1239,6 +1241,8 @@ export class DeploymentContainer extends UnnamedStreamPipesEntity { instance.containerPorts = __getCopyArrayFn(__identity<string>())(data.containerPorts); instance.envVars = __getCopyArrayFn(__identity<string>())(data.envVars); instance.labels = __getCopyObjectFn(__identity<string>())(data.labels); + instance.volumes = __getCopyArrayFn(__identity<string>())(data.volumes); + instance.dependsOnContainers = __getCopyArrayFn(__identity<string>())(data.dependsOnContainers); return instance; }
