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/debugger/devtools_client/index.js

@@ -0,0 +1,66 @@
+'use strict'
+
+const { breakpoints } = require('./state')
+const log = require('./logger')
+const session = require('./session')
+const { getLocalStateForBreakpoint, toObject } = require('./snapshot')
+const send = require('./send')
+const { ackEmitting } = require('./status')
+require('./remote_config')
+
+// The `session.connectToMainThread()` method called above doesn't "register" any active handles, so the worker thread
+// will exit with code 0 once when reaches the end of the file unless we do something to keep it alive:
+setInterval(() => {}, 1000 * 60)
+
+session.on('Debugger.paused', async ({ params }) => {
+  const start = process.hrtime.bigint()
+  const conf = breakpoints.get(params.hitBreakpoints[0]) // TODO: Handle multiple breakpoints
+  const state = conf.captureSnapshot ? await getLocalStateForBreakpoint(params) : undefined
+  await session.post('Debugger.resume')
+  const diff = process.hrtime.bigint() - start // TODO: Should we log this using some sort of telemetry?
+
+  await send({
+    probe: {
+      id: conf.id,
+      version: conf.version, // TODO: Is this the same version?
+      location: {
+        file: conf.where.sourceFile,
+        lines: conf.where.lines // TODO: Is it right to give the whole array here?
+      },
+      language: conf.language
+    },
+    service: 'watson-nodejs-demo-app', // TODO: Get this from tracer config
+    message: conf.template // TODO: Process template
+  })
+
+  ackEmitting(conf)
+
+  // TODO: Remove before shipping
+  log.debug(
+    '\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'
+  )
+})
+
+// TODO: Remove this function before shipping
+function stateToString (state) {
+  if (state === undefined) return '<not captured>'
+  state = toObject(state)
+  let str = ''
+  for (const [name, value] of Object.entries(state)) {
+    str += `${name}: ${color(value)}\n`
+  }
+  return str
+}
+
+// TODO: Remove this function before shipping
+function color (obj) {
+  return require('node:util').inspect(obj, { depth: null, colors: true })
+}

packages/dd-trace/src/debugger/devtools_client/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/debugger/devtools_client/logger.js

@@ -0,0 +1,13 @@
+'use strict'
+
+const { workerData: { logPort } } = require('node:worker_threads')
+
+const logLevels = ['debug', 'info', 'warn', 'error']
+const logger = {}
+
+for (const level of logLevels) {
+  // TODO: Only send back meesage to the parentPort if log level is active
+  logger[level] = (...args) => logPort.postMessage({ level, args })
+}
+
+module.exports = logger

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

@@ -0,0 +1,99 @@
+'use strict'
+
+const { workerData: { rcPort } } = require('node:worker_threads')
+const { scripts, probes, breakpoints } = require('./state')
+const log = require('./logger')
+const session = require('./session')
+const { ackReceived, ackInstalled, ackError } = require('./status')
+
+let sessionStarted = false
+
+rcPort.on('message', ({ action, conf }) => {
+  if (!sessionStarted) {
+    start()
+      .then(() => {
+        processMsg(action, conf).catch(processMsgErrorHandler)
+      })
+      .catch(startupErrorHandler)
+  } else {
+    processMsg(action, conf).catch(processMsgErrorHandler)
+  }
+})
+
+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
+  await session.post('Debugger.enable')
+}
+
+async function stop () {
+  sessionStarted = false
+  await session.post('Debugger.disable')
+}
+
+async function processMsg (action, conf) {
+  log.debug('Processing probe update of type "%s" with config:', action, conf)
+
+  ackReceived(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}`
+
+  try {
+    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)
+
+    ackInstalled(conf)
+  } catch (err) {
+    log.error(err) // TODO: Is there a standard format og loggering errors from the tracer?
+    ackError(err, 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)
+
+  if (breakpoints.size === 0) await stop()
+}

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

@@ -0,0 +1,68 @@
+'use strict'
+
+const { hostname } = require('node:os')
+const { inspect } = require('node:util')
+const { request } = require('node:http')
+const uuid = require('crypto-randomuuid')
+
+const host = hostname()
+
+// TODO: Get host/port from tracer config
+const url = 'http://localhost:8126/debugger/v1/input'
+
+const opts = {
+  method: 'POST',
+  headers: {
+    'Content-Type': 'application/json; charset=utf-8',
+    Accept: 'text/plain' // TODO: This seems wrong
+  }
+}
+
+// TODO: Figure out correct logger values
+const logger = {
+  name: __filename,
+  method: '<module>',
+  thread_name: `${process.argv0};pid:${process.pid}`,
+  thread_id: 42,
+  version: 2
+}
+
+const ddsource = 'dd_debugger'
+
+module.exports = async function send ({ probe: { id, version, location, language }, service, message }) {
+  const payload = {
+    service,
+    debugger: { // TODO: Technically more efficient to use "debugger.snapshot" key
+      snapshot: {
+        id: uuid(),
+        timestamp: Date.now(),
+        evaluationErrors: [],
+        // evaluationErrors: [{ expr: 'foo == 42', message: 'foo' }],
+        probe: { id, version, location },
+        language
+      }
+    },
+    host,
+    logger,
+    'dd.trace_id': null,
+    'dd.span_id': null,
+    ddsource,
+    message,
+    timestamp: Date.now()
+    // ddtags: {}
+  }
+
+  process._rawDebug('Payload:', inspect(payload, { depth: null, colors: true })) // TODO: Remove
+
+  // TODO: Do the tracer have a agent client already? (e.g. that sets up standard headers etc)
+  const req = request(url, opts, (res) => {
+    process._rawDebug('Response:', { status: res.statusCode, headers: res.headers }) // TODO: Remove
+    // TODO: Do we actually need the response for anything?
+    const buffers = []
+    res.on('data', buffers.push.bind(buffers))
+    // TODO: Remove log output
+    res.on('end', () => process._rawDebug('Response body:', JSON.parse(Buffer.concat(buffers).toString())))
+  })
+
+  req.end(JSON.stringify(payload))
+}

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

@@ -0,0 +1,7 @@
+'use strict'
+
+const inspector = require('./inspector_promises_polyfill')
+
+const session = module.exports = new inspector.Session()
+
+session.connectToMainThread()