Stop observations for async requests in Servlet filter

Prior to this commit, the `ServerHttpObservationFilter` would support
async dispatches and would do the following:

1. start the observation
2. call the filter chain
3. if async has started, do nothing
4. if not in async mode, stop the observation

This behavior would effectively rely on Async implementations to
complete and dispatch the request back to the container for an async
dispatch. This is what Spring web frameworks do and guarantee.

Some implementations complete the async request but do not dispatch
back; as a result, observations could leak as they are never stopped.

This commit changes the support of async requests. The filter now
opts-out of async dispatches - the filter will not be called for those
anymore. Instead, if the application started async mode during the
initial container dispatch, the filter will register an AsyncListener to
be notified of the outcome of the async handling.

Fixes gh-32986

Author: bclozel

spring-web/src/main/java/org/springframework/web/filter/ServerHttpObservationFilter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-2024 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.
@@ -21,9 +21,12 @@
 
 import io.micrometer.observation.Observation;
 import io.micrometer.observation.ObservationRegistry;
+import jakarta.servlet.AsyncEvent;
+import jakarta.servlet.AsyncListener;
 import jakarta.servlet.FilterChain;
 import jakarta.servlet.RequestDispatcher;
 import jakarta.servlet.ServletException;
+import jakarta.servlet.ServletRequest;
 import jakarta.servlet.http.HttpServletRequest;
 import jakarta.servlet.http.HttpServletResponse;
 
@@ -94,11 +97,6 @@ public static Optional<ServerRequestObservationContext> findObservationContext(H
 		return Optional.ofNullable((ServerRequestObservationContext) request.getAttribute(CURRENT_OBSERVATION_CONTEXT_ATTRIBUTE));
 	}
 
-	@Override
-	protected boolean shouldNotFilterAsyncDispatch() {
-		return false;
-	}
-
 	@Override
 	@SuppressWarnings("try")
 	protected void doFilterInternal(HttpServletRequest request, HttpServletResponse response, FilterChain filterChain)
@@ -114,8 +112,12 @@ protected void doFilterInternal(HttpServletRequest request, HttpServletResponse
 			throw ex;
 		}
 		finally {
-			// Only stop Observation if async processing is done or has never been started.
-			if (!request.isAsyncStarted()) {
+			// If async is started, register a listener for completion notification.
+			if (request.isAsyncStarted()) {
+				request.getAsyncContext().addListener(new ObservationAsyncListener(observation));
+			}
+			// Stop Observation right now if async processing has not been started.
+			else {
 				Throwable error = fetchException(request);
 				if (error != null) {
 					observation.error(error);
@@ -139,13 +141,44 @@ private Observation createOrFetchObservation(HttpServletRequest request, HttpSer
 		return observation;
 	}
 
-	private Throwable unwrapServletException(Throwable ex) {
+	@Nullable
+	static Throwable unwrapServletException(Throwable ex) {
 		return (ex instanceof ServletException) ? ex.getCause() : ex;
 	}
 
 	@Nullable
-	private Throwable fetchException(HttpServletRequest request) {
+	static Throwable fetchException(ServletRequest request) {
 		return (Throwable) request.getAttribute(RequestDispatcher.ERROR_EXCEPTION);
 	}
 
+	private static class ObservationAsyncListener implements AsyncListener {
+
+		private final Observation currentObservation;
+
+		public ObservationAsyncListener(Observation currentObservation) {
+			this.currentObservation = currentObservation;
+		}
+
+		@Override
+		public void onStartAsync(AsyncEvent event) {
+		}
+
+		@Override
+		public void onTimeout(AsyncEvent event) {
+			this.currentObservation.stop();
+		}
+
+		@Override
+		public void onComplete(AsyncEvent event) {
+			this.currentObservation.stop();
+		}
+
+		@Override
+		public void onError(AsyncEvent event) {
+			this.currentObservation.error(unwrapServletException(event.getThrowable()));
+			this.currentObservation.stop();
+		}
+
+	}
+
 }

spring-web/src/test/java/org/springframework/web/filter/ServerHttpObservationFilterTests.java

@@ -50,6 +50,11 @@ class ServerHttpObservationFilterTests {
 	private final MockHttpServletResponse response = new MockHttpServletResponse();
 
 
+	@Test
+	void filterShouldNotProcessAsyncDispatch() {
+		assertThat(this.filter.shouldNotFilterAsyncDispatch()).isTrue();
+	}
+
 	@Test
 	void filterShouldFillObservationContext() throws Exception {
 		this.filter.doFilter(this.request, this.response, this.mockFilterChain);
@@ -60,7 +65,7 @@ void filterShouldFillObservationContext() throws Exception {
 		assertThat(context.getCarrier()).isEqualTo(this.request);
 		assertThat(context.getResponse()).isEqualTo(this.response);
 		assertThat(context.getPathPattern()).isNull();
-		assertThatHttpObservation().hasLowCardinalityKeyValue("outcome", "SUCCESS");
+		assertThatHttpObservation().hasLowCardinalityKeyValue("outcome", "SUCCESS").hasBeenStopped();
 	}
 
 	@Test
@@ -111,6 +116,16 @@ void filterShouldSetDefaultErrorStatusForBubblingExceptions() {
 				.hasLowCardinalityKeyValue("status", "500");
 	}
 
+	@Test
+	void shouldCloseObservationAfterAsyncCompletion() throws Exception {
+		this.request.setAsyncSupported(true);
+		this.request.startAsync();
+		this.filter.doFilter(this.request, this.response, this.mockFilterChain);
+		this.request.getAsyncContext().complete();
+
+		assertThatHttpObservation().hasLowCardinalityKeyValue("outcome", "SUCCESS").hasBeenStopped();
+	}
+
 	private TestObservationRegistryAssert.TestObservationRegistryAssertReturningObservationContextAssert assertThatHttpObservation() {
 		return TestObservationRegistryAssert.assertThat(this.observationRegistry)
 				.hasObservationWithNameEqualTo("http.server.requests").that();