This is an automated email from the ASF dual-hosted git repository. riemer pushed a commit to branch STREAMPIPES-319 in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit 951ed472e765a7d6a25cbe746b31055bbdb0d0db Author: Dominik Riemer <[email protected]> AuthorDate: Tue Jul 13 09:57:03 2021 +0200 [STREAMPIPES-319] Refactor configuration management --- .../org/apache/streampipes/config/SpConfig.java | 82 ----------- .../streampipes/config/backend/BackendConfig.java | 5 +- .../streampipes/config/consul/MockSpConfig.java | 125 ----------------- .../worker/init/AdapterWorkerContainer.java | 2 +- .../connect/adapter/GroundingServiceTest.java | 23 ++- .../extensions/ExtensionsModelSubmitter.java | 2 +- .../standalone/init/StandaloneModelSubmitter.java | 2 +- .../api/InvocablePipelineElementResource.java | 8 +- .../container/config/ConfigExtractor.java | 4 +- .../container/init/DeclarersSingleton.java | 28 +++- .../container/model/SpServiceDefinition.java | 11 ++ .../model/SpServiceDefinitionBuilder.java | 28 +++- .../apache/streampipes/rest/impl/ConsulConfig.java | 5 +- .../svcdiscovery/api/ISpServiceDiscovery.java | 2 +- .../streampipes/svcdiscovery/api/SpConfig.java | 64 +++++++++ .../svcdiscovery/api/model/ConfigItem.java | 154 +++++++++++++-------- .../svcdiscovery/api/model/ConfigItemUtils.java | 17 ++- .../svcdiscovery/api/model/SpServiceTagPrefix.java | 1 + .../svcdiscovery}/consul/ConsulSpConfig.java | 75 +++------- .../consul/SpConsulServiceDiscovery.java | 22 +-- .../svcdiscovery/SpServiceDiscovery.java | 7 + .../base/StreamPipesExtensionsServiceBase.java | 14 ++ streampipes-storage-couchdb/pom.xml | 4 +- .../storage/couchdb/utils/CouchDbConfig.java | 5 +- streampipes-user-management/pom.xml | 5 + .../consul-configs/consul-configs.component.html | 2 +- .../pipeline-element-configuration.component.html | 5 +- 27 files changed, 318 insertions(+), 384 deletions(-) diff --git a/streampipes-config/src/main/java/org/apache/streampipes/config/SpConfig.java b/streampipes-config/src/main/java/org/apache/streampipes/config/SpConfig.java deleted file mode 100644 index 5167061..0000000 --- a/streampipes-config/src/main/java/org/apache/streampipes/config/SpConfig.java +++ /dev/null @@ -1,82 +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.config; - -import org.apache.streampipes.config.consul.ConsulSpConfig; -import org.apache.streampipes.svcdiscovery.api.model.ConfigItem; -import org.apache.streampipes.svcdiscovery.api.model.ConfigurationScope; -import org.apache.streampipes.svcdiscovery.consul.AbstractConsulService; - - -public abstract class SpConfig extends AbstractConsulService { - - - public SpConfig(String name) { - - } - - public SpConfig(String serviceName, SpConfigChangeCallback callback) { - - } - - public static SpConfig getSpConfig(String serviceName) { - return new ConsulSpConfig(serviceName); - } - - public static SpConfig getSpConfig(String serviceName, SpConfigChangeCallback callback) { - return new ConsulSpConfig(serviceName, callback); - } - - public abstract <T> void register(String key, T defaultValue, String description, ConfigurationScope configurationScope); - - public abstract void register(String key, boolean defaultValue, String description); - - public abstract void register(String key, int defaultValue, String description); - - public abstract void register(String key, double defaultValue, String description); - - public abstract void register(String key, String defaultValue, String description); - - public abstract void registerObject(String key, Object defaultValue, String description); - - public abstract void registerPassword(String key, String defaultValue, String description); - - public abstract boolean getBoolean(String key); - - public abstract int getInteger(String key); - - public abstract double getDouble(String key); - - public abstract String getString(String key); - - public abstract <T> T getObject(String key, Class<T> clazz, T defaultValue); - - public abstract ConfigItem getConfigItem(String key); - - public abstract void setBoolean(String key, Boolean value); - - public abstract void setInteger(String key, int value); - - public abstract void setDouble(String key, double value); - - public abstract void setString(String key, String value); - - public abstract void setObject(String key, Object value); - -} diff --git a/streampipes-config/src/main/java/org/apache/streampipes/config/backend/BackendConfig.java b/streampipes-config/src/main/java/org/apache/streampipes/config/backend/BackendConfig.java index 6540c8e..a1e8db6 100644 --- a/streampipes-config/src/main/java/org/apache/streampipes/config/backend/BackendConfig.java +++ b/streampipes-config/src/main/java/org/apache/streampipes/config/backend/BackendConfig.java @@ -20,7 +20,8 @@ package org.apache.streampipes.config.backend; import org.apache.commons.lang3.RandomStringUtils; -import org.apache.streampipes.config.SpConfig; +import org.apache.streampipes.svcdiscovery.SpServiceDiscovery; +import org.apache.streampipes.svcdiscovery.api.SpConfig; import java.io.File; import java.security.SecureRandom; @@ -32,7 +33,7 @@ public enum BackendConfig { private SpConfig config; BackendConfig() { - config = SpConfig.getSpConfig("backend"); + config = SpServiceDiscovery.getSpConfig("backend"); config.register(BackendConfigKeys.SERVICE_NAME, "Backend", "Backend Configuration"); diff --git a/streampipes-config/src/main/java/org/apache/streampipes/config/consul/MockSpConfig.java b/streampipes-config/src/main/java/org/apache/streampipes/config/consul/MockSpConfig.java deleted file mode 100644 index b7b5b6c..0000000 --- a/streampipes-config/src/main/java/org/apache/streampipes/config/consul/MockSpConfig.java +++ /dev/null @@ -1,125 +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.config.consul; - -import org.apache.streampipes.config.SpConfig; -import org.apache.streampipes.config.SpConfigChangeCallback; -import org.apache.streampipes.svcdiscovery.api.model.ConfigItem; -import org.apache.streampipes.svcdiscovery.api.model.ConfigurationScope; - -public class MockSpConfig extends SpConfig { - - public MockSpConfig(String name) { - super(name); - } - - public MockSpConfig(String serviceName, SpConfigChangeCallback callback) { - super(serviceName, callback); - } - - @Override - public <T> void register(String key, T defaultValue, String description, ConfigurationScope configurationScope) { - - } - - @Override - public void register(String key, boolean defaultValue, String description) { - - } - - @Override - public void register(String key, int defaultValue, String description) { - - } - - @Override - public void register(String key, double defaultValue, String description) { - - } - - @Override - public void register(String key, String defaultValue, String description) { - - } - - @Override - public void registerObject(String key, Object defaultValue, String description) { - - } - - @Override - public void registerPassword(String key, String defaultValue, String description) { - - } - - @Override - public boolean getBoolean(String key) { - return false; - } - - @Override - public int getInteger(String key) { - return 0; - } - - @Override - public double getDouble(String key) { - return 0; - } - - @Override - public String getString(String key) { - return null; - } - - @Override - public <T> T getObject(String key, Class<T> clazz, T defaultValue) { - return null; - } - - @Override - public ConfigItem getConfigItem(String key) { - return null; - } - - @Override - public void setBoolean(String key, Boolean value) { - - } - - @Override - public void setInteger(String key, int value) { - - } - - @Override - public void setDouble(String key, double value) { - - } - - @Override - public void setString(String key, String value) { - - } - - @Override - public void setObject(String key, Object value) { - - } -} diff --git a/streampipes-connect-container-worker/src/main/java/org/apache/streampipes/connect/container/worker/init/AdapterWorkerContainer.java b/streampipes-connect-container-worker/src/main/java/org/apache/streampipes/connect/container/worker/init/AdapterWorkerContainer.java index 4f434f8..9b88cc9 100644 --- a/streampipes-connect-container-worker/src/main/java/org/apache/streampipes/connect/container/worker/init/AdapterWorkerContainer.java +++ b/streampipes-connect-container-worker/src/main/java/org/apache/streampipes/connect/container/worker/init/AdapterWorkerContainer.java @@ -43,7 +43,7 @@ public abstract class AdapterWorkerContainer extends StreamPipesExtensionsServic } @Override - protected List<SpServiceTag> getServiceTags() { + protected List<SpServiceTag> getExtensionsServiceTags() { return new ConnectWorkerTagProvider().extractServiceTags(); } diff --git a/streampipes-connect/src/test/java/org/apache/streampipes/connect/adapter/GroundingServiceTest.java b/streampipes-connect/src/test/java/org/apache/streampipes/connect/adapter/GroundingServiceTest.java index 61e0dfb..89bd069 100644 --- a/streampipes-connect/src/test/java/org/apache/streampipes/connect/adapter/GroundingServiceTest.java +++ b/streampipes-connect/src/test/java/org/apache/streampipes/connect/adapter/GroundingServiceTest.java @@ -18,17 +18,7 @@ package org.apache.streampipes.connect.adapter; -import org.junit.Before; -import org.junit.Test; -import org.junit.runner.RunWith; -import org.powermock.api.mockito.PowerMockito; -import org.powermock.core.classloader.annotations.PrepareForTest; -import org.powermock.modules.junit4.PowerMockRunner; -import org.powermock.reflect.Whitebox; -import org.apache.streampipes.config.SpConfig; import org.apache.streampipes.config.backend.BackendConfig; -import org.apache.streampipes.config.backend.MessagingSettings; -import org.apache.streampipes.config.consul.MockSpConfig; import org.apache.streampipes.model.SpDataSet; import org.apache.streampipes.model.connect.adapter.AdapterDescription; import org.apache.streampipes.model.connect.adapter.GenericAdapterSetDescription; @@ -38,12 +28,15 @@ import org.apache.streampipes.model.grounding.EventGrounding; import org.apache.streampipes.model.grounding.KafkaTransportProtocol; import org.apache.streampipes.model.grounding.SimpleTopicDefinition; import org.apache.streampipes.model.grounding.TopicDefinition; +import org.apache.streampipes.svcdiscovery.api.SpConfig; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.powermock.api.mockito.PowerMockito; +import org.powermock.core.classloader.annotations.PrepareForTest; +import org.powermock.modules.junit4.PowerMockRunner; import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertTrue; -import static org.mockito.ArgumentMatchers.anyString; -import static org.powermock.api.mockito.PowerMockito.mock; -import static org.powermock.api.mockito.PowerMockito.when; @RunWith(PowerMockRunner.class) @PrepareForTest({ BackendConfig.class, SpConfig.class}) @@ -167,4 +160,4 @@ public class GroundingServiceTest { } -} \ No newline at end of file +} diff --git a/streampipes-container-extensions/src/main/java/org/apache/streampipes/container/extensions/ExtensionsModelSubmitter.java b/streampipes-container-extensions/src/main/java/org/apache/streampipes/container/extensions/ExtensionsModelSubmitter.java index 82addbe..a95b185 100644 --- a/streampipes-container-extensions/src/main/java/org/apache/streampipes/container/extensions/ExtensionsModelSubmitter.java +++ b/streampipes-container-extensions/src/main/java/org/apache/streampipes/container/extensions/ExtensionsModelSubmitter.java @@ -54,7 +54,7 @@ public abstract class ExtensionsModelSubmitter extends StreamPipesExtensionsServ } @Override - protected List<SpServiceTag> getServiceTags() { + protected List<SpServiceTag> getExtensionsServiceTags() { List<SpServiceTag> serviceTags = new PipelineElementServiceTagProvider().extractServiceTags(); serviceTags.addAll(new ConnectWorkerTagProvider().extractServiceTags()); diff --git a/streampipes-container-standalone/src/main/java/org/apache/streampipes/container/standalone/init/StandaloneModelSubmitter.java b/streampipes-container-standalone/src/main/java/org/apache/streampipes/container/standalone/init/StandaloneModelSubmitter.java index 547aef9..bf5cae0 100644 --- a/streampipes-container-standalone/src/main/java/org/apache/streampipes/container/standalone/init/StandaloneModelSubmitter.java +++ b/streampipes-container-standalone/src/main/java/org/apache/streampipes/container/standalone/init/StandaloneModelSubmitter.java @@ -64,7 +64,7 @@ public abstract class StandaloneModelSubmitter extends StreamPipesExtensionsServ } @Override - protected List<SpServiceTag> getServiceTags() { + protected List<SpServiceTag> getExtensionsServiceTags() { return new PipelineElementServiceTagProvider().extractServiceTags(); } diff --git a/streampipes-container/src/main/java/org/apache/streampipes/container/api/InvocablePipelineElementResource.java b/streampipes-container/src/main/java/org/apache/streampipes/container/api/InvocablePipelineElementResource.java index f391685..2e4cebd 100644 --- a/streampipes-container/src/main/java/org/apache/streampipes/container/api/InvocablePipelineElementResource.java +++ b/streampipes-container/src/main/java/org/apache/streampipes/container/api/InvocablePipelineElementResource.java @@ -71,7 +71,7 @@ public abstract class InvocablePipelineElementResource<I extends InvocableStream String runningInstanceId = getInstanceId(graph.getElementId(), elementId); if (!RunningInstances.INSTANCE.exists(runningInstanceId)) { RunningInstances.INSTANCE.add(runningInstanceId, graph, declarer.getClass().newInstance()); - Response resp = RunningInstances.INSTANCE.getInvocation(runningInstanceId).invokeRuntime(graph, getServiceId()); + Response resp = RunningInstances.INSTANCE.getInvocation(runningInstanceId).invokeRuntime(graph, getServiceGroup()); if (!resp.isSuccess()) { LOG.error("Could not invoke pipeline element {} due to the following error: {}", graph.getName(), @@ -145,7 +145,7 @@ public abstract class InvocablePipelineElementResource<I extends InvocableStream InvocableDeclarer runningInstance = RunningInstances.INSTANCE.getInvocation(runningInstanceId); if (runningInstance != null) { - Response resp = runningInstance.detachRuntime(runningInstanceId, getServiceId()); + Response resp = runningInstance.detachRuntime(runningInstanceId, getServiceGroup()); if (resp.isSuccess()) { RunningInstances.INSTANCE.remove(runningInstanceId); @@ -172,8 +172,8 @@ public abstract class InvocablePipelineElementResource<I extends InvocableStream return "true".equals(System.getenv("SP_DEBUG")); } - private String getServiceId() { - return DeclarersSingleton.getInstance().getServiceId(); + private String getServiceGroup() { + return DeclarersSingleton.getInstance().getServiceDefinition().getServiceGroup(); } } diff --git a/streampipes-container/src/main/java/org/apache/streampipes/container/config/ConfigExtractor.java b/streampipes-container/src/main/java/org/apache/streampipes/container/config/ConfigExtractor.java index a43d9d6..aaff8e1 100644 --- a/streampipes-container/src/main/java/org/apache/streampipes/container/config/ConfigExtractor.java +++ b/streampipes-container/src/main/java/org/apache/streampipes/container/config/ConfigExtractor.java @@ -17,8 +17,8 @@ */ package org.apache.streampipes.container.config; -import org.apache.streampipes.config.SpConfig; -import org.apache.streampipes.config.consul.ConsulSpConfig; +import org.apache.streampipes.svcdiscovery.api.SpConfig; +import org.apache.streampipes.svcdiscovery.consul.ConsulSpConfig; public class ConfigExtractor { diff --git a/streampipes-container/src/main/java/org/apache/streampipes/container/init/DeclarersSingleton.java b/streampipes-container/src/main/java/org/apache/streampipes/container/init/DeclarersSingleton.java index 44c61d1..1a2ea26 100644 --- a/streampipes-container/src/main/java/org/apache/streampipes/container/init/DeclarersSingleton.java +++ b/streampipes-container/src/main/java/org/apache/streampipes/container/init/DeclarersSingleton.java @@ -30,7 +30,12 @@ import org.apache.streampipes.messaging.SpProtocolManager; import org.apache.streampipes.model.grounding.TransportFormat; import org.apache.streampipes.model.grounding.TransportProtocol; import org.apache.streampipes.model.util.Cloner; +import org.apache.streampipes.svcdiscovery.SpServiceDiscovery; +import org.apache.streampipes.svcdiscovery.api.SpConfig; +import org.apache.streampipes.svcdiscovery.api.model.ConfigItem; import org.apache.streampipes.vocabulary.StreamPipes; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import java.net.URI; import java.util.*; @@ -38,12 +43,16 @@ import java.util.stream.Collectors; public class DeclarersSingleton { + private static final Logger LOG = LoggerFactory.getLogger(DeclarersSingleton.class); + private static DeclarersSingleton instance; private static final String Http = "http://";; private static final String Colon = ":"; private static final String Slash = "/"; + private SpServiceDefinition serviceDefinition; + private Map<String, SemanticEventProcessingAgentDeclarer> epaDeclarers; private Map<String, SemanticEventConsumerDeclarer> consumerDeclarers; private Map<String, PipelineTemplateDeclarer> pipelineTemplateDeclarers; @@ -83,6 +92,7 @@ public class DeclarersSingleton { public void populate(String host, Integer port, SpServiceDefinition serviceDef) { + this.serviceDefinition = serviceDef; this.setHostName(host); this.setPort(port); this.addDeclarers(serviceDef.getDeclarers()); @@ -91,14 +101,20 @@ public class DeclarersSingleton { this.registerDataFormats(serviceDef.getDataFormatFactories()); this.allAdapters = serviceDef.getSpecificAdapters(); this.allProtocols = serviceDef.getAdapterProtocols(); + this.registerConfigs(serviceDef.getServiceGroup(), serviceDef.getServiceName(), serviceDef.getKvConfigs()); + } - this.serviceId = serviceDef.getServiceId(); + private void registerConfigs(String serviceGroup, + String serviceName, + Map<String, ConfigItem> configs) { + LOG.info("Registering {} configs in key/value store", configs.size()); + SpConfig spConfig = SpServiceDiscovery.getSpConfig(serviceGroup); + configs.values().forEach(spConfig::register); + spConfig.register(ConfigItem.from("SP_SERVICE_NAME", serviceName, "")); } public void addDeclarers(List<Declarer<?>> allDeclarers) { - for (Declarer<?> d : allDeclarers) { - add(d); - } + allDeclarers.forEach(this::add); } @Deprecated @@ -309,4 +325,8 @@ public class DeclarersSingleton { // TODO create complete service definition return serviceDef; } + + public SpServiceDefinition getServiceDefinition() { + return this.serviceDefinition; + } } diff --git a/streampipes-container/src/main/java/org/apache/streampipes/container/model/SpServiceDefinition.java b/streampipes-container/src/main/java/org/apache/streampipes/container/model/SpServiceDefinition.java index b45d4d7..dfe8e3f 100644 --- a/streampipes-container/src/main/java/org/apache/streampipes/container/model/SpServiceDefinition.java +++ b/streampipes-container/src/main/java/org/apache/streampipes/container/model/SpServiceDefinition.java @@ -22,6 +22,7 @@ import org.apache.streampipes.connect.api.IProtocol; import org.apache.streampipes.container.declarer.Declarer; import org.apache.streampipes.dataformat.SpDataFormatFactory; import org.apache.streampipes.messaging.SpProtocolDefinitionFactory; +import org.apache.streampipes.svcdiscovery.api.model.ConfigItem; import java.util.*; @@ -39,6 +40,7 @@ public class SpServiceDefinition { private Map<String, IProtocol> adapterProtocols; private Map<String, IAdapter> specificAdapters; + private Map<String, ConfigItem> kvConfigs; public SpServiceDefinition() { this.serviceId = UUID.randomUUID().toString(); @@ -47,6 +49,7 @@ public class SpServiceDefinition { this.protocolDefinitionFactories = new ArrayList<>(); this.adapterProtocols = new HashMap<>(); this.specificAdapters = new HashMap<>(); + this.kvConfigs = new HashMap<>(); } public String getServiceGroup() { @@ -121,6 +124,10 @@ public class SpServiceDefinition { protocols.values().forEach(this::addAdapterProtocol); } + public void addConfig(ConfigItem configItem) { + this.kvConfigs.put(configItem.getKey(), configItem); + } + public void addSpecificAdapter(IAdapter<?> adapter) { this.specificAdapters.put(adapter.getId(), adapter); } @@ -160,4 +167,8 @@ public class SpServiceDefinition { public Map<String, IAdapter> getSpecificAdapters() { return specificAdapters; } + + public Map<String, ConfigItem> getKvConfigs() { + return kvConfigs; + } } diff --git a/streampipes-container/src/main/java/org/apache/streampipes/container/model/SpServiceDefinitionBuilder.java b/streampipes-container/src/main/java/org/apache/streampipes/container/model/SpServiceDefinitionBuilder.java index eff9d86..3ad1308 100644 --- a/streampipes-container/src/main/java/org/apache/streampipes/container/model/SpServiceDefinitionBuilder.java +++ b/streampipes-container/src/main/java/org/apache/streampipes/container/model/SpServiceDefinitionBuilder.java @@ -17,21 +17,24 @@ */ package org.apache.streampipes.container.model; -import org.apache.streampipes.config.SpConfig; -import org.apache.streampipes.config.consul.ConsulSpConfig; import org.apache.streampipes.connect.api.Connector; import org.apache.streampipes.connect.api.IAdapter; import org.apache.streampipes.connect.api.IProtocol; import org.apache.streampipes.container.declarer.Declarer; import org.apache.streampipes.dataformat.SpDataFormatFactory; import org.apache.streampipes.messaging.SpProtocolDefinitionFactory; +import org.apache.streampipes.svcdiscovery.api.model.ConfigItem; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import java.util.Arrays; public class SpServiceDefinitionBuilder { + private static final Logger LOG = LoggerFactory.getLogger(SpServiceDefinitionBuilder.class); + private SpServiceDefinition serviceDefinition; - private SpConfig config; + //private SpConfig config; public static SpServiceDefinitionBuilder create(String serviceGroup, String serviceName, @@ -49,7 +52,7 @@ public class SpServiceDefinitionBuilder { this.serviceDefinition.setServiceName(serviceName); this.serviceDefinition.setServiceDescription(serviceDescription); this.serviceDefinition.setDefaultPort(defaultPort); - this.config = new ConsulSpConfig(serviceGroup); + //this.config = new ConsulSpConfig(serviceGroup); } public SpServiceDefinitionBuilder withHostname(String hostname) { @@ -58,17 +61,22 @@ public class SpServiceDefinitionBuilder { } public SpServiceDefinitionBuilder addConfig(String key, String defaultValue, String description) { - this.config.register(key, defaultValue, description); + this.serviceDefinition.addConfig(ConfigItem.from(key, defaultValue, description)); return this; } public SpServiceDefinitionBuilder addConfig(String key, Integer defaultValue, String description) { - this.config.register(key, defaultValue, description); + this.serviceDefinition.addConfig(ConfigItem.from(key, defaultValue, description)); return this; } public SpServiceDefinitionBuilder addConfig(String key, Boolean defaultValue, String description) { - this.config.register(key, defaultValue, description); + this.serviceDefinition.addConfig(ConfigItem.from(key, defaultValue, description)); + return this; + } + + public SpServiceDefinitionBuilder addConfig(ConfigItem configItem) { + this.serviceDefinition.addConfig(configItem); return this; } @@ -120,6 +128,12 @@ public class SpServiceDefinitionBuilder { this.serviceDefinition.addDeclarers(other.getDeclarers()); this.serviceDefinition.addAdapterProtocols(other.getAdapterProtocols()); this.serviceDefinition.addSpecificAdapters(other.getSpecificAdapters()); + other.getKvConfigs().values().forEach(value -> { + if (this.serviceDefinition.getKvConfigs().containsKey(value.getKey())) { + LOG.warn("Config key {} already exists and will be overridden by merge, which might lead to strange results.", value.getKey()); + } + this.serviceDefinition.addConfig(value); + }); return this; } diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/ConsulConfig.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/ConsulConfig.java index 8d273e0..d3c7c6d 100644 --- a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/ConsulConfig.java +++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/ConsulConfig.java @@ -22,7 +22,7 @@ import com.google.gson.Gson; import com.google.gson.JsonObject; import org.apache.streampipes.config.backend.BackendConfig; import org.apache.streampipes.config.backend.MessagingSettings; -import org.apache.streampipes.config.consul.ConsulSpConfig; +import org.apache.streampipes.svcdiscovery.consul.ConsulSpConfig; import org.apache.streampipes.svcdiscovery.api.model.ConfigItem; import org.apache.streampipes.svcdiscovery.api.model.PeConfig; import org.apache.streampipes.rest.shared.annotation.GsonWithIds; @@ -48,12 +48,11 @@ public class ConsulConfig extends AbstractRestResource { @GsonWithIds public Response getAllServiceConfigs() { LOG.info("Request for all service configs"); - Map<String, String> peServices = getServiceDiscovery().getPeServices(); + Map<String, String> peServices = getServiceDiscovery().getExtensionsServiceGroups(); List<PeConfig> peConfigs = new LinkedList<>(); for (Map.Entry<String, String> entry : peServices.entrySet()) { - String serviceName = ""; String serviceStatus = entry.getValue(); String mainKey = ConsulSpConfig.SERVICE_ROUTE_PREFIX + entry.getKey(); diff --git a/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/ISpServiceDiscovery.java b/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/ISpServiceDiscovery.java index 40be32b..01a22b2 100644 --- a/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/ISpServiceDiscovery.java +++ b/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/ISpServiceDiscovery.java @@ -66,7 +66,7 @@ public interface ISpServiceDiscovery { * * @return list of pipline element service endpoints */ - Map<String, String> getPeServices(); + Map<String, String> getExtensionsServiceGroups(); /** * Deregister registered service endpoint in Consul diff --git a/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/SpConfig.java b/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/SpConfig.java new file mode 100644 index 0000000..892302f --- /dev/null +++ b/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/SpConfig.java @@ -0,0 +1,64 @@ +/* + * 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.svcdiscovery.api; + +import org.apache.streampipes.svcdiscovery.api.model.ConfigItem; +import org.apache.streampipes.svcdiscovery.api.model.ConfigurationScope; + +public interface SpConfig { + + <T> void register(String key, T defaultValue, String description, ConfigurationScope configurationScope); + + void register(String key, boolean defaultValue, String description); + + void register(String key, int defaultValue, String description); + + void register(String key, double defaultValue, String description); + + void register(String key, String defaultValue, String description); + + void register(ConfigItem configItem); + + void registerObject(String key, Object defaultValue, String description); + + void registerPassword(String key, String defaultValue, String description); + + boolean getBoolean(String key); + + int getInteger(String key); + + double getDouble(String key); + + String getString(String key); + + <T> T getObject(String key, Class<T> clazz, T defaultValue); + + ConfigItem getConfigItem(String key); + + void setBoolean(String key, Boolean value); + + void setInteger(String key, int value); + + void setDouble(String key, double value); + + void setString(String key, String value); + + void setObject(String key, Object value); + +} diff --git a/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/ConfigItem.java b/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/ConfigItem.java index edcdaf6..b44bb90 100644 --- a/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/ConfigItem.java +++ b/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/ConfigItem.java @@ -20,62 +20,100 @@ package org.apache.streampipes.svcdiscovery.api.model; public class ConfigItem { - private String key; - private String description; - private String value; - private String valueType; - private ConfigurationScope configurationScope; - private boolean isPassword; - - public ConfigItem() { - setPassword(false); - } - - public String getKey() { - return key; - } - - public void setKey(String key) { - this.key = key; - } - - public String getDescription() { - return description; - } - - public void setDescription(String description) { - this.description = description; - } - - public String getValue() { - return value; - } - - public void setValue(String value) { - this.value = value; - } - - public String getValueType() { - return valueType; - } - - public void setValueType(String valueType) { - this.valueType = valueType; - } - - public boolean isPassword() { - return isPassword; - } - - public void setPassword(boolean state) { - isPassword = state; - } - - public ConfigurationScope getConfigurationScope() { - return configurationScope; - } - - public void setConfigurationScope(ConfigurationScope configurationScope) { - this.configurationScope = configurationScope; - } + private String key; + private String description; + private String value; + private String valueType; + private ConfigurationScope configurationScope; + private boolean isPassword; + + public static <T> ConfigItem from(String key, + T defaultValue, + String description) { + return from(key, defaultValue, description, ConfigurationScope.CONTAINER_STARTUP_CONFIG, false); + } + + public static <T> ConfigItem from(String key, + T defaultValue, + String description, + ConfigurationScope configurationScope) { + return from(key, defaultValue, description, configurationScope, false); + } + + public static <T> ConfigItem from(String key, + T defaultValue, + String description, + ConfigurationScope configurationScope, + boolean isPassword) { + return from(key, defaultValue, description, ConfigItemUtils.getValueType(defaultValue), configurationScope, isPassword); + } + + public static <T> ConfigItem from(String key, + T defaultValue, + String description, + String valueType, + ConfigurationScope configurationScope, + boolean isPassword) { + ConfigItem configItem = new ConfigItem(); + configItem.setKey(key); + configItem.setValue(String.valueOf(defaultValue)); + configItem.setValueType(valueType); + configItem.setDescription(description); + configItem.setConfigurationScope(configurationScope); + configItem.setPassword(isPassword); + + return configItem; + } + + public ConfigItem() { + setPassword(false); + } + + public String getKey() { + return key; + } + + public void setKey(String key) { + this.key = key; + } + + public String getDescription() { + return description; + } + + public void setDescription(String description) { + this.description = description; + } + + public String getValue() { + return value; + } + + public void setValue(String value) { + this.value = value; + } + + public String getValueType() { + return valueType; + } + + public void setValueType(String valueType) { + this.valueType = valueType; + } + + public boolean isPassword() { + return isPassword; + } + + public void setPassword(boolean state) { + isPassword = state; + } + + public ConfigurationScope getConfigurationScope() { + return configurationScope; + } + + public void setConfigurationScope(ConfigurationScope configurationScope) { + this.configurationScope = configurationScope; + } } diff --git a/streampipes-config/src/main/java/org/apache/streampipes/config/SpConfigChangeCallback.java b/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/ConfigItemUtils.java similarity index 66% rename from streampipes-config/src/main/java/org/apache/streampipes/config/SpConfigChangeCallback.java rename to streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/ConfigItemUtils.java index fc14709..f1251b9 100644 --- a/streampipes-config/src/main/java/org/apache/streampipes/config/SpConfigChangeCallback.java +++ b/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/ConfigItemUtils.java @@ -15,10 +15,19 @@ * limitations under the License. * */ +package org.apache.streampipes.svcdiscovery.api.model; -package org.apache.streampipes.config; +public class ConfigItemUtils { -public interface SpConfigChangeCallback { - - void onChange(); + public static <T> String getValueType(T defaultValue) { + if (defaultValue instanceof Boolean) { + return "xs:boolean"; + } else if (defaultValue instanceof Integer) { + return "xs:integer"; + } else if (defaultValue instanceof Double) { + return "xs:double"; + } else { + return "xs:string"; + } + } } diff --git a/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/SpServiceTagPrefix.java b/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/SpServiceTagPrefix.java index 220f813..c22e04a 100644 --- a/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/SpServiceTagPrefix.java +++ b/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/SpServiceTagPrefix.java @@ -19,6 +19,7 @@ package org.apache.streampipes.svcdiscovery.api.model; public enum SpServiceTagPrefix { SYSTEM("sys"), + SP_GROUP("spgroup"), ADAPTER("adapter"), PROTOCOL("protocol"), DATA_STREAM("dstream"), diff --git a/streampipes-config/src/main/java/org/apache/streampipes/config/consul/ConsulSpConfig.java b/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/ConsulSpConfig.java similarity index 75% rename from streampipes-config/src/main/java/org/apache/streampipes/config/consul/ConsulSpConfig.java rename to streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/ConsulSpConfig.java index 9a3f14a..f195aa2 100644 --- a/streampipes-config/src/main/java/org/apache/streampipes/config/consul/ConsulSpConfig.java +++ b/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/ConsulSpConfig.java @@ -16,25 +16,23 @@ * */ -package org.apache.streampipes.config.consul; +package org.apache.streampipes.svcdiscovery.consul; import com.fasterxml.jackson.core.JsonProcessingException; import com.orbitz.consul.Consul; import com.orbitz.consul.KeyValueClient; -import com.orbitz.consul.model.kv.Value; -import org.apache.streampipes.config.SpConfig; -import org.apache.streampipes.config.SpConfigChangeCallback; import org.apache.streampipes.serializers.json.JacksonSerializer; +import org.apache.streampipes.svcdiscovery.api.SpConfig; import org.apache.streampipes.svcdiscovery.api.model.ConfigItem; +import org.apache.streampipes.svcdiscovery.api.model.ConfigItemUtils; import org.apache.streampipes.svcdiscovery.api.model.ConfigurationScope; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.util.HashMap; import java.util.Map; import java.util.Optional; -public class ConsulSpConfig extends SpConfig implements Runnable { +public class ConsulSpConfig extends AbstractConsulService implements SpConfig { private static final Logger LOG = LoggerFactory.getLogger(ConsulSpConfig.class); @@ -45,60 +43,17 @@ public class ConsulSpConfig extends SpConfig implements Runnable { private final KeyValueClient kvClient; // TODO Implement mechanism to update the client when some configuration parameters change in Consul - private SpConfigChangeCallback callback; private Map<String, Object> configProps; public ConsulSpConfig(String serviceName) { - super(serviceName); Consul consul = consulInstance(); this.kvClient = consul.keyValueClient(); this.serviceName = serviceName; } - public ConsulSpConfig(String serviceName, SpConfigChangeCallback callback) { - this(serviceName); - this.callback = callback; - this.configProps = new HashMap<>(); - new Thread(this).start(); - } - - @Override - public void run() { - Consul consulThread = Consul.builder().build(); // connect to Consul on localhost - KeyValueClient kvClientThread = consulThread.keyValueClient(); - while (true) { - - configProps.keySet().forEach((s) -> { - Optional<Value> te = kvClientThread.getValue(addSn(s)); - if (!te.get().getValueAsString().get().equals(configProps.get(s))) { - callback.onChange(); - configProps.put(s, te.get().getValueAsString().get()); - } - }); - - try { - Thread.sleep(10000); - } catch (InterruptedException e) { - e.printStackTrace(); - } - } - } - @Override public <T> void register(String key, T defaultValue, String description, ConfigurationScope configurationScope) { - register(key, String.valueOf(defaultValue), getValueType(defaultValue), description, configurationScope, false); - } - - private <T> String getValueType(T defaultValue) { - if (defaultValue instanceof Boolean) { - return "xs:boolean"; - } else if (defaultValue instanceof Integer) { - return "xs:integer"; - } else if (defaultValue instanceof Double) { - return "xs:double"; - } else { - return "xs:string"; - } + register(key, String.valueOf(defaultValue), ConfigItemUtils.getValueType(defaultValue), description, configurationScope, false); } @Override @@ -135,22 +90,26 @@ public class ConsulSpConfig extends SpConfig implements Runnable { register(key, defaultValue, "xs:string", description, ConfigurationScope.CONTAINER_STARTUP_CONFIG, true); } - private void register(String key, String defaultValue, String valueType, String description, ConfigurationScope configurationScope, boolean isPassword) { + @Override + public void register(ConfigItem configItem) { + String key = addSn(configItem.getKey()); + Optional<String> i = kvClient.getValueAsString(key); - Optional<String> i = kvClient.getValueAsString(addSn(key)); - ConfigItem configItem = prepareConfigItem(valueType, description, configurationScope, isPassword); - // TODO this check does not work if (!i.isPresent()) { // Set the value of environment variable as default - String envVariable = System.getenv(key); + String envVariable = System.getenv(configItem.getKey()); if (envVariable != null) { configItem.setValue(envVariable); - kvClient.putValue(addSn(key), toJson(configItem)); + kvClient.putValue(key, toJson(configItem)); } else { - configItem.setValue(defaultValue); - kvClient.putValue(addSn(key), toJson(configItem)); + kvClient.putValue(key, toJson(configItem)); } } + } + + private void register(String key, String defaultValue, String valueType, String description, ConfigurationScope configurationScope, boolean isPassword) { + ConfigItem configItem = ConfigItem.from(key, defaultValue, description, valueType, configurationScope, isPassword); + register(configItem); if (configProps != null) { configProps.put(key, getString(key)); diff --git a/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/SpConsulServiceDiscovery.java b/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/SpConsulServiceDiscovery.java index c23f760..3cc6984 100644 --- a/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/SpConsulServiceDiscovery.java +++ b/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/SpConsulServiceDiscovery.java @@ -25,10 +25,7 @@ import com.orbitz.consul.model.agent.Registration; import com.orbitz.consul.model.health.HealthCheck; import com.orbitz.consul.model.health.Service; import org.apache.streampipes.svcdiscovery.api.ISpServiceDiscovery; -import org.apache.streampipes.svcdiscovery.api.model.DefaultSpServiceGroups; -import org.apache.streampipes.svcdiscovery.api.model.DefaultSpServiceTags; -import org.apache.streampipes.svcdiscovery.api.model.SpServiceRegistrationRequest; -import org.apache.streampipes.svcdiscovery.api.model.SpServiceTag; +import org.apache.streampipes.svcdiscovery.api.model.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -46,7 +43,6 @@ public class SpConsulServiceDiscovery extends AbstractConsulService implements I private static final String HTTP_PROTOCOL = "http://";; private static final String COLON = ":"; private static final String HEALTH_CHECK_INTERVAL = "10s"; - private static final String PE_SVC_TAG = "pe"; @Override public void registerService(SpServiceRegistrationRequest req) { @@ -78,7 +74,7 @@ public class SpConsulServiceDiscovery extends AbstractConsulService implements I } @Override - public Map<String, String> getPeServices() { + public Map<String, String> getExtensionsServiceGroups() { LOG.info("Load pipeline element service status"); Consul consul = consulInstance(); AgentClient agent = consul.agentClient(); @@ -89,19 +85,29 @@ public class SpConsulServiceDiscovery extends AbstractConsulService implements I Map<String, String> peSvcs = new HashMap<>(); for (Map.Entry<String, Service> entry : services.entrySet()) { - if (entry.getValue().getTags().contains(PE_SVC_TAG)) { + if (hasExtensionsTag(entry.getValue().getTags())) { String serviceId = entry.getValue().getId(); String serviceStatus = "critical"; if (checks.containsKey("service:" + entry.getKey())) { serviceStatus = checks.get("service:" + entry.getKey()).getStatus(); } LOG.info("Service id: " + serviceId + " service status: " + serviceStatus); - peSvcs.put(serviceId, serviceStatus); + String serviceGroup = extractServiceGroup(entry.getValue().getTags()); + peSvcs.put(serviceGroup, serviceStatus); } } return peSvcs; } + private boolean hasExtensionsTag(List<String> tags) { + return tags.stream().anyMatch(tag -> tag.equals(DefaultSpServiceTags.PE.asString()) || tag.equals(DefaultSpServiceTags.CONNECT_WORKER.asString())); + } + + private String extractServiceGroup(List<String> tags) { + String groupTag = tags.stream().filter(tag -> tag.startsWith(SpServiceTagPrefix.SP_GROUP.asString())).findFirst().orElse("unknown service group"); + return groupTag.replaceAll(SpServiceTagPrefix.SP_GROUP.asString() + ":", ""); + } + @Override public void deregisterService(String svcId) { Consul consul = consulInstance(); diff --git a/streampipes-service-discovery/src/main/java/org/apache/streampipes/svcdiscovery/SpServiceDiscovery.java b/streampipes-service-discovery/src/main/java/org/apache/streampipes/svcdiscovery/SpServiceDiscovery.java index e4a87ad..2a292c8 100644 --- a/streampipes-service-discovery/src/main/java/org/apache/streampipes/svcdiscovery/SpServiceDiscovery.java +++ b/streampipes-service-discovery/src/main/java/org/apache/streampipes/svcdiscovery/SpServiceDiscovery.java @@ -19,6 +19,8 @@ package org.apache.streampipes.svcdiscovery; import org.apache.streampipes.svcdiscovery.api.ISpKvManagement; import org.apache.streampipes.svcdiscovery.api.ISpServiceDiscovery; +import org.apache.streampipes.svcdiscovery.api.SpConfig; +import org.apache.streampipes.svcdiscovery.consul.ConsulSpConfig; import org.apache.streampipes.svcdiscovery.consul.SpConsulKvManagement; import org.apache.streampipes.svcdiscovery.consul.SpConsulServiceDiscovery; @@ -31,4 +33,9 @@ public class SpServiceDiscovery { public static ISpKvManagement getKeyValueStore() { return new SpConsulKvManagement(); } + + public static SpConfig getSpConfig(String serviceGroup) { + return new ConsulSpConfig(serviceGroup); + } + } diff --git a/streampipes-service-extensions-base/src/main/java/org/apache/streampipes/service/extensions/base/StreamPipesExtensionsServiceBase.java b/streampipes-service-extensions-base/src/main/java/org/apache/streampipes/service/extensions/base/StreamPipesExtensionsServiceBase.java index d65d6e4..41adedc 100644 --- a/streampipes-service-extensions-base/src/main/java/org/apache/streampipes/service/extensions/base/StreamPipesExtensionsServiceBase.java +++ b/streampipes-service-extensions-base/src/main/java/org/apache/streampipes/service/extensions/base/StreamPipesExtensionsServiceBase.java @@ -22,11 +22,15 @@ import org.apache.streampipes.container.base.StreamPipesServiceBase; import org.apache.streampipes.container.init.DeclarersSingleton; import org.apache.streampipes.container.model.SpServiceDefinition; import org.apache.streampipes.svcdiscovery.api.model.DefaultSpServiceGroups; +import org.apache.streampipes.svcdiscovery.api.model.SpServiceTag; +import org.apache.streampipes.svcdiscovery.api.model.SpServiceTagPrefix; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import javax.annotation.PreDestroy; import java.net.UnknownHostException; +import java.util.ArrayList; +import java.util.List; public abstract class StreamPipesExtensionsServiceBase extends StreamPipesServiceBase { @@ -68,6 +72,16 @@ public abstract class StreamPipesExtensionsServiceBase extends StreamPipesServic this.afterServiceRegistered(serviceDef); } + @Override + protected List<SpServiceTag> getServiceTags() { + List<SpServiceTag> tags = new ArrayList<>(); + tags.add(SpServiceTag.create(SpServiceTagPrefix.SP_GROUP, DeclarersSingleton.getInstance().getServiceDefinition().getServiceGroup())); + tags.addAll(getExtensionsServiceTags()); + return tags; + } + + protected abstract List<SpServiceTag> getExtensionsServiceTags(); + @PreDestroy public abstract void onExit(); diff --git a/streampipes-storage-couchdb/pom.xml b/streampipes-storage-couchdb/pom.xml index 3ed20e6..7617642 100644 --- a/streampipes-storage-couchdb/pom.xml +++ b/streampipes-storage-couchdb/pom.xml @@ -31,7 +31,7 @@ <!-- StreamPipes dependencies --> <dependency> <groupId>org.apache.streampipes</groupId> - <artifactId>streampipes-config</artifactId> + <artifactId>streampipes-service-discovery</artifactId> <version>0.69.0-SNAPSHOT</version> </dependency> <dependency> @@ -73,4 +73,4 @@ <artifactId>fluent-hc</artifactId> </dependency> </dependencies> -</project> \ No newline at end of file +</project> diff --git a/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/utils/CouchDbConfig.java b/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/utils/CouchDbConfig.java index f08c795..5194615 100644 --- a/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/utils/CouchDbConfig.java +++ b/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/utils/CouchDbConfig.java @@ -18,7 +18,8 @@ package org.apache.streampipes.storage.couchdb.utils; -import org.apache.streampipes.config.SpConfig; +import org.apache.streampipes.svcdiscovery.SpServiceDiscovery; +import org.apache.streampipes.svcdiscovery.api.SpConfig; public enum CouchDbConfig { @@ -30,7 +31,7 @@ public enum CouchDbConfig { private final static String PROTOCOL = "PROTOCOL"; CouchDbConfig() { - config = SpConfig.getSpConfig("storage/couchdb"); + config = SpServiceDiscovery.getSpConfig("storage/couchdb"); config.register(COUCHDB_HOST, "couchdb", "Hostname for the couch db service"); config.register(COUCHDB_PORT, 5984, "Port for the couch db service"); config.register(PROTOCOL, "http", "Protocol the couch db service"); diff --git a/streampipes-user-management/pom.xml b/streampipes-user-management/pom.xml index 9f7be0b..f3a56c8 100644 --- a/streampipes-user-management/pom.xml +++ b/streampipes-user-management/pom.xml @@ -31,6 +31,11 @@ <!-- StreamPipes dependencies --> <dependency> <groupId>org.apache.streampipes</groupId> + <artifactId>streampipes-config</artifactId> + <version>0.69.0-SNAPSHOT</version> + </dependency> + <dependency> + <groupId>org.apache.streampipes</groupId> <artifactId>streampipes-model-client</artifactId> <version>0.69.0-SNAPSHOT</version> </dependency> diff --git a/ui/src/app/configuration/consul-configs/consul-configs.component.html b/ui/src/app/configuration/consul-configs/consul-configs.component.html index c62e6fe..f5bec99 100644 --- a/ui/src/app/configuration/consul-configs/consul-configs.component.html +++ b/ui/src/app/configuration/consul-configs/consul-configs.component.html @@ -17,7 +17,7 @@ --> <div fxLayout="row" fxFlex="100" - style="background:white;border-top:1px solid lightgray;padding:10px;margin-bottom:5px;"> + style="border-top:1px solid lightgray;padding:10px;margin-bottom:5px;"> <form (ngSubmit)="updateConfiguration()" fxFlex="100"> <mat-grid-list cols="4" rowHeight="3:1" gutterSize="10px"> <mat-grid-tile *ngFor="let configuration of consulService.configs"> diff --git a/ui/src/app/configuration/pipeline-element-configuration/pipeline-element-configuration.component.html b/ui/src/app/configuration/pipeline-element-configuration/pipeline-element-configuration.component.html index 2765efa..68df495 100644 --- a/ui/src/app/configuration/pipeline-element-configuration/pipeline-element-configuration.component.html +++ b/ui/src/app/configuration/pipeline-element-configuration/pipeline-element-configuration.component.html @@ -39,8 +39,7 @@ </ng-container> <ng-container matColumnDef="name"> - <th fxFlex="40" fxLayoutAlign="start center" mat-header-cell *matHeaderCellDef> Pipeline Element - Container + <th fxFlex="40" fxLayoutAlign="start center" mat-header-cell *matHeaderCellDef> Extensions Service Group </th> <td fxFlex="40" fxLayoutAlign="start center" mat-cell *matCellDef="let element"> {{element.name}} @@ -77,4 +76,4 @@ </table> <!-- <mat-paginator [pageSize]="20" [pageSizeOptions]="[20, 50, 100]" showFirstLastButtons></mat-paginator>--> </div> -</div> \ No newline at end of file +</div>
