Instrument reactive servers for Observability

This commit introduces a `HttpRequestsObservationWebFilter` which
instruments web frameworks using Spring's reactive `ServerHttpRequest`
and `ServerHttpResponse` interfaces.

This replaces Spring Boot's `MetricsWebFilter`.

See gh-28880

spring-web/src/main/java/org/springframework/web/observation/reactive/DefaultHttpRequestsObservationConvention.java

@@ -0,0 +1,145 @@
+/*
+ * Copyright 2002-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.web.observation.reactive;
+
+import io.micrometer.common.KeyValue;
+import io.micrometer.common.KeyValues;
+
+import org.springframework.http.HttpStatus;
+import org.springframework.web.util.pattern.PathPattern;
+
+/**
+ * Default {@link HttpRequestsObservationConvention}.
+ *
+ * @author Brian Clozel
+ * @since 6.0
+ */
+public class DefaultHttpRequestsObservationConvention implements HttpRequestsObservationConvention {
+
+	private static final String DEFAULT_NAME = "http.server.requests";
+
+	private static final KeyValue METHOD_UNKNOWN = KeyValue.of(HttpRequestsObservation.LowCardinalityKeyNames.METHOD, "UNKNOWN");
+
+	private static final KeyValue STATUS_UNKNOWN = KeyValue.of(HttpRequestsObservation.LowCardinalityKeyNames.STATUS, "UNKNOWN");
+
+	private static final KeyValue URI_UNKNOWN = KeyValue.of(HttpRequestsObservation.LowCardinalityKeyNames.URI, "UNKNOWN");
+
+	private static final KeyValue URI_ROOT = KeyValue.of(HttpRequestsObservation.LowCardinalityKeyNames.URI, "root");
+
+	private static final KeyValue URI_NOT_FOUND = KeyValue.of(HttpRequestsObservation.LowCardinalityKeyNames.URI, "NOT_FOUND");
+
+	private static final KeyValue URI_REDIRECTION = KeyValue.of(HttpRequestsObservation.LowCardinalityKeyNames.URI, "REDIRECTION");
+
+	private static final KeyValue EXCEPTION_NONE = KeyValue.of(HttpRequestsObservation.LowCardinalityKeyNames.EXCEPTION, "none");
+
+	private static final KeyValue OUTCOME_UNKNOWN = KeyValue.of(HttpRequestsObservation.LowCardinalityKeyNames.OUTCOME, "UNKNOWN");
+
+	private static final KeyValue URI_EXPANDED_UNKNOWN = KeyValue.of(HttpRequestsObservation.HighCardinalityKeyNames.URI_EXPANDED, "UNKNOWN");
+
+	private final String name;
+
+	/**
+	 * Create a convention with the default name {@code "http.server.requests"}.
+	 */
+	public DefaultHttpRequestsObservationConvention() {
+		this(DEFAULT_NAME);
+	}
+
+	/**
+	 * Create a convention with a custom name.
+	 *
+	 * @param name the observation name
+	 */
+	public DefaultHttpRequestsObservationConvention(String name) {
+		this.name = name;
+	}
+
+	@Override
+	public String getName() {
+		return this.name;
+	}
+
+	@Override
+	public KeyValues getLowCardinalityKeyValues(HttpRequestsObservationContext context) {
+		return KeyValues.of(method(context), uri(context), status(context), exception(context), outcome(context));
+	}
+
+	@Override
+	public KeyValues getHighCardinalityKeyValues(HttpRequestsObservationContext context) {
+		return KeyValues.of(uriExpanded(context));
+	}
+
+	protected KeyValue method(HttpRequestsObservationContext context) {
+		return (context.getCarrier() != null) ? KeyValue.of(HttpRequestsObservation.LowCardinalityKeyNames.METHOD, context.getCarrier().getMethod().name()) : METHOD_UNKNOWN;
+	}
+
+	protected KeyValue status(HttpRequestsObservationContext context) {
+		return (context.getResponse() != null) ? KeyValue.of(HttpRequestsObservation.LowCardinalityKeyNames.STATUS, Integer.toString(context.getResponse().getStatusCode().value())) : STATUS_UNKNOWN;
+	}
+
+	protected KeyValue uri(HttpRequestsObservationContext context) {
+		if (context.getCarrier() != null) {
+			PathPattern pattern = context.getPathPattern();
+			if (pattern != null) {
+				if (pattern.toString().isEmpty()) {
+					return URI_ROOT;
+				}
+				return KeyValue.of("uri", pattern.toString());
+			}
+			if (context.getResponse() != null) {
+				HttpStatus status = HttpStatus.resolve(context.getResponse().getStatusCode().value());
+				if (status != null) {
+					if (status.is3xxRedirection()) {
+						return URI_REDIRECTION;
+					}
+					if (status == HttpStatus.NOT_FOUND) {
+						return URI_NOT_FOUND;
+					}
+				}
+			}
+		}
+		return URI_UNKNOWN;
+	}
+
+	protected KeyValue exception(HttpRequestsObservationContext context) {
+		return context.getError().map(throwable ->
+						KeyValue.of(HttpRequestsObservation.LowCardinalityKeyNames.EXCEPTION, throwable.getClass().getSimpleName()))
+				.orElse(EXCEPTION_NONE);
+	}
+
+	protected KeyValue outcome(HttpRequestsObservationContext context) {
+		if (context.isConnectionAborted()) {
+			return OUTCOME_UNKNOWN;
+		}
+		else if (context.getResponse() != null) {
+			HttpStatus status = HttpStatus.resolve(context.getResponse().getStatusCode().value());
+			if (status != null) {
+				return KeyValue.of(HttpRequestsObservation.LowCardinalityKeyNames.OUTCOME, status.series().name());
+			}
+		}
+		return OUTCOME_UNKNOWN;
+	}
+
+	protected KeyValue uriExpanded(HttpRequestsObservationContext context) {
+		if (context.getCarrier() != null) {
+			String uriExpanded = context.getCarrier().getPath().toString();
+			return KeyValue.of(HttpRequestsObservation.HighCardinalityKeyNames.URI_EXPANDED, uriExpanded);
+		}
+		return URI_EXPANDED_UNKNOWN;
+	}
+
+}

