Spring/spring-boot
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Auto-configure Observation instrumentation for WebFlux

Browse Source 
Prior to this commit, Spring Boot would offer a specific Metrics
instrumentation for WebFlux applications through a `WebFilter` and
custom Tag providers.

As of Spring Framework 6.0, the Observation instrumentation is done
directly in WebFlux, also with a `WebFilter`. While this allows both
metrics and traces, some features cannot be supported in the same way
with this new infrastructure.
The former `WebFilter` has been removed and the Tagging infrastructure
deprecated in favor of custom Observation conventions. This commit
provides an adapter layer so that developers can refactor their custom
tagging solution to the convention way, during the deprecation phase,
without losing any feature.

Closes gh-32539
This commit is contained in:
Brian Clozel
2022-10-20 15:24:47 +02:00
parent cda63b541f
commit 685fa900f8
19 changed files with 378 additions and 704 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/metrics/web/reactive/server/CancelledServerWebExchangeException.java
View File

@@ -1,29 +0,0 @@
/*
* 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 {
}

4
spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/metrics/web/reactive/server/DefaultWebFluxTagsProvider.java
View File

@@ -30,7 +30,11 @@ import org.springframework.web.server.ServerWebExchange;
* @author Jon Schneider
* @author Andy Wilkinson
* @since 2.0.0
* @deprecated since 3.0.0 for removal in 3.2.0 in favor of
* {@link org.springframework.http.observation.reactive.ServerRequestObservationConvention}
*/
@Deprecated(since = "3.0.0", forRemoval = true)
@SuppressWarnings("removal")
public class DefaultWebFluxTagsProvider implements WebFluxTagsProvider {
private final boolean ignoreTrailingSlash;

128
spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/metrics/web/reactive/server/MetricsWebFilter.java
View File

@@ -1,128 +0,0 @@
/*
* Copyright 2012-2022 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;
import java.util.Collections;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import io.micrometer.core.annotation.Timed;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.Tag;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Mono;
import org.springframework.boot.actuate.metrics.AutoTimer;
import org.springframework.boot.actuate.metrics.annotation.TimedAnnotations;
import org.springframework.boot.web.reactive.error.ErrorAttributes;
import org.springframework.core.Ordered;
import org.springframework.core.annotation.Order;
import org.springframework.http.server.reactive.ServerHttpResponse;
import org.springframework.web.method.HandlerMethod;
import org.springframework.web.reactive.HandlerMapping;
import org.springframework.web.server.ServerWebExchange;
import org.springframework.web.server.WebFilter;
import org.springframework.web.server.WebFilterChain;
/**
* Intercepts incoming HTTP requests handled by Spring WebFlux handlers and records
* metrics about execution time and results.
*
* @author Jon Schneider
* @author Brian Clozel
* @since 2.0.0
*/
@Order(Ordered.HIGHEST_PRECEDENCE + 1)
public class MetricsWebFilter implements WebFilter {
private static Log logger = LogFactory.getLog(MetricsWebFilter.class);
private final MeterRegistry registry;
private final WebFluxTagsProvider tagsProvider;
private final String metricName;
private final AutoTimer autoTimer;
/**
* Create a new {@code MetricsWebFilter}.
* @param registry the registry to which metrics are recorded
* @param tagsProvider provider for metrics tags
* @param metricName name of the metric to record
* @param autoTimer the auto-timers to apply or {@code null} to disable auto-timing
* @since 2.2.0
*/
public MetricsWebFilter(MeterRegistry registry, WebFluxTagsProvider tagsProvider, String metricName,
AutoTimer autoTimer) {
this.registry = registry;
this.tagsProvider = tagsProvider;
this.metricName = metricName;
this.autoTimer = (autoTimer != null) ? autoTimer : AutoTimer.DISABLED;
}
@Override
public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {
return chain.filter(exchange).transformDeferred((call) -> filter(exchange, call));
}
private Publisher<Void> filter(ServerWebExchange exchange, Mono<Void> call) {
long start = System.nanoTime();
return call.doOnEach((signal) -> onTerminalSignal(exchange, signal.getThrowable(), start))
.doOnCancel(() -> onTerminalSignal(exchange, new CancelledServerWebExchangeException(), start));
}
private void onTerminalSignal(ServerWebExchange exchange, Throwable cause, long start) {
ServerHttpResponse response = exchange.getResponse();
if (response.isCommitted() || cause instanceof CancelledServerWebExchangeException) {
record(exchange, cause, start);
}
else {
response.beforeCommit(() -> {
record(exchange, cause, start);
return Mono.empty();
});
}
}
private void record(ServerWebExchange exchange, Throwable cause, long start) {
try {
cause = (cause != null) ? cause : exchange.getAttribute(ErrorAttributes.ERROR_ATTRIBUTE);
Object handler = exchange.getAttribute(HandlerMapping.BEST_MATCHING_HANDLER_ATTRIBUTE);
Set<Timed> annotations = getTimedAnnotations(handler);
Iterable<Tag> tags = this.tagsProvider.httpRequestTags(exchange, cause);
long duration = System.nanoTime() - start;
AutoTimer.apply(this.autoTimer, this.metricName, annotations,
(builder) -> builder.description("Duration of HTTP server request handling").tags(tags)
.register(this.registry).record(duration, TimeUnit.NANOSECONDS));
}
catch (Exception ex) {
logger.warn("Failed to record timer metrics", ex);
// Allow exchange to continue, unaffected by metrics problem
}
}
private Set<Timed> getTimedAnnotations(Object handler) {
if (handler instanceof HandlerMethod handlerMethod) {
return TimedAnnotations.get(handlerMethod.getMethod(), handlerMethod.getBeanType());
}
return Collections.emptySet();
}
}

6
spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/metrics/web/reactive/server/WebFluxTags.java
View File

@@ -40,7 +40,10 @@ import org.springframework.web.util.pattern.PathPattern;
* @author Michael McFadyen
* @author Brian Clozel
* @since 2.0.0
* @deprecated since 3.0.0 for removal in 3.2.0 in favor of
* {@link org.springframework.http.observation.reactive.ServerRequestObservationConvention}
*/
@Deprecated(since = "3.0.0", forRemoval = true)
public final class WebFluxTags {
private static final Tag URI_NOT_FOUND = Tag.of("uri", "NOT_FOUND");
@@ -176,8 +179,7 @@ public final class WebFluxTags {
*/
public static Tag outcome(ServerWebExchange exchange, Throwable exception) {
if (exception != null) {
if (exception instanceof CancelledServerWebExchangeException
|| DISCONNECTED_CLIENT_EXCEPTIONS.contains(exception.getClass().getSimpleName())) {
if (DISCONNECTED_CLIENT_EXCEPTIONS.contains(exception.getClass().getSimpleName())) {
return Outcome.UNKNOWN.asTag();
}
}

3
spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/metrics/web/reactive/server/WebFluxTagsContributor.java
View File

@@ -26,8 +26,11 @@ import org.springframework.web.server.ServerWebExchange;
*
* @author Andy Wilkinson
* @since 2.3.0
* @deprecated since 3.0.0 for removal in 3.2.0 in favor of
* {@link org.springframework.http.observation.reactive.ServerRequestObservationConvention}
*/
@FunctionalInterface
@Deprecated(since = "3.0.0", forRemoval = true)
public interface WebFluxTagsContributor {
/**

3
spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/metrics/web/reactive/server/WebFluxTagsProvider.java
View File

@@ -26,8 +26,11 @@ import org.springframework.web.server.ServerWebExchange;
* @author Jon Schneider
* @author Andy Wilkinson
* @since 2.0.0
* @deprecated since 3.0.0 for removal in 3.2.0 in favor of
* {@link org.springframework.http.observation.reactive.ServerRequestObservationConvention}
*/
@FunctionalInterface
@Deprecated(since = "3.0.0", forRemoval = true)
public interface WebFluxTagsProvider {
/**

1
spring-boot-project/spring-boot-actuator/src/test/java/org/springframework/boot/actuate/metrics/web/reactive/server/DefaultWebFluxTagsProviderTests.java
View File

@@ -37,6 +37,7 @@ import static org.assertj.core.api.Assertions.assertThat;
*
* @author Andy Wilkinson
*/
@SuppressWarnings("removal")
class DefaultWebFluxTagsProviderTests {
@Test

283
spring-boot-project/spring-boot-actuator/src/test/java/org/springframework/boot/actuate/metrics/web/reactive/server/MetricsWebFilterTests.java
View File

@@ -1,283 +0,0 @@
/*
* 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;
import java.io.EOFException;
import java.time.Duration;
import java.util.concurrent.atomic.AtomicBoolean;
import io.micrometer.core.annotation.Timed;
import io.micrometer.core.instrument.MockClock;
import io.micrometer.core.instrument.Tag;
import io.micrometer.core.instrument.simple.SimpleConfig;
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.boot.web.reactive.error.ErrorAttributes;
import org.springframework.mock.http.server.reactive.MockServerHttpRequest;
import org.springframework.mock.web.server.MockServerWebExchange;
import org.springframework.util.ReflectionUtils;
import org.springframework.web.method.HandlerMethod;
import org.springframework.web.reactive.HandlerMapping;
import org.springframework.web.server.ServerWebExchange;
import org.springframework.web.util.pattern.PathPatternParser;
import static org.assertj.core.api.Assertions.assertThat;
/**
* Tests for {@link MetricsWebFilter}
*
* @author Brian Clozel
* @author Madhura Bhave
*/
class MetricsWebFilterTests {
private static final String REQUEST_METRICS_NAME = "http.server.requests";
private static final String REQUEST_METRICS_NAME_PERCENTILE = REQUEST_METRICS_NAME + ".percentile";
private final FaultyWebFluxTagsProvider tagsProvider = new FaultyWebFluxTagsProvider();
private SimpleMeterRegistry registry;
private MetricsWebFilter webFilter;
@BeforeEach
void setup() {
MockClock clock = new MockClock();
this.registry = new SimpleMeterRegistry(SimpleConfig.DEFAULT, clock);
this.webFilter = new MetricsWebFilter(this.registry, this.tagsProvider, REQUEST_METRICS_NAME,
AutoTimer.ENABLED);
}
@Test
void filterAddsTagsToRegistry() {
MockServerWebExchange exchange = createExchange("/projects/spring-boot", "/projects/{project}");
this.webFilter.filter(exchange, (serverWebExchange) -> exchange.getResponse().setComplete())
.block(Duration.ofSeconds(30));
assertMetricsContainsTag("uri", "/projects/{project}");
assertMetricsContainsTag("status", "200");
}
@Test
void filterAddsTagsToRegistryForExceptions() {
MockServerWebExchange exchange = createExchange("/projects/spring-boot", "/projects/{project}");
this.webFilter.filter(exchange, (serverWebExchange) -> Mono.error(new IllegalStateException("test error")))
.onErrorResume((t) -> {
exchange.getResponse().setRawStatusCode(500);
return exchange.getResponse().setComplete();
}).block(Duration.ofSeconds(30));
assertMetricsContainsTag("uri", "/projects/{project}");
assertMetricsContainsTag("status", "500");
assertMetricsContainsTag("exception", "IllegalStateException");
}
@Test
void filterAddsNonEmptyTagsToRegistryForAnonymousExceptions() {
final Exception anonymous = new Exception("test error") {
};
MockServerWebExchange exchange = createExchange("/projects/spring-boot", "/projects/{project}");
this.webFilter.filter(exchange, (serverWebExchange) -> Mono.error(anonymous)).onErrorResume((t) -> {
exchange.getResponse().setRawStatusCode(500);
return exchange.getResponse().setComplete();
}).block(Duration.ofSeconds(30));
assertMetricsContainsTag("uri", "/projects/{project}");
assertMetricsContainsTag("status", "500");
assertMetricsContainsTag("exception", anonymous.getClass().getName());
}
@Test
void filterAddsTagsToRegistryForHandledExceptions() {
MockServerWebExchange exchange = createExchange("/projects/spring-boot", "/projects/{project}");
this.webFilter.filter(exchange, (serverWebExchange) -> {
exchange.getAttributes().put(ErrorAttributes.ERROR_ATTRIBUTE, new IllegalStateException("test error"));
return exchange.getResponse().setComplete();
}).block(Duration.ofSeconds(30));
assertMetricsContainsTag("uri", "/projects/{project}");
assertMetricsContainsTag("status", "200");
assertMetricsContainsTag("exception", "IllegalStateException");
}
@Test
void filterAddsTagsToRegistryForExceptionsAndCommittedResponse() {
MockServerWebExchange exchange = createExchange("/projects/spring-boot", "/projects/{project}");
this.webFilter.filter(exchange, (serverWebExchange) -> {
exchange.getResponse().setRawStatusCode(500);
return exchange.getResponse().setComplete().then(Mono.error(new IllegalStateException("test error")));
}).onErrorResume((t) -> Mono.empty()).block(Duration.ofSeconds(30));
assertMetricsContainsTag("uri", "/projects/{project}");
assertMetricsContainsTag("status", "500");
}
@Test
void trailingSlashShouldNotRecordDuplicateMetrics() {
MockServerWebExchange exchange1 = createExchange("/projects/spring-boot", "/projects/{project}");
MockServerWebExchange exchange2 = createExchange("/projects/spring-boot", "/projects/{project}/");
this.webFilter.filter(exchange1, (serverWebExchange) -> exchange1.getResponse().setComplete())
.block(Duration.ofSeconds(30));
this.webFilter.filter(exchange2, (serverWebExchange) -> exchange2.getResponse().setComplete())
.block(Duration.ofSeconds(30));
assertThat(this.registry.get(REQUEST_METRICS_NAME).tag("uri", "/projects/{project}").timer().count())
.isEqualTo(2);
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<Void> 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<Void> 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");
}
@Test
void filterAddsStandardTags() {
MockServerWebExchange exchange = createTimedHandlerMethodExchange("timed");
this.webFilter.filter(exchange, (serverWebExchange) -> exchange.getResponse().setComplete())
.block(Duration.ofSeconds(30));
assertMetricsContainsTag("uri", "/projects/{project}");
assertMetricsContainsTag("status", "200");
}
@Test
void filterAddsExtraTags() {
MockServerWebExchange exchange = createTimedHandlerMethodExchange("timedExtraTags");
this.webFilter.filter(exchange, (serverWebExchange) -> exchange.getResponse().setComplete())
.block(Duration.ofSeconds(30));
assertMetricsContainsTag("uri", "/projects/{project}");
assertMetricsContainsTag("status", "200");
assertMetricsContainsTag("tag1", "value1");
assertMetricsContainsTag("tag2", "value2");
}
@Test
void filterAddsExtraTagsAndException() {
MockServerWebExchange exchange = createTimedHandlerMethodExchange("timedExtraTags");
this.webFilter.filter(exchange, (serverWebExchange) -> Mono.error(new IllegalStateException("test error")))
.onErrorResume((ex) -> {
exchange.getResponse().setRawStatusCode(500);
return exchange.getResponse().setComplete();
}).block(Duration.ofSeconds(30));
assertMetricsContainsTag("uri", "/projects/{project}");
assertMetricsContainsTag("status", "500");
assertMetricsContainsTag("exception", "IllegalStateException");
assertMetricsContainsTag("tag1", "value1");
assertMetricsContainsTag("tag2", "value2");
}
@Test
void filterAddsPercentileMeters() {
MockServerWebExchange exchange = createTimedHandlerMethodExchange("timedPercentiles");
this.webFilter.filter(exchange, (serverWebExchange) -> exchange.getResponse().setComplete())
.block(Duration.ofSeconds(30));
assertMetricsContainsTag("uri", "/projects/{project}");
assertMetricsContainsTag("status", "200");
assertThat(this.registry.get(REQUEST_METRICS_NAME_PERCENTILE).tag("phi", "0.95").gauge().value()).isNotZero();
assertThat(this.registry.get(REQUEST_METRICS_NAME_PERCENTILE).tag("phi", "0.5").gauge().value()).isNotZero();
}
@Test
void whenMetricsRecordingFailsThenExchangeFilteringSucceeds() {
MockServerWebExchange exchange = createExchange("/projects/spring-boot", "/projects/{project}");
this.tagsProvider.failOnce();
this.webFilter.filter(exchange, (serverWebExchange) -> exchange.getResponse().setComplete())
.block(Duration.ofSeconds(30));
}
private MockServerWebExchange createTimedHandlerMethodExchange(String methodName) {
MockServerWebExchange exchange = createExchange("/projects/spring-boot", "/projects/{project}");
exchange.getAttributes().put(HandlerMapping.BEST_MATCHING_HANDLER_ATTRIBUTE,
new HandlerMethod(this, ReflectionUtils.findMethod(Handlers.class, methodName)));
return exchange;
}
private MockServerWebExchange createExchange(String path, String pathPattern) {
PathPatternParser parser = new PathPatternParser();
MockServerWebExchange exchange = MockServerWebExchange.from(MockServerHttpRequest.get(path).build());
exchange.getAttributes().put(HandlerMapping.BEST_MATCHING_PATTERN_ATTRIBUTE, parser.parse(pathPattern));
return exchange;
}
private void assertMetricsContainsTag(String tagKey, String tagValue) {
assertThat(this.registry.get(REQUEST_METRICS_NAME).tag(tagKey, tagValue).timer().count()).isEqualTo(1);
}
static class Handlers {
@Timed
Mono<String> timed() {
return Mono.just("test");
}
@Timed(extraTags = { "tag1", "value1", "tag2", "value2" })
Mono<String> timedExtraTags() {
return Mono.just("test");
}
@Timed(percentiles = { 0.5, 0.95 })
Mono<String> timedPercentiles() {
return Mono.just("test");
}
}
class FaultyWebFluxTagsProvider extends DefaultWebFluxTagsProvider {
private final AtomicBoolean fail = new AtomicBoolean();
FaultyWebFluxTagsProvider() {
super(true);
}
@Override
public Iterable<Tag> httpRequestTags(ServerWebExchange exchange, Throwable exception) {
if (this.fail.compareAndSet(true, false)) {
throw new RuntimeException();
}
return super.httpRequestTags(exchange, exception);
}
void failOnce() {
this.fail.set(true);
}
}
}

7
spring-boot-project/spring-boot-actuator/src/test/java/org/springframework/boot/actuate/metrics/web/reactive/server/WebFluxTagsTests.java
View File

@@ -44,6 +44,7 @@ import static org.mockito.Mockito.mock;
* @author Madhura Bhave
* @author Stephane Nicoll
*/
@SuppressWarnings("removal")
class WebFluxTagsTests {
private MockServerWebExchange exchange;
@@ -215,10 +216,4 @@ class WebFluxTagsTests {
assertThat(tag.getValue()).isEqualTo("UNKNOWN");
}
@Test
void outcomeTagIsUnknownWhenExceptionIsCancelledExchange() {
Tag tag = WebFluxTags.outcome(this.exchange, new CancelledServerWebExchangeException());
assertThat(tag.getValue()).isEqualTo("UNKNOWN");
}
}