From 51c9158837c573c5213bec2a096705facef2b891 Mon Sep 17 00:00:00 2001 From: Adrian Cole Date: Mon, 24 Feb 2020 14:34:25 +0800 Subject: [PATCH 1/4] Fixes WebClient propagation errors and adds Brave Tests (#1562) In many places, the trace context of callbacks was accidentally set to the client span, not the invocation context. I noticed a hack trying to work around this. This code fixes all the problems around context. It also removes some sporadic logging, which was only applied to a few hooks. Finally, this adds Brave tests which would have caught the problems earlier. Notably, there is still more work to do as this will not help with duplicate instrumentation, which is normal when reactor-netty is the WebClient's HTTP connector. --- .../TraceWebClientBeanPostProcessor.java | 414 +++++++----------- ...FilterFunctionHttpClientResponseTests.java | 10 +- .../pom.xml | 11 + .../web/client/WebClientBraveTests.java | 180 ++++++++ 4 files changed, 352 insertions(+), 263 deletions(-) create mode 100644 tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/WebClientBraveTests.java 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 e50f31fc0..97bb452a7 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 @@ -16,18 +16,16 @@ package org.springframework.cloud.sleuth.instrument.web.client; -import java.util.Collections; import java.util.List; import java.util.concurrent.CancellationException; import java.util.function.Consumer; import java.util.function.Function; import brave.Span; -import brave.Tracer; -import brave.Tracing; import brave.http.HttpClientHandler; import brave.http.HttpTracing; -import brave.propagation.Propagation; +import brave.propagation.CurrentTraceContext; +import brave.propagation.CurrentTraceContext.Scope; import brave.propagation.TraceContext; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -38,11 +36,10 @@ import reactor.core.publisher.Mono; import reactor.util.annotation.Nullable; import reactor.util.context.Context; -import org.springframework.beans.BeansException; import org.springframework.beans.factory.config.BeanPostProcessor; +import org.springframework.cloud.sleuth.internal.LazyBean; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.core.io.buffer.DataBuffer; -import org.springframework.web.client.RestClientException; import org.springframework.web.reactive.function.client.ClientRequest; import org.springframework.web.reactive.function.client.ClientResponse; import org.springframework.web.reactive.function.client.ExchangeFilterFunction; @@ -60,21 +57,19 @@ import static org.springframework.cloud.sleuth.instrument.reactor.ReactorSleuth. */ final class TraceWebClientBeanPostProcessor implements BeanPostProcessor { - private final ConfigurableApplicationContext springContext; + final ConfigurableApplicationContext springContext; TraceWebClientBeanPostProcessor(ConfigurableApplicationContext springContext) { this.springContext = springContext; } @Override - public Object postProcessBeforeInitialization(Object bean, String beanName) - throws BeansException { + public Object postProcessBeforeInitialization(Object bean, String beanName) { return bean; } @Override - public Object postProcessAfterInitialization(Object bean, String beanName) - throws BeansException { + public Object postProcessAfterInitialization(Object bean, String beanName) { if (bean instanceof WebClient) { WebClient webClient = (WebClient) bean; return wrapBuilder(webClient.mutate()).build(); @@ -114,25 +109,6 @@ final class TraceWebClientBeanPostProcessor implements BeanPostProcessor { final class TraceExchangeFilterFunction implements ExchangeFilterFunction { private static final Log log = LogFactory.getLog(TraceExchangeFilterFunction.class); - static final Propagation.Setter SETTER = new Propagation.Setter() { - @Override - public void put(ClientRequest.Builder carrier, String key, String value) { - carrier.headers(httpHeaders -> { - if (log.isTraceEnabled()) { - log.trace("Replacing [" + key + "] with value [" + value + "]"); - } - httpHeaders.merge(key, Collections.singletonList(value), - (oldValue, newValue) -> newValue); - }); - } - - @Override - public String toString() { - return "ClientRequest.Builder::header"; - } - }; - - static final String CLIENT_SPAN_KEY = "sleuth.webclient.clientSpan"; static final Exception CANCELLED_ERROR = new CancellationException("CANCELLED") { @Override @@ -141,20 +117,17 @@ final class TraceExchangeFilterFunction implements ExchangeFilterFunction { } }; - final ConfigurableApplicationContext springContext; + final LazyBean httpTracing; final Function, ? extends Publisher> scopePassingTransformer; - Tracer tracer; - - HttpTracing httpTracing; - + // Lazy initialized fields HttpClientHandler handler; - TraceContext.Injector injector; + CurrentTraceContext currentTraceContext; TraceExchangeFilterFunction(ConfigurableApplicationContext springContext) { - this.springContext = springContext; + this.httpTracing = LazyBean.create(springContext, HttpTracing.class); this.scopePassingTransformer = scopePassingSpanOperator(springContext); } @@ -166,80 +139,55 @@ final class TraceExchangeFilterFunction implements ExchangeFilterFunction { @Override public Mono filter(ClientRequest request, ExchangeFunction next) { HttpClientRequest wrapper = new HttpClientRequest(request); + TraceContext parent = currentTraceContext().get(); + Span clientSpan = handler().handleSend(wrapper); if (log.isDebugEnabled()) { - log.debug("Instrumenting WebClient call"); + log.debug("HttpClientHandler::handleSend: " + clientSpan); } - Span parentSpan = tracer().currentSpan(); - Span span = handler().handleSend(wrapper); - if (log.isDebugEnabled()) { - log.debug("Handled send of " + span); - } - MonoWebClientTrace trace = new MonoWebClientTrace(next, wrapper.buildRequest(), - this, span); - // TODO: investigate why this commit leaks a scope: - // 8f5bcdabd7af23df443e771432eb85597f3b3076 - tracer().withSpanInScope(parentSpan); - return trace; + return new MonoWebClientTrace(next, wrapper.buildRequest(), this, parent, + clientSpan); + } + + CurrentTraceContext currentTraceContext() { + if (this.currentTraceContext == null) { + this.currentTraceContext = httpTracing.get().tracing().currentTraceContext(); + } + return this.currentTraceContext; } - @SuppressWarnings("unchecked") HttpClientHandler handler() { if (this.handler == null) { - this.handler = HttpClientHandler - .create(this.springContext.getBean(HttpTracing.class)); + this.handler = HttpClientHandler.create(this.httpTracing.get()); } return this.handler; } - Tracer tracer() { - if (this.tracer == null) { - this.tracer = httpTracing().tracing().tracer(); - } - return this.tracer; - } - - HttpTracing httpTracing() { - if (this.httpTracing == null) { - this.httpTracing = this.springContext.getBean(HttpTracing.class); - } - return this.httpTracing; - } - - TraceContext.Injector injector() { - if (this.injector == null) { - this.injector = this.springContext.getBean(HttpTracing.class).tracing() - .propagation().injector(SETTER); - } - return this.injector; - } - private static final class MonoWebClientTrace extends Mono { final ExchangeFunction next; final ClientRequest request; - final Tracer tracer; - final HttpClientHandler handler; - final TraceContext.Injector injector; - - final Tracing tracing; + final CurrentTraceContext currentTraceContext; final Function, ? extends Publisher> scopePassingTransformer; + @Nullable + final TraceContext parent; + private final Span span; MonoWebClientTrace(ExchangeFunction next, ClientRequest request, - TraceExchangeFilterFunction parent, Span span) { + TraceExchangeFilterFunction filterFunction, @Nullable TraceContext parent, + Span span) { this.next = next; this.request = request; - this.tracer = parent.tracer(); - this.handler = parent.handler(); - this.injector = parent.injector(); - this.tracing = parent.httpTracing().tracing(); - this.scopePassingTransformer = parent.scopePassingTransformer; + this.handler = filterFunction.handler(); + this.currentTraceContext = filterFunction.currentTraceContext(); + this.scopePassingTransformer = filterFunction.scopePassingTransformer; + this.parent = parent; this.span = span; } @@ -248,177 +196,133 @@ final class TraceExchangeFilterFunction implements ExchangeFilterFunction { Context context = subscriber.currentContext(); - this.next.exchange(request).subscribe( - new WebClientTracerSubscriber(subscriber, context, span, this)); - } - - static final class WebClientTracerSubscriber - implements CoreSubscriber { - - final CoreSubscriber actual; - - final Context context; - - final Span span; - - final HttpClientHandler handler; - - final Function, ? extends Publisher> scopePassingTransformer; - - final Tracing tracing; - - boolean done; - - WebClientTracerSubscriber(CoreSubscriber actual, - Context context, Span span, MonoWebClientTrace parent) { - this.actual = actual; - this.span = span; - this.handler = parent.handler; - this.tracing = parent.tracing; - this.scopePassingTransformer = parent.scopePassingTransformer; - - if (!context.hasKey(TraceContext.class)) { - context = context.put(TraceContext.class, span.context()); - if (log.isDebugEnabled()) { - log.debug("Reactor Context got injected with the client span " - + span); - } - } - - this.context = context.put(CLIENT_SPAN_KEY, span); - } - - @Override - public void onSubscribe(Subscription subscription) { - this.actual.onSubscribe(new Subscription() { - @Override - public void request(long n) { - try (Tracer.SpanInScope ws = tracing.tracer() - .withSpanInScope(span)) { - if (log.isTraceEnabled()) { - log.trace("Request"); - } - subscription.request(n); - } - } - - @Override - public void cancel() { - try (Tracer.SpanInScope ws = tracing.tracer() - .withSpanInScope(span)) { - if (log.isTraceEnabled()) { - log.trace("Cancel"); - } - terminateSpanOnCancel(); - subscription.cancel(); - } - } - }); - } - - @Override - public void onNext(ClientResponse response) { - try (Tracer.SpanInScope ws = tracing.tracer().withSpanInScope(span)) { - this.done = true; - try { - // decorate response body - this.actual.onNext(ClientResponse.from(response) - .body(response.bodyToFlux(DataBuffer.class) - .transform(this.scopePassingTransformer)) - .build()); - } - finally { - terminateSpan(response, null); - } - } - } - - @Override - public void onError(Throwable t) { - try (Tracer.SpanInScope ws = tracing.tracer().withSpanInScope(span)) { - try { - this.actual.onError(t); - } - finally { - terminateSpan(null, t); - } - } - } - - @Override - public void onComplete() { - try (Tracer.SpanInScope ws = tracing.tracer().withSpanInScope(span)) { - try { - this.actual.onComplete(); - } - finally { - if (!this.done) { - terminateSpan(null, null); - } - } - } - } - - @Override - public Context currentContext() { - return this.context; - } - - void handleReceive(Span clientSpan, @Nullable ClientResponse res, - @Nullable Throwable error) { - if (log.isTraceEnabled()) { - log.trace("Handling receive"); - } - HttpClientResponse response = res != null ? new HttpClientResponse(res) - : null; - this.handler.handleReceive(response, error, clientSpan); - if (log.isTraceEnabled()) { - log.trace("Closed scope"); - } - } - - void terminateSpanOnCancel() { - if (log.isDebugEnabled()) { - log.debug("Subscription was cancelled. Will close the span [" - + this.span + "]"); - } - - handleReceive(this.span, null, CANCELLED_ERROR); - } - - void terminateSpan(@Nullable ClientResponse clientResponse, - @Nullable Throwable error) { - if (clientResponse == null) { - if (log.isDebugEnabled()) { - log.debug("No response was returned. Will close the span [" - + this.span + "]"); - } - handleReceive(this.span, null, error); - return; - } - int statusCode = clientResponse.rawStatusCode(); - boolean isHttpError = statusCode >= 400; - if (isHttpError) { - if (log.isDebugEnabled()) { - log.debug( - "Non positive status code was returned from the call. Will close the span [" - + this.span + "]"); - } - error = new RestClientException( - "Status code of the response is [" + statusCode + "]"); - } - handleReceive(this.span, clientResponse, error); - } - + this.next.exchange(request).subscribe(new WebClientTracerSubscriber( + subscriber, context, parent, span, this)); } } - static final class HttpClientRequest extends brave.http.HttpClientRequest { + private static final class WebClientTracerSubscriber + implements CoreSubscriber { - private final ClientRequest delegate; + final CoreSubscriber actual; - private final ClientRequest.Builder builder; + final Context context; + + @Nullable + final TraceContext parent; + + final Span clientSpan; + + final HttpClientHandler handler; + + final Function, ? extends Publisher> scopePassingTransformer; + + final CurrentTraceContext currentTraceContext; + + boolean done; + + WebClientTracerSubscriber(CoreSubscriber actual, + Context ctx, @Nullable final TraceContext parent, Span clientSpan, + MonoWebClientTrace mono) { + this.actual = actual; + this.parent = parent; + this.clientSpan = clientSpan; + this.handler = mono.handler; + this.currentTraceContext = mono.currentTraceContext; + this.scopePassingTransformer = mono.scopePassingTransformer; + this.context = parent != null + && !parent.equals(ctx.getOrDefault(TraceContext.class, null)) + ? ctx.put(TraceContext.class, parent) : ctx; + } + + @Override + public void onSubscribe(Subscription subscription) { + this.actual.onSubscribe(new Subscription() { + @Override + public void request(long n) { + try (Scope scope = currentTraceContext.maybeScope(parent)) { + subscription.request(n); + } + } + + @Override + public void cancel() { + try (Scope scope = currentTraceContext.maybeScope(parent)) { + subscription.cancel(); + } + finally { + if (log.isDebugEnabled()) { + log.debug("Subscription was cancelled. Will close the span [" + + clientSpan + "]"); + } + handleReceive(null, CANCELLED_ERROR); + } + } + }); + } + + @Override + public void onNext(ClientResponse response) { + try (Scope scope = currentTraceContext.maybeScope(parent)) { + this.done = true; + // decorate response body + this.actual + .onNext(ClientResponse.from(response) + .body(response.bodyToFlux(DataBuffer.class) + .transform(this.scopePassingTransformer)) + .build()); + } + finally { + handleReceive(response, null); + } + } + + @Override + public void onError(Throwable t) { + try (Scope scope = currentTraceContext.maybeScope(parent)) { + this.actual.onError(t); + } + finally { + handleReceive(null, t); + } + } + + @Override + public void onComplete() { + try (Scope scope = currentTraceContext.maybeScope(parent)) { + this.actual.onComplete(); + } + finally { + // TODO: onComplete should be after onNext. Why are we handling this? + if (!this.done) { // unknown state + if (log.isDebugEnabled()) { + log.debug("Reached OnComplete without finishing [" + + this.clientSpan + "]"); + } + this.clientSpan.abandon(); + } + } + } + + @Override + public Context currentContext() { + return this.context; + } + + void handleReceive(@Nullable ClientResponse res, @Nullable Throwable error) { + HttpClientResponse response = res != null ? new HttpClientResponse(res) + : null; + this.handler.handleReceive(response, error, clientSpan); + } + + } + + private static final class HttpClientRequest extends brave.http.HttpClientRequest { + + final ClientRequest delegate; + + final ClientRequest.Builder builder; HttpClientRequest(ClientRequest delegate) { this.delegate = delegate; @@ -463,7 +367,7 @@ final class TraceExchangeFilterFunction implements ExchangeFilterFunction { static final class HttpClientResponse extends brave.http.HttpClientResponse { - private final ClientResponse delegate; + final ClientResponse delegate; HttpClientResponse(ClientResponse delegate) { this.delegate = delegate; @@ -476,12 +380,8 @@ final class TraceExchangeFilterFunction implements ExchangeFilterFunction { @Override public int statusCode() { - try { - return delegate.rawStatusCode(); - } - catch (Exception dontCare) { - return 0; - } + // unlike statusCode(), this doesn't throw + return Math.max(delegate.rawStatusCode(), 0); } } diff --git a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/TraceExchangeFilterFunctionHttpClientResponseTests.java b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/TraceExchangeFilterFunctionHttpClientResponseTests.java index 031edc235..b194cf6ab 100644 --- a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/TraceExchangeFilterFunctionHttpClientResponseTests.java +++ b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/TraceExchangeFilterFunctionHttpClientResponseTests.java @@ -20,6 +20,7 @@ import org.assertj.core.api.BDDAssertions; import org.junit.Test; import org.mockito.BDDMockito; +import org.springframework.cloud.sleuth.instrument.web.client.TraceExchangeFilterFunction.HttpClientResponse; import org.springframework.web.reactive.function.client.ClientResponse; public class TraceExchangeFilterFunctionHttpClientResponseTests { @@ -27,10 +28,8 @@ public class TraceExchangeFilterFunctionHttpClientResponseTests { @Test public void should_return_0_when_invalid_status_code_is_returned() { ClientResponse clientResponse = BDDMockito.mock(ClientResponse.class); - BDDMockito.given(clientResponse.rawStatusCode()) - .willThrow(new IllegalStateException("Boom")); - TraceExchangeFilterFunction.HttpClientResponse response = new TraceExchangeFilterFunction.HttpClientResponse( - clientResponse); + BDDMockito.given(clientResponse.rawStatusCode()).willReturn(-1); + HttpClientResponse response = new HttpClientResponse(clientResponse); Integer statusCode = response.statusCode(); @@ -41,8 +40,7 @@ public class TraceExchangeFilterFunctionHttpClientResponseTests { public void should_return_status_code_when_valid_status_code_is_returned() { ClientResponse clientResponse = BDDMockito.mock(ClientResponse.class); BDDMockito.given(clientResponse.rawStatusCode()).willReturn(200); - TraceExchangeFilterFunction.HttpClientResponse response = new TraceExchangeFilterFunction.HttpClientResponse( - clientResponse); + HttpClientResponse response = new HttpClientResponse(clientResponse); Integer statusCode = response.statusCode(); diff --git a/tests/spring-cloud-sleuth-instrumentation-reactor-tests/pom.xml b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/pom.xml index 2e07b1f31..85fd25178 100644 --- a/tests/spring-cloud-sleuth-instrumentation-reactor-tests/pom.xml +++ b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/pom.xml @@ -59,6 +59,17 @@ org.springframework.cloud spring-cloud-starter-sleuth + + io.zipkin.brave + brave-instrumentation-http-tests + test + + + org.eclipse.jetty + * + + + org.springframework.boot spring-boot-starter-test 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 new file mode 100644 index 000000000..9d113a5bf --- /dev/null +++ b/tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/WebClientBraveTests.java @@ -0,0 +1,180 @@ +/* + * Copyright 2013-2019 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.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; +import org.springframework.http.client.reactive.ReactorClientHttpConnector; +import org.springframework.web.reactive.function.BodyInserters; +import org.springframework.web.reactive.function.client.ClientResponse; +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(); + } + + /** + * This uses Spring to instrument the {@link WebClient} using a + * {@link BeanPostProcessor}. + */ + @Override + protected WebClient 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(); + } + + @Override + protected void closeClient(WebClient client) { + // WebClient is not Closeable + } + + @Override + protected void get(WebClient client, String pathIncludingQuery) { + client.get().uri(pathIncludingQuery).exchange().block(); + } + + @Override + protected void post(WebClient client, String pathIncludingQuery, String body) { + client.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(); + + 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(); + } + }); + } + + @Test + @Ignore("TODO: reactor/reactor-netty#1000") + @Override + public void redirect() { + } + + @Test + @Ignore("WebClient has no portable function to retrieve the server address") + @Override + public void reportsServerAddress() { + } + + /** + * This fakes auto-configuration which wouldn't configure reactor's trace + * instrumentation. + */ + @Configuration + static class WebClientBuilderConfiguration { + + @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); + } + + @Bean + WebClient.Builder webClientBuilder(HttpClient httpClient) { + return WebClient.builder() + .clientConnector(new ReactorClientHttpConnector(httpClient)); + } + + } + +} From 547a124d41e589d7c53ec3d9f577eb166babfcc0 Mon Sep 17 00:00:00 2001 From: Adrian Cole Date: Mon, 24 Feb 2020 15:46:35 +0800 Subject: [PATCH 2/4] Adds Brave tests for reactor-netty (#1554) --- spring-cloud-sleuth-core/pom.xml | 5 + .../ReactorNettyHttpClientBraveTests.java | 187 ++++++++++++++++++ 2 files changed, 192 insertions(+) create mode 100644 spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/ReactorNettyHttpClientBraveTests.java diff --git a/spring-cloud-sleuth-core/pom.xml b/spring-cloud-sleuth-core/pom.xml index 9d7157d31..af72b90f3 100644 --- a/spring-cloud-sleuth-core/pom.xml +++ b/spring-cloud-sleuth-core/pom.xml @@ -335,6 +335,11 @@ spring-boot-starter-test test + + io.zipkin.brave + brave-instrumentation-http-tests + test + com.netflix.archaius archaius-core diff --git a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/ReactorNettyHttpClientBraveTests.java b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/ReactorNettyHttpClientBraveTests.java new file mode 100644 index 000000000..fe84c6529 --- /dev/null +++ b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/ReactorNettyHttpClientBraveTests.java @@ -0,0 +1,187 @@ +/* + * Copyright 2013-2019 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.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; +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"); + } + + /** + * This uses Spring to instrument the {@link HttpClient} using a + * {@link BeanPostProcessor}. + */ + @Override + protected HttpClient newClient(int port) { + AnnotationConfigApplicationContext result = new AnnotationConfigApplicationContext(); + result.registerBean(HttpTracing.class, () -> httpTracing); + result.registerBean(HttpClient.class, + ReactorNettyHttpClientBraveTests::testHttpClient); + result.register(HttpClientBeanPostProcessor.class); + result.refresh(); + return result.getBean(HttpClient.class).baseUrl("http://127.0.0.1:" + port); + } + + static HttpClient testHttpClient() { + return HttpClient.create() + .tcpConfiguration(tcpClient -> tcpClient + .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 1000) + .doOnConnected(conn -> conn + .addHandler(new ReadTimeoutHandler(1, TimeUnit.SECONDS)))) + .followRedirect(true); + } + + @Override + protected void closeClient(HttpClient client) { + // HttpClient is not Closeable + } + + @Override + protected void get(HttpClient client, String pathIncludingQuery) { + client.get().uri(pathIncludingQuery).response().block(); + } + + @Test + @Ignore("TODO: consider integrating TracingMapConnect with ScopePassingSpanSubscriber") + @Override + public void callbackContextIsFromInvocationTime() { + } + + @Test + @Ignore("TODO: reactor/reactor-netty#1000") + @Override + public void redirect() { + } + + @Test + @Ignore("TODO: reactor/reactor-netty#1000") + @Override + public void supportsPortableCustomization() { + } + + @Test + @Ignore("TODO: reactor/reactor-netty#1000") + @Override + public void post() { + } + + @Test + @Ignore("TODO: reactor/reactor-netty#1000") + @Override + public void customSampler() { + } + + @Test + @Ignore("TODO: reactor/reactor-netty#1000") + @Override + public void httpPathTagExcludesQueryParams() { + } + + @Override + protected void post(HttpClient client, String pathIncludingQuery, String body) { + client.post().send(ByteBufFlux.fromString(Mono.just(body))) + .uri(pathIncludingQuery).response().block(); + } + + @Override + protected void getAsync(HttpClient client, String path, Callback callback) { + Mono request = client.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(); + } + }); + } + +} From 0446918fcf0ad8482039093c7d43a68967baf620 Mon Sep 17 00:00:00 2001 From: Adrian Cole Date: Wed, 26 Feb 2020 15:11:18 +0800 Subject: [PATCH 3/4] Ensures spring context is closed in brave tests (#1572) --- .../client/HttpClientBeanPostProcessor.java | 1 + .../TraceWebClientBeanPostProcessor.java | 5 +- .../ReactorNettyHttpClientBraveTests.java | 106 +++++------------- .../web/client/TestCallbackSubscriber.java | 102 +++++++++++++++++ .../web/client/WebClientBraveTests.java | 105 ++++------------- 5 files changed, 159 insertions(+), 160 deletions(-) rename {spring-cloud-sleuth-core => tests/spring-cloud-sleuth-instrumentation-reactor-tests}/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/ReactorNettyHttpClientBraveTests.java (52%) create mode 100644 tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/TestCallbackSubscriber.java 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 From 7262790717c02c4d3451252105c2b388f06deb48 Mon Sep 17 00:00:00 2001 From: Adrian Cole Date: Wed, 26 Feb 2020 16:43:14 +0800 Subject: [PATCH 4/4] Performs minimal work to update to Brave 5.10 (#1573) A later PR will handle new features/deprecation --- benchmarks/pom.xml | 2 +- pom.xml | 2 +- spring-cloud-sleuth-dependencies/pom.xml | 2 +- .../ReactorNettyHttpClientBraveTests.java | 23 ++++++++++++------- ...r.java => TestHttpCallbackSubscriber.java} | 18 ++++++++++----- .../web/client/WebClientBraveTests.java | 11 +++++++-- 6 files changed, 39 insertions(+), 19 deletions(-) rename tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/{TestCallbackSubscriber.java => TestHttpCallbackSubscriber.java} (82%) diff --git a/benchmarks/pom.xml b/benchmarks/pom.xml index d07959a23..3b92456b6 100644 --- a/benchmarks/pom.xml +++ b/benchmarks/pom.xml @@ -33,7 +33,7 @@ 1.8 1.8 2.3.0.BUILD-SNAPSHOT - 5.9.5 + 5.10.0 3.14.6 diff --git a/pom.xml b/pom.xml index 207d75c98..e2efb8d59 100644 --- a/pom.xml +++ b/pom.xml @@ -257,7 +257,7 @@ Horsham.SR1 2.2.2.BUILD-SNAPSHOT 2.2.2.BUILD-SNAPSHOT - 5.9.5 + 5.10.0 2.1.7.RELEASE 2.2.2.BUILD-SNAPSHOT false diff --git a/spring-cloud-sleuth-dependencies/pom.xml b/spring-cloud-sleuth-dependencies/pom.xml index 11cd57f54..491afd756 100644 --- a/spring-cloud-sleuth-dependencies/pom.xml +++ b/spring-cloud-sleuth-dependencies/pom.xml @@ -31,7 +31,7 @@ spring-cloud-sleuth-dependencies Spring Cloud Sleuth Dependencies - 5.9.5 + 5.10.0 0.35.1 3.4.1 diff --git a/tests/spring-cloud-sleuth-instrumentation-reactor-tests/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 index e526f7188..af2d8ded7 100644 --- a/tests/spring-cloud-sleuth-instrumentation-reactor-tests/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 @@ -83,12 +83,6 @@ public class ReactorNettyHttpClientBraveTests public void callbackContextIsFromInvocationTime() { } - @Test - @Ignore("TODO: False negative due to NPE reading context: remove after Brave 5.10") - @Override - public void asyncRootSpan() { - } - @Test @Ignore("TODO: reactor/reactor-netty#1000") @Override @@ -101,6 +95,13 @@ public class ReactorNettyHttpClientBraveTests public void supportsPortableCustomization() { } + @Test + @Ignore("TODO: reactor/reactor-netty#1000") + @Override + @Deprecated + public void supportsDeprecatedPortableCustomization() { + } + @Test @Ignore("TODO: reactor/reactor-netty#1000") @Override @@ -119,6 +120,12 @@ public class ReactorNettyHttpClientBraveTests public void httpPathTagExcludesQueryParams() { } + @Test + @Ignore("HttpClient has no function to retrieve the wire request from the response") + @Override + public void readsRequestAtResponseTime() { + } + @Override protected void post(AnnotationConfigApplicationContext context, String pathIncludingQuery, String body) { @@ -129,11 +136,11 @@ public class ReactorNettyHttpClientBraveTests @Override protected void getAsync(AnnotationConfigApplicationContext context, String path, - Callback callback) { + Callback callback) { Mono request = context.getBean(HttpClient.class).get() .uri(path).response(); - TestCallbackSubscriber.subscribe(request, callback); + TestHttpCallbackSubscriber.subscribe(request, r -> r.status().code(), 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/TestHttpCallbackSubscriber.java similarity index 82% rename from tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/TestCallbackSubscriber.java rename to tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/TestHttpCallbackSubscriber.java index 57569eeaa..b56557f03 100644 --- 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/TestHttpCallbackSubscriber.java @@ -17,6 +17,7 @@ package org.springframework.cloud.sleuth.instrument.web.client; import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Function; import org.reactivestreams.Subscription; import reactor.core.CoreSubscriber; @@ -34,17 +35,22 @@ import zipkin2.Callback; * The implementation forwards signals to the supplied {@link Callback}, enforcing * assumptions about a non-empty, {@link Mono} subscription. */ -final class TestCallbackSubscriber implements CoreSubscriber { +final class TestHttpCallbackSubscriber implements CoreSubscriber { - static void subscribe(Mono mono, Callback callback) { - mono.subscribe(new TestCallbackSubscriber<>(callback)); + static void subscribe(Mono mono, Function statusCodeFunction, + Callback callback) { + mono.subscribe(new TestHttpCallbackSubscriber<>(statusCodeFunction, callback)); } - final Callback callback; + final Function statusCodeFunction; + + final Callback callback; final AtomicReference ref = new AtomicReference<>(); - private TestCallbackSubscriber(Callback callback) { + private TestHttpCallbackSubscriber(Function statusCodeFunction, + Callback callback) { + this.statusCodeFunction = statusCodeFunction; this.callback = callback; } @@ -63,7 +69,7 @@ final class TestCallbackSubscriber implements CoreSubscriber { @Override public void onNext(T t) { if (ref.getAndSet(null) != null) { - callback.onSuccess(null /* because Void */); + callback.onSuccess(statusCodeFunction.apply(t)); } else { // This is a Mono, which doesn't signal onNext() twice. If we reach here, 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 6f89a6e2d..d5859a3d5 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 @@ -75,10 +75,11 @@ public class WebClientBraveTests @Override protected void getAsync(AnnotationConfigApplicationContext context, String path, - Callback callback) { + Callback callback) { Mono request = client(context).get().uri(path).exchange(); - TestCallbackSubscriber.subscribe(request, callback); + TestHttpCallbackSubscriber.subscribe(request, ClientResponse::rawStatusCode, + callback); } @Test @@ -93,6 +94,12 @@ public class WebClientBraveTests public void reportsServerAddress() { } + @Test + @Ignore("TODO: maybe refactor as an ExchangeFilterFunction to get the request from response") + @Override + public void readsRequestAtResponseTime() { + } + WebClient client(AnnotationConfigApplicationContext context) { return context.getBean(WebClient.Builder.class) .baseUrl("http://127.0.0.1:" + server.getPort()).build();