spring-web/src/main/java/org/springframework/web/observation/reactive/HttpRequestsObservation.java

@@ -0,0 +1,124 @@
+/*
+ * Copyright 2002-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.web.observation.reactive;
+
+import io.micrometer.common.docs.KeyName;
+import io.micrometer.observation.Observation;
+import io.micrometer.observation.ObservationConvention;
+import io.micrometer.observation.docs.DocumentedObservation;
+
+/**
+ * Documented {@link io.micrometer.common.KeyValue KeyValues} for the HTTP server observations
+ * for Servlet-based web applications.
+ * <p>This class is used by automated tools to document KeyValues attached to the HTTP server observations.
+ * @author Brian Clozel
+ * @since 6.0
+ */
+public enum HttpRequestsObservation implements DocumentedObservation {
+
+	/**
+	 * HTTP server request observations.
+	 */
+	HTTP_REQUESTS {
+		@Override
+		public Class<? extends ObservationConvention<? extends Observation.Context>> getDefaultConvention() {
+			return DefaultHttpRequestsObservationConvention.class;
+		}
+
+		@Override
+		public KeyName[] getLowCardinalityKeyNames() {
+			return LowCardinalityKeyNames.values();
+		}
+
+		@Override
+		public KeyName[] getHighCardinalityKeyNames() {
+			return HighCardinalityKeyNames.values();
+		}
+
+	};
+
+	public enum LowCardinalityKeyNames implements KeyName {
+
+		/**
+		 * Name of HTTP request method or {@code "None"} if the request was not received properly.
+		 */
+		METHOD {
+			@Override
+			public String asString() {
+				return "method";
+			}
+
+		},
+
+		/**
+		 * HTTP response raw status code, or {@code "STATUS_UNKNOWN"} if no response was created.
+		 */
+		STATUS {
+			@Override
+			public String asString() {
+				return "status";
+			}
+		},
+
+		/**
+		 * URI pattern for the matching handler if available, falling back to {@code REDIRECTION} for 3xx responses,
+		 * {@code NOT_FOUND} for 404 responses, {@code root} for requests with no path info,
+		 * and {@code UNKNOWN} for all other requests.
+		 */
+		URI {
+			@Override
+			public String asString() {
+				return "uri";
+			}
+		},
+
+		/**
+		 * Name of the exception thrown during the exchange, or {@code "None"} if no exception happened.
+		 */
+		EXCEPTION {
+			@Override
+			public String asString() {
+				return "exception";
+			}
+		},
+
+		/**
+		 * Outcome of the HTTP server exchange.
+		 * @see org.springframework.http.HttpStatus.Series
+		 */
+		OUTCOME {
+			@Override
+			public String asString() {
+				return "outcome";
+			}
+		}
+	}
+
+	public enum HighCardinalityKeyNames implements KeyName {
+
+		/**
+		 * HTTP request URI.
+		 */
+		URI_EXPANDED {
+			@Override
+			public String asString() {
+				return "uri.expanded";
+			}
+		}
+
+	}
+}

