diff --git a/spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/metrics/web/reactive/server/CancelledServerWebExchangeException.java b/spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/metrics/web/reactive/server/CancelledServerWebExchangeException.java new file mode 100644 index 0000000000..b0e6c40c7d --- /dev/null +++ b/spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/metrics/web/reactive/server/CancelledServerWebExchangeException.java @@ -0,0 +1,29 @@ +/* + * Copyright 2012-2021 the original author or authors. + * + * Licensed 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 + * + * https://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.springframework.boot.actuate.metrics.web.reactive.server; + +/** + * Runtime exception that materializes a {@link reactor.core.publisher.SignalType#CANCEL + * cancel signal} for the WebFlux server metrics instrumentation. + * + * @author Brian Clozel + * @since 2.5.0 + * @see MetricsWebFilter + */ +public class CancelledServerWebExchangeException extends RuntimeException { + +} diff --git a/spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/metrics/web/reactive/server/DefaultWebFluxTagsProvider.java b/spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/metrics/web/reactive/server/DefaultWebFluxTagsProvider.java index 12f95984b5..5a55bf07d8 100644 --- a/spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/metrics/web/reactive/server/DefaultWebFluxTagsProvider.java +++ b/spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/metrics/web/reactive/server/DefaultWebFluxTagsProvider.java @@ -71,7 +71,8 @@ public class DefaultWebFluxTagsProvider implements WebFluxTagsProvider { @Override public Iterable httpRequestTags(ServerWebExchange exchange, Throwable exception) { Tags tags = Tags.of(WebFluxTags.method(exchange), WebFluxTags.uri(exchange, this.ignoreTrailingSlash), - WebFluxTags.exception(exception), WebFluxTags.status(exchange), WebFluxTags.outcome(exchange)); + WebFluxTags.exception(exception), WebFluxTags.status(exchange), + WebFluxTags.outcome(exchange, exception)); for (WebFluxTagsContributor contributor : this.contributors) { tags = tags.and(contributor.httpRequestTags(exchange, exception)); } diff --git a/spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/metrics/web/reactive/server/MetricsWebFilter.java b/spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/metrics/web/reactive/server/MetricsWebFilter.java index 09760ba24a..d2e216b3ed 100644 --- a/spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/metrics/web/reactive/server/MetricsWebFilter.java +++ b/spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/metrics/web/reactive/server/MetricsWebFilter.java @@ -1,5 +1,5 @@ /* - * Copyright 2012-2020 the original author or authors. + * Copyright 2012-2021 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -75,28 +75,24 @@ public class MetricsWebFilter implements WebFilter { private Publisher filter(ServerWebExchange exchange, Mono call) { long start = System.nanoTime(); - return call.doOnSuccess((done) -> onSuccess(exchange, start)) - .doOnError((cause) -> onError(exchange, start, cause)); + return call.doOnEach((signal) -> onTerminalSignal(exchange, signal.getThrowable(), start)) + .doOnCancel(() -> onTerminalSignal(exchange, new CancelledServerWebExchangeException(), start)); } - private void onSuccess(ServerWebExchange exchange, long start) { - record(exchange, start, null); - } - - private void onError(ServerWebExchange exchange, long start, Throwable cause) { + private void onTerminalSignal(ServerWebExchange exchange, Throwable cause, long start) { ServerHttpResponse response = exchange.getResponse(); - if (response.isCommitted()) { - record(exchange, start, cause); + if (response.isCommitted() || cause instanceof CancelledServerWebExchangeException) { + record(exchange, cause, start); } else { response.beforeCommit(() -> { - record(exchange, start, cause); + record(exchange, cause, start); return Mono.empty(); }); } } - private void record(ServerWebExchange exchange, long start, Throwable cause) { + private void record(ServerWebExchange exchange, Throwable cause, long start) { Iterable tags = this.tagsProvider.httpRequestTags(exchange, cause); this.autoTimer.builder(this.metricName).tags(tags).register(this.registry).record(System.nanoTime() - start, TimeUnit.NANOSECONDS); diff --git a/spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/metrics/web/reactive/server/WebFluxTags.java b/spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/metrics/web/reactive/server/WebFluxTags.java index f0d4daab27..7a6046925b 100644 --- a/spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/metrics/web/reactive/server/WebFluxTags.java +++ b/spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/metrics/web/reactive/server/WebFluxTags.java @@ -1,5 +1,5 @@ /* - * Copyright 2012-2020 the original author or authors. + * Copyright 2012-2021 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -16,6 +16,9 @@ package org.springframework.boot.actuate.metrics.web.reactive.server; +import java.util.Arrays; +import java.util.HashSet; +import java.util.Set; import java.util.regex.Pattern; import io.micrometer.core.instrument.Tag; @@ -35,6 +38,7 @@ import org.springframework.web.util.pattern.PathPattern; * @author Jon Schneider * @author Andy Wilkinson * @author Michael McFadyen + * @author Brian Clozel * @since 2.0.0 */ public final class WebFluxTags { @@ -51,6 +55,9 @@ public final class WebFluxTags { private static final Pattern FORWARD_SLASHES_PATTERN = Pattern.compile("//+"); + private static final Set DISCONNECTED_CLIENT_EXCEPTIONS = new HashSet<>( + Arrays.asList("AbortedException", "ClientAbortException", "EOFException", "EofException")); + private WebFluxTags() { } @@ -165,8 +172,28 @@ public final class WebFluxTags { * @param exchange the exchange * @return the outcome tag derived from the response status * @since 2.1.0 + * @deprecated as of 2.5.0 in favor of {@link #outcome(ServerWebExchange, Throwable)}, + * to be removed in 2.7.0 */ public static Tag outcome(ServerWebExchange exchange) { + return outcome(exchange, null); + } + + /** + * Creates an {@code outcome} tag based on the response status of the given + * {@code exchange} and the exception thrown during request processing. + * @param exchange the exchange + * @param exception the termination signal sent by the publisher + * @return the outcome tag derived from the response status + * @since 2.5.0 + */ + public static Tag outcome(ServerWebExchange exchange, Throwable exception) { + if (exception != null) { + if (exception instanceof CancelledServerWebExchangeException + || DISCONNECTED_CLIENT_EXCEPTIONS.contains(exception.getClass().getSimpleName())) { + return Outcome.UNKNOWN.asTag(); + } + } Integer statusCode = extractStatusCode(exchange); Outcome outcome = (statusCode != null) ? Outcome.forStatus(statusCode) : Outcome.SUCCESS; return outcome.asTag(); diff --git a/spring-boot-project/spring-boot-actuator/src/test/java/org/springframework/boot/actuate/metrics/web/reactive/server/MetricsWebFilterTests.java b/spring-boot-project/spring-boot-actuator/src/test/java/org/springframework/boot/actuate/metrics/web/reactive/server/MetricsWebFilterTests.java index 7f3008343d..d108c8a1b5 100644 --- a/spring-boot-project/spring-boot-actuator/src/test/java/org/springframework/boot/actuate/metrics/web/reactive/server/MetricsWebFilterTests.java +++ b/spring-boot-project/spring-boot-actuator/src/test/java/org/springframework/boot/actuate/metrics/web/reactive/server/MetricsWebFilterTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2012-2020 the original author or authors. + * Copyright 2012-2021 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -16,6 +16,7 @@ package org.springframework.boot.actuate.metrics.web.reactive.server; +import java.io.EOFException; import java.time.Duration; import io.micrometer.core.instrument.MockClock; @@ -24,6 +25,7 @@ import io.micrometer.core.instrument.simple.SimpleMeterRegistry; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import reactor.core.publisher.Mono; +import reactor.test.StepVerifier; import org.springframework.boot.actuate.metrics.AutoTimer; import org.springframework.mock.http.server.reactive.MockServerHttpRequest; @@ -116,6 +118,32 @@ class MetricsWebFilterTests { assertThat(this.registry.get(REQUEST_METRICS_NAME).tag("status", "200").timer().count()).isEqualTo(2); } + @Test + void cancelledConnectionsShouldProduceMetrics() { + MockServerWebExchange exchange = createExchange("/projects/spring-boot", "/projects/{project}"); + Mono processing = this.webFilter.filter(exchange, + (serverWebExchange) -> exchange.getResponse().setComplete()); + StepVerifier.create(processing).thenCancel().verify(Duration.ofSeconds(5)); + assertMetricsContainsTag("uri", "/projects/{project}"); + assertMetricsContainsTag("status", "200"); + assertMetricsContainsTag("outcome", "UNKNOWN"); + } + + @Test + void disconnectedExceptionShouldProduceMetrics() { + MockServerWebExchange exchange = createExchange("/projects/spring-boot", "/projects/{project}"); + Mono processing = this.webFilter + .filter(exchange, (serverWebExchange) -> Mono.error(new EOFException("Disconnected"))) + .onErrorResume((t) -> { + exchange.getResponse().setRawStatusCode(500); + return exchange.getResponse().setComplete(); + }); + StepVerifier.create(processing).expectComplete().verify(Duration.ofSeconds(5)); + assertMetricsContainsTag("uri", "/projects/{project}"); + assertMetricsContainsTag("status", "500"); + assertMetricsContainsTag("outcome", "UNKNOWN"); + } + private MockServerWebExchange createExchange(String path, String pathPattern) { PathPatternParser parser = new PathPatternParser(); MockServerWebExchange exchange = MockServerWebExchange.from(MockServerHttpRequest.get(path).build()); diff --git a/spring-boot-project/spring-boot-actuator/src/test/java/org/springframework/boot/actuate/metrics/web/reactive/server/WebFluxTagsTests.java b/spring-boot-project/spring-boot-actuator/src/test/java/org/springframework/boot/actuate/metrics/web/reactive/server/WebFluxTagsTests.java index 90c6b36504..023c62957d 100644 --- a/spring-boot-project/spring-boot-actuator/src/test/java/org/springframework/boot/actuate/metrics/web/reactive/server/WebFluxTagsTests.java +++ b/spring-boot-project/spring-boot-actuator/src/test/java/org/springframework/boot/actuate/metrics/web/reactive/server/WebFluxTagsTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2012-2020 the original author or authors. + * Copyright 2012-2021 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -16,6 +16,8 @@ package org.springframework.boot.actuate.metrics.web.reactive.server; +import java.io.EOFException; + import io.micrometer.core.instrument.Tag; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -140,7 +142,7 @@ class WebFluxTagsTests { @Test void outcomeTagIsSuccessWhenResponseStatusIsNull() { this.exchange.getResponse().setStatusCode(null); - Tag tag = WebFluxTags.outcome(this.exchange); + Tag tag = WebFluxTags.outcome(this.exchange, null); assertThat(tag.getValue()).isEqualTo("SUCCESS"); } @@ -153,56 +155,68 @@ class WebFluxTagsTests { given(response.getRawStatusCode()).willReturn(null); given(exchange.getRequest()).willReturn(request); given(exchange.getResponse()).willReturn(response); - Tag tag = WebFluxTags.outcome(exchange); + Tag tag = WebFluxTags.outcome(exchange, null); assertThat(tag.getValue()).isEqualTo("SUCCESS"); } @Test void outcomeTagIsInformationalWhenResponseIs1xx() { this.exchange.getResponse().setStatusCode(HttpStatus.CONTINUE); - Tag tag = WebFluxTags.outcome(this.exchange); + Tag tag = WebFluxTags.outcome(this.exchange, null); assertThat(tag.getValue()).isEqualTo("INFORMATIONAL"); } @Test void outcomeTagIsSuccessWhenResponseIs2xx() { this.exchange.getResponse().setStatusCode(HttpStatus.OK); - Tag tag = WebFluxTags.outcome(this.exchange); + Tag tag = WebFluxTags.outcome(this.exchange, null); assertThat(tag.getValue()).isEqualTo("SUCCESS"); } @Test void outcomeTagIsRedirectionWhenResponseIs3xx() { this.exchange.getResponse().setStatusCode(HttpStatus.MOVED_PERMANENTLY); - Tag tag = WebFluxTags.outcome(this.exchange); + Tag tag = WebFluxTags.outcome(this.exchange, null); assertThat(tag.getValue()).isEqualTo("REDIRECTION"); } @Test void outcomeTagIsClientErrorWhenResponseIs4xx() { this.exchange.getResponse().setStatusCode(HttpStatus.BAD_REQUEST); - Tag tag = WebFluxTags.outcome(this.exchange); + Tag tag = WebFluxTags.outcome(this.exchange, null); assertThat(tag.getValue()).isEqualTo("CLIENT_ERROR"); } @Test void outcomeTagIsServerErrorWhenResponseIs5xx() { this.exchange.getResponse().setStatusCode(HttpStatus.BAD_GATEWAY); - Tag tag = WebFluxTags.outcome(this.exchange); + Tag tag = WebFluxTags.outcome(this.exchange, null); assertThat(tag.getValue()).isEqualTo("SERVER_ERROR"); } @Test void outcomeTagIsClientErrorWhenResponseIsNonStandardInClientSeries() { this.exchange.getResponse().setRawStatusCode(490); - Tag tag = WebFluxTags.outcome(this.exchange); + Tag tag = WebFluxTags.outcome(this.exchange, null); assertThat(tag.getValue()).isEqualTo("CLIENT_ERROR"); } @Test void outcomeTagIsUnknownWhenResponseStatusIsInUnknownSeries() { this.exchange.getResponse().setRawStatusCode(701); - Tag tag = WebFluxTags.outcome(this.exchange); + Tag tag = WebFluxTags.outcome(this.exchange, null); + assertThat(tag.getValue()).isEqualTo("UNKNOWN"); + } + + @Test + void outcomeTagIsClientErrorWhenExceptionIsDisconnectedClient() { + Tag tag = WebFluxTags.outcome(this.exchange, new EOFException("broken pipe")); + assertThat(tag.getValue()).isEqualTo("UNKNOWN"); + } + + @Test + void outcomeTagIsClientErrorWhenExceptionIsCancelledExchange() { + Tag tag = WebFluxTags.outcome(this.exchange, new CancelledServerWebExchangeException()); assertThat(tag.getValue()).isEqualTo("UNKNOWN"); }