Skip to content

Defer parsing the extracted OpenTelemetry tracestate member #12987

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
dougqh wants to merge 1 commit into
base: master
Choose a base branch
from
Draft

Defer parsing the extracted OpenTelemetry tracestate member #12987

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
121 changes: 121 additions & 0 deletions dd-trace-core/src/main/java/datadog/trace/core/propagation/ptags/InheritedSamplingState.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,121 @@
package datadog.trace.core.propagation.ptags;

import datadog.trace.core.propagation.PropagationTags.SamplingState;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;

/**
* The sampling state for an extracted trace whose OpenTelemetry {@code ot} tracestate member hasn't
* been parsed yet. Extraction used to parse the member and check it against the inherited decision
* on the request thread, though only W3C injection, OTLP export and a local change of decision ever
* read it. This keeps the raw member and parses it on first read, so a leaf service exporting to
* the Datadog agent never pays for it.
*
* <p>Consistency checks made before the first read are recorded as pending and applied when the
* member is parsed. Each check drops the threshold when {@code rv >= th} doesn't match the sampling
* decision checked against, and dropping it is idempotent and leaves the state consistent with any
* decision. So any sequence of checks has the same effect as one per decision checked, in any
* order, and two bits record them without changing the output.
*/
final class InheritedSamplingState extends SamplingState {
static final byte NO_CHECKS = 0;
private static final byte CHECKED_SAMPLED = 1;
private static final byte CHECKED_NOT_SAMPLED = 2;

/** Marks a member that parsed to nothing, distinct from not parsed yet. */
private static final Object NO_OTEL_TRACE_STATE = new Object();

private static final AtomicReferenceFieldUpdater<InheritedSamplingState, Object>
RESOLVED_UPDATER =
AtomicReferenceFieldUpdater.newUpdater(
InheritedSamplingState.class, Object.class, "resolved");

private final CharSequence rawOtelTraceState;
private final int otelOriginalSize;
private final byte pendingChecks;

/** {@code null} until parsed, then the {@link OtelTraceState} or {@link #NO_OTEL_TRACE_STATE}. */
private volatile Object resolved;

InheritedSamplingState(
int samplingPriority,
String tracestate,
CharSequence decisionMaker,
CharSequence knuthSamplingRate,
CharSequence rawOtelTraceState,
int otelOriginalSize,
byte pendingChecks) {
super(samplingPriority, tracestate, null, decisionMaker, knuthSamplingRate);
this.rawOtelTraceState = rawOtelTraceState;
this.otelOriginalSize = otelOriginalSize;
this.pendingChecks = pendingChecks;
}

/** Returns the pending checks with one more against {@code sampled} added. */
static byte withCheck(byte pendingChecks, boolean sampled) {
return (byte) (pendingChecks | (sampled ? CHECKED_SAMPLED : CHECKED_NOT_SAMPLED));
}

/** A copy of this unparsed member under a new decision, with one more pending check. */
InheritedSamplingState withCheck(
int samplingPriority,
String tracestate,
CharSequence decisionMaker,
CharSequence knuthSamplingRate,
boolean sampled) {
return new InheritedSamplingState(
samplingPriority,
tracestate,
decisionMaker,
knuthSamplingRate,
rawOtelTraceState,
otelOriginalSize,
withCheck(pendingChecks, sampled));
}

boolean isResolved() {
return resolved != null;
}

CharSequence rawOtelTraceState() {
return rawOtelTraceState;
}

int otelOriginalSize() {
return otelOriginalSize;
}

byte pendingChecks() {
return pendingChecks;
}

/**
* Parses the member and applies the pending checks on first read. Racing readers agree on one
* instance, so identity checks on otel states keep skipping redundant work.
*/
@Override
public CharSequence getOtelTraceState() {
Object current = resolved;
if (current == null) {
current = resolve();
}
return current == NO_OTEL_TRACE_STATE ? null : (OtelTraceState) current;
}

private Object resolve() {
OtelTraceState otelTraceState = OtelTraceState.parse(rawOtelTraceState, otelOriginalSize);
if ((pendingChecks & CHECKED_SAMPLED) != 0) {
otelTraceState = checked(otelTraceState, true);
}
if ((pendingChecks & CHECKED_NOT_SAMPLED) != 0) {
otelTraceState = checked(otelTraceState, false);
}
Object built = otelTraceState == null ? NO_OTEL_TRACE_STATE : otelTraceState;
return RESOLVED_UPDATER.compareAndSet(this, null, built) ? built : resolved;
}

private static OtelTraceState checked(OtelTraceState otelTraceState, boolean sampled) {
return otelTraceState == null || otelTraceState.isConsistentWith(sampled)
? otelTraceState
: otelTraceState.withoutThreshold();
}
}
109 changes: 98 additions & 11 deletions dd-trace-core/src/main/java/datadog/trace/core/propagation/ptags/PTagsFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -202,10 +202,11 @@ static class PTags extends PropagationTags {
orgPropagationMarkerTagValue,
llmObsTagValues,
null,
null);
null,
0);
}