spring-web/src/main/java/org/springframework/web/observation/reactive/HttpRequestsObservationContext.java

@@ -0,0 +1,81 @@
+/*
+ * Copyright 2002-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.web.observation.reactive;
+
+import io.micrometer.observation.transport.RequestReplyReceiverContext;
+
+import org.springframework.http.server.reactive.ServerHttpRequest;
+import org.springframework.http.server.reactive.ServerHttpResponse;
+import org.springframework.lang.Nullable;
+import org.springframework.web.server.ServerWebExchange;
+import org.springframework.web.util.pattern.PathPattern;
+
+/**
+ * Context that holds information for metadata collection during observations for reactive web applications.
+ * <p>This context also extends {@link RequestReplyReceiverContext} for propagating
+ * tracing information with the HTTP server exchange.
+ * @author Brian Clozel
+ * @since 6.0
+ */
+public class HttpRequestsObservationContext extends RequestReplyReceiverContext<ServerHttpRequest, ServerHttpResponse> {
+
+	@Nullable
+	private PathPattern pathPattern;
+
+	private boolean connectionAborted;
+
+	public HttpRequestsObservationContext(ServerWebExchange exchange) {
+		super((request, key) -> request.getHeaders().getFirst(key));
+		this.setCarrier(exchange.getRequest());
+		this.setResponse(exchange.getResponse());
+	}
+
+	/**
+	 * Return the path pattern for the handler that matches the current request.
+	 * For example, {@code "/projects/{name}"}.
+	 * <p>Path patterns must have a low cardinality for the entire application.
+	 * @return the path pattern, or {@code null} if none found
+	 */
+	@Nullable
+	public PathPattern getPathPattern() {
+		return this.pathPattern;
+	}
+
+	/**
+	 * Set the path pattern for the handler that matches the current request.
+	 * <p>Path patterns must have a low cardinality for the entire application.
+	 * @param pathPattern the path pattern, for example {@code "/projects/{name}"}.
+	 */
+	public void setPathPattern(@Nullable PathPattern pathPattern) {
+		this.pathPattern = pathPattern;
+	}
+
+	/**
+	 * Whether the current connection was aborted by the client, resulting
+	 * in a {@link reactor.core.publisher.SignalType#CANCEL cancel signal} on te reactive chain,
+	 * or an {@code AbortedException} when reading the request.
+	 * @return if the connection has been aborted
+	 */
+	public boolean isConnectionAborted() {
+		return this.connectionAborted;
+	}
+
+	void setConnectionAborted(boolean connectionAborted) {
+		this.connectionAborted = connectionAborted;
+	}
+
+}

spring-web/src/main/java/org/springframework/web/observation/reactive/HttpRequestsObservationConvention.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2002-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.web.observation.reactive;
+
+import io.micrometer.observation.Observation;
+import io.micrometer.observation.ObservationConvention;
+
+/**
+ * Interface for an {@link ObservationConvention} related to reactive HTTP exchanges.
+ * @author Brian Clozel
+ * @since 6.0
+ */
+public interface HttpRequestsObservationConvention extends ObservationConvention<HttpRequestsObservationContext> {
+
+	@Override
+	default boolean supportsContext(Observation.Context context) {
+		return context instanceof HttpRequestsObservationContext;
+	}
+
+}

spring-web/src/main/java/org/springframework/web/observation/reactive/HttpRequestsObservationWebFilter.java

