Merge branch '2.2.x'

docs/pom.xml

@@ -31,8 +31,7 @@
 	<properties>
 		<docs.main>spring-cloud-sleuth</docs.main>
 		<!-- Comma separated list of whitelisted branches -->
-		<docs.whitelisted.branches>1.0.x,1.1.x,1.2.x,1.3.x,2.0.x,2.1.x
-		</docs.whitelisted.branches>
+		<docs.whitelisted.branches>2.1.x,2.2.x</docs.whitelisted.branches>
 		<main.basedir>${basedir}/..</main.basedir>
 		<configprops.inclusionPattern>spring.sleuth.*|spring.zipkin.*</configprops.inclusionPattern>
 	</properties>

spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientBeanPostProcessor.java

@@ -18,7 +18,6 @@ package org.springframework.cloud.sleuth.instrument.web.client;
 
 import java.util.List;
 import java.util.concurrent.CancellationException;
-import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 import java.util.function.Function;
@@ -290,8 +289,7 @@ final class TraceExchangeFilterFunction implements ExchangeFilterFunction {
 
 	}
 
-	static class TraceWebClientSubscription extends AtomicBoolean
-			implements Subscription {
+	static class TraceWebClientSubscription implements Subscription {
 
 		static final Exception CANCELLED_ERROR = new CancellationException("CANCELLED") {
 			@Override
@@ -304,6 +302,8 @@ final class TraceExchangeFilterFunction implements ExchangeFilterFunction {
 
 		final Subscription delegate;
 
+		volatile boolean requested;
+
 		TraceWebClientSubscription(Subscription delegate,
 				AtomicReference<Span> pendingSpan) {
 			this.delegate = delegate;
@@ -312,9 +312,8 @@ final class TraceExchangeFilterFunction implements ExchangeFilterFunction {
 
 		@Override
 		public void request(long n) {
-			if (compareAndSet(false, true)) {
-				delegate.request(n); // Not scoping to save overhead
-			}
+			requested = true;
+			delegate.request(n); // Not scoping to save overhead
 		}
 
 		@Override
@@ -331,7 +330,7 @@ final class TraceExchangeFilterFunction implements ExchangeFilterFunction {
 									+ span + "]");
 				}
 
-				if (!get()) { // Subscription.request() not called: Abandon the span.
+				if (!requested) { // Abandon the span.
 					span.abandon();
 				}
 				else { // Request was canceled in-flight

spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/util/BlockingQueueSpanReporter.java

@@ -0,0 +1,74 @@
+/*
+ * Copyright 2013-2020 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.cloud.sleuth.util;
+
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import zipkin2.Span;
+import zipkin2.reporter.Reporter;
+
+/**
+ * Like {@link ArrayListSpanReporter}, except appropriate for async instrumentation.
+ */
+public class BlockingQueueSpanReporter implements Reporter<Span> {
+
+	private final LinkedBlockingQueue<Span> spans = new LinkedBlockingQueue<>();
+
+	/**
+	 * Blocks until a span is reported or throws an {@link AssertionError}.
+	 * @return the first span not yet taken.
+	 */
+	public Span takeSpan() {
+		Span result = takeSpan(3_000);
+		if (result == null) {
+			throw new AssertionError("Span was not reported");
+		}
+		return result;
+	}
+
+	@Override
+	public String toString() {
+		return "BlockingQueueSpanReporter{spans=" + spans + '}';
+	}
+
+	@Override
+	public void report(Span span) {
+		spans.add(span);
+	}
+
+	/** Use this as a post-condition to ensure all spans are accounted for. */
+	public void assertEmpty() {
+		if (takeSpan(100) != null) {
+			throw new AssertionError(
+					"Span remaining in queue. Check for redundant reporting!");
+		}
+	}
+
+	private Span takeSpan(long timeout) {
+		Span result;
+		try {
+			result = spans.poll(timeout, TimeUnit.MILLISECONDS);
+		}
+		catch (InterruptedException e) {
+			Thread.currentThread().interrupt();
+			throw new AssertionError(e);
+		}
+		return result;
+	}
+
+}

spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/HttpClientBeanPostProcessorTest.java

@@ -0,0 +1,72 @@
+/*
+ * Copyright 2013-2020 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.cloud.sleuth.instrument.web.client;
+
+import brave.propagation.TraceContext;
+import io.netty.bootstrap.Bootstrap;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import reactor.core.publisher.Mono;
+import reactor.netty.Connection;
+
+import org.springframework.cloud.sleuth.instrument.web.client.HttpClientBeanPostProcessor.PendingSpan;
+import org.springframework.cloud.sleuth.instrument.web.client.HttpClientBeanPostProcessor.TracingMapConnect;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@ExtendWith(MockitoExtension.class)
+public class HttpClientBeanPostProcessorTest {
+
+	@Mock
+	Connection connection;
+
+	@Mock
+	Bootstrap bootstrap;
+
+	TraceContext traceContext = TraceContext.newBuilder().traceId(1).spanId(2)
+			.sampled(true).build();
+
+	@Test
+	void mapConnect_should_setup_reactor_context_currentTraceContext() {
+		TracingMapConnect tracingMapConnect = new TracingMapConnect(() -> traceContext);
+
+		Mono<Connection> original = Mono.just(connection).handle((t, ctx) -> {
+			assertThat(ctx.currentContext().get(TraceContext.class))
+					.isSameAs(traceContext);
+			assertThat(ctx.currentContext().get(PendingSpan.class)).isNotNull();
+		});
+
+		// Wrap and run the assertions
+		tracingMapConnect.apply(original, bootstrap).log().subscribe();
+	}
+
+	@Test
+	void mapConnect_should_setup_reactor_context_no_currentTraceContext() {
+		TracingMapConnect tracingMapConnect = new TracingMapConnect(() -> null);
+
+		Mono<Connection> original = Mono.just(connection).handle((t, ctx) -> {
+			assertThat(ctx.currentContext().getOrEmpty(TraceContext.class)).isEmpty();
+			assertThat(ctx.currentContext().get(PendingSpan.class)).isNotNull();
+		});
+
+		// Wrap and run the assertions
+		tracingMapConnect.apply(original, bootstrap).log().subscribe();
+	}
+
+}

spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientBeanPostProcessorTest.java

@@ -49,7 +49,7 @@ public class TraceWebClientBeanPostProcessorTest {
 	Span span;
 
 	@Test
-	public void should_add_filter_only_once_to_web_client() {
+	void should_add_filter_only_once_to_web_client() {
 		TraceWebClientBeanPostProcessor processor = new TraceWebClientBeanPostProcessor(
 				this.springContext);
 		WebClient client = WebClient.create();
@@ -65,7 +65,7 @@ public class TraceWebClientBeanPostProcessorTest {
 	}
 
 	@Test
-	public void should_add_filter_only_once_to_web_client_via_builder() {
+	void should_add_filter_only_once_to_web_client_via_builder() {
 		TraceWebClientBeanPostProcessor processor = new TraceWebClientBeanPostProcessor(
 				this.springContext);
 		WebClient.Builder builder = WebClient.builder();
@@ -83,7 +83,7 @@ public class TraceWebClientBeanPostProcessorTest {
 	}
 
 	@Test
-	public void should_close_span_on_cancel() {
+	void should_close_span_on_cancel() {
 		TraceWebClientSubscription traceSubscription = new TraceWebClientSubscription(
 				subscription, new AtomicReference<>(span));
 
@@ -98,7 +98,7 @@ public class TraceWebClientBeanPostProcessorTest {
 	}
 
 	@Test
-	public void should_not_crash_on_cancel_when_span_clear() {
+	void should_not_crash_on_cancel_when_span_clear() {
 		TraceWebClientSubscription traceSubscription = new TraceWebClientSubscription(
 				subscription, new AtomicReference<>());
 

spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/util/BlockingQueueSpanReporterTests.java

@@ -0,0 +1,55 @@
+/*
+ * Copyright 2013-2020 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.cloud.sleuth.util;
+
+import org.junit.jupiter.api.Test;
+import zipkin2.Span;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+public class BlockingQueueSpanReporterTests {
+
+	Span span1 = Span.newBuilder().traceId("1").id("1").build();
+
+	Span span2 = Span.newBuilder().traceId("1").id("2").build();
+
+	BlockingQueueSpanReporter reporter = new BlockingQueueSpanReporter();
+
+	@Test
+	void takeSpan_fifo_order() {
+		reporter.report(span1);
+		reporter.report(span2);
+
+		assertThat(reporter.takeSpan()).isSameAs(span1);
+		assertThat(reporter.takeSpan()).isSameAs(span2);
+	}
+
+	@Test
+	void assertEmpty() {
+		reporter.assertEmpty();
+	}
+
+	@Test
+	void assertEmpty_fails_when_not_empty() {
+		reporter.report(span1);
+
+		assertThatThrownBy(reporter::assertEmpty).isInstanceOf(AssertionError.class)
+				.hasMessage("Span remaining in queue. Check for redundant reporting!");
+	}
+
+}

tests/spring-cloud-sleuth-instrumentation-mvc-tests/src/test/java/org/springframework/cloud/sleuth/instrument/web/TraceFilterWebIntegrationTests.java

@@ -29,7 +29,6 @@ import brave.sampler.Sampler;
 import brave.sampler.SamplerFunction;
 import org.assertj.core.api.BDDAssertions;
 import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import zipkin2.Span;
@@ -39,7 +38,7 @@ import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.boot.test.system.CapturedOutput;
 import org.springframework.boot.test.system.OutputCaptureExtension;
-import org.springframework.cloud.sleuth.util.ArrayListSpanReporter;
+import org.springframework.cloud.sleuth.util.BlockingQueueSpanReporter;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.core.env.Environment;
@@ -64,10 +63,7 @@ import static org.assertj.core.api.BDDAssertions.then;
 public class TraceFilterWebIntegrationTests {
 
 	@Autowired
-	Tracing tracer;
-
-	@Autowired
-	ArrayListSpanReporter accumulator;
+	BlockingQueueSpanReporter reporter;
 
 	@Autowired
 	@HttpServerSampler
@@ -76,10 +72,9 @@ public class TraceFilterWebIntegrationTests {
 	@Autowired
 	Environment environment;
 
-	@BeforeEach
 	@AfterEach
 	public void cleanup() {
-		this.accumulator.clear();
+		this.reporter.assertEmpty();
 	}
 
 	@Test
@@ -88,8 +83,7 @@ public class TraceFilterWebIntegrationTests {
 				String.class);
 
 		then(Tracing.current().tracer().currentSpan()).isNull();
-		then(this.accumulator.getSpans()).hasSize(1);
-		then(this.accumulator.getSpans().get(0).tags()).containsKey("http.url");
+		then(this.reporter.takeSpan().tags()).containsKey("http.url");
 	}
 
 	@Test
@@ -103,8 +97,7 @@ public class TraceFilterWebIntegrationTests {
 		}
 
 		then(Tracing.current().tracer().currentSpan()).isNull();
-		then(this.accumulator.getSpans()).hasSize(1);
-		Span fromFirstTraceFilterFlow = this.accumulator.getSpans().get(0);
+		Span fromFirstTraceFilterFlow = this.reporter.takeSpan();
 		then(fromFirstTraceFilterFlow.tags()).containsEntry("http.method", "GET")
 				.containsEntry("mvc.controller.class", "ExceptionThrowingController")
 				.containsEntry("error",
@@ -130,13 +123,10 @@ public class TraceFilterWebIntegrationTests {
 		}
 
 		then(Tracing.current().tracer().currentSpan()).isNull();
-		then(this.accumulator.getSpans()).hasSize(1);
-		then(this.accumulator.getSpans().get(0).kind().ordinal())
-				.isEqualTo(Span.Kind.SERVER.ordinal());
-		then(this.accumulator.getSpans().get(0).tags()).containsEntry("http.status_code",
-				"400");
-		then(this.accumulator.getSpans().get(0).tags()).containsEntry("http.path",
-				"/test_bad_request");
+		Span span = this.reporter.takeSpan();
+		then(span.kind().ordinal()).isEqualTo(Span.Kind.SERVER.ordinal());
+		then(span.tags()).containsEntry("http.status_code", "400");
+		then(span.tags()).containsEntry("http.path", "/test_bad_request");
 	}
 
 	@Test
@@ -163,8 +153,8 @@ public class TraceFilterWebIntegrationTests {
 		}
 
 		@Bean
-		ArrayListSpanReporter reporter() {
-			return new ArrayListSpanReporter();
+		BlockingQueueSpanReporter reporter() {
+			return new BlockingQueueSpanReporter();
 		}
 
 		@Bean