// Takes tracestate/otelTraceState up-front to build SamplingState in one allocation.
// Takes tracestate and the raw otel member up-front to build SamplingState in one allocation.
PTags(
PTagsFactory factory,
List<TagElement> tagPairs,
Expand All @@ -218,15 +219,21 @@ static class PTags extends PropagationTags {
TagValue orgPropagationMarkerTagValue,
@Nonnull LLMObsTagValues llmObsTagValues,
String tracestate,
OtelTraceState otelTraceState) {
CharSequence rawOtelTraceState,
int otelOriginalSize) {
assert tagPairs == null || tagPairs.size() % 2 == 0;
this.factory = factory;
this.tagPairs = tagPairs;
this.canChangeDecisionMaker = decisionMakerTagValue == null;
this.traceSource = traceSource;
this.tracestate = tracestate;
this.samplingState =
initialSamplingState(samplingPriority, tracestate, otelTraceState, decisionMakerTagValue);
initialSamplingState(
samplingPriority,
tracestate,
rawOtelTraceState,
otelOriginalSize,
decisionMakerTagValue);
this.origin = origin;
this.lastParentId = lastParentId;
this.orgPropagationMarkerTagValue = orgPropagationMarkerTagValue;
Expand Down Expand Up @@ -255,7 +262,8 @@ static PTags withError(PTagsFactory factory, String error) {
null,
LLMObsTagValues.EMPTY,
null,
null);
null,
0);
pTags.error = error;
return pTags;
}
Expand All @@ -265,6 +273,10 @@ public void updateTraceSamplingPriority(int samplingPriority, int samplingMechan
synchronized (samplingStateLock) {
if (samplingPriority != PrioritySampling.UNSET && canChangeDecisionMaker
|| samplingMechanism == SamplingMechanism.EXTERNAL_OVERRIDE) {
if (deferOtelCheck(
samplingPriority, samplingMechanism, getKnuthSamplingRateTagValue(), true)) {
return;
}
OtelTraceState nextOtelTraceState =
reconcileOtelTraceState(getOtelTraceState(), samplingPriority, samplingMechanism);
installSamplingState(samplingPriority, samplingMechanism, nextOtelTraceState);
Expand All @@ -283,14 +295,21 @@ public boolean tryUpdateTraceSamplingPriority(
if (!allowOverride && current.getSamplingPriority() != PrioritySampling.UNSET) {
return false;
}
boolean updateDecisionMaker =
canChangeDecisionMaker || samplingMechanism == SamplingMechanism.EXTERNAL_OVERRIDE;
TagValue knuthSamplingRate = getKnuthSamplingRateTagValue();
if (deferOtelCheck(
samplingPriority, samplingMechanism, knuthSamplingRate, updateDecisionMaker)) {
return true;
}
OtelTraceState nextOtelTraceState =
reconcileOtelTraceState(getOtelTraceState(), samplingPriority, samplingMechanism);
installSamplingState(
samplingPriority,
samplingMechanism,
nextOtelTraceState,
getKnuthSamplingRateTagValue(),
canChangeDecisionMaker || samplingMechanism == SamplingMechanism.EXTERNAL_OVERRIDE);
knuthSamplingRate,
updateDecisionMaker);
return true;
}
}
Expand Down Expand Up @@ -432,16 +451,57 @@ private TagValue nextDecisionMakerTagValue(
private static SamplingState initialSamplingState(
int samplingPriority,
String tracestate,
OtelTraceState otelTraceState,
CharSequence rawOtelTraceState,
int otelOriginalSize,
TagValue decisionMakerTagValue) {
if (rawOtelTraceState != null) {
return new InheritedSamplingState(
samplingPriority,
tracestate,
decisionMakerTagValue,
null,
rawOtelTraceState,
otelOriginalSize,
InheritedSamplingState.NO_CHECKS);
}
if (samplingPriority == PrioritySampling.UNSET
&& tracestate == null
&& otelTraceState == null
&& decisionMakerTagValue == null) {
return EMPTY_SAMPLING_STATE;
}
return new SamplingState(
samplingPriority, tracestate, otelTraceState, decisionMakerTagValue, null);
return new SamplingState(samplingPriority, tracestate, null, decisionMakerTagValue, null);
}

/**
* Records a decision against an extracted otel member that hasn't been parsed, without parsing
* it. For an external override or unknown mechanism, reconciling the member is only a
* consistency check, which can wait for the first read; a local decision rewrites the member
* and parses it. Called with the sampling state lock held.
*/
private boolean deferOtelCheck(
int samplingPriority,
int samplingMechanism,
TagValue knuthSamplingRate,
boolean updateDecisionMaker) {
SamplingState current = samplingState;
if (!(current instanceof InheritedSamplingState)
|| ((InheritedSamplingState) current).isResolved()
|| (samplingMechanism != SamplingMechanism.EXTERNAL_OVERRIDE
&& samplingMechanism != SamplingMechanism.UNKNOWN)) {
return false;
}
TagValue decisionMaker =
nextDecisionMakerTagValue(samplingPriority, samplingMechanism, updateDecisionMaker);
clearCachedHeader(W3C);
samplingState =
((InheritedSamplingState) current)
.withCheck(
samplingPriority,
tracestate,
decisionMaker,
knuthSamplingRate,
samplingPriority > 0);
return true;
}

private static OtelTraceState reconcileOtelTraceState(
Expand Down Expand Up @@ -899,6 +959,11 @@ public void updateW3CTracestateFrom(PropagationTags source) {
}
PTags sourcePTags = (PTags) source;
SamplingState sourceState = sourcePTags.samplingState();
if (sourceState instanceof InheritedSamplingState
&& !((InheritedSamplingState) sourceState).isResolved()) {
setW3CTracestate(sourceState.getTracestate(), (InheritedSamplingState) sourceState);
return;
}
CharSequence sourceOtelTraceState = sourceState.getOtelTraceState();
setW3CTracestate(
sourceState.getTracestate(),
Expand Down Expand Up @@ -927,6 +992,28 @@ private void setW3CTracestate(String tracestate, OtelTraceState otelTraceState)
}
}

/** Takes over an extracted otel member that hasn't been parsed, without parsing it. */
private void setW3CTracestate(String tracestate, InheritedSamplingState source) {
synchronized (samplingStateLock) {
clearCachedHeader(W3C);
int samplingPriority = samplingState.getSamplingPriority();
byte pendingChecks = source.pendingChecks();
if (samplingPriority != PrioritySampling.UNSET) {
pendingChecks = InheritedSamplingState.withCheck(pendingChecks, samplingPriority > 0);
}
this.tracestate = tracestate;
this.samplingState =
new InheritedSamplingState(
samplingPriority,
tracestate,
getDecisionMakerTagValue(),
getKnuthSamplingRateTagValue(),
source.rawOtelTraceState(),
source.otelOriginalSize(),
pendingChecks);
}
}

private OtelTraceState getOtelTraceState() {
return (OtelTraceState) samplingState.getOtelTraceState();
}
Expand Down
Loading
Loading