@@ -0,0 +1,132 @@
+/*
+ * Copyright 2002-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.web.observation.reactive;
+
+import java.util.Optional;
+import java.util.Set;
+
+import io.micrometer.observation.Observation;
+import io.micrometer.observation.ObservationRegistry;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Mono;
+
+import org.springframework.http.server.reactive.ServerHttpResponse;
+import org.springframework.web.server.ServerWebExchange;
+import org.springframework.web.server.WebFilter;
+import org.springframework.web.server.WebFilterChain;
+
+
+/**
+ * {@link org.springframework.web.server.WebFilter} that creates {@link Observation observations}
+ * for HTTP exchanges. This collects information about the execution time and
+ * information gathered from the {@link HttpRequestsObservationContext}.
+ * <p>Web Frameworks can fetch the current {@link HttpRequestsObservationContext context}
+ * as a {@link #CURRENT_OBSERVATION_CONTEXT_ATTRIBUTE request attribute} and contribute
+ * additional information to it.
+ * The configured {@link HttpRequestsObservationConvention} will use this context to collect
+ * {@link io.micrometer.common.KeyValue metadata} and attach it to the observation.
+ *
+ * @author Brian Clozel
+ * @since 6.0
+ */
+public class HttpRequestsObservationWebFilter implements WebFilter {
+
+	/**
+	 * Name of the request attribute holding the {@link HttpRequestsObservationContext context} for the current observation.
+	 */
+	public static final String CURRENT_OBSERVATION_CONTEXT_ATTRIBUTE = HttpRequestsObservationWebFilter.class.getName() + ".context";
+
+	private static final HttpRequestsObservationConvention DEFAULT_OBSERVATION_CONVENTION = new DefaultHttpRequestsObservationConvention();
+
+	private static final Set<String> DISCONNECTED_CLIENT_EXCEPTIONS = Set.of("AbortedException",
+			"ClientAbortException", "EOFException", "EofException");
+
+	private final ObservationRegistry observationRegistry;
+
+	private final HttpRequestsObservationConvention observationConvention;
+
+	/**
+	 * Create a {@code HttpRequestsObservationWebFilter} that records observations
+	 * against the given {@link ObservationRegistry}. The default
+	 * {@link DefaultHttpRequestsObservationConvention convention} will be used.
+	 * @param observationRegistry the registry to use for recording observations
+	 */
+	public HttpRequestsObservationWebFilter(ObservationRegistry observationRegistry) {
+		this(observationRegistry, new DefaultHttpRequestsObservationConvention());
+	}
+
+	/**
+	 * Create a {@code HttpRequestsObservationWebFilter} that records observations
+	 * against the given {@link ObservationRegistry} with a custom convention.
+	 * @param observationRegistry the registry to use for recording observations
+	 * @param observationConvention the convention to use for all recorded observations
+	 */
+	public HttpRequestsObservationWebFilter(ObservationRegistry observationRegistry, HttpRequestsObservationConvention observationConvention) {
+		this.observationRegistry = observationRegistry;
+		this.observationConvention = observationConvention;
+	}
+
+	/**
+	 * Get the current {@link HttpRequestsObservationContext observation context} from the given request, if available.
+	 * @param exchange the current exchange
+	 * @return the current observation context
+	 */
+	public static Optional<HttpRequestsObservationContext> findObservationContext(ServerWebExchange exchange) {
+		return Optional.ofNullable(exchange.getAttribute(CURRENT_OBSERVATION_CONTEXT_ATTRIBUTE));
+	}
+
+	@Override
+	public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {
+		HttpRequestsObservationContext observationContext = new HttpRequestsObservationContext(exchange);
+		exchange.getAttributes().put(CURRENT_OBSERVATION_CONTEXT_ATTRIBUTE, observationContext);
+		return chain.filter(exchange).transformDeferred(call -> filter(exchange, observationContext, call));
+	}
+
+	private Publisher<Void> filter(ServerWebExchange exchange, HttpRequestsObservationContext observationContext, Mono<Void> call) {
+		Observation observation = Observation.createNotStarted(this.observationConvention, DEFAULT_OBSERVATION_CONVENTION,
+				observationContext, this.observationRegistry);
+		observation.start();
+		return call.doOnEach(signal -> {
+					Throwable throwable = signal.getThrowable();
+					if (throwable != null) {
+						if (DISCONNECTED_CLIENT_EXCEPTIONS.contains(throwable.getClass().getSimpleName())) {
+							observationContext.setConnectionAborted(true);
+						}
+						observationContext.setError(throwable);
+					}
+					onTerminalSignal(observation, exchange);
+				})
+				.doOnCancel(() -> {
+					observationContext.setConnectionAborted(true);
+					observation.stop();
+				});
+	}
+
+	private void onTerminalSignal(Observation observation, ServerWebExchange exchange) {
+		ServerHttpResponse response = exchange.getResponse();
+		if (response.isCommitted()) {
+			observation.stop();
+		}
+		else {
+			response.beforeCommit(() -> {
+				observation.stop();
+				return Mono.empty();
+			});
+		}
+	}
+
+}

spring-web/src/main/java/org/springframework/web/observation/reactive/package-info.java

@@ -0,0 +1,9 @@
+/**
+ * Instrumentation for {@link io.micrometer.observation.Observation observing} reactive web applications.
+ */
+@NonNullApi
+@NonNullFields
+package org.springframework.web.observation.reactive;
+
+import org.springframework.lang.NonNullApi;
+import org.springframework.lang.NonNullFields;