diff --git a/spring-boot-admin-server/src/main/java/de/codecentric/boot/admin/server/services/EndpointDetectionTrigger.java b/spring-boot-admin-server/src/main/java/de/codecentric/boot/admin/server/services/EndpointDetectionTrigger.java index 53132b71a8d..f7a7b944450 100644 --- a/spring-boot-admin-server/src/main/java/de/codecentric/boot/admin/server/services/EndpointDetectionTrigger.java +++ b/spring-boot-admin-server/src/main/java/de/codecentric/boot/admin/server/services/EndpointDetectionTrigger.java @@ -25,6 +25,7 @@ import de.codecentric.boot.admin.server.domain.events.InstanceEvent; import de.codecentric.boot.admin.server.domain.events.InstanceRegistrationUpdatedEvent; import de.codecentric.boot.admin.server.domain.events.InstanceStatusChangedEvent; +import de.codecentric.boot.admin.server.services.endpoints.QueryIndexEndpointStrategy; public class EndpointDetectionTrigger extends AbstractEventHandler { @@ -47,7 +48,13 @@ protected Publisher handle(Flux publisher) { protected Mono detectEndpoints(InstanceEvent event) { return this.endpointDetector.detectEndpoints(event.getInstance()).onErrorResume((e) -> { - log.warn("Unexpected error while detecting endpoints for {}", event.getInstance(), e); + if (QueryIndexEndpointStrategy.isTransportError(e)) { + log.warn("Endpoint detection for {} failed, instance unreachable: {}", event.getInstance(), + e.toString()); + } + else { + log.warn("Unexpected error while detecting endpoints for {}", event.getInstance(), e); + } return Mono.empty(); }); } diff --git a/spring-boot-admin-server/src/main/java/de/codecentric/boot/admin/server/services/endpoints/QueryIndexEndpointStrategy.java b/spring-boot-admin-server/src/main/java/de/codecentric/boot/admin/server/services/endpoints/QueryIndexEndpointStrategy.java index 602c32a02c9..8b684d1c804 100644 --- a/spring-boot-admin-server/src/main/java/de/codecentric/boot/admin/server/services/endpoints/QueryIndexEndpointStrategy.java +++ b/spring-boot-admin-server/src/main/java/de/codecentric/boot/admin/server/services/endpoints/QueryIndexEndpointStrategy.java @@ -16,10 +16,15 @@ package de.codecentric.boot.admin.server.services.endpoints; +import java.io.IOException; +import java.util.Collections; import java.util.HashMap; +import java.util.IdentityHashMap; import java.util.List; import java.util.Map; import java.util.Objects; +import java.util.Set; +import java.util.concurrent.TimeoutException; import java.util.function.Function; import com.fasterxml.jackson.annotation.JsonCreator; @@ -27,7 +32,9 @@ import lombok.Data; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.springframework.core.codec.CodecException; import org.springframework.web.reactive.function.client.ClientResponse; +import org.springframework.web.reactive.function.client.WebClientRequestException; import reactor.core.publisher.Mono; import de.codecentric.boot.admin.server.domain.entities.Instance; @@ -64,9 +71,12 @@ public Mono detectEndpoints(Instance instance) { .get() .uri(managementUrl) .exchangeToMono(this.convert(instance, managementUrl)) - .onErrorResume((e) -> { + // shortcut: failed detection is not retried until the next + // status/registration event, add scheduled re-detection if instances stay + // without endpoints + .onErrorResume((e) -> !isTransportError(e), (e) -> { log.warn("Querying actuator-index for instance {} on '{}' failed: {}", instance.getId(), managementUrl, - e.getMessage()); + e.toString()); log.debug("Querying actuator-index for instance {} on '{}' failed.", instance.getId(), managementUrl, e); return Mono.empty(); @@ -123,6 +133,21 @@ protected Mono convertResponse(Response response) { return endpoints.isEmpty() ? Mono.empty() : Mono.just(Endpoints.of(endpoints)); } + // transport errors propagate instead of falling back: instance is unreachable, so + // probing would fail as well + public static boolean isTransportError(Throwable e) { + Set seen = Collections.newSetFromMap(new IdentityHashMap<>()); + for (Throwable t = e; t != null && seen.add(t); t = t.getCause()) { + if (t instanceof CodecException) { + return false; // malformed body, instance reachable -> allow fallback + } + if (t instanceof WebClientRequestException || t instanceof TimeoutException || t instanceof IOException) { + return true; + } + } + return false; + } + @Data protected static class Response { diff --git a/spring-boot-admin-server/src/test/java/de/codecentric/boot/admin/server/services/EndpointDetectorTest.java b/spring-boot-admin-server/src/test/java/de/codecentric/boot/admin/server/services/EndpointDetectorTest.java index d1fcb8e824b..ed53c0ddf3b 100644 --- a/spring-boot-admin-server/src/test/java/de/codecentric/boot/admin/server/services/EndpointDetectorTest.java +++ b/spring-boot-admin-server/src/test/java/de/codecentric/boot/admin/server/services/EndpointDetectorTest.java @@ -104,4 +104,27 @@ void should_update_endpoints() { .verifyComplete(); } + @Test + void should_propagate_error_and_keep_endpoints() { + // given + Registration registration = Registration.create("foo", "https://health").managementUrl("https://mgmt").build(); + Endpoints originalEndpoints = Endpoints.single("info", "https://mgmt/info") + .withEndpoint("health", "https://health"); + Instance instance = Instance.create(InstanceId.of("id")) + .register(registration) + .withEndpoints(originalEndpoints) + .withStatusInfo(StatusInfo.ofUp()); + StepVerifier.create(repository.save(instance)).expectNextCount(1).verifyComplete(); + + when(strategy.detectEndpoints(any(Instance.class))).thenReturn(Mono.error(new IllegalStateException("x"))); + + // when + StepVerifier.create(detector.detectEndpoints(instance.getId())).verifyError(IllegalStateException.class); + + // then + StepVerifier.create(repository.find(instance.getId())) + .assertNext((app) -> assertThat(app.getEndpoints()).isEqualTo(originalEndpoints)) + .verifyComplete(); + } + } diff --git a/spring-boot-admin-server/src/test/java/de/codecentric/boot/admin/server/services/endpoints/ChainingStrategyTest.java b/spring-boot-admin-server/src/test/java/de/codecentric/boot/admin/server/services/endpoints/ChainingStrategyTest.java index 3c3ccb10585..388720fde16 100644 --- a/spring-boot-admin-server/src/test/java/de/codecentric/boot/admin/server/services/endpoints/ChainingStrategyTest.java +++ b/spring-boot-admin-server/src/test/java/de/codecentric/boot/admin/server/services/endpoints/ChainingStrategyTest.java @@ -16,6 +16,8 @@ package de.codecentric.boot.admin.server.services.endpoints; +import java.util.concurrent.atomic.AtomicBoolean; + import org.junit.jupiter.api.Test; import reactor.core.publisher.Mono; import reactor.test.StepVerifier; @@ -24,6 +26,7 @@ import de.codecentric.boot.admin.server.domain.values.Endpoints; import de.codecentric.boot.admin.server.domain.values.InstanceId; +import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; class ChainingStrategyTest { @@ -59,4 +62,19 @@ void should_return_empty_endpoints_when_all_empty() { StepVerifier.create(strategy.detectEndpoints(instance)).expectNext(Endpoints.empty()).verifyComplete(); } + @Test + void should_not_chain_on_error() { + AtomicBoolean subscribed = new AtomicBoolean(false); + Instance instance = Instance.create(InstanceId.of("id")); + ChainingStrategy strategy = new ChainingStrategy((a) -> Mono.error(new IllegalStateException("x")), + (a) -> Mono.fromCallable(() -> { + subscribed.set(true); + return Endpoints.single("id", "path"); + })); + + StepVerifier.create(strategy.detectEndpoints(instance)).verifyError(IllegalStateException.class); + + assertThat(subscribed).isFalse(); + } + } diff --git a/spring-boot-admin-server/src/test/java/de/codecentric/boot/admin/server/services/endpoints/QueryIndexEndpointStrategyTest.java b/spring-boot-admin-server/src/test/java/de/codecentric/boot/admin/server/services/endpoints/QueryIndexEndpointStrategyTest.java index b76622a6bde..4f13b4f7587 100644 --- a/spring-boot-admin-server/src/test/java/de/codecentric/boot/admin/server/services/endpoints/QueryIndexEndpointStrategyTest.java +++ b/spring-boot-admin-server/src/test/java/de/codecentric/boot/admin/server/services/endpoints/QueryIndexEndpointStrategyTest.java @@ -16,7 +16,11 @@ package de.codecentric.boot.admin.server.services.endpoints; +import java.io.IOException; +import java.net.ServerSocket; +import java.net.SocketException; import java.time.Duration; +import java.util.concurrent.TimeoutException; import javax.net.ssl.SSLException; @@ -27,10 +31,13 @@ import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; import org.springframework.boot.actuate.endpoint.ApiVersion; +import org.springframework.core.codec.DecodingException; import org.springframework.http.MediaType; import org.springframework.http.client.reactive.ReactorClientHttpConnector; import org.springframework.web.reactive.function.client.WebClient; +import org.springframework.web.reactive.function.client.WebClientRequestException; import reactor.netty.http.client.HttpClient; import reactor.test.StepVerifier; @@ -56,6 +63,7 @@ import static de.codecentric.boot.admin.server.web.client.InstanceExchangeFilterFunctions.timeout; import static java.util.Collections.emptyMap; import static java.util.Collections.singletonMap; +import static org.assertj.core.api.Assertions.assertThat; class QueryIndexEndpointStrategyTest { @@ -177,7 +185,7 @@ void should_return_empty_on_not_found() { } @Test - void should_return_empty_on_error() { + void should_return_error_on_transport_error() { // given Instance instance = Instance.create(InstanceId.of("id")) .register(Registration.create("test", this.wireMock.url("/mgmt/health")) @@ -192,7 +200,7 @@ void should_return_empty_on_error() { // when StepVerifier.create(strategy.detectEndpoints(instance)) // then - .verifyComplete(); + .verifyError(WebClientRequestException.class); } @Test @@ -261,6 +269,82 @@ void should_retry() { .verifyComplete(); } + @Test + void should_return_error_on_connection_refused() throws IOException { + // given + int port; + try (ServerSocket socket = new ServerSocket(0)) { + port = socket.getLocalPort(); + } + + Instance instance = Instance.create(InstanceId.of("id")) + .register(Registration.create("test", "http://localhost:" + port + "/mgmt/health") + .managementUrl("http://localhost:" + port + "/mgmt") + .build()); + + QueryIndexEndpointStrategy strategy = new QueryIndexEndpointStrategy(this.instanceWebClient, + this.apiMediaTypeHandler); + + // when + StepVerifier.create(strategy.detectEndpoints(instance)) + // then + .verifyError(WebClientRequestException.class); + } + + @Test + void should_return_error_on_timeout() { + // given + Instance instance = Instance.create(InstanceId.of("id")) + .register(Registration.create("test", this.wireMock.url("/mgmt/health")) + .managementUrl(this.wireMock.url("/mgmt")) + .build()); + + this.wireMock.stubFor(get("/mgmt").willReturn(okJson("{}").withFixedDelay(2000))); + + QueryIndexEndpointStrategy strategy = new QueryIndexEndpointStrategy(this.instanceWebClient, + this.apiMediaTypeHandler); + + // when + StepVerifier.create(strategy.detectEndpoints(instance)) + // then + .verifyError(TimeoutException.class); + } + + @Test + void should_return_empty_on_invalid_body() { + // given + Instance instance = Instance.create(InstanceId.of("id")) + .register(Registration.create("test", this.wireMock.url("/mgmt/health")) + .managementUrl(this.wireMock.url("/mgmt")) + .build()); + + this.wireMock.stubFor(get("/mgmt") + .willReturn(ok("{invalid").withHeader("Content-Type", ApiVersion.LATEST.getProducedMimeType().toString()))); + + QueryIndexEndpointStrategy strategy = new QueryIndexEndpointStrategy(this.instanceWebClient, + this.apiMediaTypeHandler); + + // when + StepVerifier.create(strategy.detectEndpoints(instance)) + // then + .verifyComplete(); + } + + @Test + @Timeout(5) + void should_detect_transport_errors() { + assertThat(QueryIndexEndpointStrategy.isTransportError(new TimeoutException())).isTrue(); + assertThat(QueryIndexEndpointStrategy.isTransportError(new RuntimeException(new SocketException("reset")))) + .isTrue(); + assertThat(QueryIndexEndpointStrategy.isTransportError(new DecodingException("bad", new IOException()))) + .isFalse(); + assertThat(QueryIndexEndpointStrategy.isTransportError(new IllegalStateException())).isFalse(); + RuntimeException a = new RuntimeException("a"); + RuntimeException b = new RuntimeException("b", a); + a.initCause(b); + assertThat(QueryIndexEndpointStrategy.isTransportError(a)).isFalse(); + } + private ReactorClientHttpConnector httpConnector() { HttpClient client = HttpClient.create().secure((ssl) -> { try {