Write global tags once on the OTLP trace resource

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

@@ -26,6 +26,10 @@ final class OtlpPayloadDispatcher implements PayloadDispatcher {
 
   @Override
   public void addTrace(List<? extends CoreSpan<?>> 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) {

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

@@ -876,6 +876,10 @@ public TraceCollector getTraceCollector() {
     return traceCollector;
   }
 
+  public TagMap getGlobalTags() {
+    return traceCollector.getTraceConfig().globalTags;
+  }
+
   public RequestContext getRequestContext() {
     return this;
   }

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

@@ -6,6 +6,7 @@
 
 import datadog.trace.api.Config;
 import datadog.trace.api.ProcessTags;
+import datadog.trace.api.TagMap;
 import java.util.HashSet;
 import java.util.LinkedHashMap;
 import java.util.List;
@@ -43,9 +44,17 @@ private OtlpResourceAttributes() {}
 
   /**
    * {@code value} is a {@link String}, except {@code datadog.process_tags}: a {@code List<String>}.
+   *
+   * <p>{@code globalTags} (none when null) are the trace's from its config snapshot, so the trace
+   * resource follows remote config tag changes, or {@code config.getGlobalTags()} for the others. A
+   * {@link TagMap} holds known tags under their Datadog name, so its tags are named like span
+   * attributes: by their OpenTelemetry name under OTel semantics, else their Datadog name.
    */
   static void visitResourceAttributes(
-      Config config, Map<String, Object> extraAttributes, BiConsumer<String, Object> visitor) {
+      Config config,
+      Map<String, ?> globalTags,
+      Map<String, Object> extraAttributes,
+      BiConsumer<String, Object> visitor) {
     String serviceName = config.getServiceName();
     String env = config.getEnv();
     String version = config.getVersion();
@@ -67,20 +76,36 @@ static void visitResourceAttributes(
     visitor.accept("telemetry.sdk.version", TRACER_VERSION);
     visitor.accept("telemetry.sdk.language", "java");
 
-    config
-        .getGlobalTags()
-        .forEach(
-            (key, value) -> {
-              // ignore global tags replaced by canonical or extra resource attributes
-              if (!IGNORED_GLOBAL_TAGS.contains(key.toLowerCase(Locale.ROOT))
-                  && !extraAttributes.containsKey(key)) {
-                visitor.accept(key, value);
-              }
-            });
+    if (globalTags instanceof TagMap) {
+      boolean otelSemantics = config.isTraceOtelSemanticsEnabled();
+      ((TagMap) globalTags)
+          .forEach(
+              entry ->
+                  visitGlobalTag(
+                      otelSemantics ? entry.openTelemetryTag() : entry.tag(),
+                      entry.stringValue(),
+                      extraAttributes,
+                      visitor));
+    } else if (globalTags != null) {
+      globalTags.forEach(
+          (key, value) -> visitGlobalTag(key, String.valueOf(value), extraAttributes, visitor));
+    }
 
     extraAttributes.forEach(visitor);
   }
 
+  private static void visitGlobalTag(
+      String key,
+      String value,
+      Map<String, Object> extraAttributes,
+      BiConsumer<String, Object> visitor) {
+    // ignore global tags replaced by canonical or extra resource attributes
+    if (!IGNORED_GLOBAL_TAGS.contains(key.toLowerCase(Locale.ROOT))
+        && !extraAttributes.containsKey(key)) {
+      visitor.accept(key, value);
+    }
+  }
+
   private static final String PROCESS_TAGS_KEY = DATADOG_PREFIX + "process_tags";
 
   /**

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

@@ -28,21 +28,28 @@ private OtlpResourceJson() {}
   public static final String RESOURCE_FRAGMENT_WITH_DATADOG_ATTRS =
       buildResourceFragment(Config.get(), datadogResourceAttributes(Config.get()));
 
-  /**
-   * Resource used by the OTLP trace export. Identical to {@link #RESOURCE_FRAGMENT} but adds 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.
-   */
-  public static final String TRACE_RESOURCE_FRAGMENT =
-      buildResourceFragment(Config.get(), traceResourceAttributes(Config.get()));
+  private static final Map<String, Object> TRACE_RESOURCE_ATTRIBUTES =
+      traceResourceAttributes(Config.get());
+
+  public static String buildTraceResourceFragment(Map<String, ?> globalTags) {
+    return buildResourceFragment(Config.get(), globalTags, TRACE_RESOURCE_ATTRIBUTES);
+  }
 
   static String buildResourceFragment(Config config, Map<String, Object> extraAttributes) {
+    return buildResourceFragment(config, config.getGlobalTags(), extraAttributes);
+  }
+
+  static String buildResourceFragment(
+      Config config, Map<String, ?> globalTags, Map<String, Object> extraAttributes) {
     try (JsonWriter writer = new JsonWriter()) {
       writer.beginObject();
       writer.name("attributes").beginArray();
 
       visitResourceAttributes(
-          config, extraAttributes, (key, value) -> writeResourceAttribute(writer, key, value));
+          config,
+          globalTags,
+          extraAttributes,
+          (key, value) -> writeResourceAttribute(writer, key, value));
 
       writer.endArray();
       writer.endObject();

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

@@ -31,19 +31,32 @@ private OtlpResourceProto() {}
   public static final byte[] RESOURCE_MESSAGE_WITH_DATADOG_ATTRS =
       buildResourceMessage(Config.get(), datadogResourceAttributes(Config.get()));
 
+  private static final Map<String, Object> TRACE_RESOURCE_ATTRIBUTES =
+      traceResourceAttributes(Config.get());
+
   /**
-   * Resource used by the OTLP trace export. Identical to {@link #RESOURCE_MESSAGE} but adds 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.
+   * Builds the resource used by the OTLP trace export, carrying the trace's {@code globalTags}.
+   * Like {@link #RESOURCE_MESSAGE} but adds 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.
    */
-  public static final byte[] TRACE_RESOURCE_MESSAGE =
-      buildResourceMessage(Config.get(), traceResourceAttributes(Config.get()));
+  public static byte[] buildTraceResourceMessage(Map<String, ?> globalTags) {
+    return buildResourceMessage(Config.get(), globalTags, TRACE_RESOURCE_ATTRIBUTES);
+  }
 
   static byte[] buildResourceMessage(Config config, Map<String, Object> extraAttributes) {
+    return buildResourceMessage(config, config.getGlobalTags(), extraAttributes);
+  }
+
+  static byte[] buildResourceMessage(
+      Config config, Map<String, ?> globalTags, Map<String, Object> extraAttributes) {
     GrowableBuffer buf = new GrowableBuffer(512);
 
     visitResourceAttributes(
-        config, extraAttributes, (key, value) -> writeResourceAttribute(buf, key, value));
+        config,
+        globalTags,
+        extraAttributes,
+        (key, value) -> writeResourceAttribute(buf, key, value));
 
     OtlpProtoBuffer protobuf = new OtlpProtoBuffer(buf.capacity());
     int numBytes = protobuf.recordMessage(buf, 1);

dd-trace-core/src/main/java/datadog/trace/core/otlp/trace/OtlpSpanTags.java

@@ -0,0 +1,67 @@
+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.
+ *
+ * <p>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.
+ *
+ * <p>Not thread-safe: each writer owns one.
+ */
+final class OtlpSpanTags<T> {
+  private final T target;
+  private final TriConsumer<T, Integer, TagMap.EntryReader> tagWriter;
+
+  // per-span state
+  private Integer direction;
+  private TagMap spanTags;
+  private TagMap spanDefaultTags;
+  private final List<String> overriddenDefaults = new ArrayList<>(4);
+
+  OtlpSpanTags(T target, TriConsumer<T, Integer, TagMap.EntryReader> 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);
+    }
+  }
+}

dd-trace-core/src/main/java/datadog/trace/core/otlp/trace/OtlpTraceCollector.java

@@ -2,13 +2,23 @@
 
 import static datadog.trace.core.DDSpanContext.SPAN_SAMPLING_MECHANISM_TAG;
 
+import datadog.trace.api.Config;
+import datadog.trace.api.TagMap;
 import datadog.trace.core.CoreSpan;
+import datadog.trace.core.DDSpan;
 import datadog.trace.core.otlp.common.OtlpPayload;
 import java.util.List;
 
 /** Collects traces ready for export. */
 public abstract class OtlpTraceCollector {
 
+  /** The startup global tags, for a trace whose config snapshot can't be read. */
+  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. */
+  protected TagMap payloadGlobalTags;
+
   /** Adds spans from the given trace to the collector. */
   public abstract void addTrace(List<? extends CoreSpan<?>> spans);
 
@@ -18,8 +28,28 @@ 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<? extends CoreSpan<?>> 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
   }
+
+  /**
+   * The trace's global tags, shared by all its spans; the startup global tags when its config
+   * snapshot can't be read.
+   */
+  protected static TagMap globalTagsOf(List<? extends CoreSpan<?>> spans) {
+    CoreSpan<?> first = spans.isEmpty() ? null : spans.get(0);
+    return first instanceof DDSpan
+        ? ((DDSpan) first).spanContext().getGlobalTags()
+        : CONFIG_GLOBAL_TAGS;
+  }
 }

dd-trace-core/src/main/java/datadog/trace/core/otlp/trace/OtlpTraceJson.java

@@ -186,12 +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<JsonWriter> spanTags;
 
     private boolean includeProcessTags;
     private boolean includeSamplingTags;
 
     public MetaWriter(JsonWriter writer) {
       this.writer = writer;
+      this.spanTags = new OtlpSpanTags<>(writer, OtlpTraceJson::writeSpanTag);
     }
 
     /** Call this to ensure process tags are written out for the next span. */
@@ -258,7 +260,8 @@ public void accept(Metadata metadata) {
         writeAttribute(writer, STRING_ATTRIBUTE, entry.getKey(), entry.getValue());
       }
 
-      tags.forEach(writer, direction, OtlpTraceJson::writeSpanTag);
+      // global tags are left to the resource
+      spanTags.write(metadata, direction);
 
       // reset for next span
       includeProcessTags = false;

dd-trace-core/src/main/java/datadog/trace/core/otlp/trace/OtlpTraceJsonCollector.java

@@ -3,10 +3,11 @@
 import static datadog.trace.core.otlp.common.OtlpCommonJson.writeScopeAndSchema;
 import static datadog.trace.core.otlp.common.OtlpPayload.JSON_CONTENT_TYPE;
 import static datadog.trace.core.otlp.common.OtlpProtoBuffer.MAX_CAPACITY_BYTES;
-import static datadog.trace.core.otlp.common.OtlpResourceJson.TRACE_RESOURCE_FRAGMENT;
+import static datadog.trace.core.otlp.common.OtlpResourceJson.buildTraceResourceFragment;
 import static datadog.trace.core.otlp.trace.OtlpTraceJson.writeSpan;
 
 import datadog.json.JsonWriter;
+import datadog.trace.api.TagMap;
 import datadog.trace.bootstrap.instrumentation.api.AgentSpanLink;
 import datadog.trace.bootstrap.otel.common.OtelInstrumentationScope;
 import datadog.trace.core.CoreSpan;
@@ -47,10 +48,15 @@ public final class OtlpTraceJsonCollector extends OtlpTraceCollector {
   private DDSpan currentSpan;
   private List<? extends AgentSpanLink> currentSpanLinks = Collections.emptyList();
 
+  // resource for the last global tags seen; they rarely change, so it's rarely rebuilt
+  private TagMap resourceGlobalTags;
+  private String resourceFragment;
+
   /** Adds the given trace spans to the collector. */
   @Override
   public void addTrace(List<? extends CoreSpan<?>> spans) {
     if (!payloadStarted) {
+      payloadGlobalTags = globalTagsOf(spans);
       start();
       payloadStarted = true;
     }
@@ -96,7 +102,7 @@ private void start() {
     writer.beginObject();
     writer.name("resourceSpans").beginArray();
     writer.beginObject();
-    writer.name("resource").jsonValue(TRACE_RESOURCE_FRAGMENT);
+    writer.name("resource").jsonValue(resourceFragment());
     writer.name("scopeSpans").beginArray();
 
     // for now put all spans under the default scope
@@ -106,6 +112,7 @@ private void start() {
   /** Cleanup elements used to collect trace data. */
   private void stop() {
     payloadStarted = false;
+    payloadGlobalTags = null;
     anySpanWritten = false;
 
     writer = null;
@@ -116,6 +123,15 @@ private void stop() {
     currentSpanLinks = Collections.emptyList();
   }
 
+  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;
+    }
+    return resourceFragment;
+  }
+
   private void visitScopedSpans(OtelInstrumentationScope scope) {
     if (currentScope != null) {
       completeScope();