Emit additional metric tags on the OTLP stats export path

The native SerializingMetricWriter path already carries user-configured
span-derived additional metric tags (DD_TRACE_STATS_ADDITIONAL_TAGS), but
the OTLP export path dropped them twice: ClientStatsAggregator built the
OTLP writer through the schema-less (EMPTY) constructor, and
OtlpStatsMetricWriter never read entry.getAdditionalTags().

Route the OTLP aggregator ctor through the schema-carrying ctor (capture is
already unconditional once the schema is real) and split each packed
"key:value" additional tag into a plain OTLP string attribute keyed by the
tag name, in both default and otel-semantics modes.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: dougqh

dd-trace-core/src/main/java/datadog/trace/common/metrics/ClientStatsAggregator.java

@@ -133,11 +133,7 @@ public ClientStatsAggregator(
     this(
         config.getWellKnownTags(),
         config.getMetricsIgnoredResources(),
-        AdditionalTagsSchema.from(
-            config.getTraceStatsAdditionalTags(),
-            config.getTraceStatsCardinalityLimit(
-                "additional_tags", MetricCardinalityLimits.ADDITIONAL_TAG_VALUE),
-            MetricCardinalityLimits.USE_BLOCKED_SENTINEL),
+        additionalTagsSchemaFrom(config),
         sharedCommunicationObjects.featuresDiscovery(config),
         healthMetrics,
         new OkHttpSink(
@@ -165,6 +161,7 @@ public ClientStatsAggregator(
       OtlpStatsMetricWriter metricWriter) {
     this(
         config.getMetricsIgnoredResources(),
+        additionalTagsSchemaFrom(config),
         sharedCommunicationObjects.featuresDiscovery(config),
         healthMetrics,
         NoOpSink.INSTANCE,
@@ -176,6 +173,14 @@ public ClientStatsAggregator(
         true);
   }
 
+  private static AdditionalTagsSchema additionalTagsSchemaFrom(Config config) {
+    return AdditionalTagsSchema.from(
+        config.getTraceStatsAdditionalTags(),
+        config.getTraceStatsCardinalityLimit(
+            "additional_tags", MetricCardinalityLimits.ADDITIONAL_TAG_VALUE),
+        MetricCardinalityLimits.USE_BLOCKED_SENTINEL);
+  }
+
   ClientStatsAggregator(
       WellKnownTags wellKnownTags,
       Set<String> ignoredResources,

dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpStatsMetricWriter.java

@@ -228,6 +228,13 @@ private void emitDataPointAttributes(
     if (entry.hasGrpcStatusCode()) {
       emitStringAttribute(metric, RPC_RESPONSE_STATUS_CODE, entry.getGrpcStatusCode());
     }
+    // Additional metric tags: user-configured span-derived dimensions, carried as packed
+    // "key:value" UTF8 strings in schema order. Emitted in both modes as plain OTLP string
+    // attributes keyed by the tag name. NOTE: the attribute-key representation (raw tag name vs a
+    // datadog.* namespace) is an open cross-team question with the OTLP/agent side -- see the PR.
+    for (UTF8BytesString additionalTag : entry.getAdditionalTags()) {
+      emitAdditionalTag(metric, additionalTag);
+    }
     // Default (Datadog) mode: emit datadog.* per-point attributes
     if (!otelSemanticsMode) {
       emitStringAttribute(metric, DATADOG_OPERATION_NAME, entry.getOperationName());
@@ -239,6 +246,19 @@ private void emitDataPointAttributes(
     }
   }
 
+  // Splits a packed "key:value" additional-tag string at the first ':' (keys cannot contain ':',
+  // values may) and emits it as an OTLP string attribute. Skips malformed slots (no ':', empty key
+  // or empty value) defensively.
+  private static void emitAdditionalTag(OtlpMetricVisitor metric, UTF8BytesString additionalTag) {
+    String packed = additionalTag.toString();
+    int separator = packed.indexOf(':');
+    if (separator <= 0 || separator == packed.length() - 1) {
+      return;
+    }
+    metric.visitAttribute(
+        STRING_ATTRIBUTE, packed.substring(0, separator), packed.substring(separator + 1));
+  }
+
   // accepts both String literals and UTF8BytesString (both CharSequence); skips null values
   private static void emitStringAttribute(
       OtlpMetricVisitor metric, String key, @Nullable CharSequence value) {

dd-trace-core/src/test/java/datadog/trace/common/metrics/AggregateEntryTestUtils.java

@@ -45,6 +45,43 @@ public static AggregateEntry of(
       @Nullable CharSequence httpMethod,
       @Nullable CharSequence httpEndpoint,
       @Nullable CharSequence grpcStatusCode) {
+    return of(
+        resource,
+        service,
+        operationName,
+        serviceSource,
+        type,
+        httpStatusCode,
+        synthetic,
+        traceRoot,
+        spanKind,
+        peerTags,
+        httpMethod,
+        httpEndpoint,
+        grpcStatusCode,
+        null);
+  }
+
+  /**
+   * Same as {@link #of} but also carries pre-packed {@code "key:value"} additional metric tags (in
+   * schema order), letting the OTLP/serializing writer tests exercise the additional-tags path
+   * without driving a full {@code AggregateTable}/{@code AdditionalTagsSchema} canonicalization.
+   */
+  public static AggregateEntry of(
+      CharSequence resource,
+      CharSequence service,
+      CharSequence operationName,
+      @Nullable CharSequence serviceSource,
+      CharSequence type,
+      int httpStatusCode,
+      boolean synthetic,
+      boolean traceRoot,
+      CharSequence spanKind,
+      @Nullable List<UTF8BytesString> peerTags,
+      @Nullable CharSequence httpMethod,
+      @Nullable CharSequence httpEndpoint,
+      @Nullable CharSequence grpcStatusCode,
+      @Nullable UTF8BytesString[] additionalTags) {
     UTF8BytesString resourceUtf = AggregateEntry.createUtf8(resource);
     UTF8BytesString serviceUtf = AggregateEntry.createUtf8(service);
     UTF8BytesString operationNameUtf = AggregateEntry.createUtf8(operationName);
@@ -56,7 +93,8 @@ public static AggregateEntry of(
     UTF8BytesString grpcUtf = AggregateEntry.createUtf8(grpcStatusCode);
     List<UTF8BytesString> peerTagsList = peerTags == null ? Collections.emptyList() : peerTags;
     UTF8BytesString[] peerTagsArr = peerTagsList.toArray(new UTF8BytesString[0]);
-    UTF8BytesString[] emptyAdditional = new UTF8BytesString[0];
+    UTF8BytesString[] additionalTagsArr =
+        additionalTags == null ? new UTF8BytesString[0] : additionalTags;
     long keyHash =
         AggregateEntry.hashOf(
             resourceUtf,
@@ -73,8 +111,8 @@ public static AggregateEntry of(
             traceRoot,
             peerTagsArr,
             peerTagsArr.length,
-            emptyAdditional,
-            0);
+            additionalTagsArr,
+            additionalTagsArr.length);
     return new AggregateEntry(
         keyHash,
         resourceUtf,
@@ -90,7 +128,7 @@ public static AggregateEntry of(
         synthetic,
         traceRoot,
         peerTagsList,
-        emptyAdditional);
+        additionalTagsArr);
   }
 
   /**

dd-trace-core/src/test/java/datadog/trace/core/otlp/metrics/OtlpStatsMetricWriterTest.java

@@ -13,6 +13,7 @@
 import com.google.protobuf.WireFormat;
 import datadog.metrics.api.Histograms;
 import datadog.metrics.impl.DDSketchHistograms;
+import datadog.trace.bootstrap.instrumentation.api.UTF8BytesString;
 import datadog.trace.common.metrics.AggregateEntry;
 import datadog.trace.common.metrics.AggregateEntryTestUtils;
 import datadog.trace.common.writer.RemoteApi;
@@ -427,6 +428,100 @@ void httpAndGrpcAttributesAppearOnlyWhenSet() throws IOException {
     assertFalse(bareAttrs.containsKey("rpc.response.status_code"));
   }
 
+  @Test
+  void additionalMetricTagsEmittedAsStringAttributes() throws IOException {
+    // Additional tags arrive on the entry pre-packed as "key:value" UTF8 strings in schema order;
+    // the writer splits each at the first ':' and emits it as a plain OTLP string attribute keyed
+    // by the tag name, in both semantics modes.
+    AggregateEntry e =
+        AggregateEntryTestUtils.of(
+            "GET /users",
+            "web",
+            "servlet.request",
+            null,
+            "web",
+            0,
+            false,
+            true,
+            "server",
+            null,
+            null,
+            null,
+            null,
+            new UTF8BytesString[] {
+              UTF8BytesString.create("region:us-east-1"),
+              UTF8BytesString.create("tenant_id:acme:corp")
+            });
+    AggregateEntryTestUtils.recordOk(e, SECONDS.toNanos(1));
+
+    Map<String, Object> attrs = writeAndDecode(false, e).dataPoints.get(0).attributes;
+    assertEquals("us-east-1", attrs.get("region"));
+    // value may itself contain ':' — only the first ':' separates key from value
+    assertEquals("acme:corp", attrs.get("tenant_id"));
+  }
+
+  @Test
+  void additionalMetricTagsEmittedInOtelSemanticsMode() throws IOException {
+    // Unlike datadog.* attributes, additional tags are user-configured dimensions and are emitted
+    // in otel-semantics mode too.
+    AggregateEntry e =
+        AggregateEntryTestUtils.of(
+            "GET /users",
+            "web",
+            "servlet.request",
+            null,
+            "web",
+            0,
+            false,
+            true,
+            "server",
+            null,
+            null,
+            null,
+            null,
+            new UTF8BytesString[] {UTF8BytesString.create("region:us-east-1")});
+    AggregateEntryTestUtils.recordOk(e, SECONDS.toNanos(1));
+
+    Map<String, Object> attrs = writeAndDecode(true, e).dataPoints.get(0).attributes;
+    assertEquals("us-east-1", attrs.get("region"));
+    assertFalse(
+        attrs.containsKey("datadog.operation.name"), "datadog.* still absent in otel-semantics");
+  }
+
+  @Test
+  void malformedAdditionalTagsAreSkipped() throws IOException {
+    // Defensive: a slot with no ':', an empty key, or an empty value is dropped rather than emitted
+    // as a malformed attribute.
+    AggregateEntry e =
+        AggregateEntryTestUtils.of(
+            "GET /users",
+            "web",
+            "servlet.request",
+            null,
+            "web",
+            0,
+            false,
+            true,
+            "server",
+            null,
+            null,
+            null,
+            null,
+            new UTF8BytesString[] {
+              UTF8BytesString.create("noseparator"),
+              UTF8BytesString.create(":emptykey"),
+              UTF8BytesString.create("emptyvalue:"),
+              UTF8BytesString.create("region:us-east-1")
+            });
+    AggregateEntryTestUtils.recordOk(e, SECONDS.toNanos(1));
+
+    Map<String, Object> attrs = writeAndDecode(false, e).dataPoints.get(0).attributes;
+    assertEquals("us-east-1", attrs.get("region"), "well-formed tag still emitted");
+    assertFalse(attrs.containsKey("noseparator"), "no-separator slot skipped");
+    assertFalse(attrs.containsKey(""), "empty-key slot skipped");
+    assertFalse(attrs.containsKey("emptyvalue"), "empty-value slot skipped");
+  }
+
   @Test
   void serviceNameEmittedOnlyForNonDefaultService() throws IOException {
     CapturingSender sender = new CapturingSender();