fix(otlp): preserve service-entry trace metrics

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,34 @@ 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)
+
+        this.#addDistribution(
+          distributions, aggStats.topLevelOkDistribution, startNano, endNano,
+          baseAttributes, true, STATUS_CODE_OK
+        )
+        this.#addDistribution(
+          distributions, aggStats.topLevelErrorDistribution, startNano, endNano,
+          baseAttributes, true, STATUS_CODE_ERROR
+        )
+        this.#addDistribution(
+          distributions, aggStats.nonTopLevelOkDistribution, startNano, endNano,
+          baseAttributes, false, STATUS_CODE_OK
+        )
+        this.#addDistribution(
+          distributions, aggStats.nonTopLevelErrorDistribution, startNano, endNano,
+          baseAttributes, false, STATUS_CODE_ERROR
+        )
+      }
+
+      for (const { sketch, startNano, endNano, attributes } of distributions.values()) {
+        this.#pushPoint(dataPoints, sketch, startNano, endNano, attributes)
       }
     }
 
@@ -123,8 +133,44 @@ 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 {boolean} topLevel
+   * @param {string} statusCode
+   * @returns {void}
+   */
+  #addDistribution (distributions, sketch, startNano, endNano, baseAttributes, topLevel, statusCode) {
     if (!sketch || sketch.count === 0) return
+
+    const attributes = {
+      ...baseAttributes,
+      'datadog.span.top_level': topLevel,
+      'status.code': statusCode,
+    }
+    const key = stableStringify(attributes)
+    const existing = distributions.get(key)
+    if (existing) {
+      existing.sketch.merge(sketch)
+    } else {
+      distributions.set(key, {
+        sketch,
+        startNano,
+        endNano,
+        attributes: this.transformAttributes(attributes),
+      })
+    }
+  }
+
+  #pushPoint (points, sketch, startNano, endNano, attributes) {
     points.push({
       attributes,
       startTimeUnixNano: startNano,
@@ -140,15 +186,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 +208,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_processor.js

@@ -6,17 +6,19 @@ const SpanSampler = require('./span_sampler')
 const GitMetadataTagger = require('./git_metadata_tagger')
 const processTags = require('./process-tags')
 const { applyHttpOtelSemantics } = require('./plugins/util/http-otel-semantics')
-const { APM_TRACING_ENABLED_KEY } = require('./constants')
+const { APM_TRACING_ENABLED_KEY, TOP_LEVEL_KEY } = require('./constants')
 
 const startedSpans = new WeakSet()
 const finishedSpans = new WeakSet()
+const servicesByTrace = new WeakMap()
 
 class SpanProcessor {
   constructor (exporter, prioritySampler, config, otlpStatsExporter) {
     this._exporter = exporter
     this._prioritySampler = prioritySampler
     this._config = config
     this._killAll = false
+    this._trackServiceBoundaries = Boolean(otlpStatsExporter)
 
     if (config.stats?.DD_TRACE_STATS_COMPUTATION_ENABLED && !config.appsec?.standalone?.enabled) {
       const { SpanStatsProcessor } = require('./span_stats')
@@ -56,8 +58,32 @@ class SpanProcessor {
 
       let isFirstSpanInChunk = true
       const stampApmDisabled = this._config.apmTracingEnabled === false
+      let serviceBySpanId
+      if (this._trackServiceBoundaries) {
+        serviceBySpanId = servicesByTrace.get(trace)
+        if (!serviceBySpanId) {
+          serviceBySpanId = new WeakMap()
+          servicesByTrace.set(trace, serviceBySpanId)
+        }
+        for (const span of started) {
+          const context = span.context()
+          if (context._spanId !== undefined) {
+            serviceBySpanId.set(context._spanId, context.getTag('service.name'))
+          }
+        }
+      }
 
       for (const span of started) {
+        if (serviceBySpanId) {
+          const context = span.context()
+          const parentId = context._parentId
+          const service = context.getTag('service.name')
+          const parentService = serviceBySpanId.get(parentId)
+          if (parentId && (!serviceBySpanId.has(parentId) ||
+            (service !== undefined && parentService !== undefined && service !== parentService))) {
+            context.setTag(TOP_LEVEL_KEY, 1)
+          }
+        }
         if (span._duration === undefined) {
           active.push(span)
         } else {