Fix concurrency problem with DDSpanContext
Also apply Java 1.8 compatability to async-tracing example.
DELETE
dd-trace-annotations/pom.xml
-<?xml version="1.0"?>
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
- <modelVersion>4.0.0</modelVersion>
- <parent>
- <groupId>com.datadoghq</groupId>
- <artifactId>dd-trace-java</artifactId>
- <version>0.1.2-SNAPSHOT</version>
- </parent>
- <artifactId>dd-trace-annotations</artifactId>
- <version>0.1.2-SNAPSHOT</version>
- <name>dd-trace-annotations</name>
- <url>https://github.com/datadog/dd-trace-java</url>
- <packaging>jar</packaging>
-</project>

MODIFY
dd-trace-examples/async-tracing/async-tracing.gradle
dd-trace-examples/async-tracing/async-tracing.gradle
apply from: "${rootDir}/gradle/java.gradle"
apply from: "${rootDir}/gradle/jacoco.gradle"
+sourceCompatibility = 1.8
+targetCompatibility = 1.8
+
description = 'async-tracing'
dependencies {
compile project(':dd-trace')
compile group: 'io.opentracing.contrib', name: 'opentracing-spanmanager', version: '0.0.5'
compile group: 'ch.qos.logback', name: 'logback-classic', version: '1.2.3'
compile group:'io.reactivex.rxjava2', name:'rxjava', version:'2.1.1'
-
}

