fix(otlp): align trace metric attributes and aggregation

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

@@ -91,11 +91,10 @@ function buildResourceAttributes (tags, { reportHostname, otelSemanticsEnabled,
 
   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]
-      }
+    // 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
@@ -116,7 +115,6 @@ function createOtlpSpanStatsExporter (config) {
     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

@@ -12,14 +12,12 @@ class OtlpStatsExporter extends OtlpHttpExporterBase {
    * @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, otelSemanticsEnabled = false, 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, otelSemanticsEnabled)
   }
 
   /**

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

@@ -12,6 +12,14 @@ 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',
+  server: 'SPAN_KIND_SERVER',
+  client: 'SPAN_KIND_CLIENT',
+  producer: 'SPAN_KIND_PRODUCER',
+  consumer: 'SPAN_KIND_CONSUMER',
+}
+
 /**
  * @param {object} sketch
  * @returns {number[]}
@@ -41,22 +49,20 @@ function getDeltaTemporality () {
   return _deltaTemporality
 }
 
-const ERROR_STATUS_ATTR = { key: 'status.code', value: { intValue: 2 } }
+const STATUS_CODE_OK_ATTR = { key: 'status.code', value: { stringValue: 'STATUS_CODE_OK' } }
+const STATUS_CODE_ERROR_ATTR = { key: 'status.code', value: { stringValue: '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, otelSemanticsEnabled = false) {
     super(resourceAttributes, protocol, 'span-stats')
     this.#otelSemanticsEnabled = otelSemanticsEnabled
-    this.#defaultService = defaultService
   }
 
   /**
@@ -96,17 +102,19 @@ class OtlpStatsTransformer extends OtlpTransformerBase {
           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])
+          this.#pushPoint(dataPoints, okDist, startNano, endNano, [...baseAttrs, STATUS_CODE_OK_ATTR])
+          this.#pushPoint(dataPoints, errDist, startNano, endNano, [...baseAttrs, STATUS_CODE_ERROR_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.topLevelOkDistribution, startNano, endNano,
+            [...tlAttrs, STATUS_CODE_OK_ATTR])
           this.#pushPoint(dataPoints, aggStats.topLevelErrorDistribution, startNano, endNano,
-            [...tlAttrs, ERROR_STATUS_ATTR])
-          this.#pushPoint(dataPoints, aggStats.nonTopLevelOkDistribution, startNano, endNano, ntlAttrs)
+            [...tlAttrs, STATUS_CODE_ERROR_ATTR])
+          this.#pushPoint(dataPoints, aggStats.nonTopLevelOkDistribution, startNano, endNano,
+            [...ntlAttrs, STATUS_CODE_OK_ATTR])
           this.#pushPoint(dataPoints, aggStats.nonTopLevelErrorDistribution, startNano, endNano,
-            [...ntlAttrs, ERROR_STATUS_ATTR])
+            [...ntlAttrs, STATUS_CODE_ERROR_ATTR])
         }
       }
     }
@@ -142,13 +150,9 @@ class OtlpStatsTransformer extends OtlpTransformerBase {
    * @param {import('../../span_stats').SpanAggKey} aggKey
    */
   #buildAttributes (aggKey) {
-    const raw = { 'span.name': aggKey.resource }
+    const raw = { 'span.name': aggKey.resource, 'service.name': aggKey.service }
 
-    if (aggKey.service && aggKey.service !== this.#defaultService) {
-      raw['service.name'] = aggKey.service
-    }
-
-    if (aggKey.spanKind) raw['span.kind'] = aggKey.spanKind
+    if (aggKey.spanKind) raw['span.kind'] = SPAN_KIND_METRIC_MAP[aggKey.spanKind] ?? 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,10 +163,13 @@ class OtlpStatsTransformer extends OtlpTransformerBase {
         : String(aggKey.rpcStatusCode).toUpperCase()
     }
 
+    // TODO: additional_metric_tags support is still evolving/TBD across most SDKs; not implemented here yet.
+
     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'
+      raw['datadog.is_trace_root'] = aggKey.isTraceRoot
     }
 
     return this.transformAttributes(raw)

packages/dd-trace/src/span_stats.js

@@ -130,6 +130,9 @@ class SpanAggKey {
     } else {
       this.rpcStatusCode = ''
     }
+
+    this.isTraceRoot = !span.parent_id || span.parent_id.toString(10) === '0'
+    // peer_tags isn't aggregated in the legacy v0.6/stats export here either; mirror it in both once added.
   }
 
   toString () {
@@ -145,6 +148,7 @@ class SpanAggKey {
       this.srvSrc,
       this.spanKind,
       this.rpcStatusCode,
+      this.isTraceRoot,
     ].join(',')
   }
 }

