Extend observations to RestClient ResponseSpec

Prior to this commit, the `RestClient` observations would be stopped as
soon as the exchange function was called. This means that all errors
related to response decoding or mapping would not be recorded by the
obsevations.

This commit extends the observation recording to the `ResponseSpec` DSL
calls as well as custom exchange functions.

Fixes gh-32575

spring-web/src/main/java/org/springframework/web/client/DefaultRestClient.java

@@ -192,7 +192,7 @@ final class DefaultRestClient implements RestClient {
 	@Nullable
 	@SuppressWarnings({"rawtypes", "unchecked"})
 	private <T> T readWithMessageConverters(ClientHttpResponse clientResponse, Runnable callback, Type bodyType,
-			Class<T> bodyClass) {
+			Class<T> bodyClass, @Nullable Observation observation) {
 
 		MediaType contentType = getContentType(clientResponse);
 
@@ -220,9 +220,13 @@ final class DefaultRestClient implements RestClient {
 					return (T) messageConverter.read((Class)bodyClass, responseWrapper);
 				}
 			}
-			throw new UnknownContentTypeException(bodyType, contentType,
+			UnknownContentTypeException unknownContentTypeException = new UnknownContentTypeException(bodyType, contentType,
 					responseWrapper.getStatusCode(), responseWrapper.getStatusText(),
 					responseWrapper.getHeaders(), RestClientUtils.getBody(responseWrapper));
+			if (observation != null) {
+				observation.error(unknownContentTypeException);
+			}
+			throw unknownContentTypeException;
 		}
 		catch (UncheckedIOException | IOException | HttpMessageNotReadableException ex) {
 			Throwable cause;
@@ -232,8 +236,17 @@ final class DefaultRestClient implements RestClient {
 			else {
 				cause = ex;
 			}
-			throw new RestClientException("Error while extracting response for type [" +
+			RestClientException restClientException = new RestClientException("Error while extracting response for type [" +
 					ResolvableType.forType(bodyType) + "] and content type [" + contentType + "]", cause);
+			if (observation != null) {
+				observation.error(restClientException);
+			}
+			throw restClientException;
+		}
+		finally {
+			if (observation != null) {
+				observation.stop();
+			}
 		}
 	}
 
@@ -475,28 +488,30 @@ final class DefaultRestClient implements RestClient {
 				}
 				clientResponse = clientRequest.execute();
 				observationContext.setResponse(clientResponse);
-				ConvertibleClientHttpResponse convertibleWrapper = new DefaultConvertibleClientHttpResponse(clientResponse);
+				ConvertibleClientHttpResponse convertibleWrapper = new DefaultConvertibleClientHttpResponse(clientResponse, observation);
 				return exchangeFunction.exchange(clientRequest, convertibleWrapper);
 			}
 			catch (IOException ex) {
 				ResourceAccessException resourceAccessException = createResourceAccessException(uri, this.httpMethod, ex);
 				if (observation != null) {
 					observation.error(resourceAccessException);
+					observation.stop();
 				}
 				throw resourceAccessException;
 			}
 			catch (Throwable error) {
 				if (observation != null) {
 					observation.error(error);
+					observation.stop();
 				}
 				throw error;
 			}
 			finally {
 				if (close && clientResponse != null) {
 					clientResponse.close();
-				}
+					if (observation != null) {
-				if (observation != null) {
+						observation.stop();
-					observation.stop();
+					}
 				}
 			}
 		}
@@ -665,7 +680,7 @@ final class DefaultRestClient implements RestClient {
 		@Nullable
 		private <T> T readBody(Type bodyType, Class<T> bodyClass) {
 			return DefaultRestClient.this.readWithMessageConverters(this.clientResponse, this::applyStatusHandlers,
-					bodyType, bodyClass);
+					bodyType, bodyClass, getCurrentObservation());
 
 		}
 
@@ -686,6 +701,15 @@ final class DefaultRestClient implements RestClient {
 				throw new UncheckedIOException(ex);
 			}
 		}
