Fix OTLP trace-metrics attribute compliance gaps

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

@@ -36,9 +36,11 @@ private OtlpResourceAttributes() {}
               "telemetry.sdk.version",
               "telemetry.sdk.language"));
 
-  /** Visits each resource attribute key/value pair with {@code visitor}. */
+  /**
+   * {@code value} is a {@link String}, except {@code datadog.process_tags}: a {@code List<String>}.
+   */
   static void visitResourceAttributes(
-      Config config, Map<String, String> extraAttributes, BiConsumer<String, String> visitor) {
+      Config config, Map<String, Object> extraAttributes, BiConsumer<String, Object> visitor) {
     String serviceName = config.getServiceName();
     String env = config.getEnv();
     String version = config.getVersion();
@@ -64,43 +66,41 @@ static void visitResourceAttributes(
         .getGlobalTags()
         .forEach(
             (key, value) -> {
-              // ignore datadog tags and their otel equivalents that we map above
-              if (!IGNORED_GLOBAL_TAGS.contains(key.toLowerCase(Locale.ROOT))) {
+              // 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);
               }
             });
 
     extraAttributes.forEach(visitor);
   }
 
+  private static final String PROCESS_TAGS_KEY = DATADOG_PREFIX + "process_tags";
+
   /**
    * 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.
    */
-  static Map<String, String> traceResourceAttributes(Config config) {
-    Map<String, String> attributes = new LinkedHashMap<>();
+  static Map<String, Object> traceResourceAttributes(Config config) {
+    Map<String, Object> attributes = new LinkedHashMap<>();
     if (config.isOtelTracesSpanMetricsEnabled()) {
       attributes.put(STATS_COMPUTED_KEY, "true");
     }
     return attributes;
   }
 
-  static Map<String, String> datadogResourceAttributes(Config config) {
-    Map<String, String> attributes = new LinkedHashMap<>();
+  static Map<String, Object> datadogResourceAttributes(Config config) {
+    Map<String, Object> attributes = new LinkedHashMap<>();
     String runtimeId = config.getRuntimeId();
     if (runtimeId != null && !runtimeId.isEmpty()) {
       attributes.put(DATADOG_PREFIX + "runtime_id", runtimeId);
     }
-    // Process tags arrive as "key:value" pairs; emit each as datadog.<key> = value.
+    // Mirrors SerializingMetricWriter's v0.6 ProcessTags shape; keep both in sync if that changes.
     List<String> processTags = ProcessTags.getTagsAsStringList();
-    if (processTags != null) {
-      for (String tag : processTags) {
-        int colon = tag.indexOf(':');
-        if (colon > 0) {
-          attributes.put(DATADOG_PREFIX + tag.substring(0, colon), tag.substring(colon + 1));
-        }
-      }
+    if (processTags != null && !processTags.isEmpty()) {
+      attributes.put(PROCESS_TAGS_KEY, processTags);
     }
     return attributes;
   }

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

@@ -1,5 +1,6 @@
 package datadog.trace.core.otlp.common;
 
+import static datadog.trace.bootstrap.otlp.common.OtlpAttributeVisitor.STRING_ARRAY_ATTRIBUTE;
 import static datadog.trace.bootstrap.otlp.common.OtlpAttributeVisitor.STRING_ATTRIBUTE;
 import static datadog.trace.core.otlp.common.OtlpCommonJson.writeAttribute;
 import static datadog.trace.core.otlp.common.OtlpResourceAttributes.datadogResourceAttributes;
@@ -9,6 +10,7 @@
 import datadog.json.JsonWriter;
 import datadog.trace.api.Config;
 import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 
 /** Provides a canned JSON fragment for OpenTelemetry's "resource.proto" JSON encoding. */
@@ -35,7 +37,7 @@ private OtlpResourceJson() {}
   public static final String TRACE_RESOURCE_FRAGMENT =
       buildResourceFragment(Config.get(), traceResourceAttributes(Config.get()));
 
-  static String buildResourceFragment(Config config, Map<String, String> extraAttributes) {
+  static String buildResourceFragment(Config config, Map<String, Object> extraAttributes) {
     try (JsonWriter writer = new JsonWriter()) {
       writer.beginObject();
       writer.name("attributes").beginArray();
@@ -49,7 +51,14 @@ static String buildResourceFragment(Config config, Map<String, String> extraAttr
     }
   }
 
-  private static void writeResourceAttribute(JsonWriter writer, String key, String value) {
-    writeAttribute(writer, STRING_ATTRIBUTE, key, value);
+  /**
+   * {@code value} is a {@link String}, except {@code datadog.process_tags}: a {@code List<String>}.
+   */
+  private static void writeResourceAttribute(JsonWriter writer, String key, Object value) {
+    if (value instanceof List) {
+      writeAttribute(writer, STRING_ARRAY_ATTRIBUTE, key, value);
+    } else {
+      writeAttribute(writer, STRING_ATTRIBUTE, key, value);
+    }
   }
 }

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

@@ -1,5 +1,6 @@
 package datadog.trace.core.otlp.common;
 
+import static datadog.trace.bootstrap.otlp.common.OtlpAttributeVisitor.STRING_ARRAY_ATTRIBUTE;
 import static datadog.trace.bootstrap.otlp.common.OtlpAttributeVisitor.STRING_ATTRIBUTE;
 import static datadog.trace.core.otlp.common.OtlpCommonProto.LEN_WIRE_TYPE;
 import static datadog.trace.core.otlp.common.OtlpCommonProto.writeAttribute;
@@ -12,6 +13,7 @@
 import datadog.communication.serialization.StreamingBuffer;
 import datadog.trace.api.Config;
 import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 
 /** Provides a canned message for OpenTelemetry's "resource.proto" wire protocol. */
@@ -38,7 +40,7 @@ private OtlpResourceProto() {}
   public static final byte[] TRACE_RESOURCE_MESSAGE =
       buildResourceMessage(Config.get(), traceResourceAttributes(Config.get()));
 
-  static byte[] buildResourceMessage(Config config, Map<String, String> extraAttributes) {
+  static byte[] buildResourceMessage(Config config, Map<String, Object> extraAttributes) {
     GrowableBuffer buf = new GrowableBuffer(512);
 
     visitResourceAttributes(
@@ -52,8 +54,15 @@ static byte[] buildResourceMessage(Config config, Map<String, String> extraAttri
     return resourceMessage;
   }
 
-  private static void writeResourceAttribute(StreamingBuffer buf, String key, String value) {
+  /**
+   * {@code value} is a {@link String}, except {@code datadog.process_tags}: a {@code List<String>}.
+   */
+  private static void writeResourceAttribute(StreamingBuffer buf, String key, Object value) {
     writeTag(buf, 1, LEN_WIRE_TYPE);
-    writeAttribute(buf, STRING_ATTRIBUTE, key, value);
+    if (value instanceof List) {
+      writeAttribute(buf, STRING_ARRAY_ATTRIBUTE, key, value);
+    } else {
+      writeAttribute(buf, STRING_ATTRIBUTE, key, value);
+    }
   }
 }

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

@@ -1,14 +1,17 @@
 package datadog.trace.core.otlp.metrics;
 
 import static datadog.trace.bootstrap.otel.metrics.OtelInstrumentType.HISTOGRAM;
+import static datadog.trace.bootstrap.otlp.common.OtlpAttributeVisitor.BOOLEAN_ATTRIBUTE;
 import static datadog.trace.bootstrap.otlp.common.OtlpAttributeVisitor.LONG_ATTRIBUTE;
+import static datadog.trace.bootstrap.otlp.common.OtlpAttributeVisitor.STRING_ARRAY_ATTRIBUTE;
 import static datadog.trace.bootstrap.otlp.common.OtlpAttributeVisitor.STRING_ATTRIBUTE;
 
 import datadog.metrics.api.Histogram;
 import datadog.trace.api.Config;
 import datadog.trace.api.config.OtlpConfig;
 import datadog.trace.api.telemetry.OtlpTelemetry;
 import datadog.trace.api.time.SystemTimeSource;
+import datadog.trace.bootstrap.instrumentation.api.Tags;
 import datadog.trace.bootstrap.instrumentation.api.UTF8BytesString;
 import datadog.trace.bootstrap.otel.common.OtelInstrumentationScope;
 import datadog.trace.bootstrap.otel.metrics.OtelInstrumentDescriptor;
@@ -49,18 +52,26 @@ public final class OtlpStatsMetricWriter implements MetricWriter {
   private static final String HTTP_ROUTE = "http.route";
   private static final String RPC_RESPONSE_STATUS_CODE = "rpc.response.status_code";
   private static final String STATUS_CODE = "status.code";
-  private static final String STATUS_CODE_ERROR = "ERROR";
+  private static final String STATUS_CODE_OK = "STATUS_CODE_OK";
+  private static final String STATUS_CODE_ERROR = "STATUS_CODE_ERROR";
+  private static final String DATADOG_ATTRIBUTE_PREFIX = "datadog.";
   private static final String DATADOG_OPERATION_NAME = "datadog.operation.name";
   private static final String DATADOG_SPAN_TYPE = "datadog.span.type";
   private static final String DATADOG_SPAN_TOP_LEVEL = "datadog.span.top_level";
+  private static final String DATADOG_IS_TRACE_ROOT = "datadog.is_trace_root";
   private static final String DATADOG_ORIGIN = "datadog.origin";
+  private static final String DATADOG_PEER_TAGS = "datadog.peer_tags";
   private static final String SYNTHETICS_ORIGIN = "synthetics";
 
+  private static final String SPAN_KIND_SERVER = "SPAN_KIND_SERVER";
+  private static final String SPAN_KIND_CLIENT = "SPAN_KIND_CLIENT";
+  private static final String SPAN_KIND_PRODUCER = "SPAN_KIND_PRODUCER";
+  private static final String SPAN_KIND_CONSUMER = "SPAN_KIND_CONSUMER";
+  private static final String SPAN_KIND_INTERNAL = "SPAN_KIND_INTERNAL";
+
   @Nullable private final OtlpSender sender;
   private final boolean otelSemanticsMode;
 
-  @Nullable private final String defaultService;
-
   // own single-thread collector; forced to DELTA since trace-stats buckets are per-interval deltas.
   private final OtlpMetricsCollector collector;
 
@@ -90,25 +101,19 @@ public OtlpStatsMetricWriter(Config config) {
     this(
         OtlpMetricsSenderFactory.create(config),
         config.getOtlpMetricsProtocol(),
-        config.isTraceOtelSemanticsEnabled(),
-        config.getServiceName());
+        config.isTraceOtelSemanticsEnabled());
   }
 
   // visible for testing: lets tests inject a capturing sender to decode the emitted payload and
-  // control the semantics mode and default service
-  OtlpStatsMetricWriter(
-      @Nullable OtlpSender sender, boolean otelSemanticsMode, @Nullable String defaultService) {
-    this(sender, OtlpConfig.Protocol.HTTP_PROTOBUF, otelSemanticsMode, defaultService);
+  // control the semantics mode
+  OtlpStatsMetricWriter(@Nullable OtlpSender sender, boolean otelSemanticsMode) {
+    this(sender, OtlpConfig.Protocol.HTTP_PROTOBUF, otelSemanticsMode);
   }
 
   private OtlpStatsMetricWriter(
-      @Nullable OtlpSender sender,
-      OtlpConfig.Protocol protocol,
-      boolean otelSemanticsMode,
-      @Nullable String defaultService) {
+      @Nullable OtlpSender sender, OtlpConfig.Protocol protocol, boolean otelSemanticsMode) {
     this.sender = sender;
     this.otelSemanticsMode = otelSemanticsMode;
-    this.defaultService = defaultService;
     // Default mode carries datadog.runtime_id / process tags on the Resource; OTel-semantics mode
     // uses the plain vendor-neutral resource (no datadog.*).
     this.collector =
@@ -205,17 +210,10 @@ private void emit(OtlpMetricsVisitor visitor) {
 
   private void emitDataPointAttributes(
       OtlpMetricVisitor metric, AggregateEntry entry, boolean error, boolean allTopLevel) {
-    if (error) {
-      emitStringAttribute(metric, STATUS_CODE, STATUS_CODE_ERROR);
-    }
-    // OTel semconv attrs are emitted in both modes
+    emitStringAttribute(metric, STATUS_CODE, error ? STATUS_CODE_ERROR : STATUS_CODE_OK);
     emitStringAttribute(metric, SPAN_NAME, entry.getResource());
-    emitStringAttribute(metric, SPAN_KIND, entry.getSpanKind());
-    // service.name on the point only when the span's service differs from the resource's default
-    UTF8BytesString service = entry.getService();
-    if (service != null && service.length() > 0 && !service.toString().equals(defaultService)) {
-      emitStringAttribute(metric, SERVICE_NAME, service);
-    }
+    emitStringAttribute(metric, SPAN_KIND, canonicalSpanKind(entry.getSpanKind()));
+    emitStringAttribute(metric, SERVICE_NAME, entry.getService());
     if (entry.hasHttpMethod()) {
       emitStringAttribute(metric, HTTP_REQUEST_METHOD, entry.getHttpMethod());
     }
@@ -228,40 +226,61 @@ 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.
+    // additional_metric_tags support is still evolving/TBD across most tracer SDKs.
     for (UTF8BytesString additionalTag : entry.getAdditionalTags()) {
-      emitAdditionalTag(metric, additionalTag);
+      emitAdditionalTag(metric, additionalTag, otelSemanticsMode);
     }
-    // Default (Datadog) mode: emit datadog.* per-point attributes
     if (!otelSemanticsMode) {
       emitStringAttribute(metric, DATADOG_OPERATION_NAME, entry.getOperationName());
       emitStringAttribute(metric, DATADOG_SPAN_TYPE, entry.getType());
-      emitLongAttribute(metric, DATADOG_SPAN_TOP_LEVEL, allTopLevel ? 1L : 0L);
+      emitBooleanAttribute(metric, DATADOG_SPAN_TOP_LEVEL, allTopLevel);
+      emitBooleanAttribute(metric, DATADOG_IS_TRACE_ROOT, entry.isTraceRoot());
       if (entry.isSynthetics()) {
         emitStringAttribute(metric, DATADOG_ORIGIN, SYNTHETICS_ORIGIN);
       }
+      emitPeerTags(metric, entry.getPeerTags());
+    }
+  }
+
+  private static void emitPeerTags(OtlpMetricVisitor metric, List<UTF8BytesString> peerTags) {
+    if (peerTags.isEmpty()) {
+      return;
+    }
+    List<String> peerTagValues = new ArrayList<>(peerTags.size());
+    for (UTF8BytesString peerTag : peerTags) {
+      peerTagValues.add(peerTag.toString());
+    }
+    metric.visitAttribute(STRING_ARRAY_ATTRIBUTE, DATADOG_PEER_TAGS, peerTagValues);
+  }
+
+  private static String canonicalSpanKind(CharSequence spanKind) {
+    if (Tags.SPAN_KIND_SERVER.contentEquals(spanKind)) {
+      return SPAN_KIND_SERVER;
+    } else if (Tags.SPAN_KIND_CLIENT.contentEquals(spanKind)) {
+      return SPAN_KIND_CLIENT;
+    } else if (Tags.SPAN_KIND_PRODUCER.contentEquals(spanKind)) {
+      return SPAN_KIND_PRODUCER;
+    } else if (Tags.SPAN_KIND_CONSUMER.contentEquals(spanKind)) {
+      return SPAN_KIND_CONSUMER;
+    } else {
+      return SPAN_KIND_INTERNAL;
     }
   }
 
-  // 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 only malformed slots with no ':' or
-  // an empty key. An empty value ("key:") is emitted as key="": the aggregation path treats an
-  // explicitly-empty tag as a distinct dimension from an absent one, so dropping it here would
-  // export two separately-aggregated rows with identical OTLP attribute sets.
-  private static void emitAdditionalTag(OtlpMetricVisitor metric, UTF8BytesString additionalTag) {
+  private static void emitAdditionalTag(
+      OtlpMetricVisitor metric, UTF8BytesString additionalTag, boolean suppressDatadogAttributes) {
     String packed = additionalTag.toString();
     int separator = packed.indexOf(':');
     if (separator <= 0) {
       return;
     }
-    metric.visitAttribute(
-        STRING_ATTRIBUTE, packed.substring(0, separator), packed.substring(separator + 1));
+    String key = packed.substring(0, separator);
+    if (suppressDatadogAttributes && key.startsWith(DATADOG_ATTRIBUTE_PREFIX)) {
+      return;
+    }
+    metric.visitAttribute(STRING_ATTRIBUTE, key, 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) {
     if (value != null) {
@@ -272,4 +291,8 @@ private static void emitStringAttribute(
   private static void emitLongAttribute(OtlpMetricVisitor metric, String key, long value) {
     metric.visitAttribute(LONG_ATTRIBUTE, key, value);
   }
+
+  private static void emitBooleanAttribute(OtlpMetricVisitor metric, String key, boolean value) {
+    metric.visitAttribute(BOOLEAN_ATTRIBUTE, key, value);
+  }
 }