MODIFY
dd-trace-examples/async-tracing/src/main/java/com/datadoghq/trace/AsyncExample.java
dd-trace-examples/async-tracing/src/main/java/com/datadoghq/trace/AsyncExample.java
@@ -16,9 +16,7 @@ public class AsyncExample {
public static Integer bar() throws Exception {
- try (ActiveSpan __ = GlobalTracer.get()
- .buildSpan("bar")
- .startActive()) {
+ try (ActiveSpan __ = GlobalTracer.get().buildSpan("bar").startActive()) {
System.out.println("bar");
Thread.sleep(1000);
}
@@ -28,27 +26,23 @@ public class AsyncExample {
public static Future<Integer> foo() throws Exception {
- try (ActiveSpan span = GlobalTracer.get()
- .buildSpan("foo")
- .startActive()) {
+ try (ActiveSpan span = GlobalTracer.get().buildSpan("foo").startActive()) {
System.out.println("foo");
Thread.sleep(500);
final ActiveSpan.Continuation cont = span.capture();
- Future<Integer> future = ex.submit(() -> {
+ Future<Integer> future =
+ ex.submit(
+ () -> {
try (ActiveSpan __ = cont.activate()) {
return bar();
}
});
-
-
return future;
-
-
}
}
@@ -63,7 +57,5 @@ public class AsyncExample {
writer.close();
ex.shutdownNow();
-
}
-
}

MODIFY
dd-trace-examples/async-tracing/src/main/java/com/datadoghq/trace/AsyncExampleReactive.java
dd-trace-examples/async-tracing/src/main/java/com/datadoghq/trace/AsyncExampleReactive.java
@@ -8,14 +8,10 @@ import io.opentracing.Tracer;
import io.opentracing.util.GlobalTracer;
import io.reactivex.Observable;
-
public class AsyncExampleReactive {
-
public static Integer bar() throws Exception {
- try (ActiveSpan __ = GlobalTracer.get()
- .buildSpan("bar")
- .startActive()) {
+ try (ActiveSpan __ = GlobalTracer.get().buildSpan("bar").startActive()) {
System.out.println("bar");
Thread.sleep(1000);
}
@@ -24,15 +20,14 @@ public class AsyncExampleReactive {
}
public static Observable<Integer> foo() throws Exception {
- try (ActiveSpan span = GlobalTracer.get()
- .buildSpan("foo")
- .startActive()) {
+ try (ActiveSpan span = GlobalTracer.get().buildSpan("foo").startActive()) {
System.out.println("foo");
Thread.sleep(500);
final ActiveSpan.Continuation cont = span.capture();
- return Observable.fromCallable(() -> {
+ return Observable.fromCallable(
+ () -> {
try (ActiveSpan __ = cont.activate()) {
return bar();
}
@@ -49,6 +44,4 @@ public class AsyncExampleReactive {
writer.close();
}
-
}
-

MODIFY
dd-trace/src/main/java/com/datadoghq/trace/DDSpanContext.java
dd-trace/src/main/java/com/datadoghq/trace/DDSpanContext.java
@@ -4,7 +4,11 @@ import com.datadoghq.trace.integration.DDSpanContextDecorator;
import com.fasterxml.jackson.annotation.JsonIgnore;
import com.google.common.collect.Maps;
import io.opentracing.tag.Tags;
-import java.util.*;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
/**
* SpanContext represents Span state that must propagate to descendant Spans and across process
@@ -47,18 +51,18 @@ public class DDSpanContext implements io.opentracing.SpanContext {
private final DDTracer tracer;
public DDSpanContext(
- long traceId,
- long spanId,
- long parentId,
- String serviceName,
- String operationName,
- String resourceName,
- Map<String, String> baggageItems,
- boolean errorFlag,
- String spanType,
- Map<String, Object> tags,
- List<DDBaseSpan<?>> trace,
- DDTracer tracer) {
+ final long traceId,
+ final long spanId,
+ final long parentId,
+ final String serviceName,
+ final String operationName,
+ final String resourceName,
+ final Map<String, String> baggageItems,
+ final boolean errorFlag,
+ final String spanType,
+ final Map<String, Object> tags,
+ final List<DDBaseSpan<?>> trace,
+ final DDTracer tracer) {
this.traceId = traceId;
this.spanId = spanId;
@@ -79,7 +83,8 @@ public class DDSpanContext implements io.opentracing.SpanContext {
this.tags = tags;
if (trace == null) {
- this.trace = new ArrayList<DDBaseSpan<?>>();
+ // TODO: figure out better concurrency model.
+ this.trace = new CopyOnWriteArrayList<>(); // must be thread safe!
} else {
this.trace = trace;
}
@@ -113,7 +118,7 @@ public class DDSpanContext implements io.opentracing.SpanContext {
return errorFlag;
}
- public void setErrorFlag(boolean errorFlag) {
+ public void setErrorFlag(final boolean errorFlag) {
this.errorFlag = errorFlag;
}
@@ -121,14 +126,14 @@ public class DDSpanContext implements io.opentracing.SpanContext {
return spanType;
}
- public void setBaggageItem(String key, String value) {
+ public void setBaggageItem(final String key, final String value) {
if (this.baggageItems.isEmpty()) {
this.baggageItems = new HashMap<String, String>();
}
this.baggageItems.put(key, value);
}
- public String getBaggageItem(String key) {
+ public String getBaggageItem(final String key) {
return this.baggageItems.get(key);
}
@@ -139,6 +144,7 @@ public class DDSpanContext implements io.opentracing.SpanContext {
/* (non-Javadoc)
* @see io.opentracing.SpanContext#baggageItems()
*/
+ @Override
public Iterable<Map.Entry<String, String>> baggageItems() {
return this.baggageItems.entrySet();
}
@@ -159,7 +165,7 @@ public class DDSpanContext implements io.opentracing.SpanContext {
* @param tag the tag-name
* @param value the value of the value
*/
- public synchronized void setTag(String tag, Object value) {
+ public synchronized void setTag(final String tag, final Object value) {
if (tag.equals(DDTags.SERVICE_NAME)) {
setServiceName(value.toString());
return;
@@ -177,9 +183,9 @@ public class DDSpanContext implements io.opentracing.SpanContext {
this.tags.put(tag, value);
//Call decorators
- List<DDSpanContextDecorator> decorators = tracer.getSpanContextDecorators(tag);
+ final List<DDSpanContextDecorator> decorators = tracer.getSpanContextDecorators(tag);
if (decorators != null) {
- for (DDSpanContextDecorator decorator : decorators) {
+ for (final DDSpanContextDecorator decorator : decorators) {
decorator.afterSetTag(this, tag, value);
}
}
@@ -216,7 +222,7 @@ public class DDSpanContext implements io.opentracing.SpanContext {
.toString();
}
- public void setOperationName(String operationName) {
+ public void setOperationName(final String operationName) {
this.operationName = operationName;
}
@@ -224,15 +230,15 @@ public class DDSpanContext implements io.opentracing.SpanContext {
return operationName;
}
- public void setServiceName(String serviceName) {
+ public void setServiceName(final String serviceName) {
this.serviceName = serviceName;
}
- public void setResourceName(String resourceName) {
+ public void setResourceName(final String resourceName) {
this.resourceName = resourceName;
}
- public void setSpanType(String spanType) {
+ public void setSpanType(final String spanType) {
this.spanType = spanType;
}
}

MODIFY
dd-trace/src/main/java/com/datadoghq/trace/writer/ListWriter.java
dd-trace/src/main/java/com/datadoghq/trace/writer/ListWriter.java
package com.datadoghq.trace.writer;
import com.datadoghq.trace.DDBaseSpan;
-import java.util.ArrayList;
import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
/** List writer used by tests mostly */
-public class ListWriter extends ArrayList<List<DDBaseSpan<?>>> implements Writer {
+public class ListWriter extends CopyOnWriteArrayList<List<DDBaseSpan<?>>> implements Writer {
public List<List<DDBaseSpan<?>>> getList() {
return this;
@@ -16,7 +16,7 @@ public class ListWriter extends ArrayList<List<DDBaseSpan<?>>> implements Writer
}
@Override
- public void write(List<DDBaseSpan<?>> trace) {
+ public void write(final List<DDBaseSpan<?>> trace) {
add(trace);
}

MODIFY
dd-trace/src/test/groovy/com/datadoghq/trace/ActiveSpanContinuationTest.groovy
dd-trace/src/test/groovy/com/datadoghq/trace/ActiveSpanContinuationTest.groovy
package com.datadoghq.trace
import com.datadoghq.trace.writer.ListWriter
+import io.opentracing.util.ThreadLocalActiveSpan
import spock.lang.Ignore
import spock.lang.Specification
import java.util.concurrent.Phaser
+import java.util.concurrent.atomic.AtomicInteger
class ActiveSpanContinuationTest extends Specification {
def traceCollector = new ListWriter()
def tracer = new DDTracer(traceCollector)
def activeSpan = tracer.buildSpan("test").startActive()
+ AtomicInteger continuationCount
+
+ def setup() {
+ def field = ThreadLocalActiveSpan.getDeclaredField("refCount")
+ field.setAccessible(true)
+ continuationCount = field.get(activeSpan)
+ }
def "calling activate from multiple continuations at once with no child spans tracks separately"() {
setup:
@@ -46,7 +55,7 @@ class ActiveSpanContinuationTest extends Specification {
traceCollector.firstTrace().size() == 1
where:
- count = 5
+ count = new Random().nextInt(50) + 5
}
def "concurrent threads with manual spans and continuations report correctly"() {
@@ -55,12 +64,13 @@ class ActiveSpanContinuationTest extends Specification {
phaser.register()
for (int i = 0; i < count; i++) {
+ String spanName = "child " + i
phaser.register()
def capture = activeSpan.capture()
new Thread({
phaser.arriveAndAwaitAdvance()
def activeSpan = capture.activate()
- def childSpan = tracer.buildSpan("child " + i).startManual()
+ def childSpan = tracer.buildSpan(spanName).startManual()
phaser.arriveAndAwaitAdvance()
childSpan.finish()
activeSpan.deactivate()
@@ -68,12 +78,15 @@ class ActiveSpanContinuationTest extends Specification {
}).start()
}
- activeSpan.deactivate() // allow the trace to be reported when all continuations deactivate
+ expect:
+ continuationCount.get() == count + 1
when:
+ activeSpan.deactivate() // allow the trace to be reported when all continuations deactivate
phaser.arriveAndAwaitAdvance() //allow threads to activate capture
then:
+ continuationCount.get() == count
traceCollector == []
when:
@@ -83,6 +96,7 @@ class ActiveSpanContinuationTest extends Specification {
traceCollector.size()
then:
+ continuationCount.get() == 0
traceCollector.size() == 1
def trace = traceCollector.remove(0)
def parent = trace.remove(0)
@@ -95,7 +109,7 @@ class ActiveSpanContinuationTest extends Specification {
}
where:
- count = 3
+ count = new Random().nextInt(50) + 5
}
def "concurrent threads with active spans and continuations report correctly"() {
@@ -118,12 +132,15 @@ class ActiveSpanContinuationTest extends Specification {
}).start()
}
- activeSpan.deactivate() // allow the trace to be reported when all continuations deactivate
+ expect:
+ continuationCount.get() == count + 1
when:
+ activeSpan.deactivate() // allow the trace to be reported when all continuations deactivate
phaser.arriveAndAwaitAdvance() //allow threads to activate capture
then:
+ continuationCount.get() == count
traceCollector == []
when:
@@ -133,6 +150,7 @@ class ActiveSpanContinuationTest extends Specification {
traceCollector.size()
then:
+ continuationCount.get() == 0
traceCollector.size() == 1
def trace = traceCollector.remove(0)
def parent = trace.remove(0)
@@ -145,7 +163,7 @@ class ActiveSpanContinuationTest extends Specification {
}
where:
- count = 3
+ count = new Random().nextInt(50) + 5
}
@Ignore("Not yet implemented in ThreadLocalActiveSpan.Continuation")

Fix concurrency problem with DDSpanContext

Also apply Java 1.8 compatability to async-tracing example.