Executor Instrumentation for Scala
MODIFY
.gitignore
.gitignore
@@ -38,6 +38,7 @@ Thumbs.db
*.iml
*.ipr
*.iws
+out/
# Visual Studio Code #
######################

MODIFY
dd-java-agent-ittests/dd-java-agent-ittests.gradle
dd-java-agent-ittests/dd-java-agent-ittests.gradle
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

ADD
dd-java-agent-ittests/src/main/scala/datadog/trace/agent/integration/executors/ScalaConcurrentTests.scala
+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
+ }
+
+ /**
+ * @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)
+ }
+}

ADD
dd-java-agent-ittests/src/test/groovy/datadog/trace/agent/integration/executors/ExecutorInstrumentationTest.groovy
+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"
+
+ 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
+ }
+
+ 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
+ }
+
+ 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
+ }
+}

ADD
dd-java-agent-ittests/src/test/java/datadog/trace/agent/integration/executors/AsyncChild.java
+package datadog.trace.agent.integration.executors;
+
+import datadog.trace.api.Trace;
+import java.util.concurrent.Callable;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class AsyncChild implements Runnable, Callable {
+ private final AtomicBoolean blockThread;
+ private final boolean doTraceableWork;
+ private final AtomicInteger numberOfWorkers = new AtomicInteger(0);
+
+ public AsyncChild() {
+ this(true, false);
+ }
+
+ public AsyncChild(boolean doTraceableWork, boolean blockThread) {
+ this.doTraceableWork = doTraceableWork;
+ this.blockThread = new AtomicBoolean(blockThread);
+ }
+
+ public void unblock() {
+ blockThread.set(false);
+ }
+
+ public int numberOfWorkers() {
+ return numberOfWorkers.get();
+ }
+
+ @Override
+ public void run() {
+ runImpl();
+ }
+
+ @Override
+ public Object call() throws Exception {
+ runImpl();
+ return null;
+ }
+
+ private void runImpl() {
+ if (doTraceableWork) {
+ asyncChild();
+ }
+ numberOfWorkers.getAndIncrement();
+ try {
+ while (blockThread.get()) {
+ // busy-wait to block thread
+ }
+ } finally {
+ numberOfWorkers.getAndDecrement();
+ }
+ }
+
+ @Trace(operationName = "asyncChild")
+ private void asyncChild() {}
+}

ADD
dd-java-agent/instrumentation/java-concurrent/java-concurrent.gradle
+apply from: "${rootDir}/gradle/java.gradle"
+
+dependencies {
+ compile project(':dd-trace-api')
+ compile project(':dd-java-agent:agent-tooling')
+
+ compile deps.bytebuddy
+ compile deps.opentracing
+ compile deps.autoservice
+}

ADD
dd-java-agent/instrumentation/java-concurrent/src/main/java/datadog/trace/instrumentation/java/concurrent/ExecutorInstrumentation.java
+package datadog.trace.instrumentation.java.concurrent;
+
+import static net.bytebuddy.matcher.ElementMatchers.hasSuperType;
+import static net.bytebuddy.matcher.ElementMatchers.isInterface;
+import static net.bytebuddy.matcher.ElementMatchers.nameMatches;
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static net.bytebuddy.matcher.ElementMatchers.not;
+import static net.bytebuddy.matcher.ElementMatchers.takesArgument;
+
+import com.google.auto.service.AutoService;
+import datadog.trace.agent.tooling.DDAdvice;
+import datadog.trace.agent.tooling.HelperInjector;
+import datadog.trace.agent.tooling.Instrumenter;
+import datadog.trace.context.ContextPropagator;
+import io.opentracing.Scope;
+import io.opentracing.util.GlobalTracer;
+import java.lang.reflect.Field;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import lombok.extern.slf4j.Slf4j;
+import net.bytebuddy.agent.builder.AgentBuilder;
+import net.bytebuddy.asm.Advice;
+
+@AutoService(Instrumenter.class)
+public final class ExecutorInstrumentation extends Instrumenter.Configurable {
+ public static final String EXEC_NAME = "java_concurrent";
+ public static final HelperInjector EXEC_HELPER_INJECTOR =
+ new HelperInjector(
+ ExecutorInstrumentation.class.getName() + "$ConcurrentUtils",
+ ExecutorInstrumentation.class.getName() + "$DatadogWrapper",
+ ExecutorInstrumentation.class.getName() + "$CallableWrapper",
+ ExecutorInstrumentation.class.getName() + "$RunnableWrapper");
+
+ public ExecutorInstrumentation() {
+ super(EXEC_NAME);
+ }
+
+ @Override
+ public AgentBuilder apply(final AgentBuilder agentBuilder) {
+ return agentBuilder
+ .type(not(isInterface()).and(hasSuperType(named(Executor.class.getName()))))
+ .transform(EXEC_HELPER_INJECTOR)
+ .transform(
+ DDAdvice.create()
+ .advice(
+ named("execute").and(takesArgument(0, Runnable.class)),
+ WrapRunnableAdvice.class.getName()))
+ .asDecorator()
+ .type(not(isInterface()).and(hasSuperType(named(ExecutorService.class.getName()))))
+ .transform(EXEC_HELPER_INJECTOR)
+ .transform(
+ DDAdvice.create()
+ .advice(
+ named("submit").and(takesArgument(0, Runnable.class)),
+ WrapRunnableAdvice.class.getName()))
+ .transform(
+ DDAdvice.create()
+ .advice(
+ named("submit").and(takesArgument(0, Callable.class)),
+ WrapCallableAdvice.class.getName()))
+ .transform(
+ DDAdvice.create()
+ .advice(
+ nameMatches("invoke(Any|All)$").and(takesArgument(0, Callable.class)),
+ WrapCallableCollectionAdvice.class.getName()))
+ .asDecorator();
+ }
+
+ public static class WrapRunnableAdvice {
+ @Advice.OnMethodEnter(suppress = Throwable.class)
+ public static DatadogWrapper wrapJob(
+ @Advice.Argument(value = 0, readOnly = false) Runnable task) {
+ final Scope scope = GlobalTracer.get().scopeManager().active();
+ if (scope instanceof ContextPropagator && task != null && !(task instanceof DatadogWrapper)) {
+ task = new RunnableWrapper(task, (ContextPropagator) scope);
+ return (DatadogWrapper) task;
+ }
+ return null;
+ }
+
+ @Advice.OnMethodExit(onThrowable = Throwable.class, suppress = Throwable.class)
+ public static void checkCancel(
+ @Advice.Enter final DatadogWrapper wrapper, @Advice.Thrown final Throwable throwable) {
+ if (null != wrapper && null != throwable) {
+ wrapper.cancel();
+ }
+ }
+ }
+
+ public static class WrapCallableAdvice {
+ @Advice.OnMethodEnter(suppress = Throwable.class)
+ public static DatadogWrapper wrapJob(
+ @Advice.Argument(value = 0, readOnly = false) Callable task) {
+ final Scope scope = GlobalTracer.get().scopeManager().active();
+ if (scope instanceof ContextPropagator && task != null && !(task instanceof DatadogWrapper)) {
+ task = new CallableWrapper(task, (ContextPropagator) scope);
+ return (DatadogWrapper) task;
+ }
+ return null;
+ }
+
+ @Advice.OnMethodExit(onThrowable = Throwable.class, suppress = Throwable.class)
+ public static void checkCancel(
+ @Advice.Enter final DatadogWrapper wrapper, @Advice.Thrown final Throwable throwable) {
+ if (null != wrapper && null != throwable) {
+ wrapper.cancel();
+ }
+ }
+ }
+
+ public static class WrapCallableCollectionAdvice {
+ @Advice.OnMethodEnter(suppress = Throwable.class)
+ public static Collection<?> wrapJob(
+ @Advice.Argument(value = 0, readOnly = false) Collection<? extends Callable<?>> tasks) {
+ final Scope scope = GlobalTracer.get().scopeManager().active();
+ if (scope instanceof ContextPropagator) {
+ Collection<Callable<?>> wrappedTasks = new ArrayList<>(tasks.size());
+ for (Callable task : tasks) {
+ if (task != null) {
+ if (!(task instanceof CallableWrapper)) {
+ task = new CallableWrapper(task, (ContextPropagator) scope);
+ }
+ wrappedTasks.add(task);
+ }
+ }
+ tasks = wrappedTasks;
+ return tasks;
+ }
+ return null;
+ }
+
+ @Advice.OnMethodExit(onThrowable = Throwable.class, suppress = Throwable.class)
+ public static void checkCancel(
+ @Advice.Enter final Collection<?> wrappedJobs, @Advice.Thrown final Throwable throwable) {
+ if (null != wrappedJobs && null != throwable) {
+ for (Object wrapper : wrappedJobs) {
+ if (wrapper instanceof DatadogWrapper) {
+ ((DatadogWrapper) wrapper).cancel();
+ }
+ }
+ }
+ }
+ }
+
+ /** Marker interface for tasks which are wrapped to propagate the trace context. */
+ @Slf4j
+ public abstract static class DatadogWrapper {
+ protected final ContextPropagator.Continuation continuation;
+
+ public DatadogWrapper(ContextPropagator scope) {
+ continuation = scope.capture(true);
+ log.debug("created continuation {} from scope {}", continuation, scope);
+ }
+
+ public void cancel() {
+ if (null != continuation) {
+ continuation.activate().close();
+ log.debug("canceled continuation {}", continuation);
+ }
+ }
+ }
+
+ @Slf4j
+ public static class RunnableWrapper extends DatadogWrapper implements Runnable {
+ private final Runnable delegatee;
+
+ public RunnableWrapper(Runnable toWrap, ContextPropagator scope) {
+ super(scope);
+ delegatee = toWrap;
+ }
+
+ @Override
+ public void run() {
+ final Scope scope = continuation.activate();
+ try {
+ delegatee.run();
+ } finally {
+ scope.close();
+ }
+ }
+ }
+
+ @Slf4j
+ public static class CallableWrapper<T> extends DatadogWrapper implements Callable<T> {
+ private final Callable<T> delegatee;
+
+ public CallableWrapper(Callable<T> toWrap, ContextPropagator scope) {
+ super(scope);
+ delegatee = toWrap;
+ }
+
+ @Override
+ public T call() throws Exception {
+ final Scope scope = continuation.activate();
+ try {
+ return delegatee.call();
+ } finally {
+ scope.close();
+ }
+ }
+ }
+
+ public static class ConcurrentUtils {
+ private static Map<Class<?>, Field> fieldCache = new ConcurrentHashMap<>();
+ private static String[] wrapperFields = {"runnable", "callable"};
+
+ public static boolean safeToWrap(Future<?> f) {
+ return null != getDatadogWrapper(f);
+ }
+
+ public static DatadogWrapper getDatadogWrapper(Future<?> f) {
+ final Field field;
+ if (fieldCache.containsKey(f.getClass())) {
+ field = fieldCache.get(f.getClass());
+ } else {
+ field = getWrapperField(f.getClass());
+ fieldCache.put(f.getClass(), field);
+ }
+
+ if (field != null) {
+ try {
+ field.setAccessible(true);
+ Object o = field.get(f);
+ if (o instanceof DatadogWrapper) {
+ return (DatadogWrapper) o;
+ }
+ } catch (IllegalArgumentException | IllegalAccessException e) {
+ } finally {
+ field.setAccessible(false);
+ }
+ }
+ return null;
+ }
+
+ private static Field getWrapperField(Class<?> clazz) {
+ Field field = null;
+ while (field == null && clazz != null) {
+ for (int i = 0; i < wrapperFields.length; ++i) {
+ try {
+ field = clazz.getDeclaredField(wrapperFields[i]);
+ break;
+ } catch (Exception e) {
+ }
+ }
+ clazz = clazz.getSuperclass();
+ }
+ return field;
+ }
+ }
+}

