Add support for Dynamic Instrumentation

packages/dd-trace/src/config.js

@@ -3,7 +3,7 @@
 const fs = require('fs')
 const os = require('os')
 const uuid = require('crypto-randomuuid') // we need to keep the old uuid dep because of cypress
-const URL = require('url').URL
+const { URL } = require('url')
 const log = require('./log')
 const pkg = require('./pkg')
 const coalesce = require('koalas')
@@ -498,6 +498,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)
@@ -589,6 +590,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,
@@ -696,6 +698,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)
@@ -837,6 +840,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/config.js

@@ -0,0 +1,26 @@
+'use strict'
+
+const { workerData: { config: parentConfig, configPort } } = require('node:worker_threads')
+const { URL, format } = require('node:url')
+
+const config = module.exports = {
+  runtimeId: parentConfig.tags['runtime-id'],
+  service: parentConfig.service
+}
+
+updateUrl(parentConfig)
+
+configPort.on('message', updateUrl)
+
+function updateUrl (updates) {
+  // if (!updates.url && !updates.hostname && !updates.port) return
+
+  const url = updates.url || new URL(format({
+    // TODO: Can this ever be anything other than `http:`, and if so, how do we get the configured value?
+    protocol: config.url?.protocol || 'http:',
+    hostname: updates.hostname || config.url?.hostname || 'localhost',
+    port: updates.port || config.url?.port
+  }))
+
+  config.url = url.toString()
+}

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

@@ -0,0 +1,42 @@
+'use strict'
+
+const uuid = require('crypto-randomuuid')
+const { breakpoints } = require('./state')
+const session = require('./session')
+const send = require('./send')
+const { ackEmitting } = require('./status')
+require('./remote_config')
+const log = require('../../log')
+
+// 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 timestamp = Date.now()
+  const probes = params.hitBreakpoints.map((id) => breakpoints.get(id))
+  await session.post('Debugger.resume')
+  const diff = process.hrtime.bigint() - start // TODO: Should we log this using some sort of telemetry?
+
+  log.debug(`Finished processing breakpoints - thread paused for: ${Number(diff) / 1000000} ms`)
+
+  // TODO: Is this the correct way of handling multiple breakpoints hit at the same time?
+  for (const probe of probes) {
+    await send({
+      message: probe.template, // TODO: Process template
+      snapshot: {
+        id: uuid(),
+        timestamp,
+        probe: {
+          id: probe.id,
+          version: probe.version, // TODO: Should this always be 2???
+          location: probe.location
+        },
+        language: 'javascript'
+      }
+    })
+
+    ackEmitting(probe)
+  }
+})

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/remote_config.js

