diff --git a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/HttpClientBeanPostProcessor.java b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/HttpClientBeanPostProcessor.java index a358784da..7e45a9b2b 100644 --- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/HttpClientBeanPostProcessor.java +++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/HttpClientBeanPostProcessor.java @@ -182,6 +182,7 @@ class HttpClientBeanPostProcessor implements BeanPostProcessor { @Override public void accept(HttpClientResponse response, Connection connection) { + // TODO: is there a way to read the request at response time? handle(response.currentContext(), response, null); } 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 97bb452a7..86d1c6f65 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 @@ -220,6 +220,7 @@ final class TraceExchangeFilterFunction implements ExchangeFilterFunction { final CurrentTraceContext currentTraceContext; + // TODO: this isn't implemented correctly. error and success could both be called boolean done; WebClientTracerSubscriber(CoreSubscriber actual, @@ -251,7 +252,8 @@ final class TraceExchangeFilterFunction implements ExchangeFilterFunction { try (Scope scope = currentTraceContext.maybeScope(parent)) { subscription.cancel(); } - finally { + finally { // TODO: this is probably incorrect as cancel happens + // routinely in unary subscription. if (log.isDebugEnabled()) { log.debug("Subscription was cancelled. Will close the span [" + clientSpan + "]"); @@ -274,6 +276,7 @@ final class TraceExchangeFilterFunction implements ExchangeFilterFunction { .build()); } finally { + // TODO: is there a way to read the request at response time? handleReceive(response, null); } } diff --git a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/ReactorNettyHttpClientBraveTests.java b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/ReactorNettyHttpClientBraveTests.java similarity index 52% rename from spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/ReactorNettyHttpClientBraveTests.java rename to tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/ReactorNettyHttpClientBraveTests.java index fe84c6529..e526f7188 100644 --- a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/ReactorNettyHttpClientBraveTests.java +++ b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/ReactorNettyHttpClientBraveTests.java @@ -17,26 +17,17 @@ package org.springframework.cloud.sleuth.instrument.web.client; import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicReference; import brave.http.HttpTracing; import brave.test.http.ITHttpAsyncClient; import io.netty.channel.ChannelOption; import io.netty.handler.timeout.ReadTimeoutHandler; -import org.junit.After; -import org.junit.Before; import org.junit.Ignore; import org.junit.Test; -import org.reactivestreams.Subscription; -import reactor.core.CoreSubscriber; -import reactor.core.publisher.Hooks; import reactor.core.publisher.Mono; -import reactor.core.publisher.Operators; -import reactor.core.scheduler.Schedulers; import reactor.netty.ByteBufFlux; import reactor.netty.http.client.HttpClient; import reactor.netty.http.client.HttpClientResponse; -import reactor.util.context.Context; import zipkin2.Callback; import org.springframework.beans.factory.config.BeanPostProcessor; @@ -46,30 +37,23 @@ import org.springframework.context.annotation.AnnotationConfigApplicationContext * This runs Brave's integration tests, ensuring common instrumentation bugs aren't * present. */ -public class ReactorNettyHttpClientBraveTests extends ITHttpAsyncClient { - @Before - @After - public void resetHooks() { - // There's an assumption some other test is leaking hooks, so we clear them all to - // prevent should_not_scope_scalar_subscribe from being interfered with. - Hooks.resetOnEachOperator(); - Hooks.resetOnLastOperator(); - Schedulers.removeExecutorServiceDecorator("sleuth"); - } +// Function of spring context so that shutdown hooks happen! +public class ReactorNettyHttpClientBraveTests + extends ITHttpAsyncClient { /** * This uses Spring to instrument the {@link HttpClient} using a * {@link BeanPostProcessor}. */ @Override - protected HttpClient newClient(int port) { + protected AnnotationConfigApplicationContext newClient(int port) { AnnotationConfigApplicationContext result = new AnnotationConfigApplicationContext(); result.registerBean(HttpTracing.class, () -> httpTracing); result.registerBean(HttpClient.class, - ReactorNettyHttpClientBraveTests::testHttpClient); + () -> testHttpClient().baseUrl("http://127.0.0.1:" + port)); result.register(HttpClientBeanPostProcessor.class); result.refresh(); - return result.getBean(HttpClient.class).baseUrl("http://127.0.0.1:" + port); + return result; } static HttpClient testHttpClient() { @@ -82,21 +66,29 @@ public class ReactorNettyHttpClientBraveTests extends ITHttpAsyncClient callback) { - Mono request = client.get().uri(path).response(); + protected void getAsync(AnnotationConfigApplicationContext context, String path, + Callback callback) { + Mono request = context.getBean(HttpClient.class).get() + .uri(path).response(); - request.subscribe(new CoreSubscriber() { - - final AtomicReference ref = new AtomicReference<>(); - - @Override - public void onSubscribe(Subscription s) { - if (Operators.validate(ref.getAndSet(s), s)) { - s.request(Long.MAX_VALUE); - } - else { - s.cancel(); - } - } - - @Override - public void onNext(HttpClientResponse t) { - Subscription s = ref.getAndSet(null); - if (s != null) { - callback.onSuccess(null); - s.cancel(); - } - else { - Operators.onNextDropped(t, currentContext()); - } - } - - @Override - public void onError(Throwable t) { - if (ref.getAndSet(null) != null) { - callback.onError(t); - } - } - - @Override - public void onComplete() { - if (ref.getAndSet(null) != null) { - callback.onSuccess(null); - } - } - - @Override - public Context currentContext() { - return Context.empty(); - } - }); + TestCallbackSubscriber.subscribe(request, callback); } } diff --git a/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/TestCallbackSubscriber.java b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/TestCallbackSubscriber.java new file mode 100644 index 000000000..57569eeaa --- /dev/null +++ b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/TestCallbackSubscriber.java @@ -0,0 +1,102 @@ +/* + * 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 java.util.concurrent.atomic.AtomicReference; + +import org.reactivestreams.Subscription; +import reactor.core.CoreSubscriber; +import reactor.core.publisher.Mono; +import reactor.core.publisher.Operators; +import reactor.util.context.Context; +import zipkin2.Callback; + +/** + * {@link #subscribe} is made for reactor-netty and WebFlux client requests used in tests. + * This does more assertions than normal, to ensure instrumentation isn't redundantly + * signalling, or missing signals. + * + *

+ * The implementation forwards signals to the supplied {@link Callback}, enforcing + * assumptions about a non-empty, {@link Mono} subscription. + */ +final class TestCallbackSubscriber implements CoreSubscriber { + + static void subscribe(Mono mono, Callback callback) { + mono.subscribe(new TestCallbackSubscriber<>(callback)); + } + + final Callback callback; + + final AtomicReference ref = new AtomicReference<>(); + + private TestCallbackSubscriber(Callback callback) { + this.callback = callback; + } + + @Override + public void onSubscribe(Subscription s) { + if (Operators.validate(ref.getAndSet(s), s)) { + s.request(Long.MAX_VALUE); + } + else { + // We don't intentionally call subscribe() multiple times in our tests. If we + // reach here, possibly instrumentation is redundantly subscribing. + callback.onError(new AssertionError("onSubscribe() called twice!")); + } + } + + @Override + public void onNext(T t) { + if (ref.getAndSet(null) != null) { + callback.onSuccess(null /* because Void */); + } + else { + // This is a Mono, which doesn't signal onNext() twice. If we reach here, + // possibly instrumentation is signaling twice. + callback.onError(new AssertionError("onNext() called twice!")); + } + } + + @Override + public void onError(Throwable t) { + if (ref.getAndSet(null) != null) { + callback.onError(t); + } + else { + // We don't expect onError() to signal twice. If we reach here, possibly + // instrumentation is signaling twice or onSuccess() threw an exception. + callback.onError(new AssertionError("onError() called twice: " + t, t)); + } + } + + @Override + public void onComplete() { + if (ref.getAndSet(null) != null) { + // Tests make a non-empty Mono subscription, which should not signal + // onComplete() before onNext(). If we reach here, possibly instrumentation + // is not signaling onNext() when it should. + callback.onError(new AssertionError("onComplete() called before onNext!")); + } + } + + @Override + public Context currentContext() { + return Context.empty(); + } + +} diff --git a/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/WebClientBraveTests.java b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/WebClientBraveTests.java index 9d113a5bf..6f89a6e2d 100644 --- a/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/WebClientBraveTests.java +++ b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/WebClientBraveTests.java @@ -16,27 +16,15 @@ package org.springframework.cloud.sleuth.instrument.web.client; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicReference; - import brave.http.HttpTracing; import brave.test.http.ITHttpAsyncClient; -import io.netty.channel.ChannelOption; -import io.netty.handler.timeout.ReadTimeoutHandler; -import org.junit.After; -import org.junit.Before; import org.junit.Ignore; import org.junit.Test; -import org.reactivestreams.Subscription; -import reactor.core.CoreSubscriber; import reactor.core.publisher.Mono; -import reactor.core.publisher.Operators; import reactor.netty.http.client.HttpClient; -import reactor.util.context.Context; import zipkin2.Callback; import org.springframework.beans.factory.config.BeanPostProcessor; -import org.springframework.cloud.sleuth.instrument.reactor.ScopePassingSpanSubscriberTests; import org.springframework.context.annotation.AnnotationConfigApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -49,94 +37,48 @@ import org.springframework.web.reactive.function.client.WebClient; * This runs Brave's integration tests without underlying instrumentation, which would * happen when a 3rd party client like Jetty is in use. */ -public class WebClientBraveTests extends ITHttpAsyncClient { - - @Before - @After - public void resetHooks() { - new ScopePassingSpanSubscriberTests().resetHooks(); - } +// Function of spring context so that shutdown hooks happen! +public class WebClientBraveTests + extends ITHttpAsyncClient { /** * This uses Spring to instrument the {@link WebClient} using a * {@link BeanPostProcessor}. */ @Override - protected WebClient newClient(int port) { + protected AnnotationConfigApplicationContext newClient(int port) { AnnotationConfigApplicationContext result = new AnnotationConfigApplicationContext(); result.registerBean(HttpTracing.class, () -> httpTracing); result.register(WebClientBuilderConfiguration.class); result.register(TraceWebClientBeanPostProcessor.class); result.refresh(); - return result.getBean(WebClient.Builder.class).baseUrl("http://127.0.0.1:" + port) - .build(); + return result; } @Override - protected void closeClient(WebClient client) { - // WebClient is not Closeable + protected void closeClient(AnnotationConfigApplicationContext context) { + context.close(); // ensures shutdown hooks fire } @Override - protected void get(WebClient client, String pathIncludingQuery) { - client.get().uri(pathIncludingQuery).exchange().block(); + protected void get(AnnotationConfigApplicationContext context, + String pathIncludingQuery) { + client(context).get().uri(pathIncludingQuery).exchange().block(); } @Override - protected void post(WebClient client, String pathIncludingQuery, String body) { - client.post().uri(pathIncludingQuery).body(BodyInserters.fromValue(body)) + protected void post(AnnotationConfigApplicationContext context, + String pathIncludingQuery, String body) { + client(context).post().uri(pathIncludingQuery).body(BodyInserters.fromValue(body)) .exchange().block(); } @Override - protected void getAsync(WebClient client, String path, Callback callback) { - Mono request = client.get().uri(path).exchange(); + protected void getAsync(AnnotationConfigApplicationContext context, String path, + Callback callback) { + Mono request = client(context).get().uri(path).exchange(); - request.subscribe(new CoreSubscriber() { - - final AtomicReference ref = new AtomicReference<>(); - - @Override - public void onSubscribe(Subscription s) { - if (Operators.validate(ref.getAndSet(s), s)) { - s.request(Long.MAX_VALUE); - } - else { - s.cancel(); - } - } - - @Override - public void onNext(ClientResponse t) { - Subscription s = ref.getAndSet(null); - if (s != null) { - callback.onSuccess(null); - s.cancel(); - } - else { - Operators.onNextDropped(t, currentContext()); - } - } - - @Override - public void onError(Throwable t) { - if (ref.getAndSet(null) != null) { - callback.onError(t); - } - } - - @Override - public void onComplete() { - if (ref.getAndSet(null) != null) { - callback.onSuccess(null); - } - } - - @Override - public Context currentContext() { - return Context.empty(); - } - }); + TestCallbackSubscriber.subscribe(request, callback); } @Test @@ -151,6 +93,11 @@ public class WebClientBraveTests extends ITHttpAsyncClient { public void reportsServerAddress() { } + WebClient client(AnnotationConfigApplicationContext context) { + return context.getBean(WebClient.Builder.class) + .baseUrl("http://127.0.0.1:" + server.getPort()).build(); + } + /** * This fakes auto-configuration which wouldn't configure reactor's trace * instrumentation. @@ -160,13 +107,7 @@ public class WebClientBraveTests extends ITHttpAsyncClient { @Bean HttpClient httpClient() { - // TODO: ReactorNettyHttpClientBraveTests.testHttpClient() #1554 - return HttpClient.create() - .tcpConfiguration(tcpClient -> tcpClient - .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 1000) - .doOnConnected(conn -> conn.addHandler( - new ReadTimeoutHandler(1, TimeUnit.SECONDS)))) - .followRedirect(true); + return ReactorNettyHttpClientBraveTests.testHttpClient(); } @Bean