feat(otel): otel sampling support in distributed tracing

.github/CODEOWNERS

@@ -333,6 +333,9 @@
 /packages/dd-trace/test/baggage.spec.js @DataDog/dd-trace-js @DataDog/apm-sdk-capabilities-js
 /packages/dd-trace/src/carrier.js @DataDog/dd-trace-js @DataDog/apm-sdk-capabilities-js
 /packages/dd-trace/test/carrier.spec.js @DataDog/dd-trace-js @DataDog/apm-sdk-capabilities-js
+/packages/dd-trace/src/knuth-hash.js @DataDog/dd-trace-js @DataDog/apm-sdk-capabilities-js
+/packages/dd-trace/src/otel-sampling.js @DataDog/dd-trace-js @DataDog/apm-sdk-capabilities-js
+/packages/dd-trace/test/opentracing/propagation/otel-sampling.spec.js @DataDog/dd-trace-js @DataDog/apm-sdk-capabilities-js
 /packages/dd-trace/src/*sampler.js @DataDog/dd-trace-js @DataDog/apm-sdk-capabilities-js
 /packages/dd-trace/test/*sampler.spec.js @DataDog/dd-trace-js @DataDog/apm-sdk-capabilities-js
 /packages/dd-trace/src/sampling_rule.js @DataDog/dd-trace-js @DataDog/apm-sdk-capabilities-js

benchmark/sirun/propagation/index.js

@@ -1,17 +1,33 @@
 'use strict'
 
-const guard = require('../startup-guard')
 // eslint-disable-next-line import/order -- The startup guard must run before every other require.
+const guard = require('../startup-guard')
 const assert = require('node:assert/strict')
 
+const { channel } = require('dc-polyfill')
+
 const { getAllBaggageItems } = require('../../../packages/dd-trace/src/baggage')
 const id = require('../../../packages/dd-trace/src/id')
+const {
+  PROPAGATED_ML_APP_KEY,
+  PROPAGATED_PARENT_AGENT_ID_KEY,
+  PROPAGATED_PARENT_AGENT_NAME_KEY,
+  PROPAGATED_PARENT_ID_KEY,
+} = require('../../../packages/dd-trace/src/llmobs/constants/tags')
 const SpanContext = require('../../../packages/dd-trace/src/opentracing/span_context')
 const TextMapPropagator = require('../../../packages/dd-trace/src/opentracing/propagation/text_map')
+const TraceState = require('../../../packages/dd-trace/src/opentracing/propagation/tracestate')
 
 const { VARIANT } = process.env
 
 const OPERATIONS = Number(process.env.OPERATIONS)
+const WITH_TRACE_TAG_REPLACEMENTS = VARIANT === 'inject-replacements'
+
+/**
+ * @typedef {object} TraceTagInjection
+ * @property {Array<string | undefined>} [traceTagReplacements]
+ * @property {number} [optionalTraceTagCount]
+ */
 
 // Duck-typed config keeps the bench out of the full `Config` singleton (telemetry
 // registration, env reads). The propagator only reads the fields below.
@@ -51,6 +67,13 @@ const EXTRACT_CARRIER_DATADOG = {
   'x-datadog-tags': '_dd.p.dm=-1,_dd.p.tid=1234567890abcdef',
 }
 
