RFC only! (ActiveSpanManager etc)

opentracing-api/pom.xml

@@ -40,5 +40,20 @@
             <version>1.10.19</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-api</artifactId>
+            <version>1.7.23</version>
+        </dependency>
+        <dependency>
+            <groupId>log4j</groupId>
+            <artifactId>log4j</artifactId>
+            <version>1.2.17</version>
+        </dependency>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-log4j12</artifactId>
+            <version>1.7.23</version>
+        </dependency>
     </dependencies>
 </project>

opentracing-api/src/main/java/io/opentracing/ActiveSpanManager.java

@@ -0,0 +1,23 @@
+package io.opentracing;
+
+public interface ActiveSpanManager {
+    // Basically a marker interface
+    public interface Snapshot {
+        Span span();
+    }
+
+    // Get the currently active Span (perhaps for this thread, etc)
+    Span active();
+
+    // Create a Snapshot encapsulating both the given Span and any state needed to activate/deactivate (see below)
+    Snapshot snapshot(Span span);
+
+    // Make the Snapshot and the Span it contains "active" per active().
+    //
+    // *Must* be paired with a subsequent call to deactivate().
+    Span activate(Snapshot snapshot);
+
+    // See activate() above.
+    void deactivate(Snapshot snapshot);
+
+}

opentracing-api/src/main/java/io/opentracing/MDCActiveSpanManager.java

@@ -0,0 +1,80 @@
+package io.opentracing;
+
+import org.slf4j.MDC;
+import sun.awt.image.ImageWatched;
+
+import java.util.Map;
+
+/**
+ * XXX: comment
+ */
+public class MDCActiveSpanManager implements io.opentracing.ActiveSpanManager {
+    private final ThreadLocal<MDCSnapshot> tlsSnapshot = new ThreadLocal<MDCSnapshot>();
+
+    class MDCSnapshot implements ActiveSpanManager.Snapshot {
+        private final Map<String, String> mdcContext;
+        private final Span span;
+        private MDCSnapshot toRestore = null;
+
+        MDCSnapshot(Span span) {
+            this.mdcContext = MDC.getCopyOfContextMap();
+            this.span = span;
+        }
+
+        public Span span() {
+            return span;
+        }
+
+        void setToRestore(MDCSnapshot toRestore) {
+            this.toRestore = toRestore;
+        }
+
+        MDCSnapshot toRestore() {
+            return toRestore;
+        }
+    }
+
+    @Override
+    public Span active() {
+        MDCSnapshot snapshot = tlsSnapshot.get();
+        if (snapshot == null) {
+            return null;
+        }
+        return snapshot.span();
+    }
+
+    @Override
+    public MDCSnapshot snapshot(Span span) {
+        return new MDCSnapshot(span);
+    }
+
+    @Override
+    public Span activate(Snapshot snapshot) {
+        if (!(snapshot instanceof MDCSnapshot)) {
+            throw new IllegalArgumentException("activate() expected MDCSnapshot");
+        }
+        ((MDCSnapshot) snapshot).setToRestore(tlsSnapshot.get());
+        tlsSnapshot.set((MDCSnapshot)snapshot);
+        return snapshot.span();
+    }
+
+    @Override
+    public void deactivate(Snapshot snapshot) {
+        if (!(snapshot instanceof MDCSnapshot)) {
+            throw new IllegalArgumentException("deactivate() expected MDCSnapshot");
+        }
+
+        if (tlsSnapshot.get() != snapshot) {
+            // do nothing
+            return;
+        }
+        MDCSnapshot nextActiveSnapshot = ((MDCSnapshot)snapshot).toRestore();
+        while (nextActiveSnapshot != null) {
+            if (!nextActiveSnapshot.span().isFinished()) {
+                break;
+            }
+            nextActiveSnapshot = nextActiveSnapshot.toRestore();
+        }
+        tlsSnapshot.set(nextActiveSnapshot);
+    }
+}

opentracing-api/src/main/java/io/opentracing/Span.java

