This is an automated email from the ASF dual-hosted git repository.
jsinovassin pushed a commit to branch unomi-3.0.x
in repository https://gitbox.apache.org/repos/asf/unomi.git
The following commit(s) were added to refs/heads/unomi-3.0.x by this push:
new 0a62a434f [UNOMI-924][UNOMI-1025] Make healthcheck concurrent, with a
provider timeout (unomi-3.0.x) (#930)
0a62a434f is described below
commit 0a62a434f088cd029ccd96829f67757405411718
Author: Jerome Blanchard <[email protected]>
AuthorDate: Fri Nov 6 15:25:54 2026 +0100
[UNOMI-924][UNOMI-1025] Make healthcheck concurrent, with a provider
timeout (unomi-3.0.x) (#930)
---
.../unomi/healthcheck/HealthCheckService.java | 117 ++++++++++++++-------
.../provider/ElasticSearchHealthCheckProvider.java | 9 +-
.../healthcheck/servlet/HealthCheckServlet.java | 2 -
.../unomi/itests/CopyPropertiesActionIT.java | 10 +-
.../org/apache/unomi/itests/HealthCheckIT.java | 38 +++++++
.../org/apache/unomi/itests/ProfileMergeIT.java | 11 +-
.../org/apache/unomi/itests/SendEventActionIT.java | 10 +-
7 files changed, 152 insertions(+), 45 deletions(-)
diff --git
a/extensions/healthcheck/src/main/java/org/apache/unomi/healthcheck/HealthCheckService.java
b/extensions/healthcheck/src/main/java/org/apache/unomi/healthcheck/HealthCheckService.java
index 9da585e14..7a5107063 100644
---
a/extensions/healthcheck/src/main/java/org/apache/unomi/healthcheck/HealthCheckService.java
+++
b/extensions/healthcheck/src/main/java/org/apache/unomi/healthcheck/HealthCheckService.java
@@ -26,9 +26,16 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import javax.servlet.ServletException;
-import java.util.*;
-import java.util.concurrent.*;
-import java.util.stream.Collectors;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import static org.apache.unomi.healthcheck.HealthCheckConfig.CONFIG_AUTH_REALM;
@@ -41,9 +48,13 @@ public class HealthCheckService {
private static final Logger LOGGER =
LoggerFactory.getLogger(HealthCheckService.class.getName());
- private final List<HealthCheckProvider> providers = new ArrayList<>();
- private ExecutorService executor;
- private boolean busy = false;
+ private final List<HealthCheckProvider> providers = new
CopyOnWriteArrayList<>();
+ private final ExecutorService executor = Executors.newCachedThreadPool();
+ private final Object cacheLock = new Object();
+ private volatile long cacheTimestamp = 0L;
+ private volatile List<HealthCheckResponse> healthCache =
Collections.emptyList();
+ private volatile boolean initialized = false;
+ private volatile boolean busy = false;
private boolean registered = false;
@Reference
@@ -58,14 +69,13 @@ public class HealthCheckService {
@Activate
public void activate() throws ServletException, NamespaceException {
LOGGER.info("Activating healthcheck service...");
- executor = Executors.newSingleThreadExecutor();
if (!registered) {
setConfig(config);
}
}
@Reference(service = HealthCheckConfig.class, policy =
ReferencePolicy.DYNAMIC, updated = "setConfig")
- private void setConfig(HealthCheckConfig config) throws ServletException,
NamespaceException {
+ protected void setConfig(HealthCheckConfig config) throws
ServletException, NamespaceException {
this.config = config;
if (httpService == null ) {
LOGGER.info("Healthcheck config with {} entrie(s) did not update
the service as not fully started yet.", config.getSize());
@@ -98,9 +108,7 @@ public class HealthCheckService {
httpService.unregister("/health/check");
registered = false;
}
- if (executor != null) {
- executor.shutdown();
- }
+ executor.shutdownNow();
}
@Reference(service = HealthCheckProvider.class, cardinality =
ReferenceCardinality.MULTIPLE, policy = ReferencePolicy.DYNAMIC, unbind =
"unbind")
@@ -114,37 +122,72 @@ public class HealthCheckService {
providers.remove(provider);
}
- public List<HealthCheckResponse> check() throws RejectedExecutionException
{
- if (config !=null && config.isEnabled()) {
- LOGGER.debug("Health check called");
- if (busy) {
- throw new RejectedExecutionException("Health check already in
progress");
- } else {
- try {
+ public List<HealthCheckResponse> check() {
+ if (config == null || !config.isEnabled()) {
+ LOGGER.warn("Healthcheck service is disabled");
+ return Collections.emptyList();
+ }
+ if (!initialized) {
+ synchronized (cacheLock) {
+ if (!initialized) {
+ refreshCache();
+ initialized = true;
+ }
+ }
+ } else if (shouldRefreshCache()) {
+ synchronized (cacheLock) {
+ if (isCacheExpired()) {
busy = true;
- List<HealthCheckResponse> health = new ArrayList<>();
- health.add(HealthCheckResponse.live("karaf"));
- for (HealthCheckProvider provider :
providers.stream().filter(p ->
config.getEnabledProviders().contains(p.name())).collect(Collectors.toList())) {
- Future<HealthCheckResponse> future =
executor.submit(provider::execute);
- try {
- HealthCheckResponse response =
future.get(config.getTimeout(), TimeUnit.MILLISECONDS);
- health.add(response);
- } catch (TimeoutException e) {
- future.cancel(true);
- health.add(provider.timeout());
- } catch (Exception e) {
- LOGGER.error("Error while executing health check",
e);
- }
+ try {
+ refreshCache();
+ } finally {
+ busy = false;
}
- return health;
- } finally {
- busy = false;
}
}
- } else {
- LOGGER.info("Healthcheck service is disabled");
- return Collections.emptyList();
}
+ return healthCache;
}
+ private boolean shouldRefreshCache() {
+ return !busy && isCacheExpired();
+ }
+
+ private boolean isCacheExpired() {
+ return (System.currentTimeMillis() - cacheTimestamp) > 1000;
+ }
+
+ private void refreshCache() {
+ try {
+ List<HealthCheckResponse> health = new ArrayList<>();
+ health.add(HealthCheckResponse.live("karaf"));
+ List<HealthCheckProvider> enabledProviders = providers.stream()
+ .filter(p ->
config.getEnabledProviders().contains(p.name()))
+ .toList();
+ List<Future<HealthCheckResponse>> futures = new ArrayList<>();
+ for (HealthCheckProvider provider : enabledProviders) {
+ futures.add(executor.submit(provider::execute));
+ }
+ long deadline = System.nanoTime() +
TimeUnit.MILLISECONDS.toNanos(config.getTimeout());
+ for (int i = 0; i < enabledProviders.size(); i++) {
+ HealthCheckProvider provider = enabledProviders.get(i);
+ Future<HealthCheckResponse> future = futures.get(i);
+ try {
+ health.add(future.get(Math.max(0, deadline -
System.nanoTime()), TimeUnit.NANOSECONDS));
+ } catch (TimeoutException e) {
+ future.cancel(true);
+ LOGGER.warn("Health check {} timed out after {} ms",
provider.name(), config.getTimeout());
+ health.add(provider.timeout());
+ } catch (Exception e) {
+ LOGGER.error("Error while executing health check", e);
+ health.add(provider.timeout());
+ }
+ }
+ health.sort(Comparator.comparing(HealthCheckResponse::getName));
+ healthCache = List.copyOf(health);
+ cacheTimestamp = System.currentTimeMillis();
+ } catch (Exception e) {
+ LOGGER.error("Error refreshing health cache", e);
+ }
+ }
}
diff --git
a/extensions/healthcheck/src/main/java/org/apache/unomi/healthcheck/provider/ElasticSearchHealthCheckProvider.java
b/extensions/healthcheck/src/main/java/org/apache/unomi/healthcheck/provider/ElasticSearchHealthCheckProvider.java
index 96a17db1a..a95f40ddf 100644
---
a/extensions/healthcheck/src/main/java/org/apache/unomi/healthcheck/provider/ElasticSearchHealthCheckProvider.java
+++
b/extensions/healthcheck/src/main/java/org/apache/unomi/healthcheck/provider/ElasticSearchHealthCheckProvider.java
@@ -22,6 +22,7 @@ import org.apache.http.HttpEntity;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
+import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.BasicCredentialsProvider;
@@ -107,7 +108,13 @@ public class ElasticSearchHealthCheckProvider implements
HealthCheckProvider {
.concat("/_cluster/health");
CloseableHttpResponse response = null;
try {
- response = httpClient.execute(new HttpGet(url));
+ HttpGet request = new HttpGet(url);
+ request.setConfig(RequestConfig.custom()
+ .setConnectTimeout(config.getTimeout())
+ .setConnectionRequestTimeout(config.getTimeout())
+ .setSocketTimeout(config.getTimeout())
+ .build());
+ response = httpClient.execute(request);
if (response != null && response.getStatusLine().getStatusCode()
== 200) {
builder.up();
HttpEntity entity = response.getEntity();
diff --git
a/extensions/healthcheck/src/main/java/org/apache/unomi/healthcheck/servlet/HealthCheckServlet.java
b/extensions/healthcheck/src/main/java/org/apache/unomi/healthcheck/servlet/HealthCheckServlet.java
index 0156d7c0d..65dbdde8d 100644
---
a/extensions/healthcheck/src/main/java/org/apache/unomi/healthcheck/servlet/HealthCheckServlet.java
+++
b/extensions/healthcheck/src/main/java/org/apache/unomi/healthcheck/servlet/HealthCheckServlet.java
@@ -32,7 +32,6 @@ import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
-import java.util.Comparator;
import java.util.List;
/**
@@ -66,7 +65,6 @@ public class HealthCheckServlet extends HttpServlet {
return;
}
List<HealthCheckResponse> checks = service.check();
- checks.sort(Comparator.comparing(HealthCheckResponse::getName));
response.getWriter().println(mapper.writeValueAsString(checks));
response.setContentType("application/json");
response.setHeader("Cache-Control", "no-cache");
diff --git
a/itests/src/test/java/org/apache/unomi/itests/CopyPropertiesActionIT.java
b/itests/src/test/java/org/apache/unomi/itests/CopyPropertiesActionIT.java
index 6acde9fe2..7d01aaf57 100644
--- a/itests/src/test/java/org/apache/unomi/itests/CopyPropertiesActionIT.java
+++ b/itests/src/test/java/org/apache/unomi/itests/CopyPropertiesActionIT.java
@@ -22,6 +22,7 @@ import org.apache.unomi.api.Metadata;
import org.apache.unomi.api.Profile;
import org.apache.unomi.api.PropertyType;
import org.apache.unomi.api.rules.Rule;
+import org.apache.unomi.api.services.EventService;
import org.apache.unomi.persistence.spi.CustomObjectMapper;
import org.junit.After;
import org.junit.Assert;
@@ -166,10 +167,13 @@ public class CopyPropertiesActionIT extends BaseIT {
Event event = new Event("copyProperties", null, profile, null, null,
profile, new Date());
event.setPersistent(false);
-
event.setProperty("urlParameters", properties);
-
- eventService.send(event);
+ int result = eventService.send(event);
+ LOGGER.info("Event processing result: {}", result);
+ if (result == EventService.ERROR) {
+ LOGGER.error("Event processing resulted in ERROR. Event details:
{}", event);
+ }
+ Assert.assertNotEquals(EventService.ERROR, result);
return event;
}
diff --git a/itests/src/test/java/org/apache/unomi/itests/HealthCheckIT.java
b/itests/src/test/java/org/apache/unomi/itests/HealthCheckIT.java
index 2ef00efa8..3233e61e9 100644
--- a/itests/src/test/java/org/apache/unomi/itests/HealthCheckIT.java
+++ b/itests/src/test/java/org/apache/unomi/itests/HealthCheckIT.java
@@ -84,6 +84,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.*;
import java.util.stream.Stream;
import static org.junit.Assert.fail;
@@ -120,6 +121,43 @@ public class HealthCheckIT extends BaseIT {
}
}
+ @Test
+ public void testConcurrentHealthCheck() {
+ final int NB_THREADS = 10;
+ final int NB_ITERATIONS = 20;
+
+ ExecutorService executorService = null;
+ try {
+ executorService = Executors.newFixedThreadPool(NB_THREADS);
+ List<Future<List<HealthCheckResponse>>> futures = new
ArrayList<>();
+ for (int i = 0; i < NB_ITERATIONS; i++) {
+ for (int j = 0; j < NB_THREADS; j++) {
+ Future<List<HealthCheckResponse>> future =
executorService.submit(() -> get(HEALTHCHECK_ENDPOINT, new TypeReference<>()
{}));
+ futures.add(future);
+ }
+ for (Future<List<HealthCheckResponse>> future : futures) {
+ List<HealthCheckResponse> health = future.get(10,
TimeUnit.SECONDS);
+ Assert.assertEquals(5, health.size());
+ Assert.assertTrue(health.stream().anyMatch(r ->
r.getName().equals("karaf") && r.getStatus() ==
HealthCheckResponse.Status.LIVE));
+ Assert.assertTrue(health.stream().anyMatch(r ->
r.getName().equals("elasticsearch") && r.getStatus() ==
HealthCheckResponse.Status.LIVE));
+ Assert.assertTrue(health.stream().anyMatch(r ->
r.getName().equals("unomi") && r.getStatus() ==
HealthCheckResponse.Status.LIVE));
+ Assert.assertTrue(health.stream().anyMatch(r ->
r.getName().equals("cluster") && r.getStatus() ==
HealthCheckResponse.Status.LIVE));
+ Assert.assertTrue(health.stream().anyMatch(r ->
r.getName().equals("persistence") && r.getStatus() ==
HealthCheckResponse.Status.LIVE));
+ }
+ Thread.sleep(10);
+ }
+ executorService.shutdown();
+ Assert.assertTrue(executorService.awaitTermination(10,
TimeUnit.SECONDS));
+ } catch (Exception e) {
+ LOGGER.error("Error while executing concurrent health check", e);
+ fail("Error while executing concurrent health check: " +
e.getMessage());
+ } finally {
+ if ( executorService != null ) {
+ executorService.shutdownNow();
+ }
+ }
+ }
+
protected <T> T get(final String url, TypeReference<T> typeReference) {
CloseableHttpResponse response = null;
try {
diff --git a/itests/src/test/java/org/apache/unomi/itests/ProfileMergeIT.java
b/itests/src/test/java/org/apache/unomi/itests/ProfileMergeIT.java
index 43c0b0429..d243f03e7 100644
--- a/itests/src/test/java/org/apache/unomi/itests/ProfileMergeIT.java
+++ b/itests/src/test/java/org/apache/unomi/itests/ProfileMergeIT.java
@@ -20,6 +20,7 @@ import org.apache.unomi.api.*;
import org.apache.unomi.api.actions.Action;
import org.apache.unomi.api.conditions.Condition;
import org.apache.unomi.api.rules.Rule;
+import org.apache.unomi.api.services.EventService;
import org.junit.After;
import org.junit.Assert;
import org.junit.Test;
@@ -27,6 +28,8 @@ import org.junit.runner.RunWith;
import org.ops4j.pax.exam.junit.PaxExam;
import org.ops4j.pax.exam.spi.reactors.ExamReactorStrategy;
import org.ops4j.pax.exam.spi.reactors.PerSuite;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.util.*;
@@ -36,6 +39,8 @@ import java.util.*;
@RunWith(PaxExam.class)
@ExamReactorStrategy(PerSuite.class)
public class ProfileMergeIT extends BaseIT {
+ private final static Logger LOGGER =
LoggerFactory.getLogger(ProfileMergeIT.class);
+
public static final String PERSONALIZATION_STRATEGY_STATUS =
"personalizationStrategyStatus";
public static final String PERSONALIZATION_STRATEGY_STATUS_ID =
"personalizationId";
public static final String PERSONALIZATION_STRATEGY_STATUS_IN_CTRL_GROUP =
"inControlGroup";
@@ -466,7 +471,11 @@ public class ProfileMergeIT extends BaseIT {
profile.setProperty("j:nodename", "michel");
profile.getSystemProperties().put("mergeIdentifier", "jose");
Event testEvent = new Event(TEST_EVENT_TYPE, null, profile, null,
null, profile, new Date());
- eventService.send(testEvent);
+ int result = eventService.send(testEvent);
+ LOGGER.info("Event processing result: {}", result);
+ if (result == EventService.ERROR) {
+ LOGGER.error("Event processing resulted in ERROR. Event details:
{}", testEvent);
+ }
return testEvent;
}
diff --git
a/itests/src/test/java/org/apache/unomi/itests/SendEventActionIT.java
b/itests/src/test/java/org/apache/unomi/itests/SendEventActionIT.java
index 821158f19..3a2086f1b 100644
--- a/itests/src/test/java/org/apache/unomi/itests/SendEventActionIT.java
+++ b/itests/src/test/java/org/apache/unomi/itests/SendEventActionIT.java
@@ -23,6 +23,7 @@ import org.apache.unomi.api.Profile;
import org.apache.unomi.api.actions.Action;
import org.apache.unomi.api.conditions.Condition;
import org.apache.unomi.api.rules.Rule;
+import org.apache.unomi.api.services.EventService;
import org.junit.After;
import org.junit.Assert;
import org.junit.Test;
@@ -30,6 +31,8 @@ import org.junit.runner.RunWith;
import org.ops4j.pax.exam.junit.PaxExam;
import org.ops4j.pax.exam.spi.reactors.ExamReactorStrategy;
import org.ops4j.pax.exam.spi.reactors.PerSuite;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.util.Collections;
import java.util.Date;
@@ -38,6 +41,7 @@ import java.util.HashMap;
@RunWith(PaxExam.class)
@ExamReactorStrategy(PerSuite.class)
public class SendEventActionIT extends BaseIT {
+ private final static Logger LOGGER =
LoggerFactory.getLogger(SendEventActionIT.class);
private final static String TEST_RULE_ID = "sendEventTest";
private final static String EVENT_ID = "sendEventTestId";
@@ -80,7 +84,11 @@ public class SendEventActionIT extends BaseIT {
profile.setProperty("j:nodename", "michel");
Event testEvent = new Event(TEST_EVENT_TYPE, null, profile, null,
null, profile, new Date());
testEvent.setItemId(EVENT_ID);
- eventService.send(testEvent);
+ int result = eventService.send(testEvent);
+ LOGGER.info("Event processing result: {}", result);
+ if (result == EventService.ERROR) {
+ LOGGER.error("Event processing resulted in ERROR. Event details:
{}", testEvent);
+ }
return testEvent;
}
