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 aa6eb82e5a5a2988cfd373d8ce6abf366fda52a2 Author: Dominik Riemer <[email protected]> AuthorDate: Sun Aug 6 17:44:32 2023 +0200 Improve service registration and deregistration --- .../management/init/DeclarersSingleton.java | 10 ---- .../svcdiscovery/SpServiceRegistration.java | 23 +++++++-- .../manager/health/ServiceHealthCheck.java | 22 ++++---- .../extensions/ExtensionsModelSubmitter.java | 4 -- .../StreamPipesExtensionsServiceBase.java | 60 ++++++++++++++++------ 5 files changed, 74 insertions(+), 45 deletions(-) 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 8b98511b8..5ebd991ab 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 @@ -101,7 +101,6 @@ public class DeclarersSingleton implements IDeclarersSingleton { public void populate(String host, Integer port, SpServiceDefinition serviceDef) { this.serviceDefinition = serviceDef; - this.registerConfigs(serviceDef.getServiceGroup(), serviceDef.getServiceName(), serviceDef.getKvConfigs()); this.setHostName(host); this.setPort(port); this.addDeclarers(serviceDef.getDeclarers()); @@ -114,15 +113,6 @@ public class DeclarersSingleton implements IDeclarersSingleton { serviceDef.getFunctions().forEach(f -> this.functions.put(f.getFunctionConfig().getFunctionId().getId(), f)); } - private void registerConfigs(String serviceGroup, - String serviceName, - List<ConfigItem> configs) { - LOG.info("Registering {} configs in key/value store", configs.size()); - StreamPipesClient client = new StreamPipesClientResolver().makeStreamPipesClientInstance(); - var serviceConfiguration = new SpServiceConfiguration(serviceGroup, serviceName, configs); - client.adminApi().registerServiceConfiguration(serviceConfiguration); - } - public void addDeclarers(List<IStreamPipesPipelineElement<?>> allPipelineElements) { allPipelineElements.forEach(this::add); } diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/extensions/svcdiscovery/SpServiceRegistration.java b/streampipes-model/src/main/java/org/apache/streampipes/model/extensions/svcdiscovery/SpServiceRegistration.java index 92b713211..2f731ba74 100644 --- a/streampipes-model/src/main/java/org/apache/streampipes/model/extensions/svcdiscovery/SpServiceRegistration.java +++ b/streampipes-model/src/main/java/org/apache/streampipes/model/extensions/svcdiscovery/SpServiceRegistration.java @@ -26,6 +26,7 @@ import java.util.List; @TsModel public class SpServiceRegistration { + private String svcType; private String svcGroup; protected @SerializedName("_rev") String rev; @@ -41,12 +42,14 @@ public class SpServiceRegistration { public SpServiceRegistration() { } - public SpServiceRegistration(String svcGroup, + public SpServiceRegistration(String svcType, + String svcGroup, String svcId, String host, int port, List<SpServiceTag> tags, String healthCheckPath) { + this.svcType = svcType; this.svcGroup = svcGroup; this.svcId = svcId; this.host = host; @@ -55,21 +58,23 @@ public class SpServiceRegistration { this.healthCheckPath = healthCheckPath; } - public static SpServiceRegistration from(String svcGroup, + public static SpServiceRegistration from(String svcType, + String svcGroup, String svcId, String host, Integer port, List<SpServiceTag> tags) { - return new SpServiceRegistration(svcGroup, svcId, host, port, tags, ""); + return new SpServiceRegistration(svcType, svcGroup, svcId, host, port, tags, ""); } - public static SpServiceRegistration from(String svcGroup, + public static SpServiceRegistration from(String svcType, + String svcGroup, String svcId, String host, Integer port, List<SpServiceTag> tags, String healthCheckPath) { - return new SpServiceRegistration(svcGroup, svcId, host, port, tags, healthCheckPath); + return new SpServiceRegistration(svcType, svcGroup, svcId, host, port, tags, healthCheckPath); } public String getSvcGroup() { @@ -151,4 +156,12 @@ public class SpServiceRegistration { public void setFirstTimeSeenUnhealthy(long firstTimeSeenUnhealthy) { this.firstTimeSeenUnhealthy = firstTimeSeenUnhealthy; } + + public String getSvcType() { + return svcType; + } + + public void setSvcType(String svcType) { + this.svcType = svcType; + } } 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 index bd2deda18..768430d5a 100644 --- 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 @@ -53,16 +53,7 @@ public class ServiceHealthCheck implements Runnable { 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); - } + processUnhealthyService(service); } else { if (!service.isHealthy()) { service.setHealthy(true); @@ -70,10 +61,21 @@ public class ServiceHealthCheck implements Runnable { } } } catch (IOException e) { + processUnhealthyService(service); + } + } + + private void processUnhealthyService(SpServiceRegistration service) { + 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); + } } private boolean shouldDeleteService(SpServiceRegistration service) { diff --git a/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/ExtensionsModelSubmitter.java b/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/ExtensionsModelSubmitter.java index 3a0297042..a67e9175e 100644 --- a/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/ExtensionsModelSubmitter.java +++ b/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/ExtensionsModelSubmitter.java @@ -24,8 +24,6 @@ import org.apache.streampipes.service.extensions.connect.ConnectWorkerRegistrati import org.apache.streampipes.service.extensions.function.StreamPipesFunctionHandler; import org.apache.streampipes.service.extensions.security.WebSecurityConfig; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Import; @@ -38,8 +36,6 @@ import java.util.List; @EnableAutoConfiguration @Import({ExtensionsResourceConfig.class, WebSecurityConfig.class}) public abstract class ExtensionsModelSubmitter extends StreamPipesExtensionsServiceBase { - private static final Logger LOG = - LoggerFactory.getLogger(ExtensionsModelSubmitter.class.getCanonicalName()); @PreDestroy public void onExit() { 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 baf41fc23..7ef2dc0cd 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 @@ -19,9 +19,12 @@ package org.apache.streampipes.service.extensions; import org.apache.streampipes.client.StreamPipesClient; +import org.apache.streampipes.commons.exceptions.SpRuntimeException; 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.model.extensions.configuration.ConfigItem; +import org.apache.streampipes.model.extensions.configuration.SpServiceConfiguration; import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceRegistration; import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceTag; import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceTagPrefix; @@ -37,10 +40,12 @@ import jakarta.annotation.PreDestroy; import java.net.UnknownHostException; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.TimeUnit; public abstract class StreamPipesExtensionsServiceBase extends StreamPipesServiceBase { private static final Logger LOG = LoggerFactory.getLogger(StreamPipesExtensionsServiceBase.class); + private static final int RETRY_INTERVAL_SECONDS = 5; public void init() { SpServiceDefinition serviceDef = provideServiceDefinition(); @@ -62,17 +67,26 @@ public abstract class StreamPipesExtensionsServiceBase extends StreamPipesServic } } - public SpServiceDefinition provideServiceDefinition() { - return null; - } + public abstract SpServiceDefinition provideServiceDefinition(); public abstract void afterServiceRegistered(SpServiceDefinition serviceDef); public void startExtensionsService(Class<?> serviceClass, SpServiceDefinition serviceDef, BaseNetworkingConfig networkingConfig) throws UnknownHostException { - this.registerService(DefaultSpServiceGroups.EXT, serviceId(), networkingConfig); + var req = SpServiceRegistration.from( + DefaultSpServiceGroups.EXT, + serviceDef.getServiceGroup(), + serviceId(), + networkingConfig.getHost(), + networkingConfig.getPort(), + getServiceTags(), + getHealthCheckPath()); + LOG.info("Registering service {} with id {} at core", req.getSvcGroup(), req.getSvcId()); + registerService(req); + + this.registerConfigs(serviceDef.getServiceGroup(), serviceDef.getServiceName(), serviceDef.getKvConfigs()); this.startStreamPipesService( serviceClass, DefaultSpServiceGroups.EXT, @@ -83,19 +97,24 @@ public abstract class StreamPipesExtensionsServiceBase extends StreamPipesServic this.afterServiceRegistered(serviceDef); } - private void registerService(String serviceGroup, - String serviceId, - BaseNetworkingConfig networkingConfig) { - SpServiceRegistration req = SpServiceRegistration.from( - serviceGroup, - serviceId, - networkingConfig.getHost(), - networkingConfig.getPort(), - getServiceTags(), - getHealthCheckPath()); - + private void registerService(SpServiceRegistration serviceRegistration) { StreamPipesClient client = new StreamPipesClientResolver().makeStreamPipesClientInstance(); - client.adminApi().registerService(req); + try { + client.adminApi().registerService(serviceRegistration); + LOG.info("Successfully registered service at core."); + } catch (SpRuntimeException e) { + LOG.warn( + "Could not register at core at url {}. Trying again in {} seconds", + client.getConnectionConfig().getBaseUrl(), + 10 + ); + try { + TimeUnit.SECONDS.sleep(RETRY_INTERVAL_SECONDS); + registerService(serviceRegistration); + } catch (InterruptedException ex) { + throw new RuntimeException(ex); + } + } } @Override @@ -115,6 +134,15 @@ public abstract class StreamPipesExtensionsServiceBase extends StreamPipesServic client.adminApi().deregisterService(serviceId); } + private void registerConfigs(String serviceGroup, + String serviceName, + List<ConfigItem> configs) { + LOG.info("Registering {} service configs for service {}", configs.size(), serviceGroup); + StreamPipesClient client = new StreamPipesClientResolver().makeStreamPipesClientInstance(); + var serviceConfiguration = new SpServiceConfiguration(serviceGroup, serviceName, configs); + client.adminApi().registerServiceConfiguration(serviceConfiguration); + } + protected abstract List<SpServiceTag> getExtensionsServiceTags(); @PreDestroy