ADD
dd-java-agent/instrumentation/java-concurrent/src/main/java/datadog/trace/instrumentation/java/concurrent/FutureInstrumentation.java
+package datadog.trace.instrumentation.java.concurrent;
+
+import static net.bytebuddy.matcher.ElementMatchers.hasSuperType;
+import static net.bytebuddy.matcher.ElementMatchers.isInterface;
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static net.bytebuddy.matcher.ElementMatchers.not;
+import static net.bytebuddy.matcher.ElementMatchers.returns;
+
+import com.google.auto.service.AutoService;
+import datadog.trace.agent.tooling.DDAdvice;
+import datadog.trace.agent.tooling.Instrumenter;
+import datadog.trace.instrumentation.java.concurrent.ExecutorInstrumentation.ConcurrentUtils;
+import datadog.trace.instrumentation.java.concurrent.ExecutorInstrumentation.DatadogWrapper;
+import java.util.concurrent.Future;
+import net.bytebuddy.agent.builder.AgentBuilder;
+import net.bytebuddy.asm.Advice;
+
+@AutoService(Instrumenter.class)
+public final class FutureInstrumentation extends Instrumenter.Configurable {
+
+ public FutureInstrumentation() {
+ super(ExecutorInstrumentation.EXEC_NAME);
+ }
+
+ @Override
+ public AgentBuilder apply(final AgentBuilder agentBuilder) {
+ return agentBuilder
+ .type(not(isInterface()).and(hasSuperType(named(Future.class.getName()))))
+ .transform(ExecutorInstrumentation.EXEC_HELPER_INJECTOR)
+ .transform(
+ DDAdvice.create()
+ .advice(
+ named("cancel").and(returns(boolean.class)),
+ CanceledFutureAdvice.class.getName()))
+ .asDecorator();
+ }
+
+ public static class CanceledFutureAdvice {
+ @Advice.OnMethodEnter(suppress = Throwable.class)
+ public static DatadogWrapper findWrapper(@Advice.This Future<?> future) {
+ return ConcurrentUtils.getDatadogWrapper(future);
+ }
+
+ @Advice.OnMethodExit(suppress = Throwable.class)
+ public static void abortTracing(
+ @Advice.Enter final DatadogWrapper wrapper, @Advice.Return boolean canceled) {
+ if (canceled && null != wrapper) {
+ wrapper.cancel();
+ }
+ }
+ }
+}

