[test visibility] Simple dynamic instrumentation - test visibility client

packages/dd-trace/src/ci-visibility/dynamic-instrumentation/index.js

@@ -0,0 +1,97 @@
+'use strict'
+
+const { join } = require('path')
+const { Worker } = require('worker_threads')
+const { randomUUID } = require('crypto')
+const log = require('../../log')
+
+const probeIdToResolveBreakpointSet = new Map()
+const probeIdToResolveBreakpointHit = new Map()
+
+class TestVisDynamicInstrumentation {
+  constructor () {
+    this.worker = null
+    this._readyPromise = new Promise(resolve => {
+      this._onReady = resolve
+    })
+    this.breakpointSetChannel = new MessageChannel()
+    this.breakpointHitChannel = new MessageChannel()
+  }
+
+  // Return 3 elements:
+  // 1. Snapshot ID
+  // 2. Promise that's resolved when the breakpoint is set
+  // 3. Promise that's resolved when the breakpoint is hit
+  addLineProbe ({ file, line }) {
+    const snapshotId = randomUUID()
+    const probeId = randomUUID()
+
+    this.breakpointSetChannel.port2.postMessage({
+      snapshotId,
+      probe: { id: probeId, file, line }
+    })
+
+    return [
+      snapshotId,
+      new Promise(resolve => {
+        probeIdToResolveBreakpointSet.set(probeId, resolve)
+      }),
+      new Promise(resolve => {
+        probeIdToResolveBreakpointHit.set(probeId, resolve)
+      })
+    ]
+  }
+
+  isReady () {
+    return this._readyPromise
+  }
+
+  start () {
+    if (this.worker) return
+
+    const { NODE_OPTIONS, ...envWithoutNodeOptions } = process.env
+
+    log.debug('Starting Test Visibility - Dynamic Instrumentation client...')
+
+    this.worker = new Worker(
+      join(__dirname, 'worker', 'index.js'),
+      {
+        execArgv: [],
+        env: envWithoutNodeOptions,
+        workerData: {
+          breakpointSetChannel: this.breakpointSetChannel.port1,
+          breakpointHitChannel: this.breakpointHitChannel.port1
+        },
+        transferList: [this.breakpointSetChannel.port1, this.breakpointHitChannel.port1]
+      }
+    )
+    this.worker.on('online', () => {
+      log.debug('Test Visibility - Dynamic Instrumentation client is ready')
+      this._onReady()
+    })
+
+    // Allow the parent to exit even if the worker is still running
+    this.worker.unref()
+
+    this.breakpointSetChannel.port2.on('message', (message) => {
+      const { probeId } = message
+      const resolve = probeIdToResolveBreakpointSet.get(probeId)
+      if (resolve) {
+        resolve()
+        probeIdToResolveBreakpointSet.delete(probeId)
+      }
+    }).unref()
+
+    this.breakpointHitChannel.port2.on('message', (message) => {
+      const { snapshot } = message
+      const { probe: { id: probeId } } = snapshot
+      const resolve = probeIdToResolveBreakpointHit.get(probeId)
+      if (resolve) {
+        resolve({ snapshot })
+        probeIdToResolveBreakpointHit.delete(probeId)
+      }
+    }).unref()
+  }
+}
+
+module.exports = new TestVisDynamicInstrumentation()

packages/dd-trace/src/ci-visibility/dynamic-instrumentation/worker/index.js

@@ -0,0 +1,90 @@
+'use strict'
+
+const { workerData: { breakpointSetChannel, breakpointHitChannel } } = require('worker_threads')
+// TODO: move debugger/devtools_client/session to common place
+const session = require('../../../debugger/devtools_client/session')
+// TODO: move debugger/devtools_client/snapshot to common place
+const { getLocalStateForCallFrame } = require('../../../debugger/devtools_client/snapshot')
+// TODO: move debugger/devtools_client/state to common place
+const {
+  findScriptFromPartialPath,
+  getStackFromCallFrames
+} = require('../../../debugger/devtools_client/state')
+const log = require('../../../log')
+
+let sessionStarted = false
+
+const breakpointIdToSnapshotId = new Map()
+const breakpointIdToProbe = new Map()
+
+session.on('Debugger.paused', async ({ params: { hitBreakpoints: [hitBreakpoint], callFrames } }) => {
+  const probe = breakpointIdToProbe.get(hitBreakpoint)
+  if (!probe) {
+    log.warn(`No probe found for breakpoint ${hitBreakpoint}`)
+    return session.post('Debugger.resume')
+  }
+
+  const stack = getStackFromCallFrames(callFrames)
+
+  const getLocalState = await getLocalStateForCallFrame(callFrames[0])
+
+  await session.post('Debugger.resume')
+
+  const snapshotId = breakpointIdToSnapshotId.get(hitBreakpoint)
+
+  const snapshot = {
+    id: snapshotId,
+    timestamp: Date.now(),
+    probe: {
+      id: probe.probeId,
+      version: '0',
+      location: probe.location
+    },
+    stack,
+    language: 'javascript'
+  }
+
+  const state = getLocalState()
+  if (state) {
+    snapshot.captures = {
+      lines: { [probe.location.lines[0]]: { locals: state } }
+    }
+  }
+
+  breakpointHitChannel.postMessage({ snapshot })
+})
+
+// TODO: add option to remove breakpoint
+breakpointSetChannel.on('message', async ({ snapshotId, probe: { id: probeId, file, line } }) => {
+  await addBreakpoint(snapshotId, { probeId, file, line })
+  breakpointSetChannel.postMessage({ probeId })
+})
+
+async function addBreakpoint (snapshotId, probe) {
+  if (!sessionStarted) await start()
+  const { file, line } = probe
+
+  probe.location = { file, lines: [String(line)] }
+
+  const script = findScriptFromPartialPath(file)
+  if (!script) throw new Error(`No loaded script found for ${file}`)
+
+  const [path, scriptId] = script
+
+  log.debug(`Adding breakpoint at ${path}:${line}`)
+
+  const { breakpointId } = await session.post('Debugger.setBreakpoint', {
+    location: {
+      scriptId,
+      lineNumber: line - 1
+    }
+  })
+
+  breakpointIdToProbe.set(breakpointId, probe)
+  breakpointIdToSnapshotId.set(breakpointId, snapshotId)
+}
+
+function start () {
+  sessionStarted = true
+  return session.post('Debugger.enable') // return instead of await to reduce number of promises created
+}

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

