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 92ebe8ac8348f3fd819e19781e9f6bcc2f9b7409 Author: Dominik Riemer <[email protected]> AuthorDate: Fri Aug 4 11:33:29 2023 +0200 Add service health check --- .../apache/streampipes/client/api/IAdminApi.java | 4 +- .../apache/streampipes/client/api/AdminApi.java | 4 +- .../svcdiscovery/ConsulSpServiceDiscoveryTest.java | 8 +- .../manager/health/ServiceHealthCheck.java | 95 ++++++++++++++++++++++ .../streampipes/rest/impl/PipelineResource.java | 1 + .../impl/admin/ServiceRegistrationResource.java | 6 +- .../service/core/StreamPipesCoreApplication.java | 13 +++ .../svcdiscovery/api/ISpServiceDiscovery.java | 4 +- ...tionRequest.java => SpServiceRegistration.java} | 60 +++++++++----- .../consul/SpConsulServiceDiscovery.java | 8 +- .../svcdiscovery/SpServiceDiscoveryCore.java | 40 +++------ .../StreamPipesExtensionsServiceBase.java | 4 +- .../streampipes/storage/api/INoSqlStorage.java | 4 +- .../storage/couchdb/CouchDbStorageManager.java | 4 +- .../couchdb/impl/ExtensionsServiceStorageImpl.java | 18 ++-- 15 files changed, 192 insertions(+), 81 deletions(-) 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 9af296e7a..59b6f8e9a 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 @@ -22,13 +22,13 @@ import org.apache.streampipes.model.configuration.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 org.apache.streampipes.svcdiscovery.api.model.SpServiceRegistration; import java.util.List; public interface IAdminApi { - void registerService(SpServiceRegistrationRequest serviceRegistration); + void registerService(SpServiceRegistration serviceRegistration); void deregisterService(String serviceId); 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 a94b70268..2948b7e60 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 @@ -24,7 +24,7 @@ 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 org.apache.streampipes.svcdiscovery.api.model.SpServiceRegistration; import java.util.List; @@ -35,7 +35,7 @@ public class AdminApi extends AbstractClientApi implements IAdminApi { } @Override - public void registerService(SpServiceRegistrationRequest serviceRegistration) { + public void registerService(SpServiceRegistration serviceRegistration) { post(getExtensionsServiceRegistrationPath(), serviceRegistration); } 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 index 0a4093c30..6f78f3874 100644 --- 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 @@ -22,7 +22,7 @@ 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.apache.streampipes.svcdiscovery.api.model.SpServiceRegistrationRequest; +import org.apache.streampipes.svcdiscovery.api.model.SpServiceRegistration; import org.apache.streampipes.svcdiscovery.api.model.SpServiceTag; import org.apache.streampipes.svcdiscovery.api.model.SpServiceTagPrefix; @@ -81,9 +81,9 @@ public class ConsulSpServiceDiscoveryTest extends AbstractConsulTest { List.of(makeServiceTag().get(0).asString())); } - private SpServiceRegistrationRequest makeRequest(List<SpServiceTag> serviceTags, - String serviceId) { - var req = new SpServiceRegistrationRequest( + private SpServiceRegistration makeRequest(List<SpServiceTag> serviceTags, + String serviceId) { + var req = new SpServiceRegistration( DefaultSpServiceGroups.EXT, serviceId, "localhost", diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/health/ServiceHealthCheck.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/health/ServiceHealthCheck.java new file mode 100644 index 000000000..719024afa --- /dev/null +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/health/ServiceHealthCheck.java @@ -0,0 +1,95 @@ +/* + * 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.manager.health; + +import org.apache.streampipes.storage.api.CRUDStorage; +import org.apache.streampipes.storage.management.StorageDispatcher; +import org.apache.streampipes.svcdiscovery.api.model.SpServiceRegistration; + +import org.apache.http.client.fluent.Request; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.List; + +public class ServiceHealthCheck implements Runnable { + + private static final Logger LOG = LoggerFactory.getLogger(ServiceHealthCheck.class); + + private static final int MAX_UNHEALTHY_DURATION_BEFORE_REMOVAL_MS = 60000; + + private final CRUDStorage<String, SpServiceRegistration> storage; + + public ServiceHealthCheck() { + this.storage = StorageDispatcher.INSTANCE.getNoSqlStore().getExtensionsServiceStorage(); + } + + @Override + public void run() { + var registeredServices = getRegisteredServices(); + registeredServices.forEach(this::checkServiceHealth); + } + + private void checkServiceHealth(SpServiceRegistration service) { + String healthCheckUrl = makeHealthCheckUrl(service); + + try { + var response = Request.Get(healthCheckUrl).execute(); + if (response.returnResponse().getStatusLine().getStatusCode() != 200) { + if (service.isHealthy()) { + service.setHealthy(false); + service.setFirstTimeSeenUnhealthy(System.currentTimeMillis()); + updateService(service); + } + if (shouldDeleteService(service)) { + LOG.info("Removing service {} which has been unhealthy for more than {} seconds.", + service.getSvcId(), MAX_UNHEALTHY_DURATION_BEFORE_REMOVAL_MS / 1000); + storage.deleteElement(service); + } + } else { + if (!service.isHealthy()) { + service.setHealthy(true); + updateService(service); + } + } + } catch (IOException e) { + service.setHealthy(false); + service.setFirstTimeSeenUnhealthy(System.currentTimeMillis()); + updateService(service); + } + } + + private boolean shouldDeleteService(SpServiceRegistration service) { + var currentTimeMillis = System.currentTimeMillis(); + return (currentTimeMillis - service.getFirstTimeSeenUnhealthy() > MAX_UNHEALTHY_DURATION_BEFORE_REMOVAL_MS); + } + + private void updateService(SpServiceRegistration service) { + storage.updateElement(service); + } + + private String makeHealthCheckUrl(SpServiceRegistration service) { + return service.getScheme() + "://" + service.getHost() + ":" + service.getPort() + service.getHealthCheckPath(); + } + + private List<SpServiceRegistration> getRegisteredServices() { + return storage.getAll(); + } +} diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineResource.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineResource.java index ac331df2a..b42f9e02d 100644 --- a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineResource.java +++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineResource.java @@ -235,6 +235,7 @@ public class PipelineResource extends AbstractAuthGuardedRestResource { } storedPipeline.setCreatedAt(System.currentTimeMillis()); storedPipeline.setPipelineCategories(pipeline.getPipelineCategories()); + storedPipeline.setHealthStatus(pipeline.getHealthStatus()); storedPipeline.setPipelineNotifications(pipeline.getPipelineNotifications()); Operations.updatePipeline(storedPipeline); SuccessMessage message = Notifications.success("Pipeline modified"); 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 index 63928cece..bfa4ccdb4 100644 --- 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 @@ -21,7 +21,7 @@ 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.apache.streampipes.svcdiscovery.api.model.SpServiceRegistration; import org.springframework.security.access.prepost.PreAuthorize; import org.springframework.stereotype.Component; @@ -38,12 +38,12 @@ import jakarta.ws.rs.core.Response; @PreAuthorize(AuthConstants.IS_ADMIN_ROLE) public class ServiceRegistrationResource extends AbstractAuthGuardedRestResource { - private final CRUDStorage<String, SpServiceRegistrationRequest> extensionsServiceStorage = + private final CRUDStorage<String, SpServiceRegistration> extensionsServiceStorage = getNoSqlStorage().getExtensionsServiceStorage(); @POST @Consumes(MediaType.APPLICATION_JSON) - public Response registerService(SpServiceRegistrationRequest serviceRegistration) { + public Response registerService(SpServiceRegistration serviceRegistration) { extensionsServiceStorage.createElement(serviceRegistration); return ok(); } 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 7ad6db969..ca7e60c2b 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 @@ -19,6 +19,7 @@ package org.apache.streampipes.service.core; import org.apache.streampipes.config.backend.BackendConfig; import org.apache.streampipes.manager.health.PipelineHealthCheck; +import org.apache.streampipes.manager.health.ServiceHealthCheck; import org.apache.streampipes.manager.monitoring.pipeline.ExtensionsServiceLogExecutor; import org.apache.streampipes.manager.operations.Operations; import org.apache.streampipes.manager.setup.AutoInstallation; @@ -82,8 +83,12 @@ public class StreamPipesCoreApplication extends StreamPipesServiceBase { private static final int HEALTH_CHECK_INTERVAL = 60; private static final TimeUnit HEALTH_CHECK_UNIT = TimeUnit.SECONDS; + private static final int SERVICE_HEALTH_CHECK_INTERVAL = 60; + private static final TimeUnit SERVICE_HEALTH_CHECK_UNIT = TimeUnit.SECONDS; + private ScheduledExecutorService executorService; private ScheduledExecutorService healthCheckExecutorService; + private ScheduledExecutorService serviceHealthCheckExecutorService; private ScheduledExecutorService logCheckExecutorService; private final Map<String, Integer> failedPipelines = new HashMap<>(); @@ -127,6 +132,7 @@ public class StreamPipesCoreApplication extends StreamPipesServiceBase { this.executorService = Executors.newSingleThreadScheduledExecutor(); this.healthCheckExecutorService = Executors.newSingleThreadScheduledExecutor(); this.logCheckExecutorService = Executors.newSingleThreadScheduledExecutor(); + this.serviceHealthCheckExecutorService = Executors.newSingleThreadScheduledExecutor(); new StreamPipesEnvChecker().updateEnvironmentVariables(); new CouchDbViewGenerator().createGenericDatabaseIfNotExists(); @@ -138,6 +144,13 @@ public class StreamPipesCoreApplication extends StreamPipesServiceBase { new MigrationsHandler().performMigrations(); executorService.schedule(this::startAllPreviouslyStoppedPipelines, 5, TimeUnit.SECONDS); + + LOG.info("Service health check will run every {} seconds", SERVICE_HEALTH_CHECK_INTERVAL); + serviceHealthCheckExecutorService.scheduleAtFixedRate(new ServiceHealthCheck(), + SERVICE_HEALTH_CHECK_INTERVAL, + SERVICE_HEALTH_CHECK_INTERVAL, + SERVICE_HEALTH_CHECK_UNIT); + LOG.info("Pipeline health check will run every {} seconds", HEALTH_CHECK_INTERVAL); healthCheckExecutorService.scheduleAtFixedRate(new PipelineHealthCheck(), HEALTH_CHECK_INTERVAL, 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 729739229..8a517945d 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,7 +17,7 @@ */ package org.apache.streampipes.svcdiscovery.api; -import org.apache.streampipes.svcdiscovery.api.model.SpServiceRegistrationRequest; +import org.apache.streampipes.svcdiscovery.api.model.SpServiceRegistration; import java.util.List; import java.util.Map; @@ -29,7 +29,7 @@ public interface ISpServiceDiscovery { * * @param serviceRegistrationRequest the service registration request */ - void registerService(SpServiceRegistrationRequest serviceRegistrationRequest); + void registerService(SpServiceRegistration serviceRegistrationRequest); /** * Get active pipeline element service endpoints 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/SpServiceRegistration.java similarity index 60% rename from streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/SpServiceRegistrationRequest.java rename to streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/SpServiceRegistration.java index 8c1e42bb2..930c46367 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/SpServiceRegistration.java @@ -21,27 +21,29 @@ import com.google.gson.annotations.SerializedName; import java.util.List; -public class SpServiceRegistrationRequest { +public class SpServiceRegistration { private String svcGroup; protected @SerializedName("_rev") String rev; private @SerializedName("_id") String svcId; + private String scheme = "http"; private String host; private int port; private List<SpServiceTag> tags; private String healthCheckPath; private boolean healthy = true; + private long firstTimeSeenUnhealthy = 0; - public SpServiceRegistrationRequest() { + public SpServiceRegistration() { } - public SpServiceRegistrationRequest(String svcGroup, - String svcId, - String host, - int port, - List<SpServiceTag> tags, - String healthCheckPath) { + public SpServiceRegistration(String svcGroup, + String svcId, + String host, + int port, + List<SpServiceTag> tags, + String healthCheckPath) { this.svcGroup = svcGroup; this.svcId = svcId; this.host = host; @@ -50,21 +52,21 @@ public class SpServiceRegistrationRequest { this.healthCheckPath = healthCheckPath; } - public static SpServiceRegistrationRequest from(String svcGroup, - String svcId, - String host, - Integer port, - List<SpServiceTag> tags) { - return new SpServiceRegistrationRequest(svcGroup, svcId, host, port, tags, ""); + public static SpServiceRegistration from(String svcGroup, + String svcId, + String host, + Integer port, + List<SpServiceTag> tags) { + return new SpServiceRegistration(svcGroup, svcId, host, port, tags, ""); } - public static SpServiceRegistrationRequest from(String svcGroup, - String svcId, - String host, - Integer port, - List<SpServiceTag> tags, - String healthCheckPath) { - return new SpServiceRegistrationRequest(svcGroup, svcId, host, port, tags, healthCheckPath); + public static SpServiceRegistration from(String svcGroup, + String svcId, + String host, + Integer port, + List<SpServiceTag> tags, + String healthCheckPath) { + return new SpServiceRegistration(svcGroup, svcId, host, port, tags, healthCheckPath); } public String getSvcGroup() { @@ -130,4 +132,20 @@ public class SpServiceRegistrationRequest { public void setHealthy(boolean healthy) { this.healthy = healthy; } + + public String getScheme() { + return scheme; + } + + public void setScheme(String scheme) { + this.scheme = scheme; + } + + public long getFirstTimeSeenUnhealthy() { + return firstTimeSeenUnhealthy; + } + + public void setFirstTimeSeenUnhealthy(long firstTimeSeenUnhealthy) { + this.firstTimeSeenUnhealthy = firstTimeSeenUnhealthy; + } } diff --git a/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/SpConsulServiceDiscovery.java b/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/SpConsulServiceDiscovery.java index a00acf575..704f80f70 100644 --- a/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/SpConsulServiceDiscovery.java +++ b/streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/SpConsulServiceDiscovery.java @@ -21,7 +21,7 @@ import org.apache.streampipes.commons.environment.Environment; 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.SpServiceRegistration; import org.apache.streampipes.svcdiscovery.api.model.SpServiceTag; import org.apache.streampipes.svcdiscovery.api.model.SpServiceTagPrefix; @@ -55,7 +55,7 @@ public class SpConsulServiceDiscovery extends AbstractConsulService implements I } @Override - public void registerService(SpServiceRegistrationRequest req) { + public void registerService(SpServiceRegistration req) { consulInstance().agentServiceRegister(createRegistrationBody(req)); LOG.info("Successfully registered service at Consul: " + req.getSvcId()); } @@ -129,7 +129,7 @@ public class SpConsulServiceDiscovery extends AbstractConsulService implements I consul.agentServiceDeregister(svcId); } - private NewService createRegistrationBody(SpServiceRegistrationRequest req) { + private NewService createRegistrationBody(SpServiceRegistration req) { var service = new NewService(); service.setId(req.getSvcId()); service.setName(req.getSvcGroup()); @@ -143,7 +143,7 @@ public class SpConsulServiceDiscovery extends AbstractConsulService implements I return service; } - private NewService.Check createServiceCheck(SpServiceRegistrationRequest req) { + private NewService.Check createServiceCheck(SpServiceRegistration req) { var serviceCheck = new NewService.Check(); serviceCheck.setHttp(HTTP_PROTOCOL + req.getHost() + COLON + req.getPort() + req.getHealthCheckPath()); 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 57ae2f054..c05f9a412 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 @@ -23,9 +23,7 @@ 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.apache.streampipes.svcdiscovery.api.model.SpServiceRegistration; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -42,14 +40,14 @@ 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; + private final CRUDStorage<String, SpServiceRegistration> serviceStorage; public SpServiceDiscoveryCore() { this.serviceStorage = StorageDispatcher.INSTANCE.getNoSqlStore().getExtensionsServiceStorage(); } @Override - public void registerService(SpServiceRegistrationRequest serviceRegistrationRequest) { + public void registerService(SpServiceRegistration serviceRegistrationRequest) { // not needed } @@ -71,7 +69,7 @@ public class SpServiceDiscoveryCore implements ISpServiceDiscovery { public List<String> getServiceEndpoints(String serviceGroup, boolean restrictToHealthy, List<String> filterByTags) { - List<SpServiceRegistrationRequest> activeServices = findService(serviceGroup, 0); + List<SpServiceRegistration> activeServices = findService(0); return activeServices .stream() @@ -92,39 +90,25 @@ public class SpServiceDiscoveryCore implements ISpServiceDiscovery { // not needed } - private List<String> asString(List<SpServiceTag> tags) { - return tags.stream().map(SpServiceTag::asString).collect(Collectors.toList()); + private String makeServiceUrl(SpServiceRegistration service) { + return service.getScheme() + "://" + service.getHost() + ":" + service.getPort(); } - 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, + private boolean allFiltersSupported(SpServiceRegistration service, List<String> filterByTags) { - return new HashSet<>(service.getTags()).containsAll(filterByTags); + return new HashSet<>(service.getTags()) + .stream() + .anyMatch(tag -> filterByTags.contains(tag.asString())); } - private List<SpServiceRegistrationRequest> findService(String serviceGroup, - int retryCount) { + private List<SpServiceRegistration> findService(int retryCount) { var services = serviceStorage.getAll(); if (services.size() == 0) { if (retryCount < MAX_RETRIES) { try { retryCount++; TimeUnit.SECONDS.sleep(10); - return findService(serviceGroup, retryCount); + return findService(retryCount); } catch (InterruptedException e) { e.printStackTrace(); return Collections.emptyList(); 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 b9b1196a3..e4df8d4a4 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 @@ -25,7 +25,7 @@ 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.SpServiceRegistration; import org.apache.streampipes.svcdiscovery.api.model.SpServiceTag; import org.apache.streampipes.svcdiscovery.api.model.SpServiceTagPrefix; @@ -86,7 +86,7 @@ public abstract class StreamPipesExtensionsServiceBase extends StreamPipesServic private void registerService(String serviceGroup, String serviceId, BaseNetworkingConfig networkingConfig) { - SpServiceRegistrationRequest req = SpServiceRegistrationRequest.from( + SpServiceRegistration req = SpServiceRegistration.from( serviceGroup, serviceId, networkingConfig.getHost(), 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 9e26868c3..b1e11a2a3 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 @@ -18,7 +18,7 @@ package org.apache.streampipes.storage.api; import org.apache.streampipes.svcdiscovery.api.model.SpServiceConfiguration; -import org.apache.streampipes.svcdiscovery.api.model.SpServiceRegistrationRequest; +import org.apache.streampipes.svcdiscovery.api.model.SpServiceRegistration; public interface INoSqlStorage { @@ -84,7 +84,7 @@ public interface INoSqlStorage { IUserActivationTokenStorage getUserActivationTokenStorage(); - CRUDStorage<String, SpServiceRegistrationRequest> getExtensionsServiceStorage(); + CRUDStorage<String, SpServiceRegistration> getExtensionsServiceStorage(); CRUDStorage<String, SpServiceConfiguration> getExtensionsServiceConfigurationStorage(); 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 e3a64167b..1aadff6c2 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 @@ -84,7 +84,7 @@ 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; +import org.apache.streampipes.svcdiscovery.api.model.SpServiceRegistration; public enum CouchDbStorageManager implements INoSqlStorage { @@ -247,7 +247,7 @@ public enum CouchDbStorageManager implements INoSqlStorage { } @Override - public CRUDStorage<String, SpServiceRegistrationRequest> getExtensionsServiceStorage() { + public CRUDStorage<String, SpServiceRegistration> getExtensionsServiceStorage() { return new ExtensionsServiceStorageImpl(); } 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 index 6e9b47797..d25f6f460 100644 --- 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 @@ -21,41 +21,41 @@ 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 org.apache.streampipes.svcdiscovery.api.model.SpServiceRegistration; import java.util.List; -public class ExtensionsServiceStorageImpl extends AbstractDao<SpServiceRegistrationRequest> - implements CRUDStorage<String, SpServiceRegistrationRequest> { +public class ExtensionsServiceStorageImpl extends AbstractDao<SpServiceRegistration> + implements CRUDStorage<String, SpServiceRegistration> { public ExtensionsServiceStorageImpl() { - super(Utils::getCouchDbExtensionsStorage, SpServiceRegistrationRequest.class); + super(Utils::getCouchDbExtensionsStorage, SpServiceRegistration.class); } @Override - public List<SpServiceRegistrationRequest> getAll() { + public List<SpServiceRegistration> getAll() { return findAll(); } @Override - public void createElement(SpServiceRegistrationRequest element) { + public void createElement(SpServiceRegistration element) { persist(element); } @Override - public SpServiceRegistrationRequest getElementById(String id) { + public SpServiceRegistration getElementById(String id) { return find(id).orElseThrow(IllegalArgumentException::new); } @Override - public SpServiceRegistrationRequest updateElement(SpServiceRegistrationRequest element) { + public SpServiceRegistration updateElement(SpServiceRegistration element) { update(element); return getElementById(element.getSvcId()); } @Override - public void deleteElement(SpServiceRegistrationRequest element) { + public void deleteElement(SpServiceRegistration element) { delete(element.getSvcId()); } }
