Introduce ContextualScopeManager

.gitignore

@@ -38,6 +38,7 @@ Thumbs.db
 *.iml
 *.ipr
 *.iws
+out/
 
 # Visual Studio Code #
 ######################

dd-java-agent-ittests/dd-java-agent-ittests.gradle

@@ -1,4 +1,6 @@
 apply from: "${rootDir}/gradle/java.gradle"
+// TODO: Test scala in separate project
+apply plugin: 'scala'
 
 description = 'dd-java-agent-ittests'
 
@@ -17,8 +19,12 @@ if (JavaVersion.current() != JavaVersion.VERSION_1_8) {
 }
 
 dependencies {
-  testCompile project(':dd-trace-api')
-  testCompile project(':dd-trace-ot')
+  compile project(':dd-trace-api')
+  compile project(':dd-trace-ot')
+
+  // calling scala classes in spock requires an explicit dependency,
+  // hence the compile instead of testCompile
+  compile group: 'org.scala-lang', name: 'scala-library', version: '2.11.12'
 
   testCompile deps.opentracingMock
   testCompile deps.testLogging
@@ -45,6 +51,7 @@ test {
   doFirst {
     // Defining here to allow jacoco to be first on the command line.
     jvmArgs "-javaagent:${project(':dd-java-agent').tasks.shadowJar.archivePath}"
+    jvmArgs "-Ddd.integration.java_concurrent.enabled=true"
   }
 
   testLogging {

dd-java-agent-ittests/src/main/scala/datadog/trace/agent/integration/executors/ScalaConcurrentTests.scala

@@ -0,0 +1,131 @@
+package datadog.trace.agent.integration.executors
+
+import datadog.trace.api.Trace
+import io.opentracing.util.GlobalTracer
+
+import scala.concurrent.duration._
+import scala.concurrent.ExecutionContext.Implicits.global
+import scala.concurrent.{Await, Future, Promise}
+
+class ScalaConcurrentTests {
+
+  /**
+    * @return Number of expected spans in the trace
+    */
+  @Trace
+  def traceWithFutureAndCallbacks() : Integer = {
+    val goodFuture: Future[Integer] = Future {
+      tracedChild("goodFuture")
+      1
+    }
+    goodFuture onSuccess {
+      case _ => tracedChild("successCallback")
+    }
+    val badFuture: Future[Integer] = Future {
+      tracedChild("badFuture")
+      throw new RuntimeException("Uh-oh")
+    }
+    badFuture onFailure {
+      case t: Throwable => tracedChild("failureCallback")
+    }
+
+    return 5
+  }
+
+  @Trace
+  def tracedAcrossThreadsWithNoTrace() :Integer = {
+    val goodFuture: Future[Integer] = Future {
+      1
+    }
+    goodFuture onSuccess {
+      case _ => Future {
+        2
+      } onSuccess {
+        case _ => tracedChild("callback")
+      }
+    }
+
+    return 2
+  }
+
+  /**
+    * @return Number of expected spans in the trace
+    */
+  @Trace
+  def traceWithPromises() : Integer = {
+    val keptPromise = Promise[Boolean]()
+    val brokenPromise = Promise[Boolean]()
+    val afterPromise = keptPromise.future
+    val afterPromise2 = keptPromise.future
+
+    val failedAfterPromise = brokenPromise.future
+
+    Future {
+      tracedChild("future1")
+      keptPromise success true
+      brokenPromise failure new IllegalStateException()
+    }
+
+    afterPromise onSuccess {
+      case b => tracedChild("keptPromise")
+    }
+    afterPromise2 onSuccess {
+      case b => tracedChild("keptPromise2")
+    }
+
+    failedAfterPromise onFailure {
+      case t => tracedChild("brokenPromise")
+    }
+
+    return 5
+  }
+
+  /**
+    * @return Number of expected spans in the trace
+    */
+  @Trace
+  def tracedWithFutureFirstCompletions() :Integer = {
+    val completedVal = Future.firstCompletedOf(
+      List(
+        Future {
+          tracedChild("timeout1")
+          false
+        },
+        Future {
+          tracedChild("timeout2")
+          false
+        },
+        Future {
+          tracedChild("timeout3")
+          true
+        }))
+    Await.result(completedVal, 30 seconds)
+    return 4
+  }
+
+  /**
+    * @return Number of expected spans in the trace
+    */
+  @Trace
+  def tracedTimeout(): Integer = {
+    val f: Future[String] = Future {
+      tracedChild("timeoutChild")
+      while(true) {
+        // never actually finish
+      }
+      "done"
+    }
+
+    try {
+      Await.result(f, 1 milliseconds)
+    } catch {
+      case e: Exception => {}
+    }
+    return 2
+  }
+
+  @Trace
+  def tracedChild(opName: String): Unit = {
+    GlobalTracer.get().activeSpan().setOperationName(opName)
+  }
+}

dd-java-agent-ittests/src/test/groovy/datadog/trace/agent/DDInfoTest.groovy

@@ -0,0 +1,31 @@
+package datadog.trace.agent
+
+import datadog.opentracing.DDTraceOTInfo
+import datadog.trace.api.DDTraceApiInfo
+
+class DDInfoTest {
+  def "info accessible from api"() {
+    expect:
+    DDTraceApiInfo.VERSION == DDTraceOTInfo.VERSION
+
+    DDTraceApiInfo.VERSION != null
+    DDTraceApiInfo.VERSION != ""
+    DDTraceApiInfo.VERSION != "unknown"
+    DDTraceOTInfo.VERSION != null
+    DDTraceOTInfo.VERSION != ""
+    DDTraceOTInfo.VERSION != "unknown"
+  }
+
+  def "info accessible from agent"() {
+    setup:
+    def clazz = Class.forName("datadog.trace.agent.tooling.DDJavaAgentInfo")
+    def versionField = clazz.getDeclaredField("VERSION")
+    def version = versionField.get(null)
+
+    expect:
+    version != null
+    version != ""
+    version != "unknown"
+    version == DDTraceApiInfo.VERSION
+  }
+}

dd-java-agent-ittests/src/test/groovy/datadog/trace/agent/integration/executors/ExecutorInstrumentationTest.groovy

@@ -0,0 +1,197 @@
+package datadog.trace.agent.integration.executors
+
+import datadog.opentracing.DDSpan
+import datadog.opentracing.DDTracer
+import datadog.trace.agent.test.IntegrationTestUtils
+import datadog.trace.api.Trace
+import datadog.trace.common.writer.ListWriter
+import spock.lang.Shared
+import spock.lang.Specification
+import spock.lang.Unroll
+
+import java.lang.reflect.Method
+import java.util.concurrent.ArrayBlockingQueue
+import java.util.concurrent.Callable
+import java.util.concurrent.ExecutorService
+import java.util.concurrent.Executor
+import java.util.concurrent.ForkJoinPool
+import java.util.concurrent.Future
+import java.util.concurrent.RejectedExecutionException
+import java.util.concurrent.ScheduledThreadPoolExecutor
+import java.util.concurrent.ThreadPoolExecutor
+import java.util.concurrent.TimeUnit
+
+class ExecutorInstrumentationTest extends Specification {
+  @Shared
+  ListWriter testWriter = new ListWriter()
+  @Shared
+  DDTracer tracer = new DDTracer(testWriter)
+  @Shared
+  Method submitMethod
+  @Shared
+  Method executeMethod
+
+  def setupSpec() {
+    IntegrationTestUtils.registerOrReplaceGlobalTracer(tracer)
+    testWriter.start()
+
+    executeMethod = Executor.getMethod("execute", Runnable)
+    submitMethod = ExecutorService.getMethod("submit", Callable)
+  }
+
+  def setup() {
+    getTestWriter().close()
+  }
+
+  @Unroll
+  // more useful name breaks java9 javac
+  // def "#poolImpl.getClass().getSimpleName() #method.getName() propagates"()
+  def "#poolImpl #method propagates"() {
+    setup:
+    def pool = poolImpl
+    def m = method
+
+    new Runnable(){
+      @Override
+      @Trace(operationName = "parent")
+      void run() {
+        // this child will have a span
+        m.invoke(pool, new AsyncChild())
+        // this child won't
+        m.invoke(pool, new AsyncChild(false, false))
+      }
+    }.run()
+
+    testWriter.waitForTraces(1)
+    List<DDSpan> trace = testWriter.get(0)
+
+    expect:
+    testWriter.size() == 1
+    trace.size() == 2
+    trace.get(0).operationName == "parent"
+    trace.get(1).operationName == "asyncChild"
+    trace.get(1).parentId == trace.get(0).spanId
+
+    cleanup:
+    pool?.shutdown()
+
+    // Unfortunately, there's no simple way to test the cross product of methods/pools.
+    where:
+    poolImpl                                                                                        | method
+    new ForkJoinPool()                                                                              | submitMethod
+    new ForkJoinPool()                                                                              | executeMethod
+    new ThreadPoolExecutor(1, 1, 1000, TimeUnit.NANOSECONDS, new ArrayBlockingQueue<Runnable>(1))   | submitMethod
+    new ThreadPoolExecutor(1, 1, 1000, TimeUnit.NANOSECONDS, new ArrayBlockingQueue<Runnable>(1))   | executeMethod
+    new ScheduledThreadPoolExecutor(1)                                                              | submitMethod
+    new ScheduledThreadPoolExecutor(1)                                                              | executeMethod
+  }
+
+  @Unroll
+  // more useful name breaks java9 javac
+  // def "#poolImpl.getClass().getSimpleName() #method.getName() propagates"()
+  def "#poolImpl reports after canceled jobs" () {
+    setup:
+    def pool = poolImpl
+    final AsyncChild child = new AsyncChild(true, true)
+    List<Future> jobFutures = new ArrayList<Future>()
+
+    new Runnable(){
+      @Override
+      @Trace(operationName = "parent")
+      void run() {
+        try {
+          for (int i = 0; i < 20; ++ i) {
+            Future f = pool.submit((Callable)child)
+            jobFutures.add(f)
+          }
+        } catch (RejectedExecutionException e) {
+        }
+
+        for (Future f : jobFutures) {
+          f.cancel(false)
+        }
+        child.unblock()
+      }
+    }.run()
+
+    testWriter.waitForTraces(1)
+
+    expect:
+    testWriter.size() == 1
+
+    where:
+    poolImpl                                                                                        | _
+    new ForkJoinPool()                                                                              | _
+    new ThreadPoolExecutor(1, 1, 1000, TimeUnit.NANOSECONDS, new ArrayBlockingQueue<Runnable>(1))   | _
+    new ScheduledThreadPoolExecutor(1)                                                              | _
+  }
+
+  def "scala futures and callbacks"() {
+    setup:
+    ScalaConcurrentTests scalaTest = new ScalaConcurrentTests()
+    int expectedNumberOfSpans = scalaTest.traceWithFutureAndCallbacks()
+    testWriter.waitForTraces(1)
+    List<DDSpan> trace = testWriter.get(0)
+
+    expect:
+    trace.size() == expectedNumberOfSpans
+    trace[0].operationName == "ScalaConcurrentTests.traceWithFutureAndCallbacks"
+    findSpan(trace, "goodFuture").context().getParentId() == trace[0].context().getSpanId()
+    findSpan(trace, "badFuture").context().getParentId() == trace[0].context().getSpanId()
+    findSpan(trace, "successCallback").context().getParentId() == trace[0].context().getSpanId()
+    findSpan(trace, "failureCallback").context().getParentId() == trace[0].context().getSpanId()
+  }
+
+  def "scala propagates across futures with no traces"() {
+    setup:
+    ScalaConcurrentTests scalaTest = new ScalaConcurrentTests()
+    int expectedNumberOfSpans = scalaTest.tracedAcrossThreadsWithNoTrace()
+    testWriter.waitForTraces(1)
+    List<DDSpan> trace = testWriter.get(0)
+
+    expect:
+    trace.size() == expectedNumberOfSpans
+    trace[0].operationName == "ScalaConcurrentTests.tracedAcrossThreadsWithNoTrace"
+    findSpan(trace, "callback").context().getParentId() == trace[0].context().getSpanId()
+  }
+
+  def "scala either promise completion"() {
+    setup:
+    ScalaConcurrentTests scalaTest = new ScalaConcurrentTests()
+    int expectedNumberOfSpans = scalaTest.traceWithPromises()
+    testWriter.waitForTraces(1)
+    List<DDSpan> trace = testWriter.get(0)
+
+    expect:
+    testWriter.size() == 1
+    trace.size() == expectedNumberOfSpans
+    trace[0].operationName == "ScalaConcurrentTests.traceWithPromises"
+    findSpan(trace, "keptPromise").context().getParentId() == trace[0].context().getSpanId()
+    findSpan(trace, "keptPromise2").context().getParentId() == trace[0].context().getSpanId()
+    findSpan(trace, "brokenPromise").context().getParentId() == trace[0].context().getSpanId()
+  }
+
+  def "scala first completed future"() {
+    setup:
+    ScalaConcurrentTests scalaTest = new ScalaConcurrentTests()
+    int expectedNumberOfSpans = scalaTest.tracedWithFutureFirstCompletions()
+    testWriter.waitForTraces(1)
+    List<DDSpan> trace = testWriter.get(0)
+
+    expect:
+    testWriter.size() == 1
+    trace.size() == expectedNumberOfSpans
+    findSpan(trace, "timeout1").context().getParentId() == trace[0].context().getSpanId()
+    findSpan(trace, "timeout2").context().getParentId() == trace[0].context().getSpanId()
+    findSpan(trace, "timeout3").context().getParentId() == trace[0].context().getSpanId()
+  }
+
+  private DDSpan findSpan(List<DDSpan> trace, String opName) {
+    for (DDSpan span : trace) {
+      if (span.getOperationName() == opName) {
+        return span
+      }
+    }
+    return null
+  }
+}