SpringCloud/spring-cloud-sleuth
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch '2.2.x'

Browse Source
This commit is contained in:
Adrian Cole
2020-02-26 16:51:58 +08:00
parent 8f4d7dd8d8 7262790717
commit db2d9aeaaa
11 changed files with 566 additions and 266 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
benchmarks/pom.xml
View File

@@ -33,7 +33,7 @@
<maven.compiler.target>1.8</maven.compiler.target>
<maven.compiler.source>1.8</maven.compiler.source>
<spring-boot.version>2.3.0.BUILD-SNAPSHOT</spring-boot.version>
<brave.version>5.9.5</brave.version>
<brave.version>5.10.0</brave.version>
<okhttp.version>3.14.6</okhttp.version>
</properties>

2
pom.xml
View File

@@ -250,7 +250,7 @@
<spring-cloud-stream.version>Horsham.BUILD-SNAPSHOT</spring-cloud-stream.version>
<spring-cloud-netflix.version>3.0.0.BUILD-SNAPSHOT</spring-cloud-netflix.version>
<spring-cloud-openfeign.version>3.0.0.BUILD-SNAPSHOT</spring-cloud-openfeign.version>
<brave.version>5.9.3</brave.version>
<brave.version>5.10.0</brave.version>
<spring-security-boot-autoconfigure.version>2.1.7.RELEASE</spring-security-boot-autoconfigure.version>
<disable.nohttp.checks>false</disable.nohttp.checks>
<okhttp.version>3.14.6</okhttp.version>

5
spring-cloud-sleuth-core/pom.xml
View File

@@ -303,6 +303,11 @@
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.zipkin.brave</groupId>
<artifactId>brave-instrumentation-http-tests</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>com.squareup.okhttp3</groupId>
<artifactId>mockwebserver</artifactId>

1
spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/HttpClientBeanPostProcessor.java
View File

@@ -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);
}

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

