[DI] Add support for loading probes from JSON file (#5941)

Add support for loading Dynamic Instrumentation / Live Debugger probes via a
JSON file instead of Remote Configuration (RC).

To use this feature, specify a path to the JSON file using either the
environment variable `DD_DYNAMIC_INSTRUMENTATION_PROBE_FILE` or the
programmatic configuration `dynamicInstrumentation.probeFile`.

This can be used in combination with RC.

The JSON should be an array of probe objects in the same format as received via
the RC `config` object, for example:

    [{
      id: '100c9a5c-45ad-49dc-818b-c570d31e11d1',
      version: 0,
      type: 'LOG_PROBE',
      where: { sourceFile: 'index.js', lines: ['25'] },
      template: 'Hello World',
      segments: [{ str: 'Hello World' }],
      captureSnapshot: true,
      capture: { maxReferenceDepth: 3 },
      sampling: { snapshotsPerSecond: 100 }
    }]

Author: watson

integration-tests/debugger/basic.spec.js

@@ -1,6 +1,8 @@
 'use strict'
 
+const { writeFileSync } = require('fs')
 const os = require('os')
+const { join } = require('path')
 
 const { assert } = require('chai')
 const { pollInterval, setup } = require('./utils')
@@ -188,7 +190,7 @@ describe('Dynamic Instrumentation', function () {
 
       it(
         'should send expected error diagnostics messages if probe doesn\'t conform to expected schema',
-        unsupporedOrInvalidProbesTest('bad config!!!', { status: 'ERROR' })
+        unsupporedOrInvalidProbesTest({ invalid: 'config' }, { status: 'ERROR' })
       )
 
       it(
@@ -493,10 +495,7 @@ describe('Dynamic Instrumentation', function () {
     })
 
     describe('input messages', function () {
-      it(
-        'should capture and send expected payload when a log line probe is triggered',
-        testBasicInputWithDD.bind(null, t)
-      )
+      it('should capture and send expected payload when a log line probe is triggered', testBasicInput.bind(null, t))
 
       it('should respond with updated message if probe message is updated', function (done) {
         const expectedMessages = ['Hello World!', 'Hello Updated World!']
@@ -743,17 +742,26 @@ describe('Dynamic Instrumentation', function () {
     })
   })
 
+  describe('probe file', function () {
+    const probeFile = join(os.tmpdir(), 'probes.json')
+    const t = setup({
+      env: { DD_DYNAMIC_INSTRUMENTATION_PROBE_FILE: probeFile },
+      dependencies: ['fastify']
+    })
+    const probe = t.generateProbeConfig()
+    writeFileSync(probeFile, JSON.stringify([probe]))
+
+    it('should install probes from a probe file', testBasicInputWithoutRC.bind(null, t, probe))
+  })
+
   describe('DD_TRACING_ENABLED=true, DD_TRACE_128_BIT_TRACEID_GENERATION_ENABLED=true', function () {
     const t = setup({
       env: { DD_TRACING_ENABLED: true, DD_TRACE_128_BIT_TRACEID_GENERATION_ENABLED: true },
       dependencies: ['fastify']
     })
 
     describe('input messages', function () {
-      it(
-        'should capture and send expected payload when a log line probe is triggered',
-        testBasicInputWithDD.bind(null, t)
-      )
+      it('should capture and send expected payload when a log line probe is triggered', testBasicInput.bind(null, t))
     })
   })
 
@@ -764,10 +772,7 @@ describe('Dynamic Instrumentation', function () {
     })
 
     describe('input messages', function () {
-      it(
-        'should capture and send expected payload when a log line probe is triggered',
-        testBasicInputWithDD.bind(null, t)
-      )
+      it('should capture and send expected payload when a log line probe is triggered', testBasicInput.bind(null, t))
     })
   })
 
@@ -786,10 +791,19 @@ describe('Dynamic Instrumentation', function () {
   })
 })
 
-function testBasicInputWithDD (t, done) {
-  let traceId, spanId, dd
-
+function testBasicInput (t, done) {
   t.triggerBreakpoint()
+  setupAssertionListeners(t, done)
+  t.agent.addRemoteConfig(t.rcConfig)
+}
+
+function testBasicInputWithoutRC (t, probe, done) {
+  t.triggerBreakpoint(false)
+  setupAssertionListeners(t, done, probe)
+}
+
+function setupAssertionListeners (t, done, probe) {
+  let traceId, spanId, dd
 
   t.agent.on('message', ({ payload }) => {
     const span = payload.find((arr) => arr[0].name === 'fastify.request')?.[0]
@@ -802,7 +816,7 @@ function testBasicInputWithDD (t, done) {
   })
 
   t.agent.on('debugger-input', ({ payload }) => {
-    assertBasicInputPayload(t, payload)
+    assertBasicInputPayload(t, payload, probe)
 
     payload = payload[0]
     assert.isObject(payload.dd)
@@ -816,8 +830,6 @@ function testBasicInputWithDD (t, done) {
     assertDD()
   })
 
-  t.agent.addRemoteConfig(t.rcConfig)
-
   function assertDD () {
     if (!traceId || !spanId || !dd) return
     assert.strictEqual(dd.trace_id, traceId)
@@ -838,7 +850,7 @@ function testBasicInputWithoutDD (t, done) {
   t.agent.addRemoteConfig(t.rcConfig)
 }
 
-function assertBasicInputPayload (t, payload) {
+function assertBasicInputPayload (t, payload, probe = t.rcConfig.config) {
   assert.isArray(payload)
   assert.lengthOf(payload, 1)
   payload = payload[0]
@@ -857,7 +869,7 @@ function assertBasicInputPayload (t, payload) {
     debugger: {
       snapshot: {
         probe: {
-          id: t.rcConfig.config.id,
+          id: probe.id,
           version: 0,
           location: { file: t.breakpoint.deployedFile, lines: [String(t.breakpoint.line)] }
         },

packages/dd-trace/src/ci-visibility/dynamic-instrumentation/index.js

@@ -62,7 +62,7 @@ class TestVisDynamicInstrumentation {
 
     log.debug('Starting Test Visibility - Dynamic Instrumentation client...')
 
-    const rcChannel = new MessageChannel() // mock channel
+    const probeChannel = new MessageChannel() // mock channel
     const configChannel = new MessageChannel() // mock channel
 
     this.worker = new Worker(
@@ -84,14 +84,14 @@ class TestVisDynamicInstrumentation {
         workerData: {
           config: this._config.serialize(),
           parentThreadId,
-          rcPort: rcChannel.port1,
+          probePort: probeChannel.port1,
           configPort: configChannel.port1,
           breakpointSetChannel: this.breakpointSetChannel.port1,
           breakpointHitChannel: this.breakpointHitChannel.port1,
           breakpointRemoveChannel: this.breakpointRemoveChannel.port1
         },
         transferList: [
-          rcChannel.port1,
+          probeChannel.port1,
           configChannel.port1,
           this.breakpointSetChannel.port1,
           this.breakpointHitChannel.port1,

packages/dd-trace/src/config.js

@@ -516,6 +516,7 @@ class Config {
     defaults['dogstatsd.port'] = '8125'
     defaults.dsmEnabled = false
     defaults['dynamicInstrumentation.enabled'] = false
+    defaults['dynamicInstrumentation.probeFile'] = undefined
     defaults['dynamicInstrumentation.redactedIdentifiers'] = []
     defaults['dynamicInstrumentation.redactionExcludedIdentifiers'] = []
     defaults['dynamicInstrumentation.uploadIntervalSeconds'] = 1
@@ -707,6 +708,7 @@ class Config {
       DD_DOGSTATSD_HOST,
       DD_DOGSTATSD_PORT,
       DD_DYNAMIC_INSTRUMENTATION_ENABLED,
+      DD_DYNAMIC_INSTRUMENTATION_PROBE_FILE,
       DD_DYNAMIC_INSTRUMENTATION_REDACTED_IDENTIFIERS,
       DD_DYNAMIC_INSTRUMENTATION_REDACTION_EXCLUDED_IDENTIFIERS,
       DD_DYNAMIC_INSTRUMENTATION_UPLOAD_INTERVAL_SECONDS,
@@ -883,6 +885,7 @@ class Config {
     this._setString(env, 'dogstatsd.port', DD_DOGSTATSD_PORT)
     this._setBoolean(env, 'dsmEnabled', DD_DATA_STREAMS_ENABLED)
     this._setBoolean(env, 'dynamicInstrumentation.enabled', DD_DYNAMIC_INSTRUMENTATION_ENABLED)
+    this._setString(env, 'dynamicInstrumentation.probeFile', DD_DYNAMIC_INSTRUMENTATION_PROBE_FILE)
     this._setArray(env, 'dynamicInstrumentation.redactedIdentifiers', DD_DYNAMIC_INSTRUMENTATION_REDACTED_IDENTIFIERS)
     this._setArray(
       env,
@@ -1108,6 +1111,7 @@ class Config {
     }
     this._setBoolean(opts, 'dsmEnabled', options.dsmEnabled)
     this._setBoolean(opts, 'dynamicInstrumentation.enabled', options.dynamicInstrumentation?.enabled)
+    this._setString(opts, 'dynamicInstrumentation.probeFile', options.dynamicInstrumentation?.probeFile)
     this._setArray(
       opts,
       'dynamicInstrumentation.redactedIdentifiers',

packages/dd-trace/src/debugger/devtools_client/remote_config.js

@@ -1,6 +1,6 @@
 'use strict'
 
-const { workerData: { rcPort } } = require('node:worker_threads')
+const { workerData: { probePort } } = require('node:worker_threads')
 const { addBreakpoint, removeBreakpoint, modifyBreakpoint } = require('./breakpoints')
 const { ackReceived, ackInstalled, ackError } = require('./status')
 const log = require('../../log')
@@ -32,16 +32,19 @@ const log = require('../../log')
 //   sampling: { snapshotsPerSecond: 5000 },
 //   evaluateAt: 'EXIT' // only used for method probes
 // }
-rcPort.on('message', async ({ action, conf: probe, ackId }) => {
+probePort.on('message', async ({ action, probe, ackId }) => {
   try {
     await processMsg(action, probe)
-    rcPort.postMessage({ ackId })
+    probePort.postMessage({ ackId })
   } catch (err) {
-    rcPort.postMessage({ ackId, error: err })
+    probePort.postMessage({ ackId, error: err })
     ackError(err, probe)
   }
 })
-rcPort.on('messageerror', (err) => log.error('[debugger:devtools_client] received "messageerror" on RC port', err))
+probePort.on(
+  'messageerror',
+  (err) => log.error('[debugger:devtools_client] received "messageerror" on probe port', err)
+)
 
 async function processMsg (action, probe) {
   log.debug(

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

@@ -1,5 +1,6 @@
 'use strict'
 
+const { readFile } = require('fs')
 const { types } = require('util')
 const { join } = require('path')
 const { Worker, MessageChannel, threadId: parentThreadId } = require('worker_threads')
@@ -23,17 +24,24 @@ function start (config, rc) {
   log.debug('[debugger] Starting Dynamic Instrumentation client...')
 
   const rcAckCallbacks = new Map()
-  const rcChannel = new MessageChannel()
+  const probeChannel = new MessageChannel()
   configChannel = new MessageChannel()
 
   process[Symbol.for('datadog:node:util:types')] = types
 
-  rc.setProductHandler('LIVE_DEBUGGING', (action, conf, id, ack) => {
+  readProbeFile(config.dynamicInstrumentation.probeFile, (probes) => {
+    const action = 'apply'
+    for (const probe of probes) {
+      probeChannel.port2.postMessage({ action, probe })
+    }
+  })
+
+  rc.setProductHandler('LIVE_DEBUGGING', (action, probe, id, ack) => {
     rcAckCallbacks.set(++ackId, ack)
-    rcChannel.port2.postMessage({ action, conf, ackId })
+    probeChannel.port2.postMessage({ action, probe, ackId })
   })
 
-  rcChannel.port2.on('message', ({ ackId, error }) => {
+  probeChannel.port2.on('message', ({ ackId, error }) => {
     const ack = rcAckCallbacks.get(ackId)
     if (ack === undefined) {
       // This should never happen, but just in case something changes in the future, we should guard against it
@@ -44,7 +52,7 @@ function start (config, rc) {
     ack(error)
     rcAckCallbacks.delete(ackId)
   })
-  rcChannel.port2.on('messageerror', (err) => log.error('[debugger] received "messageerror" on RC port', err))
+  probeChannel.port2.on('messageerror', (err) => log.error('[debugger] received "messageerror" on probe port', err))
 
   worker = new Worker(
     join(__dirname, 'devtools_client', 'index.js'),
@@ -54,10 +62,10 @@ function start (config, rc) {
       workerData: {
         config: config.serialize(),
         parentThreadId,
-        rcPort: rcChannel.port1,
+        probePort: probeChannel.port1,
         configPort: configChannel.port1
       },
-      transferList: [rcChannel.port1, configChannel.port1]
+      transferList: [probeChannel.port1, configChannel.port1]
     }
   )
 
@@ -84,8 +92,8 @@ function start (config, rc) {
   })
 
   worker.unref()
-  rcChannel.port1.unref()
-  rcChannel.port2.unref()
+  probeChannel.port1.unref()
+  probeChannel.port2.unref()
   configChannel.port1.unref()
   configChannel.port2.unref()
 }
@@ -94,3 +102,22 @@ function configure (config) {
   if (configChannel === null) return
   configChannel.port2.postMessage(config.serialize())
 }
+
+function readProbeFile (path, cb) {
+  if (!path) return
+
+  log.debug('[debugger] Reading probe file: %s', path)
+  readFile(path, 'utf8', (err, data) => {
+    if (err) {
+      log.error('[debugger] Failed to read probe file: %s', path, err)
+      return
+    }
+    try {
+      const parsedData = JSON.parse(data)
+      log.debug('[debugger] Successfully parsed probe file: %s', path)
+      cb(parsedData)
+    } catch (err) {
+      log.error('[debugger] Probe file (%s) is not valid JSON', path, err)
+    }
+  })
+}

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

@@ -56,6 +56,7 @@
     "DD_DOGSTATSD_HOST": ["A"],
     "DD_DOGSTATSD_PORT": ["A"],
     "DD_DYNAMIC_INSTRUMENTATION_ENABLED": ["A"],
+    "DD_DYNAMIC_INSTRUMENTATION_PROBE_FILE": ["A"],
     "DD_DYNAMIC_INSTRUMENTATION_REDACTED_IDENTIFIERS": ["A"],
     "DD_DYNAMIC_INSTRUMENTATION_REDACTION_EXCLUDED_IDENTIFIERS": ["A"],
     "DD_DYNAMIC_INSTRUMENTATION_UPLOAD_INTERVAL_SECONDS": ["A"],