+const TRACE_TAG_REPLACEMENTS = [
+  PROPAGATED_PARENT_ID_KEY, 'abcdef1234567890',
+  PROPAGATED_ML_APP_KEY, 'benchmark-app',
+  PROPAGATED_PARENT_AGENT_ID_KEY, '0123456789abcdef',
+  PROPAGATED_PARENT_AGENT_NAME_KEY, 'benchmark-agent',
+]
+
 const injectContext = new SpanContext({
   traceId: id('1234567890abcdef'),
   spanId: id('abcdef1234567890'),
@@ -61,8 +84,19 @@ const injectContext = new SpanContext({
     started: [],
     finished: [],
   },
+  tracestate: WITH_TRACE_TAG_REPLACEMENTS ? TraceState.fromString(EXTRACT_CARRIER_ASCII.tracestate) : undefined,
 })
 
+const injectCh = channel('dd-trace:span:inject')
+
+/** @param {TraceTagInjection} injection */
+function addTraceTagReplacements (injection) {
+  injection.traceTagReplacements = TRACE_TAG_REPLACEMENTS
+  injection.optionalTraceTagCount = 2
+}
+
+if (WITH_TRACE_TAG_REPLACEMENTS) injectCh.subscribe(addTraceTagReplacements)
+
 // Pre-flight: confirm the selected variant does real work; catches a silent
 // breakage where the duck-typed config is missing a field the propagator now reads.
 if (VARIANT === 'extract') {
@@ -81,10 +115,26 @@ if (VARIANT === 'extract') {
     '_dd.p.dm': '-1',
     '_dd.p.tid': '1234567890abcdef',
   })
-} else if (VARIANT === 'inject') {
+} else if (VARIANT === 'inject' || WITH_TRACE_TAG_REPLACEMENTS) {
   const sanityInjected = {}
   propagator.inject(injectContext, sanityInjected)
-  assert.ok(sanityInjected.traceparent && sanityInjected['x-datadog-trace-id'], 'inject populated no headers')
+  if (WITH_TRACE_TAG_REPLACEMENTS) {
+    assert.strictEqual(
+      sanityInjected['x-datadog-tags'],
+      '_dd.p.dm=-1,_dd.p.tid=1234567890abcdef,' +
+      '_dd.p.llmobs_parent_id=abcdef1234567890,_dd.p.llmobs_ml_app=benchmark-app,' +
+      '_dd.p.llmobs_pagent_span_id=0123456789abcdef,_dd.p.llmobs_pagent_name=benchmark-agent'
+    )
+    assert.strictEqual(
+      sanityInjected.tracestate,
+      'dd=t.llmobs_pagent_name:benchmark-agent;t.llmobs_pagent_span_id:0123456789abcdef;' +
+      't.llmobs_ml_app:benchmark-app;t.llmobs_parent_id:abcdef1234567890;' +
+      't.tid:1234567890abcdef;t.dm:-1;s:1;p:abc,' +
+      'vendor1=k1:v1,vendor2=k1:v1;k2:v2,foo=bar'
+    )
+  } else {
+    assert.ok(sanityInjected.traceparent && sanityInjected['x-datadog-trace-id'], 'inject populated no headers')
+  }
 }
 
 guard.loopStart()
@@ -100,9 +150,10 @@ if (VARIANT === 'extract') {
   for (let iteration = 0; iteration < OPERATIONS; iteration++) {
     propagator.extract(EXTRACT_CARRIER_DATADOG)
   }
-} else if (VARIANT === 'inject') {
+} else if (VARIANT === 'inject' || WITH_TRACE_TAG_REPLACEMENTS) {
   for (let iteration = 0; iteration < OPERATIONS; iteration++) {
     propagator.inject(injectContext, {})
   }
 }
 guard.done()
+if (WITH_TRACE_TAG_REPLACEMENTS) injectCh.unsubscribe(addTraceTagReplacements)

benchmark/sirun/propagation/meta.json

@@ -21,6 +21,15 @@
         "OPERATIONS": "3000000"
       }
     },
