Propagate context in reactive HTTP server and client

Prior to this commit, the ServerHttpObservationFilter would not add
the current observation as a key in the Reactor context, preventing
from being used or propagated during the HTTP exchange handling.

Also, the client instrumentation in `DefaultWebClient` would start
the observation once the request is fully formed and immutable,
preventing the context from being propagated through HTTP request
headers.

This commit fixes both uses cases now by:

* adding the current observation as a key in the reactor context
  on the server side
* using the `ClientRequest.Builder` as a Carrier on the client side

Closes gh-29388

spring-webflux/src/test/java/org/springframework/web/reactive/function/client/DefaultClientRequestObservationConventionTests.java

@@ -44,7 +44,8 @@ class DefaultClientRequestObservationConventionTests {
 	@Test
 	void shouldHaveContextualName() {
 		ClientRequestObservationContext context = new ClientRequestObservationContext();
-		context.setCarrier(ClientRequest.create(HttpMethod.GET, URI.create("/test")).build());
+		context.setCarrier(ClientRequest.create(HttpMethod.GET, URI.create("/test")));
+		context.setBuiltRequest(context.getCarrier().build());
 		assertThat(this.observationConvention.getContextualName(context)).isEqualTo("http get");
 	}
 
@@ -77,10 +78,11 @@ class DefaultClientRequestObservationConventionTests {
 
 	@Test
 	void shouldAddKeyValuesForRequestWithUriTemplate() {
-		ClientRequest request = ClientRequest.create(HttpMethod.GET, URI.create("/resource/42"))
-				.attribute(WebClient.class.getName() + ".uriTemplate", "/resource/{id}").build();
+		ClientRequest.Builder request = ClientRequest.create(HttpMethod.GET, URI.create("/resource/42"))
+				.attribute(WebClient.class.getName() + ".uriTemplate", "/resource/{id}");
 		ClientRequestObservationContext context = createContext(request);
 		context.setUriTemplate("/resource/{id}");
+		context.setBuiltRequest(context.getCarrier().build());
 		assertThat(this.observationConvention.getLowCardinalityKeyValues(context))
 				.contains(KeyValue.of("exception", "none"), KeyValue.of("method", "GET"), KeyValue.of("uri", "/resource/{id}"),
 						KeyValue.of("status", "200"), KeyValue.of("outcome", "SUCCESS"));
@@ -90,7 +92,8 @@ class DefaultClientRequestObservationConventionTests {
 
 	@Test
 	void shouldAddKeyValuesForRequestWithoutUriTemplate() {
-		ClientRequestObservationContext context = createContext(ClientRequest.create(HttpMethod.GET, URI.create("/resource/42")).build());
+		ClientRequestObservationContext context = createContext(ClientRequest.create(HttpMethod.GET, URI.create("/resource/42")));
+		context.setBuiltRequest(context.getCarrier().build());
 		assertThat(this.observationConvention.getLowCardinalityKeyValues(context))
 				.contains(KeyValue.of("method", "GET"), KeyValue.of("uri", "none"));
 		assertThat(this.observationConvention.getHighCardinalityKeyValues(context)).hasSize(2).contains(KeyValue.of("http.url", "/resource/42"));
@@ -98,11 +101,12 @@ class DefaultClientRequestObservationConventionTests {
 
 	@Test
 	void shouldAddClientNameKeyValueForRequestWithHost() {
-		ClientRequestObservationContext context = createContext(ClientRequest.create(HttpMethod.GET, URI.create("https://localhost:8080/resource/42")).build());
+		ClientRequestObservationContext context = createContext(ClientRequest.create(HttpMethod.GET, URI.create("https://localhost:8080/resource/42")));
+		context.setBuiltRequest(context.getCarrier().build());
 		assertThat(this.observationConvention.getHighCardinalityKeyValues(context)).contains(KeyValue.of("client.name", "localhost"));
 	}
 
-	private ClientRequestObservationContext createContext(ClientRequest request) {
+	private ClientRequestObservationContext createContext(ClientRequest.Builder request) {
 		ClientRequestObservationContext context = new ClientRequestObservationContext();
 		context.setCarrier(request);
 		context.setResponse(ClientResponse.create(HttpStatus.OK).build());

spring-webflux/src/test/java/org/springframework/web/reactive/function/client/WebClientObservationTests.java

@@ -17,7 +17,11 @@
 package org.springframework.web.reactive.function.client;
 
 import java.time.Duration;
+import java.util.Collections;
 
+import io.micrometer.observation.Observation;
+import io.micrometer.observation.ObservationHandler;
+import io.micrometer.observation.contextpropagation.ObservationThreadLocalAccessor;
 import io.micrometer.observation.tck.TestObservationRegistry;
 import io.micrometer.observation.tck.TestObservationRegistryAssert;
 import org.junit.jupiter.api.BeforeEach;
@@ -28,6 +32,7 @@ import reactor.test.StepVerifier;
 
 import org.springframework.http.HttpStatus;
 
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.BDDMockito.given;
 import static org.mockito.BDDMockito.when;
@@ -57,17 +62,35 @@ class WebClientObservationTests {
 		when(mockResponse.bodyToMono(Void.class)).thenReturn(Mono.empty());
 		given(this.exchangeFunction.exchange(this.request.capture())).willReturn(Mono.just(mockResponse));
 		this.builder = WebClient.builder().baseUrl("/base").exchangeFunction(this.exchangeFunction).observationRegistry(this.observationRegistry);
+		this.observationRegistry.observationConfig().observationHandler(new HeaderInjectingHandler());
 	}
 
-
 	@Test
 	void recordsObservationForSuccessfulExchange() {
 		this.builder.build().get().uri("/resource/{id}", 42)
 				.retrieve().bodyToMono(Void.class).block(Duration.ofSeconds(10));
-		verifyAndGetRequest();
+
+		ClientRequest clientRequest = verifyAndGetRequest();
 
 		assertThatHttpObservation().hasLowCardinalityKeyValue("outcome", "SUCCESS")
 				.hasLowCardinalityKeyValue("uri", "/resource/{id}");
+		assertThat(clientRequest.headers()).containsEntry("foo", Collections.singletonList("bar"));
+	}
+
+	@Test
+	void recordsObservationForSuccessfulExchangeWithParentObservationInReactorContext() {
+		Observation parent = Observation.start("parent", observationRegistry);
+		try {
+			this.builder.build().get().uri("/resource/{id}", 42)
+					.retrieve().bodyToMono(Void.class).contextWrite(context -> context.put(ObservationThreadLocalAccessor.KEY, parent)).block(Duration.ofSeconds(10));
+			verifyAndGetRequest();
+
+			assertThatHttpObservation().hasLowCardinalityKeyValue("outcome", "SUCCESS")
+					.hasParentObservationEqualTo(parent);
+		}
+		finally {
+			parent.stop();
+		}
 	}
 
 	@Test
@@ -102,4 +125,17 @@ class WebClientObservationTests {
 		return request.getValue();
 	}
 
+	static class HeaderInjectingHandler implements ObservationHandler<ClientRequestObservationContext> {
+
+		@Override
+		public void onStart(ClientRequestObservationContext context) {
+			context.getSetter().set(context.getCarrier(), "foo", "bar");
+		}
+
+		@Override
+		public boolean supportsContext(Observation.Context context) {
+			return context instanceof ClientRequestObservationContext;
+		}
+	}
+
 }