This is an automated email from the ASF dual-hosted git repository.
sergehuber pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/unomi.git
The following commit(s) were added to refs/heads/master by this push:
new 729ffe8f6 [UNOMI-1025] Restore the healthcheck provider timeout (#931)
729ffe8f6 is described below
commit 729ffe8f6dfd3e601aa6a90fa3f48393da560bfe
Author: Jerome Blanchard <[email protected]>
AuthorDate: Thu Oct 29 11:20:58 2026 +0100
[UNOMI-1025] Restore the healthcheck provider timeout (#931)
Co-authored-by: Serge Huber <[email protected]>
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../unomi/healthcheck/HealthCheckService.java | 49 +++++--
.../provider/ElasticSearchHealthCheckProvider.java | 9 +-
.../provider/OpenSearchHealthCheckProvider.java | 9 +-
.../unomi/healthcheck/HealthCheckServiceTest.java | 157 +++++++++++++++++++++
4 files changed, 213 insertions(+), 11 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 5a6f181ba..9d13c4052 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
@@ -30,6 +30,13 @@ 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 java.util.concurrent.atomic.AtomicInteger;
import static org.apache.unomi.healthcheck.HealthCheckConfig.CONFIG_AUTH_REALM;
@@ -42,7 +49,10 @@ public class HealthCheckService {
private static final Logger LOGGER =
LoggerFactory.getLogger(HealthCheckService.class.getName());
- private final List<HealthCheckProvider> providers = new ArrayList<>();
+ private final List<HealthCheckProvider> providers = new
CopyOnWriteArrayList<>();
+ private final AtomicInteger threadCount = new AtomicInteger();
+ private final ExecutorService executor = Executors.newCachedThreadPool(
+ r -> new Thread(r, "unomi-healthcheck-" +
threadCount.incrementAndGet()));
private final Object cacheLock = new Object();
private volatile long cacheTimestamp = 0L;
private volatile List<HealthCheckResponse> healthCache =
Collections.emptyList();
@@ -68,7 +78,7 @@ public class HealthCheckService {
}
@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());
@@ -103,6 +113,7 @@ public class HealthCheckService {
httpService.unregister("/health/check");
registered = false;
}
+ executor.shutdownNow();
}
@Reference(service = HealthCheckProvider.class, cardinality =
ReferenceCardinality.MULTIPLE, policy = ReferencePolicy.DYNAMIC, unbind =
"unbind")
@@ -130,7 +141,7 @@ public class HealthCheckService {
}
} else if (shouldRefreshCache()) {
synchronized (cacheLock) {
- if (!busy) {
+ if (isCacheExpired()) {
busy = true;
try {
refreshCache();
@@ -144,21 +155,41 @@ public class HealthCheckService {
}
private boolean shouldRefreshCache() {
- return !busy && (System.currentTimeMillis() - cacheTimestamp) > 1000;
+ 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"));
- for (HealthCheckProvider provider : providers.stream()
+ List<HealthCheckProvider> enabledProviders = providers.stream()
.filter(p ->
config.getEnabledProviders().contains(p.name()))
- .toList()) {
+ .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 {
- HealthCheckResponse response = provider.execute();
- health.add(response);
+ 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 (InterruptedException e) {
+ futures.forEach(f -> f.cancel(true));
+ Thread.currentThread().interrupt();
+ LOGGER.warn("Health check refresh interrupted, keeping the
previous results");
+ return;
} catch (Exception e) {
- LOGGER.error("Error while executing health check", e);
+ LOGGER.error("Error while executing health check {}",
provider.name(), e);
health.add(provider.timeout());
}
}
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 b64ad50a4..708e3b467 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
@@ -24,6 +24,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;
@@ -120,7 +121,13 @@ public class ElasticSearchHealthCheckProvider implements
HealthCheckProvider {
config.get(HealthCheckConfig.CONFIG_ES_ADDRESSES).split(",")[0].trim()).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/provider/OpenSearchHealthCheckProvider.java
b/extensions/healthcheck/src/main/java/org/apache/unomi/healthcheck/provider/OpenSearchHealthCheckProvider.java
index 7e3b21461..53f04f606 100644
---
a/extensions/healthcheck/src/main/java/org/apache/unomi/healthcheck/provider/OpenSearchHealthCheckProvider.java
+++
b/extensions/healthcheck/src/main/java/org/apache/unomi/healthcheck/provider/OpenSearchHealthCheckProvider.java
@@ -24,6 +24,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;
@@ -120,7 +121,13 @@ public class OpenSearchHealthCheckProvider implements
HealthCheckProvider {
config.get(HealthCheckConfig.CONFIG_OS_ADDRESSES).split(",")[0].trim()).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/test/java/org/apache/unomi/healthcheck/HealthCheckServiceTest.java
b/extensions/healthcheck/src/test/java/org/apache/unomi/healthcheck/HealthCheckServiceTest.java
new file mode 100644
index 000000000..c4ebdfc9b
--- /dev/null
+++
b/extensions/healthcheck/src/test/java/org/apache/unomi/healthcheck/HealthCheckServiceTest.java
@@ -0,0 +1,157 @@
+/*
+ * 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.unomi.healthcheck;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Supplier;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class HealthCheckServiceTest {
+
+ private static final int TIMEOUT = 200;
+
+ private final CountDownLatch neverReleased = new CountDownLatch(1);
+ private HealthCheckService service;
+
+ @BeforeEach
+ void setUp() throws Exception {
+ HealthCheckConfig config = new HealthCheckConfig();
+ config.setEnabledProviders(List.of("fast", "blocked"));
+ config.setTimeout(TIMEOUT);
+ service = new HealthCheckService();
+ service.setConfig(config);
+ }
+
+ @AfterEach
+ void tearDown() {
+ service.deactivate();
+ }
+
+ @Test
+ void blockedProviderTimesOutAndOtherProvidersAnswer() {
+ service.bind(new TestProvider("fast", () ->
HealthCheckResponse.live("fast")));
+ service.bind(new TestProvider("blocked", this::waitForever));
+
+ List<HealthCheckResponse> health =
assertTimeoutPreemptively(Duration.ofMillis(10L * TIMEOUT), service::check);
+
+ assertEquals(3, health.size());
+ assertTrue(response(health, "karaf").isLive());
+ assertTrue(response(health, "fast").isLive());
+ HealthCheckResponse blocked = response(health, "blocked");
+ assertTrue(blocked.isError());
+ assertEquals("timeout", blocked.getData().get("error.cause"));
+ }
+
+ @Test
+ void secondCallWithinOneSecondReturnsTheCachedResults() {
+ TestProvider fast = new TestProvider("fast", () ->
HealthCheckResponse.live("fast"));
+ service.bind(fast);
+
+ List<HealthCheckResponse> first = service.check();
+ List<HealthCheckResponse> second = service.check();
+
+ assertEquals(1, fast.calls.get());
+ assertEquals(first, second);
+ }
+
+ @Test
+ void concurrentCallsOnAnExpiredCacheRefreshOnlyOnce() throws Exception {
+ TestProvider slow = new TestProvider("fast", () -> {
+ try {
+ Thread.sleep(50);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ return HealthCheckResponse.live("fast");
+ });
+ service.bind(slow);
+ service.check();
+ Thread.sleep(1100);
+
+ int threads = 20;
+ CountDownLatch start = new CountDownLatch(1);
+ ExecutorService callers = Executors.newFixedThreadPool(threads);
+ try {
+ List<Future<?>> calls = new ArrayList<>();
+ for (int i = 0; i < threads; i++) {
+ calls.add(callers.submit(() -> {
+ start.await();
+ return service.check();
+ }));
+ }
+ start.countDown();
+ for (Future<?> call : calls) {
+ call.get(5, TimeUnit.SECONDS);
+ }
+ } finally {
+ callers.shutdownNow();
+ }
+
+ assertEquals(2, slow.calls.get());
+ }
+
+ private HealthCheckResponse waitForever() {
+ try {
+ neverReleased.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ return HealthCheckResponse.live("blocked");
+ }
+
+ private static HealthCheckResponse response(List<HealthCheckResponse>
health, String name) {
+ return health.stream().filter(r ->
r.getName().equals(name)).findFirst().orElseThrow();
+ }
+
+ private static class TestProvider implements HealthCheckProvider {
+
+ private final String name;
+ private final Supplier<HealthCheckResponse> execution;
+ private final AtomicInteger calls = new AtomicInteger();
+
+ TestProvider(String name, Supplier<HealthCheckResponse> execution) {
+ this.name = name;
+ this.execution = execution;
+ }
+
+ @Override
+ public String name() {
+ return name;
+ }
+
+ @Override
+ public HealthCheckResponse execute() {
+ calls.incrementAndGet();
+ return execution.get();
+ }
+ }
+}
