fix(otlp): align trace metric attributes and aggregation

packages/dd-trace/src/opentelemetry/metrics/index.js

@@ -10,6 +10,8 @@ const MeterProvider = require('./meter_provider')
 const PeriodicMetricReader = require('./periodic_metric_reader')
 const OtlpHttpMetricExporter = require('./otlp_http_metric_exporter')
 
+const RESERVED_TRACER_TAGS = new Set(['service', 'env', 'version', 'runtime_id', 'runtime-id'])
+
 /**
  * @typedef {import('../../config')} Config
  */
@@ -78,7 +80,16 @@ function initializeOpenTelemetryMetrics (config) {
   metrics.setGlobalMeterProvider(meterProvider)
 }
 
-function buildResourceAttributes (tags, { reportHostname, otelSemanticsEnabled, service, env, serviceVersion } = {}) {
+/**
+ * @param {Record<string, unknown>} tags
+ * @param {object} [options]
+ * @param {boolean} [options.reportHostname]
+ * @param {string} [options.service]
+ * @param {string} [options.env]
+ * @param {string} [options.serviceVersion]
+ * @returns {import('@opentelemetry/api').Attributes}
+ */
+function buildResourceAttributes (tags, { reportHostname, service, env, serviceVersion } = {}) {
   const attrs = {
     'telemetry.sdk.name': 'datadog',
     'telemetry.sdk.language': 'nodejs',
@@ -89,14 +100,19 @@ function buildResourceAttributes (tags, { reportHostname, otelSemanticsEnabled,
   if (env) attrs['deployment.environment.name'] = env
   if (reportHostname) attrs['host.name'] = os.hostname()
 
-  if (!otelSemanticsEnabled) {
-    if (tags?.['runtime-id']) attrs['datadog.runtime_id'] = tags['runtime-id']
-    const processTagsObject = processTags.tagsObject
-    if (processTagsObject) {
-      for (const key of Object.keys(processTagsObject)) {
-        attrs[`datadog.${key}`] = processTagsObject[key]
-      }
-    }
+  if (tags['runtime-id']) attrs['datadog.runtime_id'] = tags['runtime-id']
+  const tracerTags = []
+  for (const [key, value] of Object.entries(tags)) {
+    const valueType = typeof value
+    const supported = valueType === 'string' || valueType === 'boolean' ||
+      (valueType === 'number' && Number.isFinite(value))
+    if (!RESERVED_TRACER_TAGS.has(key) && supported) tracerTags.push(`${key}:${value}`)
+  }
+  if (tracerTags.length) attrs['datadog.tracer_tags'] = tracerTags
+  // Mirrors the legacy v0.6/stats ProcessTags shape (buildProcessTags().tagsArray); keep both in sync.
+  const processTagsArray = processTags.tagsArray
+  if (processTagsArray.length) {
+    attrs['datadog.process_tags'] = processTagsArray
   }
   return attrs
 }
@@ -106,7 +122,6 @@ function createOtlpSpanStatsExporter (config) {
   const protocol = config.OTEL_EXPORTER_OTLP_METRICS_PROTOCOL || 'http/json'
   const resourceAttributes = buildResourceAttributes(config.tags, {
     reportHostname: config.reportHostname,
-    otelSemanticsEnabled: config.DD_TRACE_OTEL_SEMANTICS_ENABLED,
     service: config.service,
     env: config.env,
     serviceVersion: config.version,
@@ -115,8 +130,6 @@ function createOtlpSpanStatsExporter (config) {
     config.OTEL_EXPORTER_OTLP_METRICS_ENDPOINT,
     protocol,
     resourceAttributes,
-    config.DD_TRACE_OTEL_SEMANTICS_ENABLED,
-    config.service,
     config.OTEL_EXPORTER_OTLP_METRICS_HEADERS,
     config.OTEL_EXPORTER_OTLP_METRICS_TIMEOUT
   )

packages/dd-trace/src/opentelemetry/metrics/otlp_span_stats_exporter.js

@@ -11,15 +11,12 @@ class OtlpStatsExporter extends OtlpHttpExporterBase {
    * @param {string} url
    * @param {string} protocol
    * @param {import('@opentelemetry/api').Attributes} resourceAttributes
-   * @param {boolean} [otelSemanticsEnabled]
-   * @param {string} [defaultService]
    * @param {Record<string, string>} [headers]
    * @param {number} [timeout]
    */
-  constructor (url, protocol, resourceAttributes, otelSemanticsEnabled = false, defaultService = '',
-    headers, timeout = 10_000) {
+  constructor (url, protocol, resourceAttributes, headers, timeout = 10_000) {
     super(url, headers, timeout, protocol, 'span-stats')
-    this.#transformer = new OtlpStatsTransformer(resourceAttributes, protocol, otelSemanticsEnabled, defaultService)
+    this.#transformer = new OtlpStatsTransformer(resourceAttributes, protocol)
   }
 
   /**

packages/dd-trace/src/opentelemetry/metrics/otlp_span_stats_transformer.js

@@ -1,17 +1,31 @@
 'use strict'
 
-const { LogCollapsingLowestDenseDDSketch } = require('../../../../../vendor/dist/@datadog/sketches-js')
 const OtlpTransformerBase = require('../otlp/otlp_transformer_base')
 const { getProtobufTypes } = require('../otlp/protobuf_loader')
 const { GRPC_STATUS_NAMES } = require('../../constants')
 
+const { stableStringify } = OtlpTransformerBase
+
 const NS_PER_S = 1e9
 
 // Must match libdatadog's EXPLICIT_BOUNDS_SECONDS and OTel spanmetrics connector defaults.
 const EXPLICIT_BOUNDS_SECONDS = [
   0.002, 0.004, 0.006, 0.008, 0.01, 0.05, 0.1, 0.2, 0.4, 0.8, 1, 1.4, 2, 5, 10, 15,
 ]
 
+const SPAN_KIND_METRIC_MAP = {
+  internal: 'SPAN_KIND_INTERNAL',
+  SPAN_KIND_INTERNAL: 'SPAN_KIND_INTERNAL',
+  server: 'SPAN_KIND_SERVER',
+  SPAN_KIND_SERVER: 'SPAN_KIND_SERVER',
+  client: 'SPAN_KIND_CLIENT',
+  SPAN_KIND_CLIENT: 'SPAN_KIND_CLIENT',
+  producer: 'SPAN_KIND_PRODUCER',
+  SPAN_KIND_PRODUCER: 'SPAN_KIND_PRODUCER',
+  consumer: 'SPAN_KIND_CONSUMER',
+  SPAN_KIND_CONSUMER: 'SPAN_KIND_CONSUMER',
+}
+
 /**
  * @param {object} sketch
  * @returns {number[]}
@@ -41,22 +55,16 @@ function getDeltaTemporality () {
   return _deltaTemporality
 }
 
-const ERROR_STATUS_ATTR = { key: 'status.code', value: { intValue: 2 } }
+const STATUS_CODE_OK = 'STATUS_CODE_OK'
+const STATUS_CODE_ERROR = 'STATUS_CODE_ERROR'
 
 class OtlpStatsTransformer extends OtlpTransformerBase {
-  #otelSemanticsEnabled
-  #defaultService
-
   /**
    * @param {import('@opentelemetry/api').Attributes} resourceAttributes
    * @param {string} protocol
-   * @param {boolean} [otelSemanticsEnabled]
-   * @param {string} [defaultService]
    */
-  constructor (resourceAttributes, protocol, otelSemanticsEnabled = false, defaultService = '') {
+  constructor (resourceAttributes, protocol) {
     super(resourceAttributes, protocol, 'span-stats')
-    this.#otelSemanticsEnabled = otelSemanticsEnabled
-    this.#defaultService = defaultService
   }
 
   /**
@@ -82,32 +90,35 @@ class OtlpStatsTransformer extends OtlpTransformerBase {
     const dataPoints = []
 
     for (const { timeNs, bucket } of drained) {
+      const distributions = new Map()
       const endTimeNs = timeNs + bucketSizeNs
       const startNano = isJson ? String(timeNs) : timeNs
       const endNano = isJson ? String(endTimeNs) : endTimeNs
 
       for (const aggStats of bucket.values()) {
-        const baseAttrs = this.#buildAttributes(aggStats.aggKey)
-
-        if (this.#otelSemanticsEnabled) {
-          const okDist = new LogCollapsingLowestDenseDDSketch()
-          okDist.merge(aggStats.topLevelOkDistribution)
-          okDist.merge(aggStats.nonTopLevelOkDistribution)
-          const errDist = new LogCollapsingLowestDenseDDSketch()
-          errDist.merge(aggStats.topLevelErrorDistribution)
-          errDist.merge(aggStats.nonTopLevelErrorDistribution)
-          this.#pushPoint(dataPoints, okDist, startNano, endNano, baseAttrs)
-          this.#pushPoint(dataPoints, errDist, startNano, endNano, [...baseAttrs, ERROR_STATUS_ATTR])
-        } else {
-          const tlAttrs = [...baseAttrs, { key: 'datadog.span.top_level', value: { boolValue: true } }]
-          const ntlAttrs = [...baseAttrs, { key: 'datadog.span.top_level', value: { boolValue: false } }]
-          this.#pushPoint(dataPoints, aggStats.topLevelOkDistribution, startNano, endNano, tlAttrs)
-          this.#pushPoint(dataPoints, aggStats.topLevelErrorDistribution, startNano, endNano,
-            [...tlAttrs, ERROR_STATUS_ATTR])
-          this.#pushPoint(dataPoints, aggStats.nonTopLevelOkDistribution, startNano, endNano, ntlAttrs)
-          this.#pushPoint(dataPoints, aggStats.nonTopLevelErrorDistribution, startNano, endNano,
-            [...ntlAttrs, ERROR_STATUS_ATTR])
-        }
+        const baseAttributes = this.#buildAttributes(aggStats.aggKey)
+        const baseKey = stableStringify(baseAttributes)
+
+        this.#addDistribution(
+          distributions, aggStats.topLevelOkDistribution, startNano, endNano,
+          baseAttributes, baseKey, true, STATUS_CODE_OK
+        )
+        this.#addDistribution(
+          distributions, aggStats.topLevelErrorDistribution, startNano, endNano,
+          baseAttributes, baseKey, true, STATUS_CODE_ERROR
+        )
+        this.#addDistribution(
+          distributions, aggStats.nonTopLevelOkDistribution, startNano, endNano,
+          baseAttributes, baseKey, false, STATUS_CODE_OK
+        )
+        this.#addDistribution(
+          distributions, aggStats.nonTopLevelErrorDistribution, startNano, endNano,
+          baseAttributes, baseKey, false, STATUS_CODE_ERROR
+        )
+      }
+
+      for (const { sketch, startNano, endNano, attributes } of distributions.values()) {
+        this.#pushPoint(dataPoints, sketch, startNano, endNano, attributes)
       }
     }
 
@@ -123,8 +134,45 @@ class OtlpStatsTransformer extends OtlpTransformerBase {
     }]
   }
 
-  #pushPoint (points, sketch, startNano, endNano, attributes) {
+  /**
+   * @param {Map<string, {
+   *   sketch: object,
+   *   startNano: string | number,
+   *   endNano: string | number,
+   *   attributes: object[]
+   * }>} distributions
+   * @param {object} sketch
+   * @param {string | number} startNano
+   * @param {string | number} endNano
+   * @param {import('@opentelemetry/api').Attributes} baseAttributes
+   * @param {string} baseKey
+   * @param {boolean} topLevel
+   * @param {string} statusCode
+   * @returns {void}
+   */
+  #addDistribution (distributions, sketch, startNano, endNano, baseAttributes, baseKey, topLevel, statusCode) {
     if (!sketch || sketch.count === 0) return
+
+    const key = `${baseKey},${topLevel},${statusCode}`
+    const existing = distributions.get(key)
+    if (existing) {
+      existing.sketch.merge(sketch)
+    } else {
+      const attributes = this.transformAttributes(baseAttributes)
+      attributes.push(
+        { key: 'datadog.span.top_level', value: { boolValue: topLevel } },
+        { key: 'status.code', value: { stringValue: statusCode } }
+      )
+      distributions.set(key, {
+        sketch,
+        startNano,
+        endNano,
+        attributes,
+      })
+    }
+  }
+
+  #pushPoint (points, sketch, startNano, endNano, attributes) {
     points.push({
       attributes,
       startTimeUnixNano: startNano,
@@ -140,15 +188,18 @@ class OtlpStatsTransformer extends OtlpTransformerBase {
 
   /**
    * @param {import('../../span_stats').SpanAggKey} aggKey
+   * @returns {import('@opentelemetry/api').Attributes}
    */
   #buildAttributes (aggKey) {
-    const raw = { 'span.name': aggKey.resource }
-
-    if (aggKey.service && aggKey.service !== this.#defaultService) {
-      raw['service.name'] = aggKey.service
+    const spanKind = Object.hasOwn(SPAN_KIND_METRIC_MAP, aggKey.spanKind)
+      ? SPAN_KIND_METRIC_MAP[aggKey.spanKind]
+      : 'SPAN_KIND_INTERNAL'
+    const raw = {
+      'span.name': aggKey.resource,
+      'service.name': aggKey.service,
+      'span.kind': spanKind,
     }
 
-    if (aggKey.spanKind) raw['span.kind'] = aggKey.spanKind
     if (aggKey.statusCode) raw['http.response.status_code'] = Number(aggKey.statusCode)
     if (aggKey.method) raw['http.request.method'] = aggKey.method
     if (aggKey.endpoint) raw['http.route'] = aggKey.endpoint
@@ -159,13 +210,15 @@ class OtlpStatsTransformer extends OtlpTransformerBase {
         : String(aggKey.rpcStatusCode).toUpperCase()
     }
 
-    if (!this.#otelSemanticsEnabled) {
-      raw['datadog.operation.name'] = aggKey.name
-      if (aggKey.type) raw['datadog.span.type'] = aggKey.type
-      if (aggKey.synthetics) raw['datadog.origin'] = 'synthetics'
-    }
+    // TODO: additional_metric_tags support is still evolving/TBD across most SDKs; not implemented here yet.
+
+    raw['datadog.operation.name'] = aggKey.name
+    if (aggKey.type) raw['datadog.span.type'] = aggKey.type
+    if (aggKey.synthetics) raw['datadog.origin'] = 'synthetics'
+    if (aggKey.srvSrc) raw['datadog.svc_src'] = aggKey.srvSrc
+    if (typeof aggKey.isTraceRoot === 'boolean') raw['datadog.is_trace_root'] = aggKey.isTraceRoot
 
-    return this.transformAttributes(raw)
+    return raw
   }
 }
 

packages/dd-trace/src/span_stats.js

@@ -14,8 +14,10 @@ const {
   GRPC_STATUS_CODE,
 } = require('../../../ext/tags')
 const { ORIGIN_KEY, TOP_LEVEL_KEY, SVC_SRC_KEY, GRPC_STATUS_NAMES } = require('./constants')
+const id = require('./id')
 
 const GRPC_STATUS_CODE_MAP = Object.fromEntries(GRPC_STATUS_NAMES.map((name, i) => [name, String(i)]))
+const ZERO_ID = id('0')
 const { version } = require('./pkg')
 const processTags = require('./process-tags')
 
@@ -150,9 +152,24 @@ class SpanAggKey {
 }
 
 class SpanBuckets extends Map {
+  #includeTraceRoot
+
+  /**
+   * @param {boolean} [includeTraceRoot]
+   */
+  constructor (includeTraceRoot = false) {
+    super()
+    this.#includeTraceRoot = includeTraceRoot
+  }
+
   forSpan (span) {
     const aggKey = new SpanAggKey(span)
-    const key = aggKey.toString()
+    const baseKey = aggKey.toString()
+    const parentId = span.parent_id
+    if (this.#includeTraceRoot && parentId !== undefined && parentId !== null) {
+      aggKey.isTraceRoot = parentId.equals(ZERO_ID)
+    }
+    const key = this.#includeTraceRoot ? `${baseKey},${aggKey.isTraceRoot}` : baseKey
 
     if (!this.has(key)) {
       this.set(key, new SpanAggStats(aggKey))
@@ -163,9 +180,19 @@ class SpanBuckets extends Map {
 }
 
 class TimeBuckets extends Map {
+  #includeTraceRoot
+
+  /**
+   * @param {boolean} [includeTraceRoot]
+   */
+  constructor (includeTraceRoot = false) {
+    super()
+    this.#includeTraceRoot = includeTraceRoot
+  }
+
   forTime (time) {
     if (!this.has(time)) {
-      this.set(time, new SpanBuckets())
+      this.set(time, new SpanBuckets(this.#includeTraceRoot))
     }
 
     return this.get(time)
@@ -192,7 +219,7 @@ class SpanStatsProcessor {
     const intervalMs = otlpExporter ? (flushIntervalMs ?? 10_000) : interval * 1e3
     this.interval = intervalMs / 1e3
     this.bucketSizeNs = intervalMs * 1e6
-    this.buckets = new TimeBuckets()
+    this.buckets = new TimeBuckets(Boolean(otlpExporter))
     this.hostname = os.hostname()
     this.enabled = enabled
     this.otlpExporter = otlpExporter || null