+    "inject-replacements": {
+      "run": "node --predictable-gc-schedule index.js",
+      "run_with_affinity": "bash -c \"taskset -c $CPU_AFFINITY node --predictable-gc-schedule index.js\"",
+      "iterations": 10,
+      "env": {
+        "VARIANT": "inject-replacements",
+        "OPERATIONS": "1500000"
+      }
+    },
     "extract-baggage-percent": {
       "env": {
         "VARIANT": "extract-baggage-percent",

packages/dd-trace/src/knuth-hash.js

@@ -0,0 +1,17 @@
+'use strict'
+
+const UINT64_MODULO = 2n ** 64n
+
+// Knuth's factor for the sampling algorithm shared across Datadog tracers.
+const SAMPLING_KNUTH_FACTOR = 1_111_111_111_111_111_111n
+
+/**
+ * Hashes the lower 64 bits of a trace ID for deterministic trace sampling.
+ *
+ * @param {bigint} traceId
+ */
+function knuthHash (traceId) {
+  return (traceId * SAMPLING_KNUTH_FACTOR) % UINT64_MODULO
+}
+
+module.exports = knuthHash

packages/dd-trace/src/opentelemetry/span_context.js

@@ -2,7 +2,9 @@
 
 const api = require('@opentelemetry/api')
 const { AUTO_KEEP } = require('../../../../ext/priority')
+const { updateOtelTraceState } = require('../otel-sampling')
 const DatadogSpanContext = require('../opentracing/span_context')
+const TraceState = require('../opentracing/propagation/tracestate')
 const id = require('../id')
 
 function newContext () {
@@ -37,8 +39,10 @@ class SpanContext {
   }
 
   get traceState () {
-    const ts = this._ddContext._tracestate
-    return api.createTraceState(ts ? ts.toString() : '')
+    this._ddContext._ensureSamplingPriority()
+    const traceState = this._ddContext._tracestate?.clone() ?? new TraceState()
+    updateOtelTraceState(this._ddContext, traceState)
+    return api.createTraceState(traceState.toString())
   }
 }
 

packages/dd-trace/src/opentelemetry/tracer.js

@@ -8,6 +8,7 @@ const tracer = require('../../')
 
 const id = require('../id')
 const log = require('../log')
+const { normalizeOtelTraceState } = require('../otel-sampling')
 const TraceState = require('../opentracing/propagation/tracestate')
 const SpanContext = require('./span_context')
 const Span = require('./span')
@@ -74,12 +75,12 @@ class Tracer {
   _convertOtelContextToDatadog (traceId, spanId, traceFlag, ts, meta = {}) {
     let origin = null
     let samplingPriority = traceFlag
+    let samplingMechanism
+    const traceStateValue = typeof ts?.serialize === 'function' ? ts.serialize() : ts?.traceparent
+    const traceState = TraceState.fromString(traceStateValue)
+    normalizeOtelTraceState(traceState)
 
-    ts = ts?.traceparent
-
-    if (ts) {
-      // Use TraceState.fromString to parse the tracestate header
-      const traceState = TraceState.fromString(ts)
+    if (traceStateValue) {
       let ddTraceStateData = null
 
       // Extract Datadog specific trace state data
@@ -89,29 +90,26 @@ class Tracer {
       })
 
       if (ddTraceStateData) {
-        // Assuming ddTraceStateData is now a Map or similar structure containing Datadog trace state data
-        // Extract values as needed, similar to the original logic
-        const samplingPriorityTs = ddTraceStateData.get('s')
-        origin = ddTraceStateData.get('o') ?? null
-        // Convert Map to object for meta
-        const otherPropagatedTags = Object.fromEntries(ddTraceStateData.entries())
-
-        // Update meta and samplingPriority based on extracted values
-        Object.assign(meta, otherPropagatedTags)
-        // Guard against an undefined/empty `s:` field that would result in NaN.
-        const tracestateSamplingPriority = samplingPriorityTs ? Math.trunc(samplingPriorityTs) : undefined
+        const priority = Number.parseInt(ddTraceStateData.get('s'), 10)
+        const tracestateSamplingPriority = Number.isInteger(priority) ? priority : undefined
+        origin = ddTraceStateData.get('o')?.replaceAll('~', '=') ?? null
+
+        const mechanism = Math.abs(Number.parseInt(ddTraceStateData.get('t.dm'), 10))
+        if (Number.isInteger(mechanism)) samplingMechanism = mechanism
+
         samplingPriority = getSamplingPriority(traceFlag, tracestateSamplingPriority, origin)
       } else {
-        log.debug('No dd list member in tracestate from incoming request:', ts)
+        log.debug('No dd list member in tracestate from incoming request:', traceStateValue)
       }
     }
 
     const spanContext = new SpanContext({
-      traceId: id(traceId, 16), spanId: id(), tags: meta, parentId: id(spanId, 16),
+      traceId: id(traceId, 16), spanId: id(), tags: meta, parentId: id(spanId, 16), tracestate: traceState,
     })
 
     spanContext._ddContext._sampling = { priority: samplingPriority }
     spanContext._ddContext._trace = { ...spanContext._ddContext._trace, origin }
+    if (samplingMechanism !== undefined) spanContext._ddContext._sampling.mechanism = samplingMechanism
     return spanContext
   }