feat(dsm,dbm): add process tags support for enhanced trace correlation (#7212)

Add propagation hash support to Data Streams Monitoring (DSM) and Database
Monitoring (DBM) to enable correlation between traces, database operations,
and data stream pathways using process and container metadata.

The propagation hash is an FNV-1a 64-bit hash combining process tags
(entrypoint info, package.json name) with container tags received from the
Datadog agent. This hash is included in DSM pathway computations and DBM
SQL comments to enable enhanced observability and trace correlation.

Key changes:
- Create propagation-hash module for FNV-1a hash computation with caching
- Update DSM pathway hash computation to include propagation hash
- Add ProcessTags field to DSM payloads sent to agent
- Add ddsh parameter to DBM SQL comments containing propagation hash
- Capture container tags from agent response headers
- Feature is opt-in via DD_EXPERIMENTAL_PROPAGATE_PROCESS_TAGS_ENABLED

This aligns with similar implementations in dd-trace-py (#15356),
dd-trace-java (#9282), and dd-trace-rb (#5208, #5209).

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>

Author: tlhunter

index.d.ts

@@ -768,6 +768,19 @@ declare namespace tracer {
      */
     dsmEnabled?: boolean
 
+    /**
+     * Configuration for Database Monitoring (DBM).
+     */
+    dbm?: {
+      /**
+       * Controls whether to inject the SQL base hash (propagation hash) in DBM SQL comments.
+       * This option requires DD_EXPERIMENTAL_PROPAGATE_PROCESS_TAGS_ENABLED=true to take effect.
+       * The propagation hash enables correlation between traces and database operations.
+       * @default false
+       */
+      injectSqlBaseHash?: boolean
+    }
+
     /**
      * Configuration of the AppSec protection. Can be a boolean as an alias to `appsec.enabled`.
      */

packages/dd-trace/src/config/defaults.js

@@ -62,6 +62,7 @@ module.exports = {
   'codeOriginForSpans.enabled': true,
   'codeOriginForSpans.experimental.exit_spans.enabled': false,
   dbmPropagationMode: 'disabled',
+  'dbm.injectSqlBaseHash': false,
   'dogstatsd.hostname': '127.0.0.1',
   'dogstatsd.port': '8125',
   dsmEnabled: false,

packages/dd-trace/src/config/index.js

@@ -283,6 +283,7 @@ class Config {
       DD_CODE_ORIGIN_FOR_SPANS_EXPERIMENTAL_EXIT_SPANS_ENABLED,
       DD_DATA_STREAMS_ENABLED,
       DD_DBM_PROPAGATION_MODE,
+      DD_DBM_INJECT_SQL_BASEHASH,
       DD_DOGSTATSD_HOST,
       DD_DOGSTATSD_PORT,
       DD_DYNAMIC_INSTRUMENTATION_CAPTURE_TIMEOUT_MS,
@@ -574,6 +575,7 @@ class Config {
       DD_CODE_ORIGIN_FOR_SPANS_EXPERIMENTAL_EXIT_SPANS_ENABLED
     )
     setString(target, 'dbmPropagationMode', DD_DBM_PROPAGATION_MODE)
+    setBoolean(target, 'dbm.injectSqlBaseHash', DD_DBM_INJECT_SQL_BASEHASH)
     setString(target, 'dogstatsd.hostname', DD_DOGSTATSD_HOST)
     setString(target, 'dogstatsd.port', DD_DOGSTATSD_PORT)
     setBoolean(target, 'dsmEnabled', DD_DATA_STREAMS_ENABLED)
@@ -900,6 +902,7 @@ class Config {
       options.codeOriginForSpans?.experimental?.exit_spans?.enabled
     )
     setString(opts, 'dbmPropagationMode', options.dbmPropagationMode)
+    setBoolean(opts, 'dbm.injectSqlBaseHash', options.dbm?.injectSqlBaseHash)
     if (options.dogstatsd) {
       setString(opts, 'dogstatsd.hostname', options.dogstatsd.hostname)
       setString(opts, 'dogstatsd.port', options.dogstatsd.port)

packages/dd-trace/src/config/supported-configurations.json

@@ -64,6 +64,7 @@
     "DD_CUSTOM_TRACE_ID": ["A"],
     "DD_DATA_STREAMS_ENABLED": ["A"],
     "DD_DBM_PROPAGATION_MODE": ["A"],
+    "DD_DBM_INJECT_SQL_BASEHASH": ["A"],
     "DD_DOGSTATSD_HOST": ["A"],
     "DD_DOGSTATSD_PORT": ["A"],
     "DD_DYNAMIC_INSTRUMENTATION_CAPTURE_TIMEOUT_MS": ["A"],

packages/dd-trace/src/datastreams/pathway.js

@@ -26,17 +26,36 @@ function shaHash (checkpointString) {
  * @param {string} env
  * @param {string[]} edgeTags
  * @param {Buffer} parentHash
+ * @param {bigint | null} propagationHashBigInt - Optional propagation hash for process/container tags
  */
-function computeHash (service, env, edgeTags, parentHash) {
+function computeHash (service, env, edgeTags, parentHash, propagationHashBigInt = null) {
   edgeTags.sort()
   const hashableEdgeTags = edgeTags.filter(item => item !== 'manual_checkpoint:true')
 
-  const key = `${service}${env}${hashableEdgeTags.join('')}${parentHash}`
+  // Cache key includes parentHash to handle fan-in/fan-out scenarios where the same
+  // service+env+tags+propagationHash can have different parents. This ensures we cache
+  // the complete pathway context, not just the current node's identity.
+  const propagationPart = propagationHashBigInt ? `:${propagationHashBigInt.toString(16)}` : ''
+  const key = `${service}${env}${hashableEdgeTags.join('')}${parentHash}${propagationPart}`
+
   let value = cache.get(key)
   if (value) {
     return value
   }
-  const currentHash = shaHash(`${service}${env}` + hashableEdgeTags.join(''))
+
+  // Key vs hashInput distinction:
+  // - 'key' (above) is used for caching and includes parentHash to differentiate pathways
+  //   with the same node but different parents (e.g., multiple queues feeding one consumer)
+  // - 'hashInput' (below) excludes parentHash to compute only the current node's identity hash,
+  //   which is then XORed with parentHash (line 54) to build the complete pathway hash
+  // This two-step approach (hash current node independently, then combine with parent) is
+  // required for proper pathway construction in the DSM protocol.
+  const baseString = `${service}${env}` + hashableEdgeTags.join('')
+  const hashInput = propagationHashBigInt
+    ? `${baseString}:${propagationHashBigInt.toString(16)}`
+    : baseString
+
+  const currentHash = shaHash(hashInput)
   const buf = Buffer.concat([currentHash, parentHash], 16)
   value = shaHash(buf.toString())
   cache.set(key, value)

packages/dd-trace/src/datastreams/processor.js

@@ -6,6 +6,8 @@ const pkg = require('../../../../package.json')
 const { LogCollapsingLowestDenseDDSketch } = require('../../../../vendor/dist/@datadog/sketches-js')
 const { PATHWAY_HASH } = require('../../../../ext/tags')
 const log = require('../log')
+const processTags = require('../process-tags')
+const propagationHash = require('../propagation-hash')
 const { DsmPathwayCodec } = require('./pathway')
 const { DataStreamsWriter } = require('./writer')
 const { computePathwayHash } = require('./pathway')
@@ -162,6 +164,7 @@ class DataStreamsProcessor {
   onInterval () {
     const { Stats } = this._serializeBuckets()
     if (Stats.length === 0) return
+
     const payload = {
       Env: this.env,
       Service: this.service,
@@ -171,6 +174,12 @@ class DataStreamsProcessor {
       Lang: 'javascript',
       Tags: Object.entries(this.tags).map(([key, value]) => `${key}:${value}`),
     }
+
+    // Add ProcessTags only if feature is enabled and process tags exist
+    if (propagationHash.isEnabled() && processTags.serialized) {
+      payload.ProcessTags = processTags.serialized.split(',')
+    }
+
     this.writer.flush(payload)
   }
 
@@ -234,7 +243,11 @@ class DataStreamsProcessor {
         edgeTags
       )
     }
-    const hash = computePathwayHash(this.service, this.env, edgeTags, parentHash)
+
+    // Get propagation hash if enabled
+    const propagationHashValue = propagationHash.isEnabled() ? propagationHash.getHash() : null
+
+    const hash = computePathwayHash(this.service, this.env, edgeTags, parentHash, propagationHashValue)
     const edgeLatencyNs = nowNs - edgeStartNs
     const pathwayLatencyNs = nowNs - pathwayStartNs
     const dataStreamsContext = {

packages/dd-trace/src/exporters/agent/writer.js

@@ -1,13 +1,12 @@
 'use strict'
 
-const { inspect } = require('util')
-
 const request = require('../common/request')
 const { startupLog } = require('../../startup-log')
 const runtimeMetrics = require('../../runtime_metrics')
 const log = require('../../log')
 const tracerVersion = require('../../../../../package.json').version
 const BaseWriter = require('../common/writer')
+const propagationHash = require('../../propagation-hash')
 
 const METRIC_PREFIX = 'datadog.tracer.node.exporter.agent'
 
@@ -29,10 +28,7 @@ class AgentWriter extends BaseWriter {
     runtimeMetrics.increment(`${METRIC_PREFIX}.requests`, true)
 
     const { _headers, _lookup, _protocolVersion, _url } = this
-    makeRequest(_protocolVersion, data, count, _url, _headers, _lookup, (err, res, status) => {
-      // Note that logging will only happen once, regardless of how many times this is called.
-      startupLog(status !== 404 && status !== 200 ? { status, message: err?.message ?? inspect(err) } : undefined)
-
+    makeRequest(_protocolVersion, data, count, _url, _headers, _lookup, true, (err, res, status, headers) => {
       if (status) {
         runtimeMetrics.increment(`${METRIC_PREFIX}.responses`, true)
         runtimeMetrics.increment(`${METRIC_PREFIX}.responses.by.status`, `status:${status}`, true)
@@ -53,6 +49,16 @@ class AgentWriter extends BaseWriter {
 
       log.debug('Response from the agent: %s', res)
 
+      // Capture container tags hash from agent response headers
+      // The hash is sent by the agent only when Datadog-Container-ID is present in the request
+      // (Datadog-Container-ID is automatically injected by docker.inject() in exporters/common/request.js)
+      if (headers) {
+        const containerTagsHash = headers['Datadog-Container-Tags-Hash']
+        if (containerTagsHash) {
+          propagationHash.updateContainerTagsHash(containerTagsHash)
+        }
+      }
+
       try {
         this._prioritySampler.update(JSON.parse(res).rate_by_service)
       } catch (e) {
@@ -72,7 +78,7 @@ function getEncoder (protocolVersion) {
     : require('../../encode/0.4').AgentEncoder
 }
 
-function makeRequest (version, data, count, url, headers, lookup, cb) {
+function makeRequest (version, data, count, url, headers, lookup, needsStartupLog, cb) {
   const options = {
     path: `/v${version}/traces`,
     method: 'PUT',
@@ -91,7 +97,15 @@ function makeRequest (version, data, count, url, headers, lookup, cb) {
 
   log.debug('Request to the agent: %j', options)
 
-  request(data, options, cb)
+  request(data, options, (err, res, status, headers) => {
+    if (needsStartupLog) {
+      // Note that logging will only happen once, regardless of how many times this is called.
+      startupLog({
+        agentError: status !== 404 && status !== 200 ? err : undefined,
+      })
+    }
+    cb(err, res, status, headers)
+  })
 }
 
 module.exports = AgentWriter

packages/dd-trace/src/exporters/common/request.js

@@ -88,13 +88,13 @@ function request (data, options, callback) {
           zlib.gunzip(buffer, (err, result) => {
             if (err) {
               log.error('Could not gunzip response: %s', err.message)
-              callback(null, '', res.statusCode)
+              callback(null, '', res.statusCode, res.headers)
             } else {
-              callback(null, result.toString(), res.statusCode)
+              callback(null, result.toString(), res.statusCode, res.headers)
             }
           })
         } else {
-          callback(null, buffer.toString(), res.statusCode)
+          callback(null, buffer.toString(), res.statusCode, res.headers)
         }
       } else {
         let errorMessage = ''
@@ -115,7 +115,7 @@ function request (data, options, callback) {
         const error = new log.NoTransmitError(errorMessage)
         error.status = res.statusCode
 
-        callback(error, null, res.statusCode)
+        callback(error, null, res.statusCode, res.headers)
       }
     })
   }

packages/dd-trace/src/plugins/database.js

@@ -1,6 +1,7 @@
 'use strict'
 
 const { PEER_SERVICE_KEY, PEER_SERVICE_SOURCE_KEY } = require('../constants')
+const propagationHash = require('../propagation-hash')
 const StoragePlugin = require('./storage')
 
 class DatabasePlugin extends StoragePlugin {
@@ -59,13 +60,25 @@ class DatabasePlugin extends StoragePlugin {
     const dbmService = this.#getDbmServiceName(serviceName, peerData)
     const servicePropagation = this.#createDBMPropagationCommentService(dbmService, span, peerData)
 
+    let dbmComment = servicePropagation
+
+    // Add propagation hash if both process tags and SQL base hash injection are enabled
+    if (propagationHash.isEnabled() && this.config['dbm.injectSqlBaseHash']) {
+      const hashBase64 = propagationHash.getHashBase64()
+      if (hashBase64) {
+        dbmComment += `,ddsh='${hashBase64}'`
+        // Add hash to span meta as a tag
+        span.setTag('_dd.dbm.propagation_hash', hashBase64)
+      }
+    }
+
     if (disableFullMode || mode === 'service') {
-      return servicePropagation
+      return dbmComment
     } else if (mode === 'full') {
       span.setTag('_dd.dbm_trace_injected', 'true')
       span._processor.sample(span)
       const traceparent = span._spanContext.toTraceparent()
-      return `${servicePropagation},traceparent='${traceparent}'`
+      return `${dbmComment},traceparent='${traceparent}'`
     }
   }