Fix/status update trigger (#2719)

Author: SteKoe

spring-boot-admin-server/src/main/java/de/codecentric/boot/admin/server/services/AbstractEventHandler.java

@@ -26,7 +26,6 @@
 import reactor.core.publisher.Flux;
 import reactor.core.scheduler.Scheduler;
 import reactor.core.scheduler.Schedulers;
-import reactor.util.retry.Retry;
 
 import de.codecentric.boot.admin.server.domain.events.InstanceEvent;
 
@@ -58,7 +57,7 @@ public void start() {
 			.ofType(this.eventType)
 			.cast(this.eventType)
 			.transform(this::handle)
-			.retryWhen(Retry.indefinitely().doBeforeRetry((s) -> this.log.warn("Unexpected error", s.failure())))
+			.onErrorContinue((throwable, o) -> this.log.warn("Unexpected error", throwable))
 			.subscribe();
 	}
 

spring-boot-admin-server/src/main/java/de/codecentric/boot/admin/server/services/IntervalCheck.java

@@ -23,8 +23,9 @@
 import java.util.function.Function;
 import java.util.logging.Level;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import lombok.Getter;
+import lombok.Setter;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.lang.Nullable;
 import reactor.core.Disposable;
 import reactor.core.publisher.Flux;
@@ -42,18 +43,20 @@
  *
  * @author Johannes Edmeier
  */
+@Slf4j
 public class IntervalCheck {
 
-	private static final Logger log = LoggerFactory.getLogger(IntervalCheck.class);
-
 	private final String name;
 
 	private final Map<InstanceId, Instant> lastChecked = new ConcurrentHashMap<>();
 
 	private final Function<InstanceId, Mono<Void>> checkFn;
 
+	@Getter
+	@Setter
 	private Duration interval;
 
+	@Setter
 	private Duration minRetention;
 
 	@Nullable
@@ -83,7 +86,7 @@ public void start() {
 			.concatMap((i) -> this.checkAllInstances())
 			.retryWhen(Retry.backoff(Long.MAX_VALUE, Duration.ofSeconds(1))
 				.doBeforeRetry((s) -> log.warn("Unexpected error in {}-check", this.name, s.failure())))
-			.subscribe();
+			.subscribe(null, (error) -> log.error("Unexpected error in {}-check", name, error));
 	}
 
 	public void markAsChecked(InstanceId instanceId) {
@@ -94,7 +97,7 @@ protected Mono<Void> checkAllInstances() {
 		log.debug("check {} for all instances", this.name);
 		Instant expiration = Instant.now().minus(this.minRetention);
 		return Flux.fromIterable(this.lastChecked.entrySet())
-			.filter((e) -> e.getValue().isBefore(expiration))
+			.filter((entry) -> entry.getValue().isBefore(expiration))
 			.map(Map.Entry::getKey)
 			.flatMap(this.checkFn)
 			.then();
@@ -111,12 +114,4 @@ public void stop() {
 		}
 	}
 
-	public void setInterval(Duration interval) {
-		this.interval = interval;
-	}
-
-	public void setMinRetention(Duration minRetention) {
-		this.minRetention = minRetention;
-	}
-
 }

spring-boot-admin-server/src/main/java/de/codecentric/boot/admin/server/services/StatusUpdateTrigger.java

@@ -52,10 +52,13 @@ protected Publisher<Void> handle(Flux<InstanceEvent> publisher) {
 	}
 
 	protected Mono<Void> updateStatus(InstanceId instanceId) {
-		return this.statusUpdater.updateStatus(instanceId).onErrorResume((e) -> {
-			log.warn("Unexpected error while updating status for {}", instanceId, e);
-			return Mono.empty();
-		}).doFinally((s) -> this.intervalCheck.markAsChecked(instanceId));
+		return this.statusUpdater.timeout(this.intervalCheck.getInterval())
+			.updateStatus(instanceId)
+			.onErrorResume((e) -> {
+				log.warn("Unexpected error while updating status for {}", instanceId, e);
+				return Mono.empty();
+			})
+			.doFinally((s) -> this.intervalCheck.markAsChecked(instanceId));
 	}
 
 	@Override

spring-boot-admin-server/src/main/java/de/codecentric/boot/admin/server/services/StatusUpdater.java

@@ -16,14 +16,15 @@
 
 package de.codecentric.boot.admin.server.services;
 
+import java.time.Duration;
 import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.Objects;
 import java.util.logging.Level;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.core.ParameterizedTypeReference;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.HttpStatusCode;
@@ -46,10 +47,10 @@
  *
  * @author Johannes Edmeier
  */
+@Slf4j
+@RequiredArgsConstructor
 public class StatusUpdater {
 
-	private static final Logger log = LoggerFactory.getLogger(StatusUpdater.class);
-
 	private static final ParameterizedTypeReference<Map<String, Object>> RESPONSE_TYPE = new ParameterizedTypeReference<>() {
 	};
 
@@ -59,16 +60,15 @@ public class StatusUpdater {
 
 	private final ApiMediaTypeHandler apiMediaTypeHandler;
 
-	public StatusUpdater(InstanceRepository repository, InstanceWebClient instanceWebClient,
-			ApiMediaTypeHandler apiMediaTypeHandler) {
-		this.repository = repository;
-		this.instanceWebClient = instanceWebClient;
-		this.apiMediaTypeHandler = apiMediaTypeHandler;
+	private Duration timeout = Duration.ofSeconds(10);
+
+	public StatusUpdater timeout(Duration timeout) {
+		this.timeout = timeout;
+		return this;
 	}
 
 	public Mono<Void> updateStatus(InstanceId id) {
 		return this.repository.computeIfPresent(id, (key, instance) -> this.doUpdateStatus(instance)).then();
-
 	}
 
 	protected Mono<Instance> doUpdateStatus(Instance instance) {
@@ -82,11 +82,20 @@ protected Mono<Instance> doUpdateStatus(Instance instance) {
 			.uri(Endpoint.HEALTH)
 			.exchangeToMono(this::convertStatusInfo)
 			.log(log.getName(), Level.FINEST)
+			.timeout(getTimeoutWithMargin())
 			.doOnError((ex) -> logError(instance, ex))
 			.onErrorResume(this::handleError)
 			.map(instance::withStatusInfo);
 	}
 
+	/*
+	 * return a timeout less than the given one to prevent backdrops in concurrent get
+	 * request. This prevents flakyness of health checks.
+	 */
+	private Duration getTimeoutWithMargin() {
+		return this.timeout.minusSeconds(1).abs();
+	}
+
 	protected Mono<StatusInfo> convertStatusInfo(ClientResponse response) {
 		boolean hasCompatibleContentType = response.headers()
 			.contentType()

spring-boot-admin-server/src/test/java/de/codecentric/boot/admin/server/services/StatusUpdateTriggerTest.java

@@ -56,6 +56,7 @@ public class StatusUpdateTriggerTest {
 	@BeforeEach
 	public void setUp() throws Exception {
 		when(this.updater.updateStatus(any(InstanceId.class))).thenReturn(Mono.empty());
+		when(this.updater.timeout(any())).thenReturn(this.updater);
 
 		this.trigger = new StatusUpdateTrigger(this.updater, this.events.flux());
 		this.trigger.start();