diff --git a/docs/pom.xml b/docs/pom.xml index ed0386d7c..dbff83b99 100644 --- a/docs/pom.xml +++ b/docs/pom.xml @@ -31,8 +31,7 @@ spring-cloud-sleuth - 1.0.x,1.1.x,1.2.x,1.3.x,2.0.x,2.1.x - + 2.1.x,2.2.x ${basedir}/.. spring.sleuth.*|spring.zipkin.* diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientBeanPostProcessor.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientBeanPostProcessor.java index 924976843..ce95437d0 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientBeanPostProcessor.java +++ b/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 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 diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/util/BlockingQueueSpanReporter.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/util/BlockingQueueSpanReporter.java new file mode 100644 index 000000000..eff2686dc --- /dev/null +++ b/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 { + + private final LinkedBlockingQueue 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; + } + +} diff --git a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/HttpClientBeanPostProcessorTest.java b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/HttpClientBeanPostProcessorTest.java new file mode 100644 index 000000000..75ae91690 --- /dev/null +++ b/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 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 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(); + } + +} diff --git a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientBeanPostProcessorTest.java b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientBeanPostProcessorTest.java index 2622c36f9..d99b7f6c2 100644 --- a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientBeanPostProcessorTest.java +++ b/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<>()); diff --git a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/util/BlockingQueueSpanReporterTests.java b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/util/BlockingQueueSpanReporterTests.java new file mode 100644 index 000000000..190388a29 --- /dev/null +++ b/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!"); + } + +} diff --git a/tests/spring-cloud-sleuth-instrumentation-mvc-tests/src/test/java/org/springframework/cloud/sleuth/instrument/web/TraceFilterWebIntegrationTests.java b/tests/spring-cloud-sleuth-instrumentation-mvc-tests/src/test/java/org/springframework/cloud/sleuth/instrument/web/TraceFilterWebIntegrationTests.java index 001fbc7a3..bd42e9532 100644 --- a/tests/spring-cloud-sleuth-instrumentation-mvc-tests/src/test/java/org/springframework/cloud/sleuth/instrument/web/TraceFilterWebIntegrationTests.java +++ b/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