diff --git a/dd-trace-core/src/main/java/datadog/trace/core/otlp/trace/OtlpSpanAttributes.java b/dd-trace-core/src/main/java/datadog/trace/core/otlp/trace/OtlpSpanAttributes.java new file mode 100644 index 00000000000..75e0cc33be4 --- /dev/null +++ b/dd-trace-core/src/main/java/datadog/trace/core/otlp/trace/OtlpSpanAttributes.java @@ -0,0 +1,94 @@ +package datadog.trace.core.otlp.trace; + +import static datadog.trace.bootstrap.instrumentation.api.InstrumentationTags.DD_MEASURED; +import static datadog.trace.bootstrap.instrumentation.api.InstrumentationTags.DD_PARTIAL_VERSION; +import static datadog.trace.bootstrap.instrumentation.api.InstrumentationTags.DD_TOP_LEVEL; +import static datadog.trace.bootstrap.instrumentation.api.InstrumentationTags.DD_WAS_LONG_RUNNING; +import static datadog.trace.common.writer.ddagent.TraceMapper.ORIGIN_KEY; +import static datadog.trace.common.writer.ddagent.TraceMapper.PROCESS_TAGS_KEY; +import static datadog.trace.common.writer.ddagent.TraceMapper.SAMPLING_PRIORITY_KEY; +import static datadog.trace.common.writer.ddagent.TraceMapper.THREAD_ID; +import static datadog.trace.common.writer.ddagent.TraceMapper.THREAD_NAME; + +import datadog.trace.api.KnownTagCodec; +import datadog.trace.api.KnownTags; +import datadog.trace.api.TagMap; +import datadog.trace.bootstrap.instrumentation.api.UTF8BytesString; +import java.util.Arrays; +import java.util.HashSet; +import java.util.Set; + +/** Span attribute keys shared by the OTLP protobuf and JSON trace writers. */ +final class OtlpSpanAttributes { + + static final UTF8BytesString RESOURCE_NAME = UTF8BytesString.create("resource.name"); + static final UTF8BytesString OPERATION_NAME = UTF8BytesString.create("operation.name"); + static final UTF8BytesString SPAN_TYPE = UTF8BytesString.create("span.type"); + + /* + * Span fields and intercepted Metadata fields bypass TagMap entry projection. Resolve their + * keys through the registry here so OpenTelemetry names stay consistent with entry tags. + */ + static final UTF8BytesString SERVICE_NAME_KEY = otelKey(KnownTags.SERVICE_ID); + + static final UTF8BytesString HTTP_STATUS_CODE_KEY = otelKey(KnownTags.HTTP_STATUS_CODE_ID); + + static final UTF8BytesString HTTP_STATUS_CODE_KEY_DD = + UTF8BytesString.create(KnownTags.HTTP_STATUS_CODE_NAME); + + /** + * Keys that baggage must not export as attributes: the attributes the writers emit directly from + * span fields and intercepted Metadata fields (the span's own value wins), plus _dd.p.tid, whose + * upper 64 bits the OTLP traceId already carries. + */ + static final Set SKIPPED_BAGGAGE_KEYS = + new HashSet<>( + Arrays.asList( + "_dd.p.tid", + SERVICE_NAME_KEY.toString(), + RESOURCE_NAME.toString(), + OPERATION_NAME.toString(), + SPAN_TYPE.toString(), + SAMPLING_PRIORITY_KEY.toString(), + DD_MEASURED.toString(), + DD_TOP_LEVEL.toString(), + DD_PARTIAL_VERSION.toString(), + DD_WAS_LONG_RUNNING.toString(), + THREAD_ID.toString(), + THREAD_NAME.toString(), + HTTP_STATUS_CODE_KEY.toString(), + HTTP_STATUS_CODE_KEY_DD.toString(), + ORIGIN_KEY.toString(), + PROCESS_TAGS_KEY.toString())); + + /** + * True if a span tag is exported under exactly {@code key}, so baggage under that key would + * duplicate it. TagMap lookups resolve aliases (http.request.method finds http.method), so the + * matching entry's exported name is compared with the key: it depends on OTel semantics and, for + * direction-scoped tags, on the span's direction (peer.port is server.port on a client). + */ + static boolean isExportedTagKey(TagMap tags, String key, int direction, boolean otelSemantics) { + if (exportsAs(tags.getEntry(key), key, direction, otelSemantics)) { + return true; + } + if (otelSemantics) { + long directionalTagId = KnownTagCodec.directionalKeyOf(key, direction); + return directionalTagId != 0 + && exportsAs(tags.getEntry(directionalTagId), key, direction, true); + } + return false; + } + + private static boolean exportsAs( + TagMap.Entry entry, String key, int direction, boolean otelSemantics) { + return entry != null + && key.equals(otelSemantics ? entry.openTelemetryTag(direction) : entry.tag()); + } + + /** The OpenTelemetry-namespace key for a known tag, as named by the registry. */ + private static UTF8BytesString otelKey(long tagId) { + return UTF8BytesString.create(KnownTagCodec.openTelemetryTagOf(tagId)); + } + + private OtlpSpanAttributes() {} +} diff --git a/dd-trace-core/src/main/java/datadog/trace/core/otlp/trace/OtlpTraceJson.java b/dd-trace-core/src/main/java/datadog/trace/core/otlp/trace/OtlpTraceJson.java index 1575b26d228..68c80100d61 100644 --- a/dd-trace-core/src/main/java/datadog/trace/core/otlp/trace/OtlpTraceJson.java +++ b/dd-trace-core/src/main/java/datadog/trace/core/otlp/trace/OtlpTraceJson.java @@ -14,20 +14,27 @@ import static datadog.trace.common.writer.ddagent.TraceMapper.SAMPLING_PRIORITY_KEY; import static datadog.trace.common.writer.ddagent.TraceMapper.THREAD_ID; import static datadog.trace.common.writer.ddagent.TraceMapper.THREAD_NAME; -import static datadog.trace.core.CoreSpan.OWN_PROPAGATION_TAGS; +import static datadog.trace.core.CoreSpan.ROOT_PROPAGATION_TAGS; import static datadog.trace.core.otlp.common.OtlpCommonJson.hexSpanId; import static datadog.trace.core.otlp.common.OtlpCommonJson.hexTraceId; import static datadog.trace.core.otlp.common.OtlpCommonJson.writeAttribute; import static datadog.trace.core.otlp.common.OtlpTraceFlags.NO_TRACE_FLAGS; import static datadog.trace.core.otlp.common.OtlpTraceFlags.REMOTE_TRACE_FLAG; import static datadog.trace.core.otlp.common.OtlpTraceFlags.SAMPLED_TRACE_FLAG; +import static datadog.trace.core.otlp.trace.OtlpSpanAttributes.HTTP_STATUS_CODE_KEY; +import static datadog.trace.core.otlp.trace.OtlpSpanAttributes.HTTP_STATUS_CODE_KEY_DD; +import static datadog.trace.core.otlp.trace.OtlpSpanAttributes.OPERATION_NAME; +import static datadog.trace.core.otlp.trace.OtlpSpanAttributes.RESOURCE_NAME; +import static datadog.trace.core.otlp.trace.OtlpSpanAttributes.SERVICE_NAME_KEY; +import static datadog.trace.core.otlp.trace.OtlpSpanAttributes.SKIPPED_BAGGAGE_KEYS; +import static datadog.trace.core.otlp.trace.OtlpSpanAttributes.SPAN_TYPE; +import static datadog.trace.core.otlp.trace.OtlpSpanAttributes.isExportedTagKey; import static datadog.trace.core.otlp.trace.OtlpSpanKind.spanKind; import datadog.json.JsonWriter; import datadog.trace.api.Config; import datadog.trace.api.DDTags; import datadog.trace.api.KnownTagCodec; -import datadog.trace.api.KnownTags; import datadog.trace.api.TagMap; import datadog.trace.bootstrap.instrumentation.api.AgentSpanLink; import datadog.trace.bootstrap.instrumentation.api.UTF8BytesString; @@ -43,27 +50,6 @@ /** Provides writers for OpenTelemetry's "trace.proto" JSON encoding. */ public final class OtlpTraceJson { - private static final UTF8BytesString RESOURCE_NAME = UTF8BytesString.create("resource.name"); - private static final UTF8BytesString OPERATION_NAME = UTF8BytesString.create("operation.name"); - private static final UTF8BytesString SPAN_TYPE = UTF8BytesString.create("span.type"); - - /* - * Span fields and intercepted Metadata fields bypass TagMap entry projection. Resolve their - * keys through the registry here so JSON and protobuf use the same OpenTelemetry names. - */ - private static final UTF8BytesString SERVICE_NAME_KEY = otelKey(KnownTags.SERVICE_ID); - - private static final UTF8BytesString HTTP_STATUS_CODE_KEY = - otelKey(KnownTags.HTTP_STATUS_CODE_ID); - - private static final UTF8BytesString HTTP_STATUS_CODE_KEY_DD = - UTF8BytesString.create(KnownTags.HTTP_STATUS_CODE_NAME); - - /** The OpenTelemetry-namespace key for a known tag, as named by the registry. */ - private static UTF8BytesString otelKey(long tagId) { - return UTF8BytesString.create(KnownTagCodec.openTelemetryTagOf(tagId)); - } - private OtlpTraceJson() {} /** Writes one complete {@code Span} JSON object. */ @@ -113,7 +99,10 @@ public static void writeSpan( if (span.getSpanType() != null) { writeSpanTag(writer, SPAN_TYPE, span.getSpanType()); } - span.processTagsAndBaggage(metaWriter, OWN_PROPAGATION_TAGS); + // Export root propagation tags on the span selected by the collector for sampling tags, + // including when the exported trace contains no local root span. + span.processTagsAndBaggage( + metaWriter, metaWriter.includeSamplingTags ? ROOT_PROPAGATION_TAGS : 0); writer.endArray(); if (!links.isEmpty()) { @@ -252,9 +241,23 @@ public void accept(Metadata metadata) { writeSpanTag(writer, PROCESS_TAGS_KEY, metadata.processTags()); } - // The direction is computed once per span; boxing -1..2 hits the Integer cache. TagMap tags = metadata.getTags(); + // The direction is computed once per span; boxing -1..2 hits the Integer cache. Integer direction = KnownTagCodec.directionOf(tags); + boolean otelSemantics = Config.get().isTraceOtelSemanticsEnabled(); + + // Baggage carries the propagation tags (e.g. the _dd.p.dm sampling decision maker) and any + // baggage items injected as tags, as in the Datadog mappers. Span tags and the attributes + // written directly above win when they are exported under the same key, which keeps OTLP + // attribute keys unique without dropping baggage whose exported key differs. + for (Map.Entry entry : metadata.getBaggage().entrySet()) { + if (SKIPPED_BAGGAGE_KEYS.contains(entry.getKey()) + || isExportedTagKey(tags, entry.getKey(), direction, otelSemantics)) { + continue; + } + writeAttribute(writer, STRING_ATTRIBUTE, entry.getKey(), entry.getValue()); + } + tags.forEach(writer, direction, OtlpTraceJson::writeSpanTag); // reset for next span diff --git a/dd-trace-core/src/main/java/datadog/trace/core/otlp/trace/OtlpTraceProto.java b/dd-trace-core/src/main/java/datadog/trace/core/otlp/trace/OtlpTraceProto.java index ab64bde46cd..865411dbe03 100644 --- a/dd-trace-core/src/main/java/datadog/trace/core/otlp/trace/OtlpTraceProto.java +++ b/dd-trace-core/src/main/java/datadog/trace/core/otlp/trace/OtlpTraceProto.java @@ -14,7 +14,7 @@ import static datadog.trace.common.writer.ddagent.TraceMapper.SAMPLING_PRIORITY_KEY; import static datadog.trace.common.writer.ddagent.TraceMapper.THREAD_ID; import static datadog.trace.common.writer.ddagent.TraceMapper.THREAD_NAME; -import static datadog.trace.core.CoreSpan.OWN_PROPAGATION_TAGS; +import static datadog.trace.core.CoreSpan.ROOT_PROPAGATION_TAGS; import static datadog.trace.core.otlp.common.OtlpCommonProto.I32_WIRE_TYPE; import static datadog.trace.core.otlp.common.OtlpCommonProto.I64_WIRE_TYPE; import static datadog.trace.core.otlp.common.OtlpCommonProto.LEN_WIRE_TYPE; @@ -30,6 +30,14 @@ import static datadog.trace.core.otlp.common.OtlpTraceFlags.NO_TRACE_FLAGS; import static datadog.trace.core.otlp.common.OtlpTraceFlags.REMOTE_TRACE_FLAG; import static datadog.trace.core.otlp.common.OtlpTraceFlags.SAMPLED_TRACE_FLAG; +import static datadog.trace.core.otlp.trace.OtlpSpanAttributes.HTTP_STATUS_CODE_KEY; +import static datadog.trace.core.otlp.trace.OtlpSpanAttributes.HTTP_STATUS_CODE_KEY_DD; +import static datadog.trace.core.otlp.trace.OtlpSpanAttributes.OPERATION_NAME; +import static datadog.trace.core.otlp.trace.OtlpSpanAttributes.RESOURCE_NAME; +import static datadog.trace.core.otlp.trace.OtlpSpanAttributes.SERVICE_NAME_KEY; +import static datadog.trace.core.otlp.trace.OtlpSpanAttributes.SKIPPED_BAGGAGE_KEYS; +import static datadog.trace.core.otlp.trace.OtlpSpanAttributes.SPAN_TYPE; +import static datadog.trace.core.otlp.trace.OtlpSpanAttributes.isExportedTagKey; import static datadog.trace.core.otlp.trace.OtlpSpanKind.spanKind; import static java.nio.charset.StandardCharsets.UTF_8; @@ -39,7 +47,6 @@ import datadog.trace.api.DDTags; import datadog.trace.api.DDTraceId; import datadog.trace.api.KnownTagCodec; -import datadog.trace.api.KnownTags; import datadog.trace.api.TagMap; import datadog.trace.bootstrap.instrumentation.api.AgentSpanLink; import datadog.trace.bootstrap.instrumentation.api.UTF8BytesString; @@ -51,31 +58,11 @@ import datadog.trace.core.otlp.common.OtlpProtoBuffer; import datadog.trace.core.propagation.PropagationTags; import datadog.trace.core.propagation.PropagationTags.SamplingState; +import java.util.Map; /** Provides optimized writers for OpenTelemetry's "trace.proto" wire protocol. */ public final class OtlpTraceProto { - private static final UTF8BytesString RESOURCE_NAME = UTF8BytesString.create("resource.name"); - private static final UTF8BytesString OPERATION_NAME = UTF8BytesString.create("operation.name"); - private static final UTF8BytesString SPAN_TYPE = UTF8BytesString.create("span.type"); - - /* - * Span fields and intercepted Metadata fields bypass TagMap entry projection. Resolve their - * keys through the registry here so OpenTelemetry names stay consistent with entry tags. - */ - private static final UTF8BytesString SERVICE_NAME_KEY = otelKey(KnownTags.SERVICE_ID); - - private static final UTF8BytesString HTTP_STATUS_CODE_KEY = - otelKey(KnownTags.HTTP_STATUS_CODE_ID); - - private static final UTF8BytesString HTTP_STATUS_CODE_KEY_DD = - UTF8BytesString.create(KnownTags.HTTP_STATUS_CODE_NAME); - - /** The OpenTelemetry-namespace key for a known tag, as named by the registry. */ - private static UTF8BytesString otelKey(long tagId) { - return UTF8BytesString.create(KnownTagCodec.openTelemetryTagOf(tagId)); - } - private OtlpTraceProto() {} /** Records a scoped spans message after its nested span messages have been recorded. */ @@ -160,7 +147,10 @@ public static int recordSpanMessage( writeSpanTag(buf, SPAN_TYPE, span.getSpanType()); } - span.processTagsAndBaggage(metaWriter, OWN_PROPAGATION_TAGS); + // Export root propagation tags on the span selected by the collector for sampling tags, + // including when the exported trace contains no local root span. + span.processTagsAndBaggage( + metaWriter, metaWriter.includeSamplingTags ? ROOT_PROPAGATION_TAGS : 0); if (span.isError()) { int stateSize = 2; @@ -329,9 +319,24 @@ public void accept(Metadata metadata) { writeSpanTag(buf, PROCESS_TAGS_KEY, metadata.processTags()); } - // The direction is computed once per span; boxing -1..2 hits the Integer cache. TagMap tags = metadata.getTags(); + // The direction is computed once per span; boxing -1..2 hits the Integer cache. Integer direction = KnownTagCodec.directionOf(tags); + boolean otelSemantics = Config.get().isTraceOtelSemanticsEnabled(); + + // Baggage carries the propagation tags (e.g. the _dd.p.dm sampling decision maker) and any + // baggage items injected as tags, as in the Datadog mappers. Span tags and the attributes + // written directly above win when they are exported under the same key, which keeps OTLP + // attribute keys unique without dropping baggage whose exported key differs. + for (Map.Entry entry : metadata.getBaggage().entrySet()) { + if (SKIPPED_BAGGAGE_KEYS.contains(entry.getKey()) + || isExportedTagKey(tags, entry.getKey(), direction, otelSemantics)) { + continue; + } + writeTag(buf, 9, LEN_WIRE_TYPE); + writeAttribute(buf, STRING_ATTRIBUTE, entry.getKey(), entry.getValue()); + } + tags.forEach(buf, direction, OtlpTraceProto::writeSpanTag); // reset for next span diff --git a/dd-trace-core/src/test/java/datadog/trace/core/otlp/trace/OtlpTraceJsonCollectorTest.java b/dd-trace-core/src/test/java/datadog/trace/core/otlp/trace/OtlpTraceJsonCollectorTest.java index 46de746960b..5964ae125c2 100644 --- a/dd-trace-core/src/test/java/datadog/trace/core/otlp/trace/OtlpTraceJsonCollectorTest.java +++ b/dd-trace-core/src/test/java/datadog/trace/core/otlp/trace/OtlpTraceJsonCollectorTest.java @@ -9,6 +9,7 @@ import static java.util.Arrays.asList; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -462,6 +463,213 @@ void multipleSpansInATraceAreAllWritten() throws IOException { assertEquals(hexSpanId(((DDSpan) parent).getSpanId()), parsedChild.get("parentSpanId")); } + @Test + void propagationTagsAreWrittenOncePerTraceOnTheSpanSelectedForSamplingTags() throws IOException { + AgentSpan root = startRootWithPropagationTags("root.op"); + AgentSpan child = startChild(root, "child.op"); + child.finish(); + root.finish(); + + OtlpTraceJsonCollector collector = new OtlpTraceJsonCollector(); + collector.addTrace(asList((CoreSpan) root, (CoreSpan) child)); + List> parsedSpans = allSpans(collector.collectTraces()); + + // the collector selects the last exported span of each trace + assertEquals(asList("root.op", "child.op"), spanNames(parsedSpans)); + assertPropagationTagsOnlyOn(parsedSpans, "child.op", "root.op", "child.op"); + } + + @Test + void partialFlushWithoutTheLocalRootStillCarriesPropagationTags() throws IOException { + AgentSpan root = startRootWithPropagationTags("root.op"); + AgentSpan first = startChild(root, "first.child"); + AgentSpan second = startChild(root, "second.child"); + first.finish(); + second.finish(); + + // a partial flush exports finished children while the local root is still open + OtlpTraceJsonCollector collector = new OtlpTraceJsonCollector(); + collector.addTrace(asList((CoreSpan) first, (CoreSpan) second)); + List> parsedSpans = allSpans(collector.collectTraces()); + root.finish(); + + assertEquals(asList("first.child", "second.child"), spanNames(parsedSpans)); + assertPropagationTagsOnlyOn(parsedSpans, "second.child", "first.child", "second.child"); + } + + @Test + void propagationTagsAreWrittenOnceForEachTraceInThePayload() throws IOException { + AgentSpan rootA = startRootWithPropagationTags("a.root"); + AgentSpan childA = startChild(rootA, "a.child"); + AgentSpan rootB = startRootWithPropagationTags("b.root"); + AgentSpan firstB = startChild(rootB, "b.first"); + AgentSpan secondB = startChild(rootB, "b.second"); + AgentSpan rootC = startRootWithPropagationTags("c.root"); + assertNotEquals(rootA.getTraceId(), rootB.getTraceId(), "trace IDs must be distinct"); + assertNotEquals(rootB.getTraceId(), rootC.getTraceId(), "trace IDs must be distinct"); + childA.finish(); + rootA.finish(); + firstB.finish(); + secondB.finish(); + rootC.finish(); + + OtlpTraceJsonCollector collector = new OtlpTraceJsonCollector(); + collector.addTrace(asList((CoreSpan) rootA, (CoreSpan) childA)); + // partial flush of trace B, without its local root + collector.addTrace(asList((CoreSpan) firstB, (CoreSpan) secondB)); + collector.addTrace(asList((CoreSpan) rootC)); + List> parsedSpans = allSpans(collector.collectTraces()); + rootB.finish(); + + assertEquals( + asList("a.root", "a.child", "b.first", "b.second", "c.root"), spanNames(parsedSpans)); + assertPropagationTagsOnlyOn(parsedSpans, "a.child", "a.root", "a.child"); + assertPropagationTagsOnlyOn(parsedSpans, "b.second", "b.first", "b.second"); + assertPropagationTagsOnlyOn(parsedSpans, "c.root", "c.root"); + } + + /** Starts a local root span with a probability sampling decision and a baggage item. */ + private static AgentSpan startRootWithPropagationTags(String resourceName) { + AgentSpan root = TRACER.startSpan("test", "op.root"); + root.setResourceName(resourceName); + // 128-bit trace ids are on by default, so the root carries a _dd.p.tid propagation tag + assertNotEquals(0, root.getTraceId().toHighOrderLong()); + ((DDSpan) root) + .spanContext() + .getPropagationTags() + .tryUpdateProbabilitySamplingDecision( + PrioritySampling.USER_KEEP, + SamplingMechanism.REMOTE_ADAPTIVE_RULE, + 0.5, + false, + root.getTraceId().toLong(), + true); + root.setBaggageItem("baggage.key", "baggage.value"); + return root; + } + + private static AgentSpan startChild(AgentSpan parent, String resourceName) { + AgentSpan child = TRACER.startSpan("test", "op.child", parent.spanContext()); + child.setResourceName(resourceName); + return child; + } + + private static List spanNames(List> spans) { + List names = new ArrayList<>(); + for (Map span : spans) { + names.add(span.get("name")); + } + return names; + } + + /** + * Asserts that, among the exported spans of one trace, only {@code carrier} has the root + * propagation tags (once each), and that every span keeps its baggage. + */ + private static void assertPropagationTagsOnlyOn( + List> spans, String carrier, String... traceSpans) { + for (String spanName : traceSpans) { + Map span = spanNamed(spans, spanName); + if (spanName.equals(carrier)) { + assertEquals(asList("-12"), attributeValues(span, "_dd.p.dm"), spanName); + assertEquals(asList("0.5"), attributeValues(span, "_dd.p.ksr"), spanName); + // the OTLP traceId already carries the upper 64 bits + assertFalse(attributeKeys(span).contains("_dd.p.tid"), "_dd.p.tid must not be exported"); + } else { + for (String key : attributeKeys(span)) { + assertFalse(key.startsWith("_dd.p."), spanName + " must not carry " + key); + } + } + assertEquals(asList("baggage.value"), attributeValues(span, "baggage.key"), spanName); + } + } + + @Test + void baggageDoesNotDuplicateDirectlyWrittenAttributes() throws IOException { + AgentSpan span = TRACER.startSpan("test", "op.direct"); + span.setResourceName("direct.op"); + span.setBaggageItem("resource.name", "baggage.resource"); + span.setBaggageItem("operation.name", "baggage.operation"); + span.setBaggageItem("thread.name", "baggage.thread"); + span.finish(); + + OtlpTraceJsonCollector collector = new OtlpTraceJsonCollector(); + collector.addTrace(asList((CoreSpan) span)); + Map parsed = onlySpan(collector.collectTraces()); + + // each key must appear once, with the span's own value + assertEquals(asList("direct.op"), attributeValues(parsed, "resource.name")); + assertEquals(asList("op.direct"), attributeValues(parsed, "operation.name")); + assertEquals(asList(Thread.currentThread().getName()), attributeValues(parsed, "thread.name")); + } + + @TableTest({ + "scenario | kind | tag | otelName ", + "client hostname | client | peer.hostname | server.address ", + "server hostname | server | http.hostname | server.address ", + "server client ip | server | network.client.ip | network.peer.address" + }) + void baggageDoesNotDuplicateDirectionScopedAttributes(String kind, String tag, String otelName) + throws IOException { + Map parsed = exportWithDirectionalBaggage(true, kind, tag, otelName); + + // the tag is exported under its direction-specific name, once, with the span's own value + assertEquals(asList("span.value"), attributeValues(parsed, otelName)); + assertTrue(attributeValues(parsed, tag).isEmpty(), "expected no " + tag); + } + + @Test + void baggageIsKeptWhenOtelSemanticsDisabled() throws IOException { + // without the rename the tag keeps its Datadog name, so the baggage key is distinct + Map parsed = + exportWithDirectionalBaggage(false, "client", "peer.hostname", "server.address"); + + assertEquals(asList("span.value"), attributeValues(parsed, "peer.hostname")); + assertEquals(asList("baggage.value"), attributeValues(parsed, "server.address")); + } + + @TableTest({ + "scenario | otelSemantics | baggageKey | httpMethod | httpRequestMethod", + "off, otel name | false | http.request.method | span.value | baggage.value ", + "on, otel name | true | http.request.method | | span.value ", + "on, dd name | true | http.method | baggage.value | span.value " + }) + void baggageIsSkippedOnlyWhenATagIsExportedUnderTheSameKey( + boolean otelSemantics, String baggageKey, String httpMethod, String httpRequestMethod) + throws IOException { + // http.method and http.request.method are aliases of one tag: lookups by either name find it, + // but only one of them is exported, so baggage under the other name must be kept + Map parsed = + exportWithDirectionalBaggage(otelSemantics, "server", "http.method", baggageKey); + + assertEquals(expectedValues(httpMethod), attributeValues(parsed, "http.method")); + assertEquals(expectedValues(httpRequestMethod), attributeValues(parsed, "http.request.method")); + } + + private static List expectedValues(String value) { + return value == null || value.isEmpty() ? new ArrayList<>() : asList((Object) value); + } + + private static Map exportWithDirectionalBaggage( + boolean otelSemanticsEnabled, String kind, String tag, String baggageKey) throws IOException { + Config realConfig = Config.get(); + try (MockedStatic configMock = mockStatic(Config.class)) { + Config config = mock(Config.class, AdditionalAnswers.delegatesTo(realConfig)); + when(config.isTraceOtelSemanticsEnabled()).thenReturn(otelSemanticsEnabled); + configMock.when(Config::get).thenReturn(config); + + AgentSpan span = TRACER.startSpan("test", "op.directionalBaggage"); + span.setTag(SPAN_KIND, kind); + span.setTag(tag, "span.value"); + span.setBaggageItem(baggageKey, "baggage.value"); + span.finish(); + + OtlpTraceJsonCollector collector = new OtlpTraceJsonCollector(); + collector.addTrace(asList((CoreSpan) span)); + return onlySpan(collector.collectTraces()); + } + } + @Test void poisonedSpanResetsCollectorForNextTrace() throws IOException { // mid-trace exception (e.g. from a malformed span) must not leave partial state behind @@ -611,6 +819,13 @@ private static List> allSpans(OtlpPayload payload) throws IO return result; } + private static Map spanNamed(List> spans, String name) { + return spans.stream() + .filter(s -> name.equals(s.get("name"))) + .findFirst() + .orElseThrow(() -> new AssertionError(name + " span not found")); + } + @SuppressWarnings("unchecked") private static Map onlyLink(Map span) { List links = (List) span.get("links"); diff --git a/dd-trace-core/src/test/java/datadog/trace/core/otlp/trace/OtlpTraceProtoTest.java b/dd-trace-core/src/test/java/datadog/trace/core/otlp/trace/OtlpTraceProtoTest.java index d4ed0db45cd..8b80b3b4acc 100644 --- a/dd-trace-core/src/test/java/datadog/trace/core/otlp/trace/OtlpTraceProtoTest.java +++ b/dd-trace-core/src/test/java/datadog/trace/core/otlp/trace/OtlpTraceProtoTest.java @@ -46,6 +46,7 @@ import java.util.ArrayList; import java.util.HashMap; import java.util.HashSet; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Set; @@ -53,6 +54,7 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.CsvSource; import org.junit.jupiter.params.provider.MethodSource; import org.mockito.AdditionalAnswers; import org.mockito.MockedStatic; @@ -739,6 +741,309 @@ void poisonedSpanResetsCollectorForNextTrace() { assertTrue(payload.getContentLength() > 0, "normal trace after reset must still export"); } + @Test + void propagationTagsAreWrittenOncePerTraceOnTheSpanSelectedForSamplingTags() throws IOException { + AgentSpan root = startRootWithPropagationTags("root.op"); + AgentSpan child = startChild(root, "child.op"); + child.finish(); + root.finish(); + + OtlpTraceProtoCollector collector = new OtlpTraceProtoCollector(); + collector.addTrace(asList((CoreSpan) root, (CoreSpan) child)); + Map> attributesBySpan = + parseStringAttributeListsBySpanName(collector.collectTraces()); + + // the payload buffer is prepending, so the collector selects the first exported span + assertEquals(asList("root.op", "child.op"), new ArrayList<>(attributesBySpan.keySet())); + assertPropagationTagsOnlyOn(attributesBySpan, "root.op", "root.op", "child.op"); + } + + @Test + void partialFlushWithoutTheLocalRootStillCarriesPropagationTags() throws IOException { + AgentSpan root = startRootWithPropagationTags("root.op"); + AgentSpan first = startChild(root, "first.child"); + AgentSpan second = startChild(root, "second.child"); + first.finish(); + second.finish(); + + // a partial flush exports finished children while the local root is still open + OtlpTraceProtoCollector collector = new OtlpTraceProtoCollector(); + collector.addTrace(asList((CoreSpan) first, (CoreSpan) second)); + Map> attributesBySpan = + parseStringAttributeListsBySpanName(collector.collectTraces()); + root.finish(); + + assertEquals(asList("first.child", "second.child"), new ArrayList<>(attributesBySpan.keySet())); + assertPropagationTagsOnlyOn(attributesBySpan, "first.child", "first.child", "second.child"); + } + + @Test + void propagationTagsAreWrittenOnceForEachTraceInThePayload() throws IOException { + AgentSpan rootA = startRootWithPropagationTags("a.root"); + AgentSpan childA = startChild(rootA, "a.child"); + AgentSpan rootB = startRootWithPropagationTags("b.root"); + AgentSpan firstB = startChild(rootB, "b.first"); + AgentSpan secondB = startChild(rootB, "b.second"); + AgentSpan rootC = startRootWithPropagationTags("c.root"); + assertNotEquals(rootA.getTraceId(), rootB.getTraceId(), "trace IDs must be distinct"); + assertNotEquals(rootB.getTraceId(), rootC.getTraceId(), "trace IDs must be distinct"); + childA.finish(); + rootA.finish(); + firstB.finish(); + secondB.finish(); + rootC.finish(); + + OtlpTraceProtoCollector collector = new OtlpTraceProtoCollector(); + collector.addTrace(asList((CoreSpan) rootA, (CoreSpan) childA)); + // partial flush of trace B, without its local root + collector.addTrace(asList((CoreSpan) firstB, (CoreSpan) secondB)); + collector.addTrace(asList((CoreSpan) rootC)); + Map> attributesBySpan = + parseStringAttributeListsBySpanName(collector.collectTraces()); + rootB.finish(); + + assertEquals(5, attributesBySpan.size(), "all spans must be exported"); + assertPropagationTagsOnlyOn(attributesBySpan, "a.root", "a.root", "a.child"); + assertPropagationTagsOnlyOn(attributesBySpan, "b.first", "b.first", "b.second"); + assertPropagationTagsOnlyOn(attributesBySpan, "c.root", "c.root"); + } + + /** Starts a local root span with a probability sampling decision and a baggage item. */ + private static AgentSpan startRootWithPropagationTags(String resourceName) { + AgentSpan root = TRACER.startSpan("test", "op.root"); + root.setResourceName(resourceName); + // 128-bit trace ids are on by default, so the root carries a _dd.p.tid propagation tag + assertNotEquals(0, root.getTraceId().toHighOrderLong()); + ((DDSpan) root) + .spanContext() + .getPropagationTags() + .tryUpdateProbabilitySamplingDecision( + PrioritySampling.USER_KEEP, + SamplingMechanism.REMOTE_USER_RULE, + 0.5, + false, + root.getTraceId().toLong(), + true); + root.setBaggageItem("baggage.key", "baggage.value"); + return root; + } + + private static AgentSpan startChild(AgentSpan parent, String resourceName) { + AgentSpan child = TRACER.startSpan("test", "op.child", parent.spanContext()); + child.setResourceName(resourceName); + return child; + } + + /** + * Asserts that, among the exported spans of one trace, only {@code carrier} has the root + * propagation tags (once each), and that every span keeps its baggage. + */ + private static void assertPropagationTagsOnlyOn( + Map> attributesBySpan, String carrier, String... traceSpans) { + for (String spanName : traceSpans) { + List attributes = attributesBySpan.get(spanName); + assertNotNull(attributes, spanName + " must be exported"); + if (spanName.equals(carrier)) { + assertEquals(asList("-11"), stringAttributeValues(attributes, "_dd.p.dm"), spanName); + assertEquals(asList("0.5"), stringAttributeValues(attributes, "_dd.p.ksr"), spanName); + // the OTLP traceId already carries the upper 64 bits + assertEquals( + emptyList(), stringAttributeValues(attributes, "_dd.p.tid"), "_dd.p.tid not exported"); + } else { + for (String[] attribute : attributes) { + assertFalse( + attribute[0].startsWith("_dd.p."), spanName + " must not carry " + attribute[0]); + } + } + assertEquals(asList("baggage.value"), stringAttributeValues(attributes, "baggage.key")); + } + } + + @Test + void baggageDoesNotDuplicateDirectlyWrittenAttributes() throws IOException { + AgentSpan span = TRACER.startSpan("test", "op.direct"); + span.setResourceName("direct.op"); + span.setBaggageItem("resource.name", "baggage.resource"); + span.setBaggageItem("operation.name", "baggage.operation"); + span.setBaggageItem("thread.name", "baggage.thread"); + span.finish(); + + OtlpTraceProtoCollector collector = new OtlpTraceProtoCollector(); + collector.addTrace(asList((CoreSpan) span)); + List attributes = + parseStringAttributeListsBySpanName(collector.collectTraces()).get("direct.op"); + assertNotNull(attributes, "span must be exported"); + + // each key must appear once, with the span's own value + assertEquals(asList("direct.op"), stringAttributeValues(attributes, "resource.name")); + assertEquals(asList("op.direct"), stringAttributeValues(attributes, "operation.name")); + assertEquals( + asList(Thread.currentThread().getName()), stringAttributeValues(attributes, "thread.name")); + } + + @ParameterizedTest(name = "{0} {1} -> {2}") + @CsvSource({ + "client, peer.hostname, server.address", + "server, http.hostname, server.address", + "server, network.client.ip, network.peer.address" + }) + void baggageDoesNotDuplicateDirectionScopedAttributes(String kind, String tag, String otelName) + throws IOException { + List attributes = exportWithDirectionalBaggage(true, kind, tag, otelName); + + // the tag is exported under its direction-specific name, once, with the span's own value + assertEquals(asList("span.value"), stringAttributeValues(attributes, otelName)); + assertTrue(stringAttributeValues(attributes, tag).isEmpty(), "expected no " + tag); + } + + @Test + void baggageIsKeptWhenOtelSemanticsDisabled() throws IOException { + // without the rename the tag keeps its Datadog name, so the baggage key is distinct + List attributes = + exportWithDirectionalBaggage(false, "client", "peer.hostname", "server.address"); + + assertEquals(asList("span.value"), stringAttributeValues(attributes, "peer.hostname")); + assertEquals(asList("baggage.value"), stringAttributeValues(attributes, "server.address")); + } + + @ParameterizedTest(name = "otel semantics {0}, baggage {1}") + @CsvSource({ + "false, http.request.method, span.value, baggage.value", + "true, http.request.method, '', span.value", + "true, http.method, baggage.value, span.value" + }) + void baggageIsSkippedOnlyWhenATagIsExportedUnderTheSameKey( + boolean otelSemantics, String baggageKey, String httpMethod, String httpRequestMethod) + throws IOException { + // http.method and http.request.method are aliases of one tag: lookups by either name find it, + // but only one of them is exported, so baggage under the other name must be kept + List attributes = + exportWithDirectionalBaggage(otelSemantics, "server", "http.method", baggageKey); + + assertEquals(expectedValues(httpMethod), stringAttributeValues(attributes, "http.method")); + assertEquals( + expectedValues(httpRequestMethod), + stringAttributeValues(attributes, "http.request.method")); + } + + private static List expectedValues(String value) { + return value == null || value.isEmpty() ? new ArrayList<>() : asList(value); + } + + private static List exportWithDirectionalBaggage( + boolean otelSemanticsEnabled, String kind, String tag, String baggageKey) throws IOException { + Config realConfig = Config.get(); + try (MockedStatic configMock = mockStatic(Config.class)) { + Config config = mock(Config.class, AdditionalAnswers.delegatesTo(realConfig)); + when(config.isTraceOtelSemanticsEnabled()).thenReturn(otelSemanticsEnabled); + configMock.when(Config::get).thenReturn(config); + + AgentSpan span = TRACER.startSpan("test", "op.directionalBaggage"); + span.setResourceName("directional.baggage"); + span.setTag(SPAN_KIND, kind); + span.setTag(tag, "span.value"); + span.setBaggageItem(baggageKey, "baggage.value"); + span.finish(); + + OtlpTraceProtoCollector collector = new OtlpTraceProtoCollector(); + collector.addTrace(asList((CoreSpan) span)); + List attributes = + parseStringAttributeListsBySpanName(collector.collectTraces()).get("directional.baggage"); + assertNotNull(attributes, "span must be exported"); + return attributes; + } + } + + /** Returns the values of every attribute with the given key, keeping duplicates. */ + private static List stringAttributeValues(List attributes, String key) { + List values = new ArrayList<>(); + for (String[] attribute : attributes) { + if (key.equals(attribute[0])) { + values.add(attribute[1]); + } + } + return values; + } + + /** + * Maps each span name, in wire order, to its raw string-valued attributes, as {key, value} pairs + * keeping their order and duplicate keys. + */ + private static Map> parseStringAttributeListsBySpanName( + OtlpPayload payload) throws IOException { + CodedInputStream tracesData = CodedInputStream.newInstance(payload.getContent()); + tracesData.readTag(); // field 1: TracesData.resource_spans + CodedInputStream resourceSpans = tracesData.readBytes().newCodedInput(); + Map> result = new LinkedHashMap<>(); + while (!resourceSpans.isAtEnd()) { + int tag = resourceSpans.readTag(); + if (WireFormat.getTagFieldNumber(tag) != 2) { + resourceSpans.skipField(tag); + continue; + } + CodedInputStream scopeSpans = resourceSpans.readBytes().newCodedInput(); + while (!scopeSpans.isAtEnd()) { + int scopeTag = scopeSpans.readTag(); + if (WireFormat.getTagFieldNumber(scopeTag) != 2) { + scopeSpans.skipField(scopeTag); + continue; + } + CodedInputStream spanData = scopeSpans.readBytes().newCodedInput(); + String name = null; + List attributes = new ArrayList<>(); + while (!spanData.isAtEnd()) { + int spanTag = spanData.readTag(); + switch (WireFormat.getTagFieldNumber(spanTag)) { + case 5: + name = spanData.readString(); + break; + case 9: + String[] attribute = readStringKeyValue(spanData.readBytes().newCodedInput()); + if (attribute != null) { + attributes.add(attribute); + } + break; + default: + spanData.skipField(spanTag); + } + } + result.put(name, attributes); + } + } + return result; + } + + /** + * Reads a {@code KeyValue { string key = 1; AnyValue value = 2; }} as {key, value} when its value + * is an {@code AnyValue { string string_value = 1; }}, otherwise returns null. + */ + private static String[] readStringKeyValue(CodedInputStream keyValue) throws IOException { + String key = null; + String value = null; + while (!keyValue.isAtEnd()) { + int tag = keyValue.readTag(); + switch (WireFormat.getTagFieldNumber(tag)) { + case 1: + key = keyValue.readString(); + break; + case 2: + CodedInputStream anyValue = keyValue.readBytes().newCodedInput(); + while (!anyValue.isAtEnd()) { + int valueTag = anyValue.readTag(); + if (WireFormat.getTagFieldNumber(valueTag) == 1) { + value = anyValue.readString(); + } else { + anyValue.skipField(valueTag); + } + } + break; + default: + keyValue.skipField(tag); + } + } + return key != null && value != null ? new String[] {key, value} : null; + } + @Test void testSpanOrderInTracePreserved() throws IOException { // Verifies that spans appear in the payload with the same order as the original trace