From bdd95f09a49bf02f4fef3d83815303e7d924f898 Mon Sep 17 00:00:00 2001 From: Brian Clozel Date: Thu, 5 Jul 2018 15:58:34 +0200 Subject: [PATCH] Polish WebClient metrics support This commit improves the for Actuator Metrics in WebClient. Unlike the server couterpart of WebFlux, using a `retry` operator on a `WebClient` pipeline does resubscribes to the whole chain. The previous implementation recorded start time at the time of pipeline build phase, but outside of it. This doesn't work since retrying the same pipeline doesn't update the recorded start time and the duration of sequential calls are cumulative. This is now fixed using the Reactor `Context`, since we're now recording the start time at subscription time and record metrics on `onNext` and `onError` signals. Closes gh-12228 --- .../MetricsWebClientFilterFunction.java | 27 ++++++++++++------- .../MetricsWebClientFilterFunctionTests.java | 21 +++++++++++++++ 2 files changed, 38 insertions(+), 10 deletions(-) diff --git a/spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/metrics/web/reactive/client/MetricsWebClientFilterFunction.java b/spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/metrics/web/reactive/client/MetricsWebClientFilterFunction.java index 89468ff246..c884e3e04b 100644 --- a/spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/metrics/web/reactive/client/MetricsWebClientFilterFunction.java +++ b/spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/metrics/web/reactive/client/MetricsWebClientFilterFunction.java @@ -37,6 +37,9 @@ import org.springframework.web.reactive.function.client.ExchangeFunction; */ public class MetricsWebClientFilterFunction implements ExchangeFilterFunction { + private static final String METRICS_WEBCLIENT_START_TIME = MetricsWebClientFilterFunction.class + .getName() + ".START_TIME"; + private final MeterRegistry meterRegistry; private final WebClientExchangeTagsProvider tagProvider; @@ -53,16 +56,20 @@ public class MetricsWebClientFilterFunction implements ExchangeFilterFunction { @Override public Mono filter(ClientRequest clientRequest, ExchangeFunction exchangeFunction) { - long startTime = System.nanoTime(); - return exchangeFunction.exchange(clientRequest) - .doOnSuccessOrError((clientResponse, throwable) -> { - Iterable tags = this.tagProvider.tags(clientRequest, - clientResponse, throwable); - Timer.builder(this.metricName).tags(tags) - .description("Timer of WebClient operation") - .register(this.meterRegistry) - .record(System.nanoTime() - startTime, TimeUnit.NANOSECONDS); - }); + return exchangeFunction.exchange(clientRequest).doOnEach((signal) -> { + if (!signal.isOnComplete()) { + Long startTime = signal.getContext().get(METRICS_WEBCLIENT_START_TIME); + ClientResponse clientResponse = signal.get(); + Throwable throwable = signal.getThrowable(); + Iterable tags = this.tagProvider.tags(clientRequest, clientResponse, + throwable); + Timer.builder(this.metricName).tags(tags) + .description("Timer of WebClient operation") + .register(this.meterRegistry) + .record(System.nanoTime() - startTime, TimeUnit.NANOSECONDS); + } + }).subscriberContext((context) -> context.put(METRICS_WEBCLIENT_START_TIME, + System.nanoTime())); } } diff --git a/spring-boot-project/spring-boot-actuator/src/test/java/org/springframework/boot/actuate/metrics/web/reactive/client/MetricsWebClientFilterFunctionTests.java b/spring-boot-project/spring-boot-actuator/src/test/java/org/springframework/boot/actuate/metrics/web/reactive/client/MetricsWebClientFilterFunctionTests.java index 533df63d24..0e1eea7187 100644 --- a/spring-boot-project/spring-boot-actuator/src/test/java/org/springframework/boot/actuate/metrics/web/reactive/client/MetricsWebClientFilterFunctionTests.java +++ b/spring-boot-project/spring-boot-actuator/src/test/java/org/springframework/boot/actuate/metrics/web/reactive/client/MetricsWebClientFilterFunctionTests.java @@ -18,6 +18,8 @@ package org.springframework.boot.actuate.metrics.web.reactive.client; import java.io.IOException; import java.net.URI; +import java.time.Duration; +import java.util.concurrent.TimeUnit; import io.micrometer.core.instrument.MeterRegistry; import io.micrometer.core.instrument.MockClock; @@ -117,4 +119,23 @@ public class MetricsWebClientFilterFunctionTests { .timer().count()).isEqualTo(1); } + @Test + public void filterWhenExceptionAndRetryShouldNotCumulateRecordTime() { + ClientRequest request = ClientRequest.create(HttpMethod.GET, + URI.create("http://example.com/projects/spring-boot")).build(); + ExchangeFunction exchange = (r) -> Mono.error(new IllegalArgumentException()) + .delaySubscription(Duration.ofMillis(300)).cast(ClientResponse.class); + this.filterFunction.filter(request, exchange).retry(1) + .onErrorResume(IllegalArgumentException.class, (t) -> Mono.empty()) + .block(); + assertThat(this.registry + .get("http.client.requests").tags("method", "GET", "uri", + "/projects/spring-boot", "status", "CLIENT_ERROR") + .timer().count()).isEqualTo(2); + assertThat(this.registry.get("http.client.requests") + .tags("method", "GET", "uri", "/projects/spring-boot", "status", + "CLIENT_ERROR") + .timer().max(TimeUnit.MILLISECONDS)).isLessThan(600); + } + }