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)
@@ -591,6 +592,7 @@ class Config {
       DD_DOGSTATSD_PORT,
       DD_ENV,
       DD_EXPERIMENTAL_APPSEC_STANDALONE_ENABLED,
+      DD_EXPERIMENTAL_DYNAMIC_INSTRUMENTATION_ENABLED,
       DD_EXPERIMENTAL_PROFILING_ENABLED,
       JEST_WORKER_ID,
       DD_IAST_DEDUPLICATION_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_EXPERIMENTAL_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,11 +840,11 @@ class Config {
       this._setString(opts, 'dogstatsd.port', options.dogstatsd.port)
     }
     this._setBoolean(opts, 'dsmEnabled', options.dsmEnabled)
+    this._setBoolean(opts, 'dynamicInstrumentationEnabled', options.experimental?.dynamicInstrumentationEnabled)
     this._setString(opts, 'env', options.env || tags.env)
-    this._setBoolean(opts, 'experimental.enableGetRumData',
-      options.experimental && options.experimental.enableGetRumData)
-    this._setString(opts, 'experimental.exporter', options.experimental && options.experimental.exporter)
-    this._setBoolean(opts, 'experimental.runtimeId', options.experimental && options.experimental.runtimeId)
+    this._setBoolean(opts, 'experimental.enableGetRumData', options.experimental?.enableGetRumData)
+    this._setString(opts, 'experimental.exporter', options.experimental?.exporter)
+    this._setBoolean(opts, 'experimental.runtimeId', options.experimental?.runtimeId)
     this._setValue(opts, 'flushInterval', maybeInt(options.flushInterval))
     this._optsUnprocessed.flushInterval = options.flushInterval
     this._setValue(opts, 'flushMinSpans', maybeInt(options.flushMinSpans))

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

@@ -0,0 +1,24 @@
+'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) {
+  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 inside `session.js` 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 this be recored as telemetry?
+
+  log.debug(`Finished processing breakpoints - main 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,132 @@
+'use strict'
+
+const { workerData: { rcPort } } = require('node:worker_threads')
+const { getScriptId, probes, breakpoints } = require('./state')
+const session = require('./session')
+const { ackReceived, ackInstalled, ackError } = require('./status')
+const log = require('../../log')
+
+let sessionStarted = false
+
+// Example log line probe (simplified):
+// {
+//   id: '100c9a5c-45ad-49dc-818b-c570d31e11d1',
+//   version: 0,
+//   type: 'LOG_PROBE',
+//   where: { sourceFile: 'index.js', lines: ['25'] }, // only use first array element
+//   template: 'Hello World 2',
+//   segments: [...],
+//   captureSnapshot: true,
+//   capture: { maxReferenceDepth: 1 },
+//   sampling: { snapshotsPerSecond: 1 },
+//   evaluateAt: 'EXIT' // only used for method probes
+// }
+//
+// Example log method probe (simplified):
+// {
+//   id: 'd692ee6d-5734-4df7-9d86-e3bc6449cc8c',
+//   version: 0,
+//   type: 'LOG_PROBE',
+//   where: { typeName: 'index.js', methodName: 'handlerA' },
+//   template: 'Executed index.js.handlerA, it took {@duration}ms',
+//   segments: [...],
+//   captureSnapshot: false,
+//   capture: { maxReferenceDepth: 3 },
+//   sampling: { snapshotsPerSecond: 5000 },
+//   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})`)
+  }
+  if (!probe.where.sourceFile && !probe.where.lines) {
+    throw new Error(
+      // eslint-disable-next-line max-len
+      `Unsupported probe insertion point! Only line-based probes are supported (id: ${probe.id}, version: ${probe.version})`
+    )
+  }
+
+  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()
+
+  const file = probe.where.sourceFile
+  const line = Number(probe.where.lines[0]) // Tracer doesn't support multiple-line breakpoints
+
+  // Optimize for sending data to /debugger/v1/input endpoint
+  probe.location = { file, lines: [line] }
+  delete probe.where
+
+  const scriptId = getScriptId(file)
+
+  if (!scriptId) throw new Error(`No script id found for ${file}`)
+
+  const { breakpointId } = await session.post('Debugger.setBreakpoint', {
+    location: {
+      scriptId,
+      lineNumber: line - 1 // Beware! lineNumber is zero-indexed
+    }
+  })
+
+  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 fail unexpected
+    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,51 @@
+'use strict'
+
+const { inspect } = require('node:util')
+const { threadId } = require('node:worker_threads')
+const config = require('./config')
+const log = require('../../log')
+const request = require('../../exporters/common/request')
+
+module.exports = send
+
+const service = config.service
+const ddsource = 'dd_debugger'
+
+// TODO: Figure out correct logger values
+const logger = {
+  name: __filename, // name of the class/type/file emitting the snapshot
+  method: send.name, // name of the method/function emitting the snapshot
+  version: 2, // version of the snapshot format (not currently used or enforced)
+  thread_id: threadId, // current thread/process id emitting the snapshot
+  thread_name: `${process.argv0};pid:${process.pid}` // name of the current thread emitting the snapshot
+}
+
+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' }
+  }
+
+  const payload = {
+    service,
+    ddsource,
+    message,
+    logger,
+    'debugger.snapshot': {
+      id,
+      timestamp,
+      captures,
+      probe,
+      language
+    }
+  }
+
+  process._rawDebug('Payload:', inspect(payload, { depth: null, colors: true })) // TODO: Remove
+
+  request(JSON.stringify(payload), opts, (err, data, statusCode) => {
+    if (err) log.error(err)
+    process._rawDebug('%s response:', statusCode, 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,29 @@
+'use strict'
+
+const session = require('./session')
+
+const scripts = []
+
+module.exports = {
+  probes: new Map(),
+  breakpoints: new Map(),
+  // Algorithm: Find the sortest url that ends in the requested path
+  getScriptId (path) {
+    return scripts
+      .filter(([url]) => url.endsWith(path))
+      .sort(([a], [b]) => a.length - b.length)[0]?.[1]
+  }
+}
+
+// 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.push([params.url, params.scriptId])
+  }
+})