MODIFY
dd-trace-api/dd-trace-api.gradle
dd-trace-api/dd-trace-api.gradle
@@ -5,5 +5,6 @@ apply from: "${rootDir}/gradle/jacoco.gradle"
description = 'dd-trace-api'
dependencies {
compile deps.slf4j
+ compile deps.opentracing
testCompile deps.junit
}

ADD
dd-trace-api/src/main/java/datadog/trace/context/ContextPropagator.java
+package datadog.trace.context;
+
+import io.opentracing.Scope;
+
+/** An object when can propagate a datadog trace across multiple threads. */
+public interface ContextPropagator {
+ /**
+ * Prevent the trace attached to this ContextPropagator from reporting until the returned
+ * Continuation finishes.
+ */
+ public Continuation capture(boolean finishOnClose);
+
+ /** Used to pass async context between workers. */
+ public static interface Continuation {
+ /**
+ * Activate the continuation. When the returned scope is closed, the continuation will no longer
+ * prevent the trace from reporting.
+ */
+ public Scope activate();
+ }
+}

MODIFY
dd-trace-ot/src/main/java/datadog/opentracing/PendingTrace.java
dd-trace-ot/src/main/java/datadog/opentracing/PendingTrace.java
@@ -44,9 +44,16 @@ public class PendingTrace extends ConcurrentLinkedDeque<DDSpan> {
log.warn("{} - span registered for wrong trace ({})", span, traceId);
return;
}
+ synchronized (span) {
+ if (null == span.ref) {
span.ref = new WeakReference<DDSpan>(span, referenceQueue);
weakReferences.add(span.ref);
- pendingReferenceCount.incrementAndGet();
+ final int count = pendingReferenceCount.incrementAndGet();
+ log.debug("traceId: {} -- registered span {}. count = {}", traceId, span, count);
+ } else {
+ log.debug("span {} already registered in trace {}", span, traceId);
+ }
+ }
}
private void expireSpan(final DDSpan span) {
@@ -54,11 +61,17 @@ public class PendingTrace extends ConcurrentLinkedDeque<DDSpan> {
log.warn("{} - span expired for wrong trace ({})", span, traceId);
return;
}
+ synchronized (span) {
+ if (null == span.ref) {
+ log.debug("span {} not registered in trace {}", span, traceId);
+ } else {
weakReferences.remove(span.ref);
span.ref.clear();
span.ref = null;
expireReference();
}
+ }
+ }
public void addSpan(final DDSpan span) {
if (span.getDurationNano() == 0) {
@@ -83,27 +96,39 @@ public class PendingTrace extends ConcurrentLinkedDeque<DDSpan> {
* completed, so we need to wait till continuations are de-referenced before reporting.
*/
public void registerContinuation(final ContinuableScope.Continuation continuation) {
+ synchronized (continuation) {
+ if (continuation.ref == null) {
continuation.ref =
new WeakReference<ContinuableScope.Continuation>(continuation, referenceQueue);
weakReferences.add(continuation.ref);
- pendingReferenceCount.incrementAndGet();
+ final int count = pendingReferenceCount.incrementAndGet();
+ log.debug(
+ "traceId: {} -- registered continuation {}. count = {}", traceId, continuation, count);
+ } else {
+ log.debug("continuation {} already registered in trace {}", continuation, traceId);
+ }
+ }
}
public void cancelContinuation(final ContinuableScope.Continuation continuation) {
synchronized (continuation) {
- if (continuation.ref != null) {
+ if (continuation.ref == null) {
+ log.debug("continuation {} not registered in trace {}", continuation, traceId);
+ } else {
weakReferences.remove(continuation.ref);
continuation.ref.clear();
continuation.ref = null;
+ expireReference();
}
}
- expireReference();
}
private void expireReference() {
- if (pendingReferenceCount.decrementAndGet() == 0) {
+ final int count = pendingReferenceCount.decrementAndGet();
+ if (count == 0) {
write();
}
+ log.debug("traceId: {} -- Expired reference. count = {}", traceId, count);
}
private void write() {

MODIFY
dd-trace-ot/src/main/java/datadog/opentracing/scopemanager/ContinuableScope.java
dd-trace-ot/src/main/java/datadog/opentracing/scopemanager/ContinuableScope.java
@@ -2,6 +2,7 @@ package datadog.opentracing.scopemanager;
import datadog.opentracing.DDSpanContext;
import datadog.opentracing.PendingTrace;
+import datadog.trace.context.ContextPropagator;
import io.opentracing.Scope;
import io.opentracing.Span;
import io.opentracing.noop.NoopScopeManager;
@@ -12,7 +13,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import lombok.extern.slf4j.Slf4j;
@Slf4j
-public class ContinuableScope implements Scope {
+public class ContinuableScope implements Scope, ContextPropagator {
final ContextualScopeManager scopeManager;
final AtomicInteger refCount;
private final Span wrapped;
@@ -66,7 +67,7 @@ public class ContinuableScope implements Scope {
return new Continuation(this.finishOnClose && finishOnClose);
}
- public class Continuation implements Closeable {
+ public class Continuation implements Closeable, ContextPropagator.Continuation {
public WeakReference<Continuation> ref;
private final AtomicBoolean used = new AtomicBoolean(false);
@@ -109,21 +110,21 @@ public class ContinuableScope implements Scope {
}
private class ClosingScope implements Scope {
- private final Scope wrapped;
+ private final Scope wrappedScope;
- private ClosingScope(final Scope wrapped) {
- this.wrapped = wrapped;
+ private ClosingScope(final Scope wrappedScope) {
+ this.wrappedScope = wrappedScope;
}
@Override
public void close() {
- wrapped.close();
+ wrappedScope.close();
Continuation.this.close();
}
@Override
public Span span() {
- return wrapped.span();
+ return wrappedScope.span();
}
}
}

MODIFY
dd-trace-ot/src/test/groovy/datadog/opentracing/scopemanager/ScopeManagerTest.groovy
dd-trace-ot/src/test/groovy/datadog/opentracing/scopemanager/ScopeManagerTest.groovy
@@ -211,7 +211,7 @@ class ScopeManagerTest extends Specification {
def newScope = continuation.activate()
then:
- scopeManager.active() == newScope.wrapped
+ scopeManager.active() == newScope.wrappedScope
newScope != childScope && newScope != parentScope
newScope.span() == childSpan
!spanFinished(childSpan)
@@ -241,7 +241,7 @@ class ScopeManagerTest extends Specification {
expect:
newScope != scope
- scopeManager.active() == newScope.wrapped
+ scopeManager.active() == newScope.wrappedScope
spanFinished(span)
writer == []

MODIFY
settings.gradle
settings.gradle
@@ -14,6 +14,7 @@ include ':dd-java-agent:instrumentation:apache-httpclient-4.3'
include ':dd-java-agent:instrumentation:aws-sdk'
include ':dd-java-agent:instrumentation:datastax-cassandra-3.2'
include ':dd-java-agent:instrumentation:jax-rs'
+include ':dd-java-agent:instrumentation:java-concurrent'
include ':dd-java-agent:instrumentation:jboss-classloading'
include ':dd-java-agent:instrumentation:jdbc'
include ':dd-java-agent:instrumentation:jedis-1.4'

Executor Instrumentation for Scala