Fix priority sampling implementation bugs
use correct sampling flag name
set sampling rate
set sampling flags on root span metrics only
tests to assert correct behavior
MODIFY
dd-trace-ot/src/main/java/datadog/opentracing/DDSpan.java
dd-trace-ot/src/main/java/datadog/opentracing/DDSpan.java
@@ -4,8 +4,6 @@ import static io.opentracing.log.Fields.ERROR_OBJECT;
import com.fasterxml.jackson.annotation.JsonGetter;
import com.fasterxml.jackson.annotation.JsonIgnore;
-import com.fasterxml.jackson.annotation.JsonInclude;
-import com.fasterxml.jackson.annotation.JsonInclude.Include;
import datadog.trace.api.DDTags;
import datadog.trace.api.interceptor.MutableSpan;
import datadog.trace.api.sampling.PrioritySampling;
@@ -255,7 +253,7 @@ public class DDSpan implements Span, MutableSpan {
}
/**
- * Set the sampling priority of the span.
+ * Set the sampling priority of the root span of this span's trace
*
* <p>Has no effect if the span priority has been propagated (injected or extracted).
*/
@@ -290,6 +288,16 @@ public class DDSpan implements Span, MutableSpan {
return meta;
}
+ /**
+ * Span metrics.
+ *
+ * @return metrics for this span
+ */
+ @JsonGetter
+ public Map<String, Number> getMetrics() {
+ return context.getMetrics();
+ }
+
@Override
@JsonGetter("start")
public long getStartTime() {
@@ -336,8 +344,7 @@ public class DDSpan implements Span, MutableSpan {
}
@Override
- @JsonGetter("sampling_priority")
- @JsonInclude(Include.NON_NULL)
+ @JsonIgnore
public Integer getSamplingPriority() {
final int samplingPriority = context.getSamplingPriority();
if (samplingPriority == PrioritySampling.UNSET) {

MODIFY
dd-trace-ot/src/main/java/datadog/opentracing/DDSpanContext.java
dd-trace-ot/src/main/java/datadog/opentracing/DDSpanContext.java
@@ -4,11 +4,9 @@ import com.fasterxml.jackson.annotation.JsonIgnore;
import datadog.opentracing.decorators.AbstractDecorator;
import datadog.trace.api.DDTags;
import datadog.trace.api.sampling.PrioritySampling;
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
-import java.util.TreeMap;
+import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicReference;
import lombok.extern.slf4j.Slf4j;
/**
@@ -21,6 +19,10 @@ import lombok.extern.slf4j.Slf4j;
*/
@Slf4j
public class DDSpanContext implements io.opentracing.SpanContext {
+ public static final String PRIORITY_SAMPLING_KEY = "_sampling_priority_v1";
+ public static final String SAMPLE_RATE_KEY = "_sample_rate";
+
+ private static final Map<String, Number> EMPTY_METRICS = Collections.emptyMap();
// Shared with other span contexts
/** For technical reasons, the ref to the original tracer */
@@ -50,10 +52,15 @@ public class DDSpanContext implements io.opentracing.SpanContext {
private volatile String spanType;
/** True indicates that the span reports an error */
private volatile boolean errorFlag;
- /** The sampling priority of the trace */
- private volatile int samplingPriority = PrioritySampling.UNSET;
- /** When true, the samplingPriority cannot be changed. */
- private volatile boolean samplingPriorityLocked = false;
+ /**
+ * When true, the samplingPriority cannot be changed. This prevents the sampling flag from
+ * changing after the context has propagated.
+ *
+ * <p>For thread safety, this boolean is only modified or accessed under instance lock.
+ */
+ private boolean samplingPriorityLocked = false;
+ /** Metrics on the span */
+ private final AtomicReference<Map<String, Number>> metrics = new AtomicReference<>();
// Additional Metadata
private final String threadName = Thread.currentThread().getName();
@@ -96,9 +103,12 @@ public class DDSpanContext implements io.opentracing.SpanContext {
this.serviceName = serviceName;
this.operationName = operationName;
this.resourceName = resourceName;
- this.samplingPriority = samplingPriority;
this.errorFlag = errorFlag;
this.spanType = spanType;
+
+ if (samplingPriority != PrioritySampling.UNSET) {
+ setSamplingPriority(samplingPriority);
+ }
}
public long getTraceId() {
@@ -156,19 +166,41 @@ public class DDSpanContext implements io.opentracing.SpanContext {
}
public void setSamplingPriority(final int newPriority) {
+ if (trace != null) {
+ DDSpan rootSpan = trace.getRootSpan();
+ if (null != rootSpan && rootSpan.context() != this) {
+ rootSpan.context().setSamplingPriority(newPriority);
+ return;
+ }
+ }
+ if (newPriority == PrioritySampling.UNSET) {
+ log.debug("{}: Refusing to set samplingPriority to UNSET", this);
+ return;
+ }
+ // sync with lockSamplingPriority
+ synchronized (this) {
if (samplingPriorityLocked) {
log.debug(
- "samplingPriority locked at {}. Refusing to set to {}", samplingPriority, newPriority);
+ "samplingPriority locked at {}. Refusing to set to {}",
+ getMetrics().get(PRIORITY_SAMPLING_KEY),
+ newPriority);
} else {
- synchronized (this) {
- // sync with lockSamplingPriority
- this.samplingPriority = newPriority;
+ setMetric(PRIORITY_SAMPLING_KEY, newPriority);
+ log.debug("Set sampling priority to {}", getMetrics().get(PRIORITY_SAMPLING_KEY));
}
}
}
+ /** @return the sampling priority of this span's trace, or null if no priority has been set */
public int getSamplingPriority() {
- return samplingPriority;
+ if (trace != null) {
+ DDSpan rootSpan = trace.getRootSpan();
+ if (null != rootSpan && rootSpan.context() != this) {
+ return rootSpan.context().getSamplingPriority();
+ }
+ }
+ final Number val = getMetrics().get(PRIORITY_SAMPLING_KEY);
+ return null == val ? PrioritySampling.UNSET : val.intValue();
}
/**
@@ -181,19 +213,24 @@ public class DDSpanContext implements io.opentracing.SpanContext {
* @return true if the sampling priority was locked.
*/
public boolean lockSamplingPriority() {
- if (!samplingPriorityLocked) {
- synchronized (this) {
- // sync with setSamplingPriority
- if (samplingPriority == PrioritySampling.UNSET) {
- log.debug("{} : refusing to lock unset samplingPriority", this);
- } else {
- this.samplingPriorityLocked = true;
- log.debug("{} : locked samplingPriority to {}", this, this.samplingPriority);
+ if (trace != null) {
+ DDSpan rootSpan = trace.getRootSpan();
+ if (null != rootSpan && rootSpan.context() != this) {
+ return rootSpan.context().lockSamplingPriority();
}
}
+ // sync with setSamplingPriority
+ synchronized (this) {
+ if (getMetrics().get(PRIORITY_SAMPLING_KEY) == null) {
+ log.debug("{} : refusing to lock unset samplingPriority", this);
+ } else if (samplingPriorityLocked == false) {
+ samplingPriorityLocked = true;
+ log.debug(
+ "{} : locked samplingPriority to {}", this, getMetrics().get(PRIORITY_SAMPLING_KEY));
}
return samplingPriorityLocked;
}
+ }
public void setBaggageItem(final String key, final String value) {
this.baggageItems.put(key, value);
@@ -225,6 +262,17 @@ public class DDSpanContext implements io.opentracing.SpanContext {
return this.tracer;
}
+ public Map<String, Number> getMetrics() {
+ final Map<String, Number> metrics = this.metrics.get();
+ return metrics == null ? EMPTY_METRICS : metrics;
+ }
+
+ public void setMetric(String key, Number value) {
+ if (metrics.get() == null) {
+ metrics.compareAndSet(null, new ConcurrentHashMap<String, Number>());
+ }
+ metrics.get().put(key, value);
+ }
/**
* Add a tag to the span. Tags are not propagated to the children
*
@@ -284,10 +332,9 @@ public class DDSpanContext implements io.opentracing.SpanContext {
.append("/")
.append(getOperationName())
.append("/")
- .append(getResourceName());
- if (getSamplingPriority() != PrioritySampling.UNSET) {
- s.append(" samplingPriority=").append(getSamplingPriority());
- }
+ .append(getResourceName())
+ .append(" metrics=")
+ .append(new TreeMap(getMetrics()));
if (errorFlag) {
s.append(" *errored*");
}

MODIFY
dd-trace-ot/src/main/java/datadog/opentracing/DDTracer.java
dd-trace-ot/src/main/java/datadog/opentracing/DDTracer.java
@@ -490,7 +490,7 @@ public class DDTracer implements io.opentracing.Tracer {
parentSpanId = ddsc.getSpanId();
baggage = ddsc.getBaggageItems();
parentTrace = ddsc.getTrace();
- samplingPriority = ddsc.getSamplingPriority();
+ samplingPriority = PrioritySampling.UNSET;
if (this.serviceName == null) this.serviceName = ddsc.getServiceName();
if (this.spanType == null) this.spanType = ddsc.getSpanType();

MODIFY
dd-trace-ot/src/main/java/datadog/opentracing/PendingTrace.java
dd-trace-ot/src/main/java/datadog/opentracing/PendingTrace.java
@@ -17,7 +17,6 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
-
import lombok.extern.slf4j.Slf4j;
@Slf4j

MODIFY
dd-trace-ot/src/main/java/datadog/trace/common/sampling/RateByServiceSampler.java
dd-trace-ot/src/main/java/datadog/trace/common/sampling/RateByServiceSampler.java
@@ -3,6 +3,7 @@ package datadog.trace.common.sampling;
import com.fasterxml.jackson.databind.JsonNode;
import datadog.opentracing.DDSpan;
+import datadog.opentracing.DDSpanContext;
import datadog.trace.api.sampling.PrioritySampling;
import datadog.trace.common.writer.DDApi.ResponseListener;
import java.util.HashMap;
@@ -47,17 +48,18 @@ public class RateByServiceSampler implements Sampler, ResponseListener {
final String serviceName = span.getServiceName();
final String env = getSpanEnv(span);
final String key = "service:" + serviceName + ",env:" + env;
- boolean agentSample;
+ final RateSampler sampler;
if (serviceRates.containsKey(key)) {
- agentSample = serviceRates.get(key).sample(span);
+ sampler = serviceRates.get(key);
} else {
- agentSample = baseSampler.sample(span);
+ sampler = baseSampler;
}
- if (agentSample) {
+ if (sampler.sample(span)) {
span.setSamplingPriority(PrioritySampling.SAMPLER_KEEP);
} else {
span.setSamplingPriority(PrioritySampling.SAMPLER_DROP);
}
+ span.context().setMetric(DDSpanContext.SAMPLE_RATE_KEY, sampler.getSampleRate());
}
private static String getSpanEnv(DDSpan span) {

MODIFY
dd-trace-ot/src/test/groovy/datadog/opentracing/DDSpanSerializationTest.groovy
dd-trace-ot/src/test/groovy/datadog/opentracing/DDSpanSerializationTest.groovy
@@ -9,7 +9,7 @@ import spock.lang.Specification
class DDSpanSerializationTest extends Specification {
- def "serialize spans"() throws Exception {
+ def "serialize spans with sampling #samplingPriority"() throws Exception {
setup:
final Map<String, String> baggage = new HashMap<>()
baggage.put("a-baggage", "value")
@@ -24,9 +24,12 @@ class DDSpanSerializationTest extends Specification {
expected.put("name", "operation")
expected.put("duration", 33000)
expected.put("resource", "operation")
+ final Map<String, Number> metrics = new HashMap<>()
if (samplingPriority != PrioritySampling.UNSET) {
- expected.put("sampling_priority", samplingPriority)
+ metrics.put("_sampling_priority_v1", Integer.valueOf(samplingPriority))
+ metrics.put("_sample_rate", Double.valueOf(1.0))
}
+ expected.put("metrics", metrics)
expected.put("start", 100000)
expected.put("span_id", 2l)
expected.put("parent_id", 0l)
@@ -55,6 +58,9 @@ class DDSpanSerializationTest extends Specification {
baggage.put(DDTags.SPAN_TYPE, context.getSpanType())
DDSpan span = new DDSpan(100L, context)
+ if (samplingPriority != PrioritySampling.UNSET) {
+ span.context().setMetric("_sample_rate", Double.valueOf(1.0))
+ }
span.finish(133L)
ObjectMapper serializer = new ObjectMapper()

MODIFY
dd-trace-ot/src/test/groovy/datadog/opentracing/DDSpanTest.groovy
dd-trace-ot/src/test/groovy/datadog/opentracing/DDSpanTest.groovy
package datadog.opentracing
import datadog.trace.api.sampling.PrioritySampling
+import datadog.trace.common.sampling.RateByServiceSampler
import datadog.trace.common.writer.ListWriter
import spock.lang.Specification
@@ -8,7 +9,7 @@ import java.util.concurrent.TimeUnit
class DDSpanTest extends Specification {
def writer = new ListWriter()
- def tracer = new DDTracer(writer)
+ def tracer = new DDTracer(DDTracer.UNASSIGNED_DEFAULT_SERVICE_NAME, writer, new RateByServiceSampler())
def "getters and setters"() {
setup:
@@ -157,4 +158,29 @@ class DDSpanTest extends Specification {
expect:
span.durationNano == 1
}
+
+ def "priority sampling metric set only on root span" () {
+ setup:
+ def parent = tracer.buildSpan("testParent").start()
+ def child1 = tracer.buildSpan("testChild1").asChildOf(parent).start()
+
+ child1.setSamplingPriority(PrioritySampling.SAMPLER_KEEP)
+ child1.context().lockSamplingPriority()
+ parent.setSamplingPriority(PrioritySampling.SAMPLER_DROP)
+ child1.finish()
+ def child2 = tracer.buildSpan("testChild2").asChildOf(parent).start()
+ child2.finish()
+ parent.finish()
+
+ expect:
+ parent.context().samplingPriority == PrioritySampling.SAMPLER_KEEP
+ parent.getMetrics().get(DDSpanContext.PRIORITY_SAMPLING_KEY) == PrioritySampling.SAMPLER_KEEP
+ parent.getMetrics().get(DDSpanContext.SAMPLE_RATE_KEY) == 1.0
+ child1.getSamplingPriority() == parent.getSamplingPriority()
+ child1.getMetrics().get(DDSpanContext.PRIORITY_SAMPLING_KEY) == null
+ child1.getMetrics().get(DDSpanContext.SAMPLE_RATE_KEY) == null
+ child2.getSamplingPriority() == parent.getSamplingPriority()
+ child2.getMetrics().get(DDSpanContext.PRIORITY_SAMPLING_KEY) == null
+ child2.getMetrics().get(DDSpanContext.SAMPLE_RATE_KEY) == null
+ }
}

MODIFY
dd-trace-ot/src/test/groovy/datadog/trace/DDSpanContextTest.groovy
dd-trace-ot/src/test/groovy/datadog/trace/DDSpanContextTest.groovy
@@ -18,7 +18,7 @@ class DDSpanContextTest extends Specification {
context.serviceName == "fakeService"
context.resourceName == "fakeResource"
context.spanType == "fakeType"
- context.toString() == "Span [ t_id=1, s_id=1, p_id=0] trace=fakeService/fakeOperation/fakeResource *errored* tags={${extra}span.type=${context.getSpanType()}, thread.id=${Thread.currentThread().id}, thread.name=${Thread.currentThread().name}}"
+ context.toString() == "Span [ t_id=1, s_id=1, p_id=0] trace=fakeService/fakeOperation/fakeResource metrics={} *errored* tags={${extra}span.type=${context.getSpanType()}, thread.id=${Thread.currentThread().id}, thread.name=${Thread.currentThread().name}}"
where:
name | extra | tags
@@ -35,7 +35,7 @@ class DDSpanContextTest extends Specification {
def thread = Thread.currentThread()
def expectedTags = [(DDTags.THREAD_NAME): thread.name, (DDTags.THREAD_ID): thread.id, (DDTags.SPAN_TYPE): context.getSpanType()]
- def expectedTrace = "Span [ t_id=1, s_id=1, p_id=0] trace=$details tags={span.type=${context.getSpanType()}, thread.id=$thread.id, thread.name=$thread.name}"
+ def expectedTrace = "Span [ t_id=1, s_id=1, p_id=0] trace=$details metrics={} tags={span.type=${context.getSpanType()}, thread.id=$thread.id, thread.name=$thread.name}"
expect:
context.getTags() == expectedTags
@@ -62,7 +62,7 @@ class DDSpanContextTest extends Specification {
(DDTags.THREAD_NAME): thread.name,
(DDTags.THREAD_ID) : thread.id
]
- context.toString() == "Span [ t_id=1, s_id=1, p_id=0] trace=fakeService/fakeOperation/fakeResource tags={span.type=${context.getSpanType()}, $name=$value, thread.id=$thread.id, thread.name=$thread.name}"
+ context.toString() == "Span [ t_id=1, s_id=1, p_id=0] trace=fakeService/fakeOperation/fakeResource metrics={} tags={span.type=${context.getSpanType()}, $name=$value, thread.id=$thread.id, thread.name=$thread.name}"
where:
name | value

MODIFY
dd-trace-ot/src/test/groovy/datadog/trace/api/sampling/RateByServiceSamplerTest.groovy
dd-trace-ot/src/test/groovy/datadog/trace/api/sampling/RateByServiceSamplerTest.groovy
@@ -65,5 +65,6 @@ class RateByServiceSamplerTest extends Specification {
expect:
// sets correctly on root span
span.getSamplingPriority() == PrioritySampling.SAMPLER_KEEP
+ span.getMetrics().get("_sample_rate") == 1.0
}
}

MODIFY
dd-trace-ot/src/test/groovy/datadog/trace/api/writer/DDApiTest.groovy
dd-trace-ot/src/test/groovy/datadog/trace/api/writer/DDApiTest.groovy
@@ -99,6 +99,7 @@ class DDApiTest extends Specification {
"duration" : 0,
"error" : 0,
"meta" : ["span.type": "fakeType", "thread.name": Thread.currentThread().getName(), "thread.id": "${Thread.currentThread().id}"],
+ "metrics" : [:],
"name" : "fakeOperation",
"parent_id": 0,
"resource" : "fakeResource",
@@ -112,6 +113,7 @@ class DDApiTest extends Specification {
"duration" : 0,
"error" : 0,
"meta" : ["span.type": "fakeType", "thread.name": Thread.currentThread().getName(), "thread.id": "${Thread.currentThread().id}"],
+ "metrics" : [:],
"name" : "fakeOperation",
"parent_id": 0,
"resource" : "my-resource",

Fix priority sampling implementation bugs

- use correct sampling flag name
- set sampling rate
- set sampling flags on root span metrics only
- tests to assert correct behavior