@@ -0,0 +1,119 @@
+'use strict'
+
+const { workerData: { rcPort } } = require('node:worker_threads')
+const { scripts, probes, breakpoints } = require('./state')
+const session = require('./session')
+const { ackReceived, ackInstalled, ackError } = require('./status')
+const log = require('../../log')
+
+let sessionStarted = false
+
+// Example `probe`:
+// {
+//   id: '100c9a5c-45ad-49dc-818b-c570d31e11d1',
+//   version: 0,
+//   type: 'LOG_PROBE',
+//   language: 'javascript', // ignore
+//   where: {
+//     sourceFile: 'index.js',
+//     lines: ['25'] // only use first array element
+//   },
+//   tags: [], // not used
+//   template: 'Hello World 2',
+//   segments: [{ str: 'Hello World 2' }],
+//   captureSnapshot: true,
+//   capture: { maxReferenceDepth: 1 },
+//   sampling: { snapshotsPerSecond: 1 },
+//   evaluateAt: 'EXIT' // only used for method probes
+// }
+rcPort.on('message', async ({ action, conf: probe }) => {
+  try {
+    await processMsg(action, probe)
+  } catch (err) {
+    ackError(err, probe)
+  }
+})
+
+async function start () {
+  sessionStarted = true
+  await session.post('Debugger.enable')
+}
+
+async function stop () {
+  sessionStarted = false
+  await session.post('Debugger.disable')
+}
+
+async function processMsg (action, probe) {
+  log.debug(`Received request to ${action} ${probe.type} probe (id: ${probe.id}, version: ${probe.version})`)
+
+  ackReceived(probe)
+
+  if (probe.type !== 'LOG_PROBE') {
+    throw new Error(`Unsupported probe type: ${probe.type} (id: ${probe.id}, version: ${probe.version})`)
+    // TODO: Throw also if method log probe
+  }
+
+  switch (action) {
+    case 'unapply':
+      await removeBreakpoint(probe)
+      break
+    case 'apply':
+      await addBreakpoint(probe)
+      break
+    case 'modify':
+      // TODO: Can we modify in place?
+      await removeBreakpoint(probe)
+      await addBreakpoint(probe)
+      break
+    default:
+      throw new Error(
+        `Cannot process probe ${probe.id} (version: ${probe.version}) - unknown remote configuration action: ${action}`
+      )
+  }
+}
+
+async function addBreakpoint (probe) {
+  if (!sessionStarted) await start()
+
+  // Optimize for sending data to /debugger/v1/input endpoint
+  probe.location = {
+    file: probe.where.sourceFile,
+    lines: [Number(probe.where.lines[0])] // Tracer doesn't support multiple-line breakpoints
+  }
+  delete probe.where
+
+  // TODO: Figure out what to do about the full path
+  const path = `file:///Users/thomas.watson/go/src/github.com/DataDog/debugger-demos/nodejs/${probe.location.file}`
+
+  const { breakpointId } = await session.post('Debugger.setBreakpoint', {
+    location: {
+      scriptId: scripts.get(path),
+      lineNumber: probe.location.lines[0] - 1 // Beware! lineNumber is zero-indexed
+    }
+    // TODO: Support conditions
+    // condition: "request.params.name === 'break'"
+  })
+
+  probes.set(probe.id, breakpointId)
+  breakpoints.set(breakpointId, probe)
+
+  ackInstalled(probe)
+}
+
+async function removeBreakpoint ({ id }) {
+  if (!sessionStarted) {
+    // We should not get in this state, but abort if we do so the code doesn't throw
+    throw Error(`Cannot remove probe ${id}: Debugger not started`)
+  }
+  if (!probes.has(id)) {
+    throw Error(`Unknown probe id: ${id}`)
+  }
+
+  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,60 @@
+'use strict'
+
+const { hostname } = require('node:os')
+const { inspect } = require('node:util')
+const config = require('./config')
+const request = require('../../exporters/common/request')
+
+const host = hostname()
+const service = config.service
+const ddsource = 'dd_debugger'
+
+// TODO: Figure out correct logger values
+const logger = {
+  name: __filename,
+  method: '<module>',
+  thread_name: `${process.argv0};pid:${process.pid}`,
+  thread_id: 42,
+  version: 2
+}
+
+module.exports = async function send ({ message, snapshot: { id, timestamp, captures, probe, language } }) {
+  const opts = {
+    method: 'POST',
+    url: config.url,
+    path: '/debugger/v1/input',
+    headers: {
+      'Content-Type': 'application/json; charset=utf-8'
+      // Accept: 'text/plain' // TODO: This seems wrong (from Python tracer)
+    }
+  }
+
+  const payload = {
+    service,
+    ddsource,
+    message,
+    logger,
+    // 'dd.trace_id': null,
+    // 'dd.span_id': null,
+    // method,
+    'debugger.snapshot': {
+      id,
+      timestamp,
+      captures,
+      // evaluationErrors: [{ expr: 'foo == 42', message: 'foo' }], // TODO: This doesn't seem to be documented?
+      probe,
+      language
+    },
+    host, // TODO: This doesn't seem to be documented?
+    timestamp: Date.now() // TODO: This doesn't seem to be documented?
+    // ddtags: {} // TODO: This doesn't seem to be documented?
+  }
+
+  process._rawDebug('Payload:', inspect(payload, { depth: null, colors: true })) // TODO: Remove
+
+  request(JSON.stringify(payload), opts, (err, data, statusCode) => {
+    if (err) throw err // TODO: Handle error
+    process._rawDebug('Response:', { statusCode }) // TODO: Remove
+    process._rawDebug('Response body:', JSON.parse(data)) // TODO: Remove
+  })
+}

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()

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

@@ -0,0 +1,24 @@
+'use strict'
+
+const session = require('./session')
+
+const scripts = new Map()
+
+module.exports = {
+  scripts,
+  probes: new Map(),
+  breakpoints: new Map()
+}
+
+// Known params.url protocols:
+// - `node:` - Ignored, as we don't want to instrument Node.js internals
+// - `wasm:` - Ignored, as we don't support instrumenting WebAssembly
+// - `file:` - Regular on disk file
+// Unknown params.url values:
+// - `structured-stack` - Not sure what this is, but should just be ignored
+// - `` - Not sure what this is, but should just be ignored
+session.on('Debugger.scriptParsed', ({ params }) => {
+  if (params.url.startsWith('file:')) {
+    scripts.set(params.url, params.scriptId)
+  }
+})