This is an automated email from the ASF dual-hosted git repository. riemer pushed a commit to branch remove-consul in repository https://gitbox.apache.org/repos/asf/streampipes.git
commit 451cda9a1427b28d4a94fbcecbe019756d82b908 Author: Dominik Riemer <[email protected]> AuthorDate: Sun Aug 6 18:19:05 2023 +0200 Cleanup, delete obsolete code --- .../apache/streampipes/commons/constants/Envs.java | 4 + .../commons/environment/DefaultEnvironment.java | 10 ++ .../commons/environment/Environment.java | 2 + .../config/backend/BackendConfigKeys.java | 5 - ...StreamPipesClientRuntimeConnectionResolver.java | 5 +- .../svcdiscovery/ConsulKvManagementTest.java | 63 --------- .../svcdiscovery/ConsulSpServiceDiscoveryTest.java | 105 -------------- .../ExtensionsServiceEndpointGenerator.java | 4 +- .../ExtensionsServiceEndpointProvider.java | 18 --- .../pipeline/ExtensionsServiceLogExecutor.java | 4 +- .../rest/core/base/impl/AbstractRestResource.java | 24 ---- .../service/base/StreamPipesServiceBase.java | 10 -- .../service/core/StreamPipesCoreApplication.java | 20 --- .../svcdiscovery/api/ISpKvManagement.java | 30 ---- .../svcdiscovery/api/ISpServiceDiscovery.java | 32 ----- .../api/model/DefaultSpServiceTags.java | 6 +- ...rviceGroups.java => DefaultSpServiceTypes.java} | 3 +- .../consul/ConsulHealthServiceManager.java | 96 ------------- .../svcdiscovery/consul/SpConsulKvManagement.java | 74 ---------- .../consul/SpConsulServiceDiscovery.java | 156 --------------------- .../svcdiscovery/SpKvManagementCore.java | 41 ------ .../svcdiscovery/SpServiceDiscovery.java | 9 -- .../svcdiscovery/SpServiceDiscoveryCore.java | 27 +--- .../StreamPipesExtensionsServiceBase.java | 7 +- 24 files changed, 29 insertions(+), 726 deletions(-) diff --git a/streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/Envs.java b/streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/Envs.java index ab1a87890..bf766979e 100644 --- a/streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/Envs.java +++ b/streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/Envs.java @@ -32,6 +32,10 @@ public enum Envs { SP_CONSUL_HOST("SP_CONSUL_HOST", DefaultEnvValues.CONSUL_HOST_DEFAULT, DefaultEnvValues.LOCALHOST), + + SP_CORE_HOST("SP_CORE_HOST", "backend", "localhost"), + SP_CORE_PORT("SP_CORE_PORT", "8030", "8030"), + SP_CONSUL_PORT("SP_CONSUL_PORT", DefaultEnvValues.CONSUL_PORT_DEFAULT), SP_KAFKA_RETENTION_MS("SP_KAFKA_RETENTION_MS", DefaultEnvValues.SP_KAFKA_RETENTION_MS_DEFAULT), SP_PRIORITIZED_PROTOCOL("SP_PRIORITIZED_PROTOCOL", "kafka"), diff --git a/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/DefaultEnvironment.java b/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/DefaultEnvironment.java index 9b5e695e3..58e2b9157 100644 --- a/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/DefaultEnvironment.java +++ b/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/DefaultEnvironment.java @@ -50,6 +50,16 @@ public class DefaultEnvironment implements Environment { return new IntEnvironmentVariable(Envs.SP_PORT); } + @Override + public StringEnvironmentVariable getSpCoreHost() { + return new StringEnvironmentVariable(Envs.SP_CORE_HOST); + } + + @Override + public IntEnvironmentVariable getSpCorePort() { + return new IntEnvironmentVariable(Envs.SP_CORE_PORT); + } + @Override public StringEnvironmentVariable getTsStorageProtocol() { return new StringEnvironmentVariable(Envs.SP_TS_STORAGE_PROTOCOL); diff --git a/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/Environment.java b/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/Environment.java index 8abc73154..481120ebb 100644 --- a/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/Environment.java +++ b/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/Environment.java @@ -40,6 +40,8 @@ public interface Environment { IntEnvironmentVariable getServicePort(); + StringEnvironmentVariable getSpCoreHost(); + IntEnvironmentVariable getSpCorePort(); // Time series storage env variables diff --git a/streampipes-config/src/main/java/org/apache/streampipes/config/backend/BackendConfigKeys.java b/streampipes-config/src/main/java/org/apache/streampipes/config/backend/BackendConfigKeys.java index 0fe2db152..b0268f46c 100644 --- a/streampipes-config/src/main/java/org/apache/streampipes/config/backend/BackendConfigKeys.java +++ b/streampipes-config/src/main/java/org/apache/streampipes/config/backend/BackendConfigKeys.java @@ -41,9 +41,4 @@ public class BackendConfigKeys { public static final String LOCAL_AUTH_CONFIG = "SP_LOCAL_AUTH_CONFIG"; public static final String EMAIL_CONFIG = "SP_EMAIL_CONFIG"; public static final String GENERAL_CONFIG = "SP_GENERAL_CONFIG"; - - - public static final String SERVICE_NAME = "SP_SERVICE_NAME"; - - public static final String PRIORITIZED_PROTOCOL = "SP_PRIORITIZED_PROTOCOL"; } diff --git a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/client/StreamPipesClientRuntimeConnectionResolver.java b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/client/StreamPipesClientRuntimeConnectionResolver.java index 8d53455ba..550643bdb 100644 --- a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/client/StreamPipesClientRuntimeConnectionResolver.java +++ b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/client/StreamPipesClientRuntimeConnectionResolver.java @@ -51,7 +51,7 @@ public class StreamPipesClientRuntimeConnectionResolver implements ClientConnect if (baseUrls.size() > 0) { if (env.getSpDebug().getValueOrDefault()) { try { - return "http://"; + Networking.getHostname() + ":" + 8030; + return "http://"; + Networking.getHostname() + ":" + env.getSpCorePort().getValueOrDefault(); } catch (UnknownHostException e) { LOG.error("Could not auto-resolve host address - using http://localhost:8030";); return "http://localhost:8030";; @@ -72,7 +72,6 @@ public class StreamPipesClientRuntimeConnectionResolver implements ClientConnect } private List<String> findClientServices() { - // TODO check how to identify backend - return List.of("backend"); + return List.of(env.getSpCoreHost().getValueOrDefault()); } } diff --git a/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/svcdiscovery/ConsulKvManagementTest.java b/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/svcdiscovery/ConsulKvManagementTest.java deleted file mode 100644 index 62ebdaf40..000000000 --- a/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/svcdiscovery/ConsulKvManagementTest.java +++ /dev/null @@ -1,63 +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.integration.svcdiscovery; - -import org.apache.streampipes.model.extensions.configuration.ConfigItem; -import org.apache.streampipes.svcdiscovery.SpServiceDiscovery; -import org.apache.streampipes.svcdiscovery.api.ISpKvManagement; - -import com.google.gson.Gson; -import org.junit.Before; -import org.junit.Test; - -import java.util.Map; - -import static org.junit.Assert.assertEquals; - -public class ConsulKvManagementTest extends AbstractConsulTest { - - private ISpKvManagement kvManagement; - - @Before - public void before() { - var environment = mockEnvironment(); - this.kvManagement = SpServiceDiscovery - .getKeyValueStore(environment); - } - - @Test - public void testKv() { - var test1Key = "a/test"; - var test2Key = "a/test2"; - var configItem1 = makeConfigItem(test1Key, "abc"); - var configItem2 = makeConfigItem(test2Key, "def"); - kvManagement.updateConfig(test1Key, serialize(configItem1), false); - kvManagement.updateConfig(test2Key, serialize(configItem2), false); - - Map<String, String> result = kvManagement.getKeyValue("a"); - - assertEquals(2, result.size()); - assertEquals(serialize(configItem1), result.get(test1Key)); - assertEquals(serialize(configItem2), result.get(test2Key)); - } - - private String serialize(ConfigItem item) { - return new Gson().toJson(item); - } -} diff --git a/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/svcdiscovery/ConsulSpServiceDiscoveryTest.java b/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/svcdiscovery/ConsulSpServiceDiscoveryTest.java deleted file mode 100644 index d776e318d..000000000 --- a/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/svcdiscovery/ConsulSpServiceDiscoveryTest.java +++ /dev/null @@ -1,105 +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.integration.svcdiscovery; - -import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceRegistration; -import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceTag; -import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceTagPrefix; -import org.apache.streampipes.svcdiscovery.SpServiceDiscovery; -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.junit.Before; -import org.junit.Test; - -import java.util.List; - -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertTrue; - -public class ConsulSpServiceDiscoveryTest extends AbstractConsulTest { - - private ISpServiceDiscovery serviceDiscovery; - - @Before - public void before() { - var env = mockEnvironment(); - this.serviceDiscovery = SpServiceDiscovery - .getServiceDiscovery(env); - } - - @Test - public void testServiceRegistration() { - - serviceDiscovery.registerService(makeRequest(makeServiceTag(), "abc")); - - var endpoints = getEndpoints(false); - assertEquals(1, endpoints.size()); - - serviceDiscovery.deregisterService("abc"); - - endpoints = getEndpoints(false); - assertEquals(0, endpoints.size()); - } - - @Test - public void testGetExtensionsServiceGroups() { - serviceDiscovery.registerService(makeRequest(List.of(DefaultSpServiceTags.PE, groupTag()), "abc")); - serviceDiscovery.registerService(makeRequest( - List.of(DefaultSpServiceTags.CORE, groupTag()), "def")); - - var serviceGroups = serviceDiscovery.getExtensionsServiceGroups(); - - assertEquals(1, serviceGroups.size()); - assertTrue(serviceGroups.containsKey("my-group")); - - serviceDiscovery.deregisterService("abc"); - serviceDiscovery.deregisterService("def"); - } - - private List<String> getEndpoints(boolean healthy) { - return serviceDiscovery.getServiceEndpoints( - DefaultSpServiceGroups.EXT, - healthy, - List.of(makeServiceTag().get(0).asString())); - } - - private SpServiceRegistration makeRequest(List<SpServiceTag> serviceTags, - String serviceId) { - var req = new SpServiceRegistration( - DefaultSpServiceGroups.EXT, - serviceId, - "localhost", - 80, - serviceTags, - "/" - ); - - return req; - } - - private SpServiceTag groupTag() { - return SpServiceTag.create(SpServiceTagPrefix.SP_GROUP, "my-group"); - } - - private List<SpServiceTag> makeServiceTag() { - return List.of(DefaultSpServiceTags.PE); - } -} diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/endpoint/ExtensionsServiceEndpointGenerator.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/endpoint/ExtensionsServiceEndpointGenerator.java index 8b47fae45..970d81a0d 100644 --- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/endpoint/ExtensionsServiceEndpointGenerator.java +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/endpoint/ExtensionsServiceEndpointGenerator.java @@ -20,7 +20,7 @@ package org.apache.streampipes.manager.execution.endpoint; import org.apache.streampipes.commons.exceptions.NoServiceEndpointsAvailableException; import org.apache.streampipes.model.base.NamedStreamPipesEntity; import org.apache.streampipes.svcdiscovery.SpServiceDiscovery; -import org.apache.streampipes.svcdiscovery.api.model.DefaultSpServiceGroups; +import org.apache.streampipes.svcdiscovery.api.model.DefaultSpServiceTypes; import org.apache.streampipes.svcdiscovery.api.model.SpServiceUrlProvider; import org.slf4j.Logger; @@ -59,7 +59,7 @@ public class ExtensionsServiceEndpointGenerator { return SpServiceDiscovery .getServiceDiscovery() .getServiceEndpoints( - DefaultSpServiceGroups.EXT, + DefaultSpServiceTypes.EXT, true, Collections .singletonList( diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/endpoint/ExtensionsServiceEndpointProvider.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/endpoint/ExtensionsServiceEndpointProvider.java index 92578a178..6e76fedfe 100644 --- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/endpoint/ExtensionsServiceEndpointProvider.java +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/endpoint/ExtensionsServiceEndpointProvider.java @@ -18,15 +18,8 @@ package org.apache.streampipes.manager.execution.endpoint; -import org.apache.streampipes.commons.constants.GlobalStreamPipesConstants; import org.apache.streampipes.commons.exceptions.NoServiceEndpointsAvailableException; import org.apache.streampipes.model.base.InvocableStreamPipesEntity; -import org.apache.streampipes.svcdiscovery.SpServiceDiscovery; -import org.apache.streampipes.svcdiscovery.api.model.DefaultSpServiceGroups; -import org.apache.streampipes.svcdiscovery.api.model.DefaultSpServiceTags; - -import java.util.Collections; -import java.util.List; public class ExtensionsServiceEndpointProvider { @@ -36,15 +29,4 @@ public class ExtensionsServiceEndpointProvider { ExtensionsServiceEndpointUtils.getPipelineElementType(g)) .getEndpointResourceUrl(); } - - private String getConnectMasterSourcesUrl() throws NoServiceEndpointsAvailableException { - List<String> connectMasterEndpoints = SpServiceDiscovery.getServiceDiscovery() - .getServiceEndpoints(DefaultSpServiceGroups.CORE, true, - Collections.singletonList(DefaultSpServiceTags.CONNECT_MASTER.asString())); - if (connectMasterEndpoints.size() > 0) { - return connectMasterEndpoints.get(0) + GlobalStreamPipesConstants.CONNECT_MASTER_SOURCES_ENDPOINT; - } else { - throw new NoServiceEndpointsAvailableException("Could not find any available connect master service endpoint"); - } - } } diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/monitoring/pipeline/ExtensionsServiceLogExecutor.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/monitoring/pipeline/ExtensionsServiceLogExecutor.java index 83eba204b..ee8e19885 100644 --- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/monitoring/pipeline/ExtensionsServiceLogExecutor.java +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/monitoring/pipeline/ExtensionsServiceLogExecutor.java @@ -25,8 +25,8 @@ import org.apache.streampipes.model.monitoring.SpEndpointMonitoringInfo; import org.apache.streampipes.resource.management.SpResourceManager; import org.apache.streampipes.serializers.json.JacksonSerializer; import org.apache.streampipes.svcdiscovery.SpServiceDiscovery; -import org.apache.streampipes.svcdiscovery.api.model.DefaultSpServiceGroups; import org.apache.streampipes.svcdiscovery.api.model.DefaultSpServiceTags; +import org.apache.streampipes.svcdiscovery.api.model.DefaultSpServiceTypes; import com.fasterxml.jackson.core.JsonProcessingException; import org.apache.http.client.fluent.Request; @@ -71,7 +71,7 @@ public class ExtensionsServiceLogExecutor implements Runnable { private List<String> getActiveExtensionsEndpoints() { return SpServiceDiscovery.getServiceDiscovery().getServiceEndpoints( - DefaultSpServiceGroups.EXT, + DefaultSpServiceTypes.EXT, true, List.of(DefaultSpServiceTags.PE.asString(), DefaultSpServiceTags.CONNECT_WORKER.asString()) ); diff --git a/streampipes-rest-core-base/src/main/java/org/apache/streampipes/rest/core/base/impl/AbstractRestResource.java b/streampipes-rest-core-base/src/main/java/org/apache/streampipes/rest/core/base/impl/AbstractRestResource.java index d3edd0090..b231450db 100644 --- a/streampipes-rest-core-base/src/main/java/org/apache/streampipes/rest/core/base/impl/AbstractRestResource.java +++ b/streampipes-rest-core-base/src/main/java/org/apache/streampipes/rest/core/base/impl/AbstractRestResource.java @@ -37,9 +37,6 @@ import org.apache.streampipes.storage.api.IUserStorage; import org.apache.streampipes.storage.api.IVisualizationStorage; import org.apache.streampipes.storage.management.StorageDispatcher; import org.apache.streampipes.storage.management.StorageManager; -import org.apache.streampipes.svcdiscovery.SpServiceDiscovery; -import org.apache.streampipes.svcdiscovery.api.ISpKvManagement; -import org.apache.streampipes.svcdiscovery.api.ISpServiceDiscovery; import org.apache.http.client.ClientProtocolException; @@ -115,19 +112,6 @@ public abstract class AbstractRestResource extends AbstractSharedRestInterface { return statusMessage(new ErrorMessage(notifications)); } -// protected String getCurrentUsername() throws AuthenticationException { -// if (SecurityContext.getSubject().isAuthenticated()) { -// return SecurityUtils.getSubject().getPrincipal().toString(); -// } -// throw new AuthenticationException("Not authenticated"); -// } - - protected boolean authorized(String username) { - // TODO SEC - //return username.equals(SecurityUtils.getSubject().getPrincipal().toString()); - return true; - } - @SuppressWarnings("deprecation") protected String decode(String encodedString) { return URLDecoder.decode(encodedString); @@ -150,14 +134,6 @@ public abstract class AbstractRestResource extends AbstractSharedRestInterface { return Response.status(Response.Status.UNAUTHORIZED).build(); } - protected ISpKvManagement getKeyValueStore() { - return SpServiceDiscovery.getKeyValueStore(); - } - - protected ISpServiceDiscovery getServiceDiscovery() { - return SpServiceDiscovery.getServiceDiscovery(); - } - protected SpResourceManager getSpResourceManager() { return new SpResourceManager(); } diff --git a/streampipes-service-base/src/main/java/org/apache/streampipes/service/base/StreamPipesServiceBase.java b/streampipes-service-base/src/main/java/org/apache/streampipes/service/base/StreamPipesServiceBase.java index 98a351c78..0d516c3a8 100644 --- a/streampipes-service-base/src/main/java/org/apache/streampipes/service/base/StreamPipesServiceBase.java +++ b/streampipes-service-base/src/main/java/org/apache/streampipes/service/base/StreamPipesServiceBase.java @@ -17,25 +17,17 @@ */ package org.apache.streampipes.service.base; -import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceTag; - import org.apache.commons.lang3.RandomStringUtils; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import org.springframework.boot.SpringApplication; import java.net.UnknownHostException; import java.util.Collections; -import java.util.List; public abstract class StreamPipesServiceBase { public static final String AUTO_GENERATED_SERVICE_ID = RandomStringUtils.randomAlphanumeric(6); - private static final Logger LOG = LoggerFactory.getLogger(StreamPipesServiceBase.class); public void startStreamPipesService(Class<?> serviceClass, - String serviceGroup, - String serviceId, BaseNetworkingConfig networkingConfig) throws UnknownHostException { runApplication(serviceClass, networkingConfig.getPort()); } @@ -47,8 +39,6 @@ public abstract class StreamPipesServiceBase { app.run(); } - protected abstract List<SpServiceTag> getServiceTags(); - protected String getHealthCheckPath() { return "/svchealth/" + AUTO_GENERATED_SERVICE_ID; } diff --git a/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/StreamPipesCoreApplication.java b/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/StreamPipesCoreApplication.java index 23f766369..32d6c0416 100644 --- a/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/StreamPipesCoreApplication.java +++ b/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/StreamPipesCoreApplication.java @@ -30,7 +30,6 @@ import org.apache.streampipes.messaging.kafka.SpKafkaProtocolFactory; import org.apache.streampipes.messaging.mqtt.SpMqttProtocolFactory; import org.apache.streampipes.messaging.nats.SpNatsProtocolFactory; import org.apache.streampipes.messaging.pulsar.SpPulsarProtocolFactory; -import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceTag; import org.apache.streampipes.model.pipeline.Pipeline; import org.apache.streampipes.model.pipeline.PipelineOperationStatus; import org.apache.streampipes.rest.security.SpPermissionEvaluator; @@ -40,8 +39,6 @@ import org.apache.streampipes.service.core.migrations.MigrationsHandler; import org.apache.streampipes.storage.api.IPipelineStorage; import org.apache.streampipes.storage.couchdb.utils.CouchDbViewGenerator; import org.apache.streampipes.storage.management.StorageDispatcher; -import org.apache.streampipes.svcdiscovery.api.model.DefaultSpServiceGroups; -import org.apache.streampipes.svcdiscovery.api.model.DefaultSpServiceTags; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -54,7 +51,6 @@ import jakarta.annotation.PostConstruct; import jakarta.annotation.PreDestroy; import java.net.UnknownHostException; -import java.util.Arrays; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -110,8 +106,6 @@ public class StreamPipesCoreApplication extends StreamPipesServiceBase { registerProtocols(supportedProtocols); BaseNetworkingConfig networkingConfig = BaseNetworkingConfig.defaultResolution(8030); startStreamPipesService(StreamPipesCoreApplication.class, - DefaultSpServiceGroups.CORE, - serviceId(), networkingConfig); } catch (UnknownHostException e) { LOG.error( @@ -124,10 +118,6 @@ public class StreamPipesCoreApplication extends StreamPipesServiceBase { protocols.getSupportedProtocols().forEach(SpProtocolManager.INSTANCE::register); } - private String serviceId() { - return DefaultSpServiceGroups.CORE + "-" + AUTO_GENERATED_SERVICE_ID; - } - @PostConstruct public void init() { this.executorService = Executors.newSingleThreadScheduledExecutor(); @@ -298,16 +288,6 @@ public class StreamPipesCoreApplication extends StreamPipesServiceBase { .getPipelineStorageAPI(); } - - @Override - protected List<SpServiceTag> getServiceTags() { - return Arrays.asList( - DefaultSpServiceTags.CORE, - DefaultSpServiceTags.CONNECT_MASTER, - DefaultSpServiceTags.STREAMPIPES_CLIENT - ); - } - @Override protected String getHealthCheckPath() { return "/streampipes-backend/api/svchealth/" + AUTO_GENERATED_SERVICE_ID; diff --git a/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/ISpKvManagement.java b/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/ISpKvManagement.java deleted file mode 100644 index 4b859f3c6..000000000 --- a/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/ISpKvManagement.java +++ /dev/null @@ -1,30 +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.svcdiscovery.api; - -import java.util.Map; - -public interface ISpKvManagement { - - Map<String, String> getKeyValue(String route); - - void updateConfig(String key, String entry, boolean password); - - void deleteConfig(String key); - -} 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 fb2eea73b..4f4350f7d 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 @@ -17,20 +17,10 @@ */ package org.apache.streampipes.svcdiscovery.api; -import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceRegistration; - import java.util.List; -import java.util.Map; public interface ISpServiceDiscovery { - /** - * Register service. - * - * @param serviceRegistrationRequest the service registration request - */ - void registerService(SpServiceRegistration serviceRegistrationRequest); - /** * Get active pipeline element service endpoints * @@ -38,13 +28,6 @@ public interface ISpServiceDiscovery { */ List<String> getActivePipelineElementEndpoints(); - /** - * Get active StreamPipes Connect worker endpoints - * - * @return list of StreamPipes Connect worker endpoints - */ - List<String> getActiveConnectWorkerEndpoints(); - /** * Get service endpoints * @@ -56,19 +39,4 @@ public interface ISpServiceDiscovery { List<String> getServiceEndpoints(String svcGroup, boolean restrictToHealthy, List<String> filterByTags); - - /** - * Get all pipeline element service endpoints - * - * @return list of pipline element service endpoints - */ - Map<String, String> getExtensionsServiceGroups(); - - /** - * Deregister registered service endpoint in Consul - * - * @param svcId service id of endpoint to be deregistered - */ - void deregisterService(String svcId); - } diff --git a/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/DefaultSpServiceTags.java b/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/DefaultSpServiceTags.java index 3d2a4bc52..b2b509180 100644 --- a/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/DefaultSpServiceTags.java +++ b/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/DefaultSpServiceTags.java @@ -24,8 +24,6 @@ public class DefaultSpServiceTags { public static final SpServiceTag CORE = SpServiceTag.create(SpServiceTagPrefix.SYSTEM, "core"); public static final SpServiceTag PE = SpServiceTag.create(SpServiceTagPrefix.SYSTEM, "pe"); - public static final SpServiceTag CONNECT_MASTER = SpServiceTag.create(SpServiceTagPrefix.SYSTEM, "connect-master"); - public static final SpServiceTag CONNECT_WORKER = SpServiceTag.create(SpServiceTagPrefix.SYSTEM, "connect-worker"); - public static final SpServiceTag STREAMPIPES_CLIENT = - SpServiceTag.create(SpServiceTagPrefix.SYSTEM, "streampipes-client"); + public static final SpServiceTag CONNECT_WORKER = SpServiceTag + .create(SpServiceTagPrefix.SYSTEM, "connect-worker"); } diff --git a/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/DefaultSpServiceGroups.java b/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/DefaultSpServiceTypes.java similarity index 91% rename from streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/DefaultSpServiceGroups.java rename to streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/DefaultSpServiceTypes.java index 5fd3452f3..6e343a809 100644 --- a/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/DefaultSpServiceGroups.java +++ b/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/DefaultSpServiceTypes.java @@ -17,9 +17,8 @@ */ package org.apache.streampipes.svcdiscovery.api.model; -public class DefaultSpServiceGroups { +public class DefaultSpServiceTypes { - public static final String CORE = "core"; public static final String EXT = "ext"; } diff --git a/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/ConsulHealthServiceManager.java b/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/ConsulHealthServiceManager.java deleted file mode 100644 index cb1710483..000000000 --- a/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/ConsulHealthServiceManager.java +++ /dev/null @@ -1,96 +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.svcdiscovery.consul; - -import org.apache.streampipes.commons.environment.Environment; - -import com.ecwid.consul.v1.QueryParams; -import com.ecwid.consul.v1.Response; -import com.ecwid.consul.v1.health.HealthServicesRequest; -import com.ecwid.consul.v1.health.model.Check; -import com.ecwid.consul.v1.health.model.HealthService; - -import java.util.Collections; -import java.util.HashSet; -import java.util.List; -import java.util.concurrent.TimeUnit; -import java.util.stream.Collectors; - -public class ConsulHealthServiceManager extends AbstractConsulService { - - private static final int MAX_RETRIES = 3; - - public ConsulHealthServiceManager(Environment environment) { - super(environment); - initializeCache(); - } - - public void initializeCache() { - - } - - public List<String> getServiceEndpoints(String serviceGroup, - boolean restrictToHealthy, - List<String> filterByTags) { - List<HealthService> activeServices = findService(serviceGroup, 0); - - return activeServices - .stream() - .filter(service -> allFiltersSupported(service, filterByTags)) - .filter(service -> !restrictToHealthy - || service.getChecks().stream().allMatch(check -> check.getStatus() == Check.CheckStatus.PASSING)) - .map(this::makeServiceUrl) - .collect(Collectors.toList()); - } - - private String makeServiceUrl(HealthService service) { - return service.getService().getAddress() + ":" + service.getService().getPort(); - } - - private boolean allFiltersSupported(HealthService service, - List<String> filterByTags) { - return new HashSet<>(service.getService().getTags()).containsAll(filterByTags); - } - - private List<HealthService> findService(String serviceGroup, int retryCount) { - HealthServicesRequest request = HealthServicesRequest.newBuilder() - .setPassing(true) - .setQueryParams(QueryParams.DEFAULT) - .build(); - Response<List<HealthService>> healthyServicesResp = consulInstance().getHealthServices(serviceGroup, request); - var healthyServices = healthyServicesResp.getValue(); - if (healthyServices.size() == 0) { - if (retryCount < MAX_RETRIES) { - try { - retryCount++; - TimeUnit.SECONDS.sleep(1); - return findService(serviceGroup, retryCount); - } catch (InterruptedException e) { - e.printStackTrace(); - return Collections.emptyList(); - } - } else { - return Collections.emptyList(); - } - } else { - return healthyServices; - } - } - - -} diff --git a/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/SpConsulKvManagement.java b/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/SpConsulKvManagement.java deleted file mode 100644 index 31247feca..000000000 --- a/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/SpConsulKvManagement.java +++ /dev/null @@ -1,74 +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.svcdiscovery.consul; - -import org.apache.streampipes.commons.environment.Environment; -import org.apache.streampipes.svcdiscovery.api.ISpKvManagement; - -import com.ecwid.consul.v1.Response; -import com.ecwid.consul.v1.kv.model.GetValue; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.util.HashMap; -import java.util.List; -import java.util.Map; - -public class SpConsulKvManagement extends AbstractConsulService implements ISpKvManagement { - - private static final Logger LOG = LoggerFactory.getLogger(SpConsulKvManagement.class); - - private static final String CONSUL_NAMESPACE = "/sp/v1/"; - - public SpConsulKvManagement(Environment environment) { - super(environment); - } - - public Map<String, String> getKeyValue(String route) { - var consul = consulInstance(); - - Map<String, String> keyValues = new HashMap<>(); - Response<List<GetValue>> consulResponseWithValues = consul.getKVValues(route); - - if (consulResponseWithValues.getValue() != null) { - for (GetValue value : consulResponseWithValues.getValue()) { - String key = value.getKey(); - String v = ""; - if (value.getValue() != null) { - v = value.getDecodedValue(); - } - keyValues.put(key, v); - } - } - return keyValues; - } - - public void updateConfig(String key, String entry, boolean password) { - var consul = consulInstance(); - if (!password) { - LOG.info("Updated config - key:" + key + " value: " + entry); - consul.setKVValue(key, entry); - } - } - - public void deleteConfig(String key) { - var consul = consulInstance(); - LOG.info("Delete config: {}", key); - consul.deleteKVValue(CONSUL_NAMESPACE + 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 deleted file mode 100644 index df8fb3837..000000000 --- a/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/SpConsulServiceDiscovery.java +++ /dev/null @@ -1,156 +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.svcdiscovery.consul; - -import org.apache.streampipes.commons.environment.Environment; -import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceRegistration; -import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceTag; -import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceTagPrefix; -import org.apache.streampipes.svcdiscovery.api.ISpServiceDiscovery; -import org.apache.streampipes.svcdiscovery.api.model.DefaultSpServiceGroups; -import org.apache.streampipes.svcdiscovery.api.model.DefaultSpServiceTags; - -import com.ecwid.consul.v1.Response; -import com.ecwid.consul.v1.agent.model.Check; -import com.ecwid.consul.v1.agent.model.NewService; -import com.ecwid.consul.v1.agent.model.Service; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.util.Collections; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.stream.Collectors; - - -public class SpConsulServiceDiscovery extends AbstractConsulService implements ISpServiceDiscovery { - - private static final Logger LOG = LoggerFactory.getLogger(SpConsulServiceDiscovery.class); - - private static final String HTTP_PROTOCOL = "http://";; - private static final String COLON = ":"; - private static final String HEALTH_CHECK_INTERVAL = "10s"; - - private ConsulHealthServiceManager healthServiceManager; - - public SpConsulServiceDiscovery(Environment environment) { - super(environment); - this.healthServiceManager = new ConsulHealthServiceManager(environment); - } - - @Override - public void registerService(SpServiceRegistration req) { - consulInstance().agentServiceRegister(createRegistrationBody(req)); - LOG.info("Successfully registered service at Consul: " + req.getSvcId()); - } - - private List<String> asString(List<SpServiceTag> tags) { - return tags.stream().map(SpServiceTag::asString).collect(Collectors.toList()); - } - - @Override - public List<String> getActivePipelineElementEndpoints() { - LOG.info("Discovering active pipeline element service endpoints"); - return getServiceEndpoints(DefaultSpServiceGroups.EXT, true, - Collections.singletonList(DefaultSpServiceTags.PE.asString())); - } - - @Override - public List<String> getActiveConnectWorkerEndpoints() { - LOG.info("Discovering active StreamPipes Connect worker service endpoints"); - return getServiceEndpoints(DefaultSpServiceGroups.EXT, true, - Collections.singletonList(DefaultSpServiceTags.CONNECT_WORKER.asString())); - } - - @Override - public List<String> getServiceEndpoints(String svcGroup, boolean restrictToHealthy, List<String> filterByTags) { - return healthServiceManager.getServiceEndpoints(svcGroup, restrictToHealthy, filterByTags); - } - - @Override - public Map<String, String> getExtensionsServiceGroups() { - LOG.info("Load pipeline element service status"); - var consul = consulInstance(); - - Response<Map<String, Service>> servicesResp = consul.getAgentServices(); - Response<Map<String, Check>> checksResp = consul.getAgentChecks(); - - Map<String, String> peSvcs = new HashMap<>(); - if (servicesResp.getValue() != null) { - var services = servicesResp.getValue(); - var checks = checksResp.getValue(); - for (Map.Entry<String, Service> entry : services.entrySet()) { - 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().name(); - } - LOG.info("Service id: " + serviceId + " service status: " + 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) { - var consul = consulInstance(); - LOG.info("Deregister service: " + svcId); - consul.agentServiceDeregister(svcId); - } - - private NewService createRegistrationBody(SpServiceRegistration req) { - var service = new NewService(); - service.setId(req.getSvcId()); - service.setName(req.getSvcGroup()); - service.setPort(req.getPort()); - service.setAddress(HTTP_PROTOCOL + req.getHost()); - service.setCheck(createServiceCheck(req)); - - service.setTags(asString(req.getTags())); - service.setEnableTagOverride(true); - - return service; - } - - private NewService.Check createServiceCheck(SpServiceRegistration req) { - var serviceCheck = new NewService.Check(); - - serviceCheck.setHttp(HTTP_PROTOCOL + req.getHost() + COLON + req.getPort() + req.getHealthCheckPath()); - serviceCheck.setInterval(HEALTH_CHECK_INTERVAL); - serviceCheck.setDeregisterCriticalServiceAfter("120s"); - serviceCheck.setStatus("passing"); - - return serviceCheck; - } -} diff --git a/streampipes-service-discovery/src/main/java/org/apache/streampipes/svcdiscovery/SpKvManagementCore.java b/streampipes-service-discovery/src/main/java/org/apache/streampipes/svcdiscovery/SpKvManagementCore.java deleted file mode 100644 index 4d2a36f68..000000000 --- a/streampipes-service-discovery/src/main/java/org/apache/streampipes/svcdiscovery/SpKvManagementCore.java +++ /dev/null @@ -1,41 +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.svcdiscovery; - -import org.apache.streampipes.svcdiscovery.api.ISpKvManagement; - -import java.util.Map; - -public class SpKvManagementCore implements ISpKvManagement { - - @Override - public Map<String, String> getKeyValue(String route) { - return null; - } - - @Override - public void updateConfig(String key, String entry, boolean password) { - - } - - @Override - public void deleteConfig(String key) { - - } -} 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 a488b4310..e46169a04 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,7 +19,6 @@ package org.apache.streampipes.svcdiscovery; import org.apache.streampipes.commons.environment.Environment; import org.apache.streampipes.commons.environment.Environments; -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; @@ -34,14 +33,6 @@ public class SpServiceDiscovery { return new SpServiceDiscoveryCore(); } - public static ISpKvManagement getKeyValueStore() { - return new SpKvManagementCore(); - } - - public static ISpKvManagement getKeyValueStore(Environment environment) { - return new SpKvManagementCore(); - } - public static SpConfig getSpConfig(String serviceGroup) { return getSpConfig(serviceGroup, Environments.getEnvironment()); } diff --git a/streampipes-service-discovery/src/main/java/org/apache/streampipes/svcdiscovery/SpServiceDiscoveryCore.java b/streampipes-service-discovery/src/main/java/org/apache/streampipes/svcdiscovery/SpServiceDiscoveryCore.java index 357b53f23..2ad6bed1f 100644 --- a/streampipes-service-discovery/src/main/java/org/apache/streampipes/svcdiscovery/SpServiceDiscoveryCore.java +++ b/streampipes-service-discovery/src/main/java/org/apache/streampipes/svcdiscovery/SpServiceDiscoveryCore.java @@ -22,8 +22,8 @@ import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceRegistratio import org.apache.streampipes.storage.api.CRUDStorage; import org.apache.streampipes.storage.management.StorageDispatcher; 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.DefaultSpServiceTypes; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -31,7 +31,6 @@ import org.slf4j.LoggerFactory; import java.util.Collections; import java.util.HashSet; import java.util.List; -import java.util.Map; import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; @@ -46,25 +45,13 @@ public class SpServiceDiscoveryCore implements ISpServiceDiscovery { this.serviceStorage = StorageDispatcher.INSTANCE.getNoSqlStore().getExtensionsServiceStorage(); } - @Override - public void registerService(SpServiceRegistration serviceRegistrationRequest) { - // not needed - } - @Override public List<String> getActivePipelineElementEndpoints() { LOG.info("Discovering active pipeline element service endpoints"); - return getServiceEndpoints(DefaultSpServiceGroups.EXT, true, + return getServiceEndpoints(DefaultSpServiceTypes.EXT, true, Collections.singletonList(DefaultSpServiceTags.PE.asString())); } - @Override - public List<String> getActiveConnectWorkerEndpoints() { - LOG.info("Discovering active StreamPipes Connect worker service endpoints"); - return getServiceEndpoints(DefaultSpServiceGroups.EXT, true, - Collections.singletonList(DefaultSpServiceTags.CONNECT_WORKER.asString())); - } - @Override public List<String> getServiceEndpoints(String serviceGroup, boolean restrictToHealthy, @@ -80,16 +67,6 @@ public class SpServiceDiscoveryCore implements ISpServiceDiscovery { .collect(Collectors.toList()); } - @Override - public Map<String, String> getExtensionsServiceGroups() { - return null; - } - - @Override - public void deregisterService(String svcId) { - // not needed - } - private String makeServiceUrl(SpServiceRegistration service) { return service.getScheme() + "://" + service.getHost() + ":" + service.getPort(); } diff --git a/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/StreamPipesExtensionsServiceBase.java b/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/StreamPipesExtensionsServiceBase.java index 7ef2dc0cd..0c4226e6c 100644 --- a/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/StreamPipesExtensionsServiceBase.java +++ b/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/StreamPipesExtensionsServiceBase.java @@ -30,7 +30,7 @@ import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceTag; import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceTagPrefix; import org.apache.streampipes.service.base.BaseNetworkingConfig; import org.apache.streampipes.service.base.StreamPipesServiceBase; -import org.apache.streampipes.svcdiscovery.api.model.DefaultSpServiceGroups; +import org.apache.streampipes.svcdiscovery.api.model.DefaultSpServiceTypes; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -75,7 +75,7 @@ public abstract class StreamPipesExtensionsServiceBase extends StreamPipesServic SpServiceDefinition serviceDef, BaseNetworkingConfig networkingConfig) throws UnknownHostException { var req = SpServiceRegistration.from( - DefaultSpServiceGroups.EXT, + DefaultSpServiceTypes.EXT, serviceDef.getServiceGroup(), serviceId(), networkingConfig.getHost(), @@ -89,8 +89,6 @@ public abstract class StreamPipesExtensionsServiceBase extends StreamPipesServic this.registerConfigs(serviceDef.getServiceGroup(), serviceDef.getServiceName(), serviceDef.getKvConfigs()); this.startStreamPipesService( serviceClass, - DefaultSpServiceGroups.EXT, - serviceId(), networkingConfig ); @@ -117,7 +115,6 @@ public abstract class StreamPipesExtensionsServiceBase extends StreamPipesServic } } - @Override protected List<SpServiceTag> getServiceTags() { List<SpServiceTag> tags = new ArrayList<>(); if (DeclarersSingleton.getInstance().getServiceDefinition() != null) {
