This is an automated email from the ASF dual-hosted git repository. jayblanc pushed a commit to branch UNOMI-1025-healthcheck-provider-timeout in repository https://gitbox.apache.org/repos/asf/unomi.git
commit f70eb79117f5d9199f5bcd6092e24887f4bc27d8 Author: Jérôme Blanchard <[email protected]> AuthorDate: Mon Oct 26 23:15:59 2026 +0100 [UNOMI-1025] Restore the healthcheck provider timeout Run the providers in an executor and wait for them until one common deadline, healthcheck.timeout. Set the same timeout on the _cluster/health request of the ElasticSearch and OpenSearch providers. Check the cache age again inside cacheLock, and use a CopyOnWriteArrayList for the providers. --- .../unomi/healthcheck/HealthCheckService.java | 39 +++++-- .../provider/ElasticSearchHealthCheckProvider.java | 9 +- .../provider/OpenSearchHealthCheckProvider.java | 9 +- .../unomi/healthcheck/HealthCheckServiceTest.java | 116 +++++++++++++++++++++ 4 files changed, 163 insertions(+), 10 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..2021c4398 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,12 @@ 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; @@ -42,7 +48,8 @@ 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 ExecutorService executor = Executors.newCachedThreadPool(); private final Object cacheLock = new Object(); private volatile long cacheTimestamp = 0L; private volatile List<HealthCheckResponse> healthCache = Collections.emptyList(); @@ -68,7 +75,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 +110,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 +138,7 @@ public class HealthCheckService { } } else if (shouldRefreshCache()) { synchronized (cacheLock) { - if (!busy) { + if (isCacheExpired()) { busy = true; try { refreshCache(); @@ -144,19 +152,34 @@ 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 (Exception e) { LOGGER.error("Error while executing health check", 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..da03074ac --- /dev/null +++ b/extensions/healthcheck/src/test/java/org/apache/unomi/healthcheck/HealthCheckServiceTest.java @@ -0,0 +1,116 @@ +/* + * 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.List; +import java.util.concurrent.CountDownLatch; +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); + } + + 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(); + } + } +}
