Introducing Tags/OTLP Resource Attributes for OTLP Adoption Metrics

communication/src/main/java/datadog/communication/serialization/msgpack/MsgPackWriter.java

@@ -101,6 +101,8 @@ public <T> boolean format(T message, Mapper<T> mapper) {
         }
       }
       buffer.reset();
+      // the buffer is now empty, so drop any mapper state from the rejected message
+      mapper.reset();
       return false;
     }
   }

communication/src/test/java/datadog/communication/serialization/msgpack/MsgPackWriterTest.java

@@ -14,6 +14,7 @@
 import datadog.communication.serialization.Mapper;
 import datadog.communication.serialization.MessageFormatter;
 import datadog.communication.serialization.StreamingBuffer;
+import datadog.communication.serialization.Writable;
 import datadog.trace.bootstrap.instrumentation.api.UTF8BytesString;
 import datadog.trace.util.stacktrace.StackTraceEvent;
 import datadog.trace.util.stacktrace.StackTraceFrame;
@@ -65,6 +66,38 @@ public void testInsertAfterOverflow() {
         packer.format("abcdefghijklmnopqrstuvwxy", mapper), "data fits in buffer after overflow");
   }
 
+  @Test
+  public void testMapperResetWhenOversizedMessageRejectedFromEmptyBuffer() {
+    CountingResetMapper mapper = new CountingResetMapper();
+    MessageFormatter packer = new MsgPackWriter(newBuffer(2 + 25, (messageCount, buffer) -> {}));
+    assertFalse(packer.format("abcdefghijklmnopqrstuvwxyz", mapper));
+    assertEquals(1, mapper.resets);
+  }
+
+  @Test
+  public void testMapperResetWhenOversizedMessageRejectedAfterFlush() {
+    CountingResetMapper mapper = new CountingResetMapper();
+    MessageFormatter packer = new MsgPackWriter(newBuffer(2 + 25, (messageCount, buffer) -> {}));
+    assertTrue(packer.format("abc", mapper));
+    assertFalse(packer.format("abcdefghijklmnopqrstuvwxyz", mapper));
+    // once before the retry, once after the retry is rejected
+    assertEquals(2, mapper.resets);
+  }
+
+  private static final class CountingResetMapper implements Mapper<String> {
+    int resets;
+
+    @Override
+    public void map(String data, Writable writable) {
+      writable.writeString(data, null);
+    }
+
+    @Override
+    public void reset() {
+      resets++;
+    }
+  }
+
   @Test
   public void testFlushOfOverflow() {
     final List<String> flushed = new ArrayList<>();

dd-trace-api/src/main/java/datadog/trace/api/DDTags.java

@@ -105,4 +105,5 @@ public class DDTags {
   public static final String PROCESS_TAGS = "_dd.tags.process";
   public static final String DD_INTEGRATION = "_dd.integration";
   public static final String DD_SVC_SRC = "_dd.svc_src";
+  public static final String SDK_OTLP_EXPORT = "_dd.sdk.otlp_export";
 }

dd-trace-core/src/main/java/datadog/trace/common/writer/ddagent/TraceMapper.java

@@ -12,4 +12,5 @@ public interface TraceMapper extends RemoteMapper {
       UTF8BytesString.create(DDSpanContext.PRIORITY_SAMPLING_KEY);
   static final UTF8BytesString ORIGIN_KEY = UTF8BytesString.create(DDTags.ORIGIN_KEY);
   static final UTF8BytesString PROCESS_TAGS_KEY = UTF8BytesString.create(DDTags.PROCESS_TAGS);
+  static final UTF8BytesString SDK_OTLP_EXPORT_KEY = UTF8BytesString.create(DDTags.SDK_OTLP_EXPORT);
 }

dd-trace-core/src/main/java/datadog/trace/common/writer/ddagent/TraceMapperV0_4.java

@@ -101,12 +101,14 @@ public void accept(Metadata metadata) {
       final boolean writeSamplingPriority =
           firstSpanInTrace || lastSpanInTrace || metadata.topLevel();
       final UTF8BytesString processTags = firstSpanInPayload ? metadata.processTags() : null;
+      final UTF8BytesString otlpExport = firstSpanInPayload ? metadata.otlpExportMarker() : null;
       int metaSize =
           metadata.getBaggage().size()
               + tags.size()
               + (UNSET_STATUS == metadata.getHttpStatusCode() ? 0 : 1)
               + (null == metadata.getOrigin() ? 0 : 1)
               + (null == processTags ? 0 : 1)
+              + (null == otlpExport ? 0 : 1)
               + 1;
       int metricsSize =
           (writeSamplingPriority && metadata.hasSamplingPriority() ? 1 : 0)
@@ -206,6 +208,10 @@ public void accept(Metadata metadata) {
         writable.writeUTF8(PROCESS_TAGS_KEY);
         writable.writeUTF8(processTags);
       }
+      if (otlpExport != null) {
+        writable.writeUTF8(SDK_OTLP_EXPORT_KEY);
+        writable.writeUTF8(otlpExport);
+      }
 
       tags.forEach(
           writable,

dd-trace-core/src/main/java/datadog/trace/common/writer/ddagent/TraceMapperV0_5.java

@@ -35,6 +35,7 @@ public final class TraceMapperV0_5 implements TraceMapper {
   private final GrowableBuffer dictionary;
 
   private final MetaWriter metaWriter = new MetaWriter();
+
   private final int size;
   private boolean firstSpanWritten;
 
@@ -220,6 +221,7 @@ public void accept(Metadata metadata) {
       final boolean writeSamplingPriority =
           firstSpanInTrace || lastSpanInTrace || metadata.topLevel();
       final UTF8BytesString processTags = firstSpanInPayload ? metadata.processTags() : null;
+      final UTF8BytesString otlpExport = firstSpanInPayload ? metadata.otlpExportMarker() : null;
 
       TagMap tags = metadata.getTags();
 
@@ -229,6 +231,7 @@ public void accept(Metadata metadata) {
               + (UNSET_STATUS == metadata.getHttpStatusCode() ? 0 : 1)
               + (null == metadata.getOrigin() ? 0 : 1)
               + (null == processTags ? 0 : 1)
+              + (null == otlpExport ? 0 : 1)
               + 1;
       int metricsSize =
           (writeSamplingPriority && metadata.hasSamplingPriority() ? 1 : 0)
@@ -272,6 +275,10 @@ public void accept(Metadata metadata) {
         writeDictionaryEncoded(writable, PROCESS_TAGS_KEY);
         writeDictionaryEncoded(writable, processTags);
       }
+      if (null != otlpExport) {
+        writeDictionaryEncoded(writable, SDK_OTLP_EXPORT_KEY);
+        writeDictionaryEncoded(writable, otlpExport);
+      }
 
       for (TagMap.EntryReader entry : tags) {
         if (entry.isNumber()) continue;

dd-trace-core/src/main/java/datadog/trace/common/writer/ddagent/TraceMapperV1.java

@@ -665,8 +665,11 @@ private ByteBuffer buildHeader() {
 
     // attributes = 10, a collection of key to value pairs common in all `chunks`
     CharSequence processTags = ProcessTags.getTagsForSerialization();
-    Map<String, Object> tags =
-        processTags != null ? singletonMap(DDTags.PROCESS_TAGS, processTags) : emptyMap();
+    Map<String, Object> tags = new HashMap<>(4);
+    tags.put(DDTags.SDK_OTLP_EXPORT, String.valueOf(cfg.isOtlpTracesExportEnabled()));
+    if (processTags != null) {
+      tags.put(DDTags.PROCESS_TAGS, processTags);
+    }
     encodeAttributes(headerWriter, 10, tags);
 
     // chunks = 11, a list of trace `chunks`, value is written by PayloadV1

dd-trace-core/src/main/java/datadog/trace/core/DDSpanContext.java

@@ -71,6 +71,9 @@ public class DDSpanContext
   public static final String SPAN_SAMPLING_RULE_RATE_TAG = "_dd.span_sampling.rule_rate";
   public static final String SPAN_SAMPLING_MAX_PER_SECOND_TAG = "_dd.span_sampling.max_per_second";
 
+  private static final UTF8BytesString OTLP_EXPORT_TRUE = UTF8BytesString.create("true");
+  private static final UTF8BytesString OTLP_EXPORT_FALSE = UTF8BytesString.create("false");
+
   private static final DDCache<String, UTF8BytesString> THREAD_NAMES =
       DDCaches.newFixedSizeCache(256);
 
@@ -1416,6 +1419,7 @@ void processTagsAndBaggage(
               getOrigin(),
               longRunningVersion,
               ProcessTags.getTagsForSerialization(),
+              Config.get().isOtlpTracesExportEnabled() ? OTLP_EXPORT_TRUE : OTLP_EXPORT_FALSE,
               restrictedSpan.getLinks()));
     }
   }

dd-trace-core/src/main/java/datadog/trace/core/Metadata.java

@@ -23,6 +23,7 @@ public final class Metadata {
   private final CharSequence origin;
   private final int longRunningVersion;
   private final UTF8BytesString processTags;
+  private final UTF8BytesString otlpExportMarker;
   private final List<? extends AgentSpanLink> spanLinks;
 
   public Metadata(
@@ -37,6 +38,7 @@ public Metadata(
       CharSequence origin,
       int longRunningVersion,
       UTF8BytesString processTags,
+      UTF8BytesString otlpExportMarker,
       List<? extends AgentSpanLink> spanLinks) {
     this.threadId = threadId;
     this.threadName = threadName;
@@ -49,6 +51,7 @@ public Metadata(
     this.origin = origin;
     this.longRunningVersion = longRunningVersion;
     this.processTags = processTags;
+    this.otlpExportMarker = otlpExportMarker;
     this.spanLinks = spanLinks == null ? emptyList() : spanLinks;
   }
 
@@ -121,6 +124,10 @@ public UTF8BytesString processTags() {
     return processTags;
   }
 
+  public UTF8BytesString otlpExportMarker() {
+    return otlpExportMarker;
+  }
+
   public List<? extends AgentSpanLink> getSpanLinks() {
     return spanLinks;
   }

dd-trace-core/src/main/java/datadog/trace/core/otlp/common/OtlpResourceAttributes.java

@@ -1,6 +1,7 @@
 package datadog.trace.core.otlp.common;
 
 import static datadog.communication.ddagent.TracerVersion.TRACER_VERSION;
+import static datadog.trace.api.DDTags.SDK_OTLP_EXPORT;
 import static java.util.Arrays.asList;
 
 import datadog.trace.api.Config;
@@ -23,6 +24,8 @@ private OtlpResourceAttributes() {}
   /** Marks that the Agent should not recompute trace metrics from the exported spans. */
   private static final String STATS_COMPUTED_KEY = "_dd.stats_computed";
 
+  private static final String SDK_SEMANTICS_KEY = "datadog.sdk.semantics";
+
   private static final Set<String> IGNORED_GLOBAL_TAGS =
       new HashSet<>(
           asList(
@@ -34,7 +37,9 @@ private OtlpResourceAttributes() {}
               "service.version",
               "telemetry.sdk.name",
               "telemetry.sdk.version",
-              "telemetry.sdk.language"));
+              "telemetry.sdk.language",
+              SDK_SEMANTICS_KEY,
+              SDK_OTLP_EXPORT));
 
   /**
    * {@code value} is a {@link String}, except {@code datadog.process_tags}: a {@code List<String>}.
@@ -81,10 +86,15 @@ static void visitResourceAttributes(
   /**
    * Builds the extra resource attributes for the OTLP trace export: the {@code _dd.stats_computed}
    * marker when the SDK is computing OTLP span metrics, so a downstream Agent does not recompute
-   * them from the exported spans.
+   * them from the exported spans; {@code datadog.sdk.semantics}, recording whether the SDK applied
+   * Datadog or OTel semantics; and {@code _dd.sdk.otlp_export}, which is always {@code "true"} here
+   * because reaching this encoder means the payload is leaving over OTLP.
    */
   static Map<String, Object> traceResourceAttributes(Config config) {
     Map<String, Object> attributes = new LinkedHashMap<>();
+    attributes.put(
+        "datadog.sdk.semantics", config.isTraceOtelSemanticsEnabled() ? "otel" : "datadog");
+    attributes.put(SDK_OTLP_EXPORT, "true");
     if (config.isOtelTracesSpanMetricsEnabled()) {
       attributes.put(STATS_COMPUTED_KEY, "true");
     }

dd-trace-core/src/test/java/datadog/trace/common/writer/DDAgentApiTest.java

@@ -1,5 +1,6 @@
 package datadog.trace.common.writer;
 
+import static datadog.trace.api.DDTags.SDK_OTLP_EXPORT;
 import static datadog.trace.api.ProtocolVersion.V0_5;
 import static java.util.Collections.emptyList;
 import static java.util.Collections.emptyMap;
@@ -252,6 +253,8 @@ void testContentIsSentAsMsgpackServiceSpan() throws IOException {
           && ProcessTags.getTagsForSerialization() != null) {
         meta.put("_dd.tags.process", ProcessTags.getTagsForSerialization().toString());
       }
+      // payload-scoped marker, written on the first span of the first non-empty chunk
+      meta.put(SDK_OTLP_EXPORT, "false");
       Map<String, Object> metrics = new TreeMap<>();
       metrics.put(DDSpanContext.PRIORITY_SAMPLING_KEY, 1);
       metrics.put(InstrumentationTags.DD_TOP_LEVEL.toString(), 1);
@@ -328,6 +331,8 @@ void testContentIsSentAsMsgpackResourceSpan() throws IOException {
           && ProcessTags.getTagsForSerialization() != null) {
         meta.put("_dd.tags.process", ProcessTags.getTagsForSerialization().toString());
       }
+      // payload-scoped marker, written on the first span of the first non-empty chunk
+      meta.put(SDK_OTLP_EXPORT, "false");
       Map<String, Object> metrics = new TreeMap<>();
       metrics.put(DDSpanContext.PRIORITY_SAMPLING_KEY, 1);
       metrics.put(InstrumentationTags.DD_TOP_LEVEL.toString(), 1);

dd-trace-core/src/test/java/datadog/trace/common/writer/DDAgentWriterCombinedTest.java

@@ -269,8 +269,11 @@ void testDefaultBufferSizeFor(String agentVersion) {
 
     TraceMapper mapper =
         agentVersion.equals("v0.5/traces") ? new TraceMapperV0_5() : new TraceMapperV0_4();
-    int traceSize = calculateSize(minimalTrace, mapper);
-    int maxedPayloadTraceCount = (mapper.messageBufferSize() / traceSize);
+    // The first trace of a payload is larger: it carries the payload-scoped _dd.sdk.otlp_export
+    // marker on its first span. Size both cases so the overflow point is exact.
+    int firstTraceSize = calculateSize(minimalTrace, mapper, true);
+    int traceSize = calculateSize(minimalTrace, mapper, false);
+    int maxedPayloadTraceCount = 1 + (mapper.messageBufferSize() - firstTraceSize) / traceSize;
 
     when(discovery.getTraceEndpoint()).thenReturn(agentVersion);
     when(api.sendSerializedTraces(
@@ -824,13 +827,25 @@ void statsdCommFailure() throws Exception {
     healthMetrics.close();
   }
 
-  static int calculateSize(List<DDSpan> trace, TraceMapper mapper) {
+  /**
+   * Serialized size of {@code trace}, either as the first trace of a payload (which carries the
+   * payload-scoped markers on its first span) or as any later trace. Uses a throwaway mapper of the
+   * same kind so the caller's mapper keeps its state.
+   */
+  static int calculateSize(List<DDSpan> trace, TraceMapper mapper, boolean firstInPayload) {
     AtomicInteger size = new AtomicInteger();
     MsgPackWriter packer =
         new MsgPackWriter(
             new FlushingBuffer(
                 1024, (messageCount, buffer) -> size.set(buffer.limit() - buffer.position())));
-    packer.format(trace, mapper);
+    TraceMapper sizingMapper =
+        mapper instanceof TraceMapperV0_5 ? new TraceMapperV0_5() : new TraceMapperV0_4();
+    if (!firstInPayload) {
+      // burn the payload-scoped markers on a throwaway trace
+      packer.format(trace, sizingMapper);
+      packer.flush();
+    }
+    packer.format(trace, sizingMapper);
     packer.flush();
     return size.get();
   }

dd-trace-core/src/test/java/datadog/trace/common/writer/FileBasedPayloadDispatcherTest.java

@@ -313,6 +313,7 @@ private static CoreSpan<?> mockSpan(CharSequence type, Map<String, Object> tags)
             null,
             0,
             null,
+            null,
             null);
     doAnswer(
             inv -> {

dd-trace-core/src/test/java/datadog/trace/common/writer/TraceGenerator.java

@@ -3,6 +3,7 @@
 import static datadog.trace.api.sampling.PrioritySampling.UNSET;
 import static java.util.Collections.emptyList;
 
+import datadog.trace.api.Config;
 import datadog.trace.api.DDSpanId;
 import datadog.trace.api.DDTags;
 import datadog.trace.api.DDTraceId;
@@ -242,6 +243,7 @@ public PojoSpan(
               origin,
               0,
               ProcessTags.getTagsForSerialization(),
+              UTF8BytesString.create(String.valueOf(Config.get().isOtlpTracesExportEnabled())),
               spanLinks);
     }