Add support for Dynamic Instrumentation

packages/dd-trace/src/config.js

@@ -508,6 +508,7 @@ class Config {
     this._setValue(defaults, 'dogstatsd.hostname', '127.0.0.1')
     this._setValue(defaults, 'dogstatsd.port', '8125')
     this._setValue(defaults, 'dsmEnabled', false)
+    this._setValue(defaults, 'dynamicInstrumentationEnabled', false)
     this._setValue(defaults, 'env', undefined)
     this._setValue(defaults, 'experimental.enableGetRumData', false)
     this._setValue(defaults, 'experimental.exporter', undefined)
@@ -599,6 +600,7 @@ class Config {
       DD_DBM_PROPAGATION_MODE,
       DD_DOGSTATSD_HOSTNAME,
       DD_DOGSTATSD_PORT,
+      DD_DYNAMIC_INSTRUMENTATION_ENABLED,
       DD_ENV,
       DD_EXPERIMENTAL_APPSEC_STANDALONE_ENABLED,
       DD_EXPERIMENTAL_PROFILING_ENABLED,
@@ -706,6 +708,7 @@ class Config {
     this._setString(env, 'dogstatsd.hostname', DD_DOGSTATSD_HOSTNAME)
     this._setString(env, 'dogstatsd.port', DD_DOGSTATSD_PORT)
     this._setBoolean(env, 'dsmEnabled', DD_DATA_STREAMS_ENABLED)
+    this._setBoolean(env, 'dynamicInstrumentationEnabled', DD_DYNAMIC_INSTRUMENTATION_ENABLED)
     this._setString(env, 'env', DD_ENV || tags.env)
     this._setBoolean(env, 'experimental.enableGetRumData', DD_TRACE_EXPERIMENTAL_GET_RUM_DATA_ENABLED)
     this._setString(env, 'experimental.exporter', DD_TRACE_EXPERIMENTAL_EXPORTER)
@@ -847,6 +850,7 @@ class Config {
       this._setString(opts, 'dogstatsd.port', options.dogstatsd.port)
     }
     this._setBoolean(opts, 'dsmEnabled', options.dsmEnabled)
+    this._setBoolean(opts, 'dynamicInstrumentationEnabled', options.dynamicInstrumentationEnabled)
     this._setString(opts, 'env', options.env || tags.env)
     this._setBoolean(opts, 'experimental.enableGetRumData',
       options.experimental && options.experimental.enableGetRumData)

packages/dd-trace/src/debugging/devtools_client.js

@@ -0,0 +1,224 @@
+'use strict'
+
+const { hrtime } = require('node:process')
+const { parentPort } = require('node:worker_threads')
+const inspector = require('./inspector_promises_polyfill')
+
+const scripts = new Map()
+const probes = new Map()
+const breakpoints = new Map()
+let sessionStarted = false
+
+parentPort.on('message', (msg) => {
+  if (!sessionStarted) {
+    start()
+      .then(() => {
+        processMsg(msg).catch(processMsgErrorHandler)
+      })
+      .catch(startupErrorHandler)
+  } else {
+    processMsg(msg).catch(processMsgErrorHandler)
+  }
+})
+
+const session = new inspector.Session()
+session.connectToMainThread()
+session.on('Debugger.scriptParsed', ({ params }) => {
+  if (params.url.startsWith('file:')) {
+    scripts.set(params.url, params.scriptId)
+  }
+})
+
+function startupErrorHandler (err) {
+  // TODO: Handle error: If we can't start, we should disable all of DI
+  throw err
+}
+
+function processMsgErrorHandler (err) {
+  // TODO: Handle error
+  throw err
+}
+
+async function start () {
+  sessionStarted = true
+
+  session.on('Debugger.paused', async ({ params }) => {
+    const start = hrtime.bigint()
+    const state = await getLocalStateForBreakpoint(params)
+    await session.post('Debugger.resume')
+    const diff = hrtime.bigint() - start
+
+    log(
+      '\nLocal state:\n' +
+      '--------------------------------------------------\n' +
+      stateToString(state) +
+      '--------------------------------------------------\n' +
+      '\nStats:\n' +
+      '--------------------------------------------------\n' +
+      `   Total state JSON size: ${JSON.stringify(state).length} bytes\n` +
+      `Processed was paused for: ${Number(diff) / 1000000} ms\n` +
+      '--------------------------------------------------\n'
+    )
+  })
+
+  await session.post('Debugger.enable')
+}
+
+async function processMsg (msg) {
+  const { action, conf } = JSON.parse(msg)
+  log(action, color(conf))
+
+  switch (action) {
+    case 'unapply':
+      await removeBreakpoint(conf)
+      break
+    case 'apply':
+      await addBreakpoint(conf)
+      break
+    case 'modify':
+      await removeBreakpoint(conf)
+      await addBreakpoint(conf)
+      break
+    default:
+      throw new Error(`Unknown remote configuration action: ${action}`)
+  }
+}
+
+async function addBreakpoint (conf) {
+  // TODO: Figure out what to do about the full path
+  const path = `file:///Users/thomas.watson/go/src/github.com/DataDog/debugger-demos/nodejs/${conf.where.sourceFile}`
+
+  const { breakpointId } = await session.post('Debugger.setBreakpoint', {
+    location: {
+      scriptId: scripts.get(path),
+      // TODO: Support multiple lines
+      lineNumber: conf.where.lines[0] - 1 // Beware! lineNumber is zero-indexed
+    }
+    // TODO: Support conditions
+    // condition: "request.params.name === 'break'"
+  })
+
+  probes.set(conf.id, breakpointId)
+  breakpoints.set(breakpointId, conf)
+}
+
+async function removeBreakpoint ({ id }) {
+  if (!probes.has(id)) {
+    throw new Error(`Unknown prope id: ${id}`) // TODO: Log error instead
+  }
+  const breakpointId = probes.get(id)
+  await session.post('Debugger.removeBreakpoint', { breakpointId })
+  probes.delete(id)
+  breakpoints.delete(breakpointId)
+}
+
+function stateToString (state) {
+  state = toObject(state)
+  let str = ''
+  for (const [name, value] of Object.entries(state)) {
+    str += `${name}: ${color(value)}\n`
+  }
+  return str
+}
+
+function toObject (state) {
+  if (state === undefined) return '<object>'
+  const obj = {}
+  for (const prop of state) {
+    obj[prop.name] = getPropVal(prop)
+  }
+  return obj
+}
+
+function toArray (state) {
+  if (state === undefined) return '<array>'
+  const arr = []
+  for (const elm of state) {
+    if (elm.enumerable === false) continue // the value of the `length` property should not be part of the array
+    arr.push(getPropVal(elm))
+  }
+  return arr
+}
+
+function getPropVal (prop) {
+  const value = prop.value ?? prop.get
+  switch (value.type) {
+    case 'object':
+      return getObjVal(value)
+    case 'symbol':
+      return value.description // TODO: Should we really send this as a string?
+    case 'function':
+      return '<function>'
+    case 'undefined':
+      return undefined // TODO: We can't send undefined values over JSON
+    case 'boolean':
+    case 'string':
+    case 'number':
+      return value.value
+    default:
+      log('-- unknown type:', prop)
+      throw new Error('unknown type')
+  }
+}
+
+function getObjVal (obj) {
+  switch (obj.subtype) {
+    case undefined:
+      return toObject(obj.properties)
+    case 'array':
+      return toArray(obj.properties)
+    case 'null':
+      return null
+    // case 'set': // TODO: Verify if we need set as well
+    case 'map':
+    case 'regexp':
+      return obj.description // TODO: How should we actually handle this?
+    default:
+      log('-- unknown subtype:', obj)
+      throw new Error('unknown subtype')
+  }
+}
+
+async function getLocalStateForBreakpoint (params) {
+  const scope = params.callFrames[0].scopeChain[0]
+  if (scope.type !== 'local') {
+    throw new Error(`Unexpcted scope type: ${scope.type}`)
+  }
+  const conf = breakpoints.get(params.hitBreakpoints[0]) // TODO: Handle multiple breakpoints
+  return await getObjectWithChildren(scope.object.objectId, conf)
+}
+
+async function getObjectWithChildren (objectId, conf, depth = 0) {
+  const { result } = (await session.post('Runtime.getProperties', {
+    objectId,
+    ownProperties: true
+    // accessorPropertiesOnly: true, // true: only return get/set accessor properties
+    // generatePreview: true, // true: generate `value.preview` object with details (including content) of maps and sets
+    // nonIndexedPropertiesOnly: true // true: do not get array elements
+  }))
+
+  // TODO: Deside if we should filter out enumerable properties or not:
+  // result = result.filter((e) => e.enumerable)
+
+  if (depth < conf.capture.maxReferenceDepth) {
+    for (const entry of result) {
+      if (entry?.value?.type === 'object' && entry?.value?.objectId) {
+        entry.value.properties = await getObjectWithChildren(entry.value.objectId, conf, depth + 1)
+      }
+    }
+  }
+
+  return result
+}
+
+function log (...args) {
+  process._rawDebug(...args)
+}
+
+function color (obj) {
+  return require('node:util').inspect(obj, { depth: null, colors: true })
+}
+
+// TODO: Find better way to keep the worker thread alive
+// TODO: Are we properly killing the worker thread once the parent dies?
+setInterval(() => {}, 1000 * 60)

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

@@ -0,0 +1,41 @@
+'use strict'
+
+const { join } = require('node:path')
+const { Worker } = require('node:worker_threads')
+const log = require('../log')
+
+let worker = null
+
+exports.start = function (rc) {
+  if (worker !== null) return
+
+  log.debug('Starting Dynamic Instrumentation client...')
+
+  rc.on('LIVE_DEBUGGING', (action, conf) => {
+    worker.postMessage(JSON.stringify({ action, conf }))
+  })
+
+  rc.on('LIVE_DEBUGGING_SYMBOL_DB', (action, conf) => {
+    // TODO: Implement
+    console.log('-- RC: LIVE_DEBUGGING_SYMBOL_DB', action, conf)
+  })
+
+  worker = new Worker(join(__dirname, 'devtools_client.js'), {
+    execArgv: [] // Avoid worker thread inheriting the `-r` command line argument
+  })
+
+  worker.unref()
+
+  worker.on('online', () => {
+    log.debug(`Dynamic Instrumentation worker thread started successfully (thread id: ${worker.threadId})`)
+  })
+
+  // TODO: How should we handle errors?
+  worker.on('error', (err) => process._rawDebug('DevTools client error:', err))
+  worker.on('exit', (code) => {
+    log.debug(`Dynamic Instrumentation worker thread exited with code ${code}`)
+    if (code !== 0) {
+      throw new Error(`DevTools client stopped with unexpected exit code: ${code}`)
+    }
+  })
+}

packages/dd-trace/src/debugging/inspector_promises_polyfill.js

@@ -0,0 +1,23 @@
+'use strict'
+
+const { builtinModules } = require('node:module')
+
+if (builtinModules.includes('inspector/promises')) {
+  module.exports = require('node:inspector/promises')
+} else {
+  const inspector = require('node:inspector')
+  const { promisify } = require('node:util')
+
+  // The rest of the code in this file is lifted from:
+  // https://github.com/nodejs/node/blob/1d4d76ff3fb08f9a0c55a1d5530b46c4d5d550c7/lib/inspector/promises.js
+  class Session extends inspector.Session {
+    constructor () { super() } // eslint-disable-line no-useless-constructor
+  }
+
+  Session.prototype.post = promisify(inspector.Session.prototype.post)
+
+  module.exports = {
+    ...inspector,
+    Session
+  }
+}

packages/dd-trace/src/proxy.js

@@ -5,6 +5,7 @@ const Config = require('./config')
 const runtimeMetrics = require('./runtime_metrics')
 const log = require('./log')
 const { setStartupLogPluginManager } = require('./startup-log')
+const DynamicInstrumentation = require('./debugging')
 const telemetry = require('./telemetry')
 const nomenclature = require('./service-naming')
 const PluginManager = require('./plugin_manager')
@@ -111,6 +112,10 @@ class Tracer extends NoopProxy {
           this._flare.enable(config)
           this._flare.module.send(conf.args)
         })
+
+        if (config.dynamicInstrumentationEnabled) {
+          DynamicInstrumentation.start(rc)
+        }
       }
 
       if (config.isGCPFunction || config.isAzureFunction) {