Add support for Dynamic Instrumentation

integration-tests/debugger/index.spec.js

@@ -0,0 +1,207 @@
+'use strict'
+
+const path = require('path')
+const getPort = require('get-port')
+const Axios = require('axios')
+const { assert } = require('chai')
+const { assertObjectContains, assertUUID, createSandbox, FakeAgent, spawnProc } = require('../helpers')
+
+const probeId = 'my-uuid'
+const probeFile = 'debugger/target-app/index.js'
+const probeLineNo = 13
+
+describe('Dynamic Instrumentation', function () {
+  let axios, sandbox, cwd, appPort, appFile, agent, proc
+  const probeConfig = {
+    product: 'LIVE_DEBUGGING',
+    id: `logProbe_${probeId}`,
+    config: generateProbeConfig()
+  }
+
+  before(async function () {
+    sandbox = await createSandbox(['fastify'])
+    cwd = sandbox.folder
+    appFile = path.join(cwd, 'debugger', 'target-app', 'index.js')
+  })
+
+  after(async function () {
+    await sandbox.remove()
+  })
+
+  beforeEach(async function () {
+    appPort = await getPort()
+    agent = await new FakeAgent().start()
+    proc = await spawnProc(appFile, {
+      cwd,
+      env: {
+        APP_PORT: appPort,
+        DD_TRACE_AGENT_PORT: agent.port,
+        DD_EXPERIMENTAL_DYNAMIC_INSTRUMENTATION_ENABLED: true
+      }
+    })
+    axios = Axios.create({
+      baseURL: `http://localhost:${appPort}`
+    })
+  })
+
+  afterEach(async function () {
+    proc.kill()
+    await agent.stop()
+  })
+
+  it('base case: target app should work as expected if no test probe has been added', async function () {
+    const response = await axios.get('/foo')
+    assert.strictEqual(response.status, 200)
+    assert.deepStrictEqual(response.data, { hello: 'foo' })
+  })
+
+  describe('diagnostics messages', function () {
+    it('should send expected diagnostics messages if valid probe is received and triggered', function (done) {
+      const expectedPayloads = [{
+        ddsource: 'dd_debugger',
+        service: 'node',
+        debugger: { diagnostics: { probeId, version: 0, status: 'RECEIVED' } }
+      }, {
+        ddsource: 'dd_debugger',
+        service: 'node',
+        debugger: { diagnostics: { probeId, version: 0, status: 'INSTALLED' } }
+      }, {
+        ddsource: 'dd_debugger',
+        service: 'node',
+        debugger: { diagnostics: { probeId, version: 0, status: 'EMITTING' } }
+      }]
+
+      agent.on('debugger-diagnostics', async ({ payload }) => {
+        try {
+          const expected = expectedPayloads.shift()
+          assertObjectContains(payload, expected)
+          assertUUID(payload.debugger.diagnostics.runtimeId)
+
+          if (payload.debugger.diagnostics.status === 'INSTALLED') {
+            const response = await axios.get('/foo')
+            assert.strictEqual(response.status, 200)
+            assert.deepStrictEqual(response.data, { hello: 'foo' })
+          }
+
+          if (expectedPayloads.length === 0) done()
+        } catch (err) {
+          // Nessecary hack: Any errors throw inside of an async function is invisible to Mocha unless the outer `it`
+          // callback is also `async` (which we can't do in this case since we rely on the `done` callback).
+          done(err)
+        }
+      })
+
+      agent.addRemoteConfig(probeConfig)
+    })
+
+    const unsupporedOrInvalidProbes = [[
+      'should send expected error diagnostics messages if probe doesn\'t conform to expected schema',
+      'bad config!!!',
+      { status: 'ERROR' }
+    ], [
+      'should send expected error diagnostics messages if probe type isn\'t supported',
+      generateProbeConfig({ type: 'INVALID_PROBE' })
+    ], [
+      'should send expected error diagnostics messages if it isn\'t a line-probe',
+      generateProbeConfig({ where: { foo: 'bar' } }) // TODO: Use valid schema for method probe instead
+    ]]
+
+    for (const [title, config, costumErrorDiagnosticsObj] of unsupporedOrInvalidProbes) {
+      // TODO: Test that we report the error via the RC client as well
+      it(title, function (done) {
+        const expectedPayloads = [{
+          ddsource: 'dd_debugger',
+          service: 'node',
+          debugger: { diagnostics: { status: 'RECEIVED' } }
+        }, {
+          ddsource: 'dd_debugger',
+          service: 'node',
+          debugger: { diagnostics: costumErrorDiagnosticsObj ?? { probeId, version: 0, status: 'ERROR' } }
+        }]
+
+        agent.on('debugger-diagnostics', ({ payload }) => {
+          const expected = expectedPayloads.shift()
+          assertObjectContains(payload, expected)
+          const { diagnostics } = payload.debugger
+          assertUUID(diagnostics.runtimeId)
+
+          if (diagnostics.status === 'ERROR') {
+            assert.property(diagnostics, 'exception')
+            assert.hasAllKeys(diagnostics.exception, ['message', 'stacktrace'])
+            assert.typeOf(diagnostics.exception.message, 'string')
+            assert.typeOf(diagnostics.exception.stacktrace, 'string')
+          }
+
+          if (expectedPayloads.length === 0) done()
+        })
+
+        agent.addRemoteConfig({
+          product: 'LIVE_DEBUGGING',
+          id: `logProbe_${probeId}`,
+          config
+        })
+      })
+    }
+  })
+
+  describe('input messages', function () {
+    it('should capture and send expected snapshot when a log line probe is triggered', function (done) {
+      agent.on('debugger-diagnostics', ({ payload }) => {
+        if (payload.debugger.diagnostics.status === 'INSTALLED') {
+          axios.get('/foo')
+        }
+      })
+
+      agent.on('debugger-input', ({ payload }) => {
+        const expected = {
+          ddsource: 'dd_debugger',
+          service: 'node',
+          message: 'Hello World!',
+          logger: {
+            method: 'send',
+            version: 2,
+            thread_id: 1
+          },
+          'debugger.snapshot': {
+            probe: {
+              id: probeId,
+              version: 0,
+              location: { file: probeFile, lines: [probeLineNo] }
+            },
+            language: 'javascript'
+          }
+        }
+
+        assertObjectContains(payload, expected)
+        assert.isTrue(payload.logger.name.endsWith(path.join('src', 'debugger', 'devtools_client', 'send.js')))
+        assert.match(payload.logger.thread_name, new RegExp(`${process.argv0};pid:\\d+$`))
+        assertUUID(payload['debugger.snapshot'].id)
+        assert.typeOf(payload['debugger.snapshot'].timestamp, 'number')
+        assert.isTrue(payload['debugger.snapshot'].timestamp > Date.now() - 1000 * 60)
+        assert.isTrue(payload['debugger.snapshot'].timestamp <= Date.now())
+
+        done()
+      })
+
+      agent.addRemoteConfig(probeConfig)
+    })
+  })
+})
+
+function generateProbeConfig (overrides) {
+  return {
+    id: probeId,
+    version: 0,
+    type: 'LOG_PROBE',
+    language: 'javascript',
+    where: { sourceFile: probeFile, lines: [String(probeLineNo)] },
+    tags: [],
+    template: 'Hello World!',
+    segments: [{ str: 'Hello World!' }],
+    captureSnapshot: false,
+    capture: { maxReferenceDepth: 3 },
+    sampling: { snapshotsPerSecond: 5000 },
+    evaluateAt: 'EXIT',
+    ...overrides
+  }
+}

