Added list of threads to ignore for rx (#294)

fixes #274

* Updates following code review

docs/src/main/asciidoc/spring-cloud-sleuth.adoc

@@ -414,6 +414,9 @@ that wraps all `Action0` instances into their Sleuth representative -
 the `TraceAction`. The hook either starts or continues a span depending on the fact whether tracing was already going
 on before the Action was scheduled. To disable the custom RxJavaSchedulersHook set the `spring.sleuth.rxjava.schedulers.hook.enabled` to `false`.
 
+You can define a list of regular expressions for thread names, for which you don't want a Span to be created. Just provide a comma separated list
+of regular expressions in the `spring.sleuth.rxjava.schedulers.ignoredthreads` property.
+
 === HTTP integration
 
 Features from this section can be disabled by providing the `spring.sleuth.web.enabled` property with value equal to `false`.

spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/rxjava/RxJavaAutoConfiguration.java

@@ -1,5 +1,10 @@
 package org.springframework.cloud.sleuth.instrument.rxjava;
 
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.autoconfigure.AutoConfigureAfter;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
@@ -9,6 +14,8 @@ import org.springframework.cloud.sleuth.Tracer;
 import org.springframework.cloud.sleuth.autoconfig.TraceAutoConfiguration;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.util.StringUtils;
+
 import rx.plugins.RxJavaSchedulersHook;
 
 /**
@@ -25,8 +32,26 @@ import rx.plugins.RxJavaSchedulersHook;
 @ConditionalOnProperty(value = "spring.sleuth.rxjava.schedulers.hook.enabled", matchIfMissing = true)
 public class RxJavaAutoConfiguration {
 
+	/**
+	 * Contains a list of thread names for which spans will not be sampled. Extracted to a constant
+	 * for readability reasons.
+	 */
+	private static final List<String> DEFAULT_IGNORED_THREADS = Arrays.asList("HystrixMetricPoller", "^RxComputation.*$");
+
 	@Bean
-	SleuthRxJavaSchedulersHook sleuthRxJavaSchedulersHook(Tracer tracer, TraceKeys traceKeys) {
-		return new SleuthRxJavaSchedulersHook(tracer, traceKeys);
+	SleuthRxJavaSchedulersHook sleuthRxJavaSchedulersHook(Tracer tracer, TraceKeys traceKeys,
+			// Comma separated list of thread name matchers
+			@Value("${spring.sleuth.rxjava.schedulers.ignoredthreads:}") String threadsToSample) {
+		return new SleuthRxJavaSchedulersHook(tracer, traceKeys, threads(threadsToSample));
+	}
+
+	private List<String> threads(String threadsToSample) {
+		List<String> threads = new ArrayList<>();
+		if (StringUtils.isEmpty(threadsToSample)) {
+			threads.addAll(DEFAULT_IGNORED_THREADS);
+		} else {
+			threads.addAll(Arrays.asList(threadsToSample.split(",")));
+		}
+		return threads;
 	}
 }

spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/rxjava/SleuthRxJavaSchedulersHook.java

@@ -1,10 +1,13 @@
 package org.springframework.cloud.sleuth.instrument.rxjava;
 
+import java.util.List;
+
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.springframework.cloud.sleuth.Span;
 import org.springframework.cloud.sleuth.TraceKeys;
 import org.springframework.cloud.sleuth.Tracer;
+
 import rx.functions.Action0;
 import rx.plugins.RxJavaErrorHandler;
 import rx.plugins.RxJavaObservableExecutionHook;
@@ -25,11 +28,14 @@ class SleuthRxJavaSchedulersHook extends RxJavaSchedulersHook {
 	private static final String RXJAVA_COMPONENT = "rxjava";
 	private final Tracer tracer;
 	private final TraceKeys traceKeys;
+	private final List<String> threadsToSample;
 	private RxJavaSchedulersHook delegate;
 
-	SleuthRxJavaSchedulersHook(Tracer tracer, TraceKeys traceKeys) {
+	SleuthRxJavaSchedulersHook(Tracer tracer, TraceKeys traceKeys,
+			List<String> threadsToSample) {
 		this.tracer = tracer;
 		this.traceKeys = traceKeys;
+		this.threadsToSample = threadsToSample;
 		try {
 			this.delegate = RxJavaPlugins.getInstance().getSchedulersHook();
 			if (this.delegate instanceof SleuthRxJavaSchedulersHook) {
@@ -68,7 +74,8 @@ class SleuthRxJavaSchedulersHook extends RxJavaSchedulersHook {
 		if (wrappedAction instanceof TraceAction) {
 			return action;
 		}
-		return super.onSchedule(new TraceAction(this.tracer, this.traceKeys, wrappedAction));
+		return super.onSchedule(new TraceAction(this.tracer, this.traceKeys, wrappedAction,
+				this.threadsToSample));
 	}
 
 	static class TraceAction implements Action0 {
@@ -77,16 +84,31 @@ class SleuthRxJavaSchedulersHook extends RxJavaSchedulersHook {
 		private Tracer tracer;
 		private TraceKeys traceKeys;
 		private Span parent;
+		private final List<String> threadsToIgnore;
 
-		public TraceAction(Tracer tracer, TraceKeys traceKeys, Action0 actual) {
+		public TraceAction(Tracer tracer, TraceKeys traceKeys, Action0 actual,
+				List<String> threadsToIgnore) {
 			this.tracer = tracer;
 			this.traceKeys = traceKeys;
+			this.threadsToIgnore = threadsToIgnore;
 			this.parent = tracer.getCurrentSpan();
 			this.actual = actual;
 		}
 
+		@SuppressWarnings("Duplicates")
 		@Override
 		public void call() {
+			// don't create a span if the thread name is on a list of threads to ignore
+			for (String threadToIgnore : this.threadsToIgnore) {
+				String threadName = Thread.currentThread().getName();
+				if (threadName.matches(threadToIgnore)) {
+					log.debug(String.format(
+							"Thread with name [%s] matches the regex [%s]. A span will not be created for this Thread.",
+							threadName, threadToIgnore));
+					this.actual.call();
+					return;
+				}
+			}
 			Span span = this.parent;
 			boolean created = false;
 			if (span != null) {

spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/assertions/SpanAssert.java

@@ -132,4 +132,14 @@ public class SpanAssert extends AbstractAssert<SpanAssert, Span> {
 		}
 		return this;
 	}
+
+	public SpanAssert isExportable() {
+		isNotNull();
+		if (!this.actual.isExportable()) {
+			String message = "The span is supposed to be exportable but it's not!";
+			log.error(message);
+			failWithMessage(message);
+		}
+		return this;
+	}
 }

spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/rxjava/SleuthRxJavaSchedulersHookTests.java

@@ -1,11 +1,20 @@
 package org.springframework.cloud.sleuth.instrument.rxjava;
 
-import static org.assertj.core.api.BDDAssertions.then;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ThreadFactory;
 
 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.runners.MockitoJUnitRunner;
 import org.springframework.cloud.sleuth.TraceKeys;
@@ -17,6 +26,11 @@ import rx.plugins.RxJavaObservableExecutionHook;
 import rx.plugins.RxJavaPlugins;
 import rx.plugins.RxJavaSchedulersHook;
 
+import static org.assertj.core.api.BDDAssertions.then;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyString;
+import static org.mockito.Mockito.never;
+
 /**
  *
  * @author Shivang Shah
@@ -24,8 +38,8 @@ import rx.plugins.RxJavaSchedulersHook;
 @RunWith(MockitoJUnitRunner.class)
 public class SleuthRxJavaSchedulersHookTests {
 
-	@Mock
-	Tracer tracer;
+	List<String> threadsToIgnore = new ArrayList<>();
+	@Mock Tracer tracer;
 	TraceKeys traceKeys = new TraceKeys();
 
 	private static StringBuilder caller;
@@ -41,7 +55,7 @@ public class SleuthRxJavaSchedulersHookTests {
 	public void should_not_override_existing_custom_hooks() {
 		RxJavaPlugins.getInstance().registerErrorHandler(new MyRxJavaErrorHandler());
 		RxJavaPlugins.getInstance().registerObservableExecutionHook(new MyRxJavaObservableExecutionHook());
-		new SleuthRxJavaSchedulersHook(this.tracer, this.traceKeys);
+		new SleuthRxJavaSchedulersHook(this.tracer, this.traceKeys, threadsToIgnore);
 		then(RxJavaPlugins.getInstance().getErrorHandler()).isExactlyInstanceOf(MyRxJavaErrorHandler.class);
 		then(RxJavaPlugins.getInstance().getObservableExecutionHook()).isExactlyInstanceOf(MyRxJavaObservableExecutionHook.class);
 	}
@@ -50,7 +64,7 @@ public class SleuthRxJavaSchedulersHookTests {
 	public void should_wrap_delegates_action_in_wrapped_action_when_delegate_is_present_on_schedule() {
 		RxJavaPlugins.getInstance().registerSchedulersHook(new MyRxJavaSchedulersHook());
 		SleuthRxJavaSchedulersHook schedulersHook = new SleuthRxJavaSchedulersHook(
-			this.tracer, this.traceKeys);
+			this.tracer, this.traceKeys, threadsToIgnore);
 		Action0 action = schedulersHook.onSchedule(() -> {
 			caller = new StringBuilder("hello");
 		});
@@ -59,6 +73,37 @@ public class SleuthRxJavaSchedulersHookTests {
 		then(caller.toString()).isEqualTo("called_from_schedulers_hook");
 	}
 
+	@Test
+	public void should_not_create_a_span_when_current_thread_should_be_ignored()
+			throws ExecutionException, InterruptedException {
+		String threadNameToIgnore = "^MyCustomThread.*$";
+		RxJavaPlugins.getInstance().registerSchedulersHook(new MyRxJavaSchedulersHook());
+		SleuthRxJavaSchedulersHook schedulersHook = new SleuthRxJavaSchedulersHook(
+			this.tracer, this.traceKeys, Collections.singletonList(threadNameToIgnore));
+		Future<Void> hello = executorService().submit((Callable<Void>) () -> {
+			Action0 action = schedulersHook.onSchedule(() -> {
+				caller = new StringBuilder("hello");
+			});
+			action.call();
+			return null;
+		});
+
+		hello.get();
+
+		BDDMockito.then(this.tracer).should(never()).createSpan(anyString());
+		BDDMockito.then(this.tracer).should(never()).continueSpan(any());
+	}
+
+	private ExecutorService executorService() {
+		ThreadFactory threadFactory = r -> {
+			Thread thread = new Thread(r);
+			thread.setName("MyCustomThread10");
+			return thread;
+		};
+		return Executors
+				.newSingleThreadExecutor(threadFactory);
+	}
+
 	static class MyRxJavaObservableExecutionHook extends RxJavaObservableExecutionHook {
 	}
 

spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/rxjava/SleuthRxJavaTests.java

@@ -25,11 +25,11 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
 import rx.Observable;
 import rx.functions.Action0;
+import rx.plugins.RxJavaPlugins;
 import rx.schedulers.Schedulers;
 
 import static com.jayway.awaitility.Awaitility.await;
 import static org.springframework.cloud.sleuth.assertions.SleuthAssertions.then;
-import rx.plugins.RxJavaPlugins;
 
 @RunWith(SpringJUnit4ClassRunner.class)
 @SpringApplicationConfiguration(classes = {SleuthRxJavaTests.TestConfig.class})
@@ -67,6 +67,7 @@ public class SleuthRxJavaTests {
 		then(this.tracer.getCurrentSpan()).isNull();
 		await().until(() -> then(this.listener.getEvents()).hasSize(1));
 		then(this.listener.getEvents().get(0)).hasNameEqualTo("rxjava");
+		then(this.listener.getEvents().get(0)).isExportable();
 		then(this.listener.getEvents().get(0)).hasATag(Span.SPAN_LOCAL_COMPONENT_TAG_NAME, "rxjava");
 		then(this.listener.getEvents().get(0)).isALocalComponentSpan();
 	}
@@ -86,6 +87,7 @@ public class SleuthRxJavaTests {
 		//making sure here that no new spans were created or reported as closed
 		then(this.listener.getEvents()).isEmpty();
 		then(spanInCurrentThread).hasNameEqualTo(spanInCurrentThread.getName());
+		then(spanInCurrentThread).isExportable();
 		then(spanInCurrentThread).hasATag(Span.SPAN_LOCAL_COMPONENT_TAG_NAME, "current_span");
 		then(spanInCurrentThread).isALocalComponentSpan();
 	}

spring-cloud-sleuth-core/src/test/resources/application.yml

@@ -9,4 +9,6 @@ exceptionService.ribbon:
   ConnectTimeout: 1
   ReadTimeout: 1
 
-spring.sleuth.scheduled.skipPattern: "^org.*TestBeanWithScheduledMethodToBeIgnored$"
+spring.sleuth.scheduled.skipPattern: "^org.*TestBeanWithScheduledMethodToBeIgnored$"
+# comma separated list of matchers
+spring.sleuth.rxjava.schedulers.ignoredthreads: HystixMetricPoller,^MyCustomThread.*$