Instrument HystrixCommand and HystrixThreadPool

dd-java-agent/agent-tooling/src/main/java/datadog/trace/agent/tooling/AgentInstaller.java

@@ -48,7 +48,7 @@ public static ResettableClassFileTransformer installBytebuddyAgent(
             .or(nameStartsWith("datadog.opentracing."))
             .or(nameStartsWith("datadog.slf4j."))
             .or(nameStartsWith("java.").and(not(nameStartsWith("java.util.concurrent."))))
-            .or(nameStartsWith("com.sun.").and(not(nameStartsWith("com.sun.proxy."))))
+            .or(nameStartsWith("com.sun."))
             .or(nameStartsWith("sun."))
             .or(nameStartsWith("jdk."))
             .or(nameStartsWith("org.aspectj."))

dd-java-agent/instrumentation/hystrix-1.4/hystrix-1.4.gradle

@@ -0,0 +1,28 @@
+apply plugin: 'version-scan'
+
+versionScan {
+  group = "com.netflix.hystrix"
+  module = 'hystrix-core'
+  versions = "[1.4.0,)"
+  verifyPresent = [
+    "com.netflix.hystrix.AbstractCommand": null,
+  ]
+}
+
+apply from: "${rootDir}/gradle/java.gradle"
+
+dependencies {
+//  compileOnly group: 'com.netflix.hystrix', name: 'hystrix-core', version: '1.5.12'
+
+  compile project(':dd-trace-ot')
+  compile project(':dd-java-agent:agent-tooling')
+
+  compile deps.bytebuddy
+  compile deps.opentracing
+  compile deps.autoservice
+
+  testCompile project(':dd-java-agent:testing')
+  testCompile project(':dd-java-agent:instrumentation:java-concurrent')
+  testCompile project(':dd-java-agent:instrumentation:trace-annotation')
+  testCompile group: 'com.netflix.hystrix', name: 'hystrix-core', version: '1.4.0'
+}

dd-java-agent/instrumentation/hystrix-1.4/src/main/java/datadog/trace/instrumentation/hystrix/HystrixCommandInstrumentation.java

@@ -0,0 +1,80 @@
+package datadog.trace.instrumentation.hystrix;
+
+import static io.opentracing.log.Fields.ERROR_OBJECT;
+import static net.bytebuddy.matcher.ElementMatchers.hasSuperType;
+import static net.bytebuddy.matcher.ElementMatchers.isInterface;
+import static net.bytebuddy.matcher.ElementMatchers.isMethod;
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static net.bytebuddy.matcher.ElementMatchers.not;
+
+import com.google.auto.service.AutoService;
+import datadog.trace.agent.tooling.DDAdvice;
+import datadog.trace.agent.tooling.DDTransformers;
+import datadog.trace.agent.tooling.Instrumenter;
+import io.opentracing.Scope;
+import io.opentracing.Span;
+import io.opentracing.tag.Tags;
+import io.opentracing.util.GlobalTracer;
+import java.lang.reflect.Method;
+import java.util.Collections;
+import net.bytebuddy.agent.builder.AgentBuilder;
+import net.bytebuddy.asm.Advice;
+
+@AutoService(Instrumenter.class)
+public class HystrixCommandInstrumentation extends Instrumenter.Configurable {
+
+  public HystrixCommandInstrumentation() {
+    super("hystrix");
+  }
+
+  @Override
+  protected boolean defaultEnabled() {
+    return false;
+  }
+
+  @Override
+  public AgentBuilder apply(final AgentBuilder agentBuilder) {
+    return agentBuilder
+        .type(not(isInterface()).and(hasSuperType(named("com.netflix.hystrix.HystrixCommand"))))
+        // Not adding a version restriction because this should work with any version and add some benefit.
+        .transform(DDTransformers.defaultTransformers())
+        .transform(
+            DDAdvice.create()
+                .advice(
+                    isMethod().and(named("run").or(named("getFallback"))),
+                    TraceAdvice.class.getName()))
+        .asDecorator();
+  }
+
+  public static class TraceAdvice {
+
+    @Advice.OnMethodEnter(suppress = Throwable.class)
+    public static Scope startSpan(@Advice.Origin final Method method) {
+      final Class<?> declaringClass = method.getDeclaringClass();
+      String className = declaringClass.getSimpleName();
+      if (className.isEmpty()) {
+        className = declaringClass.getName();
+        if (declaringClass.getPackage() != null) {
+          final String pkgName = declaringClass.getPackage().getName();
+          if (!pkgName.isEmpty()) {
+            className = declaringClass.getName().replace(pkgName, "").substring(1);
+          }
+        }
+      }
+      final String operationName = className + "." + method.getName();
+
+      return GlobalTracer.get().buildSpan(operationName).startActive(true);
+    }
+
+    @Advice.OnMethodExit(onThrowable = Throwable.class, suppress = Throwable.class)
+    public static void stopSpan(
+        @Advice.Enter final Scope scope, @Advice.Thrown final Throwable throwable) {
+      if (throwable != null) {
+        final Span span = scope.span();
+        Tags.ERROR.set(span, true);
+        span.log(Collections.singletonMap(ERROR_OBJECT, throwable));
+      }
+      scope.close();
+    }
+  }
+}

dd-java-agent/instrumentation/hystrix-1.4/src/main/java/datadog/trace/instrumentation/hystrix/HystrixThreadPoolInstrumentation.java

@@ -0,0 +1,70 @@
+package datadog.trace.instrumentation.hystrix;
+
+import static datadog.trace.agent.tooling.ClassLoaderMatcher.classLoaderHasClasses;
+import static net.bytebuddy.matcher.ElementMatchers.isMethod;
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
+
+import com.google.auto.service.AutoService;
+import datadog.trace.agent.tooling.DDAdvice;
+import datadog.trace.agent.tooling.DDTransformers;
+import datadog.trace.agent.tooling.Instrumenter;
+import datadog.trace.context.TraceScope;
+import io.opentracing.Scope;
+import io.opentracing.util.GlobalTracer;
+import net.bytebuddy.agent.builder.AgentBuilder;
+import net.bytebuddy.asm.Advice;
+
+@AutoService(Instrumenter.class)
+public class HystrixThreadPoolInstrumentation extends Instrumenter.Configurable {
+
+  public HystrixThreadPoolInstrumentation() {
+    super("hystrix");
+  }
+
+  @Override
+  protected boolean defaultEnabled() {
+    return false;
+  }
+
+  @Override
+  public AgentBuilder apply(final AgentBuilder agentBuilder) {
+    return agentBuilder
+        .type(
+            named(
+                "com.netflix.hystrix.strategy.concurrency.HystrixContextScheduler$ThreadPoolWorker"),
+            classLoaderHasClasses("com.netflix.hystrix.AbstractCommand"))
+        .transform(DDTransformers.defaultTransformers())
+        .transform(
+            DDAdvice.create()
+                .advice(
+                    isMethod().and(named("schedule")).and(takesArguments(1)),
+                    EnableAsyncAdvice.class.getName()))
+        .asDecorator();
+  }
+
+  public static class EnableAsyncAdvice {
+
+    @Advice.OnMethodEnter(suppress = Throwable.class)
+    public static boolean enableAsyncTracking() {
+      final Scope scope = GlobalTracer.get().scopeManager().active();
+      if (scope instanceof TraceScope) {
+        if (!((TraceScope) scope).isAsyncPropagating()) {
+          ((TraceScope) scope).setAsyncPropagation(true);
+          return true;
+        }
+      }
+      return false;
+    }
+
+    @Advice.OnMethodExit(suppress = Throwable.class)
+    public static void disableAsyncTracking(@Advice.Enter final boolean wasEnabled) {
+      if (wasEnabled) {
+        final Scope scope = GlobalTracer.get().scopeManager().active();
+        if (scope instanceof TraceScope) {
+          ((TraceScope) scope).setAsyncPropagation(false);
+        }
+      }
+    }
+  }
+}

dd-java-agent/instrumentation/hystrix-1.4/src/test/groovy/HystrixTest.groovy

@@ -0,0 +1,163 @@
+import com.netflix.hystrix.HystrixCommand
+import datadog.trace.agent.test.AgentTestRunner
+import datadog.trace.api.Trace
+import spock.lang.Unroll
+
+import java.util.concurrent.BlockingQueue
+import java.util.concurrent.LinkedBlockingQueue
+
+import static com.netflix.hystrix.HystrixCommandGroupKey.Factory.asKey
+import static datadog.trace.agent.test.ListWriterAssert.assertTraces
+import static datadog.trace.agent.test.TestUtils.runUnderTrace
+
+class HystrixTest extends AgentTestRunner {
+  static {
+    System.setProperty("dd.integration.hystrix.enabled", "true")
+    // Uncomment for debugging:
+//    System.setProperty("hystrix.command.default.execution.timeout.enabled", "false")
+  }
+
+  @Unroll
+  def "test command #action"() {
+    setup:
+    def command = new HystrixCommand(asKey("ExampleGroup")) {
+      @Override
+      protected Object run() throws Exception {
+        return tracedMethod()
+      }
+
+      @Trace
+      private String tracedMethod() {
+        return "Hello!"
+      }
+    }
+    def result = runUnderTrace("parent") {
+      operation(command)
+    }
+    expect:
+    result == "Hello!"
+
+    assertTraces(TEST_WRITER, 1) {
+      trace(0, 3) {
+        span(0) {
+          serviceName "unnamed-java-app"
+          operationName "parent"
+          resourceName "parent"
+          spanType null
+          parent()
+          errored false
+          tags {
+            defaultTags()
+          }
+        }
+        span(1) {
+          serviceName "unnamed-java-app"
+          operationName "HystrixTest\$1.run"
+          resourceName "HystrixTest\$1.run"
+          spanType null
+          childOf span(0)
+          errored false
+          tags {
+            defaultTags()
+          }
+        }
+        span(2) {
+          serviceName "unnamed-java-app"
+          operationName "HystrixTest\$1.tracedMethod"
+          resourceName "HystrixTest\$1.tracedMethod"
+          spanType null
+          childOf span(1)
+          errored false
+          tags {
+            defaultTags()
+          }
+        }
+      }
+    }
+
+    where:
+    action    | operation
+    "execute" | { HystrixCommand cmd -> cmd.execute() }
+    "queue"   | { HystrixCommand cmd -> cmd.queue().get() }
+    "observe" | { HystrixCommand cmd -> cmd.observe().toBlocking().first() }
+    "observe" | { HystrixCommand cmd ->
+      BlockingQueue queue = new LinkedBlockingQueue()
+      cmd.observe().subscribe { next ->
+        queue.put(next)
+      }
+      queue.poll()
+    }
+  }
+
+  @Unroll
+  def "test command #action fallback"() {
+    setup:
+    def command = new HystrixCommand(asKey("ExampleGroup")) {
+      @Override
+      protected Object run() throws Exception {
+        throw new IllegalArgumentException()
+      }
+
+      protected String getFallback() {
+        return "Fallback!"
+      }
+    }
+    def result = runUnderTrace("parent") {
+      operation(command)
+    }
+    expect:
+    result == "Fallback!"
+
+    assertTraces(TEST_WRITER, 1) {
+      trace(0, 3) {
+        span(0) {
+          serviceName "unnamed-java-app"
+          operationName "parent"
+          resourceName "parent"
+          spanType null
+          parent()
+          errored false
+          tags {
+            defaultTags()
+          }
+        }
+        span(1) {
+          serviceName "unnamed-java-app"
+          operationName "HystrixTest\$2.getFallback"
+          resourceName "HystrixTest\$2.getFallback"
+          spanType null
+          childOf span(0)
+          errored false
+          tags {
+            defaultTags()
+          }
+        }
+        span(2) {
+          serviceName "unnamed-java-app"
+          operationName "HystrixTest\$2.run"
+          resourceName "HystrixTest\$2.run"
+          spanType null
+          childOf span(0)
+          errored true
+          tags {
+            errorTags(IllegalArgumentException)
+            defaultTags()
+          }
+        }
+      }
+    }
+
+    where:
+    action    | operation
+    "execute" | { HystrixCommand cmd -> cmd.execute() }
+    "queue"   | { HystrixCommand cmd -> cmd.queue().get() }
+    "observe" | { HystrixCommand cmd -> cmd.observe().toBlocking().first() }
+    "observe" | { HystrixCommand cmd ->
+      BlockingQueue queue = new LinkedBlockingQueue()
+      cmd.observe().subscribe { next ->
+        queue.put(next)
+      }
+      queue.poll()
+    }
+  }
+}

dd-java-agent/instrumentation/jms-1/src/test/groovy/JMS1Test.groovy

@@ -8,6 +8,8 @@ import spock.lang.Shared
 import spock.lang.Unroll
 
 import javax.jms.Connection
+import javax.jms.Message
+import javax.jms.MessageListener
 import javax.jms.Session
 import javax.jms.TextMessage
 import java.util.concurrent.CountDownLatch
@@ -152,9 +154,12 @@ class JMS1Test extends AgentTestRunner {
     def messageRef = new AtomicReference<TextMessage>()
     def producer = session.createProducer(destination)
     def consumer = session.createConsumer(destination)
-    consumer.setMessageListener { message ->
-      lock.await() // ensure the producer trace is reported first.
-      messageRef.set(message)
+    consumer.setMessageListener new MessageListener() {
+      @Override
+      void onMessage(Message message) {
+        lock.await() // ensure the producer trace is reported first.
+        messageRef.set(message)
+      }
     }
 
     def message = session.createTextMessage("a message")