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

Ensures that there is no double instrumentation for executor services

Browse Source 
fixes gh-1463
This commit is contained in:
Marcin Grzejszczak
2019-10-22 13:06:50 +02:00
parent f586661b78
commit 54a65bc9f3
2 changed files with 97 additions and 116 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -63,16 +63,14 @@ class ExecutorBeanPostProcessor implements BeanPostProcessor {
}
@Override
public Object postProcessBeforeInitialization(Object bean, String beanName)
throws BeansException {
public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
return bean;
}
@Override
public Object postProcessAfterInitialization(Object bean, String beanName)
throws BeansException {
if (bean instanceof ThreadPoolTaskExecutor
&& !(bean instanceof LazyTraceThreadPoolTaskExecutor)) {
public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
boolean alreadyTraced = alreadyTraced(bean);
if (bean instanceof ThreadPoolTaskExecutor && !alreadyTraced) {
if (isProxyNeeded(beanName)) {
return wrapThreadPoolTaskExecutor(bean);
}
@@ -80,8 +78,7 @@ class ExecutorBeanPostProcessor implements BeanPostProcessor {
log.info("Not instrumenting bean " + beanName);
}
}
else if (bean instanceof ExecutorService
&& !(bean instanceof TraceableExecutorService)) {
else if (bean instanceof ExecutorService && !alreadyTraced) {
if (isProxyNeeded(beanName)) {
return wrapExecutorService(bean);
}
@@ -89,8 +86,7 @@ class ExecutorBeanPostProcessor implements BeanPostProcessor {
log.info("Not instrumenting bean " + beanName);
}
}
else if (bean instanceof AsyncTaskExecutor
&& !(bean instanceof LazyTraceAsyncTaskExecutor)) {
else if (bean instanceof AsyncTaskExecutor && !alreadyTraced) {
if (isProxyNeeded(beanName)) {
return wrapAsyncTaskExecutor(bean);
}
@@ -98,30 +94,32 @@ class ExecutorBeanPostProcessor implements BeanPostProcessor {
log.info("Not instrumenting bean " + beanName);
}
}
else if (bean instanceof Executor && !(bean instanceof LazyTraceExecutor)) {
else if (bean instanceof Executor && !alreadyTraced) {
return wrapExecutor(bean);
}
return bean;
}
private boolean alreadyTraced(Object bean) {
return bean instanceof LazyTraceThreadPoolTaskExecutor
|| bean instanceof TraceableExecutorService || bean instanceof LazyTraceAsyncTaskExecutor
|| bean instanceof LazyTraceExecutor;
}
private Object wrapExecutor(Object bean) {
Executor executor = (Executor) bean;
boolean methodFinal = anyFinalMethods(executor, Executor.class);
boolean classFinal = Modifier.isFinal(bean.getClass().getModifiers());
boolean cglibProxy = !methodFinal && !classFinal;
try {
return createProxy(bean, cglibProxy,
new ExecutorMethodInterceptor<>(executor, this.beanFactory));
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);
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));
return createProxy(bean, false, new ExecutorMethodInterceptor<>(executor, this.beanFactory));
}
throw ex;
}
@@ -156,8 +154,7 @@ class ExecutorBeanPostProcessor implements BeanPostProcessor {
return !sleuthAsyncProperties.getIgnoredBeans().contains(beanName);
}
Object createThreadPoolTaskExecutorProxy(Object bean, boolean cglibProxy,
ThreadPoolTaskExecutor executor) {
Object createThreadPoolTaskExecutorProxy(Object bean, boolean cglibProxy, ThreadPoolTaskExecutor executor) {
if (!cglibProxy) {
return new LazyTraceThreadPoolTaskExecutor(this.beanFactory, executor);
}
@@ -165,45 +162,37 @@ class ExecutorBeanPostProcessor implements BeanPostProcessor {
() -> new LazyTraceThreadPoolTaskExecutor(this.beanFactory, executor));
}
Supplier<Executor> createThreadPoolTaskSchedulerProxy(
ThreadPoolTaskScheduler executor) {
Supplier<Executor> createThreadPoolTaskSchedulerProxy(ThreadPoolTaskScheduler executor) {
return () -> new LazyTraceThreadPoolTaskScheduler(this.beanFactory, executor);
}
Supplier<Executor> createScheduledThreadPoolExecutorProxy(
ScheduledThreadPoolExecutor executor) {
return () -> new LazyTraceScheduledThreadPoolExecutor(executor.getCorePoolSize(),
executor.getThreadFactory(), executor.getRejectedExecutionHandler(),
this.beanFactory, executor);
Supplier<Executor> createScheduledThreadPoolExecutorProxy(ScheduledThreadPoolExecutor executor) {
return () -> new LazyTraceScheduledThreadPoolExecutor(executor.getCorePoolSize(), executor.getThreadFactory(),
executor.getRejectedExecutionHandler(), this.beanFactory, executor);
}
Object createExecutorServiceProxy(Object bean, boolean cglibProxy,
ExecutorService executor) {
Object createExecutorServiceProxy(Object bean, boolean cglibProxy, ExecutorService executor) {
return getProxiedObject(bean, cglibProxy, executor,
() -> new TraceableExecutorService(this.beanFactory, executor));
}
Object createAsyncTaskExecutorProxy(Object bean, boolean cglibProxy,
AsyncTaskExecutor executor) {
Object createAsyncTaskExecutorProxy(Object bean, boolean cglibProxy, AsyncTaskExecutor executor) {
return getProxiedObject(bean, cglibProxy, executor, () -> {
if (bean instanceof ThreadPoolTaskScheduler) {
return new LazyTraceThreadPoolTaskScheduler(this.beanFactory,
(ThreadPoolTaskScheduler) executor);
return new LazyTraceThreadPoolTaskScheduler(this.beanFactory, (ThreadPoolTaskScheduler) executor);
}
return new LazyTraceAsyncTaskExecutor(this.beanFactory, executor);
});
}
private Object getProxiedObject(Object bean, boolean cglibProxy, Executor executor,
Supplier<Executor> supplier) {
private Object getProxiedObject(Object bean, boolean cglibProxy, Executor executor, Supplier<Executor> supplier) {
ProxyFactoryBean factory = proxyFactoryBean(bean, cglibProxy, executor, supplier);
try {
return getObject(factory);
}
catch (Exception ex) {
if (log.isDebugEnabled()) {
log.debug(
"Exception occurred while trying to get a proxy. Will fallback to a different implementation",
log.debug("Exception occurred while trying to get a proxy. Will fallback to a different implementation",
ex);
}
try {
@@ -212,40 +201,35 @@ class ExecutorBeanPostProcessor implements BeanPostProcessor {
log.debug(
"Will wrap ThreadPoolTaskScheduler in its tracing representation due to previous errors");
}
return createThreadPoolTaskSchedulerProxy(
(ThreadPoolTaskScheduler) bean).get();
return createThreadPoolTaskSchedulerProxy((ThreadPoolTaskScheduler) bean).get();
}
else if (bean instanceof ScheduledThreadPoolExecutor) {
if (log.isDebugEnabled()) {
log.debug(
"Will wrap ScheduledThreadPoolExecutor in its tracing representation due to previous errors");
}
return createScheduledThreadPoolExecutorProxy(
(ScheduledThreadPoolExecutor) bean).get();
return createScheduledThreadPoolExecutorProxy((ScheduledThreadPoolExecutor) bean).get();
}
}
catch (Exception ex2) {
if (log.isDebugEnabled()) {
log.debug(
"Fallback for special wrappers failed, will try the tracing representation instead",
ex2);
log.debug("Fallback for special wrappers failed, will try the tracing representation instead", ex2);
}
}
return supplier.get();
}
}
private ProxyFactoryBean proxyFactoryBean(Object bean, boolean cglibProxy,
Executor executor, Supplier<Executor> supplier) {
private ProxyFactoryBean proxyFactoryBean(Object bean, boolean cglibProxy, Executor executor,
Supplier<Executor> supplier) {
ProxyFactoryBean factory = new ProxyFactoryBean();
factory.setProxyTargetClass(cglibProxy);
factory.addAdvice(
new ExecutorMethodInterceptor<Executor>(executor, this.beanFactory) {
@Override
Executor executor(BeanFactory beanFactory, Executor executor) {
return supplier.get();
}
});
factory.addAdvice(new ExecutorMethodInterceptor<Executor>(executor, this.beanFactory) {
@Override
Executor executor(BeanFactory beanFactory, Executor executor) {
return supplier.get();
}
});
factory.setTarget(bean);
return factory;
}
@@ -265,21 +249,17 @@ class ExecutorBeanPostProcessor implements BeanPostProcessor {
private SleuthAsyncProperties asyncConfigurationProperties() {
if (this.sleuthAsyncProperties == null) {
this.sleuthAsyncProperties = this.beanFactory
.getBean(SleuthAsyncProperties.class);
this.sleuthAsyncProperties = this.beanFactory.getBean(SleuthAsyncProperties.class);
}
return this.sleuthAsyncProperties;
}
private static <T> boolean anyFinalMethods(T object, Class<T> iface) {
AtomicBoolean finalMethodPresent = new AtomicBoolean();
ReflectionUtils.doWithMethods(iface, method -> finalMethodPresent.set(true),
method -> {
Method m = ReflectionUtils.findMethod(object.getClass(),
method.getName(), method.getParameterTypes());
return m != null && !ReflectionUtils.isObjectMethod(m)
&& Modifier.isFinal(m.getModifiers());
});
ReflectionUtils.doWithMethods(iface, method -> finalMethodPresent.set(true), method -> {
Method m = ReflectionUtils.findMethod(object.getClass(), method.getName(), method.getParameterTypes());
return m != null && !ReflectionUtils.isObjectMethod(m) && Modifier.isFinal(m.getModifiers());
});
return finalMethodPresent.get();
}
@@ -321,8 +301,7 @@ class ExecutorMethodInterceptor<T extends Executor> implements MethodInterceptor
private Method getMethod(MethodInvocation invocation, Object object) {
Method method = invocation.getMethod();
return ReflectionUtils.findMethod(object.getClass(), method.getName(),
method.getParameterTypes());
return ReflectionUtils.findMethod(object.getClass(), method.getName(), method.getParameterTypes());
}
T executor(BeanFactory beanFactory, T executor) {

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

@@ -33,6 +33,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
import brave.Tracing;
import org.aopalliance.aop.Advice;
import org.assertj.core.api.BDDAssertions;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
@@ -71,8 +72,7 @@ public class ExecutorBeanPostProcessorTests {
@Before
public void setup() {
this.sleuthAsyncProperties = new SleuthAsyncProperties();
Mockito.when(this.beanFactory.getBean(SleuthAsyncProperties.class))
.thenReturn(this.sleuthAsyncProperties);
Mockito.when(this.beanFactory.getBean(SleuthAsyncProperties.class)).thenReturn(this.sleuthAsyncProperties);
}
@After
@@ -82,28 +82,24 @@ public class ExecutorBeanPostProcessorTests {
@Test
public void should_create_a_cglib_proxy_by_default() throws Exception {
Object o = new ExecutorBeanPostProcessor(this.beanFactory)
.postProcessAfterInitialization(new Foo(), "foo");
Object o = new ExecutorBeanPostProcessor(this.beanFactory).postProcessAfterInitialization(new Foo(), "foo");
then(o).isInstanceOf(Foo.class);
then(AopUtils.isCglibProxy(o)).isTrue();
}
@Test
public void should_fallback_to_sleuth_implementation_when_cglib_cannot_be_created()
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");
Object o = new ExecutorBeanPostProcessor(this.beanFactory).postProcessAfterInitialization(service, "foo");
then(o).isInstanceOf(TraceableExecutorService.class);
service.shutdown();
}
@Test
public void should_fallback_to_default_implementation_when_exception_thrown()
throws Exception {
public void should_fallback_to_default_implementation_when_exception_thrown() throws Exception {
ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor();
ExecutorBeanPostProcessor bpp = new ExecutorBeanPostProcessor(this.beanFactory) {
@@ -121,8 +117,7 @@ public class ExecutorBeanPostProcessorTests {
}
@Test
public void should_create_a_cglib_proxy_by_default_for_ThreadPoolTaskExecutor()
throws Exception {
public void should_create_a_cglib_proxy_by_default_for_ThreadPoolTaskExecutor() throws Exception {
Object o = new ExecutorBeanPostProcessor(this.beanFactory)
.postProcessAfterInitialization(new FooThreadPoolTaskExecutor(), "foo");
@@ -136,8 +131,7 @@ public class ExecutorBeanPostProcessorTests {
ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
ExecutorBeanPostProcessor bpp = new ExecutorBeanPostProcessor(this.beanFactory) {
@Override
Object createThreadPoolTaskExecutorProxy(Object bean, boolean cglibProxy,
ThreadPoolTaskExecutor executor) {
Object createThreadPoolTaskExecutorProxy(Object bean, boolean cglibProxy, ThreadPoolTaskExecutor executor) {
throw new AopConfigException("foo");
}
};
@@ -167,8 +161,7 @@ public class ExecutorBeanPostProcessorTests {
ExecutorService service = exceptionThrowingExecutorService();
ExecutorBeanPostProcessor bpp = new ExecutorBeanPostProcessor(this.beanFactory);
ExecutorService o = (ExecutorService) bpp.postProcessAfterInitialization(service,
"foo");
ExecutorService o = (ExecutorService) bpp.postProcessAfterInitialization(service, "foo");
thenThrownBy(() -> o.submit((Callable<Object>) () -> "hello")).hasMessage("foo")
.isInstanceOf(IllegalStateException.class);
@@ -202,8 +195,7 @@ public class ExecutorBeanPostProcessorTests {
}
@Override
public boolean awaitTermination(long timeout, TimeUnit unit)
throws InterruptedException {
public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
return false;
}
@@ -223,14 +215,13 @@ public class ExecutorBeanPostProcessorTests {
}
@Override
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
throws InterruptedException {
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 {
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
throws InterruptedException {
return null;
}
@@ -241,8 +232,7 @@ public class ExecutorBeanPostProcessorTests {
}
@Override
public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout,
TimeUnit unit)
public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException {
return null;
}
@@ -251,11 +241,9 @@ public class ExecutorBeanPostProcessorTests {
@Test
public void should_use_jdk_proxy_when_executor_has_final_methods() {
ExecutorBeanPostProcessor beanPostProcessor = new ExecutorBeanPostProcessor(
this.beanFactory);
ExecutorBeanPostProcessor beanPostProcessor = new ExecutorBeanPostProcessor(this.beanFactory);
Executor executor = Runnable::run;
Executor wrappedExecutor = (Executor) beanPostProcessor
.postProcessAfterInitialization(executor, "executor");
Executor wrappedExecutor = (Executor) beanPostProcessor.postProcessAfterInitialization(executor, "executor");
then(AopUtils.isJdkDynamicProxy(wrappedExecutor)).isTrue();
then(AopUtils.isCglibProxy(wrappedExecutor)).isFalse();
@@ -268,10 +256,8 @@ public class ExecutorBeanPostProcessorTests {
}
@Test
public void should_use_jdk_proxy_when_executor_service_has_final_methods()
throws Exception {
ExecutorBeanPostProcessor beanPostProcessor = new ExecutorBeanPostProcessor(
this.beanFactory);
public void should_use_jdk_proxy_when_executor_service_has_final_methods() throws Exception {
ExecutorBeanPostProcessor beanPostProcessor = new ExecutorBeanPostProcessor(this.beanFactory);
ExecutorService executorService = new DelegatingSecurityContextExecutorService(
Executors.newSingleThreadExecutor());
ExecutorService wrappedExecutor = (ExecutorService) beanPostProcessor
@@ -284,10 +270,8 @@ public class ExecutorBeanPostProcessorTests {
}
@Test
public void should_use_jdk_proxy_when_async_task_executor_has_final_methods()
throws Exception {
ExecutorBeanPostProcessor beanPostProcessor = new ExecutorBeanPostProcessor(
this.beanFactory);
public void should_use_jdk_proxy_when_async_task_executor_has_final_methods() throws Exception {
ExecutorBeanPostProcessor beanPostProcessor = new ExecutorBeanPostProcessor(this.beanFactory);
AsyncTaskExecutor wrappedExecutor = (AsyncTaskExecutor) beanPostProcessor
.postProcessAfterInitialization(new DirectTaskExecutor(), "taskExecutor");
@@ -299,13 +283,11 @@ public class ExecutorBeanPostProcessorTests {
@Test
public void should_fallback_to_sleuth_impl_when_thread_pool_task_executor_has_final_methods() {
ExecutorBeanPostProcessor postProcessor = new ExecutorBeanPostProcessor(
this.beanFactory);
ExecutorBeanPostProcessor postProcessor = new ExecutorBeanPostProcessor(this.beanFactory);
ThreadPoolTaskExecutor threadPoolTaskExecutor = new PoolTaskExecutor();
ThreadPoolTaskExecutor wrappedTaskExecutor = (ThreadPoolTaskExecutor) postProcessor
.postProcessAfterInitialization(threadPoolTaskExecutor,
"threadPoolTaskExecutor");
.postProcessAfterInitialization(threadPoolTaskExecutor, "threadPoolTaskExecutor");
then(wrappedTaskExecutor).isInstanceOf(LazyTraceThreadPoolTaskExecutor.class);
then(AopUtils.isCglibProxy(wrappedTaskExecutor)).isFalse();
@@ -315,31 +297,26 @@ public class ExecutorBeanPostProcessorTests {
@Test
public void proxy_is_not_needed() throws Exception {
this.sleuthAsyncProperties
.setIgnoredBeans(Collections.singletonList("fooExecutor"));
this.sleuthAsyncProperties.setIgnoredBeans(Collections.singletonList("fooExecutor"));
boolean isProxyNeeded = new ExecutorBeanPostProcessor(this.beanFactory)
.isProxyNeeded("fooExecutor");
boolean isProxyNeeded = new ExecutorBeanPostProcessor(this.beanFactory).isProxyNeeded("fooExecutor");
then(isProxyNeeded).isFalse();
}
@Test
public void proxy_is_needed() throws Exception {
boolean isProxyNeeded = new ExecutorBeanPostProcessor(this.beanFactory)
.isProxyNeeded("fooExecutor");
boolean isProxyNeeded = new ExecutorBeanPostProcessor(this.beanFactory).isProxyNeeded("fooExecutor");
then(isProxyNeeded).isTrue();
}
@Test
public void should_not_create_proxy() throws Exception {
this.sleuthAsyncProperties
.setIgnoredBeans(Collections.singletonList("fooExecutor"));
this.sleuthAsyncProperties.setIgnoredBeans(Collections.singletonList("fooExecutor"));
Object o = new ExecutorBeanPostProcessor(this.beanFactory)
.postProcessAfterInitialization(new ThreadPoolTaskExecutor(),
"fooExecutor");
.postProcessAfterInitialization(new ThreadPoolTaskExecutor(), "fooExecutor");
then(o).isInstanceOf(ThreadPoolTaskExecutor.class);
then(AopUtils.isCglibProxy(o)).isFalse();
@@ -348,8 +325,7 @@ public class ExecutorBeanPostProcessorTests {
@Test
public void should_throw_real_exception_when_using_proxy() throws Exception {
Object o = new ExecutorBeanPostProcessor(this.beanFactory)
.postProcessAfterInitialization(new RejectedExecutionExecutor(),
"fooExecutor");
.postProcessAfterInitialization(new RejectedExecutionExecutor(), "fooExecutor");
then(o).isInstanceOf(RejectedExecutionExecutor.class);
then(AopUtils.isCglibProxy(o)).isTrue();
@@ -357,6 +333,32 @@ public class ExecutorBeanPostProcessorTests {
})).isInstanceOf(RejectedExecutionException.class).hasMessage("rejected");
}
// #1463
@Test
public void should_not_double_instrument_traced_executors() throws Exception {
LazyTraceThreadPoolTaskExecutor lazyTraceThreadPoolTaskExecutor = BDDMockito
.mock(LazyTraceThreadPoolTaskExecutor.class);
Object o = new ExecutorBeanPostProcessor(this.beanFactory)
.postProcessAfterInitialization(lazyTraceThreadPoolTaskExecutor, "executor");
BDDAssertions.then(o).isSameAs(lazyTraceThreadPoolTaskExecutor);
TraceableExecutorService traceableExecutorService = BDDMockito.mock(TraceableExecutorService.class);
o = new ExecutorBeanPostProcessor(this.beanFactory).postProcessAfterInitialization(traceableExecutorService,
"executor");
BDDAssertions.then(o).isSameAs(traceableExecutorService);
LazyTraceAsyncTaskExecutor lazyTraceAsyncTaskExecutor = BDDMockito.mock(LazyTraceAsyncTaskExecutor.class);
o = new ExecutorBeanPostProcessor(this.beanFactory).postProcessAfterInitialization(lazyTraceAsyncTaskExecutor,
"executor");
BDDAssertions.then(o).isSameAs(lazyTraceAsyncTaskExecutor);
LazyTraceExecutor lazyTraceExecutor = BDDMockito.mock(LazyTraceExecutor.class);
o = new ExecutorBeanPostProcessor(this.beanFactory).postProcessAfterInitialization(lazyTraceExecutor,
"executor");
BDDAssertions.then(o).isSameAs(lazyTraceExecutor);
}
class Foo implements Executor {
@Override