@@ -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<ClientRequest.Builder, String> SETTER = new Propagation.Setter<ClientRequest.Builder, String>() {
@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> httpTracing;
final Function<? super Publisher<DataBuffer>, ? extends Publisher<DataBuffer>> scopePassingTransformer;
Tracer tracer;
HttpTracing httpTracing;
// Lazy initialized fields
HttpClientHandler<brave.http.HttpClientRequest, brave.http.HttpClientResponse> handler;
TraceContext.Injector<ClientRequest.Builder> 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<ClientResponse> 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<brave.http.HttpClientRequest, brave.http.HttpClientResponse> 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<ClientRequest.Builder> 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<ClientResponse> {
final ExchangeFunction next;
final ClientRequest request;
final Tracer tracer;
final HttpClientHandler<brave.http.HttpClientRequest, brave.http.HttpClientResponse> handler;
final TraceContext.Injector<ClientRequest.Builder> injector;
final Tracing tracing;
final CurrentTraceContext currentTraceContext;
final Function<? super Publisher<DataBuffer>, ? extends Publisher<DataBuffer>> 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,136 @@ 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<ClientResponse> {
final CoreSubscriber<? super ClientResponse> actual;
final Context context;
final Span span;
final HttpClientHandler<brave.http.HttpClientRequest, brave.http.HttpClientResponse> handler;
final Function<? super Publisher<DataBuffer>, ? extends Publisher<DataBuffer>> scopePassingTransformer;
final Tracing tracing;
boolean done;
WebClientTracerSubscriber(CoreSubscriber<? super ClientResponse> 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<ClientResponse> {
private final ClientRequest delegate;
final CoreSubscriber<? super ClientResponse> actual;
private final ClientRequest.Builder builder;
final Context context;
@Nullable
final TraceContext parent;
final Span clientSpan;
final HttpClientHandler<brave.http.HttpClientRequest, brave.http.HttpClientResponse> handler;
final Function<? super Publisher<DataBuffer>, ? extends Publisher<DataBuffer>> scopePassingTransformer;
final CurrentTraceContext currentTraceContext;
// TODO: this isn't implemented correctly. error and success could both be called
boolean done;
WebClientTracerSubscriber(CoreSubscriber<? super ClientResponse> 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 { // 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 + "]");
}
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 {
// TODO: is there a way to read the request at response time?
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 +370,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 +383,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);
}
}

10
spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/TraceExchangeFilterFunctionHttpClientResponseTests.java
View File

@@ -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();

2
spring-cloud-sleuth-dependencies/pom.xml
View File

@@ -31,7 +31,7 @@
<name>spring-cloud-sleuth-dependencies</name>
<description>Spring Cloud Sleuth Dependencies</description>
<properties>
<brave.version>5.9.5</brave.version>
<brave.version>5.10.0</brave.version>
<brave.opentracing.version>0.35.1</brave.opentracing.version>
<grpc.spring.boot.version>3.4.1</grpc.spring.boot.version>
</properties>

11
tests/spring-cloud-sleuth-instrumentation-reactor-tests/pom.xml
View File

@@ -59,6 +59,17 @@
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-sleuth</artifactId>
</dependency>
<dependency>
<groupId>io.zipkin.brave</groupId>
<artifactId>brave-instrumentation-http-tests</artifactId>
<scope>test</scope>
<exclusions>
<exclusion>
<groupId>org.eclipse.jetty</groupId>
<artifactId>*</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>

146
tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/ReactorNettyHttpClientBraveTests.java Normal file
View File

@@ -0,0 +1,146 @@
/*
* 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 brave.http.HttpTracing;
import brave.test.http.ITHttpAsyncClient;
import io.netty.channel.ChannelOption;
import io.netty.handler.timeout.ReadTimeoutHandler;
import org.junit.Ignore;
import org.junit.Test;
import reactor.core.publisher.Mono;
import reactor.netty.ByteBufFlux;
import reactor.netty.http.client.HttpClient;
import reactor.netty.http.client.HttpClientResponse;
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.
*/
// Function of spring context so that shutdown hooks happen!
public class ReactorNettyHttpClientBraveTests
extends ITHttpAsyncClient<AnnotationConfigApplicationContext> {
/**
* This uses Spring to instrument the {@link HttpClient} using a
* {@link BeanPostProcessor}.
*/
@Override
protected AnnotationConfigApplicationContext newClient(int port) {
AnnotationConfigApplicationContext result = new AnnotationConfigApplicationContext();
result.registerBean(HttpTracing.class, () -> httpTracing);
result.registerBean(HttpClient.class,
() -> testHttpClient().baseUrl("http://127.0.0.1:" + port));
result.register(HttpClientBeanPostProcessor.class);
result.refresh();
return result;
}
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(AnnotationConfigApplicationContext context) {
context.close(); // ensures shutdown hooks fire
}
@Override
protected void get(AnnotationConfigApplicationContext context,
String pathIncludingQuery) {
context.getBean(HttpClient.class).get().uri(pathIncludingQuery).response()
.block();
}
@Test
@Ignore("TODO: NPE reading context: 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
@Deprecated
public void supportsDeprecatedPortableCustomization() {
}
@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() {
}
@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) {
context.getBean(HttpClient.class).post()
.send(ByteBufFlux.fromString(Mono.just(body))).uri(pathIncludingQuery)
.response().block();
}
@Override
protected void getAsync(AnnotationConfigApplicationContext context, String path,
Callback<Integer> callback) {
Mono<HttpClientResponse> request = context.getBean(HttpClient.class).get()
.uri(path).response();
TestHttpCallbackSubscriber.subscribe(request, r -> r.status().code(), callback);
}
}

108
tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/TestHttpCallbackSubscriber.java Normal file
View File

@@ -0,0 +1,108 @@
/*
* 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 java.util.function.Function;
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.
*
* <p>
* The implementation forwards signals to the supplied {@link Callback}, enforcing
* assumptions about a non-empty, {@link Mono} subscription.
*/
final class TestHttpCallbackSubscriber<T> implements CoreSubscriber<T> {
static <T> void subscribe(Mono<T> mono, Function<T, Integer> statusCodeFunction,
Callback<Integer> callback) {
mono.subscribe(new TestHttpCallbackSubscriber<>(statusCodeFunction, callback));
}
final Function<T, Integer> statusCodeFunction;
final Callback<Integer> callback;
final AtomicReference<Subscription> ref = new AtomicReference<>();
private TestHttpCallbackSubscriber(Function<T, Integer> statusCodeFunction,
Callback<Integer> callback) {
this.statusCodeFunction = statusCodeFunction;
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(statusCodeFunction.apply(t));
}
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();
}
}

128
tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/WebClientBraveTests.java Normal file
View File

@@ -0,0 +1,128 @@
/*
* 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 brave.http.HttpTracing;
import brave.test.http.ITHttpAsyncClient;
import org.junit.Ignore;
import org.junit.Test;
import reactor.core.publisher.Mono;
import reactor.netty.http.client.HttpClient;
import zipkin2.Callback;
import org.springframework.beans.factory.config.BeanPostProcessor;
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.
*/
// Function of spring context so that shutdown hooks happen!
public class WebClientBraveTests
extends ITHttpAsyncClient<AnnotationConfigApplicationContext> {
/**
* This uses Spring to instrument the {@link WebClient} using a
* {@link BeanPostProcessor}.
*/
@Override
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;
}
@Override
protected void closeClient(AnnotationConfigApplicationContext context) {
context.close(); // ensures shutdown hooks fire
}
@Override
protected void get(AnnotationConfigApplicationContext context,
String pathIncludingQuery) {
client(context).get().uri(pathIncludingQuery).exchange().block();
}
@Override
protected void post(AnnotationConfigApplicationContext context,
String pathIncludingQuery, String body) {
client(context).post().uri(pathIncludingQuery).body(BodyInserters.fromValue(body))
.exchange().block();
}
@Override
protected void getAsync(AnnotationConfigApplicationContext context, String path,
Callback<Integer> callback) {
Mono<ClientResponse> request = client(context).get().uri(path).exchange();
TestHttpCallbackSubscriber.subscribe(request, ClientResponse::rawStatusCode,
callback);
}
@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() {
}
@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();
}
/**
* This fakes auto-configuration which wouldn't configure reactor's trace
* instrumentation.
*/
@Configuration
static class WebClientBuilderConfiguration {
@Bean
HttpClient httpClient() {
return ReactorNettyHttpClientBraveTests.testHttpClient();
}
@Bean
WebClient.Builder webClientBuilder(HttpClient httpClient) {
return WebClient.builder()
.clientConnector(new ReactorClientHttpConnector(httpClient));
}
}
}