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 692298aaf1ea86fa262cf16ae6d92573ca0709c1 Author: Patrick Wiener <[email protected]> AuthorDate: Sat Apr 10 13:52:35 2021 +0200 refactored node controller, enhanced DockerContainerBuilder --- ...ckerContainer.java => ContainerEnvBuilder.java} | 37 ++++--- .../model/node/container/ContainerLabels.java | 24 ++--- .../model/node/container/ContainerType.java | 6 +- ...Container.java => ContainerVolumesBuilder.java} | 40 ++++--- .../model/node/container/DeploymentContainer.java | 24 ++++- .../model/node/container/DockerContainer.java | 4 +- .../node/container/DockerContainerBuilder.java | 16 +-- .../streampipes/model/node/container/Ports.java | 8 +- ...nningInstances.java => NodeControllerInit.java} | 18 ++-- .../{container => }/api/AbstractResource.java | 2 +- .../{container => }/api/ConnectResource.java | 4 +- .../api/ContainerDeploymentResource.java | 22 ++-- .../api/DataStreamRelayResource.java | 4 +- .../{container => }/api/HealthCheckResource.java | 4 +- .../api/InvocableEntityResource.java | 19 +--- .../api/NodeControllerResourceConfig.java | 2 +- .../api/NodeInfoDescriptionResource.java | 8 +- .../{container => }/config/ConfigKeys.java | 2 +- .../config/NodeControllerConfig.java | 2 +- .../container/DockerExtensionsContainer.java | 44 ++++++++ .../controller/container/DockerKafkaContainer.java | 57 ++++++++++ .../container/DockerMosquittoContainer.java | 43 ++++++++ .../container/DockerZookeeperContainer.java | 40 +++++++ .../orchestrator/docker/DockerConstants.java | 118 --------------------- .../NodeControllerSubmitter.java} | 43 +++++--- .../management/connect/ConnectManager.java | 7 +- .../management/janitor/JanitorManager.java | 6 +- .../management/node/NodeConstants.java | 8 +- .../management/node/NodeManager.java | 5 +- .../orchestrator/DockerOrchestratorManager.java} | 44 ++++---- .../orchestrator/IContainerOrchestrator.java} | 5 +- .../docker/AbstractStreamPipesDockerContainer.java | 56 ++++++++++ .../docker/DockerContainerDeclarerSingleton.java | 63 +++++++++++ .../docker/IDockerContainerDeclarer.java} | 8 +- .../docker/model/ContainerStatus.java} | 6 +- .../orchestrator/docker/model/DockerInfo.java | 2 +- .../orchestrator/docker/utils/DockerUtils.java | 54 ++++++---- .../status}/ContainerDeploymentStatus.java | 12 +-- .../storage}/RunningContainerInstances.java | 10 +- .../pe/IPipelineElementLifeCycle.java} | 4 +- .../management/pe/InvocableElementManager.java | 9 +- .../pe/storage}/RunningInvocableInstances.java | 8 +- .../management/relay/DataStreamRelayManager.java | 6 +- .../relay/base/AbstractEventRelay.java} | 8 +- .../relay/base}/EventRelay.java | 4 +- .../relay/bridges/EventRelayStrategy.java | 2 +- .../management/relay/bridges/JmsToJmsBridge.java | 2 +- .../management/relay/bridges/JmsToKafkaBridge.java | 2 +- .../management/relay/bridges/JmsToMqttBridge.java | 2 +- .../management/relay/bridges/KafkaToJmsBridge.java | 2 +- .../relay/bridges/KafkaToKafkaBridge.java | 2 +- .../relay/bridges/KafkaToMqttBridge.java | 2 +- .../management/relay/bridges/MqttToJmsBridge.java | 2 +- .../relay/bridges/MqttToKafkaBridge.java | 2 +- .../management/relay/bridges/MqttToMqttBridge.java | 2 +- .../relay/bridges/MultiBrokerBridge.java | 4 +- .../relay/storage}/RunningRelayInstances.java | 7 +- .../management/resource/ResourceManager.java | 8 +- .../management/resource/model/ResourceMetrics.java | 2 +- .../{container => }/storage/CRUDStorage.java | 2 +- .../IRunningInstances.java} | 4 +- .../{container => }/storage/MapDBImpl.java | 2 +- .../apache/streampipes/vocabulary/StreamPipes.java | 1 + 63 files changed, 589 insertions(+), 377 deletions(-) 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/ContainerEnvBuilder.java similarity index 55% copy from streampipes-model/src/main/java/org/apache/streampipes/model/node/container/DockerContainer.java copy to streampipes-model/src/main/java/org/apache/streampipes/model/node/container/ContainerEnvBuilder.java index 3d97fe9..00b8346 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/ContainerEnvBuilder.java @@ -17,29 +17,32 @@ */ package org.apache.streampipes.model.node.container; -import io.fogsy.empire.annotations.RdfsClass; -import org.apache.streampipes.model.shared.annotation.TsModel; -import org.apache.streampipes.vocabulary.StreamPipes; - -import javax.persistence.Entity; +import java.util.ArrayList; import java.util.List; -import java.util.Map; -@RdfsClass(StreamPipes.DEPLOYMENT_DOCKER_CONTAINER) -@Entity -@TsModel -public class DockerContainer extends DeploymentContainer { +public class ContainerEnvBuilder { + + private final List<String> envVariables; + + public ContainerEnvBuilder() { + this.envVariables = new ArrayList<>(); + } + + public static ContainerEnvBuilder create() { + return new ContainerEnvBuilder(); + } - public DockerContainer(String elementId) { - super(elementId); + public ContainerEnvBuilder addNodeEnvs(List<String> nodeEnvVariables) { + this.envVariables.addAll(nodeEnvVariables); + return this; } - public DockerContainer() { - super(); + public ContainerEnvBuilder add(String key, String value) { + this.envVariables.add(String.format("%s=%s", key, value)); + return this; } - public DockerContainer(String imageURI, String containerName, String serviceId, String[] containerPorts, - List<String> envVars, Map<String, String> labels) { - super(imageURI, containerName, serviceId, containerPorts, envVars, labels); + public List<String> build() { + return envVariables; } } diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/RunningInstances.java b/streampipes-model/src/main/java/org/apache/streampipes/model/node/container/ContainerLabels.java similarity index 60% copy from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/RunningInstances.java copy to streampipes-model/src/main/java/org/apache/streampipes/model/node/container/ContainerLabels.java index 934f917..9ec07e8 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/RunningInstances.java +++ b/streampipes-model/src/main/java/org/apache/streampipes/model/node/container/ContainerLabels.java @@ -15,19 +15,17 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.management; +package org.apache.streampipes.model.node.container; -import java.util.List; +import java.util.HashMap; +import java.util.Map; -public interface RunningInstances<T> { - - void add(String id, T value); - - boolean isRunning(String id); - - T get(String id); - - List<T> getAll(); - - void remove(String id); +public class ContainerLabels { + public static Map<String, String> with (String containerId, String nodeType, ContainerType containerType) { + return new HashMap<String, String>() {{ + put("org.apache.streampipes.service.id", containerId); + put("org.apache.streampipes.node.type",nodeType); + put("org.apache.streampipes.container.type", containerType.name()); + }}; + } } diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/bridges/EventRelayStrategy.java b/streampipes-model/src/main/java/org/apache/streampipes/model/node/container/ContainerType.java similarity index 85% copy from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/bridges/EventRelayStrategy.java copy to streampipes-model/src/main/java/org/apache/streampipes/model/node/container/ContainerType.java index 8e2f508..b930ed6 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/bridges/EventRelayStrategy.java +++ b/streampipes-model/src/main/java/org/apache/streampipes/model/node/container/ContainerType.java @@ -15,8 +15,8 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.management.relay.bridges; +package org.apache.streampipes.model.node.container; -public enum EventRelayStrategy { - BUFFER, PURGE +public enum ContainerType { + EXTENSIONS, BROKER } 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/ContainerVolumesBuilder.java similarity index 53% copy from streampipes-model/src/main/java/org/apache/streampipes/model/node/container/DockerContainer.java copy to streampipes-model/src/main/java/org/apache/streampipes/model/node/container/ContainerVolumesBuilder.java index 3d97fe9..8733b3e 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/ContainerVolumesBuilder.java @@ -17,29 +17,35 @@ */ package org.apache.streampipes.model.node.container; -import io.fogsy.empire.annotations.RdfsClass; -import org.apache.streampipes.model.shared.annotation.TsModel; -import org.apache.streampipes.vocabulary.StreamPipes; - -import javax.persistence.Entity; +import java.util.ArrayList; import java.util.List; -import java.util.Map; -@RdfsClass(StreamPipes.DEPLOYMENT_DOCKER_CONTAINER) -@Entity -@TsModel -public class DockerContainer extends DeploymentContainer { +public class ContainerVolumesBuilder { + private final List<String> volumes; + + public ContainerVolumesBuilder() { + this.volumes = new ArrayList<>(); + } + + public static ContainerVolumesBuilder create() { + return new ContainerVolumesBuilder(); + } - public DockerContainer(String elementId) { - super(elementId); + public ContainerVolumesBuilder addNodeEnvs(List<String> volumes) { + this.volumes.addAll(volumes); + return this; } - public DockerContainer() { - super(); + public ContainerVolumesBuilder add(String key, String value, boolean readOnly) { + if (readOnly) { + this.volumes.add(String.format("%s:%s:%s", key, value, "ro")); + } else { + this.volumes.add(String.format("%s:%s", key, value)); + } + return this; } - public DockerContainer(String imageURI, String containerName, String serviceId, String[] containerPorts, - List<String> envVars, Map<String, String> labels) { - super(imageURI, containerName, serviceId, containerPorts, envVars, labels); + public List<String> build() { + return volumes; } } 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 cd8f0b2..c4f1cf4 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 @@ -25,6 +25,8 @@ import org.apache.streampipes.model.shared.annotation.TsModel; import org.apache.streampipes.vocabulary.StreamPipes; import javax.persistence.*; +import java.util.ArrayList; +import java.util.HashMap; import java.util.List; import java.util.Map; @@ -58,11 +60,22 @@ public abstract class DeploymentContainer extends UnnamedStreamPipesEntity { @RdfProperty(StreamPipes.DEPLOYMENT_CONTAINER_LABELS) private Map<String, String> labels; + @OneToMany(fetch = FetchType.EAGER, + cascade = {CascadeType.ALL}) + @RdfProperty(StreamPipes.DEPLOYMENT_CONTAINER_VOLUMES) + private List<String> volumes; + public DeploymentContainer() { + this.envVars = new ArrayList<>(); + this.labels = new HashMap<>(); + this.volumes = new ArrayList<>(); } public DeploymentContainer(String elementId) { super(elementId); + this.envVars = new ArrayList<>(); + this.labels = new HashMap<>(); + this.volumes = new ArrayList<>(); } public DeploymentContainer(DeploymentContainer other) { @@ -70,13 +83,14 @@ 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> envVars, Map<String, String> labels, List<String> volumes) { this.imageUri = imageUri; this.containerName = containerName; this.serviceId = serviceId; this.containerPorts = containerPorts; this.envVars = envVars; this.labels = labels; + this.volumes = volumes; } public String getImageUri() { @@ -125,4 +139,12 @@ public abstract class DeploymentContainer extends UnnamedStreamPipesEntity { public void setLabels(Map<String, String> labels) { this.labels = labels; } + + public List<String> getVolumes() { + return volumes; + } + + public void setVolumes(List<String> volumes) { + this.volumes = volumes; + } } 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 3d97fe9..b2e6bb7 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,7 @@ public class DockerContainer extends DeploymentContainer { } public DockerContainer(String imageURI, String containerName, String serviceId, String[] containerPorts, - List<String> envVars, Map<String, String> labels) { - super(imageURI, containerName, serviceId, containerPorts, envVars, labels); + List<String> envVars, Map<String, String> labels, List<String> volumes) { + super(imageURI, containerName, serviceId, containerPorts, envVars, labels, volumes); } } 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 53329a6..c7d33a7 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 @@ -20,21 +20,10 @@ import java.util.*; public class DockerContainerBuilder { private final DockerContainer dockerContainer; - private String imageURI; - private String containerName; - private String serviceId; - private String [] containerPorts; - private List<String> envVars; - private Map<String, String> labels; public DockerContainerBuilder(String id) { this.dockerContainer = new DockerContainer(); this.dockerContainer.setServiceId(id); -// this.imageURI = ""; -// this.containerName = ""; -// this.containerPorts = new String[]{}; -// this.envVars = new ArrayList<>(); -// this.labels = new HashMap<>(); } public static DockerContainerBuilder create(String id) { @@ -66,6 +55,11 @@ public class DockerContainerBuilder { return this; } + public DockerContainerBuilder withVolumes(List<String> volumes) { + this.dockerContainer.setVolumes(volumes); + return this; + } + public DockerContainer build() { return dockerContainer; } diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/bridges/EventRelayStrategy.java b/streampipes-model/src/main/java/org/apache/streampipes/model/node/container/Ports.java similarity index 83% copy from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/bridges/EventRelayStrategy.java copy to streampipes-model/src/main/java/org/apache/streampipes/model/node/container/Ports.java index 8e2f508..55a64bb 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/bridges/EventRelayStrategy.java +++ b/streampipes-model/src/main/java/org/apache/streampipes/model/node/container/Ports.java @@ -15,8 +15,10 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.management.relay.bridges; +package org.apache.streampipes.model.node.container; -public enum EventRelayStrategy { - BUFFER, PURGE +public class Ports { + public static String[] withMapping(String... port) { + return port; + } } diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/RunningInstances.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/NodeControllerInit.java similarity index 73% copy from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/RunningInstances.java copy to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/NodeControllerInit.java index 934f917..e069041 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/RunningInstances.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/NodeControllerInit.java @@ -15,19 +15,13 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.management; +package org.apache.streampipes.node.controller; -import java.util.List; +import org.apache.streampipes.node.controller.management.NodeControllerSubmitter; -public interface RunningInstances<T> { +public class NodeControllerInit extends NodeControllerSubmitter { - void add(String id, T value); - - boolean isRunning(String id); - - T get(String id); - - List<T> getAll(); - - void remove(String id); + public static void main(String... args) { + new NodeControllerInit().init(); + } } diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/api/AbstractResource.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/api/AbstractResource.java similarity index 96% rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/api/AbstractResource.java rename to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/api/AbstractResource.java index dc18520..5b2fa53 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/api/AbstractResource.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/api/AbstractResource.java @@ -15,7 +15,7 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.api; +package org.apache.streampipes.node.controller.api; import org.apache.streampipes.model.message.Message; diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/api/ConnectResource.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/api/ConnectResource.java similarity index 97% rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/api/ConnectResource.java rename to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/api/ConnectResource.java index 2865cc1..8970551 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/api/ConnectResource.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/api/ConnectResource.java @@ -15,14 +15,14 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.api; +package org.apache.streampipes.node.controller.api; import org.apache.streampipes.model.connect.adapter.AdapterDescription; import org.apache.streampipes.model.connect.adapter.AdapterSetDescription; import org.apache.streampipes.model.connect.adapter.AdapterStreamDescription; import org.apache.streampipes.model.connect.worker.ConnectWorkerContainer; import org.apache.streampipes.model.runtime.RuntimeOptionsRequest; -import org.apache.streampipes.node.controller.container.management.connect.ConnectManager; +import org.apache.streampipes.node.controller.management.connect.ConnectManager; import org.apache.streampipes.rest.shared.annotation.JacksonSerialized; import javax.ws.rs.*; diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/api/ContainerDeploymentResource.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/api/ContainerDeploymentResource.java similarity index 65% rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/api/ContainerDeploymentResource.java rename to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/api/ContainerDeploymentResource.java index 69809e7..b7a980d 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/api/ContainerDeploymentResource.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/api/ContainerDeploymentResource.java @@ -15,14 +15,14 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.api; +package org.apache.streampipes.node.controller.api; import org.apache.streampipes.model.node.container.DockerContainer; -import org.apache.streampipes.node.controller.container.management.node.NodeManager; -import org.apache.streampipes.node.controller.container.management.orchestrator.ContainerDeploymentStatus; -import org.apache.streampipes.node.controller.container.management.orchestrator.docker.DockerConstants; -import org.apache.streampipes.node.controller.container.management.orchestrator.docker.DockerContainerManager; -import org.apache.streampipes.node.controller.container.management.pe.InvocableElementManager; +import org.apache.streampipes.node.controller.management.node.NodeManager; +import org.apache.streampipes.node.controller.management.orchestrator.status.ContainerDeploymentStatus; +import org.apache.streampipes.node.controller.management.orchestrator.docker.model.ContainerStatus; +import org.apache.streampipes.node.controller.management.orchestrator.DockerOrchestratorManager; +import org.apache.streampipes.node.controller.management.pe.InvocableElementManager; import javax.ws.rs.*; import javax.ws.rs.core.MediaType; @@ -33,16 +33,16 @@ public class ContainerDeploymentResource extends AbstractResource { @GET @Produces(MediaType.APPLICATION_JSON) public javax.ws.rs.core.Response getPipelineElementContainer(){ - return ok(DockerContainerManager.getInstance().list()); + return ok(DockerOrchestratorManager.getInstance().list()); } @POST @Path("/deploy") @Consumes(MediaType.APPLICATION_JSON) public javax.ws.rs.core.Response deployPipelineElementContainer(DockerContainer container) { - ContainerDeploymentStatus status = DockerContainerManager.getInstance().deploy(container); + ContainerDeploymentStatus status = DockerOrchestratorManager.getInstance().deploy(container); - if (status.getStatus() == DockerConstants.ContainerStatus.DEPLOYED) { + if (status.getStatus() == ContainerStatus.DEPLOYED) { NodeManager.getInstance().addToRegisteredContainers(status.getContainer()); } return ok(status); @@ -52,9 +52,9 @@ public class ContainerDeploymentResource extends AbstractResource { @Path("/remove") @Consumes(MediaType.APPLICATION_JSON) public javax.ws.rs.core.Response removePipelineElementContainer(DockerContainer container) { - ContainerDeploymentStatus status = DockerContainerManager.getInstance().remove(container); + ContainerDeploymentStatus status = DockerOrchestratorManager.getInstance().remove(container); - if (status.getStatus() == DockerConstants.ContainerStatus.REMOVED) { + if (status.getStatus() == ContainerStatus.REMOVED) { InvocableElementManager.getInstance().unregister(); NodeManager.getInstance().removeFromRegisteredContainers(status.getContainer()); } diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/api/DataStreamRelayResource.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/api/DataStreamRelayResource.java similarity index 91% rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/api/DataStreamRelayResource.java rename to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/api/DataStreamRelayResource.java index 0883a68..53022dc 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/api/DataStreamRelayResource.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/api/DataStreamRelayResource.java @@ -15,10 +15,10 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.api; +package org.apache.streampipes.node.controller.api; import org.apache.streampipes.model.eventrelay.SpDataStreamRelayContainer; -import org.apache.streampipes.node.controller.container.management.relay.DataStreamRelayManager; +import org.apache.streampipes.node.controller.management.relay.DataStreamRelayManager; import org.apache.streampipes.rest.shared.annotation.JacksonSerialized; import javax.ws.rs.*; diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/api/HealthCheckResource.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/api/HealthCheckResource.java similarity index 89% rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/api/HealthCheckResource.java rename to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/api/HealthCheckResource.java index 6a4f88e..05bf038 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/api/HealthCheckResource.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/api/HealthCheckResource.java @@ -15,9 +15,9 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.api; +package org.apache.streampipes.node.controller.api; -import org.apache.streampipes.node.controller.container.config.NodeControllerConfig; +import org.apache.streampipes.node.controller.config.NodeControllerConfig; import javax.ws.rs.GET; import javax.ws.rs.Path; diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/api/InvocableEntityResource.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/api/InvocableEntityResource.java similarity index 80% rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/api/InvocableEntityResource.java rename to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/api/InvocableEntityResource.java index e54e6dd..473d426 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/api/InvocableEntityResource.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/api/InvocableEntityResource.java @@ -15,33 +15,20 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.api; +package org.apache.streampipes.node.controller.api; -import com.fasterxml.jackson.core.JsonProcessingException; -import com.fasterxml.jackson.databind.ObjectMapper; -import org.apache.kafka.clients.producer.KafkaProducer; -import org.apache.streampipes.commons.exceptions.SpRuntimeException; import org.apache.streampipes.container.model.node.InvocableRegistration; -import org.apache.streampipes.messaging.EventProducer; -import org.apache.streampipes.messaging.jms.ActiveMQPublisher; -import org.apache.streampipes.messaging.kafka.SpKafkaProducer; -import org.apache.streampipes.messaging.mqtt.MqttPublisher; import org.apache.streampipes.model.Response; import org.apache.streampipes.model.base.InvocableStreamPipesEntity; -import org.apache.streampipes.model.grounding.JmsTransportProtocol; -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.pipeline.PipelineElementReconfigurationEntity; -import org.apache.streampipes.node.controller.container.management.pe.InvocableElementManager; -import org.apache.streampipes.node.controller.container.management.pe.RunningInvocableInstances; +import org.apache.streampipes.node.controller.management.pe.InvocableElementManager; +import org.apache.streampipes.node.controller.management.pe.storage.RunningInvocableInstances; import org.apache.streampipes.rest.shared.annotation.JacksonSerialized; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import javax.ws.rs.*; import javax.ws.rs.core.*; -import java.nio.charset.StandardCharsets; @Path("/api/v2/node/element") public class InvocableEntityResource extends AbstractResource { diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/api/NodeControllerResourceConfig.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/api/NodeControllerResourceConfig.java similarity index 95% rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/api/NodeControllerResourceConfig.java rename to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/api/NodeControllerResourceConfig.java index 893bcb3..9cf65c2 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/api/NodeControllerResourceConfig.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/api/NodeControllerResourceConfig.java @@ -15,7 +15,7 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.api; +package org.apache.streampipes.node.controller.api; import org.glassfish.jersey.server.ResourceConfig; import org.springframework.stereotype.Component; diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/api/NodeInfoDescriptionResource.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/api/NodeInfoDescriptionResource.java similarity index 88% rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/api/NodeInfoDescriptionResource.java rename to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/api/NodeInfoDescriptionResource.java index 6bbe2f5..aee6545 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/api/NodeInfoDescriptionResource.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/api/NodeInfoDescriptionResource.java @@ -15,12 +15,12 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.api; +package org.apache.streampipes.node.controller.api; import org.apache.streampipes.model.node.NodeInfoDescription; -import org.apache.streampipes.node.controller.container.management.node.NodeManager; -import org.apache.streampipes.node.controller.container.management.relay.DataStreamRelayManager;; -import org.apache.streampipes.node.controller.container.management.resource.ResourceManager; +import org.apache.streampipes.node.controller.management.node.NodeManager; +import org.apache.streampipes.node.controller.management.relay.DataStreamRelayManager;; +import org.apache.streampipes.node.controller.management.resource.ResourceManager; import org.apache.streampipes.rest.shared.annotation.JacksonSerialized; import javax.ws.rs.*; diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/config/ConfigKeys.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/config/ConfigKeys.java similarity index 97% rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/config/ConfigKeys.java rename to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/config/ConfigKeys.java index 00d3091..0dccd71 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/config/ConfigKeys.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/config/ConfigKeys.java @@ -15,7 +15,7 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.config; +package org.apache.streampipes.node.controller.config; public class ConfigKeys { // required diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/config/NodeControllerConfig.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/config/NodeControllerConfig.java similarity index 99% rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/config/NodeControllerConfig.java rename to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/config/NodeControllerConfig.java index 9a9f6e8..0659259 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/config/NodeControllerConfig.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/config/NodeControllerConfig.java @@ -15,7 +15,7 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.config; +package org.apache.streampipes.node.controller.config; import org.apache.streampipes.config.SpConfig; import org.apache.streampipes.model.node.resources.fielddevice.FieldDeviceAccessResource; 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 new file mode 100644 index 0000000..f527307 --- /dev/null +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/DockerExtensionsContainer.java @@ -0,0 +1,44 @@ +/* + * 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.node.controller.container; + +import org.apache.streampipes.model.node.container.*; +import org.apache.streampipes.node.controller.config.NodeControllerConfig; +import org.apache.streampipes.node.controller.management.orchestrator.docker.AbstractStreamPipesDockerContainer; + +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) + .withImage("apachestreampipes/extensions-all-jvm:" + getStreamPipesVersion()) + .withName("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") + .build()) + .withLabels(ContainerLabels.with(SP_CONTAINER_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 new file mode 100644 index 0000000..781d13c --- /dev/null +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/DockerKafkaContainer.java @@ -0,0 +1,57 @@ +/* + * 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.node.controller.container; + +import org.apache.streampipes.model.node.container.*; +import org.apache.streampipes.node.controller.config.NodeControllerConfig; +import org.apache.streampipes.node.controller.management.orchestrator.docker.AbstractStreamPipesDockerContainer; + + +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) + .withImage("fogsyio/kafka:2.2.0") + .withName("streampipes-node-broker") + .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") + .build()) + .withLabels(ContainerLabels.with(SP_CONTAINER_ID, retrieveNodeType(), ContainerType.BROKER)) + .withVolumes( + ContainerVolumesBuilder.create() + .add("/var/lib/streampipes/kafka", "/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 new file mode 100644 index 0000000..e869a9f --- /dev/null +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/DockerMosquittoContainer.java @@ -0,0 +1,43 @@ +/* + * 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.node.controller.container; + +import org.apache.streampipes.model.node.container.*; +import org.apache.streampipes.node.controller.management.orchestrator.docker.AbstractStreamPipesDockerContainer; + +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) + .withImage("eclipse-mosquitto:1.6.12") + .withName("streampipes-node-broker") + .withExposedPorts(Ports.withMapping("1883")) + .withEnvironmentVariables( + ContainerEnvBuilder.create() + .addNodeEnvs(generateStreamPipesNodeEnvs()) + .build()) + .withLabels(ContainerLabels.with(SP_CONTAINER_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 new file mode 100644 index 0000000..76382a4 --- /dev/null +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/DockerZookeeperContainer.java @@ -0,0 +1,40 @@ +/* + * 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.node.controller.container; + +import org.apache.streampipes.model.node.container.*; +import org.apache.streampipes.node.controller.management.orchestrator.docker.AbstractStreamPipesDockerContainer; + +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) + .withImage("fogsyio/zookeeper:3.4.13") + .withName("streampipes-node-zookeeper") + .withExposedPorts(Ports.withMapping("2181")) + .withEnvironmentVariables( + ContainerEnvBuilder.create() + .addNodeEnvs(generateStreamPipesNodeEnvs()) + .build()) + .withLabels(ContainerLabels.with(SP_CONTAINER_ID, retrieveNodeType(), ContainerType.BROKER)) + .build(); + } +} diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/orchestrator/docker/DockerConstants.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/orchestrator/docker/DockerConstants.java deleted file mode 100644 index 6a54c93..0000000 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/orchestrator/docker/DockerConstants.java +++ /dev/null @@ -1,118 +0,0 @@ -/* - * 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.node.controller.container.management.orchestrator.docker; - -import org.apache.streampipes.model.node.container.DockerContainer; -import org.apache.streampipes.model.node.container.DockerContainerBuilder; -import org.apache.streampipes.node.controller.container.config.ConfigKeys; -import org.apache.streampipes.node.controller.container.config.NodeControllerConfig; -import org.apache.streampipes.node.controller.container.management.node.NodeManager; - -import java.util.*; - -public class DockerConstants { - - public static final String SP_VERSION = - !NodeControllerConfig.INSTANCE.getSpVersion().equals("") ? - NodeControllerConfig.INSTANCE.getSpVersion() : NodeManager.getInstance().getStreamPipesVersion(); - public static final String SP_CONTAINER_PREFIX = "streampipes-"; - public static final String SP_CONTAINER_NETWORK = "spnet"; - - // extensions container - public static final String SP_EXTENSIONS_ID = "pe/org.apache.streampipes.extensions.all.jvm"; - public static final String SP_IMG_EXTENSIONS = "apachestreampipes/extensions-all-jvm:" + SP_VERSION; - public static final String SP_CONTAINER_EXTENSIONS_NAME = SP_CONTAINER_PREFIX + "extensions"; - public static final String[] SP_CONTAINER_EXTENSIONS_PORT = new String[]{"8090"}; - - //broker container - public static final String SP_BROKER_ID = "pe/org.apache.streampipes.node.broker"; - public static final String SP_IMG_MOSQUITTO = "eclipse-mosquitto:1.6.12"; - public static final String SP_CONTAINER_BROKER_NAME = SP_CONTAINER_PREFIX + "node-broker"; - public static final String[] SP_CONTAINER_BROKER_PORT = - System.getenv(ConfigKeys.NODE_BROKER_CONTAINER_PORT) != null ? - new String[]{System.getenv(ConfigKeys.NODE_BROKER_CONTAINER_PORT)} : new String[]{"1883"}; - - // container env variables - public static final List<String> SP_CONTAINER_ENV_VARIABLES = Arrays.asList( - ConfigKeys.NODE_CONTROLLER_ID + "=" + NodeControllerConfig.INSTANCE.getNodeControllerId(), - ConfigKeys.NODE_CONTROLLER_CONTAINER_HOST + "=" + NodeControllerConfig.INSTANCE.getNodeHost(), - ConfigKeys.NODE_CONTROLLER_CONTAINER_PORT + "=" + NodeControllerConfig.INSTANCE.getNodeControllerPort(), - ConfigKeys.NODE_BROKER_CONTAINER_HOST + "=" + NodeControllerConfig.INSTANCE.getNodeBrokerHost(), - ConfigKeys.NODE_BROKER_CONTAINER_PORT + "=" + NodeControllerConfig.INSTANCE.getNodeBrokerPort(), - //TODO: remove once PE does not need it for KV registration - "CONSUL_LOCATION=" + NodeControllerConfig.INSTANCE.consulLocation(), - "SP_HOST=" + NodeControllerConfig.INSTANCE.getNodeHost(), - "SP_PORT=8090" - ); - - // extension container labels - public static final Map<String, String> SP_CONTAINER_EXTENSIONS_LABELS = new HashMap<String, String>() {{ - put("org.apache.streampipes.service.id", SP_EXTENSIONS_ID); - put("org.apache.streampipes.node.type", NodeControllerConfig.INSTANCE.getNodeType()); - put("org.apache.streampipes.container.type", "extensions"); - }}; - - // mosquitto container labels - public static final Map<String, String> SP_CONTAINER_BROKER_LABELS = new HashMap<String, String>() {{ - put("org.apache.streampipes.service.id", SP_BROKER_ID); - put("org.apache.streampipes.node.type", NodeControllerConfig.INSTANCE.getNodeType()); - put("org.apache.streampipes.container.type", "broker"); - }}; - - - public enum ContainerStatus { - DEPLOYED, RUNNING, STOPPED, REMOVED, UNKNOWN, FAILED - } - - public enum ContainerType { - EXTENSIONS, BROKER - } - - public enum NodeContainer { - INSTANCE; - - public List<DockerContainer> getAllStreamPipesContainer() { - List<DockerContainer> nodeContainers = new ArrayList<>(); - - // Extensions container - DockerContainer extensions = DockerContainerBuilder.create(SP_EXTENSIONS_ID) - .withImage(DockerConstants.SP_IMG_EXTENSIONS) - .withName(DockerConstants.SP_CONTAINER_EXTENSIONS_NAME) - .withExposedPorts(DockerConstants.SP_CONTAINER_EXTENSIONS_PORT) - .withEnvironmentVariables(SP_CONTAINER_ENV_VARIABLES) - .withLabels(DockerConstants.SP_CONTAINER_EXTENSIONS_LABELS) - .build(); - - // Node broker container - DockerContainer nodeBroker = DockerContainerBuilder.create(SP_BROKER_ID) - .withImage(DockerConstants.SP_IMG_MOSQUITTO) - .withName(DockerConstants.SP_CONTAINER_BROKER_NAME) - .withExposedPorts(DockerConstants.SP_CONTAINER_BROKER_PORT) - .withEnvironmentVariables(SP_CONTAINER_ENV_VARIABLES) - .withLabels(SP_CONTAINER_BROKER_LABELS) - .build(); - - nodeContainers.add(extensions); - nodeContainers.add(nodeBroker); - - return nodeContainers; - } - - } - -} diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/NodeControllerInit.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/NodeControllerSubmitter.java similarity index 61% rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/NodeControllerInit.java rename to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/NodeControllerSubmitter.java index 539a05f..5c2ff0c 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/NodeControllerInit.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/NodeControllerSubmitter.java @@ -15,16 +15,20 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container; +package org.apache.streampipes.node.controller.management; import org.apache.streampipes.commons.exceptions.SpRuntimeException; -import org.apache.streampipes.node.controller.container.management.orchestrator.docker.DockerContainerManager; -import org.apache.streampipes.node.controller.container.api.NodeControllerResourceConfig; -import org.apache.streampipes.node.controller.container.config.NodeControllerConfig; -import org.apache.streampipes.node.controller.container.management.node.NodeManager; -import org.apache.streampipes.node.controller.container.management.janitor.JanitorManager; -import org.apache.streampipes.node.controller.container.management.pe.InvocableElementManager; -import org.apache.streampipes.node.controller.container.management.resource.ResourceManager; +import org.apache.streampipes.node.controller.api.NodeControllerResourceConfig; +import org.apache.streampipes.node.controller.config.NodeControllerConfig; +import org.apache.streampipes.node.controller.container.DockerExtensionsContainer; +import org.apache.streampipes.node.controller.container.DockerKafkaContainer; +import org.apache.streampipes.node.controller.container.DockerMosquittoContainer; +import org.apache.streampipes.node.controller.container.DockerZookeeperContainer; +import org.apache.streampipes.node.controller.management.janitor.JanitorManager; +import org.apache.streampipes.node.controller.management.node.NodeManager; +import org.apache.streampipes.node.controller.management.orchestrator.DockerOrchestratorManager; +import org.apache.streampipes.node.controller.management.orchestrator.docker.DockerContainerDeclarerSingleton; +import org.apache.streampipes.node.controller.management.resource.ResourceManager; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.boot.SpringApplication; @@ -32,23 +36,22 @@ import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Import; -import javax.annotation.PostConstruct; import javax.annotation.PreDestroy; import java.util.Collections; @Configuration @EnableAutoConfiguration @Import({ NodeControllerResourceConfig.class }) -public class NodeControllerInit { +public abstract class NodeControllerSubmitter { private static final Logger LOG = - LoggerFactory.getLogger(NodeControllerInit.class.getCanonicalName()); + LoggerFactory.getLogger(NodeControllerSubmitter.class.getCanonicalName()); - public static void main(String [] args) { + public void init() { NodeControllerConfig conf = NodeControllerConfig.INSTANCE; - SpringApplication app = new SpringApplication(NodeControllerInit.class); + SpringApplication app = new SpringApplication(NodeControllerSubmitter.class); app.setDefaultProperties(Collections.singletonMap("server.port", conf.getNodeControllerPort())); app.run(); @@ -68,11 +71,16 @@ public class NodeControllerInit { ResourceManager.getInstance().run(); if (!"true".equals(System.getenv("SP_DEBUG"))) { - LOG.info("Auto-deploy extensions container"); - DockerContainerManager.getInstance().init(); -// LOG.info("Checking for pipeline elements to be started ..."); -// InvocableElementManager.getInstance().invokePipelineElementsOnSystemRebootOrRestart(); + LOG.info("Register auto-deploy containers"); + DockerContainerDeclarerSingleton.getInstance() + .register(new DockerExtensionsContainer()) + .register(new DockerMosquittoContainer()) + .register(new DockerKafkaContainer()) + .register(new DockerZookeeperContainer()); + + LOG.info("Auto-deploy extensions container"); + DockerOrchestratorManager.getInstance().init(); LOG.info("Start janitor manager"); JanitorManager.getInstance().run(); @@ -83,4 +91,5 @@ public class NodeControllerInit { @PreDestroy public void onExit(){ } + } diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/connect/ConnectManager.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/connect/ConnectManager.java similarity index 96% rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/connect/ConnectManager.java rename to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/connect/ConnectManager.java index 8063b13..6a69a7a 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/connect/ConnectManager.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/connect/ConnectManager.java @@ -15,7 +15,7 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.management.connect; +package org.apache.streampipes.node.controller.management.connect; import com.fasterxml.jackson.core.JsonProcessingException; import org.apache.http.HttpResponse; @@ -31,8 +31,7 @@ import org.apache.streampipes.model.connect.adapter.AdapterStreamDescription; import org.apache.streampipes.model.connect.guess.GuessSchema; import org.apache.streampipes.model.connect.worker.ConnectWorkerContainer; import org.apache.streampipes.model.runtime.RuntimeOptionsRequest; -import org.apache.streampipes.node.controller.container.config.NodeControllerConfig; -import org.apache.streampipes.node.controller.container.management.orchestrator.docker.DockerConstants; +import org.apache.streampipes.node.controller.config.NodeControllerConfig; import org.apache.streampipes.serializers.json.JacksonSerializer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -53,7 +52,7 @@ public class ConnectManager { // Connect adapter base route // TODO: get from registered extensions or connect adapater config - private static final String CONNECT_WORKER_HOST = DockerConstants.SP_CONTAINER_EXTENSIONS_NAME; + private static final String CONNECT_WORKER_HOST = "streampipes-extensions"; private static final int CONNECT_WORKER_PORT = 8090; private static final String CONNECT_WORKER_BASE_ROUTE = "/api/v1/{username}/worker"; private static final String STREAM_ROUTE = "/stream"; diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/janitor/JanitorManager.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/janitor/JanitorManager.java similarity index 88% rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/janitor/JanitorManager.java rename to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/janitor/JanitorManager.java index 818e7eb..a966b04 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/janitor/JanitorManager.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/janitor/JanitorManager.java @@ -15,10 +15,10 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.management.janitor; +package org.apache.streampipes.node.controller.management.janitor; -import org.apache.streampipes.node.controller.container.config.NodeControllerConfig; -import org.apache.streampipes.node.controller.container.management.orchestrator.docker.utils.DockerUtils; +import org.apache.streampipes.node.controller.config.NodeControllerConfig; +import org.apache.streampipes.node.controller.management.orchestrator.docker.utils.DockerUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/node/NodeConstants.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/node/NodeConstants.java similarity index 95% rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/node/NodeConstants.java rename to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/node/NodeConstants.java index 4c04463..19b5985 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/node/NodeConstants.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/node/NodeConstants.java @@ -15,7 +15,7 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.management.node; +package org.apache.streampipes.node.controller.management.node; import org.apache.streampipes.model.node.container.DeploymentContainer; import org.apache.streampipes.model.node.container.DockerContainer; @@ -27,9 +27,9 @@ import org.apache.streampipes.model.node.resources.hardware.CPU; import org.apache.streampipes.model.node.resources.hardware.DISK; import org.apache.streampipes.model.node.resources.hardware.GPU; import org.apache.streampipes.model.node.resources.hardware.MEM; -import org.apache.streampipes.node.controller.container.config.NodeControllerConfig; -import org.apache.streampipes.node.controller.container.management.orchestrator.docker.model.DockerInfo; -import org.apache.streampipes.node.controller.container.management.orchestrator.docker.utils.DockerUtils; +import org.apache.streampipes.node.controller.config.NodeControllerConfig; +import org.apache.streampipes.node.controller.management.orchestrator.docker.model.DockerInfo; +import org.apache.streampipes.node.controller.management.orchestrator.docker.utils.DockerUtils; import oshi.SystemInfo; import oshi.hardware.ComputerSystem; import oshi.hardware.HardwareAbstractionLayer; diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/node/NodeManager.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/node/NodeManager.java similarity index 98% rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/node/NodeManager.java rename to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/node/NodeManager.java index 5af5a83..4025d0c 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/node/NodeManager.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/node/NodeManager.java @@ -15,7 +15,7 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.management.node; +package org.apache.streampipes.node.controller.management.node; import org.apache.http.client.fluent.Request; import org.apache.http.client.fluent.Response; @@ -29,8 +29,7 @@ import org.apache.streampipes.model.node.*; import org.apache.streampipes.model.node.container.DeploymentContainer; import org.apache.streampipes.model.node.container.DockerContainer; import org.apache.streampipes.model.node.meta.GeoLocation; -import org.apache.streampipes.model.pipeline.Pipeline; -import org.apache.streampipes.node.controller.container.config.NodeControllerConfig; +import org.apache.streampipes.node.controller.config.NodeControllerConfig; import org.apache.streampipes.serializers.json.JacksonSerializer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/orchestrator/docker/DockerContainerManager.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/orchestrator/DockerOrchestratorManager.java similarity index 79% rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/orchestrator/docker/DockerContainerManager.java rename to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/orchestrator/DockerOrchestratorManager.java index dea27b7..2dec572 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/orchestrator/docker/DockerContainerManager.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/orchestrator/DockerOrchestratorManager.java @@ -15,41 +15,41 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.management.orchestrator.docker; +package org.apache.streampipes.node.controller.management.orchestrator; import com.fasterxml.jackson.core.JsonProcessingException; -import com.google.gson.Gson; import com.spotify.docker.client.exceptions.DockerException; import com.spotify.docker.client.exceptions.NotFoundException; 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.container.management.node.NodeManager; -import org.apache.streampipes.node.controller.container.management.orchestrator.ContainerOrchestrator; -import org.apache.streampipes.node.controller.container.management.orchestrator.ContainerDeploymentStatus; -import org.apache.streampipes.node.controller.container.management.orchestrator.docker.utils.DockerUtils; +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; +import org.apache.streampipes.node.controller.management.orchestrator.docker.utils.DockerUtils; +import org.apache.streampipes.node.controller.management.orchestrator.status.ContainerDeploymentStatus; import org.apache.streampipes.serializers.json.JacksonSerializer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.*; -public class DockerContainerManager implements ContainerOrchestrator { +public class DockerOrchestratorManager implements IContainerOrchestrator { private static final Logger LOG = - LoggerFactory.getLogger(DockerContainerManager.class.getCanonicalName()); + LoggerFactory.getLogger(DockerOrchestratorManager.class.getCanonicalName()); private final DockerUtils docker = DockerUtils.getInstance(); - private static DockerContainerManager instance = null; + private static DockerOrchestratorManager instance = null; - private DockerContainerManager() {} + private DockerOrchestratorManager() {} - public static DockerContainerManager getInstance() { + public static DockerOrchestratorManager getInstance() { if (instance == null) { - synchronized (DockerContainerManager.class) { + synchronized (DockerOrchestratorManager.class) { if (instance == null) - instance = new DockerContainerManager(); + instance = new DockerOrchestratorManager(); } } return instance; @@ -57,10 +57,10 @@ public class DockerContainerManager implements ContainerOrchestrator { @Override public void init() { - DockerConstants.NodeContainer.INSTANCE.getAllStreamPipesContainer().forEach(container -> { + DockerContainerDeclarerSingleton.getInstance().getAllDockerContainerAsList().forEach(container -> { ContainerDeploymentStatus status = deploy(container); - if (status.getStatus() == DockerConstants.ContainerStatus.DEPLOYED) { + if (status.getStatus() == ContainerStatus.DEPLOYED) { NodeManager.getInstance().addToRegisteredContainers(status.getContainer()); } }); @@ -78,13 +78,13 @@ public class DockerContainerManager implements ContainerOrchestrator { containerId = deployPipelineElementContainer(container); } catch (Exception e) { LOG.error("Could not deploy pipeline element. {}", e.toString()); - return generateContainerStatus(containerId, container, DockerConstants.ContainerStatus.FAILED); + return generateContainerStatus(containerId, container, ContainerStatus.FAILED); } LOG.info("Finished pull image and deployed pipeline element container"); - return generateContainerStatus(containerId, container, DockerConstants.ContainerStatus.DEPLOYED); + return generateContainerStatus(containerId, container, ContainerStatus.DEPLOYED); } LOG.info("Container already running {}", container.getContainerName()); - return generateContainerStatus(containerOptional.get().id(), container, DockerConstants.ContainerStatus.RUNNING); + return generateContainerStatus(containerOptional.get().id(), container, ContainerStatus.RUNNING); } @Override @@ -100,12 +100,10 @@ public class DockerContainerManager implements ContainerOrchestrator { ConsulUtil.deregisterService(container.getServiceId()); ConsulUtil.deleteConfig(container.getServiceId()); - return generateContainerStatus(containerOptional.get().id(), container, - DockerConstants.ContainerStatus.REMOVED); + return generateContainerStatus(containerOptional.get().id(), container, ContainerStatus.REMOVED); } - return generateContainerStatus(containerOptional.get().id(), container, - DockerConstants.ContainerStatus.UNKNOWN); + return generateContainerStatus(containerOptional.get().id(), container, ContainerStatus.UNKNOWN); } @Override @@ -154,7 +152,7 @@ public class DockerContainerManager implements ContainerOrchestrator { } private ContainerDeploymentStatus generateContainerStatus(String containerId, DockerContainer container, - DockerConstants.ContainerStatus containerStatus) { + ContainerStatus containerStatus) { ContainerDeploymentStatus status = new ContainerDeploymentStatus(); status.setTimestamp(System.currentTimeMillis()); diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/orchestrator/ContainerOrchestrator.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/orchestrator/IContainerOrchestrator.java similarity index 82% rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/orchestrator/ContainerOrchestrator.java rename to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/orchestrator/IContainerOrchestrator.java index 9fe151a..5a53b92 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/orchestrator/ContainerOrchestrator.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/orchestrator/IContainerOrchestrator.java @@ -15,11 +15,12 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.management.orchestrator; +package org.apache.streampipes.node.controller.management.orchestrator; import org.apache.streampipes.model.node.container.DockerContainer; +import org.apache.streampipes.node.controller.management.orchestrator.status.ContainerDeploymentStatus; -public interface ContainerOrchestrator { +public interface IContainerOrchestrator { void init(); diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/orchestrator/docker/AbstractStreamPipesDockerContainer.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/orchestrator/docker/AbstractStreamPipesDockerContainer.java new file mode 100644 index 0000000..af14b5d --- /dev/null +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/orchestrator/docker/AbstractStreamPipesDockerContainer.java @@ -0,0 +1,56 @@ +/* + * 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.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.config.NodeControllerConfig; +import org.apache.streampipes.node.controller.management.node.NodeManager; + +import java.util.*; + +public abstract class AbstractStreamPipesDockerContainer { + + + protected abstract DockerContainer declareDockerContainer(); + + public static String getStreamPipesVersion() { + return !NodeControllerConfig.INSTANCE.getSpVersion().equals("") ? + NodeControllerConfig.INSTANCE.getSpVersion() : NodeManager.getInstance().getStreamPipesVersion(); + } + + public List<String> generateStreamPipesNodeEnvs() { + return new ArrayList<>(Arrays.asList( + toEnv(ConfigKeys.NODE_CONTROLLER_ID, NodeControllerConfig.INSTANCE.getNodeControllerId()), + toEnv(ConfigKeys.NODE_CONTROLLER_CONTAINER_HOST, NodeControllerConfig.INSTANCE.getNodeHost()), + toEnv( ConfigKeys.NODE_CONTROLLER_CONTAINER_PORT, NodeControllerConfig.INSTANCE.getNodeControllerPort()), + toEnv(ConfigKeys.NODE_BROKER_CONTAINER_HOST, NodeControllerConfig.INSTANCE.getNodeBrokerHost()), + toEnv(ConfigKeys.NODE_BROKER_CONTAINER_PORT, NodeControllerConfig.INSTANCE.getNodeBrokerPort()) + )); + } + + public String retrieveNodeType(){ + return NodeControllerConfig.INSTANCE.getNodeType(); + } + + // Helper + + private <T>String toEnv(String key, T value) { + return String.format("%s=%s", key, value); + } +} 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 new file mode 100644 index 0000000..ad17d07 --- /dev/null +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/orchestrator/docker/DockerContainerDeclarerSingleton.java @@ -0,0 +1,63 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ +package org.apache.streampipes.node.controller.management.orchestrator.docker; + + +import org.apache.streampipes.model.node.container.DockerContainer; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +public class DockerContainerDeclarerSingleton { + private static final Logger LOG = + LoggerFactory.getLogger(DockerContainerDeclarerSingleton.class.getCanonicalName()); + + private static DockerContainerDeclarerSingleton instance; + private final Map<String, DockerContainer> dockerContainers; + + private DockerContainerDeclarerSingleton() { + this.dockerContainers = new HashMap<>(); + } + + public static DockerContainerDeclarerSingleton getInstance() { + if (DockerContainerDeclarerSingleton.instance == null) { + DockerContainerDeclarerSingleton.instance = new DockerContainerDeclarerSingleton(); + } + return DockerContainerDeclarerSingleton.instance; + } + + public DockerContainerDeclarerSingleton register(AbstractStreamPipesDockerContainer container) { + LOG.info("Register container for auto-deploy: " + container.declareDockerContainer().getServiceId()); + this.dockerContainers.put( + container.declareDockerContainer().getServiceId(), + container.declareDockerContainer()); + return getInstance(); + } + + public Map<String, DockerContainer> getDockerContainers() { + return new HashMap<>(dockerContainers); + } + + public List<DockerContainer> getAllDockerContainerAsList() { + return new ArrayList<>(dockerContainers.values()); + } +} diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/bridges/EventRelayStrategy.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/orchestrator/docker/IDockerContainerDeclarer.java similarity index 77% copy from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/bridges/EventRelayStrategy.java copy to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/orchestrator/docker/IDockerContainerDeclarer.java index 8e2f508..29aee0b 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/bridges/EventRelayStrategy.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/orchestrator/docker/IDockerContainerDeclarer.java @@ -15,8 +15,10 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.management.relay.bridges; +package org.apache.streampipes.node.controller.management.orchestrator.docker; -public enum EventRelayStrategy { - BUFFER, PURGE +import org.apache.streampipes.model.node.container.DockerContainer; + +public interface IDockerContainerDeclarer { + DockerContainer declareDockerContainer(); } diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/bridges/EventRelayStrategy.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/orchestrator/docker/model/ContainerStatus.java similarity index 82% copy from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/bridges/EventRelayStrategy.java copy to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/orchestrator/docker/model/ContainerStatus.java index 8e2f508..0e933ca 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/bridges/EventRelayStrategy.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/orchestrator/docker/model/ContainerStatus.java @@ -15,8 +15,8 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.management.relay.bridges; +package org.apache.streampipes.node.controller.management.orchestrator.docker.model; -public enum EventRelayStrategy { - BUFFER, PURGE +public enum ContainerStatus { + DEPLOYED, RUNNING, STOPPED, REMOVED, UNKNOWN, FAILED } diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/orchestrator/docker/model/DockerInfo.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/orchestrator/docker/model/DockerInfo.java similarity index 96% rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/orchestrator/docker/model/DockerInfo.java rename to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/orchestrator/docker/model/DockerInfo.java index ee5939a..d362395 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/orchestrator/docker/model/DockerInfo.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/orchestrator/docker/model/DockerInfo.java @@ -15,7 +15,7 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.management.orchestrator.docker.model; +package org.apache.streampipes.node.controller.management.orchestrator.docker.model; public class DockerInfo { diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/orchestrator/docker/utils/DockerUtils.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/orchestrator/docker/utils/DockerUtils.java similarity index 87% rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/orchestrator/docker/utils/DockerUtils.java rename to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/orchestrator/docker/utils/DockerUtils.java index 5ee6a2b..f57dd98 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/orchestrator/docker/utils/DockerUtils.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/orchestrator/docker/utils/DockerUtils.java @@ -15,7 +15,7 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.management.orchestrator.docker.utils; +package org.apache.streampipes.node.controller.management.orchestrator.docker.utils; import com.google.gson.JsonObject; import com.google.gson.JsonParser; @@ -26,8 +26,7 @@ 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 org.apache.streampipes.model.node.container.DockerContainer; -import org.apache.streampipes.node.controller.container.management.orchestrator.docker.DockerConstants; -import org.apache.streampipes.node.controller.container.management.orchestrator.docker.model.DockerInfo; +import org.apache.streampipes.node.controller.management.orchestrator.docker.model.DockerInfo; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -42,6 +41,8 @@ public class DockerUtils { private static final String BLANK_SPACE = " "; private static final String DOCKER_UNIX_SOCK = "/var/run/docker.sock"; + private static final String SP_CONTAINER_NETWORK = "spnet"; + private static final String SP_CONTAINER_PREFIX = "streampipes-"; private static DockerUtils instance; private static DockerClient docker; @@ -128,8 +129,7 @@ public class DockerUtils { } private String verifyContainerName(String containerName) { - return containerName.startsWith(DockerConstants.SP_CONTAINER_PREFIX) ? - containerName : DockerConstants.SP_CONTAINER_PREFIX + containerName; + return containerName.startsWith(SP_CONTAINER_PREFIX) ? containerName : SP_CONTAINER_PREFIX + containerName; } private ContainerConfig getContainerConfig(DockerContainer p) { @@ -139,8 +139,8 @@ public class DockerUtils { .image(p.getImageUri()) .labels(p.getLabels()) .env(p.getEnvVars()) - .hostConfig(getHostConfig(DockerConstants.SP_CONTAINER_NETWORK, p.getContainerPorts())) - .networkingConfig(getNetworkingConfig(DockerConstants.SP_CONTAINER_NETWORK, p.getContainerName())) + .hostConfig(getHostConfig(SP_CONTAINER_NETWORK, p.getContainerPorts(), p.getVolumes())) + .networkingConfig(getNetworkingConfig(SP_CONTAINER_NETWORK, p.getContainerName())) .build(); } @@ -174,23 +174,35 @@ public class DockerUtils { } private static HostConfig getHostConfig(String network, String[] ports) { - Map<String, List<PortBinding>> portBindings = new HashMap<>(); - for (String port : ports) { - portBindings.put(port + "/tcp", Collections.singletonList(PortBinding.of("0.0.0.0", port))); - } - - return HostConfig.builder() - .portBindings(portBindings) - .networkMode(network) - .restartPolicy(HostConfig.RestartPolicy.unlessStopped()) - //.extraHosts() - .build(); + return getHostConfig(network, ports, null); } private static HostConfig getHostConfig(String network) { - return HostConfig.builder() - .networkMode(network) - .build(); + return getHostConfig(network, null, null); + } + + private static HostConfig getHostConfig(String network, String[] ports, List<String> volumes) { + 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))); + } + } + + if (volumes != null) { + return HostConfig.builder() + .portBindings(portBindings) + .networkMode(network) + .restartPolicy(HostConfig.RestartPolicy.unlessStopped()) + .appendBinds(volumes) + .build(); + } else { + return HostConfig.builder() + .portBindings(portBindings) + .networkMode(network) + .restartPolicy(HostConfig.RestartPolicy.unlessStopped()) + .build(); + } } private static ContainerConfig.NetworkingConfig getNetworkingConfig(String network, String containerName) { diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/orchestrator/ContainerDeploymentStatus.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/orchestrator/status/ContainerDeploymentStatus.java similarity index 81% rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/orchestrator/ContainerDeploymentStatus.java rename to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/orchestrator/status/ContainerDeploymentStatus.java index 6e7ba29..c6b2cb0 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/orchestrator/ContainerDeploymentStatus.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/orchestrator/status/ContainerDeploymentStatus.java @@ -15,23 +15,23 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.management.orchestrator; +package org.apache.streampipes.node.controller.management.orchestrator.status; import org.apache.streampipes.model.node.container.DockerContainer; -import org.apache.streampipes.node.controller.container.management.orchestrator.docker.DockerConstants; +import org.apache.streampipes.node.controller.management.orchestrator.docker.model.ContainerStatus; public class ContainerDeploymentStatus { private String containerId; private DockerContainer container; private long timestamp; - private DockerConstants.ContainerStatus status; + private ContainerStatus status; public ContainerDeploymentStatus() { } public ContainerDeploymentStatus(String containerId, DockerContainer container, long timestamp, - DockerConstants.ContainerStatus status) { + ContainerStatus status) { this.containerId = containerId; this.container = container; this.timestamp = timestamp; @@ -62,11 +62,11 @@ public class ContainerDeploymentStatus { this.timestamp = timestamp; } - public DockerConstants.ContainerStatus getStatus() { + public ContainerStatus getStatus() { return status; } - public void setStatus(DockerConstants.ContainerStatus status) { + public void setStatus(ContainerStatus status) { this.status = status; } } diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/orchestrator/RunningContainerInstances.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/orchestrator/storage/RunningContainerInstances.java similarity index 81% rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/orchestrator/RunningContainerInstances.java rename to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/orchestrator/storage/RunningContainerInstances.java index f1fdb63..f6404d5 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/orchestrator/RunningContainerInstances.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/orchestrator/storage/RunningContainerInstances.java @@ -15,18 +15,16 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.management.orchestrator; +package org.apache.streampipes.node.controller.management.orchestrator.storage; import org.apache.streampipes.model.node.container.DockerContainer; -import org.apache.streampipes.node.controller.container.management.RunningInstances; -import org.apache.streampipes.node.controller.container.storage.MapDBImpl; +import org.apache.streampipes.node.controller.storage.IRunningInstances; +import org.apache.streampipes.node.controller.storage.MapDBImpl; import java.io.File; -import java.util.HashMap; import java.util.List; -import java.util.Map; -public enum RunningContainerInstances implements RunningInstances<DockerContainer> { +public enum RunningContainerInstances implements IRunningInstances<DockerContainer> { INSTANCE; private final MapDBImpl mapDB; diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/pe/PipelineElementLifeCycle.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/pe/IPipelineElementLifeCycle.java similarity index 92% rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/pe/PipelineElementLifeCycle.java rename to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/pe/IPipelineElementLifeCycle.java index 5bdc6e6..f8d35e9 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/pe/PipelineElementLifeCycle.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/pe/IPipelineElementLifeCycle.java @@ -15,14 +15,14 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.management.pe; +package org.apache.streampipes.node.controller.management.pe; import org.apache.streampipes.container.model.node.InvocableRegistration; import org.apache.streampipes.model.Response; import org.apache.streampipes.model.base.InvocableStreamPipesEntity; import org.apache.streampipes.model.pipeline.PipelineElementReconfigurationEntity; -public interface PipelineElementLifeCycle { +public interface IPipelineElementLifeCycle { void register(InvocableRegistration registration); diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/pe/InvocableElementManager.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/pe/InvocableElementManager.java similarity index 97% rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/pe/InvocableElementManager.java rename to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/pe/InvocableElementManager.java index 5d1f833..abc4841 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/pe/InvocableElementManager.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/pe/InvocableElementManager.java @@ -15,7 +15,7 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.management.pe; +package org.apache.streampipes.node.controller.management.pe; import com.fasterxml.jackson.core.JsonProcessingException; import com.google.gson.JsonSyntaxException; @@ -35,8 +35,9 @@ import org.apache.streampipes.model.node.NodeInfoDescription; import org.apache.streampipes.model.pipeline.PipelineElementReconfigurationEntity; import org.apache.streampipes.model.staticproperty.FreeTextStaticProperty; import org.apache.streampipes.model.staticproperty.StaticProperty; -import org.apache.streampipes.node.controller.container.config.NodeControllerConfig; -import org.apache.streampipes.node.controller.container.management.node.NodeManager; +import org.apache.streampipes.node.controller.config.NodeControllerConfig; +import org.apache.streampipes.node.controller.management.node.NodeManager; +import org.apache.streampipes.node.controller.management.pe.storage.RunningInvocableInstances; import org.apache.streampipes.serializers.json.JacksonSerializer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -49,7 +50,7 @@ import java.net.URL; import java.nio.charset.StandardCharsets; import java.util.*; -public class InvocableElementManager implements PipelineElementLifeCycle { +public class InvocableElementManager implements IPipelineElementLifeCycle { private static final Logger LOG = LoggerFactory.getLogger(InvocableElementManager.class.getCanonicalName()); diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/pe/RunningInvocableInstances.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/pe/storage/RunningInvocableInstances.java similarity index 83% rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/pe/RunningInvocableInstances.java rename to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/pe/storage/RunningInvocableInstances.java index 6c7d923..14937f0 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/pe/RunningInvocableInstances.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/pe/storage/RunningInvocableInstances.java @@ -15,16 +15,16 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.management.pe; +package org.apache.streampipes.node.controller.management.pe.storage; import org.apache.streampipes.model.base.InvocableStreamPipesEntity; -import org.apache.streampipes.node.controller.container.storage.MapDBImpl; -import org.apache.streampipes.node.controller.container.management.RunningInstances; +import org.apache.streampipes.node.controller.storage.MapDBImpl; +import org.apache.streampipes.node.controller.storage.IRunningInstances; import java.io.File; import java.util.List; -public enum RunningInvocableInstances implements RunningInstances<InvocableStreamPipesEntity> { +public enum RunningInvocableInstances implements IRunningInstances<InvocableStreamPipesEntity> { INSTANCE; private final MapDBImpl mapDB; diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/DataStreamRelayManager.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/relay/DataStreamRelayManager.java similarity index 93% rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/DataStreamRelayManager.java rename to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/relay/DataStreamRelayManager.java index 0fbe538..ef8f07a 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/DataStreamRelayManager.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/relay/DataStreamRelayManager.java @@ -15,7 +15,7 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.management.relay; +package org.apache.streampipes.node.controller.management.relay; import org.apache.streampipes.commons.exceptions.SpRuntimeException; import org.apache.streampipes.model.Response; @@ -25,6 +25,8 @@ import org.apache.streampipes.model.eventrelay.SpDataStreamRelayContainer; import org.apache.streampipes.model.graph.DataProcessorInvocation; import org.apache.streampipes.model.grounding.TransportProtocol; import org.apache.streampipes.model.eventrelay.metrics.RelayMetrics; +import org.apache.streampipes.node.controller.management.relay.base.EventRelay; +import org.apache.streampipes.node.controller.management.relay.storage.RunningRelayInstances; import java.util.*; @@ -59,7 +61,7 @@ public class DataStreamRelayManager { // start data stream relay // 1:1 mapping -> remote forward - // 1:n mamping -> remote fan-out + // 1:n mapping -> remote fan-out desc.getOutputStreamRelays().forEach(relay -> { // add new relay if not running if(!runningRelayExists(id, relay)) { diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/BaseEventRelay.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/relay/base/AbstractEventRelay.java similarity index 92% rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/BaseEventRelay.java rename to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/relay/base/AbstractEventRelay.java index b1f7aa1..1b85eba 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/BaseEventRelay.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/relay/base/AbstractEventRelay.java @@ -15,19 +15,19 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.management.relay; +package org.apache.streampipes.node.controller.management.relay.base; import org.apache.streampipes.model.grounding.JmsTransportProtocol; 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.node.controller.container.management.relay.bridges.*; +import org.apache.streampipes.node.controller.management.relay.bridges.*; -public abstract class BaseEventRelay { +public abstract class AbstractEventRelay { protected final MultiBrokerBridge<?,?> multiBrokerBridge; - public BaseEventRelay(TransportProtocol source, TransportProtocol target, String relayStrategy) { + public AbstractEventRelay(TransportProtocol source, TransportProtocol target, String relayStrategy) { this.multiBrokerBridge = getBrokerToBrokerRelaySetup(source, target, relayStrategy); } diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/EventRelay.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/relay/base/EventRelay.java similarity index 93% rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/EventRelay.java rename to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/relay/base/EventRelay.java index f22a815..d1423c2 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/EventRelay.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/relay/base/EventRelay.java @@ -15,13 +15,13 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.management.relay; +package org.apache.streampipes.node.controller.management.relay.base; import org.apache.streampipes.commons.exceptions.SpRuntimeException; import org.apache.streampipes.model.grounding.*; import org.apache.streampipes.model.eventrelay.metrics.RelayMetrics; -public class EventRelay extends BaseEventRelay { +public class EventRelay extends AbstractEventRelay { private static final String DEFAULT_EVENT_RELAY_STRATEGY = "buffer"; diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/bridges/EventRelayStrategy.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/relay/bridges/EventRelayStrategy.java similarity index 91% rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/bridges/EventRelayStrategy.java rename to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/relay/bridges/EventRelayStrategy.java index 8e2f508..18d45fa 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/bridges/EventRelayStrategy.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/relay/bridges/EventRelayStrategy.java @@ -15,7 +15,7 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.management.relay.bridges; +package org.apache.streampipes.node.controller.management.relay.bridges; public enum EventRelayStrategy { BUFFER, PURGE diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/bridges/JmsToJmsBridge.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/relay/bridges/JmsToJmsBridge.java similarity index 95% rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/bridges/JmsToJmsBridge.java rename to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/relay/bridges/JmsToJmsBridge.java index 67b0192..94ab702 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/bridges/JmsToJmsBridge.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/relay/bridges/JmsToJmsBridge.java @@ -15,7 +15,7 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.management.relay.bridges; +package org.apache.streampipes.node.controller.management.relay.bridges; import org.apache.streampipes.messaging.jms.ActiveMQConsumer; import org.apache.streampipes.messaging.jms.ActiveMQPublisher; diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/bridges/JmsToKafkaBridge.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/relay/bridges/JmsToKafkaBridge.java similarity index 95% rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/bridges/JmsToKafkaBridge.java rename to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/relay/bridges/JmsToKafkaBridge.java index 5c5e8af..46c3747 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/bridges/JmsToKafkaBridge.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/relay/bridges/JmsToKafkaBridge.java @@ -15,7 +15,7 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.management.relay.bridges; +package org.apache.streampipes.node.controller.management.relay.bridges; import org.apache.streampipes.messaging.jms.ActiveMQConsumer; import org.apache.streampipes.messaging.kafka.SpKafkaProducer; diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/bridges/JmsToMqttBridge.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/relay/bridges/JmsToMqttBridge.java similarity index 95% rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/bridges/JmsToMqttBridge.java rename to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/relay/bridges/JmsToMqttBridge.java index 839b014..fc5ddee 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/bridges/JmsToMqttBridge.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/relay/bridges/JmsToMqttBridge.java @@ -15,7 +15,7 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.management.relay.bridges; +package org.apache.streampipes.node.controller.management.relay.bridges; import org.apache.streampipes.messaging.jms.ActiveMQConsumer; import org.apache.streampipes.messaging.mqtt.MqttPublisher; diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/bridges/KafkaToJmsBridge.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/relay/bridges/KafkaToJmsBridge.java similarity index 95% rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/bridges/KafkaToJmsBridge.java rename to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/relay/bridges/KafkaToJmsBridge.java index c6ea274..1b613a0 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/bridges/KafkaToJmsBridge.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/relay/bridges/KafkaToJmsBridge.java @@ -15,7 +15,7 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.management.relay.bridges; +package org.apache.streampipes.node.controller.management.relay.bridges; import org.apache.streampipes.messaging.jms.ActiveMQPublisher; import org.apache.streampipes.messaging.kafka.SpKafkaConsumer; diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/bridges/KafkaToKafkaBridge.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/relay/bridges/KafkaToKafkaBridge.java similarity index 95% rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/bridges/KafkaToKafkaBridge.java rename to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/relay/bridges/KafkaToKafkaBridge.java index 6ee1fd8..034a89a 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/bridges/KafkaToKafkaBridge.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/relay/bridges/KafkaToKafkaBridge.java @@ -15,7 +15,7 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.management.relay.bridges; +package org.apache.streampipes.node.controller.management.relay.bridges; import org.apache.streampipes.messaging.kafka.SpKafkaConsumer; import org.apache.streampipes.messaging.kafka.SpKafkaProducer; diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/bridges/KafkaToMqttBridge.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/relay/bridges/KafkaToMqttBridge.java similarity index 95% rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/bridges/KafkaToMqttBridge.java rename to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/relay/bridges/KafkaToMqttBridge.java index ab67174..eb77a52 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/bridges/KafkaToMqttBridge.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/relay/bridges/KafkaToMqttBridge.java @@ -15,7 +15,7 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.management.relay.bridges; +package org.apache.streampipes.node.controller.management.relay.bridges; import org.apache.streampipes.messaging.kafka.SpKafkaConsumer; import org.apache.streampipes.messaging.mqtt.MqttPublisher; diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/bridges/MqttToJmsBridge.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/relay/bridges/MqttToJmsBridge.java similarity index 95% rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/bridges/MqttToJmsBridge.java rename to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/relay/bridges/MqttToJmsBridge.java index 13d357d..2d4b7e0 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/bridges/MqttToJmsBridge.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/relay/bridges/MqttToJmsBridge.java @@ -15,7 +15,7 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.management.relay.bridges; +package org.apache.streampipes.node.controller.management.relay.bridges; import org.apache.streampipes.messaging.jms.ActiveMQPublisher; import org.apache.streampipes.messaging.mqtt.MqttConsumer; diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/bridges/MqttToKafkaBridge.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/relay/bridges/MqttToKafkaBridge.java similarity index 95% rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/bridges/MqttToKafkaBridge.java rename to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/relay/bridges/MqttToKafkaBridge.java index 22ec587..90bd6c6 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/bridges/MqttToKafkaBridge.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/relay/bridges/MqttToKafkaBridge.java @@ -15,7 +15,7 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.management.relay.bridges; +package org.apache.streampipes.node.controller.management.relay.bridges; import org.apache.streampipes.messaging.kafka.SpKafkaProducer; import org.apache.streampipes.messaging.mqtt.MqttConsumer; diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/bridges/MqttToMqttBridge.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/relay/bridges/MqttToMqttBridge.java similarity index 95% rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/bridges/MqttToMqttBridge.java rename to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/relay/bridges/MqttToMqttBridge.java index c5a7f5d..40fd631 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/bridges/MqttToMqttBridge.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/relay/bridges/MqttToMqttBridge.java @@ -15,7 +15,7 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.management.relay.bridges; +package org.apache.streampipes.node.controller.management.relay.bridges; import org.apache.streampipes.messaging.mqtt.MqttConsumer; import org.apache.streampipes.messaging.mqtt.MqttPublisher; diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/bridges/MultiBrokerBridge.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/relay/bridges/MultiBrokerBridge.java similarity index 97% rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/bridges/MultiBrokerBridge.java rename to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/relay/bridges/MultiBrokerBridge.java index 09b9057..e278c40 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/bridges/MultiBrokerBridge.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/relay/bridges/MultiBrokerBridge.java @@ -15,7 +15,7 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.management.relay.bridges; +package org.apache.streampipes.node.controller.management.relay.bridges; import org.apache.streampipes.commons.exceptions.SpRuntimeException; import org.apache.streampipes.messaging.EventConsumer; @@ -23,7 +23,7 @@ import org.apache.streampipes.messaging.EventProducer; import org.apache.streampipes.messaging.EventRelay; import org.apache.streampipes.model.grounding.*; import org.apache.streampipes.model.eventrelay.metrics.RelayMetrics; -import org.apache.streampipes.node.controller.container.config.NodeControllerConfig; +import org.apache.streampipes.node.controller.config.NodeControllerConfig; import org.apache.streampipes.sdk.helpers.Tuple3; import org.slf4j.Logger; import org.slf4j.LoggerFactory; diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/RunningRelayInstances.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/relay/storage/RunningRelayInstances.java similarity index 85% rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/RunningRelayInstances.java rename to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/relay/storage/RunningRelayInstances.java index 63baf7b..8b81e73 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/relay/RunningRelayInstances.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/relay/storage/RunningRelayInstances.java @@ -15,14 +15,15 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.management.relay; +package org.apache.streampipes.node.controller.management.relay.storage; -import org.apache.streampipes.node.controller.container.management.RunningInstances; +import org.apache.streampipes.node.controller.storage.IRunningInstances; +import org.apache.streampipes.node.controller.management.relay.base.EventRelay; import java.util.*; import java.util.stream.Collectors; -public enum RunningRelayInstances implements RunningInstances<Map<String,EventRelay>> { +public enum RunningRelayInstances implements IRunningInstances<Map<String, EventRelay>> { INSTANCE; private final Map<String, Map<String, EventRelay>> runningInstances = new HashMap<>(); diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/resource/ResourceManager.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/resource/ResourceManager.java similarity index 95% rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/resource/ResourceManager.java rename to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/resource/ResourceManager.java index e9d49e4..d274fba 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/resource/ResourceManager.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/resource/ResourceManager.java @@ -15,13 +15,11 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.management.resource; +package org.apache.streampipes.node.controller.management.resource; import com.fasterxml.jackson.core.JsonProcessingException; -import com.google.gson.Gson; -import org.apache.streampipes.commons.exceptions.SpRuntimeException; -import org.apache.streampipes.node.controller.container.config.NodeControllerConfig; -import org.apache.streampipes.node.controller.container.management.resource.model.ResourceMetrics; +import org.apache.streampipes.node.controller.config.NodeControllerConfig; +import org.apache.streampipes.node.controller.management.resource.model.ResourceMetrics; import org.apache.streampipes.serializers.json.JacksonSerializer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/resource/model/ResourceMetrics.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/resource/model/ResourceMetrics.java similarity index 97% rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/resource/model/ResourceMetrics.java rename to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/resource/model/ResourceMetrics.java index 3595895..ddaaff6 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/resource/model/ResourceMetrics.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/resource/model/ResourceMetrics.java @@ -15,7 +15,7 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.management.resource.model; +package org.apache.streampipes.node.controller.management.resource.model; public class ResourceMetrics { diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/storage/CRUDStorage.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/storage/CRUDStorage.java similarity index 94% rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/storage/CRUDStorage.java rename to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/storage/CRUDStorage.java index 0937366..e553bf6 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/storage/CRUDStorage.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/storage/CRUDStorage.java @@ -15,7 +15,7 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.storage; +package org.apache.streampipes.node.controller.storage; import java.util.List; diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/RunningInstances.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/storage/IRunningInstances.java similarity index 90% rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/RunningInstances.java rename to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/storage/IRunningInstances.java index 934f917..54b33f6 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/RunningInstances.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/storage/IRunningInstances.java @@ -15,11 +15,11 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.management; +package org.apache.streampipes.node.controller.storage; import java.util.List; -public interface RunningInstances<T> { +public interface IRunningInstances<T> { void add(String id, T value); diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/storage/MapDBImpl.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/storage/MapDBImpl.java similarity index 97% rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/storage/MapDBImpl.java rename to streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/storage/MapDBImpl.java index 6824c87..ebe5518 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/storage/MapDBImpl.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/storage/MapDBImpl.java @@ -15,7 +15,7 @@ * limitations under the License. * */ -package org.apache.streampipes.node.controller.container.storage; +package org.apache.streampipes.node.controller.storage; import org.mapdb.DB; import org.mapdb.DBMaker; 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 7570cf7..4ab49c0 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 @@ -368,6 +368,7 @@ public class StreamPipes { public static final String DEPLOYMENT_CONTAINER_PORTS = NS + "dockerContainerPorts"; 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"; // UI Rendering