packages/dd-trace/test/opentelemetry/metrics/otlp_span_stats_exporter.spec.js

@@ -11,6 +11,7 @@ const { OtlpStatsExporter } = require('../../../src/opentelemetry/metrics/otlp_s
 const { buildResourceAttributes, createOtlpSpanStatsExporter } = require('../../../src/opentelemetry/metrics')
 const { SpanBuckets } = require('../../../src/span_stats')
 const { HTTP_STATUS_CODE } = require('../../../../../ext/tags')
+const processTags = require('../../../src/process-tags')
 
 const RESOURCE_ATTRS = { 'service.name': 'svc' }
 const BUCKET_SIZE_NS = 10 * 1e9
@@ -59,6 +60,13 @@ describe('buildResourceAttributes', () => {
     const attrs = buildResourceAttributes({ 'runtime-id': 'abc-123' }, { otelSemanticsEnabled: true })
     assert.ok(!Object.keys(attrs).some(k => k.startsWith('datadog.')))
   })
+
+  it('includes datadog.process_tags as a single array attribute when otelSemanticsEnabled is false', () => {
+    processTags.initialize()
+    const attrs = buildResourceAttributes({}, { otelSemanticsEnabled: false })
+    assert.ok(Array.isArray(attrs['datadog.process_tags']))
+    assert.ok(!('datadog.entrypoint.type' in attrs))
+  })
 })
 
 describe('createOtlpSpanStatsExporter', () => {

packages/dd-trace/test/opentelemetry/metrics/otlp_span_stats_transformer.spec.js

@@ -20,7 +20,6 @@ const RESOURCE_ATTRS = {
   'service.version': '1.2.3',
   'deployment.environment.name': 'test',
 }
-const DEFAULT_SERVICE = 'svc'
 const BUCKET_SIZE_NS = 10 * 1e9
 
 function makeSpan (overrides = {}) {
@@ -81,7 +80,7 @@ describe('OtlpStatsTransformer', () => {
     let transformer
 
     before(() => {
-      transformer = new OtlpStatsTransformer(RESOURCE_ATTRS, 'http/json', false, DEFAULT_SERVICE)
+      transformer = new OtlpStatsTransformer(RESOURCE_ATTRS, 'http/json', false)
     })
 
     it('emits a single histogram metric with the correct name, unit and temporality', () => {
@@ -110,15 +109,18 @@ describe('OtlpStatsTransformer', () => {
 
       assert.deepStrictEqual(attrMapOf(dataPointsOf(payload)[0]), {
         'span.name': 'GET /foo',
-        'span.kind': 'server',
+        'service.name': 'svc',
+        'span.kind': 'SPAN_KIND_SERVER',
         'http.response.status_code': 404,
         'http.request.method': 'POST',
         'http.route': '/users/:id',
         'rpc.response.status_code': 'OK',
+        'status.code': 'STATUS_CODE_OK',
         'datadog.operation.name': 'test.op',
         'datadog.span.type': 'web',
         'datadog.origin': 'synthetics',
         'datadog.span.top_level': false,
+        'datadog.is_trace_root': true,
       })
     })
 
@@ -162,14 +164,14 @@ describe('OtlpStatsTransformer', () => {
       assert.strictEqual(dp.bucketCounts.filter(c => c > 0).length, 2)
     })
 
-    it('marks error data points with status.code=ERROR and ok data points without it', () => {
+    it('marks error data points with status.code=STATUS_CODE_ERROR and ok data points with STATUS_CODE_OK', () => {
       const spans = [makeTopLevelSpan(), makeTopLevelSpan({ error: 1 })]
       const payload = JSON.parse(transformer.transform(makeDrained(12340000000000, spans), BUCKET_SIZE_NS))
       const points = dataPointsOf(payload)
 
-      const ok = points.find(dp => attrMapOf(dp)['datadog.span.top_level'] === true && !attrMapOf(dp)['status.code'])
-      const err = points.find(dp => attrMapOf(dp)['status.code'] === 2)
-      assert.ok(ok, 'ok data point should carry no status.code')
+      const ok = points.find(dp => attrMapOf(dp)['status.code'] === 'STATUS_CODE_OK')
+      const err = points.find(dp => attrMapOf(dp)['status.code'] === 'STATUS_CODE_ERROR')
+      assert.ok(ok, 'ok data point should carry status.code=STATUS_CODE_OK')
       assert.strictEqual(attrMapOf(err)['datadog.span.top_level'], true)
     })
 
@@ -179,8 +181,8 @@ describe('OtlpStatsTransformer', () => {
       const points = dataPointsOf(payload)
 
       assert.strictEqual(points.length, 2)
-      const ok = points.find(dp => !attrMapOf(dp)['status.code'])
-      const err = points.find(dp => attrMapOf(dp)['status.code'] === 2)
+      const ok = points.find(dp => attrMapOf(dp)['status.code'] === 'STATUS_CODE_OK')
+      const err = points.find(dp => attrMapOf(dp)['status.code'] === 'STATUS_CODE_ERROR')
       assert.strictEqual(ok.count, 2)
       assert.strictEqual(err.count, 1)
       assert.strictEqual(attrMapOf(ok)['datadog.span.top_level'], true)
@@ -221,7 +223,7 @@ describe('OtlpStatsTransformer', () => {
       assert.strictEqual(resourceAttrs['deployment.environment.name'], 'test')
     })
 
-    it('emits a single scopeMetrics and tags data points whose service differs from the default', () => {
+    it('emits a single scopeMetrics and tags every data point with service.name, including the default service', () => {
       const drained = makeDrained(12340000000000, [
         makeSpan({ service: 'svc', resource: 'GET /foo' }),
         makeSpan({ service: 'svc-other', resource: 'GET /bar' }),
@@ -234,7 +236,7 @@ describe('OtlpStatsTransformer', () => {
       const serviceByResource = Object.fromEntries(
         dataPointsOf(payload).map(dp => [attrMapOf(dp)['span.name'], attrMapOf(dp)['service.name']])
       )
-      assert.strictEqual(serviceByResource['GET /foo'], undefined)
+      assert.strictEqual(serviceByResource['GET /foo'], 'svc')
       assert.strictEqual(serviceByResource['GET /bar'], 'svc-other')
     })
 
@@ -262,7 +264,7 @@ describe('OtlpStatsTransformer', () => {
     let transformer
 
     before(() => {
-      transformer = new OtlpStatsTransformer(RESOURCE_ATTRS, 'http/json', true, DEFAULT_SERVICE)
+      transformer = new OtlpStatsTransformer(RESOURCE_ATTRS, 'http/json', true)
     })
 
     it('emits only OTel attributes (no dd.*) while keeping status.code on errors', () => {
@@ -279,7 +281,7 @@ describe('OtlpStatsTransformer', () => {
       )
       assert.deepStrictEqual(
         { name: attrs['span.name'], method: attrs['http.request.method'], status: attrs['status.code'] },
-        { name: 'GET /foo', method: 'GET', status: 2 }
+        { name: 'GET /foo', method: 'GET', status: 'STATUS_CODE_ERROR' }
       )
     })
   })
@@ -288,7 +290,7 @@ describe('OtlpStatsTransformer', () => {
     let transformer
 
     before(() => {
-      transformer = new OtlpStatsTransformer(RESOURCE_ATTRS, 'http/protobuf', false, DEFAULT_SERVICE)
+      transformer = new OtlpStatsTransformer(RESOURCE_ATTRS, 'http/protobuf', false)
     })
 
     it('emits a valid ExportMetricsServiceRequest with a single duration metric', () => {
@@ -310,10 +312,10 @@ describe('OtlpStatsTransformer', () => {
       assert.strictEqual(metric.histogram.aggregationTemporality, delta)
       const okNotTopLevel = metric.histogram.dataPoints.find(dp =>
         dp.attributes.some(a => a.key === 'datadog.span.top_level' && a.value.boolValue === false) &&
-        !dp.attributes.some(a => a.key === 'status.code')
+        dp.attributes.some(a => a.key === 'status.code' && a.value.stringValue === 'STATUS_CODE_OK')
       )
       const errTopLevel = metric.histogram.dataPoints.find(dp =>
-        dp.attributes.some(a => a.key === 'status.code' && Number(a.value.intValue) === 2) &&
+        dp.attributes.some(a => a.key === 'status.code' && a.value.stringValue === 'STATUS_CODE_ERROR') &&
         dp.attributes.some(a => a.key === 'datadog.span.top_level' && a.value.boolValue === true)
       )
       assert.ok(okNotTopLevel, 'should have ok not-top-level data point')

packages/dd-trace/test/span_stats.spec.js

@@ -105,24 +105,24 @@ describe('SpanAggKey', () => {
   it('should make aggregation key for a basic span', () => {
     const key = new SpanAggKey(basicSpan)
     assert.strictEqual(
-      key.toString(), 'basic-span,service-name,resource-name,span-type,200,false,,,integration,,')
+      key.toString(), 'basic-span,service-name,resource-name,span-type,200,false,,,integration,,,true')
   })
 
   it('should make aggregation key for a synthetic span', () => {
     const key = new SpanAggKey(syntheticSpan)
     assert.strictEqual(
-      key.toString(), 'synthetic-span,service-name,resource-name,span-type,200,true,,,integration,,')
+      key.toString(), 'synthetic-span,service-name,resource-name,span-type,200,true,,,integration,,,true')
   })
 
   it('should make aggregation key for an error span', () => {
     const key = new SpanAggKey(errorSpan)
     assert.strictEqual(
-      key.toString(), 'error-span,service-name,resource-name,span-type,500,false,,,integration,,')
+      key.toString(), 'error-span,service-name,resource-name,span-type,500,false,,,integration,,,true')
   })
 
   it('should use sensible defaults', () => {
     const key = new SpanAggKey({ meta: {}, metrics: {} })
-    assert.strictEqual(key.toString(), `${DEFAULT_SPAN_NAME},${DEFAULT_SERVICE_NAME},,,0,false,,,,,`)
+    assert.strictEqual(key.toString(), `${DEFAULT_SPAN_NAME},${DEFAULT_SERVICE_NAME},,,0,false,,,,,,true`)
   })
 
   it('should include HTTP method and route in aggregation key', () => {
@@ -136,7 +136,7 @@ describe('SpanAggKey', () => {
     }
     const key = new SpanAggKey(span)
     assert.strictEqual(
-      key.toString(), 'basic-span,service-name,resource-name,span-type,200,false,GET,/users/:id,integration,,')
+      key.toString(), 'basic-span,service-name,resource-name,span-type,200,false,GET,/users/:id,integration,,,true')
   })
 
   it('should include HTTP method and endpoint in aggregation key', () => {
@@ -151,7 +151,7 @@ describe('SpanAggKey', () => {
     const key = new SpanAggKey(span)
     assert.strictEqual(
       key.toString(),
-      'basic-span,service-name,resource-name,span-type,200,false,POST,/users/{param:int},integration,,')
+      'basic-span,service-name,resource-name,span-type,200,false,POST,/users/{param:int},integration,,,true')
   })
 
   it('should prioritize http.route over http.endpoint', () => {
@@ -166,7 +166,7 @@ describe('SpanAggKey', () => {
     }
     const key = new SpanAggKey(span)
     assert.strictEqual(
-      key.toString(), 'basic-span,service-name,resource-name,span-type,200,false,GET,/users/:id,integration,,')
+      key.toString(), 'basic-span,service-name,resource-name,span-type,200,false,GET,/users/:id,integration,,,true')
   })
 
   it('should include service source in aggregation key', () => {
@@ -179,28 +179,36 @@ describe('SpanAggKey', () => {
     }
     const key = new SpanAggKey(span)
     assert.strictEqual(
-      key.toString(), 'basic-span,service-name,resource-name,span-type,200,false,,,opt.plugin,,')
+      key.toString(), 'basic-span,service-name,resource-name,span-type,200,false,,,opt.plugin,,,true')
   })
 
   it('should include span kind in aggregation key', () => {
     const span = { ...basicSpan, meta: { ...basicSpan.meta, [SPAN_KIND]: 'server' } }
     const key = new SpanAggKey(span)
     assert.strictEqual(
-      key.toString(), 'basic-span,service-name,resource-name,span-type,200,false,,,integration,server,')
+      key.toString(), 'basic-span,service-name,resource-name,span-type,200,false,,,integration,server,,true')
   })
 
   it('should normalize gRPC status name to numeric string in aggregation key', () => {
     const span = { ...basicSpan, meta: { ...basicSpan.meta, [GRPC_STATUS_CODE]: 'NOT_FOUND' } }
     const key = new SpanAggKey(span)
     assert.strictEqual(
-      key.toString(), 'basic-span,service-name,resource-name,span-type,200,false,,,integration,,5')
+      key.toString(), 'basic-span,service-name,resource-name,span-type,200,false,,,integration,,5,true')
   })
 
   it('should keep numeric gRPC status code as numeric string in aggregation key', () => {
     const span = { ...basicSpan, meta: {}, metrics: { [GRPC_STATUS_CODE]: 14 } }
     const key = new SpanAggKey(span)
     assert.strictEqual(
-      key.toString(), 'basic-span,service-name,resource-name,span-type,0,false,,,,,14')
+      key.toString(), 'basic-span,service-name,resource-name,span-type,0,false,,,,,14,true')
+  })
+
+  it('should mark isTraceRoot false when parent_id is a non-zero Identifier', () => {
+    const span = { ...basicSpan, parent_id: { toString: (radix) => (123).toString(radix) } }
+    const key = new SpanAggKey(span)
+    assert.strictEqual(key.isTraceRoot, false)
+    assert.strictEqual(
+      key.toString(), 'basic-span,service-name,resource-name,span-type,200,false,,,integration,,,false')
   })
 
   it('should use rpc.grpc.status_code OTel alias when grpc.status.code is absent', () => {