@@ -5,7 +5,7 @@ const { breakpoints } = require('./state')
 const session = require('./session')
 const { getLocalStateForCallFrame } = require('./snapshot')
 const send = require('./send')
-const { getScriptUrlFromId } = require('./state')
+const { getStackFromCallFrames } = require('./state')
 const { ackEmitting, ackError } = require('./status')
 const { parentThreadId } = require('./config')
 const log = require('../../log')
@@ -66,16 +66,7 @@ session.on('Debugger.paused', async ({ params }) => {
     thread_name: threadName
   }
 
-  const stack = params.callFrames.map((frame) => {
-    let fileName = getScriptUrlFromId(frame.location.scriptId)
-    if (fileName.startsWith('file://')) fileName = fileName.substr(7) // TODO: This might not be required
-    return {
-      fileName,
-      function: frame.functionName,
-      lineNumber: frame.location.lineNumber + 1, // Beware! lineNumber is zero-indexed
-      columnNumber: frame.location.columnNumber + 1 // Beware! columnNumber is zero-indexed
-    }
-  })
+  const stack = getStackFromCallFrames(params.callFrames)
 
   // TODO: Send multiple probes in one HTTP request as an array (DEBUG-2848)
   for (const probe of probes) {

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

@@ -32,8 +32,17 @@ module.exports = {
       .sort(([a], [b]) => a.length - b.length)[0]
   },
 
-  getScriptUrlFromId (id) {
-    return scriptUrls.get(id)
+  getStackFromCallFrames (callFrames) {
+    return callFrames.map((frame) => {
+      let fileName = scriptUrls.get(frame.location.scriptId)
+      if (fileName.startsWith('file://')) fileName = fileName.substr(7) // TODO: This might not be required
+      return {
+        fileName,
+        function: frame.functionName,
+        lineNumber: frame.location.lineNumber + 1, // Beware! lineNumber is zero-indexed
+        columnNumber: frame.location.columnNumber + 1 // Beware! columnNumber is zero-indexed
+      }
+    })
   }
 }
 

packages/dd-trace/test/ci-visibility/dynamic-instrumentation/dynamic-instrumentation.spec.js

@@ -0,0 +1,47 @@
+'use strict'
+
+require('../../../../dd-trace/test/setup/tap')
+
+const { fork } = require('child_process')
+const path = require('path')
+
+const { assert } = require('chai')
+
+describe('test visibility with dynamic instrumentation', () => {
+  // Dynamic Instrumentation - Test Visibility not currently supported for windows
+  if (process.platform === 'win32') {
+    return
+  }
+  let childProcess
+
+  afterEach(() => {
+    if (childProcess) {
+      childProcess.kill()
+    }
+  })
+
+  it('can grab local variables', (done) => {
+    childProcess = fork(path.join(__dirname, 'target-app', 'test-visibility-dynamic-instrumentation-script.js'))
+
+    childProcess.on('message', ({ snapshot: { language, stack, probe, captures }, snapshotId }) => {
+      assert.exists(snapshotId)
+      assert.exists(probe)
+      assert.exists(stack)
+      assert.equal(language, 'javascript')
+
+      assert.deepEqual(captures, {
+        lines: {
+          9: {
+            locals: {
+              a: { type: 'number', value: '1' },
+              b: { type: 'number', value: '2' },
+              localVar: { type: 'number', value: '1' }
+            }
+          }
+        }
+      })
+
+      done()
+    })
+  })
+})

packages/dd-trace/test/ci-visibility/dynamic-instrumentation/target-app/di-dependency.js

@@ -0,0 +1,10 @@
+'use strict'
+
+module.exports = function (a, b) {
+  // eslint-disable-next-line no-console
+  const localVar = 1
+  if (a > 10) {
+    throw new Error('a is too big')
+  }
+  return a + b + localVar // location of the breakpoint
+}

packages/dd-trace/test/ci-visibility/dynamic-instrumentation/target-app/test-visibility-dynamic-instrumentation-script.js

@@ -0,0 +1,29 @@
+'use strict'
+
+const path = require('path')
+const tvDynamicInstrumentation = require('../../../../src/ci-visibility/dynamic-instrumentation')
+const sum = require('./di-dependency')
+
+// keep process alive
+const intervalId = setInterval(() => {}, 5000)
+
+tvDynamicInstrumentation.start()
+
+tvDynamicInstrumentation.isReady().then(() => {
+  const [
+    snapshotId,
+    breakpointSetPromise,
+    breakpointHitPromise
+  ] = tvDynamicInstrumentation.addLineProbe({ file: path.join(__dirname, 'di-dependency.js'), line: 9 })
+
+  breakpointHitPromise.then(({ snapshot }) => {
+    // once the breakpoint is hit, we can grab the snapshot and send it to the parent process
+    process.send({ snapshot, snapshotId })
+    clearInterval(intervalId)
+  })
+
+  // We run the code once the breakpoint is set
+  breakpointSetPromise.then(() => {
+    sum(1, 2)
+  })
+})