+
+		@Nullable
+		private Observation getCurrentObservation() {
+			if (this.clientResponse instanceof DefaultConvertibleClientHttpResponse convertibleResponse) {
+				return convertibleResponse.observation;
+			}
+			return null;
+		}
+
 	}
 
 
@@ -693,16 +717,19 @@ final class DefaultRestClient implements RestClient {
 
 		private final ClientHttpResponse delegate;
 
+		private final Observation observation;
 
-		public DefaultConvertibleClientHttpResponse(ClientHttpResponse delegate) {
+
+		public DefaultConvertibleClientHttpResponse(ClientHttpResponse delegate, Observation observation) {
 			this.delegate = delegate;
+			this.observation = observation;
 		}
 
 
 		@Nullable
 		@Override
 		public <T> T bodyTo(Class<T> bodyType) {
-			return readWithMessageConverters(this.delegate, () -> {} , bodyType, bodyType);
+			return readWithMessageConverters(this.delegate, () -> {} , bodyType, bodyType, this.observation);
 		}
 
 		@Nullable
@@ -710,7 +737,7 @@ final class DefaultRestClient implements RestClient {
 		public <T> T bodyTo(ParameterizedTypeReference<T> bodyType) {
 			Type type = bodyType.getType();
 			Class<T> bodyClass = bodyClass(type);
-			return readWithMessageConverters(this.delegate, () -> {} , type, bodyClass);
+			return readWithMessageConverters(this.delegate, () -> {}, type, bodyClass, this.observation);
 		}
 
 		@Override
@@ -736,6 +763,7 @@ final class DefaultRestClient implements RestClient {
 		@Override
 		public void close() {
 			this.delegate.close();
+			this.observation.stop();
 		}
 
 	}

spring-web/src/test/java/org/springframework/web/client/RestClientObservationTests.java

@@ -19,6 +19,7 @@ package org.springframework.web.client;
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.net.URI;
+import java.nio.charset.StandardCharsets;
 import java.util.Map;
 import java.util.UUID;
 
@@ -40,6 +41,7 @@ import org.springframework.http.client.observation.ClientRequestObservationConte
 import org.springframework.http.client.observation.ClientRequestObservationConvention;
 import org.springframework.http.client.observation.DefaultClientRequestObservationConvention;
 import org.springframework.http.converter.HttpMessageConverter;
+import org.springframework.util.StreamUtils;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
@@ -84,7 +86,7 @@ class RestClientObservationTests {
 	}
 
 	@Test
-	void executeVarArgsAddsUriTemplateAsKeyValue() throws Exception {
+	void shouldContributeTemplateWhenUriVariables() throws Exception {
 		mockSentRequest(GET, "https://example.com/hotels/42/bookings/21");
 		mockResponseStatus(HttpStatus.OK);
 
@@ -95,7 +97,7 @@ class RestClientObservationTests {
 	}
 
 	@Test
-	void executeArgsMapAddsUriTemplateAsKeyValue() throws Exception {
+	void shouldContributeTemplateWhenMap() throws Exception {
 		mockSentRequest(GET, "https://example.com/hotels/42/bookings/21");
 		mockResponseStatus(HttpStatus.OK);
 
@@ -107,9 +109,8 @@ class RestClientObservationTests {
 		assertThatHttpObservation().hasLowCardinalityKeyValue("uri", "/hotels/{hotel}/bookings/{booking}");
 	}
 
-
 	@Test
-	void executeAddsSuccessAsOutcome() throws Exception {
+	void shouldContributeSuccessOutcome() throws Exception {
 		mockSentRequest(GET, "https://example.org");
 		mockResponseStatus(HttpStatus.OK);
 		mockResponseBody("Hello World", MediaType.TEXT_PLAIN);
@@ -120,7 +121,7 @@ class RestClientObservationTests {
 	}
 
 	@Test
-	void executeAddsServerErrorAsOutcome() throws Exception {
+	void shouldContributeServerErrorOutcome() throws Exception {
 		String url = "https://example.org";
 		mockSentRequest(GET, url);
 		mockResponseStatus(HttpStatus.INTERNAL_SERVER_ERROR);
@@ -134,7 +135,7 @@ class RestClientObservationTests {
 	}
 
 	@Test
-	void executeAddsExceptionAsKeyValue() throws Exception {
+	void shouldContributeDecodingError() throws Exception {
 		mockSentRequest(POST, "https://example.org/resource");
 		mockResponseStatus(HttpStatus.OK);
 
@@ -150,7 +151,7 @@ class RestClientObservationTests {
 	}
 
 	@Test
-	void executeWithIoExceptionAddsUnknownOutcome() throws Exception {
+	void shouldContributeIOError() throws Exception {
 		String url = "https://example.org/resource";
 		mockSentRequest(GET, url);
 		given(request.execute()).willThrow(new IOException("Socket failure"));
@@ -161,7 +162,7 @@ class RestClientObservationTests {
 	}
 
 	@Test
-	void executeWithCustomConventionUsesCustomObservationName() throws Exception {
+	void shouldUseCustomConvention() throws Exception {
 		ClientRequestObservationConvention observationConvention =
 				new DefaultClientRequestObservationConvention("custom.requests");
 		RestClient restClient = this.client.mutate().observationConvention(observationConvention).build();
@@ -174,6 +175,56 @@ class RestClientObservationTests {
 				.hasObservationWithNameEqualTo("custom.requests");
 	}
 
+	@Test
+	void shouldAddClientDecodingErrorAsException() throws Exception {
+		String url = "https://example.org";
+		mockSentRequest(GET, url);
+		mockResponseStatus(HttpStatus.OK);
+		mockResponseBody("INVALID", MediaType.APPLICATION_JSON);
+
+		assertThatExceptionOfType(RestClientException.class).isThrownBy(() ->
+				client.get().uri(url).retrieve().body(User.class));
+
+		assertThatHttpObservation().hasLowCardinalityKeyValue("exception", "RestClientException");
+	}
+
+	@Test
+	void shouldAddUnknownContentTypeErrorAsException() throws Exception {
+		String url = "https://example.org";
+		mockSentRequest(GET, url);
+		mockResponseStatus(HttpStatus.OK);
+		mockResponseBody("Not Found", MediaType.TEXT_HTML);
+
+		assertThatExceptionOfType(RestClientException.class).isThrownBy(() ->
+				client.get().uri(url).retrieve().body(User.class));
+
+		assertThatHttpObservation().hasLowCardinalityKeyValue("exception", "UnknownContentTypeException");
+	}
+
+	@Test
+	void registerObservationWhenReadingBody() throws Exception {
+		mockSentRequest(GET, "https://example.org");
+		mockResponseStatus(HttpStatus.OK);
+		mockResponseBody("Hello World", MediaType.TEXT_PLAIN);
+
+		client.get().uri("https://example.org").exchange((request, response) -> response.bodyTo(String.class));
+		assertThatHttpObservation().hasLowCardinalityKeyValue("outcome", "SUCCESS");
+	}
+
+	@Test
+	void registerObservationWhenReadingStream() throws Exception {
+		mockSentRequest(GET, "https://example.org");
+		mockResponseStatus(HttpStatus.OK);
+		mockResponseBody("Hello World", MediaType.TEXT_PLAIN);
+
+		client.get().uri("https://example.org").exchange((request, response) -> {
+			String result = StreamUtils.copyToString(response.getBody(), StandardCharsets.UTF_8);
+			response.close();
+			return result;
+		}, false);
+		assertThatHttpObservation().hasLowCardinalityKeyValue("outcome", "SUCCESS");
+	}
+
 
 	private void mockSentRequest(HttpMethod method, String uri) throws Exception {
 		mockSentRequest(method, uri, new HttpHeaders());
@@ -220,4 +271,8 @@ class RestClientObservationTests {
 		}
 	}
 
+	record User(String name) {
+
+	}
+
 }