integration-tests/debugger/target-app/index.js

@@ -0,0 +1,22 @@
+'use strict'
+
+const Fastify = require('fastify')
+const tracer = require('dd-trace')
+
+tracer.init({
+  flushInterval: 0
+})
+
+const fastify = Fastify()
+
+fastify.get('/:name', function handler (request) {
+  return { hello: request.params.name }
+})
+
+fastify.listen({ port: process.env.APP_PORT }, (err) => {
+  if (err) {
+    fastify.log.error(err)
+    process.exit(1)
+  }
+  process.send({ port: process.env.APP_PORT })
+})

integration-tests/helpers/fake-agent.js

@@ -272,6 +272,22 @@ function buildExpressServer (agent) {
     })
   })
 
+  app.post('/debugger/v1/input', (req, res) => {
+    res.status(200).send()
+    agent.emit('debugger-input', {
+      headers: req.headers,
+      payload: req.body
+    })
+  })
+
+  app.post('/debugger/v1/diagnostics', upload.any(), (req, res) => {
+    res.status(200).send()
+    agent.emit('debugger-diagnostics', {
+      headers: req.headers,
+      payload: JSON.parse(req.files[0].buffer.toString())
+    })
+  })
+
   app.post('/profiling/v1/input', upload.any(), (req, res) => {
     res.status(200).send()
     agent.emit('message', {

integration-tests/helpers/index.js

@@ -334,12 +334,32 @@ function useSandbox (...args) {
     return oldSandbox.remove()
   })
 }
+
 function sandboxCwd () {
   return sandbox.folder
 }
 
+function assertObjectContains (actual, expected) {
+  for (const [key, val] of Object.entries(expected)) {
+    if (val !== null && typeof val === 'object') {
+      assert.ok(key in actual)
+      assert.notStrictEqual(actual[key], null)
+      assert.strictEqual(typeof actual[key], 'object')
+      assertObjectContains(actual[key], val)
+    } else {
+      assert.strictEqual(actual[key], expected[key])
+    }
+  }
+}
+
+function assertUUID (actual, msg = 'not a valid UUID') {
+  assert.match(actual, /^[\da-f]{8}-[\da-f]{4}-[\da-f]{4}-[\da-f]{4}-[\da-f]{12}$/, msg)
+}
+
 module.exports = {
   FakeAgent,
+  assertObjectContains,
+  assertUUID,
   spawnProc,
   runAndCheckWithTelemetry,
   createSandbox,

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')
@@ -423,6 +423,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)
@@ -531,6 +532,7 @@ class Config {
       DD_ENV,
       DD_EXPERIMENTAL_API_SECURITY_ENABLED,
       DD_EXPERIMENTAL_APPSEC_STANDALONE_ENABLED,
+      DD_EXPERIMENTAL_DYNAMIC_INSTRUMENTATION_ENABLED,
       DD_EXPERIMENTAL_PROFILING_ENABLED,
       JEST_WORKER_ID,
       DD_IAST_DEDUPLICATION_ENABLED,
@@ -657,6 +659,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)
@@ -824,11 +827,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,38 @@
+'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')
+
+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(
+      probe.template, // TODO: Process template
+      {
+        id: uuid(),
+        timestamp,
+        probe: {
+          id: probe.id,
+          version: probe.version,
+          location: probe.location
+        },
+        language: 'javascript'
+      }
+    )
+
+    ackEmitting(probe)
+  }
+})