@@ -41,6 +41,9 @@ public interface Span extends Closeable {
      */
     void finish();
 
+    // XXX: comment
+    boolean isFinished();
+
     /**
      * Sets an explicit end timestamp and records the span.
      *

opentracing-api/src/main/java/io/opentracing/Tracer.java

@@ -41,6 +41,9 @@ public interface Tracer {
    */
   SpanBuilder buildSpan(String operationName);
 
+  void setActiveSpanManager(ActiveSpanManager mgr);
+  ActiveSpanManager activeSpanManager();
+
   /**
    * Inject a SpanContext into a `carrier` of a given type, presumably for propagation across process boundaries.
    *
@@ -88,7 +91,8 @@ public interface Tracer {
    */
   <C> SpanContext extract(Format<C> format, C carrier);
 
-
+  // XXX(bhs): could make this an abstract class. In any case, by default a SpanBuilder will have an asChildOf pointer
+  // to the ActiveSpanManager's active Span's SpanContext.
   interface SpanBuilder extends SpanContext {
 
       /**

opentracing-impl/src/main/java/io/opentracing/concurrent/TracedCallable.java

@@ -0,0 +1,38 @@
+package io.opentracing.concurrent;
+
+import io.opentracing.ActiveSpanManager;
+import io.opentracing.Span;
+import io.opentracing.Tracer;
+import io.opentracing.impl.GlobalTracer;
+
+import java.util.concurrent.Callable;
+
+public class TracedCallable<T> implements Callable<T> {
+    private ActiveSpanManager.Snapshot snapshot;
+    private ActiveSpanManager manager;
+    private Callable<T> callable;
+
+    public TracedCallable(Callable<T> callable) {
+        this(callable, GlobalTracer.get().activeSpanManager());
+    }
+
+    public TracedCallable(Callable<T> callable, ActiveSpanManager manager) {
+        this(callable, manager.active(), manager);
+    }
+
+    public TracedCallable(Callable<T> callable, Span span, ActiveSpanManager manager) {
+        if (callable == null) throw new NullPointerException("Callable is <null>.");
+        this.callable= callable;
+        this.manager = manager;
+        this.snapshot = manager.snapshot(span);
+    }
+
+    public T call() throws Exception {
+        final Span span = manager.activate(snapshot);
+        try {
+            return callable.call();
+        } finally {
+            manager.deactivate(snapshot);
+        }
+    }
+}

opentracing-impl/src/main/java/io/opentracing/concurrent/TracedExecutorService.java

@@ -0,0 +1,102 @@
+package io.opentracing.concurrent;
+
+import io.opentracing.ActiveSpanManager;
+import io.opentracing.Span;
+import io.opentracing.Tracer;
+import io.opentracing.impl.GlobalTracer;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.*;
+
+public class TracedExecutorService implements ExecutorService {
+    private ExecutorService executor;
+    private ActiveSpanManager manager;
+
+    public TracedExecutorService(ExecutorService executor){
+        this(executor, GlobalTracer.get().activeSpanManager());
+    }
+
+    public TracedExecutorService(ExecutorService executor, ActiveSpanManager manager) {
+        if (executor == null) throw new NullPointerException("Executor is <null>.");
+        if (manager == null) throw new NullPointerException("ActiveSpanManager is <null>.");
+        this.executor = executor;
+        this.manager = manager;
+    }
+
+    @Override
+    public void execute(Runnable command) {
+        executor.execute(new TracedRunnable(command, manager));
+    }
+
+    @Override
+    public Future<?> submit(Runnable task) {
+        return executor.submit(new TracedRunnable(task, manager));
+    }
+
+    @Override
+    public <T> Future<T> submit(Runnable task, T result) {
+        return executor.submit(new TracedRunnable(task, manager), result);
+    }
+
+    @Override
+    public <T> Future<T> submit(Callable<T> task) {
+        return executor.submit(new TracedCallable(task, manager));
+    }
+
+    @Override
+    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException {
+        return executor.invokeAll(tasksWithTracing(tasks));
+    }
+
+    @Override
+    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
+        throws InterruptedException {
+        return executor.invokeAll(tasksWithTracing(tasks), timeout, unit);
+    }
+
+    @Override
+    public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException {
+        return executor.invokeAny(tasksWithTracing(tasks));
+    }
+
+    @Override
+    public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
+        throws InterruptedException, ExecutionException, TimeoutException {
+        return executor.invokeAny(tasksWithTracing(tasks), timeout, unit);
+    }
+
+    @Override
+    public void shutdown() {
+        executor.shutdown();
+    }
+
+    @Override
+    public List<Runnable> shutdownNow() {
+        return executor.shutdownNow();
+    }
+
+    @Override
+    public boolean isShutdown() {
+        return executor.isShutdown();
+    }
+
+    @Override
+    public boolean isTerminated() {
+        return executor.isTerminated();
+    }
+
+    @Override
+    public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
+        return executor.awaitTermination(timeout, unit);
+    }
+
+    private <T> Collection<? extends Callable<T>> tasksWithTracing(
+        Collection<? extends Callable<T>> tasks) {
+        if (tasks == null) throw new NullPointerException("Collection of tasks is <null>.");
+        Collection<Callable<T>> result = new ArrayList<Callable<T>>(tasks.size());
+        for (Callable<T> task : tasks) result.add(new TracedCallable(task, manager));
+        return result;
+    }
+}

opentracing-impl/src/main/java/io/opentracing/concurrent/TracedRunnable.java

@@ -0,0 +1,42 @@
+package io.opentracing.concurrent;
+
+import io.opentracing.ActiveSpanManager;
+import io.opentracing.Span;
+import io.opentracing.Tracer;
+import io.opentracing.impl.GlobalTracer;
+
+
+public class TracedRunnable implements Runnable {
+    private Runnable runnable;
+    private ActiveSpanManager manager;
+    private ActiveSpanManager.Snapshot snapshot;
+
+    public TracedRunnable(Runnable runnable) {
+        this(runnable, GlobalTracer.get().activeSpanManager());
+    }
+
+    public TracedRunnable(Runnable runnable, Span span) {
+        this(runnable, span, GlobalTracer.get().activeSpanManager());
+    }
+
+    public TracedRunnable(Runnable runnable, ActiveSpanManager manager) {
+        this(runnable, manager.active(), manager);
+    }
+
+    public TracedRunnable(Runnable runnable, Span span, ActiveSpanManager manager) {
+        if (runnable == null) throw new NullPointerException("Runnable is <null>.");
+        this.runnable = runnable;
+        this.manager = manager;
+        this.snapshot = manager.snapshot(span);
+    }
+
+    @Override
+    public void run() {
+        final Span span = manager.activate(this.snapshot);
+        try {
+            runnable.run();
+        } finally {
+            manager.deactivate(this.snapshot);
+        }
+    }
+}

opentracing-impl/src/main/java/io/opentracing/impl/AbstractSpan.java

@@ -55,6 +55,11 @@ public void finish() {
         duration = Duration.between(start, Instant.now());
     }
 
+    @Override
+    public boolean isFinished() {
+        return null != duration;
+    }
+
     @Override
     public void finish(long finishMicros) {
         long finishEpochSeconds = TimeUnit.MICROSECONDS.toSeconds(finishMicros);

opentracing-impl/src/main/java/io/opentracing/impl/AbstractTracer.java

@@ -13,6 +13,8 @@
  */
 package io.opentracing.impl;
 
+import io.opentracing.ActiveSpanManager;
+import io.opentracing.Span;
 import io.opentracing.SpanContext;
 import io.opentracing.Tracer;
 import io.opentracing.propagation.Extractor;
@@ -28,6 +30,7 @@ abstract class AbstractTracer implements Tracer {
     static final boolean BAGGAGE_ENABLED = !Boolean.getBoolean("opentracing.propagation.dropBaggage");
 
     private final PropagationRegistry registry = new PropagationRegistry();
+    private ActiveSpanManager manager;
 
     protected AbstractTracer() {
         registry.register(Format.Builtin.TEXT_MAP, new TextMapInjectorImpl(this));
@@ -36,6 +39,16 @@ protected AbstractTracer() {
 
     abstract AbstractSpanBuilder createSpanBuilder(String operationName);
 
+    @Override
+    public void setActiveSpanManager(ActiveSpanManager manager) {
+        this.manager = manager;
+    }
+
+    @Override
+    public ActiveSpanManager activeSpanManager() {
+        return this.manager;
+    }
+
     @Override
     public SpanBuilder buildSpan(String operationName){
         return createSpanBuilder(operationName);