From 237ae233dc295273fdef80335ba8b809c4fdfbd3 Mon Sep 17 00:00:00 2001 From: Matthew Li Date: Wed, 28 Oct 2026 13:51:50 -0400 Subject: [PATCH 1/2] Merge OtlpSpanTags into OtlpSpanAttributes as a nested Writer Both classes decide which span tags become OTLP attributes. Keep OtlpSpanAttributes as the static helper for keys and baggage rules, and nest the per-span tag writer as OtlpSpanAttributes.Writer. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../core/otlp/trace/OtlpSpanAttributes.java | 69 ++++++++++++++++++- .../trace/core/otlp/trace/OtlpSpanTags.java | 67 ------------------ .../trace/core/otlp/trace/OtlpTraceJson.java | 4 +- .../trace/core/otlp/trace/OtlpTraceProto.java | 4 +- ...sTest.java => OtlpSpanAttributesTest.java} | 2 +- 5 files changed, 73 insertions(+), 73 deletions(-) delete mode 100644 dd-trace-core/src/main/java/datadog/trace/core/otlp/trace/OtlpSpanTags.java rename dd-trace-core/src/test/java/datadog/trace/core/otlp/trace/{OtlpSpanTagsTest.java => OtlpSpanAttributesTest.java} (99%) 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 index 75e0cc33be4..8fc58e13e48 100644 --- 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 @@ -13,12 +13,19 @@ import datadog.trace.api.KnownTagCodec; import datadog.trace.api.KnownTags; import datadog.trace.api.TagMap; +import datadog.trace.api.function.TriConsumer; import datadog.trace.bootstrap.instrumentation.api.UTF8BytesString; +import datadog.trace.core.Metadata; +import java.util.ArrayList; import java.util.Arrays; import java.util.HashSet; +import java.util.List; import java.util.Set; -/** Span attribute keys shared by the OTLP protobuf and JSON trace writers. */ +/** + * Span attributes shared by the OTLP protobuf and JSON trace writers: the keys and baggage rules + * they use, and a {@link Writer} for the tags a span carries as attributes. + */ final class OtlpSpanAttributes { static final UTF8BytesString RESOURCE_NAME = UTF8BytesString.create("resource.name"); @@ -90,5 +97,65 @@ private static UTF8BytesString otelKey(long tagId) { return UTF8BytesString.create(KnownTagCodec.openTelemetryTagOf(tagId)); } + /** + * Writes the tags an OTLP span carries as attributes: the span's direct tags, plus the span + * defaults ({@code DD_TRACE_SPAN_TAGS}, {@code env}, ...) it doesn't set itself. The global tags + * are left to the resource, which carries them once per payload. + * + *

A span default is only written while the span still carries it: one the span removed, or + * that interception consumed into a span field (such as {@code resource.name}), is skipped. When + * the tracer tags need interception they're copied into each span, so global tags then also + * appear on the span, as before. + * + *

Not thread-safe: each writer owns one. + */ + static final class Writer { + private final T target; + private final TriConsumer tagWriter; + + // per-span state + private Integer direction; + private TagMap spanTags; + private TagMap spanDefaultTags; + private final List overriddenDefaults = new ArrayList<>(4); + + Writer(T target, TriConsumer tagWriter) { + this.target = target; + this.tagWriter = tagWriter; + } + + /** Writes the span's attribute tags, naming them for the span's {@code direction}. */ + void write(Metadata metadata, Integer direction) { + this.direction = direction; + this.spanTags = metadata.getTags(); + this.spanDefaultTags = metadata.getSpanDefaultTags(); + try { + spanTags.forEachLocal(this, Writer::writeDirectTag); + if (!spanDefaultTags.isEmpty()) { + spanDefaultTags.forEach(this, Writer::writeDefaultTag); + } + } finally { + this.spanTags = null; + this.spanDefaultTags = null; + overriddenDefaults.clear(); + } + } + + private void writeDirectTag(TagMap.EntryReader entry) { + if (!spanDefaultTags.isEmpty() && spanDefaultTags.containsKey(entry.tag())) { + overriddenDefaults.add(entry.tag()); + } + tagWriter.accept(target, direction, entry); + } + + private void writeDefaultTag(TagMap.EntryReader entry) { + // skip defaults the span set itself, or no longer carries (removed, or consumed by + // interception) + if (!overriddenDefaults.contains(entry.tag()) && spanTags.getEntry(entry.tag()) != null) { + tagWriter.accept(target, direction, entry); + } + } + } + private OtlpSpanAttributes() {} } diff --git a/dd-trace-core/src/main/java/datadog/trace/core/otlp/trace/OtlpSpanTags.java b/dd-trace-core/src/main/java/datadog/trace/core/otlp/trace/OtlpSpanTags.java deleted file mode 100644 index 624b95b96bb..00000000000 --- a/dd-trace-core/src/main/java/datadog/trace/core/otlp/trace/OtlpSpanTags.java +++ /dev/null @@ -1,67 +0,0 @@ -package datadog.trace.core.otlp.trace; - -import datadog.trace.api.TagMap; -import datadog.trace.api.function.TriConsumer; -import datadog.trace.core.Metadata; -import java.util.ArrayList; -import java.util.List; - -/** - * Visits the tags an OTLP span carries as attributes: the span's direct tags, plus the span - * defaults ({@code DD_TRACE_SPAN_TAGS}, {@code env}, ...) it doesn't set itself. The global tags - * are left to the resource, which carries them once per payload. - * - *

A span default is only written while the span still carries it: one the span removed, or that - * interception consumed into a span field (such as {@code resource.name}), is skipped. When the - * tracer tags need interception they're copied into each span, so global tags then also appear on - * the span, as before. - * - *

Not thread-safe: each writer owns one. - */ -final class OtlpSpanTags { - private final T target; - private final TriConsumer tagWriter; - - // per-span state - private Integer direction; - private TagMap spanTags; - private TagMap spanDefaultTags; - private final List overriddenDefaults = new ArrayList<>(4); - - OtlpSpanTags(T target, TriConsumer tagWriter) { - this.target = target; - this.tagWriter = tagWriter; - } - - /** Writes the span's attribute tags, naming them for the span's {@code direction}. */ - void write(Metadata metadata, Integer direction) { - this.direction = direction; - this.spanTags = metadata.getTags(); - this.spanDefaultTags = metadata.getSpanDefaultTags(); - try { - spanTags.forEachLocal(this, OtlpSpanTags::writeDirectTag); - if (!spanDefaultTags.isEmpty()) { - spanDefaultTags.forEach(this, OtlpSpanTags::writeDefaultTag); - } - } finally { - this.spanTags = null; - this.spanDefaultTags = null; - overriddenDefaults.clear(); - } - } - - private void writeDirectTag(TagMap.EntryReader entry) { - if (!spanDefaultTags.isEmpty() && spanDefaultTags.containsKey(entry.tag())) { - overriddenDefaults.add(entry.tag()); - } - tagWriter.accept(target, direction, entry); - } - - private void writeDefaultTag(TagMap.EntryReader entry) { - // skip defaults the span set itself, or no longer carries (removed, or consumed by - // interception) - if (!overriddenDefaults.contains(entry.tag()) && spanTags.getEntry(entry.tag()) != null) { - tagWriter.accept(target, direction, entry); - } - } -} 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 18d18c01b4f..d849f27a86a 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 @@ -186,14 +186,14 @@ private static void writeSpanTag(JsonWriter writer, UTF8BytesString key, long va public static class MetaWriter implements MetadataConsumer { private final JsonWriter writer; - private final OtlpSpanTags spanTags; + private final OtlpSpanAttributes.Writer spanTags; private boolean includeProcessTags; private boolean includeSamplingTags; public MetaWriter(JsonWriter writer) { this.writer = writer; - this.spanTags = new OtlpSpanTags<>(writer, OtlpTraceJson::writeSpanTag); + this.spanTags = new OtlpSpanAttributes.Writer<>(writer, OtlpTraceJson::writeSpanTag); } /** Call this to ensure process tags are written out for the 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 fa8c6b7a548..fd1f593c4db 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 @@ -264,14 +264,14 @@ private static void writeSpanTag(StreamingBuffer buf, UTF8BytesString key, long public static class MetaWriter implements MetadataConsumer { private final StreamingBuffer buf; - private final OtlpSpanTags spanTags; + private final OtlpSpanAttributes.Writer spanTags; private boolean includeProcessTags; private boolean includeSamplingTags; public MetaWriter(StreamingBuffer buf) { this.buf = buf; - this.spanTags = new OtlpSpanTags<>(buf, OtlpTraceProto::writeSpanTag); + this.spanTags = new OtlpSpanAttributes.Writer<>(buf, OtlpTraceProto::writeSpanTag); } /** Call this to ensure process tags are written out for the next span. */ diff --git a/dd-trace-core/src/test/java/datadog/trace/core/otlp/trace/OtlpSpanTagsTest.java b/dd-trace-core/src/test/java/datadog/trace/core/otlp/trace/OtlpSpanAttributesTest.java similarity index 99% rename from dd-trace-core/src/test/java/datadog/trace/core/otlp/trace/OtlpSpanTagsTest.java rename to dd-trace-core/src/test/java/datadog/trace/core/otlp/trace/OtlpSpanAttributesTest.java index f5905f4e1c1..56400716462 100644 --- a/dd-trace-core/src/test/java/datadog/trace/core/otlp/trace/OtlpSpanTagsTest.java +++ b/dd-trace-core/src/test/java/datadog/trace/core/otlp/trace/OtlpSpanAttributesTest.java @@ -30,7 +30,7 @@ */ @WithConfig(key = GeneralConfig.TAGS, value = "team:core,region:us1") @WithConfig(key = TracerConfig.SPAN_TAGS, value = "owner:alice,tier:gold") -class OtlpSpanTagsTest { +class OtlpSpanAttributesTest { private CoreTracer tracer; From 5142b32f7d98faf26c6a357162629721c5169383 Mon Sep 17 00:00:00 2001 From: Matthew Li Date: Wed, 28 Oct 2026 15:17:12 -0400 Subject: [PATCH 2/2] Keep traces with different global tags in one OTLP payload Instead of flushing the payload whenever a trace's global tags differ from the payload's, start a new resource within the same payload. Each resource carries the global tags of its traces, and the collectors keep the current and previous resource cached so traces interleaving old and new tags after a remote config change don't rebuild them each time. Co-Authored-By: Claude Opus 5.5 --- .../common/writer/OtlpPayloadDispatcher.java | 4 - .../core/otlp/trace/OtlpTraceCollector.java | 11 +- .../otlp/trace/OtlpTraceJsonCollector.java | 93 ++++++++++---- .../otlp/trace/OtlpTraceProtoCollector.java | 92 ++++++++++---- .../otlp/trace/OtlpSpanAttributesTest.java | 113 +++++++++++++++--- 5 files changed, 241 insertions(+), 72 deletions(-) diff --git a/dd-trace-core/src/main/java/datadog/trace/common/writer/OtlpPayloadDispatcher.java b/dd-trace-core/src/main/java/datadog/trace/common/writer/OtlpPayloadDispatcher.java index 08246bb22ad..ed69f08c37e 100644 --- a/dd-trace-core/src/main/java/datadog/trace/common/writer/OtlpPayloadDispatcher.java +++ b/dd-trace-core/src/main/java/datadog/trace/common/writer/OtlpPayloadDispatcher.java @@ -26,10 +26,6 @@ final class OtlpPayloadDispatcher implements PayloadDispatcher { @Override public void addTrace(List> trace) { - // a payload carries one resource: start a new one when the trace's global tags differ - if (!collector.sharesResource(trace)) { - flush(); - } collector.addTrace(trace); // flush proactively to keep payload size bounded if (collector.sizeInBytes() >= FLUSH_THRESHOLD_BYTES) { diff --git a/dd-trace-core/src/main/java/datadog/trace/core/otlp/trace/OtlpTraceCollector.java b/dd-trace-core/src/main/java/datadog/trace/core/otlp/trace/OtlpTraceCollector.java index 083b572641b..a082c64752a 100644 --- a/dd-trace-core/src/main/java/datadog/trace/core/otlp/trace/OtlpTraceCollector.java +++ b/dd-trace-core/src/main/java/datadog/trace/core/otlp/trace/OtlpTraceCollector.java @@ -16,7 +16,7 @@ public abstract class OtlpTraceCollector { private static final TagMap CONFIG_GLOBAL_TAGS = TagMap.fromMapImmutable(Config.get().getGlobalTags()); - /** The global tags of the payload being collected; null until its first trace. */ + /** The global tags of the resource being collected; null until the payload's first span. */ protected TagMap payloadGlobalTags; /** Adds spans from the given trace to the collector. */ @@ -28,15 +28,6 @@ public abstract class OtlpTraceCollector { /** Returns the number of bytes buffered since the last collection. */ public abstract int sizeInBytes(); - /** - * Whether the given trace can join the payload being collected. A payload carries the global tags - * once, on its resource, so a trace with different global tags (after a remote config tag change) - * needs a payload of its own. - */ - public final boolean sharesResource(List> spans) { - return payloadGlobalTags == null || globalTagsOf(spans) == payloadGlobalTags; - } - protected final boolean shouldExport(CoreSpan span) { return span.samplingPriority() > 0 // trace-level sampling priority || span.getTag(SPAN_SAMPLING_MECHANISM_TAG) != null; // span-level sampling priority diff --git a/dd-trace-core/src/main/java/datadog/trace/core/otlp/trace/OtlpTraceJsonCollector.java b/dd-trace-core/src/main/java/datadog/trace/core/otlp/trace/OtlpTraceJsonCollector.java index 8ddf62cc472..9be62c62d40 100644 --- a/dd-trace-core/src/main/java/datadog/trace/core/otlp/trace/OtlpTraceJsonCollector.java +++ b/dd-trace-core/src/main/java/datadog/trace/core/otlp/trace/OtlpTraceJsonCollector.java @@ -31,6 +31,9 @@ * one-span lookahead: that decision (and the corresponding {@code MetadataConsumer.accept} call) * can only be made once we see the next span (or reach a scope/payload boundary), since it depends * on whether the held-back span is the last one in its trace. + * + *

Each resource carries the global tags of its traces. A trace whose global tags differ (after a + * remote config tag change) completes the current resource and starts a new one. */ public final class OtlpTraceJsonCollector extends OtlpTraceCollector { @@ -48,19 +51,22 @@ public final class OtlpTraceJsonCollector extends OtlpTraceCollector { private DDSpan currentSpan; private List currentSpanLinks = Collections.emptyList(); - // resource for the last global tags seen; they rarely change, so it's rarely rebuilt - private TagMap resourceGlobalTags; - private String resourceFragment; + // global tags of the trace being added + private TagMap traceGlobalTags; + + // resources for the last two global tags seen, see CachedResource + private CachedResource currentResource; + private CachedResource previousResource; /** Adds the given trace spans to the collector. */ @Override public void addTrace(List> spans) { if (!payloadStarted) { - payloadGlobalTags = globalTagsOf(spans); start(); payloadStarted = true; } + traceGlobalTags = globalTagsOf(spans); try { for (CoreSpan span : spans) { visitSpan(span); @@ -94,25 +100,23 @@ public OtlpPayload collectTraces() { } } - /** Prepare temporary elements to collect trace data. */ + /** + * Prepare temporary elements to collect trace data. Lazily allocate Resource when spans are + * added. + */ private void start() { writer = new JsonWriter(); metaWriter = new OtlpTraceJson.MetaWriter(writer); writer.beginObject(); writer.name("resourceSpans").beginArray(); - writer.beginObject(); - writer.name("resource").jsonValue(resourceFragment()); - writer.name("scopeSpans").beginArray(); - - // for now put all spans under the default scope - visitScopedSpans(DEFAULT_TRACE_SCOPE); } /** Cleanup elements used to collect trace data. */ private void stop() { payloadStarted = false; payloadGlobalTags = null; + traceGlobalTags = null; anySpanWritten = false; writer = null; @@ -124,12 +128,31 @@ private void stop() { } private String resourceFragment() { - // Update resourceGlobalTags when RC updates the DD_TAGS value (reflected by payloadGlobalTags) - if (resourceFragment == null || resourceGlobalTags != payloadGlobalTags) { - resourceFragment = buildTraceResourceFragment(payloadGlobalTags); - resourceGlobalTags = payloadGlobalTags; + // Rebuild when the open resource's tags (payloadGlobalTags) differ from the cached one + if (currentResource == null || currentResource.globalTags != payloadGlobalTags) { + CachedResource replaced = currentResource; + currentResource = + previousResource != null && previousResource.globalTags == payloadGlobalTags + ? previousResource + : new CachedResource( + payloadGlobalTags, buildTraceResourceFragment(payloadGlobalTags)); + previousResource = replaced; } - return resourceFragment; + return currentResource.fragment; + } + + private void visitResource(TagMap globalTags) { + if (currentScope != null) { + completeResource(); + } + payloadGlobalTags = globalTags; + + writer.beginObject(); + writer.name("resource").jsonValue(resourceFragment()); + writer.name("scopeSpans").beginArray(); + + // for now put all spans under the default scope + visitScopedSpans(DEFAULT_TRACE_SCOPE); } private void visitScopedSpans(OtelInstrumentationScope scope) { @@ -148,6 +171,11 @@ private void visitSpan(CoreSpan span) { if (!shouldExport(span)) { return; } + if (currentScope == null || traceGlobalTags != payloadGlobalTags) { + // if payload's first exported span, start new resource/scope. Otherwise (global tags + // changed), close old resource/scope and start new resource/scope + visitResource(traceGlobalTags); + } if (currentSpan != null) { // ensure last span written at trace boundary includes sampling tags if (!span.getTraceId().equals(currentSpan.getTraceId())) { @@ -159,14 +187,12 @@ private void visitSpan(CoreSpan span) { currentSpanLinks = currentSpan.getLinks(); } - // called once we've processed all scopes and span messages + // called once we've processed all resources and span messages private OtlpPayload completePayload() { if (currentScope != null) { - completeScope(); + completeResource(); } - writer.endArray(); // scopeSpans - writer.endObject(); // resourceSpans[0] writer.endArray(); // resourceSpans writer.endObject(); // root @@ -178,6 +204,14 @@ private OtlpPayload completePayload() { return new OtlpPayload(ByteBuffer.wrap(bytes), JSON_CONTENT_TYPE); } + // called once we've processed all spans with the same global tags + private void completeResource() { + completeScope(); + + writer.endArray(); // scopeSpans + writer.endObject(); // resourceSpans[i] + } + // called once we've processed all spans in a specific scope private void completeScope() { if (currentSpan != null) { @@ -213,4 +247,23 @@ private void completeSpan() { "OTLP payload exceeds maximum buffer size of " + MAX_CAPACITY_BYTES + " bytes"); } } + + /** + * A resource built for one global tags map, cached so it's only rebuilt when the tags change. + * + *

A trace takes its global tags from the config snapshot when it starts, so after a remote + * config tag change, traces started before the change (such as long-lived traces still sending + * partial flushes) keep arriving with the old tags, interleaved with new traces. The collector + * keeps the current and the previous resource, so switching back and forth between the two swaps + * them instead of rebuilding each time. + */ + private static final class CachedResource { + final TagMap globalTags; + final String fragment; + + CachedResource(TagMap globalTags, String fragment) { + this.globalTags = globalTags; + this.fragment = fragment; + } + } } diff --git a/dd-trace-core/src/main/java/datadog/trace/core/otlp/trace/OtlpTraceProtoCollector.java b/dd-trace-core/src/main/java/datadog/trace/core/otlp/trace/OtlpTraceProtoCollector.java index 0c5b70ce363..10f0fd72c50 100644 --- a/dd-trace-core/src/main/java/datadog/trace/core/otlp/trace/OtlpTraceProtoCollector.java +++ b/dd-trace-core/src/main/java/datadog/trace/core/otlp/trace/OtlpTraceProtoCollector.java @@ -27,8 +27,9 @@ * we chunk all span-links for a given span. Once the span is complete we add the first part of the * span message and its chunked span-links to the scoped chunks. Once the scope is complete we add * the first part of the scoped spans message and all its chunks (span messages and any span-links) - * to the payload. Once all the span data has been chunked we add the enclosing resource span - * message, carrying the payload's global tags, to the start of the payload. + * to the resource chunks. Once the resource is complete we add the enclosing resource spans + * message, carrying the global tags of its traces, to the payload. A trace whose global tags differ + * (after a remote config tag change) completes the current resource and starts a new one. */ public final class OtlpTraceProtoCollector extends OtlpTraceCollector { @@ -43,25 +44,29 @@ public final class OtlpTraceProtoCollector extends OtlpTraceCollector { // total number of chunked bytes at different nesting levels private int payloadBytes; + private int resourceBytes; private int scopedBytes; private int spanBytes; private OtelInstrumentationScope currentScope; private DDSpan currentSpan; - // resource for the last global tags seen; they rarely change, so it's rarely rebuilt - private TagMap resourceGlobalTags; - private byte[] resourceMessage; + // global tags of the trace being added + private TagMap traceGlobalTags; + + // resources for the last two global tags seen, see CachedResource + private CachedResource currentResource; + private CachedResource previousResource; /** Adds the given trace spans to the collector. */ @Override public void addTrace(List> spans) { if (!payloadStarted) { - payloadGlobalTags = globalTagsOf(spans); start(); payloadStarted = true; } + traceGlobalTags = globalTagsOf(spans); try { // OtlpProtoBuffer collects spans in reverse for (int i = spans.size() - 1; i >= 0; i--) { @@ -97,20 +102,19 @@ public OtlpPayload collectTraces() { private void start() { // remove stale entries from caches OtlpCommonProto.recalibrateCaches(); - - // for now put all spans under the default scope - visitScopedSpans(DEFAULT_TRACE_SCOPE); } /** Cleanup elements used to collect trace data. */ private void stop() { payloadStarted = false; payloadGlobalTags = null; + traceGlobalTags = null; buf.reset(); protobuf.reset(); payloadBytes = 0; + resourceBytes = 0; scopedBytes = 0; spanBytes = 0; @@ -118,6 +122,16 @@ private void stop() { currentSpan = null; } + private void visitResource(TagMap globalTags) { + if (currentScope != null) { + completeResource(); + } + payloadGlobalTags = globalTags; + + // for now put all spans under the default scope + visitScopedSpans(DEFAULT_TRACE_SCOPE); + } + private void visitScopedSpans(OtelInstrumentationScope scope) { if (currentScope != null) { completeScope(); @@ -129,6 +143,10 @@ private void visitSpan(CoreSpan span) { if (!shouldExport(span)) { return; } + if (currentScope == null || traceGlobalTags != payloadGlobalTags) { + // spans with different global tags need a resource of their own + visitResource(traceGlobalTags); + } if (currentSpan != null) { // ensure last span written at trace boundary includes sampling tags // payload buffer is prepending, so last span written appears first! @@ -145,30 +163,43 @@ private void visitSpanLink(AgentSpanLink spanLink) { spanBytes += recordSpanLinkMessage(buf, spanLink, protobuf); } - // called once we've processed all scopes and span messages + // called once we've processed all resources and span messages private OtlpPayload completePayload() { if (currentScope != null) { - completeScope(); + completeResource(); } if (payloadBytes == 0) { return OtlpPayload.EMPTY; } - // prepend the resource chunk, which carries the payload's global tags - payloadBytes += protobuf.recordMessage(resourceMessage()); - - // finally prepend the total length of all collected chunks - protobuf.recordMessage(buf, 1, payloadBytes); return protobuf.toPayload(); } + // called once we've processed all spans with the same global tags + private void completeResource() { + completeScope(); + + // prepend the resource chunk, which carries the global tags + resourceBytes += protobuf.recordMessage(resourceMessage()); + + // then prepend the total length of the resource's chunks + payloadBytes += protobuf.recordMessage(buf, 1, resourceBytes); + + // reset temporary elements for next resource + resourceBytes = 0; + } + private byte[] resourceMessage() { - if (resourceMessage == null || resourceGlobalTags != payloadGlobalTags) { - resourceMessage = buildTraceResourceMessage(payloadGlobalTags); - resourceGlobalTags = payloadGlobalTags; + if (currentResource == null || currentResource.globalTags != payloadGlobalTags) { + CachedResource replaced = currentResource; + currentResource = + previousResource != null && previousResource.globalTags == payloadGlobalTags + ? previousResource + : new CachedResource(payloadGlobalTags, buildTraceResourceMessage(payloadGlobalTags)); + previousResource = replaced; } - return resourceMessage; + return currentResource.message; } // called once we've processed all spans in a specific scope @@ -182,7 +213,7 @@ private void completeScope() { } if (scopedBytes > 0) { - payloadBytes += recordScopedSpansMessage(buf, currentScope, scopedBytes, protobuf); + resourceBytes += recordScopedSpansMessage(buf, currentScope, scopedBytes, protobuf); } // reset temporary elements for next scope @@ -199,4 +230,23 @@ private void completeSpan() { currentSpan = null; spanBytes = 0; } + + /** + * A resource built for one global tags map, cached so it's only rebuilt when the tags change. + * + *

A trace takes its global tags from the config snapshot when it starts, so after a remote + * config tag change, traces started before the change (such as long-lived traces still sending + * partial flushes) keep arriving with the old tags, interleaved with new traces. The collector + * keeps the current and the previous resource, so switching back and forth between the two swaps + * them instead of rebuilding each time. + */ + private static final class CachedResource { + final TagMap globalTags; + final byte[] message; + + CachedResource(TagMap globalTags, byte[] message) { + this.globalTags = globalTags; + this.message = message; + } + } } diff --git a/dd-trace-core/src/test/java/datadog/trace/core/otlp/trace/OtlpSpanAttributesTest.java b/dd-trace-core/src/test/java/datadog/trace/core/otlp/trace/OtlpSpanAttributesTest.java index 56400716462..090ae0a5503 100644 --- a/dd-trace-core/src/test/java/datadog/trace/core/otlp/trace/OtlpSpanAttributesTest.java +++ b/dd-trace-core/src/test/java/datadog/trace/core/otlp/trace/OtlpSpanAttributesTest.java @@ -4,11 +4,13 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNull; -import static org.junit.jupiter.api.Assertions.assertTrue; +import com.google.protobuf.CodedInputStream; +import com.google.protobuf.WireFormat; import datadog.json.JsonMapper; import datadog.trace.api.config.GeneralConfig; import datadog.trace.api.config.TracerConfig; +import datadog.trace.api.sampling.PrioritySampling; import datadog.trace.common.writer.ListWriter; import datadog.trace.core.CoreSpan; import datadog.trace.core.CoreTracer; @@ -17,6 +19,7 @@ import datadog.trace.test.junit.utils.config.WithConfig; import java.io.IOException; import java.nio.charset.StandardCharsets; +import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -27,6 +30,7 @@ /** * End-to-end check that the global tags are exported once, as OTLP resource attributes taken from * the trace's config snapshot, while span defaults and the span's own tags stay on each span. + * Traces with different global tags share a payload under resources of their own. */ @WithConfig(key = GeneralConfig.TAGS, value = "team:core,region:us1") @WithConfig(key = TracerConfig.SPAN_TAGS, value = "owner:alice,tier:gold") @@ -96,22 +100,52 @@ void consumedSpanDefaultIsNotWrittenTwice() throws IOException { } @Test - void traceWithOtherGlobalTagsNeedsItsOwnPayload() { + void traceWithOtherGlobalTagsGetsItsOwnResource() throws IOException { CoreTracer otherTracer = CoreTracer.builder().writer(new ListWriter()).build(); try { - List> trace = finishedTrace(tracer); - List> sameTagsTrace = finishedTrace(tracer); // another tracer's snapshot holds its own global tags map - List> otherTagsTrace = finishedTrace(otherTracer); + List>> traces = + asList( + droppedTrace(otherTracer), // no exported spans, so no resource of its own + finishedTrace(tracer), + finishedTrace(tracer), + finishedTrace(otherTracer), + finishedTrace(tracer)); OtlpTraceJsonCollector collector = new OtlpTraceJsonCollector(); - assertTrue(collector.sharesResource(trace)); // empty payload takes any trace - collector.addTrace(trace); - assertTrue(collector.sharesResource(sameTagsTrace)); - assertFalse(collector.sharesResource(otherTagsTrace)); + traces.forEach(collector::addTrace); + List> resourceSpans = allResourceSpans(collector.collectTraces()); + + assertEquals(3, resourceSpans.size()); + assertEquals(2, spans(resourceSpans.get(0)).size()); + assertEquals(1, spans(resourceSpans.get(1)).size()); + assertEquals(1, spans(resourceSpans.get(2)).size()); + for (Map resource : resourceSpans) { + assertEquals("core", attributes(resource(resource)).get("team")); + } + } finally { + otherTracer.close(); + } + } - collector.collectTraces(); - assertTrue(collector.sharesResource(otherTagsTrace)); // a new payload takes it + @Test + void traceWithOtherGlobalTagsGetsItsOwnProtoResource() throws IOException { + CoreTracer otherTracer = CoreTracer.builder().writer(new ListWriter()).build(); + try { + List>> traces = + asList( + droppedTrace(otherTracer), // no exported spans, so no resource of its own + finishedTrace(tracer), + finishedTrace(tracer), + finishedTrace(otherTracer), + finishedTrace(tracer)); + + OtlpTraceProtoCollector collector = new OtlpTraceProtoCollector(); + traces.forEach(collector::addTrace); + OtlpPayload payload = collector.collectTraces(); + + // the protobuf payload is prepended, so its resources come out last first + assertEquals(asList(1, 1, 2), protoSpansPerResource(payload)); } finally { otherTracer.close(); } @@ -129,14 +163,54 @@ private static List> finishedTrace(CoreTracer tracer) { return asList((CoreSpan) span); } - @SuppressWarnings("unchecked") + private static List> droppedTrace(CoreTracer tracer) { + DDSpan span = (DDSpan) tracer.startSpan("test", "op"); + span.setSamplingPriority(PrioritySampling.USER_DROP); + span.finish(); + return asList((CoreSpan) span); + } + private static Map onlyResourceSpans(OtlpPayload payload) throws IOException { + List> resourceSpans = allResourceSpans(payload); + assertEquals(1, resourceSpans.size()); + return resourceSpans.get(0); + } + + @SuppressWarnings("unchecked") + private static List> allResourceSpans(OtlpPayload payload) + throws IOException { byte[] bytes = new byte[payload.getContentLength()]; payload.getContent().get(bytes); Map root = JsonMapper.fromJsonToMap(new String(bytes, StandardCharsets.UTF_8)); - List resourceSpans = (List) root.get("resourceSpans"); - assertEquals(1, resourceSpans.size()); - return (Map) resourceSpans.get(0); + return (List>) root.get("resourceSpans"); + } + + /** Counts the spans in each TracesData.resource_spans entry of a protobuf payload. */ + private static List protoSpansPerResource(OtlpPayload payload) throws IOException { + List counts = new ArrayList<>(); + CodedInputStream tracesData = CodedInputStream.newInstance(payload.getContent()); + while (!tracesData.isAtEnd()) { + assertEquals(1, WireFormat.getTagFieldNumber(tracesData.readTag())); // resource_spans + CodedInputStream resourceSpans = tracesData.readBytes().newCodedInput(); + int count = 0; + while (!resourceSpans.isAtEnd()) { + int tag = resourceSpans.readTag(); + if (WireFormat.getTagFieldNumber(tag) != 2) { // scope_spans + resourceSpans.skipField(tag); + continue; + } + CodedInputStream scopeSpans = resourceSpans.readBytes().newCodedInput(); + while (!scopeSpans.isAtEnd()) { + int scopeTag = scopeSpans.readTag(); + if (WireFormat.getTagFieldNumber(scopeTag) == 2) { // spans + count++; + } + scopeSpans.skipField(scopeTag); + } + } + counts.add(count); + } + return counts; } @SuppressWarnings("unchecked") @@ -146,12 +220,17 @@ private static Map resource(Map resourceSpans) { @SuppressWarnings("unchecked") private static Map onlySpan(Map resourceSpans) { - List scopeSpans = (List) resourceSpans.get("scopeSpans"); - List spans = (List) ((Map) scopeSpans.get(0)).get("spans"); + List spans = spans(resourceSpans); assertEquals(1, spans.size()); return (Map) spans.get(0); } + @SuppressWarnings("unchecked") + private static List spans(Map resourceSpans) { + List scopeSpans = (List) resourceSpans.get("scopeSpans"); + return (List) ((Map) scopeSpans.get(0)).get("spans"); + } + /** Flattens OTLP JSON {@code KeyValue} attributes, failing on duplicate keys. */ @SuppressWarnings("unchecked") private static Map attributes(Map owner) {