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 99231f4783c7df2efed30ef1d7c0cf5f133b5be7 Author: Dominik Riemer <[email protected]> AuthorDate: Wed Aug 2 09:08:27 2023 +0200 Remove consul dependency from service discovery --- streampipes-client-api/pom.xml | 5 + .../apache/streampipes/client/api/IAdminApi.java | 9 ++ .../streampipes/client/api/AbstractClientApi.java | 6 + .../apache/streampipes/client/api/AdminApi.java | 29 +++++ .../streampipes/client/http/PostRequest.java | 2 +- .../client/http/PostRequestWithoutPayload.java | 27 ++-- ...StreamPipesClientRuntimeConnectionResolver.java | 13 +- .../management/init/DeclarersSingleton.java | 11 +- .../shared/impl/AbstractSharedRestInterface.java | 4 + .../streampipes/rest/impl/admin/ConsulConfig.java | 9 +- .../impl/admin/ServiceConfigurationResource.java | 53 ++++++++ .../impl/admin/ServiceRegistrationResource.java | 62 +++++++++ .../service/base/StreamPipesServiceBase.java | 24 ---- .../service/core/StreamPipesCoreApplication.java | 5 +- .../service/core/StreamPipesResourceConfig.java | 4 + streampipes-service-discovery-api/pom.xml | 6 + .../api/model/SpServiceConfiguration.java | 75 +++++++++++ .../api/model/SpServiceRegistrationRequest.java | 26 +++- .../svcdiscovery/api/model/SpServiceTag.java | 24 +++- streampipes-service-discovery/pom.xml | 7 +- .../streampipes/svcdiscovery/SpConfigCore.java | 121 ++++++++++++++++++ .../svcdiscovery/SpKvManagementCore.java | 26 ++-- .../svcdiscovery/SpServiceDiscovery.java | 14 +-- .../svcdiscovery/SpServiceDiscoveryCore.java | 139 +++++++++++++++++++++ .../StreamPipesExtensionsServiceBase.java | 27 ++++ streampipes-storage-api/pom.xml | 5 + .../streampipes/storage/api/INoSqlStorage.java | 7 ++ streampipes-storage-couchdb/pom.xml | 2 +- .../storage/couchdb/CouchDbStorageManager.java | 15 +++ .../impl/ExtensionsServiceConfigStorageImpl.java | 60 +++++++++ .../couchdb/impl/ExtensionsServiceStorageImpl.java | 61 +++++++++ .../streampipes/storage/couchdb/utils/Utils.java | 8 ++ 32 files changed, 800 insertions(+), 86 deletions(-) diff --git a/streampipes-client-api/pom.xml b/streampipes-client-api/pom.xml index d54bbbb9a..d3f983564 100644 --- a/streampipes-client-api/pom.xml +++ b/streampipes-client-api/pom.xml @@ -45,6 +45,11 @@ <artifactId>streampipes-model</artifactId> <version>0.93.0-SNAPSHOT</version> </dependency> + <dependency> + <groupId>org.apache.streampipes</groupId> + <artifactId>streampipes-service-discovery-api</artifactId> + <version>0.93.0-SNAPSHOT</version> + </dependency> <dependency> <groupId>org.apache.httpcomponents</groupId> <artifactId>fluent-hc</artifactId> diff --git a/streampipes-client-api/src/main/java/org/apache/streampipes/client/api/IAdminApi.java b/streampipes-client-api/src/main/java/org/apache/streampipes/client/api/IAdminApi.java index d8a4ee591..14f3fb3b6 100644 --- a/streampipes-client-api/src/main/java/org/apache/streampipes/client/api/IAdminApi.java +++ b/streampipes-client-api/src/main/java/org/apache/streampipes/client/api/IAdminApi.java @@ -21,10 +21,19 @@ package org.apache.streampipes.client.api; import org.apache.streampipes.model.config.MessagingSettings; import org.apache.streampipes.model.connect.adapter.AdapterDescription; import org.apache.streampipes.model.function.FunctionDefinition; +import org.apache.streampipes.svcdiscovery.api.model.SpServiceConfiguration; +import org.apache.streampipes.svcdiscovery.api.model.SpServiceRegistrationRequest; import java.util.List; public interface IAdminApi { + + void registerService(SpServiceRegistrationRequest serviceRegistration); + + void deregisterService(String serviceId); + + void registerServiceConfiguration(SpServiceConfiguration serviceConfiguration); + void registerAdapters(List<AdapterDescription> adapters); void registerFunctions(List<FunctionDefinition> functions); diff --git a/streampipes-client/src/main/java/org/apache/streampipes/client/api/AbstractClientApi.java b/streampipes-client/src/main/java/org/apache/streampipes/client/api/AbstractClientApi.java index 2297b5302..78c9d02bc 100644 --- a/streampipes-client/src/main/java/org/apache/streampipes/client/api/AbstractClientApi.java +++ b/streampipes-client/src/main/java/org/apache/streampipes/client/api/AbstractClientApi.java @@ -20,6 +20,7 @@ package org.apache.streampipes.client.api; import org.apache.streampipes.client.http.DeleteRequest; import org.apache.streampipes.client.http.GetRequest; import org.apache.streampipes.client.http.PostRequestWithPayloadResponse; +import org.apache.streampipes.client.http.PostRequestWithoutPayload; import org.apache.streampipes.client.http.PostRequestWithoutPayloadResponse; import org.apache.streampipes.client.http.PutRequest; import org.apache.streampipes.client.model.StreamPipesClientConfig; @@ -41,6 +42,11 @@ public class AbstractClientApi { return new PostRequestWithPayloadResponse<>(clientConfig, apiPath, serializer, responseClass).executeRequest(); } + protected void post(StreamPipesApiPath apiPath) { + ObjectSerializer<Void, Void> serializer = new ObjectSerializer<>(); + new PostRequestWithoutPayload<>(clientConfig, apiPath, serializer).executeRequest(); + } + protected <T> void post(StreamPipesApiPath apiPath, T object) { ObjectSerializer<T, Void> serializer = new ObjectSerializer<>(); new PostRequestWithoutPayloadResponse<>(clientConfig, apiPath, serializer, object).executeRequest(); diff --git a/streampipes-client/src/main/java/org/apache/streampipes/client/api/AdminApi.java b/streampipes-client/src/main/java/org/apache/streampipes/client/api/AdminApi.java index 74c89fbef..90f81f8f6 100644 --- a/streampipes-client/src/main/java/org/apache/streampipes/client/api/AdminApi.java +++ b/streampipes-client/src/main/java/org/apache/streampipes/client/api/AdminApi.java @@ -23,6 +23,8 @@ import org.apache.streampipes.model.config.MessagingSettings; import org.apache.streampipes.model.connect.adapter.AdapterDescription; import org.apache.streampipes.model.function.FunctionDefinition; import org.apache.streampipes.model.message.SuccessMessage; +import org.apache.streampipes.svcdiscovery.api.model.SpServiceConfiguration; +import org.apache.streampipes.svcdiscovery.api.model.SpServiceRegistrationRequest; import java.util.List; @@ -32,6 +34,21 @@ public class AdminApi extends AbstractClientApi implements IAdminApi { super(clientConfig); } + @Override + public void registerService(SpServiceRegistrationRequest serviceRegistration) { + post(getExtensionsServiceRegistrationPath(), serviceRegistration); + } + + @Override + public void deregisterService(String serviceId) { + post(getExtensionsServiceRegistrationPath().addToPath(serviceId)); + } + + @Override + public void registerServiceConfiguration(SpServiceConfiguration serviceConfiguration) { + post(getExtensionsServiceConfigurationPath(), serviceConfiguration); + } + @Override public void registerAdapters(List<AdapterDescription> adapters) { post(getConnectPath(), adapters); @@ -52,6 +69,18 @@ public class AdminApi extends AbstractClientApi implements IAdminApi { return getSingle(getMessagingSettingsPath(), MessagingSettings.class); } + private StreamPipesApiPath getExtensionsServiceRegistrationPath() { + return StreamPipesApiPath + .fromBaseApiPath() + .addToPath("extensions-services"); + } + + private StreamPipesApiPath getExtensionsServiceConfigurationPath() { + return StreamPipesApiPath + .fromBaseApiPath() + .addToPath("extensions-services-configurations"); + } + private StreamPipesApiPath getMessagingSettingsPath() { return StreamPipesApiPath .fromBaseApiPath() diff --git a/streampipes-client/src/main/java/org/apache/streampipes/client/http/PostRequest.java b/streampipes-client/src/main/java/org/apache/streampipes/client/http/PostRequest.java index fc6de1374..15293e1b5 100644 --- a/streampipes-client/src/main/java/org/apache/streampipes/client/http/PostRequest.java +++ b/streampipes-client/src/main/java/org/apache/streampipes/client/http/PostRequest.java @@ -27,7 +27,7 @@ import org.apache.http.entity.ContentType; public abstract class PostRequest<K, V, T> extends HttpRequest<K, V, T> { private K body; - private boolean withBody; + private final boolean withBody; public PostRequest(StreamPipesClientConfig clientConfig, StreamPipesApiPath apiPath, diff --git a/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/SpServiceTag.java b/streampipes-client/src/main/java/org/apache/streampipes/client/http/PostRequestWithoutPayload.java similarity index 54% copy from streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/SpServiceTag.java copy to streampipes-client/src/main/java/org/apache/streampipes/client/http/PostRequestWithoutPayload.java index baabf90f3..d685e2169 100644 --- a/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/SpServiceTag.java +++ b/streampipes-client/src/main/java/org/apache/streampipes/client/http/PostRequestWithoutPayload.java @@ -15,25 +15,24 @@ * limitations under the License. * */ -package org.apache.streampipes.svcdiscovery.api.model; -public class SpServiceTag { +package org.apache.streampipes.client.http; - private static final String COLON = ":"; - private final SpServiceTagPrefix prefix; - private final String value; +import org.apache.streampipes.client.model.StreamPipesClientConfig; +import org.apache.streampipes.client.serializer.Serializer; +import org.apache.streampipes.client.util.StreamPipesApiPath; - private SpServiceTag(SpServiceTagPrefix prefix, String value) { - this.prefix = prefix; - this.value = value; - } +import org.apache.http.HttpEntity; + +public class PostRequestWithoutPayload<K, V> extends PostRequest<K, V, Void> { - public static SpServiceTag create(SpServiceTagPrefix prefix, - String value) { - return new SpServiceTag(prefix, value); + public PostRequestWithoutPayload(StreamPipesClientConfig clientConfig, + StreamPipesApiPath apiPath, Serializer<K, V, Void> serializer) { + super(clientConfig, apiPath, serializer); } - public String asString() { - return prefix.asString() + COLON + value; + @Override + protected Void afterRequest(Serializer<K, V, Void> serializer, HttpEntity entity) { + return null; } } 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 cd1735f18..8d53455ba 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 @@ -24,15 +24,11 @@ import org.apache.streampipes.commons.environment.Environment; import org.apache.streampipes.commons.environment.Environments; import org.apache.streampipes.commons.exceptions.SpRuntimeException; import org.apache.streampipes.commons.networking.Networking; -import org.apache.streampipes.svcdiscovery.SpServiceDiscovery; -import org.apache.streampipes.svcdiscovery.api.model.DefaultSpServiceGroups; -import org.apache.streampipes.svcdiscovery.api.model.DefaultSpServiceTags; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.net.UnknownHostException; -import java.util.Collections; import java.util.List; public class StreamPipesClientRuntimeConnectionResolver implements ClientConnectionUrlResolver { @@ -76,12 +72,7 @@ public class StreamPipesClientRuntimeConnectionResolver implements ClientConnect } private List<String> findClientServices() { - return SpServiceDiscovery - .getServiceDiscovery() - .getServiceEndpoints( - DefaultSpServiceGroups.CORE, - true, - Collections.singletonList(DefaultSpServiceTags.STREAMPIPES_CLIENT.asString()) - ); + // TODO check how to identify backend + return List.of("backend"); } } diff --git a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/init/DeclarersSingleton.java b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/init/DeclarersSingleton.java index 80f4eca7a..4a925dc41 100644 --- a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/init/DeclarersSingleton.java +++ b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/init/DeclarersSingleton.java @@ -18,6 +18,7 @@ package org.apache.streampipes.extensions.management.init; +import org.apache.streampipes.client.StreamPipesClient; import org.apache.streampipes.dataformat.SpDataFormatFactory; import org.apache.streampipes.dataformat.SpDataFormatManager; import org.apache.streampipes.extensions.api.connect.StreamPipesAdapter; @@ -27,15 +28,15 @@ import org.apache.streampipes.extensions.api.pe.IStreamPipesDataSink; import org.apache.streampipes.extensions.api.pe.IStreamPipesDataStream; import org.apache.streampipes.extensions.api.pe.IStreamPipesPipelineElement; import org.apache.streampipes.extensions.api.pe.runtime.IStreamPipesRuntimeProvider; +import org.apache.streampipes.extensions.management.client.StreamPipesClientResolver; import org.apache.streampipes.extensions.management.model.SpServiceDefinition; import org.apache.streampipes.messaging.SpProtocolDefinitionFactory; 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.svcdiscovery.api.model.SpServiceConfiguration; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -115,9 +116,9 @@ public class DeclarersSingleton implements IDeclarersSingleton { 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, "")); + StreamPipesClient client = new StreamPipesClientResolver().makeStreamPipesClientInstance(); + var serviceConfiguration = new SpServiceConfiguration(serviceGroup, serviceName, configs); + client.adminApi().registerServiceConfiguration(serviceConfiguration); } public void addDeclarers(List<IStreamPipesPipelineElement<?>> allPipelineElements) { diff --git a/streampipes-rest-shared/src/main/java/org/apache/streampipes/rest/shared/impl/AbstractSharedRestInterface.java b/streampipes-rest-shared/src/main/java/org/apache/streampipes/rest/shared/impl/AbstractSharedRestInterface.java index 80a3d4ecb..2f7211807 100644 --- a/streampipes-rest-shared/src/main/java/org/apache/streampipes/rest/shared/impl/AbstractSharedRestInterface.java +++ b/streampipes-rest-shared/src/main/java/org/apache/streampipes/rest/shared/impl/AbstractSharedRestInterface.java @@ -58,6 +58,10 @@ public abstract class AbstractSharedRestInterface { return Response.ok().build(); } + protected Response created() { + return Response.status(201).build(); + } + protected Response fail() { return Response.serverError().build(); } diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/admin/ConsulConfig.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/admin/ConsulConfig.java index 17bdb22a8..ab1acffac 100644 --- a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/admin/ConsulConfig.java +++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/admin/ConsulConfig.java @@ -21,13 +21,12 @@ package org.apache.streampipes.rest.impl.admin; import org.apache.streampipes.config.backend.BackendConfig; import org.apache.streampipes.model.config.MessagingSettings; -import org.apache.streampipes.rest.core.base.impl.AbstractRestResource; +import org.apache.streampipes.rest.core.base.impl.AbstractAuthGuardedRestResource; import org.apache.streampipes.rest.security.AuthConstants; import org.apache.streampipes.rest.shared.annotation.JacksonSerialized; import org.apache.streampipes.svcdiscovery.api.ISpKvManagement; import org.apache.streampipes.svcdiscovery.api.model.ConfigItem; import org.apache.streampipes.svcdiscovery.api.model.PeConfig; -import org.apache.streampipes.svcdiscovery.consul.ConsulSpConfig; import com.google.gson.Gson; import com.google.gson.JsonObject; @@ -52,7 +51,7 @@ import java.util.Map; @Path("/v2/consul") @Component -public class ConsulConfig extends AbstractRestResource { +public class ConsulConfig extends AbstractAuthGuardedRestResource { private static final Logger LOG = LoggerFactory.getLogger(ConsulConfig.class); @@ -68,7 +67,7 @@ public class ConsulConfig extends AbstractRestResource { for (Map.Entry<String, String> entry : peServices.entrySet()) { String serviceStatus = entry.getValue(); - String mainKey = ConsulSpConfig.SERVICE_ROUTE_PREFIX + entry.getKey(); + String mainKey = entry.getKey(); Map<String, String> meta = new HashMap<>(); meta.put("status", serviceStatus); @@ -181,7 +180,7 @@ public class ConsulConfig extends AbstractRestResource { } public List<ConfigItem> getConfigForService(String serviceId) { - Map<String, String> keyValues = getKeyValueStore().getKeyValue(ConsulSpConfig.SERVICE_ROUTE_PREFIX + serviceId); + Map<String, String> keyValues = getKeyValueStore().getKeyValue(serviceId); List<ConfigItem> configItems = new LinkedList<>(); diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/admin/ServiceConfigurationResource.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/admin/ServiceConfigurationResource.java new file mode 100644 index 000000000..b73fe7455 --- /dev/null +++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/admin/ServiceConfigurationResource.java @@ -0,0 +1,53 @@ +/* + * 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.rest.impl.admin; + +import org.apache.streampipes.rest.core.base.impl.AbstractAuthGuardedRestResource; +import org.apache.streampipes.rest.security.AuthConstants; +import org.apache.streampipes.storage.api.CRUDStorage; +import org.apache.streampipes.svcdiscovery.api.model.SpServiceConfiguration; + +import org.springframework.security.access.prepost.PreAuthorize; +import org.springframework.stereotype.Component; + +import jakarta.ws.rs.Consumes; +import jakarta.ws.rs.POST; +import jakarta.ws.rs.Path; +import jakarta.ws.rs.core.MediaType; +import jakarta.ws.rs.core.Response; + +@Path("/v2/extensions-services-configurations") +@Component +@PreAuthorize(AuthConstants.IS_ADMIN_ROLE) +public class ServiceConfigurationResource extends AbstractAuthGuardedRestResource { + + private final CRUDStorage<String, SpServiceConfiguration> extensionsServicesConfigStorage = + getNoSqlStorage().getExtensionsServiceConfigurationStorage(); + + @POST + @Consumes(MediaType.APPLICATION_JSON) + public Response registerServiceConfiguration(SpServiceConfiguration serviceConfiguration) { + if (extensionsServicesConfigStorage.getElementById(serviceConfiguration.getServiceGroup()) == null) { + extensionsServicesConfigStorage.createElement(serviceConfiguration); + return created(); + } else { + return ok(); + } + } +} diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/admin/ServiceRegistrationResource.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/admin/ServiceRegistrationResource.java new file mode 100644 index 000000000..63928cece --- /dev/null +++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/admin/ServiceRegistrationResource.java @@ -0,0 +1,62 @@ +/* + * 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.rest.impl.admin; + +import org.apache.streampipes.rest.core.base.impl.AbstractAuthGuardedRestResource; +import org.apache.streampipes.rest.security.AuthConstants; +import org.apache.streampipes.storage.api.CRUDStorage; +import org.apache.streampipes.svcdiscovery.api.model.SpServiceRegistrationRequest; + +import org.springframework.security.access.prepost.PreAuthorize; +import org.springframework.stereotype.Component; + +import jakarta.ws.rs.Consumes; +import jakarta.ws.rs.POST; +import jakarta.ws.rs.Path; +import jakarta.ws.rs.PathParam; +import jakarta.ws.rs.core.MediaType; +import jakarta.ws.rs.core.Response; + +@Path("/v2/extensions-services") +@Component +@PreAuthorize(AuthConstants.IS_ADMIN_ROLE) +public class ServiceRegistrationResource extends AbstractAuthGuardedRestResource { + + private final CRUDStorage<String, SpServiceRegistrationRequest> extensionsServiceStorage = + getNoSqlStorage().getExtensionsServiceStorage(); + + @POST + @Consumes(MediaType.APPLICATION_JSON) + public Response registerService(SpServiceRegistrationRequest serviceRegistration) { + extensionsServiceStorage.createElement(serviceRegistration); + return ok(); + } + + @POST + @Path("/{serviceId}") + public Response unregisterService(@PathParam("serviceId") String serviceId) { + try { + var serviceRegistration = extensionsServiceStorage.getElementById(serviceId); + extensionsServiceStorage.deleteElement(serviceRegistration); + return ok(); + } catch (IllegalArgumentException e) { + return badRequest("Could not find registered service with id " + serviceId); + } + } +} 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 5b3a5f84c..14a05e56d 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,8 +17,6 @@ */ package org.apache.streampipes.service.base; -import org.apache.streampipes.svcdiscovery.SpServiceDiscovery; -import org.apache.streampipes.svcdiscovery.api.model.SpServiceRegistrationRequest; import org.apache.streampipes.svcdiscovery.api.model.SpServiceTag; import org.apache.commons.lang3.RandomStringUtils; @@ -39,7 +37,6 @@ public abstract class StreamPipesServiceBase { String serviceGroup, String serviceId, BaseNetworkingConfig networkingConfig) throws UnknownHostException { - registerService(serviceGroup, serviceId, networkingConfig); runApplication(serviceClass, networkingConfig.getPort()); } @@ -50,29 +47,8 @@ public abstract class StreamPipesServiceBase { app.run(); } - private void registerService(String serviceGroup, - String serviceId, - BaseNetworkingConfig networkingConfig) { - SpServiceRegistrationRequest req = SpServiceRegistrationRequest.from( - serviceGroup, - serviceId, - networkingConfig.getHost(), - networkingConfig.getPort(), - getServiceTags(), - getHealthCheckPath()); - - SpServiceDiscovery - .getServiceDiscovery() - .registerService(req); - } - protected abstract List<SpServiceTag> getServiceTags(); - protected void deregisterService(String serviceId) { - LOG.info("Deregistering service (id={})...", serviceId); - SpServiceDiscovery.getServiceDiscovery().deregisterService(serviceId); - } - 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 295f60bd9..969b760ea 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 @@ -128,7 +128,8 @@ public class StreamPipesCoreApplication extends StreamPipesServiceBase { this.healthCheckExecutorService = Executors.newSingleThreadScheduledExecutor(); this.logCheckExecutorService = Executors.newSingleThreadScheduledExecutor(); - new StreamPipesEnvChecker().updateEnvironmentVariables(); + // TODO + //new StreamPipesEnvChecker().updateEnvironmentVariables(); new CouchDbViewGenerator().createGenericDatabaseIfNotExists(); if (!isConfigured()) { @@ -205,8 +206,6 @@ public class StreamPipesCoreApplication extends StreamPipesServiceBase { } }); - deregisterService(serviceId()); - LOG.info("Thanks for using Apache StreamPipes - see you next time!"); } diff --git a/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/StreamPipesResourceConfig.java b/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/StreamPipesResourceConfig.java index a27595a94..63da27b65 100644 --- a/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/StreamPipesResourceConfig.java +++ b/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/StreamPipesResourceConfig.java @@ -64,6 +64,8 @@ import org.apache.streampipes.rest.impl.admin.ExtensionsServiceEndpointResource; import org.apache.streampipes.rest.impl.admin.GeneralConfigurationResource; import org.apache.streampipes.rest.impl.admin.PermissionResource; import org.apache.streampipes.rest.impl.admin.PipelineElementImport; +import org.apache.streampipes.rest.impl.admin.ServiceConfigurationResource; +import org.apache.streampipes.rest.impl.admin.ServiceRegistrationResource; import org.apache.streampipes.rest.impl.admin.UserAdminResource; import org.apache.streampipes.rest.impl.admin.UserGroupResource; import org.apache.streampipes.rest.impl.connect.AdapterResource; @@ -152,7 +154,9 @@ public class StreamPipesResourceConfig extends BaseResourceConfig { Setup.class, ResetResource.class, RestorePasswordResource.class, + ServiceConfigurationResource.class, ServiceHealthResource.class, + ServiceRegistrationResource.class, UserResource.class, UserAdminResource.class, Version.class, diff --git a/streampipes-service-discovery-api/pom.xml b/streampipes-service-discovery-api/pom.xml index f56faa912..67464858e 100644 --- a/streampipes-service-discovery-api/pom.xml +++ b/streampipes-service-discovery-api/pom.xml @@ -27,6 +27,12 @@ <artifactId>streampipes-service-discovery-api</artifactId> + <dependencies> + <dependency> + <groupId>com.google.code.gson</groupId> + <artifactId>gson</artifactId> + </dependency> + </dependencies> <build> <plugins> <plugin> diff --git a/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/SpServiceConfiguration.java b/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/SpServiceConfiguration.java new file mode 100644 index 000000000..eaa0c8797 --- /dev/null +++ b/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/SpServiceConfiguration.java @@ -0,0 +1,75 @@ +/* + * 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.model; + +import com.google.gson.annotations.SerializedName; + +import java.util.Map; + +public class SpServiceConfiguration { + + protected @SerializedName("_rev") String rev; + private @SerializedName("_id") String serviceGroup; + + String serviceName; + Map<String, ConfigItem> configs; + + public SpServiceConfiguration() { + } + + public SpServiceConfiguration(String serviceGroup, + String serviceName, + Map<String, ConfigItem> configs) { + this.serviceGroup = serviceGroup; + this.serviceName = serviceName; + this.configs = configs; + } + + public String getServiceGroup() { + return serviceGroup; + } + + public void setServiceGroup(String serviceGroup) { + this.serviceGroup = serviceGroup; + } + + public String getServiceName() { + return serviceName; + } + + public void setServiceName(String serviceName) { + this.serviceName = serviceName; + } + + public Map<String, ConfigItem> getConfigs() { + return configs; + } + + public void setConfigs(Map<String, ConfigItem> configs) { + this.configs = configs; + } + + public String getRev() { + return rev; + } + + public void setRev(String rev) { + this.rev = rev; + } +} diff --git a/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/SpServiceRegistrationRequest.java b/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/SpServiceRegistrationRequest.java index 0b0e6c0d2..8c1e42bb2 100644 --- a/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/SpServiceRegistrationRequest.java +++ b/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/SpServiceRegistrationRequest.java @@ -17,16 +17,24 @@ */ package org.apache.streampipes.svcdiscovery.api.model; +import com.google.gson.annotations.SerializedName; + import java.util.List; public class SpServiceRegistrationRequest { private String svcGroup; - private String svcId; + + protected @SerializedName("_rev") String rev; + private @SerializedName("_id") String svcId; private String host; private int port; private List<SpServiceTag> tags; private String healthCheckPath; + private boolean healthy = true; + + public SpServiceRegistrationRequest() { + } public SpServiceRegistrationRequest(String svcGroup, String svcId, @@ -106,4 +114,20 @@ public class SpServiceRegistrationRequest { public void setHealthCheckPath(String healthCheckPath) { this.healthCheckPath = healthCheckPath; } + + public String getRev() { + return rev; + } + + public void setRev(String rev) { + this.rev = rev; + } + + public boolean isHealthy() { + return healthy; + } + + public void setHealthy(boolean healthy) { + this.healthy = healthy; + } } diff --git a/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/SpServiceTag.java b/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/SpServiceTag.java index baabf90f3..450ebcbc4 100644 --- a/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/SpServiceTag.java +++ b/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/SpServiceTag.java @@ -20,8 +20,12 @@ package org.apache.streampipes.svcdiscovery.api.model; public class SpServiceTag { private static final String COLON = ":"; - private final SpServiceTagPrefix prefix; - private final String value; + private SpServiceTagPrefix prefix; + private String value; + + public SpServiceTag() { + + } private SpServiceTag(SpServiceTagPrefix prefix, String value) { this.prefix = prefix; @@ -36,4 +40,20 @@ public class SpServiceTag { public String asString() { return prefix.asString() + COLON + value; } + + public SpServiceTagPrefix getPrefix() { + return prefix; + } + + public void setPrefix(SpServiceTagPrefix prefix) { + this.prefix = prefix; + } + + public String getValue() { + return value; + } + + public void setValue(String value) { + this.value = value; + } } diff --git a/streampipes-service-discovery/pom.xml b/streampipes-service-discovery/pom.xml index c32772c1a..3f1944a47 100644 --- a/streampipes-service-discovery/pom.xml +++ b/streampipes-service-discovery/pom.xml @@ -28,6 +28,11 @@ <artifactId>streampipes-service-discovery</artifactId> <dependencies> + <dependency> + <groupId>org.apache.streampipes</groupId> + <artifactId>streampipes-commons</artifactId> + <version>0.93.0-SNAPSHOT</version> + </dependency> <dependency> <groupId>org.apache.streampipes</groupId> <artifactId>streampipes-service-discovery-api</artifactId> @@ -35,7 +40,7 @@ </dependency> <dependency> <groupId>org.apache.streampipes</groupId> - <artifactId>streampipes-service-discovery-consul</artifactId> + <artifactId>streampipes-storage-management</artifactId> <version>0.93.0-SNAPSHOT</version> </dependency> </dependencies> diff --git a/streampipes-service-discovery/src/main/java/org/apache/streampipes/svcdiscovery/SpConfigCore.java b/streampipes-service-discovery/src/main/java/org/apache/streampipes/svcdiscovery/SpConfigCore.java new file mode 100644 index 000000000..399033997 --- /dev/null +++ b/streampipes-service-discovery/src/main/java/org/apache/streampipes/svcdiscovery/SpConfigCore.java @@ -0,0 +1,121 @@ +/* + * 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.SpConfig; +import org.apache.streampipes.svcdiscovery.api.model.ConfigItem; +import org.apache.streampipes.svcdiscovery.api.model.ConfigurationScope; + +public class SpConfigCore implements SpConfig { + + @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 register(ConfigItem configItem) { + + } + + @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-client-api/src/main/java/org/apache/streampipes/client/api/IAdminApi.java b/streampipes-service-discovery/src/main/java/org/apache/streampipes/svcdiscovery/SpKvManagementCore.java similarity index 61% copy from streampipes-client-api/src/main/java/org/apache/streampipes/client/api/IAdminApi.java copy to streampipes-service-discovery/src/main/java/org/apache/streampipes/svcdiscovery/SpKvManagementCore.java index d8a4ee591..4d2a36f68 100644 --- a/streampipes-client-api/src/main/java/org/apache/streampipes/client/api/IAdminApi.java +++ b/streampipes-service-discovery/src/main/java/org/apache/streampipes/svcdiscovery/SpKvManagementCore.java @@ -16,20 +16,26 @@ * */ -package org.apache.streampipes.client.api; +package org.apache.streampipes.svcdiscovery; -import org.apache.streampipes.model.config.MessagingSettings; -import org.apache.streampipes.model.connect.adapter.AdapterDescription; -import org.apache.streampipes.model.function.FunctionDefinition; +import org.apache.streampipes.svcdiscovery.api.ISpKvManagement; -import java.util.List; +import java.util.Map; -public interface IAdminApi { - void registerAdapters(List<AdapterDescription> adapters); +public class SpKvManagementCore implements ISpKvManagement { - void registerFunctions(List<FunctionDefinition> functions); + @Override + public Map<String, String> getKeyValue(String route) { + return null; + } - void deregisterFunction(String functionId); + @Override + public void updateConfig(String key, String entry, boolean password) { - MessagingSettings getMessagingSettings(); + } + + @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 cead1334b..e192ad76b 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 @@ -22,26 +22,23 @@ 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; -import org.apache.streampipes.svcdiscovery.consul.SpConsulKvManagement; -import org.apache.streampipes.svcdiscovery.consul.SpConsulServiceDiscovery; public class SpServiceDiscovery { public static ISpServiceDiscovery getServiceDiscovery() { - return new SpConsulServiceDiscovery(Environments.getEnvironment()); + return new SpServiceDiscoveryCore(); } public static ISpServiceDiscovery getServiceDiscovery(Environment environment) { - return new SpConsulServiceDiscovery(environment); + return new SpServiceDiscoveryCore(); } public static ISpKvManagement getKeyValueStore() { - return new SpConsulKvManagement(Environments.getEnvironment()); + return new SpKvManagementCore(); } public static ISpKvManagement getKeyValueStore(Environment environment) { - return new SpConsulKvManagement(environment); + return new SpKvManagementCore(); } public static SpConfig getSpConfig(String serviceGroup) { @@ -50,7 +47,8 @@ public class SpServiceDiscovery { public static SpConfig getSpConfig(String serviceGroup, Environment environment) { - return new ConsulSpConfig(serviceGroup, environment); + // TODO can probably be removed? + return null; } } 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 new file mode 100644 index 000000000..57ae2f054 --- /dev/null +++ b/streampipes-service-discovery/src/main/java/org/apache/streampipes/svcdiscovery/SpServiceDiscoveryCore.java @@ -0,0 +1,139 @@ +/* + * 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.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.SpServiceRegistrationRequest; +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 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; + +public class SpServiceDiscoveryCore implements ISpServiceDiscovery { + + private static final Logger LOG = LoggerFactory.getLogger(SpServiceDiscoveryCore.class); + private static final int MAX_RETRIES = 3; + + private final CRUDStorage<String, SpServiceRegistrationRequest> serviceStorage; + + public SpServiceDiscoveryCore() { + this.serviceStorage = StorageDispatcher.INSTANCE.getNoSqlStore().getExtensionsServiceStorage(); + } + + @Override + public void registerService(SpServiceRegistrationRequest serviceRegistrationRequest) { + // not needed + } + + @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 serviceGroup, + boolean restrictToHealthy, + List<String> filterByTags) { + List<SpServiceRegistrationRequest> activeServices = findService(serviceGroup, 0); + + return activeServices + .stream() + .filter(service -> allFiltersSupported(service, filterByTags)) + .filter(service -> !restrictToHealthy + || service.isHealthy()) + .map(this::makeServiceUrl) + .collect(Collectors.toList()); + } + + @Override + public Map<String, String> getExtensionsServiceGroups() { + return null; + } + + @Override + public void deregisterService(String svcId) { + // not needed + } + + private List<String> asString(List<SpServiceTag> tags) { + return tags.stream().map(SpServiceTag::asString).collect(Collectors.toList()); + } + + 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() + ":", ""); + } + + private String makeServiceUrl(SpServiceRegistrationRequest service) { + return service.getHost() + ":" + service.getPort(); + } + + private boolean allFiltersSupported(SpServiceRegistrationRequest service, + List<String> filterByTags) { + return new HashSet<>(service.getTags()).containsAll(filterByTags); + } + + private List<SpServiceRegistrationRequest> findService(String serviceGroup, + int retryCount) { + var services = serviceStorage.getAll(); + if (services.size() == 0) { + if (retryCount < MAX_RETRIES) { + try { + retryCount++; + TimeUnit.SECONDS.sleep(10); + return findService(serviceGroup, retryCount); + } catch (InterruptedException e) { + e.printStackTrace(); + return Collections.emptyList(); + } + } else { + return Collections.emptyList(); + } + } else { + return services; + } + } +} 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 b63498916..b9b1196a3 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 @@ -18,11 +18,14 @@ package org.apache.streampipes.service.extensions; +import org.apache.streampipes.client.StreamPipesClient; +import org.apache.streampipes.extensions.management.client.StreamPipesClientResolver; import org.apache.streampipes.extensions.management.init.DeclarersSingleton; import org.apache.streampipes.extensions.management.model.SpServiceDefinition; 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.SpServiceRegistrationRequest; import org.apache.streampipes.svcdiscovery.api.model.SpServiceTag; import org.apache.streampipes.svcdiscovery.api.model.SpServiceTagPrefix; @@ -68,15 +71,33 @@ public abstract class StreamPipesExtensionsServiceBase extends StreamPipesServic public void startExtensionsService(Class<?> serviceClass, SpServiceDefinition serviceDef, BaseNetworkingConfig networkingConfig) throws UnknownHostException { + this.registerService(DefaultSpServiceGroups.EXT, serviceId(), networkingConfig); + this.startStreamPipesService( serviceClass, DefaultSpServiceGroups.EXT, serviceId(), networkingConfig ); + this.afterServiceRegistered(serviceDef); } + private void registerService(String serviceGroup, + String serviceId, + BaseNetworkingConfig networkingConfig) { + SpServiceRegistrationRequest req = SpServiceRegistrationRequest.from( + serviceGroup, + serviceId, + networkingConfig.getHost(), + networkingConfig.getPort(), + getServiceTags(), + getHealthCheckPath()); + + StreamPipesClient client = new StreamPipesClientResolver().makeStreamPipesClientInstance(); + client.adminApi().registerService(req); + } + @Override protected List<SpServiceTag> getServiceTags() { List<SpServiceTag> tags = new ArrayList<>(); @@ -88,6 +109,12 @@ public abstract class StreamPipesExtensionsServiceBase extends StreamPipesServic return tags; } + protected void deregisterService(String serviceId) { + LOG.info("Deregistering service (id={})...", serviceId); + StreamPipesClient client = new StreamPipesClientResolver().makeStreamPipesClientInstance(); + client.adminApi().deregisterService(serviceId); + } + protected abstract List<SpServiceTag> getExtensionsServiceTags(); @PreDestroy diff --git a/streampipes-storage-api/pom.xml b/streampipes-storage-api/pom.xml index 754c9704b..15ae3f044 100644 --- a/streampipes-storage-api/pom.xml +++ b/streampipes-storage-api/pom.xml @@ -39,6 +39,11 @@ <artifactId>streampipes-model-client</artifactId> <version>0.93.0-SNAPSHOT</version> </dependency> + <dependency> + <groupId>org.apache.streampipes</groupId> + <artifactId>streampipes-service-discovery-api</artifactId> + <version>0.93.0-SNAPSHOT</version> + </dependency> </dependencies> <build> <plugins> diff --git a/streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/INoSqlStorage.java b/streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/INoSqlStorage.java index 6b4af153a..eee65bd0f 100644 --- a/streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/INoSqlStorage.java +++ b/streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/INoSqlStorage.java @@ -17,6 +17,9 @@ */ package org.apache.streampipes.storage.api; +import org.apache.streampipes.svcdiscovery.api.model.SpServiceConfiguration; +import org.apache.streampipes.svcdiscovery.api.model.SpServiceRegistrationRequest; + public interface INoSqlStorage { IGenericStorage getGenericStorage(); @@ -80,4 +83,8 @@ public interface INoSqlStorage { IPasswordRecoveryTokenStorage getPasswordRecoveryTokenStorage(); IUserActivationTokenStorage getUserActivationTokenStorage(); + + CRUDStorage<String, SpServiceRegistrationRequest> getExtensionsServiceStorage(); + + CRUDStorage<String, SpServiceConfiguration> getExtensionsServiceConfigurationStorage(); } diff --git a/streampipes-storage-couchdb/pom.xml b/streampipes-storage-couchdb/pom.xml index 963d0a25c..30d1efb7b 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-service-discovery</artifactId> + <artifactId>streampipes-service-discovery-api</artifactId> <version>0.93.0-SNAPSHOT</version> </dependency> <dependency> diff --git a/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/CouchDbStorageManager.java b/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/CouchDbStorageManager.java index fb9c28ca6..6318552f8 100644 --- a/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/CouchDbStorageManager.java +++ b/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/CouchDbStorageManager.java @@ -17,6 +17,7 @@ */ package org.apache.streampipes.storage.couchdb; +import org.apache.streampipes.storage.api.CRUDStorage; import org.apache.streampipes.storage.api.IAdapterStorage; import org.apache.streampipes.storage.api.IAssetDashboardStorage; import org.apache.streampipes.storage.api.ICategoryStorage; @@ -60,7 +61,9 @@ import org.apache.streampipes.storage.couchdb.impl.DataLakeStorageImpl; import org.apache.streampipes.storage.couchdb.impl.DataProcessorStorageImpl; import org.apache.streampipes.storage.couchdb.impl.DataSinkStorageImpl; import org.apache.streampipes.storage.couchdb.impl.DataStreamStorageImpl; +import org.apache.streampipes.storage.couchdb.impl.ExtensionsServiceConfigStorageImpl; import org.apache.streampipes.storage.couchdb.impl.ExtensionsServiceEndpointStorageImpl; +import org.apache.streampipes.storage.couchdb.impl.ExtensionsServiceStorageImpl; import org.apache.streampipes.storage.couchdb.impl.FileMetadataStorageImpl; import org.apache.streampipes.storage.couchdb.impl.GenericStorageImpl; import org.apache.streampipes.storage.couchdb.impl.ImageStorageImpl; @@ -78,6 +81,8 @@ import org.apache.streampipes.storage.couchdb.impl.UserActivationTokenImpl; import org.apache.streampipes.storage.couchdb.impl.UserGroupStorageImpl; import org.apache.streampipes.storage.couchdb.impl.UserStorage; import org.apache.streampipes.storage.couchdb.impl.VisualizationStorageImpl; +import org.apache.streampipes.svcdiscovery.api.model.SpServiceConfiguration; +import org.apache.streampipes.svcdiscovery.api.model.SpServiceRegistrationRequest; public enum CouchDbStorageManager implements INoSqlStorage { @@ -239,5 +244,15 @@ public enum CouchDbStorageManager implements INoSqlStorage { return new UserActivationTokenImpl(); } + @Override + public CRUDStorage<String, SpServiceRegistrationRequest> getExtensionsServiceStorage() { + return new ExtensionsServiceStorageImpl(); + } + + @Override + public CRUDStorage<String, SpServiceConfiguration> getExtensionsServiceConfigurationStorage() { + return new ExtensionsServiceConfigStorageImpl(); + } + } diff --git a/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/ExtensionsServiceConfigStorageImpl.java b/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/ExtensionsServiceConfigStorageImpl.java new file mode 100644 index 000000000..b1f45aaa5 --- /dev/null +++ b/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/ExtensionsServiceConfigStorageImpl.java @@ -0,0 +1,60 @@ +/* + * 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.storage.couchdb.impl; + +import org.apache.streampipes.storage.api.CRUDStorage; +import org.apache.streampipes.storage.couchdb.dao.AbstractDao; +import org.apache.streampipes.storage.couchdb.utils.Utils; +import org.apache.streampipes.svcdiscovery.api.model.SpServiceConfiguration; + +import java.util.List; + +public class ExtensionsServiceConfigStorageImpl extends AbstractDao<SpServiceConfiguration> + implements CRUDStorage<String, SpServiceConfiguration> { + + public ExtensionsServiceConfigStorageImpl() { + super(Utils::getCouchDbServiceConfigStorage, SpServiceConfiguration.class); + } + + @Override + public List<SpServiceConfiguration> getAll() { + return findAll(); + } + + @Override + public void createElement(SpServiceConfiguration element) { + persist(element); + } + + @Override + public SpServiceConfiguration getElementById(String id) { + return findWithNullIfEmpty(id); + } + + @Override + public SpServiceConfiguration updateElement(SpServiceConfiguration element) { + update(element); + return getElementById(element.getServiceName()); + } + + @Override + public void deleteElement(SpServiceConfiguration element) { + delete(element.getServiceName()); + } +} diff --git a/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/ExtensionsServiceStorageImpl.java b/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/ExtensionsServiceStorageImpl.java new file mode 100644 index 000000000..6e9b47797 --- /dev/null +++ b/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/ExtensionsServiceStorageImpl.java @@ -0,0 +1,61 @@ +/* + * 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.storage.couchdb.impl; + +import org.apache.streampipes.storage.api.CRUDStorage; +import org.apache.streampipes.storage.couchdb.dao.AbstractDao; +import org.apache.streampipes.storage.couchdb.utils.Utils; +import org.apache.streampipes.svcdiscovery.api.model.SpServiceRegistrationRequest; + +import java.util.List; + +public class ExtensionsServiceStorageImpl extends AbstractDao<SpServiceRegistrationRequest> + implements CRUDStorage<String, SpServiceRegistrationRequest> { + + public ExtensionsServiceStorageImpl() { + super(Utils::getCouchDbExtensionsStorage, SpServiceRegistrationRequest.class); + } + + + @Override + public List<SpServiceRegistrationRequest> getAll() { + return findAll(); + } + + @Override + public void createElement(SpServiceRegistrationRequest element) { + persist(element); + } + + @Override + public SpServiceRegistrationRequest getElementById(String id) { + return find(id).orElseThrow(IllegalArgumentException::new); + } + + @Override + public SpServiceRegistrationRequest updateElement(SpServiceRegistrationRequest element) { + update(element); + return getElementById(element.getSvcId()); + } + + @Override + public void deleteElement(SpServiceRegistrationRequest element) { + delete(element.getSvcId()); + } +} diff --git a/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/utils/Utils.java b/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/utils/Utils.java index 125270494..30da73daf 100644 --- a/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/utils/Utils.java +++ b/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/utils/Utils.java @@ -64,6 +64,14 @@ public class Utils { return getCouchDbGsonClient("category"); } + public static CouchDbClient getCouchDbServiceConfigStorage() { + return getCouchDbGsonClient("extensions-services-configurations"); + } + + public static CouchDbClient getCouchDbExtensionsStorage() { + return getCouchDbGsonClient("extensions-services"); + } + public static CouchDbClient getCouchDbLabelClient() { return getCouchDbGsonClient("label"); }
