Fix propagation and add onUnsubscribe handler
Ensures that span is closed when Observable is unsubscribed from.
Also added retransform error logger since retransforms might be missed if an exception is thrown, leading to odd behavior.
MODIFY
dd-java-agent/agent-tooling/src/main/java/datadog/trace/agent/tooling/AgentInstaller.java
dd-java-agent/agent-tooling/src/main/java/datadog/trace/agent/tooling/AgentInstaller.java
@@ -12,6 +12,7 @@ import static net.bytebuddy.matcher.ElementMatchers.not;
import datadog.trace.api.Config;
import datadog.trace.bootstrap.WeakMap;
import java.lang.instrument.Instrumentation;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.ServiceLoader;
@@ -60,9 +61,10 @@ public class AgentInstaller {
new AgentBuilder.Default()
.disableClassFormatChanges()
.with(AgentBuilder.RedefinitionStrategy.RETRANSFORMATION)
+ .with(new RedefinitionLoggingListener())
.with(AgentBuilder.DescriptionStrategy.Default.POOL_ONLY)
.with(POOL_STRATEGY)
- .with(new LoggingListener())
+ .with(new TransformLoggingListener())
.with(new ClassLoadListener())
.with(LOCATION_STRATEGY)
// FIXME: we cannot enable it yet due to BB/JVM bug, see
@@ -159,7 +161,33 @@ public class AgentInstaller {
}
@Slf4j
- static class LoggingListener implements AgentBuilder.Listener {
+ static class RedefinitionLoggingListener implements AgentBuilder.RedefinitionStrategy.Listener {
+
+ @Override
+ public void onBatch(final int index, final List<Class<?>> batch, final List<Class<?>> types) {}
+
+ @Override
+ public Iterable<? extends List<Class<?>>> onError(
+ final int index,
+ final List<Class<?>> batch,
+ final Throwable throwable,
+ final List<Class<?>> types) {
+ if (log.isDebugEnabled()) {
+ log.debug(
+ "Exception while retransforming " + batch.size() + " classes: " + batch, throwable);
+ }
+ return Collections.emptyList();
+ }
+
+ @Override
+ public void onComplete(
+ final int amount,
+ final List<Class<?>> types,
+ final Map<List<Class<?>>, Throwable> failures) {}
+ }
+
+ @Slf4j
+ static class TransformLoggingListener implements AgentBuilder.Listener {
@Override
public void onError(

MODIFY
dd-java-agent/instrumentation/hystrix-1.4/src/main/java/datadog/trace/instrumentation/hystrix/HystrixInstrumentation.java
dd-java-agent/instrumentation/hystrix-1.4/src/main/java/datadog/trace/instrumentation/hystrix/HystrixInstrumentation.java
@@ -16,6 +16,7 @@ import io.opentracing.Tracer;
import io.opentracing.util.GlobalTracer;
import java.util.HashMap;
import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
import net.bytebuddy.asm.Advice;
import net.bytebuddy.description.method.MethodDescription;
import net.bytebuddy.description.type.TypeDescription;
@@ -23,6 +24,7 @@ import net.bytebuddy.matcher.ElementMatcher;
import rx.DDTracingUtil;
import rx.Observable;
import rx.Subscriber;
+import rx.Subscription;
@AutoService(Instrumenter.class)
public class HystrixInstrumentation extends Instrumenter.Default {
@@ -46,6 +48,7 @@ public class HystrixInstrumentation extends Instrumenter.Default {
"rx.DDTracingUtil",
"datadog.trace.agent.decorator.BaseDecorator",
packageName + ".HystrixDecorator",
+ packageName + ".HystrixInstrumentation$SpanFinishingSubscription",
packageName + ".HystrixInstrumentation$TracedSubscriber",
packageName + ".HystrixInstrumentation$TracedOnSubscribe",
};
@@ -127,6 +130,9 @@ public class HystrixInstrumentation extends Instrumenter.Default {
DECORATE.onCommand(span, command, methodName);
try (final Scope scope = tracer.scopeManager().activate(span, false)) {
+ if (!((TraceScope) scope).isAsyncPropagating()) {
+ ((TraceScope) scope).setAsyncPropagation(true);
+ }
delegate.call(new TracedSubscriber(span, subscriber));
}
}
@@ -135,26 +141,35 @@ public class HystrixInstrumentation extends Instrumenter.Default {
public static class TracedSubscriber<T> extends Subscriber<T> {
private final ScopeManager scopeManager = GlobalTracer.get().scopeManager();
- private final Span span;
+ private final AtomicReference<Span> spanRef;
private final Subscriber<T> delegate;
public TracedSubscriber(final Span span, final Subscriber<T> delegate) {
- this.span = span;
+ spanRef = new AtomicReference<>(span);
this.delegate = delegate;
+ final SpanFinishingSubscription subscription = new SpanFinishingSubscription(spanRef);
+ delegate.add(subscription);
}
@Override
public void onStart() {
+ final Span span = spanRef.get();
+ if (span != null) {
try (final Scope scope = scopeManager.activate(span, false)) {
if (scope instanceof TraceScope) {
((TraceScope) scope).setAsyncPropagation(true);
}
delegate.onStart();
}
+ } else {
+ delegate.onStart();
+ }
}
@Override
public void onNext(final T value) {
+ final Span span = spanRef.get();
+ if (span != null) {
try (final Scope scope = scopeManager.activate(span, false)) {
if (scope instanceof TraceScope) {
((TraceScope) scope).setAsyncPropagation(true);
@@ -163,10 +178,15 @@ public class HystrixInstrumentation extends Instrumenter.Default {
} catch (final Throwable e) {
onError(e);
}
+ } else {
+ delegate.onNext(value);
+ }
}
@Override
public void onCompleted() {
+ final Span span = spanRef.getAndSet(null);
+ if (span != null) {
boolean errored = false;
try (final Scope scope = scopeManager.activate(span, false)) {
if (scope instanceof TraceScope) {
@@ -174,6 +194,8 @@ public class HystrixInstrumentation extends Instrumenter.Default {
}
delegate.onCompleted();
} catch (final Throwable e) {
+ // Repopulate the spanRef for onError
+ spanRef.compareAndSet(null, span);
onError(e);
errored = true;
} finally {
@@ -183,10 +205,15 @@ public class HystrixInstrumentation extends Instrumenter.Default {
span.finish();
}
}
+ } else {
+ delegate.onCompleted();
+ }
}
@Override
public void onError(final Throwable e) {
+ final Span span = spanRef.getAndSet(null);
+ if (span != null) {
try (final Scope scope = scopeManager.activate(span, false)) {
if (scope instanceof TraceScope) {
((TraceScope) scope).setAsyncPropagation(true);
@@ -201,6 +228,32 @@ public class HystrixInstrumentation extends Instrumenter.Default {
DECORATE.beforeFinish(span);
span.finish();
}
+ } else {
+ delegate.onError(e);
+ }
+ }
+ }
+
+ public static class SpanFinishingSubscription implements Subscription {
+
+ private final AtomicReference<Span> spanRef;
+
+ public SpanFinishingSubscription(final AtomicReference<Span> spanRef) {
+ this.spanRef = spanRef;
+ }
+
+ @Override
+ public void unsubscribe() {
+ final Span span = spanRef.getAndSet(null);
+ if (span != null) {
+ DECORATE.beforeFinish(span);
+ span.finish();
+ }
+ }
+
+ @Override
+ public boolean isUnsubscribed() {
+ return spanRef.get() == null;
}
}
}

MODIFY
dd-java-agent/instrumentation/hystrix-1.4/src/test/groovy/HystrixObservableChainTest.groovy
dd-java-agent/instrumentation/hystrix-1.4/src/test/groovy/HystrixObservableChainTest.groovy
@@ -3,19 +3,29 @@ import datadog.trace.agent.test.AgentTestRunner
import datadog.trace.api.Trace
import io.opentracing.tag.Tags
import rx.Observable
+import rx.schedulers.Schedulers
+import spock.lang.Retry
+import spock.lang.Timeout
import static com.netflix.hystrix.HystrixCommandGroupKey.Factory.asKey
import static datadog.trace.agent.test.utils.TraceUtils.runUnderTrace
+@Retry
+@Timeout(5)
class HystrixObservableChainTest extends AgentTestRunner {
+ static {
+ // Disable so failure testing below doesn't inadvertently change the behavior.
+ System.setProperty("hystrix.command.default.circuitBreaker.enabled", "false")
+
// Uncomment for debugging:
- // static {
// System.setProperty("hystrix.command.default.execution.timeout.enabled", "false")
- // }
+ }
def "test command #action"() {
setup:
- def command = new HystrixObservableCommand<String>(asKey("ExampleGroup")) {
+
+ def result = runUnderTrace("parent") {
+ def val = new HystrixObservableCommand<String>(asKey("ExampleGroup")) {
@Trace
private String tracedMethod() {
return "Hello"
@@ -26,16 +36,17 @@ class HystrixObservableChainTest extends AgentTestRunner {
Observable.defer {
Observable.just(tracedMethod())
}
+ .subscribeOn(Schedulers.immediate())
}
- }
-
- def result = runUnderTrace("parent") {
- command.toObservable().map {
+ }.toObservable()
+ .subscribeOn(Schedulers.io())
+ .map {
it.toUpperCase()
}.flatMap { str ->
new HystrixObservableCommand<String>(asKey("OtherGroup")) {
@Trace
private String tracedMethod() {
+ blockUntilChildSpansFinished(2)
return "$str!"
}
@@ -44,9 +55,15 @@ class HystrixObservableChainTest extends AgentTestRunner {
Observable.defer {
Observable.just(tracedMethod())
}
+ .subscribeOn(Schedulers.computation())
}
}.toObservable()
+ .subscribeOn(Schedulers.trampoline())
}.toBlocking().first()
+ // when this is running in different threads, we don't know when the other span is done
+ // adding sleep to improve ordering consistency
+ blockUntilChildSpansFinished(4)
+ return val
}
expect:
@@ -68,13 +85,13 @@ class HystrixObservableChainTest extends AgentTestRunner {
span(1) {
serviceName "unnamed-java-app"
operationName "hystrix.cmd"
- resourceName "ExampleGroup.HystrixObservableChainTest\$1.execute"
+ resourceName "OtherGroup.HystrixObservableChainTest\$2.execute"
spanType null
- childOf span(0)
+ childOf span(3)
errored false
tags {
- "hystrix.command" "HystrixObservableChainTest\$1"
- "hystrix.group" "ExampleGroup"
+ "hystrix.command" "HystrixObservableChainTest\$2"
+ "hystrix.group" "OtherGroup"
"hystrix.circuit-open" false
"$Tags.COMPONENT.key" "hystrix"
defaultTags()
@@ -82,28 +99,28 @@ class HystrixObservableChainTest extends AgentTestRunner {
}
span(2) {
serviceName "unnamed-java-app"
- operationName "hystrix.cmd"
- resourceName "OtherGroup.HystrixObservableChainTest\$2.execute"
+ operationName "HystrixObservableChainTest\$2.tracedMethod"
+ resourceName "HystrixObservableChainTest\$2.tracedMethod"
spanType null
childOf span(1)
errored false
tags {
- "hystrix.command" "HystrixObservableChainTest\$2"
- "hystrix.group" "OtherGroup"
- "hystrix.circuit-open" false
- "$Tags.COMPONENT.key" "hystrix"
+ "$Tags.COMPONENT.key" "trace"
defaultTags()
}
}
span(3) {
serviceName "unnamed-java-app"
- operationName "HystrixObservableChainTest\$2.tracedMethod"
- resourceName "HystrixObservableChainTest\$2.tracedMethod"
+ operationName "hystrix.cmd"
+ resourceName "ExampleGroup.HystrixObservableChainTest\$1.execute"
spanType null
- childOf span(2)
+ childOf span(0)
errored false
tags {
- "$Tags.COMPONENT.key" "trace"
+ "hystrix.command" "HystrixObservableChainTest\$1"
+ "hystrix.group" "ExampleGroup"
+ "hystrix.circuit-open" false
+ "$Tags.COMPONENT.key" "hystrix"
defaultTags()
}
}
@@ -112,7 +129,7 @@ class HystrixObservableChainTest extends AgentTestRunner {
operationName "HystrixObservableChainTest\$1.tracedMethod"
resourceName "HystrixObservableChainTest\$1.tracedMethod"
spanType null
- childOf span(1)
+ childOf span(3)
errored false
tags {
"$Tags.COMPONENT.key" "trace"

MODIFY
dd-java-agent/instrumentation/hystrix-1.4/src/test/groovy/HystrixObservableTest.groovy
dd-java-agent/instrumentation/hystrix-1.4/src/test/groovy/HystrixObservableTest.groovy
import com.netflix.hystrix.HystrixObservable
import com.netflix.hystrix.HystrixObservableCommand
+import com.netflix.hystrix.exception.HystrixRuntimeException
import datadog.trace.agent.test.AgentTestRunner
import datadog.trace.api.Trace
import io.opentracing.tag.Tags
import rx.Observable
+import rx.schedulers.Schedulers
+import spock.lang.Retry
+import spock.lang.Timeout
import java.util.concurrent.BlockingQueue
import java.util.concurrent.LinkedBlockingQueue
@@ -11,15 +15,23 @@ import java.util.concurrent.LinkedBlockingQueue
import static com.netflix.hystrix.HystrixCommandGroupKey.Factory.asKey
import static datadog.trace.agent.test.utils.TraceUtils.runUnderTrace
+@Retry
+@Timeout(5)
class HystrixObservableTest extends AgentTestRunner {
+ static {
+ // Disable so failure testing below doesn't inadvertently change the behavior.
+ System.setProperty("hystrix.command.default.circuitBreaker.enabled", "false")
+
// Uncomment for debugging:
- // static {
// System.setProperty("hystrix.command.default.execution.timeout.enabled", "false")
- // }
+ }
def "test command #action"() {
setup:
- def command = new HystrixObservableCommand<String>(asKey("ExampleGroup")) {
+ def observeOnFn = observeOn
+ def subscribeOnFn = subscribeOn
+ def result = runUnderTrace("parent") {
+ def val = operation new HystrixObservableCommand<String>(asKey("ExampleGroup")) {
@Trace
private String tracedMethod() {
return "Hello!"
@@ -27,14 +39,22 @@ class HystrixObservableTest extends AgentTestRunner {
@Override
protected Observable<String> construct() {
- Observable.defer {
- Observable.just(tracedMethod())
+ def obs = Observable.defer {
+ Observable.just(tracedMethod()).repeat(1)
}
+ if (observeOnFn) {
+ obs = obs.observeOn(observeOnFn)
}
+ if (subscribeOnFn) {
+ obs = obs.subscribeOn(subscribeOnFn)
}
-
- def result = runUnderTrace("parent") {
- operation(command)
+ return obs
+ }
+ }
+ // when this is running in different threads, we don't know when the other span is done
+ // adding sleep to improve ordering consistency
+ blockUntilChildSpansFinished(2)
+ return val
}
expect:
@@ -85,35 +105,66 @@ class HystrixObservableTest extends AgentTestRunner {
}
where:
- action | operation
- "toObservable" | { HystrixObservable cmd -> cmd.toObservable().toBlocking().first() }
- "observe" | { HystrixObservable cmd -> cmd.observe().toBlocking().first() }
- "observe block" | { HystrixObservable cmd ->
+ action | observeOn | subscribeOn | operation
+ "toObservable" | null | null | { HystrixObservable cmd -> cmd.toObservable().toBlocking().first() }
+ "toObservable-I" | Schedulers.immediate() | null | { HystrixObservable cmd -> cmd.toObservable().toBlocking().first() }
+ "toObservable-T" | Schedulers.trampoline() | null | { HystrixObservable cmd -> cmd.toObservable().toBlocking().first() }
+ "toObservable-C" | Schedulers.computation() | null | { HystrixObservable cmd -> cmd.toObservable().toBlocking().first() }
+ "toObservable-IO" | Schedulers.io() | null | { HystrixObservable cmd -> cmd.toObservable().toBlocking().first() }
+ "toObservable-NT" | Schedulers.newThread() | null | { HystrixObservable cmd -> cmd.toObservable().toBlocking().first() }
+ "toObservable+I" | null | Schedulers.immediate() | { HystrixObservable cmd -> cmd.toObservable().toBlocking().first() }
+ "toObservable+T" | null | Schedulers.trampoline() | { HystrixObservable cmd -> cmd.toObservable().toBlocking().first() }
+ "toObservable+C" | null | Schedulers.computation() | { HystrixObservable cmd -> cmd.toObservable().toBlocking().first() }
+ "toObservable+IO" | null | Schedulers.io() | { HystrixObservable cmd -> cmd.toObservable().toBlocking().first() }
+ "toObservable+NT" | null | Schedulers.newThread() | { HystrixObservable cmd -> cmd.toObservable().toBlocking().first() }
+ "observe" | null | null | { HystrixObservable cmd -> cmd.observe().toBlocking().first() }
+ "observe-I" | Schedulers.immediate() | null | { HystrixObservable cmd -> cmd.observe().toBlocking().first() }
+ "observe-T" | Schedulers.trampoline() | null | { HystrixObservable cmd -> cmd.observe().toBlocking().first() }
+ "observe-C" | Schedulers.computation() | null | { HystrixObservable cmd -> cmd.observe().toBlocking().first() }
+ "observe-IO" | Schedulers.io() | null | { HystrixObservable cmd -> cmd.observe().toBlocking().first() }
+ "observe-NT" | Schedulers.newThread() | null | { HystrixObservable cmd -> cmd.observe().toBlocking().first() }
+ "observe+I" | null | Schedulers.immediate() | { HystrixObservable cmd -> cmd.observe().toBlocking().first() }
+ "observe+T" | null | Schedulers.trampoline() | { HystrixObservable cmd -> cmd.observe().toBlocking().first() }
+ "observe+C" | null | Schedulers.computation() | { HystrixObservable cmd -> cmd.observe().toBlocking().first() }
+ "observe+IO" | null | Schedulers.io() | { HystrixObservable cmd -> cmd.observe().toBlocking().first() }
+ "observe+NT" | null | Schedulers.newThread() | { HystrixObservable cmd -> cmd.observe().toBlocking().first() }
+ "toObservable block" | Schedulers.computation() | Schedulers.newThread() | { HystrixObservable cmd ->
BlockingQueue queue = new LinkedBlockingQueue()
- cmd.observe().subscribe { next ->
+ def subscription = cmd.toObservable().subscribe { next ->
queue.put(next)
}
- queue.take()
+ def val = queue.take()
+ subscription.unsubscribe()
+ return val
}
}
def "test command #action fallback"() {
setup:
- def command = new HystrixObservableCommand<String>(asKey("ExampleGroup")) {
+ def observeOnFn = observeOn
+ def subscribeOnFn = subscribeOn
+ def result = runUnderTrace("parent") {
+ def val = operation new HystrixObservableCommand<String>(asKey("ExampleGroup")) {
@Override
protected Observable<String> construct() {
- Observable.defer {
- Observable.error(new IllegalArgumentException())
+ def err = Observable.defer {
+ Observable.error(new IllegalArgumentException()).repeat(1)
+ }
+ if (observeOnFn) {
+ err = err.observeOn(observeOnFn)
}
+ if (subscribeOnFn) {
+ err = err.subscribeOn(subscribeOnFn)
+ }
+ return err
}
protected Observable<String> resumeWithFallback() {
- return Observable.just("Fallback!")
+ return Observable.just("Fallback!").repeat(1)
}
}
-
- def result = runUnderTrace("parent") {
- operation(command)
+ blockUntilChildSpansFinished(2) // Improve span ordering consistency
+ return val
}
expect:
@@ -168,15 +219,160 @@ class HystrixObservableTest extends AgentTestRunner {
}
where:
- action | operation
- "toObservable" | { HystrixObservable cmd -> cmd.toObservable().toBlocking().first() }
- "observe" | { HystrixObservable cmd -> cmd.observe().toBlocking().first() }
- "observe block" | { HystrixObservable cmd ->
+ action | observeOn | subscribeOn | operation
+ "toObservable" | null | null | { HystrixObservable cmd -> cmd.toObservable().toBlocking().first() }
+ "toObservable-I" | Schedulers.immediate() | null | { HystrixObservable cmd -> cmd.toObservable().toBlocking().first() }
+ "toObservable-T" | Schedulers.trampoline() | null | { HystrixObservable cmd -> cmd.toObservable().toBlocking().first() }
+ "toObservable-C" | Schedulers.computation() | null | { HystrixObservable cmd -> cmd.toObservable().toBlocking().first() }
+ "toObservable-IO" | Schedulers.io() | null | { HystrixObservable cmd -> cmd.toObservable().toBlocking().first() }
+ "toObservable-NT" | Schedulers.newThread() | null | { HystrixObservable cmd -> cmd.toObservable().toBlocking().first() }
+ "toObservable+I" | null | Schedulers.immediate() | { HystrixObservable cmd -> cmd.toObservable().toBlocking().first() }
+ "toObservable+T" | null | Schedulers.trampoline() | { HystrixObservable cmd -> cmd.toObservable().toBlocking().first() }
+ "toObservable+C" | null | Schedulers.computation() | { HystrixObservable cmd -> cmd.toObservable().toBlocking().first() }
+ "toObservable+IO" | null | Schedulers.io() | { HystrixObservable cmd -> cmd.toObservable().toBlocking().first() }
+ "toObservable+NT" | null | Schedulers.newThread() | { HystrixObservable cmd -> cmd.toObservable().toBlocking().first() }
+ "observe" | null | null | { HystrixObservable cmd -> cmd.observe().toBlocking().first() }
+ "observe-I" | Schedulers.immediate() | null | { HystrixObservable cmd -> cmd.observe().toBlocking().first() }
+ "observe-T" | Schedulers.trampoline() | null | { HystrixObservable cmd -> cmd.observe().toBlocking().first() }
+ "observe-C" | Schedulers.computation() | null | { HystrixObservable cmd -> cmd.observe().toBlocking().first() }
+ "observe-IO" | Schedulers.io() | null | { HystrixObservable cmd -> cmd.observe().toBlocking().first() }
+ "observe-NT" | Schedulers.newThread() | null | { HystrixObservable cmd -> cmd.observe().toBlocking().first() }
+ "observe+I" | null | Schedulers.immediate() | { HystrixObservable cmd -> cmd.observe().toBlocking().first() }
+ "observe+T" | null | Schedulers.trampoline() | { HystrixObservable cmd -> cmd.observe().toBlocking().first() }
+ "observe+C" | null | Schedulers.computation() | { HystrixObservable cmd -> cmd.observe().toBlocking().first() }
+ "observe+IO" | null | Schedulers.io() | { HystrixObservable cmd -> cmd.observe().toBlocking().first() }
+ "observe+NT" | null | Schedulers.newThread() | { HystrixObservable cmd -> cmd.observe().toBlocking().first() }
+ "toObservable block" | null | null | { HystrixObservable cmd ->
BlockingQueue queue = new LinkedBlockingQueue()
- cmd.observe().subscribe { next ->
+ def subscription = cmd.toObservable().subscribe { next ->
queue.put(next)
}
- queue.take()
+ def val = queue.take()
+ subscription.unsubscribe()
+ return val
+ }
+ }
+
+ def "test no fallback results in error for #action"() {
+ setup:
+ def observeOnFn = observeOn
+ def subscribeOnFn = subscribeOn
+
+ when:
+ runUnderTrace("parent") {
+ try {
+ operation new HystrixObservableCommand<String>(asKey("FailingGroup")) {
+
+ @Override
+ protected Observable<String> construct() {
+ def err = Observable.defer {
+ Observable.error(new IllegalArgumentException())
+ }
+ if (observeOnFn) {
+ err = err.observeOn(observeOnFn)
+ }
+ if (subscribeOnFn) {
+ err = err.subscribeOn(subscribeOnFn)
+ }
+ return err
+ }
+ }
+ } finally {
+ // when this is running in different threads, we don't know when the other span is done
+ // adding sleep to improve ordering consistency
+ // Also an exception is being thrown here, so we need to wrap it in a try block.
+ blockUntilChildSpansFinished(2)
+ }
+ }
+
+ then:
+ TRANSFORMED_CLASSES.contains("HystrixObservableTest\$3")
+ def err = thrown HystrixRuntimeException
+ err.cause instanceof IllegalArgumentException
+
+ assertTraces(1) {
+ trace(0, 3) {
+ span(0) {
+ serviceName "unnamed-java-app"
+ operationName "parent"
+ resourceName "parent"
+ spanType null
+ parent()
+ errored true
+ tags {
+ errorTags(HystrixRuntimeException, "HystrixObservableTest\$3 failed and no fallback available.")
+ defaultTags()
+ }
+ }
+ span(1) {
+ serviceName "unnamed-java-app"
+ operationName "hystrix.cmd"
+ resourceName "FailingGroup.HystrixObservableTest\$3.execute"
+ spanType null
+ childOf span(0)
+ errored true
+ tags {
+ "hystrix.command" "HystrixObservableTest\$3"
+ "hystrix.group" "FailingGroup"
+ "hystrix.circuit-open" false
+ "$Tags.COMPONENT.key" "hystrix"
+ errorTags(IllegalArgumentException)
+ defaultTags()
+ }
+ }
+ span(2) {
+ serviceName "unnamed-java-app"
+ operationName "hystrix.cmd"
+ resourceName "FailingGroup.HystrixObservableTest\$3.fallback"
+ spanType null
+ childOf span(1)
+ errored true
+ tags {
+ "hystrix.command" "HystrixObservableTest\$3"
+ "hystrix.group" "FailingGroup"
+ "hystrix.circuit-open" false
+ "$Tags.COMPONENT.key" "hystrix"
+ errorTags(UnsupportedOperationException, "No fallback available.")
+ defaultTags()
+ }
+ }
+ }
+ }
+
+ where:
+ action | observeOn | subscribeOn | operation
+ "toObservable" | null | null | { HystrixObservable cmd -> cmd.toObservable().toBlocking().first() }
+ "toObservable-I" | Schedulers.immediate() | null | { HystrixObservable cmd -> cmd.toObservable().toBlocking().first() }
+ "toObservable-T" | Schedulers.trampoline() | null | { HystrixObservable cmd -> cmd.toObservable().toBlocking().first() }
+ "toObservable-C" | Schedulers.computation() | null | { HystrixObservable cmd -> cmd.toObservable().toBlocking().first() }
+ "toObservable-IO" | Schedulers.io() | null | { HystrixObservable cmd -> cmd.toObservable().toBlocking().first() }
+ "toObservable-NT" | Schedulers.newThread() | null | { HystrixObservable cmd -> cmd.toObservable().toBlocking().first() }
+ "toObservable+I" | null | Schedulers.immediate() | { HystrixObservable cmd -> cmd.toObservable().toBlocking().first() }
+ "toObservable+T" | null | Schedulers.trampoline() | { HystrixObservable cmd -> cmd.toObservable().toBlocking().first() }
+ "toObservable+C" | null | Schedulers.computation() | { HystrixObservable cmd -> cmd.toObservable().toBlocking().first() }
+ "toObservable+IO" | null | Schedulers.io() | { HystrixObservable cmd -> cmd.toObservable().toBlocking().first() }
+ "toObservable+NT" | null | Schedulers.newThread() | { HystrixObservable cmd -> cmd.toObservable().toBlocking().first() }
+ "observe" | null | null | { HystrixObservable cmd -> cmd.observe().toBlocking().first() }
+ "observe-I" | Schedulers.immediate() | null | { HystrixObservable cmd -> cmd.observe().toBlocking().first() }
+ "observe-T" | Schedulers.trampoline() | null | { HystrixObservable cmd -> cmd.observe().toBlocking().first() }
+ "observe-C" | Schedulers.computation() | null | { HystrixObservable cmd -> cmd.observe().toBlocking().first() }
+ "observe-IO" | Schedulers.io() | null | { HystrixObservable cmd -> cmd.observe().toBlocking().first() }
+ "observe-NT" | Schedulers.newThread() | null | { HystrixObservable cmd -> cmd.observe().toBlocking().first() }
+ "observe+I" | null | Schedulers.immediate() | { HystrixObservable cmd -> cmd.observe().toBlocking().first() }
+ "observe+T" | null | Schedulers.trampoline() | { HystrixObservable cmd -> cmd.observe().toBlocking().first() }
+ "observe+C" | null | Schedulers.computation() | { HystrixObservable cmd -> cmd.observe().toBlocking().first() }
+ "observe+IO" | null | Schedulers.io() | { HystrixObservable cmd -> cmd.observe().toBlocking().first() }
+ "observe+NT" | null | Schedulers.newThread() | { HystrixObservable cmd -> cmd.observe().toBlocking().first() }
+ "toObservable block" | Schedulers.computation() | Schedulers.newThread() | { HystrixObservable cmd ->
+ def queue = new LinkedBlockingQueue<Throwable>()
+ def subscription = cmd.toObservable().subscribe({ next ->
+ queue.put(new Exception("Unexpectedly got a next"))
+ }, { next ->
+ queue.put(next)
+ })
+ Throwable ex = queue.take()
+ subscription.unsubscribe()
+ throw ex
}
}
}

MODIFY
dd-java-agent/instrumentation/hystrix-1.4/src/test/groovy/HystrixTest.groovy
dd-java-agent/instrumentation/hystrix-1.4/src/test/groovy/HystrixTest.groovy
@@ -2,6 +2,7 @@ import com.netflix.hystrix.HystrixCommand
import datadog.trace.agent.test.AgentTestRunner
import datadog.trace.api.Trace
import io.opentracing.tag.Tags
+import spock.lang.Timeout
import java.util.concurrent.BlockingQueue
import java.util.concurrent.LinkedBlockingQueue
@@ -9,11 +10,15 @@ import java.util.concurrent.LinkedBlockingQueue
import static com.netflix.hystrix.HystrixCommandGroupKey.Factory.asKey
import static datadog.trace.agent.test.utils.TraceUtils.runUnderTrace
+@Timeout(10)
class HystrixTest extends AgentTestRunner {
+ static {
+ // Disable so failure testing below doesn't inadvertently change the behavior.
+ System.setProperty("hystrix.command.default.circuitBreaker.enabled", "false")
+
// Uncomment for debugging:
- // static {
// System.setProperty("hystrix.command.default.execution.timeout.enabled", "false")
- // }
+ }
def "test command #action"() {
setup:

MODIFY
dd-java-agent/instrumentation/java-concurrent/src/main/java/datadog/trace/instrumentation/java/concurrent/AsyncPropagatingDisableInstrumentation.java
dd-java-agent/instrumentation/java-concurrent/src/main/java/datadog/trace/instrumentation/java/concurrent/AsyncPropagatingDisableInstrumentation.java
package datadog.trace.instrumentation.java.concurrent;
+import static datadog.trace.agent.tooling.ByteBuddyElementMatchers.safeHasSuperType;
import static java.util.Collections.singletonMap;
-import static net.bytebuddy.matcher.ElementMatchers.isConstructor;
import static net.bytebuddy.matcher.ElementMatchers.named;
import com.google.auto.service.AutoService;
@@ -31,17 +31,19 @@ import net.bytebuddy.matcher.ElementMatchers;
public final class AsyncPropagatingDisableInstrumentation implements Instrumenter {
private static final Map<
- ElementMatcher<? super TypeDescription>, ElementMatcher<MethodDescription>>
+ ElementMatcher<? super TypeDescription>, ElementMatcher<? super MethodDescription>>
CLASS_AND_METHODS =
new ImmutableMap.Builder<
- ElementMatcher<? super TypeDescription>, ElementMatcher<MethodDescription>>()
- .put(named("rx.internal.util.ObjectPool"), isConstructor())
+ ElementMatcher<? super TypeDescription>,
+ ElementMatcher<? super MethodDescription>>()
+ .put(safeHasSuperType(named("rx.Scheduler$Worker")), named("schedulePeriodically"))
.build();
@Override
public AgentBuilder instrument(AgentBuilder agentBuilder) {
- for (final Map.Entry<ElementMatcher<? super TypeDescription>, ElementMatcher<MethodDescription>>
+ for (final Map.Entry<
+ ElementMatcher<? super TypeDescription>, ElementMatcher<? super MethodDescription>>
entry : CLASS_AND_METHODS.entrySet()) {
agentBuilder =
new DisableAsyncInstrumentation(entry.getKey(), entry.getValue())
@@ -54,7 +56,7 @@ public final class AsyncPropagatingDisableInstrumentation implements Instrumente
public static class DisableAsyncInstrumentation extends Default {
private final ElementMatcher<? super TypeDescription> typeMatcher;
- private final ElementMatcher<MethodDescription> methodMatcher;
+ private final ElementMatcher<? super MethodDescription> methodMatcher;
/** No-arg constructor only used by muzzle and tests. */
public DisableAsyncInstrumentation() {
@@ -63,7 +65,7 @@ public final class AsyncPropagatingDisableInstrumentation implements Instrumente
public DisableAsyncInstrumentation(
final ElementMatcher<? super TypeDescription> typeMatcher,
- final ElementMatcher<MethodDescription> methodMatcher) {
+ final ElementMatcher<? super MethodDescription> methodMatcher) {
super(AbstractExecutorInstrumentation.EXEC_NAME);
this.typeMatcher = typeMatcher;
this.methodMatcher = methodMatcher;
@@ -75,7 +77,7 @@ public final class AsyncPropagatingDisableInstrumentation implements Instrumente
}
@Override
- public Map<ElementMatcher<MethodDescription>, String> transformers() {
+ public Map<? extends ElementMatcher<? super MethodDescription>, String> transformers() {
return singletonMap(methodMatcher, DisableAsyncAdvice.class.getName());
}
}

MODIFY
dd-java-agent/testing/src/main/groovy/datadog/trace/agent/test/AgentTestRunner.java
dd-java-agent/testing/src/main/groovy/datadog/trace/agent/test/AgentTestRunner.java
@@ -5,6 +5,7 @@ import ch.qos.logback.classic.Logger;
import com.google.common.collect.Sets;
import datadog.opentracing.DDSpan;
import datadog.opentracing.DDTracer;
+import datadog.opentracing.PendingTrace;
import datadog.trace.agent.test.asserts.ListWriterAssert;
import datadog.trace.agent.test.utils.GlobalTracerUtils;
import datadog.trace.agent.tooling.AgentInstaller;
@@ -197,6 +198,15 @@ public abstract class AgentTestRunner extends Specification {
ListWriterAssert.assertTraces(TEST_WRITER, size, spec);
}
+ public void blockUntilChildSpansFinished(final int numberOfSpans) throws InterruptedException {
+ final DDSpan span = (DDSpan) io.opentracing.util.GlobalTracer.get().activeSpan();
+ final PendingTrace pendingTrace = span.context().getTrace();
+
+ while (pendingTrace.size() < numberOfSpans) {
+ Thread.sleep(10);
+ }
+ }
+
public static class TestRunnerListener implements AgentBuilder.Listener {
private static final List<AgentTestRunner> activeTests = new CopyOnWriteArrayList<>();

MODIFY
dd-java-agent/testing/src/test/groovy/AgentTestRunnerTest.groovy
dd-java-agent/testing/src/test/groovy/AgentTestRunnerTest.groovy
@@ -97,6 +97,28 @@ class AgentTestRunnerTest extends AgentTestRunner {
new config.exclude.packagename.SomeClass.NestedClass() | _
}
+ def "test unblocked by completed span"() {
+ setup:
+ runUnderTrace("parent") {
+ runUnderTrace("child") {}
+ blockUntilChildSpansFinished(1)
+ }
+
+ expect:
+ assertTraces(1) {
+ trace(0, 2) {
+ span(0) {
+ operationName "parent"
+ parent()
+ }
+ span(1) {
+ operationName "child"
+ childOf(span(0))
+ }
+ }
+ }
+ }
+
private static getAgentTransformer() {
Field f
try {

Fix propagation and add onUnsubscribe handler

Ensures that span is closed when Observable is unsubscribed from.

Also added retransform error logger since retransforms might be missed if an exception is thrown, leading to odd behavior.