Expose the trace's global tags and span defaults to serializers (OTLP trace/resource-level split)

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

@@ -1529,11 +1529,14 @@ void processTagsAndBaggage(
                 : serializedPropagationTags.createTagMap();
       }
 
+      CoreTracer.ConfigSnapshot traceConfig = traceCollector.getTraceConfig();
       consumer.accept(
           new Metadata(
               threadId,
               threadName,
               unsafeTags,
+              traceConfig.globalTags,
+              traceConfig.spanDefaultTags,
               baggageItemsWithPropagationTags,
               getSamplingPriority(),
               measured,

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

@@ -14,6 +14,8 @@ public final class Metadata {
   private final UTF8BytesString threadName;
   private final int httpStatusCode;
   private final TagMap tags;
+  private final TagMap globalTags;
+  private final TagMap spanDefaultTags;
   private final Map<String, String> baggage;
 
   private final int samplingPriority;
@@ -30,6 +32,8 @@ public Metadata(
       long threadId,
       UTF8BytesString threadName,
       TagMap tags,
+      TagMap globalTags,
+      TagMap spanDefaultTags,
       Map<String, String> baggage,
       int samplingPriority,
       boolean measured,
@@ -45,6 +49,8 @@ public Metadata(
     this.threadName = threadName;
     this.httpStatusCode = httpStatusCode;
     this.tags = tags;
+    this.globalTags = globalTags;
+    this.spanDefaultTags = spanDefaultTags;
     this.baggage = baggage;
     this.samplingPriority = samplingPriority;
     this.measured = measured;
@@ -98,6 +104,23 @@ public TagMap getTags() {
     return this.tags;
   }
 
+  /**
+   * The trace's global tags ({@code DD_TAGS}, or their remote-config override). {@link #getTags()}
+   * already includes them, below the span defaults and the span's own tags; this lets a serializer
+   * that writes them once per payload, such as an OTLP resource, tell them apart.
+   */
+  public TagMap getGlobalTags() {
+    return globalTags;
+  }
+
+  /**
+   * The trace's span defaults ({@code DD_SPAN_TAGS} and tracer-defined tags such as {@code env}),
+   * without the global tags. {@link #getTags()} already includes them, below the span's own tags.
+   */
+  public TagMap getSpanDefaultTags() {
+    return spanDefaultTags;
+  }
+
   public Map<String, String> getBaggage() {
     return baggage;
   }

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

@@ -307,6 +307,8 @@ private static CoreSpan<?> mockSpan(CharSequence type, Map<String, Object> tags)
             Thread.currentThread().getId(),
             null,
             TagMap.fromMap(tags),
+            TagMap.EMPTY,
+            TagMap.EMPTY,
             Collections.<String, String>emptyMap(),
             0,
             false,

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

@@ -181,6 +181,12 @@ void testTraceAndSpanCountsAreResetAfterAccess() {
   }
 
   DDSpan realSpan() throws Exception {
+    // A real config snapshot, since serialization reads the trace's tag levels from it. Its type
+    // isn't visible from this package, so it's held as an Object and matched by name below.
+    CoreTracer tracer = CoreTracer.builder().writer(new ListWriter()).build();
+    Object traceConfig = tracer.captureTraceConfig();
+    tracer.close();
+
     // getTracer() and mapServiceName() are package-private in TraceCollector; use a custom
     // Answer to handle them at runtime without compile-time accessibility issues
     PendingTrace trace =
@@ -193,6 +199,9 @@ DDSpan realSpan() throws Exception {
                 // behavior)
                 return mock(CoreTracer.class);
               }
+              if (returnType.getSimpleName().equals("ConfigSnapshot")) {
+                return traceConfig;
+              }
               if (returnType == String.class) {
                 Object[] args = invocation.getArguments();
                 // mapServiceName(String) - return the argument unchanged

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

@@ -238,6 +238,8 @@ public PojoSpan(
               Thread.currentThread().getId(),
               UTF8BytesString.create(Thread.currentThread().getName()),
               TagMap.fromMap(tags),
+              TagMap.EMPTY,
+              TagMap.EMPTY,
               baggage,
               samplingPriority,
               measured,

dd-trace-core/src/test/java/datadog/trace/core/CoreTracerTest.java

@@ -6,7 +6,9 @@
 import static org.junit.jupiter.api.Assertions.assertInstanceOf;
 import static org.junit.jupiter.api.Assertions.assertNotEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNotSame;
 import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assumptions.assumeFalse;
 import static org.mockito.ArgumentMatchers.any;
@@ -722,6 +724,39 @@ void configSnapshotSeparatesGlobalTagsFromSpanDefaults() throws Exception {
     }
   }
 
+  /**
+   * Serializers get the global tags and span defaults of the snapshot the trace started with, so a
+   * config change mid-trace can't split one trace across two sets.
+   */
+  @Test
+  void metadataCarriesTheTraceConfigTagLevels() {
+    Properties properties = new Properties();
+    properties.setProperty(GeneralConfig.TAGS, "global_key:global_val");
+    properties.setProperty(TracerConfig.SPAN_TAGS, "span_key:span_val");
+    CoreTracer tracer = CoreTracer.builder().withProperties(properties).build();
+    unclosedTracers.add(tracer);
+
+    try {
+      ConfigSnapshot traceStartConfig = tracer.captureTraceConfig();
+      DDSpan span = (DDSpan) tracer.buildSpan("datadog", "my_span").start();
+
+      Properties changed = new Properties();
+      changed.setProperty(GeneralConfig.TAGS, "other_key:other_val");
+      tracer.rebuildTraceConfig(Config.get(changed));
+      assertNotSame(traceStartConfig.globalTags, tracer.captureTraceConfig().globalTags);
+
+      Metadata[] captured = {null};
+      span.processTagsAndBaggage(metadata -> captured[0] = metadata, 0);
+
+      assertSame(traceStartConfig.globalTags, captured[0].getGlobalTags());
+      assertSame(traceStartConfig.spanDefaultTags, captured[0].getSpanDefaultTags());
+      assertEquals(buildStringMap("global_key", "global_val"), captured[0].getGlobalTags());
+      assertEquals(buildStringMap("span_key", "span_val"), captured[0].getSpanDefaultTags());
+    } finally {
+      tracer.close();
+    }
+  }
+
   static final String ACTION_JSON =
       "'{\"action\": \"enable\", \"lib_config\":"
           + "{\"tracing_sampling_rate\": null,"

dd-trace-core/src/test/java/datadog/trace/core/MetadataTest.java

@@ -54,6 +54,8 @@ private static Metadata metadataWithStatus(int status) {
         Thread.currentThread().getId(),
         UTF8BytesString.create("main"),
         TagMap.fromMap(emptyMap()),
+        TagMap.EMPTY,
+        TagMap.EMPTY,
         emptyMap(),
         0,
         false,

dd-trace-core/src/traceAgentTest/java/TraceGenerator.java

@@ -175,6 +175,8 @@ static class PojoSpan implements CoreSpan<PojoSpan> {
               currentThread().getId(),
               UTF8BytesString.create(currentThread().getName()),
               fromMap(tags),
+              TagMap.EMPTY,
+              TagMap.EMPTY,
               baggage,
               UNSET,
               measured,