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

WIP

Browse Source
This commit is contained in:
Marcin Grzejszczak
2018-10-19 11:43:17 +02:00
parent ef5efd8041
commit 6e80f3f06a
4 changed files with 312 additions and 270 deletions
Download Patch File Download Diff File Expand all files Collapse all files

121
spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/async/ExecutorBeanPostProcessor.java
View File

@@ -20,7 +20,11 @@ import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.lang.reflect.Modifier;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.function.Function;
import java.util.function.Supplier;
import org.aopalliance.aop.Advice;
import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
import org.apache.commons.logging.Log;
@@ -64,42 +68,64 @@ class ExecutorBeanPostProcessor implements BeanPostProcessor {
@Override
public Object postProcessAfterInitialization(Object bean, String beanName)
throws BeansException {
if (bean instanceof Executor && !(bean instanceof ThreadPoolTaskExecutor)) {
Method execute = ReflectionUtils.findMethod(bean.getClass(), "execute",
Runnable.class);
boolean methodFinal = Modifier.isFinal(execute.getModifiers());
boolean classFinal = Modifier.isFinal(bean.getClass().getModifiers());
boolean cglibProxy = !methodFinal && !classFinal;
Executor executor = (Executor) bean;
try {
return createProxy(bean, cglibProxy, executor);
}
catch (AopConfigException ex) {
if (cglibProxy) {
if (log.isDebugEnabled()) {
log.debug(
"Exception occurred while trying to create a proxy, falling back to JDK proxy",
ex);
}
return createProxy(bean, false, executor);
}
throw ex;
}
}
else if (bean instanceof ThreadPoolTaskExecutor) {
if (bean instanceof ThreadPoolTaskExecutor) {
if (isProxyNeeded(beanName)) {
boolean classFinal = Modifier.isFinal(bean.getClass().getModifiers());
boolean cglibProxy = !classFinal;
ThreadPoolTaskExecutor executor = (ThreadPoolTaskExecutor) bean;
return createThreadPoolTaskExecutorProxy(bean, cglibProxy, executor);
return wrapThreadPoolTaskExecutor(bean);
}
else {
log.info("Not instrumenting bean " + beanName);
}
} else if (bean instanceof ExecutorService) {
if (isProxyNeeded(beanName)) {
return wrapExecutorService(bean);
}
else {
log.info("Not instrumenting bean " + beanName);
}
} else if (bean instanceof Executor) {
return wrapExecutor(bean);
}
return bean;
}
private Object wrapExecutor(Object bean) {
Method execute = ReflectionUtils.findMethod(bean.getClass(), "execute",
Runnable.class);
boolean methodFinal = Modifier.isFinal(execute.getModifiers());
boolean classFinal = Modifier.isFinal(bean.getClass().getModifiers());
boolean cglibProxy = !methodFinal && !classFinal;
Executor executor = (Executor) bean;
try {
return createProxy(bean, cglibProxy,
new ExecutorMethodInterceptor(executor, this.beanFactory));
}
catch (AopConfigException ex) {
if (cglibProxy) {
if (log.isDebugEnabled()) {
log.debug(
"Exception occurred while trying to create a proxy, falling back to JDK proxy",
ex);
}
return createProxy(bean, false, new ExecutorMethodInterceptor(executor, this.beanFactory));
}
throw ex;
}
}
private Object wrapThreadPoolTaskExecutor(Object bean) {
boolean classFinal = Modifier.isFinal(bean.getClass().getModifiers());
boolean cglibProxy = !classFinal;
ThreadPoolTaskExecutor executor = (ThreadPoolTaskExecutor) bean;
return createThreadPoolTaskExecutorProxy(bean, cglibProxy, executor);
}
private Object wrapExecutorService(Object bean) {
boolean classFinal = Modifier.isFinal(bean.getClass().getModifiers());
boolean cglibProxy = !classFinal;
ExecutorService executor = (ExecutorService) bean;
return createExecutorServiceProxy(bean, cglibProxy, executor);
}
boolean isProxyNeeded(String beanName) {
SleuthAsyncProperties sleuthAsyncProperties = asyncConfigurationProperties();
return !sleuthAsyncProperties.getIgnoredBeans().contains(beanName);
@@ -107,24 +133,47 @@ class ExecutorBeanPostProcessor implements BeanPostProcessor {
Object createThreadPoolTaskExecutorProxy(Object bean, boolean cglibProxy,
ThreadPoolTaskExecutor executor) {
return getProxiedObject(bean, cglibProxy, executor,
() -> new LazyTraceThreadPoolTaskExecutor(this.beanFactory, executor));
}
Object createExecutorServiceProxy(Object bean, boolean cglibProxy,
ExecutorService executor) {
return getProxiedObject(bean, cglibProxy, executor,
() -> new TraceableExecutorService(this.beanFactory, executor));
}
private Object getProxiedObject(Object bean, boolean cglibProxy, Executor executor,
Supplier<Executor> supplier) {
ProxyFactoryBean factory = new ProxyFactoryBean();
factory.setProxyTargetClass(cglibProxy);
factory.addAdvice(new ExecutorMethodInterceptor<ThreadPoolTaskExecutor>(executor,
factory.addAdvice(new ExecutorMethodInterceptor<Executor>(executor,
this.beanFactory) {
@Override
Executor executor(BeanFactory beanFactory, ThreadPoolTaskExecutor executor) {
return new LazyTraceThreadPoolTaskExecutor(beanFactory, executor);
<T extends Executor> T executor(BeanFactory beanFactory, T executor) {
return (T) supplier.get();
}
});
factory.setTarget(bean);
try {
return getObject(factory);
} catch (Exception e) {
if (log.isDebugEnabled()) {
log.debug("Exception occurred while trying to get a proxy. Will fallback to a different implementation", e);
}
return supplier.get();
}
}
Object getObject(ProxyFactoryBean factory) {
return factory.getObject();
}
@SuppressWarnings("unchecked")
Object createProxy(Object bean, boolean cglibProxy, Executor executor) {
Object createProxy(Object bean, boolean cglibProxy, Advice advice) {
ProxyFactoryBean factory = new ProxyFactoryBean();
factory.setProxyTargetClass(cglibProxy);
factory.addAdvice(new ExecutorMethodInterceptor(executor, this.beanFactory));
factory.addAdvice(advice);
factory.setTarget(bean);
if (JavaVersion.current().isJava11Compatible()) {
if (log.isDebugEnabled()) {
@@ -134,7 +183,7 @@ class ExecutorBeanPostProcessor implements BeanPostProcessor {
});
}
return factory.getObject();
return getObject(factory);
}
private SleuthAsyncProperties asyncConfigurationProperties() {
@@ -166,7 +215,7 @@ class ExecutorMethodInterceptor<T extends Executor> implements MethodInterceptor
@Override
public Object invoke(MethodInvocation invocation) throws Throwable {
Executor executor = executor(this.beanFactory, this.delegate);
T executor = executor(this.beanFactory, this.delegate);
Method methodOnTracedBean = getMethod(invocation, executor);
if (methodOnTracedBean != null) {
try {
@@ -187,8 +236,8 @@ class ExecutorMethodInterceptor<T extends Executor> implements MethodInterceptor
method.getParameterTypes());
}
Executor executor(BeanFactory beanFactory, T executor) {
return new LazyTraceExecutor(beanFactory, executor);
<T extends Executor> T executor(BeanFactory beanFactory, T executor) {
return (T) new LazyTraceExecutor(beanFactory, executor);
}
}

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

@@ -19,10 +19,7 @@ package org.springframework.cloud.sleuth.instrument.web.client;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BiFunction;
import java.util.function.BiConsumer;
import brave.Span;
import brave.Tracer;
@@ -34,18 +31,17 @@ import brave.propagation.Propagation;
import brave.propagation.TraceContext;
import brave.spring.web.TracingClientHttpRequestInterceptor;
import io.netty.handler.codec.http.HttpHeaders;
import io.netty.handler.codec.http.HttpMethod;
import io.netty.handler.codec.http.HttpVersion;
import io.netty.handler.codec.http.cookie.Cookie;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import io.netty.util.Attribute;
import io.netty.util.AttributeKey;
import org.apache.http.impl.client.HttpClientBuilder;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Pointcut;
import org.reactivestreams.Publisher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.netty.Connection;
import reactor.netty.http.client.HttpClient;
import reactor.netty.http.client.HttpClientRequest;
import reactor.netty.http.client.HttpClientResponse;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.ListableBeanFactory;
@@ -70,11 +66,6 @@ import org.springframework.http.client.ClientHttpResponse;
import org.springframework.security.oauth2.client.OAuth2RestTemplate;
import org.springframework.web.client.RestTemplate;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Mono;
import reactor.netty.NettyOutbound;
import reactor.netty.http.client.HttpClient;
import reactor.netty.http.client.HttpClientRequest;
import reactor.netty.http.client.HttpClientResponse;
/**
* {@link org.springframework.boot.autoconfigure.EnableAutoConfiguration
@@ -165,8 +156,8 @@ public class TraceWebClientAutoConfiguration {
static class NettyConfiguration {
@Bean
public NettyAspect traceNetyAspect(HttpTracing httpTracing) {
return new NettyAspect(httpTracing);
public HttpClientBeanPostProcessor httpClientBeanPostProcessor(BeanFactory beanFactory) {
return new HttpClientBeanPostProcessor(beanFactory);
}
}
@@ -318,40 +309,31 @@ class LazyTracingClientHttpRequestInterceptor implements ClientHttpRequestInterc
}
return this.interceptor;
}
}
@Aspect
class NettyAspect {
private final TracingHttpClientInstrumentation instrumentation;
class HttpClientBeanPostProcessor implements BeanPostProcessor {
NettyAspect(HttpTracing httpTracing) {
this.instrumentation = TracingHttpClientInstrumentation.create(httpTracing);
private final BeanFactory beanFactory;
HttpClientBeanPostProcessor(BeanFactory beanFactory) {
this.beanFactory = beanFactory;
}
@Pointcut("execution(public * reactor.netty.http.client.HttpClient.RequestSender.send(..)) && args(function)")
private void anyHttpClientRequestSending(
BiFunction<? super HttpClientRequest, ? super NettyOutbound, ? extends Publisher<Void>> function) {
} // NOSONAR
@Around("anyHttpClientRequestSending(function)")
public Object wrapHttpClientRequestSending(ProceedingJoinPoint pjp,
BiFunction<? super HttpClientRequest, ? super NettyOutbound, ? extends Publisher<Void>> function)
throws Throwable {
return Mono.defer(() -> {
try {
return this.instrumentation.wrapHttpClientRequestSending(pjp, function);
}
catch (Throwable e) {
return Mono.error(e);
}
});
@Override
public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
if (bean instanceof HttpClient) {
return ((HttpClient) bean)
.doOnRequest(TracingDoOnRequest.create(this.beanFactory))
.doOnResponse(TracingDoOnResponse.create(this.beanFactory));
}
return bean;
}
}
class TracingHttpClientInstrumentation {
class TracingDoOnRequest implements BiConsumer<HttpClientRequest, Connection> {
private static final Logger log = LoggerFactory.getLogger(TracingDoOnRequest.class);
static final Propagation.Setter<HttpHeaders, String> SETTER = new Propagation.Setter<HttpHeaders, String>() {
@Override
@@ -378,206 +360,95 @@ class TracingHttpClientInstrumentation {
}
};
private static final Log log = LogFactory
.getLog(TracingHttpClientInstrumentation.class);
final Tracer tracer;
final HttpClientHandler<HttpClientRequest, HttpClientResponse> handler;
final TraceContext.Injector<HttpHeaders> injector;
final HttpTracing httpTracing;
TracingHttpClientInstrumentation(HttpTracing httpTracing) {
TracingDoOnRequest(HttpTracing httpTracing) {
this.tracer = httpTracing.tracing().tracer();
this.handler = HttpClientHandler.create(httpTracing, new HttpAdapter());
this.injector = httpTracing.tracing().propagation().injector(SETTER);
this.httpTracing = httpTracing;
}
static TracingHttpClientInstrumentation create(HttpTracing httpTracing) {
return new TracingHttpClientInstrumentation(httpTracing);
static TracingDoOnRequest create(BeanFactory beanFactory) {
return new TracingDoOnRequest(beanFactory.getBean(HttpTracing.class));
}
Mono<HttpClientResponse> wrapHttpClientRequestSending(ProceedingJoinPoint pjp,
BiFunction<? super HttpClientRequest, ? super NettyOutbound, ? extends Publisher<Void>> function)
throws Throwable {
// add headers and set CS
@Override
public void accept(HttpClientRequest req, Connection connection) {
final Span currentSpan = this.tracer.currentSpan();
final AtomicReference<Span> span = new AtomicReference<>();
BiFunction<HttpClientRequest, NettyOutbound, Publisher<Void>> combinedFunction = (
req, nettyOutbound) -> {
try (Tracer.SpanInScope spanInScope = this.tracer
.withSpanInScope(currentSpan)) {
io.netty.handler.codec.http.HttpHeaders originalHeaders = req
.requestHeaders().copy();
io.netty.handler.codec.http.HttpHeaders tracedHeaders = req
.requestHeaders();
span.set(this.handler.handleSend(this.injector, tracedHeaders, req));
if (log.isDebugEnabled()) {
log.debug("Handled send of " + span.get());
}
io.netty.handler.codec.http.HttpHeaders addedHeaders = tracedHeaders
.copy();
originalHeaders.forEach(header -> addedHeaders.remove(header.getKey()));
try (Tracer.SpanInScope clientInScope = this.tracer
.withSpanInScope(span.get())) {
if (log.isDebugEnabled()) {
log.debug("Created a new client span for Netty client");
}
return handle(function,
new TracedHttpClientRequest(req, addedHeaders),
nettyOutbound);
}
try (Tracer.SpanInScope spanInScope = this.tracer
.withSpanInScope(currentSpan)) {
if (log.isDebugEnabled()) {
log.debug("Wrapping do on request");
}
};
// run
Mono<HttpClientResponse> responseMono = (Mono<HttpClientResponse>) pjp
.proceed(new Object[] { combinedFunction });
// get response
return responseMono.doOnSuccessOrError((httpClientResponse, throwable) -> {
try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(span.get())) {
// status codes and CR
if (span.get() != null) {
this.handler.handleReceive(httpClientResponse, throwable, span.get());
if (log.isDebugEnabled()) {
log.debug("Setting client sent spans");
}
}
Span span = this.handler.handleSend(this.injector, req
.requestHeaders(), req);
Attribute<Object> attribute = connection.channel()
.attr(AttributeKey.valueOf("span"));
attribute.set(span);
}
}
}
class TracingDoOnResponse implements BiConsumer<HttpClientResponse, Connection> {
private static final Logger log = LoggerFactory.getLogger(TracingDoOnResponse.class);
final Tracer tracer;
final HttpClientHandler<HttpClientRequest, HttpClientResponse> handler;
TracingDoOnResponse(HttpTracing httpTracing) {
this.tracer = httpTracing.tracing().tracer();
this.handler = HttpClientHandler.create(httpTracing, new HttpAdapter());
}
static TracingDoOnResponse create(BeanFactory beanFactory) {
return new TracingDoOnResponse(beanFactory.getBean(HttpTracing.class));
}
@Override
public void accept(HttpClientResponse httpClientResponse, Connection connection) {
Attribute<Object> spanAttr = connection.channel().attr(AttributeKey.valueOf("span"));
Span span = (Span) spanAttr.get();
if (span == null) {
return;
}
try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(span)) {
if (log.isDebugEnabled()) {
log.debug("Setting client sent spans");
}
});
// status codes and CR
// TODO: Add throwable
this.handler.handleReceive(httpClientResponse, null, span);
}
}
}
class HttpAdapter
extends brave.http.HttpClientAdapter<HttpClientRequest, HttpClientResponse> {
@Override
public String method(HttpClientRequest request) {
return request.method().name();
}
private Publisher<Void> handle(
BiFunction<? super HttpClientRequest, ? super NettyOutbound, ? extends Publisher<Void>> handler,
HttpClientRequest req, NettyOutbound nettyOutbound) {
if (handler != null) {
return handler.apply(req, nettyOutbound);
}
return nettyOutbound;
@Override
public String url(HttpClientRequest request) {
return request.uri();
}
/**
* The `org.springframework.cloud.gateway.filter.NettyRoutingFilter` in SC Gateway is
* adding only these headers that were set when the request came in. That means that
* adding any additional headers (via instrumentation) is completely ignored. That's
* why we're wrapping the `HttpClientRequest` in such a wrapper that when `setHeaders`
* is called (that clears any current headers), will also add the tracing headers
*/
static class TracedHttpClientRequest implements HttpClientRequest {
private final io.netty.handler.codec.http.HttpHeaders addedHeaders;
private HttpClientRequest delegate;
TracedHttpClientRequest(HttpClientRequest delegate, HttpHeaders addedHeaders) {
this.delegate = delegate;
this.addedHeaders = addedHeaders;
}
@Override
public HttpClientRequest addCookie(Cookie cookie) {
this.delegate = this.delegate.addCookie(cookie);
return this;
}
@Override
public HttpClientRequest addHeader(CharSequence name, CharSequence value) {
this.delegate = this.delegate.addHeader(name, value);
return this;
}
@Override
public HttpClientRequest header(CharSequence name, CharSequence value) {
this.delegate = this.delegate.header(name, value);
return this;
}
@Override
public HttpClientRequest headers(HttpHeaders headers) {
HttpHeaders copy = headers.copy();
copy.add(this.addedHeaders);
this.delegate = this.delegate.headers(copy);
return this;
}
@Override
public boolean isFollowRedirect() {
return this.delegate.isFollowRedirect();
}
@Override
public String[] redirectedFrom() {
return this.delegate.redirectedFrom();
}
@Override
public HttpHeaders requestHeaders() {
return this.delegate.requestHeaders();
}
@Override
public Map<CharSequence, Set<Cookie>> cookies() {
return this.delegate.cookies();
}
@Override
public boolean isKeepAlive() {
return this.delegate.isKeepAlive();
}
@Override
public boolean isWebsocket() {
return this.delegate.isWebsocket();
}
@Override
public HttpMethod method() {
return this.delegate.method();
}
@Override
public String path() {
return this.delegate.path();
}
@Override
public String uri() {
return this.delegate.uri();
}
@Override
public HttpVersion version() {
return this.delegate.version();
}
@Override
public String requestHeader(HttpClientRequest request, String name) {
Object result = request.requestHeaders().get(name);
return result != null ? result.toString() : "";
}
static final class HttpAdapter
extends brave.http.HttpClientAdapter<HttpClientRequest, HttpClientResponse> {
@Override
public String method(HttpClientRequest request) {
return request.method().name();
}
@Override
public String url(HttpClientRequest request) {
return request.uri();
}
@Override
public String requestHeader(HttpClientRequest request, String name) {
Object result = request.requestHeaders().get(name);
return result != null ? result.toString() : "";
}
@Override
public Integer statusCode(HttpClientResponse response) {
return response.status().code();
}
@Override
public Integer statusCode(HttpClientResponse response) {
return response.status().code();
}
}

140
spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/async/ExecutorBeanPostProcessorTests.java
View File

@@ -16,22 +16,36 @@
package org.springframework.cloud.sleuth.instrument.async;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import brave.Tracer;
import brave.Tracing;
import org.aopalliance.aop.Advice;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.BDDMockito;
import org.mockito.Mock;
import org.mockito.Mockito;
import org.mockito.junit.MockitoJUnitRunner;
import org.springframework.aop.framework.AopConfigException;
import org.springframework.aop.framework.ProxyFactoryBean;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.cloud.sleuth.DefaultSpanNamer;
import org.springframework.cloud.sleuth.SpanName;
import org.springframework.cloud.sleuth.SpanNamer;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
@@ -49,6 +63,7 @@ public class ExecutorBeanPostProcessorTests {
@Mock
BeanFactory beanFactory;
Tracing tracing = Tracing.newBuilder().build();
private SleuthAsyncProperties sleuthAsyncProperties;
@@ -57,6 +72,15 @@ public class ExecutorBeanPostProcessorTests {
this.sleuthAsyncProperties = new SleuthAsyncProperties();
Mockito.when(beanFactory.getBean(SleuthAsyncProperties.class))
.thenReturn(this.sleuthAsyncProperties);
Mockito.when(beanFactory.getBean(Tracing.class))
.thenReturn(this.tracing);
Mockito.when(beanFactory.getBean(SpanNamer.class))
.thenReturn(new DefaultSpanNamer());
}
@After
public void clear() {
this.tracing.close();
}
@Test
@@ -78,30 +102,32 @@ public class ExecutorBeanPostProcessorTests {
}
@Test
public void should_create_jdk_proxy_when_cglib_fails_to_be_done() throws Exception {
public void should_fallback_to_sleuth_implementation_when_cglib_cannot_be_created() throws Exception {
ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor();
Object o = new ExecutorBeanPostProcessor(this.beanFactory)
.postProcessAfterInitialization(service, "foo");
then(o).isInstanceOf(ScheduledExecutorService.class);
then(ClassUtils.isCglibProxy(o)).isFalse();
then(o).isInstanceOf(TraceableExecutorService.class);
service.shutdown();
}
@Test
public void should_throw_exception_when_it_is_not_possible_to_create_any_proxy()
public void should_fallback_to_default_implementation_when_exception_thrown()
throws Exception {
ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor();
ExecutorBeanPostProcessor bpp = new ExecutorBeanPostProcessor(this.beanFactory) {
@Override
Object createProxy(Object bean, boolean cglibProxy, Executor executor) {
Object createProxy(Object bean, boolean cglibProxy, Advice advice) {
throw new AopConfigException("foo");
}
};
thenThrownBy(() -> bpp.postProcessAfterInitialization(service, "foo"))
.isInstanceOf(AopConfigException.class).hasMessage("foo");
Object wrappedService = bpp.postProcessAfterInitialization(service, "foo");
then(wrappedService).isInstanceOf(TraceableExecutorService.class);
service.shutdown();
}
@@ -120,7 +146,7 @@ public class ExecutorBeanPostProcessorTests {
}
@Test
public void should_throw_exception_when_it_is_not_possible_to_create_any_proxyfor_ThreadPoolTaskExecutor()
public void should_throw_exception_when_it_is_not_possible_to_create_any_proxy_for_ThreadPoolTaskExecutor()
throws Exception {
ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
ExecutorBeanPostProcessor bpp = new ExecutorBeanPostProcessor(this.beanFactory) {
@@ -135,6 +161,104 @@ public class ExecutorBeanPostProcessorTests {
.isInstanceOf(AopConfigException.class).hasMessage("foo");
}
@Test
public void should_fallback_to_sleuth_impl_when_it_is_not_possible_to_create_any_proxy_for_ExecutorService()
throws Exception {
ExecutorService service = BDDMockito.mock(ExecutorService.class);
ExecutorBeanPostProcessor bpp = new ExecutorBeanPostProcessor(this.beanFactory) {
@Override
Object getObject(ProxyFactoryBean factory) {
throw new AopConfigException("foo");
}
};
Object o = bpp.postProcessAfterInitialization(service, "foo");
then(o).isInstanceOf(TraceableExecutorService.class);
}
@Test
public void should_throw_exception_from_the_wrapped_object()
throws Exception {
ExecutorService service = exceptionThrowingExecutorService();
ExecutorBeanPostProcessor bpp = new ExecutorBeanPostProcessor(this.beanFactory);
ExecutorService o = (ExecutorService) bpp.postProcessAfterInitialization(service, "foo");
thenThrownBy(() -> o.submit((Callable<Object>) () -> "hello"))
.hasMessage("foo")
.isInstanceOf(IllegalStateException.class);
}
private ExecutorService exceptionThrowingExecutorService() {
return new ExecutorService() {
@Override
public void execute(Runnable command) {
}
@Override
public void shutdown() {
}
@Override
public List<Runnable> shutdownNow() {
return null;
}
@Override
public boolean isShutdown() {
return false;
}
@Override
public boolean isTerminated() {
return false;
}
@Override
public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
return false;
}
@Override
public <T> Future<T> submit(Callable<T> task) {
throw new IllegalStateException("foo");
}
@Override
public <T> Future<T> submit(Runnable task, T result) {
return null;
}
@Override
public Future<?> submit(Runnable task) {
return null;
}
@Override
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException {
return null;
}
@Override
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException {
return null;
}
@Override
public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException {
return null;
}
@Override
public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
return null;
}
};
}
@Test
public void proxy_is_not_needed() throws Exception {
this.sleuthAsyncProperties

2
spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/integration/WebClientTests.java
View File

@@ -280,8 +280,6 @@ public class WebClientTests {
then(this.reporter.getSpans()).isNotEmpty();
}
// TODO: Fix me
@Ignore("reactor is broken")
@Test
@SuppressWarnings("unchecked")
public void shouldAttachTraceIdWhenCallingAnotherServiceForNettyHttpClient()