Merge pull request #48 from DataDog/tyler/async-testing
Use ThreadLocalActiveSpanSource instead of DDActiveSpan
This commit is contained in:
Tyler Benson
GitHub
commit
9bcfe1ef34
|
|
@ -1,14 +0,0 @@
|
|||
<?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>
|
||||
|
|
@ -1,8 +1,13 @@
|
|||
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'
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,61 @@
|
|||
package com.datadoghq.trace;
|
||||
|
||||
import com.datadoghq.trace.sampling.AllSampler;
|
||||
import com.datadoghq.trace.writer.DDAgentWriter;
|
||||
import com.datadoghq.trace.writer.DDApi;
|
||||
import io.opentracing.ActiveSpan;
|
||||
import io.opentracing.Tracer;
|
||||
import io.opentracing.util.GlobalTracer;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.Future;
|
||||
|
||||
public class AsyncExample {
|
||||
|
||||
private static ExecutorService ex = Executors.newFixedThreadPool(1);
|
||||
|
||||
public static Integer bar() throws Exception {
|
||||
|
||||
try (ActiveSpan __ = GlobalTracer.get().buildSpan("bar").startActive()) {
|
||||
System.out.println("bar");
|
||||
Thread.sleep(1000);
|
||||
}
|
||||
|
||||
return 42;
|
||||
}
|
||||
|
||||
public static Future<Integer> foo() throws Exception {
|
||||
|
||||
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(
|
||||
() -> {
|
||||
try (ActiveSpan __ = cont.activate()) {
|
||||
|
||||
return bar();
|
||||
}
|
||||
});
|
||||
|
||||
return future;
|
||||
}
|
||||
}
|
||||
|
||||
public static void main(String[] args) throws Exception {
|
||||
|
||||
DDAgentWriter writer = new DDAgentWriter(new DDApi("localhost", 8126));
|
||||
Tracer tracer = new DDTracer("dd-trace-test-app", writer, new AllSampler());
|
||||
// Tracer tracer = new DDTracer(new LoggingWriter(), new AllSampler());
|
||||
GlobalTracer.register(tracer);
|
||||
|
||||
System.out.printf("%d%n", foo().get());
|
||||
|
||||
writer.close();
|
||||
ex.shutdownNow();
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,47 @@
|
|||
package com.datadoghq.trace;
|
||||
|
||||
import com.datadoghq.trace.sampling.AllSampler;
|
||||
import com.datadoghq.trace.writer.DDAgentWriter;
|
||||
import com.datadoghq.trace.writer.DDApi;
|
||||
import io.opentracing.ActiveSpan;
|
||||
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()) {
|
||||
System.out.println("bar");
|
||||
Thread.sleep(1000);
|
||||
}
|
||||
|
||||
return 42;
|
||||
}
|
||||
|
||||
public static Observable<Integer> foo() throws Exception {
|
||||
try (ActiveSpan span = GlobalTracer.get().buildSpan("foo").startActive()) {
|
||||
|
||||
System.out.println("foo");
|
||||
Thread.sleep(500);
|
||||
|
||||
final ActiveSpan.Continuation cont = span.capture();
|
||||
return Observable.fromCallable(
|
||||
() -> {
|
||||
try (ActiveSpan __ = cont.activate()) {
|
||||
return bar();
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
public static void main(String[] args) throws Exception {
|
||||
DDAgentWriter writer = new DDAgentWriter(new DDApi("localhost", 8126));
|
||||
Tracer tracer = new DDTracer("dd-trace-test-app", writer, new AllSampler());
|
||||
GlobalTracer.register(tracer);
|
||||
|
||||
foo().subscribe(System.out::println);
|
||||
|
||||
writer.close();
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,16 @@
|
|||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<configuration>
|
||||
|
||||
<appender name="console" class="ch.qos.logback.core.ConsoleAppender">
|
||||
<layout class="ch.qos.logback.classic.PatternLayout">
|
||||
<Pattern>
|
||||
%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n
|
||||
</Pattern>
|
||||
</layout>
|
||||
</appender>
|
||||
|
||||
<root level="debug">
|
||||
<appender-ref ref="console"/>
|
||||
</root>
|
||||
|
||||
</configuration>
|
||||
|
|
@ -12,6 +12,7 @@ minimumBranchCoverage = 0.3
|
|||
minimumInstructionCoverage = 0.5
|
||||
whitelistedInstructionClasses += whitelistedBranchClasses += [
|
||||
"com.datadoghq.trace.integration.*",
|
||||
'com.datadoghq.trace.writer.ListWriter',
|
||||
'com.datadoghq.trace.DDTags'
|
||||
]
|
||||
|
||||
|
|
|
|||
|
|
@ -79,7 +79,7 @@ If child spans are unfinished when the parent attempts to finish, the span will
|
|||
|
||||
|
||||
Now, you are able to create, start and stop very simple spans.
|
||||
You can manipulate them at any time and add contextual information using tags.
|
||||
You can manipulate them at any time and add extra contextual information using the tags.
|
||||
|
||||
### Tags
|
||||
|
||||
|
|
|
|||
|
|
@ -1,72 +0,0 @@
|
|||
package com.datadoghq.trace;
|
||||
|
||||
import com.fasterxml.jackson.annotation.JsonIgnore;
|
||||
import io.opentracing.ActiveSpan;
|
||||
|
||||
/** Base implementation for opentracing {@link ActiveSpan} */
|
||||
public class DDActiveSpan extends DDBaseSpan<ActiveSpan> implements ActiveSpan {
|
||||
|
||||
protected final DDActiveSpan parent;
|
||||
protected boolean deactivated = false;
|
||||
|
||||
protected DDActiveSpan(DDActiveSpan parent, DDSpan span) {
|
||||
super(span.startTimeMicro, span.context());
|
||||
this.startTimeNano = span.startTimeNano;
|
||||
this.durationNano = span.durationNano;
|
||||
this.parent = parent;
|
||||
}
|
||||
|
||||
protected DDActiveSpan(DDActiveSpan parent, long timestampMicro, DDSpanContext context) {
|
||||
super(timestampMicro, context);
|
||||
this.parent = parent;
|
||||
}
|
||||
|
||||
/** @return the generating parent if not null */
|
||||
@JsonIgnore
|
||||
public DDActiveSpan getParent() {
|
||||
return parent;
|
||||
}
|
||||
|
||||
/** @return true if the span has already been deactivated */
|
||||
@JsonIgnore
|
||||
public boolean isDeactivated() {
|
||||
return deactivated;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void deactivate() {
|
||||
DDTracer tracer = context().getTracer();
|
||||
if (tracer != null) {
|
||||
tracer.deactivate(this);
|
||||
}
|
||||
finish();
|
||||
deactivated = true;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Continuation capture() {
|
||||
return new DDContinuation();
|
||||
}
|
||||
|
||||
@Override
|
||||
protected ActiveSpan thisInstance() {
|
||||
return this;
|
||||
}
|
||||
|
||||
public class DDContinuation implements Continuation {
|
||||
|
||||
@Override
|
||||
public ActiveSpan activate() {
|
||||
//Reactivate the current span
|
||||
context().getTracer().makeActive(DDActiveSpan.this);
|
||||
|
||||
//And return the encapsulating ActiveSpan
|
||||
return DDActiveSpan.this;
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
deactivate();
|
||||
}
|
||||
}
|
||||
|
|
@ -304,6 +304,10 @@ public abstract class DDBaseSpan<S extends BaseSpan> implements BaseSpan<S> {
|
|||
|
||||
@Override
|
||||
public String toString() {
|
||||
return context.toString();
|
||||
return new StringBuilder()
|
||||
.append(context.toString())
|
||||
.append(", duration_ns=")
|
||||
.append(durationNano)
|
||||
.toString();
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
}
|
||||
}
|
||||
|
|
@ -200,22 +206,23 @@ public class DDSpanContext implements io.opentracing.SpanContext {
|
|||
|
||||
@Override
|
||||
public String toString() {
|
||||
return "Span [ "
|
||||
+ traceId
|
||||
+ " ] [ "
|
||||
+ spanId
|
||||
+ " | "
|
||||
+ parentId
|
||||
+ " ] [ "
|
||||
+ getServiceName()
|
||||
+ " | "
|
||||
+ getOperationName()
|
||||
+ " | "
|
||||
+ getResourceName()
|
||||
+ " ]";
|
||||
return new StringBuilder()
|
||||
.append("Span [ t_id=")
|
||||
.append(traceId)
|
||||
.append(", s_id=")
|
||||
.append(spanId)
|
||||
.append(", p_id=")
|
||||
.append(parentId)
|
||||
.append("] trace=")
|
||||
.append(getServiceName())
|
||||
.append("/")
|
||||
.append(getOperationName())
|
||||
.append("/")
|
||||
.append(getResourceName())
|
||||
.toString();
|
||||
}
|
||||
|
||||
public void setOperationName(String operationName) {
|
||||
public void setOperationName(final String operationName) {
|
||||
this.operationName = operationName;
|
||||
}
|
||||
|
||||
|
|
@ -223,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;
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -7,16 +7,18 @@ import com.datadoghq.trace.sampling.AllSampler;
|
|||
import com.datadoghq.trace.sampling.Sampler;
|
||||
import com.datadoghq.trace.writer.LoggingWriter;
|
||||
import com.datadoghq.trace.writer.Writer;
|
||||
import io.opentracing.ActiveSpan;
|
||||
import io.opentracing.ActiveSpanSource;
|
||||
import io.opentracing.BaseSpan;
|
||||
import io.opentracing.Span;
|
||||
import io.opentracing.SpanContext;
|
||||
import io.opentracing.propagation.Format;
|
||||
import io.opentracing.util.ThreadLocalActiveSpanSource;
|
||||
import java.util.*;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
/** DDTracer makes it easy to send traces and span to DD using the OpenTracing integration. */
|
||||
public class DDTracer implements io.opentracing.Tracer {
|
||||
public class DDTracer extends ThreadLocalActiveSpanSource implements io.opentracing.Tracer {
|
||||
|
||||
public static final String JAVA_VERSION = System.getProperty("java.version", "unknown");
|
||||
public static final String CURRENT_VERSION;
|
||||
|
|
@ -93,7 +95,7 @@ public class DDTracer implements io.opentracing.Tracer {
|
|||
}
|
||||
|
||||
public DDSpanBuilder buildSpan(String operationName) {
|
||||
return new DDSpanBuilder(operationName);
|
||||
return new DDSpanBuilder(operationName, this);
|
||||
}
|
||||
|
||||
public <T> void inject(SpanContext spanContext, Format<T> format, T carrier) {
|
||||
|
|
@ -136,53 +138,9 @@ public class DDTracer implements io.opentracing.Tracer {
|
|||
writer.close();
|
||||
}
|
||||
|
||||
private final ThreadLocal<DDActiveSpan> currentActiveSpan = new ThreadLocal<DDActiveSpan>();
|
||||
|
||||
@Override
|
||||
public DDActiveSpan activeSpan() {
|
||||
return currentActiveSpan.get();
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the newly created active span as the active one from the Tracer's perspective
|
||||
*
|
||||
* @param activeSpan
|
||||
*/
|
||||
protected void makeActive(DDActiveSpan activeSpan) {
|
||||
//We cannot make active a preably deactivated span
|
||||
if (activeSpan != null && activeSpan.isDeactivated()) currentActiveSpan.set(null);
|
||||
else currentActiveSpan.set(activeSpan);
|
||||
}
|
||||
|
||||
/**
|
||||
* Deactivate the current span (if active) and make the parent active (again)
|
||||
*
|
||||
* @param activeSpan
|
||||
*/
|
||||
protected void deactivate(DDActiveSpan activeSpan) {
|
||||
DDActiveSpan current = activeSpan();
|
||||
if (current == activeSpan) {
|
||||
//The parent becomes the active span
|
||||
makeActive(activeSpan.getParent());
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public DDActiveSpan makeActive(Span span) {
|
||||
if (!(span instanceof DDSpan))
|
||||
throw new IllegalArgumentException(
|
||||
"Cannot transform a non DDSpan into a DDActiveSpan. Provided class: " + span.getClass());
|
||||
|
||||
//Wrap the provided manual span into an active one with the current parent
|
||||
DDActiveSpan activeSpan = new DDActiveSpan(activeSpan(), (DDSpan) span);
|
||||
|
||||
makeActive(activeSpan);
|
||||
|
||||
return activeSpan;
|
||||
}
|
||||
|
||||
/** Spans are built using this builder */
|
||||
public class DDSpanBuilder implements SpanBuilder {
|
||||
private final ActiveSpanSource spanSource;
|
||||
|
||||
/** Each span must have an operationName according to the opentracing specification */
|
||||
private String operationName;
|
||||
|
|
@ -203,32 +161,20 @@ public class DDTracer implements io.opentracing.Tracer {
|
|||
return this;
|
||||
}
|
||||
|
||||
private DDSpan startSpan() {
|
||||
return new DDSpan(this.timestamp, buildSpanContext());
|
||||
}
|
||||
|
||||
@Override
|
||||
public DDActiveSpan startActive() {
|
||||
//Set the active span as parent if ignoreActiveSpan==true
|
||||
DDActiveSpan activeParent = null;
|
||||
if (!ignoreActiveSpan) {
|
||||
DDActiveSpan current = activeSpan();
|
||||
if (current != null) {
|
||||
activeParent = current;
|
||||
|
||||
//Ensure parent inheritance
|
||||
asChildOf(activeParent);
|
||||
}
|
||||
}
|
||||
|
||||
//Create the active span
|
||||
DDActiveSpan activeSpan = new DDActiveSpan(activeParent, this.timestamp, buildSpanContext());
|
||||
public ActiveSpan startActive() {
|
||||
ActiveSpan activeSpan = spanSource.makeActive(startSpan());
|
||||
logger.debug("{} - Starting a new active span.", activeSpan);
|
||||
|
||||
makeActive(activeSpan);
|
||||
|
||||
return activeSpan;
|
||||
}
|
||||
|
||||
@Override
|
||||
public DDSpan startManual() {
|
||||
DDSpan span = new DDSpan(this.timestamp, buildSpanContext());
|
||||
DDSpan span = startSpan();
|
||||
logger.debug("{} - Starting a new manuel span.", span);
|
||||
return span;
|
||||
}
|
||||
|
|
@ -240,12 +186,12 @@ public class DDTracer implements io.opentracing.Tracer {
|
|||
}
|
||||
|
||||
@Override
|
||||
public DDTracer.DDSpanBuilder withTag(String tag, Number number) {
|
||||
public DDSpanBuilder withTag(String tag, Number number) {
|
||||
return withTag(tag, (Object) number);
|
||||
}
|
||||
|
||||
@Override
|
||||
public DDTracer.DDSpanBuilder withTag(String tag, String string) {
|
||||
public DDSpanBuilder withTag(String tag, String string) {
|
||||
if (tag.equals(DDTags.SERVICE_NAME)) {
|
||||
return withServiceName(string);
|
||||
} else if (tag.equals(DDTags.RESOURCE_NAME)) {
|
||||
|
|
@ -258,36 +204,37 @@ public class DDTracer implements io.opentracing.Tracer {
|
|||
}
|
||||
|
||||
@Override
|
||||
public DDTracer.DDSpanBuilder withTag(String tag, boolean bool) {
|
||||
public DDSpanBuilder withTag(String tag, boolean bool) {
|
||||
return withTag(tag, (Object) bool);
|
||||
}
|
||||
|
||||
public DDSpanBuilder(String operationName) {
|
||||
public DDSpanBuilder(String operationName, ActiveSpanSource spanSource) {
|
||||
this.operationName = operationName;
|
||||
this.spanSource = spanSource;
|
||||
}
|
||||
|
||||
@Override
|
||||
public DDTracer.DDSpanBuilder withStartTimestamp(long timestampMillis) {
|
||||
public DDSpanBuilder withStartTimestamp(long timestampMillis) {
|
||||
this.timestamp = timestampMillis;
|
||||
return this;
|
||||
}
|
||||
|
||||
public DDTracer.DDSpanBuilder withServiceName(String serviceName) {
|
||||
public DDSpanBuilder withServiceName(String serviceName) {
|
||||
this.serviceName = serviceName;
|
||||
return this;
|
||||
}
|
||||
|
||||
public DDTracer.DDSpanBuilder withResourceName(String resourceName) {
|
||||
public DDSpanBuilder withResourceName(String resourceName) {
|
||||
this.resourceName = resourceName;
|
||||
return this;
|
||||
}
|
||||
|
||||
public DDTracer.DDSpanBuilder withErrorFlag() {
|
||||
public DDSpanBuilder withErrorFlag() {
|
||||
this.errorFlag = true;
|
||||
return this;
|
||||
}
|
||||
|
||||
public DDTracer.DDSpanBuilder withSpanType(String spanType) {
|
||||
public DDSpanBuilder withSpanType(String spanType) {
|
||||
this.spanType = spanType;
|
||||
return this;
|
||||
}
|
||||
|
|
@ -300,24 +247,24 @@ public class DDTracer implements io.opentracing.Tracer {
|
|||
}
|
||||
|
||||
@Override
|
||||
public DDTracer.DDSpanBuilder asChildOf(BaseSpan<?> span) {
|
||||
public DDSpanBuilder asChildOf(BaseSpan<?> span) {
|
||||
return asChildOf(span == null ? null : span.context());
|
||||
}
|
||||
|
||||
@Override
|
||||
public DDTracer.DDSpanBuilder asChildOf(SpanContext spanContext) {
|
||||
public DDSpanBuilder asChildOf(SpanContext spanContext) {
|
||||
this.parent = spanContext;
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public DDTracer.DDSpanBuilder addReference(String referenceType, SpanContext spanContext) {
|
||||
public DDSpanBuilder addReference(String referenceType, SpanContext spanContext) {
|
||||
logger.debug("`addReference` method is not implemented. Doing nothing");
|
||||
return this;
|
||||
}
|
||||
|
||||
// Private methods
|
||||
private DDTracer.DDSpanBuilder withTag(String tag, Object value) {
|
||||
private DDSpanBuilder withTag(String tag, Object value) {
|
||||
if (this.tags.isEmpty()) {
|
||||
this.tags = new HashMap<String, Object>();
|
||||
}
|
||||
|
|
@ -336,22 +283,38 @@ public class DDTracer implements io.opentracing.Tracer {
|
|||
* @return the context
|
||||
*/
|
||||
private DDSpanContext buildSpanContext() {
|
||||
long generatedId = generateNewId();
|
||||
final long traceId;
|
||||
final long spanId = generateNewId();
|
||||
final long parentSpanId;
|
||||
final Map<String, String> baggage;
|
||||
final List<DDBaseSpan<?>> parentTrace;
|
||||
|
||||
DDSpanContext context;
|
||||
DDSpanContext p = this.parent != null ? (DDSpanContext) this.parent : null;
|
||||
|
||||
String spanType = this.spanType;
|
||||
if (spanType == null && this.parent != null) {
|
||||
spanType = p.getSpanType();
|
||||
SpanContext parentContext = this.parent;
|
||||
if (parentContext == null && !ignoreActiveSpan) {
|
||||
// use the ActiveSpan as parent unless overridden or ignored.
|
||||
ActiveSpan activeSpan = activeSpan();
|
||||
if (activeSpan != null) parentContext = activeSpan.context();
|
||||
}
|
||||
|
||||
String serviceName = this.serviceName;
|
||||
if (serviceName == null) {
|
||||
if (p != null && p.getServiceName() != null) {
|
||||
serviceName = p.getServiceName();
|
||||
if (parentContext instanceof DDSpanContext) {
|
||||
DDSpanContext ddsc = (DDSpanContext) parentContext;
|
||||
traceId = ddsc.getTraceId();
|
||||
parentSpanId = ddsc.getSpanId();
|
||||
baggage = ddsc.getBaggageItems();
|
||||
parentTrace = ddsc.getTrace();
|
||||
|
||||
if (this.serviceName == null) this.serviceName = ddsc.getServiceName();
|
||||
if (this.spanType == null) this.spanType = ddsc.getSpanType();
|
||||
} else {
|
||||
serviceName = defaultServiceName;
|
||||
traceId = spanId;
|
||||
parentSpanId = 0L;
|
||||
baggage = null;
|
||||
parentTrace = null;
|
||||
}
|
||||
|
||||
if (serviceName == null) {
|
||||
serviceName = defaultServiceName;
|
||||
}
|
||||
|
||||
String operationName = this.operationName != null ? this.operationName : this.resourceName;
|
||||
|
|
@ -361,17 +324,17 @@ public class DDTracer implements io.opentracing.Tracer {
|
|||
// some attributes are inherited from the parent
|
||||
context =
|
||||
new DDSpanContext(
|
||||
this.parent == null ? generatedId : p.getTraceId(),
|
||||
generatedId,
|
||||
this.parent == null ? 0L : p.getSpanId(),
|
||||
traceId,
|
||||
spanId,
|
||||
parentSpanId,
|
||||
serviceName,
|
||||
operationName,
|
||||
this.resourceName,
|
||||
this.parent == null ? null : p.getBaggageItems(),
|
||||
baggage,
|
||||
errorFlag,
|
||||
spanType,
|
||||
this.tags,
|
||||
this.parent == null ? null : p.getTrace(),
|
||||
parentTrace,
|
||||
DDTracer.this);
|
||||
|
||||
// Force the lang meta
|
||||
|
|
|
|||
|
|
@ -1,34 +1,32 @@
|
|||
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 implements Writer {
|
||||
|
||||
protected List<List<DDBaseSpan<?>>> list = new ArrayList<List<DDBaseSpan<?>>>();
|
||||
public class ListWriter extends CopyOnWriteArrayList<List<DDBaseSpan<?>>> implements Writer {
|
||||
|
||||
public List<List<DDBaseSpan<?>>> getList() {
|
||||
return list;
|
||||
return this;
|
||||
}
|
||||
|
||||
public List<DDBaseSpan<?>> firstTrace() {
|
||||
return list.get(0);
|
||||
return get(0);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void write(List<DDBaseSpan<?>> trace) {
|
||||
list.add(trace);
|
||||
public void write(final List<DDBaseSpan<?>> trace) {
|
||||
add(trace);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void start() {
|
||||
list.clear();
|
||||
clear();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
list.clear();
|
||||
clear();
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,192 @@
|
|||
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:
|
||||
def phaser = new Phaser()
|
||||
phaser.register()
|
||||
|
||||
for (int i = 0; i < count; i++) {
|
||||
phaser.register()
|
||||
def capture = activeSpan.capture()
|
||||
new Thread({
|
||||
phaser.arriveAndAwaitAdvance()
|
||||
def activeSpan = capture.activate()
|
||||
phaser.arriveAndAwaitAdvance()
|
||||
activeSpan.deactivate()
|
||||
phaser.arriveAndDeregister()
|
||||
}).start()
|
||||
}
|
||||
|
||||
activeSpan.deactivate() // allow the trace to be reported when all continuations deactivate
|
||||
|
||||
when:
|
||||
phaser.arriveAndAwaitAdvance() //allow threads to activate capture
|
||||
|
||||
then:
|
||||
traceCollector == []
|
||||
|
||||
when:
|
||||
phaser.arriveAndAwaitAdvance() //allow threads to deactivate their span
|
||||
phaser.arriveAndAwaitAdvance() // wait till all threads have deactivated
|
||||
|
||||
then:
|
||||
traceCollector.size() == 1
|
||||
traceCollector.firstTrace().size() == 1
|
||||
|
||||
where:
|
||||
count = new Random().nextInt(50) + 5
|
||||
}
|
||||
|
||||
def "concurrent threads with manual spans and continuations report correctly"() {
|
||||
setup:
|
||||
def phaser = new Phaser()
|
||||
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(spanName).startManual()
|
||||
phaser.arriveAndAwaitAdvance()
|
||||
childSpan.finish()
|
||||
activeSpan.deactivate()
|
||||
phaser.arriveAndDeregister()
|
||||
}).start()
|
||||
}
|
||||
|
||||
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:
|
||||
phaser.arriveAndAwaitAdvance() //allow threads to deactivate their span
|
||||
phaser.arriveAndAwaitAdvance() // wait till all threads have deactivated
|
||||
|
||||
traceCollector.size()
|
||||
|
||||
then:
|
||||
continuationCount.get() == 0
|
||||
traceCollector.size() == 1
|
||||
def trace = traceCollector.remove(0)
|
||||
def parent = trace.remove(0)
|
||||
|
||||
trace.size() == count
|
||||
parent.context.parentId == 0
|
||||
|
||||
trace.every {
|
||||
it.context.parentId == parent.context.spanId
|
||||
}
|
||||
|
||||
where:
|
||||
count = new Random().nextInt(50) + 5
|
||||
}
|
||||
|
||||
def "concurrent threads with active spans and continuations report correctly"() {
|
||||
setup:
|
||||
def phaser = new Phaser()
|
||||
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(spanName).startActive()
|
||||
phaser.arriveAndAwaitAdvance()
|
||||
childSpan.deactivate()
|
||||
activeSpan.deactivate()
|
||||
phaser.arriveAndDeregister()
|
||||
}).start()
|
||||
}
|
||||
|
||||
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:
|
||||
phaser.arriveAndAwaitAdvance() //allow threads to deactivate their span
|
||||
phaser.arriveAndAwaitAdvance() // wait till all threads have deactivated
|
||||
|
||||
traceCollector.size()
|
||||
|
||||
then:
|
||||
continuationCount.get() == 0
|
||||
traceCollector.size() == 1
|
||||
def trace = traceCollector.remove(0)
|
||||
def parent = trace.remove(0)
|
||||
|
||||
trace.size() == count
|
||||
parent.context.parentId == 0
|
||||
|
||||
trace.every {
|
||||
it.context.parentId == parent.context.spanId
|
||||
}
|
||||
|
||||
where:
|
||||
count = new Random().nextInt(50) + 5
|
||||
}
|
||||
|
||||
@Ignore("Not yet implemented in ThreadLocalActiveSpan.Continuation")
|
||||
def "calling activate more than once results in an error"() {
|
||||
setup:
|
||||
def capture = activeSpan.capture()
|
||||
|
||||
when:
|
||||
activeSpan.deactivate()
|
||||
|
||||
then:
|
||||
traceCollector == []
|
||||
|
||||
when:
|
||||
capture.activate().deactivate()
|
||||
// parent span should be finished at this point.
|
||||
then:
|
||||
traceCollector == []
|
||||
|
||||
when:
|
||||
capture.activate().deactivate()
|
||||
|
||||
then:
|
||||
thrown(RuntimeException)
|
||||
}
|
||||
}
|
||||
|
|
@ -1,130 +0,0 @@
|
|||
package com.datadoghq.trace;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
import com.datadoghq.trace.writer.ListWriter;
|
||||
import io.opentracing.ActiveSpan.Continuation;
|
||||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
|
||||
public class DDActiveSpanTest {
|
||||
|
||||
private ListWriter listWriter = new ListWriter();
|
||||
private DDTracer ddTracer = new DDTracer(listWriter);
|
||||
|
||||
@Before
|
||||
public void setUp() {
|
||||
listWriter.start();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testThreadContextPropagation() {
|
||||
|
||||
DDActiveSpan span1 = ddTracer.buildSpan("op1").startActive();
|
||||
|
||||
assertThat(span1.getOperationName()).isEqualTo("op1");
|
||||
assertThat(ddTracer.activeSpan()).isEqualTo(span1);
|
||||
|
||||
DDActiveSpan span2 = ddTracer.buildSpan("op2").startActive();
|
||||
assertThat(span2.getOperationName()).isEqualTo("op2");
|
||||
assertThat(span2.getParent()).isEqualTo(span1);
|
||||
assertThat(span2.context().getParentId()).isEqualTo(span1.getSpanId());
|
||||
assertThat(span2.context().getTraceId()).isEqualTo(span1.getTraceId());
|
||||
assertThat(ddTracer.activeSpan()).isEqualTo(span2);
|
||||
|
||||
span2.deactivate();
|
||||
assertThat(ddTracer.activeSpan()).isEqualTo(span1);
|
||||
|
||||
assertThat(listWriter.getList().size()).isEqualTo(0);
|
||||
span1.deactivate();
|
||||
assertThat(ddTracer.activeSpan()).isNull();
|
||||
|
||||
assertThat(listWriter.getList().size()).isEqualTo(1);
|
||||
assertThat(listWriter.getList().get(0).size()).isEqualTo(2);
|
||||
assertThat(listWriter.getList().get(0).get(0)).isEqualTo(span1);
|
||||
assertThat(listWriter.getList().get(0).get(1)).isEqualTo(span2);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testParentDeactivationFirst() {
|
||||
|
||||
DDActiveSpan span1 = ddTracer.buildSpan("op1").startActive();
|
||||
|
||||
assertThat(span1.getOperationName()).isEqualTo("op1");
|
||||
|
||||
DDActiveSpan span2 = ddTracer.buildSpan("op2").startActive();
|
||||
|
||||
assertThat(ddTracer.activeSpan()).isEqualTo(span2);
|
||||
|
||||
span1.deactivate();
|
||||
|
||||
//If parent span deactivated first => Span 2 remains the active span
|
||||
assertThat(ddTracer.activeSpan()).isEqualTo(span2);
|
||||
|
||||
span2.deactivate();
|
||||
|
||||
//If span2 comes to be deactivated
|
||||
assertThat(ddTracer.activeSpan()).isNull();
|
||||
|
||||
assertThat(listWriter.getList().size()).isEqualTo(1);
|
||||
assertThat(listWriter.getList().get(0).size()).isEqualTo(2);
|
||||
assertThat(listWriter.getList().get(0).get(0)).isEqualTo(span1);
|
||||
assertThat(listWriter.getList().get(0).get(1)).isEqualTo(span2);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testContinuation() throws Exception {
|
||||
final DDActiveSpan span1 = ddTracer.buildSpan("op1").startActive();
|
||||
|
||||
final Continuation continuation = span1.capture();
|
||||
|
||||
Thread t =
|
||||
new Thread(
|
||||
new Runnable() {
|
||||
@Override
|
||||
public void run() {
|
||||
assertThat(ddTracer.activeSpan()).isNull();
|
||||
continuation.activate();
|
||||
assertThat(ddTracer.activeSpan()).isEqualTo(span1);
|
||||
DDActiveSpan span2 = ddTracer.buildSpan("op2").startActive();
|
||||
assertThat(ddTracer.activeSpan()).isEqualTo(span2);
|
||||
span2.deactivate();
|
||||
assertThat(ddTracer.activeSpan()).isEqualTo(span1);
|
||||
}
|
||||
});
|
||||
|
||||
t.start();
|
||||
t.join();
|
||||
|
||||
span1.deactivate();
|
||||
assertThat(ddTracer.activeSpan()).isNull();
|
||||
|
||||
assertThat(listWriter.getList().size()).isEqualTo(1);
|
||||
assertThat(listWriter.getList().get(0).size()).isEqualTo(2);
|
||||
assertThat(listWriter.getList().get(0).get(0)).isEqualTo(span1);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testMakeSpanActive() {
|
||||
final DDSpan manualSpan = ddTracer.buildSpan("op1").startManual();
|
||||
|
||||
DDActiveSpan span1 = ddTracer.makeActive(manualSpan);
|
||||
|
||||
assertThat(span1.getOperationName()).isEqualTo("op1");
|
||||
assertThat(ddTracer.activeSpan()).isEqualTo(span1);
|
||||
assertThat(manualSpan.context()).isEqualTo(span1.context());
|
||||
assertThat(manualSpan.startTimeNano).isEqualTo(span1.startTimeNano);
|
||||
assertThat(manualSpan.startTimeMicro).isEqualTo(span1.startTimeMicro);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testCrossContextPropagation() {
|
||||
final DDSpan manualSpan = ddTracer.buildSpan("op1").startManual();
|
||||
|
||||
DDActiveSpan activeSpan =
|
||||
ddTracer.buildSpan("op2").asChildOf(manualSpan.context()).startActive();
|
||||
|
||||
assertThat(activeSpan.getParentId()).isEqualTo(manualSpan.getSpanId());
|
||||
assertThat(activeSpan.getTraceId()).isEqualTo(manualSpan.getTraceId());
|
||||
}
|
||||
}
|
||||
|
|
@ -13,7 +13,6 @@ public class DDSpanSerializationTest {
|
|||
|
||||
ObjectMapper serializer;
|
||||
DDSpan span;
|
||||
DDActiveSpan activeSpan;
|
||||
Map<String, Object> expected = Maps.newHashMap();
|
||||
|
||||
@Before
|
||||
|
|
@ -56,9 +55,6 @@ public class DDSpanSerializationTest {
|
|||
|
||||
span = new DDSpan(100L, context);
|
||||
span.finish(133L);
|
||||
|
||||
activeSpan = new DDActiveSpan(null, 100L, context);
|
||||
activeSpan.deactivate();
|
||||
serializer = new ObjectMapper();
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue