You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampipes.apache.org by ri...@apache.org on 2023/08/06 16:19:46 UTC
[streampipes] 10/11: Improve service registration and deregistration
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 <do...@gmail.com>
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