perf(debugger): sample probes in breakpoint conditions (#8967)

Move probe sampling into generated breakpoint conditions so unsampled hits can avoid pausing
the application. The devtools worker now reads sampled probe indexes from a shared buffer
instead of evaluating conditions and sampling after every pause.

Install a runtime sampler in the debuggee from the debugger bootstrap, hand the shared buffer
to the worker, and uninstall it on cleanup so the sampler survives debugger stop/start. Track
probe indexes in worker state and clear per-probe sampler state when a probe is removed. This
keeps per-probe sampling, user conditions, and the global snapshot sampling limit in the
condition path.

Rework the sirun debugger benchmark to exercise the new condition path: add a second
breakpoint target, make the startup-guard share and operation counts configurable, drop the
install-only variant, and honor per-variant iteration counts in the overview collector.

Add focused coverage for sampler expressions, breakpoint condition rebuilding, pause handling,
overflow handling, stop/start, and stale sampled probe indexes.

Author: watson

benchmark/sirun/collect-overview.js

@@ -101,6 +101,7 @@ for (const name of benches) {
     try { fs.unlinkSync(SG_FILE) } catch {}
 
     const variantCfg = meta.variants?.[variant] || {}
+    const variantIters = variantCfg.iterations || configIters
     const inner = innerCount(variantCfg.env)
     let perIterMs = '-'
     let stddevPct = '-'
@@ -120,7 +121,7 @@ for (const name of benches) {
           const { mean, stddevPct: sd } = stats(times)
           perIterMs = (mean / 1000).toFixed(1)
           stddevPct = sd.toFixed(1)
-          totalS = ((mean / 1e6) * configIters).toFixed(0)
+          totalS = ((mean / 1e6) * variantIters).toFixed(0)
         }
       } catch { perIterMs = 'parse-err' }
     } else {
@@ -131,7 +132,7 @@ for (const name of benches) {
 
     fs.appendFileSync(OUT,
       `| ${name} | ${variant} | ${categoryOf(name)} | ${meaningOf(name)} | ${inner} | ` +
-      `${configIters} | ${perIterMs} | ${stddevPct} | ${totalS} | ${startupPct} |\n`)
+      `${variantIters} | ${perIterMs} | ${stddevPct} | ${totalS} | ${startupPct} |\n`)
     console.log(`${name}/${variant} ${perIterMs}ms sd=${stddevPct}% ${totalS}s start=${startupPct}%`)
   }
 

benchmark/sirun/debugger/app.js

@@ -1,12 +1,13 @@
 'use strict'
 
-// WARNING: the breakpoint target below is referenced by line number from
+// WARNING: the breakpoint targets below are referenced by line number from
 // meta.json (BREAKPOINT_LINE). Update the BREAKPOINT_LINE values there if you
-// move the `data.n = n` line.
+// move the `data.n = n` line or the unreachable `return n` line.
 
 const guard = require('../startup-guard')
 
 const OPERATIONS = Number(process.env.OPERATIONS)
+const STARTUP_GUARD_MAX_SHARE = Number(process.env.STARTUP_GUARD_MAX_SHARE)
 
 if (process.env.DD_DYNAMIC_INSTRUMENTATION_ENABLED === 'true') {
   // The devtools worker and its ports are unref'd, so nothing holds the event
@@ -16,9 +17,7 @@ if (process.env.DD_DYNAMIC_INSTRUMENTATION_ENABLED === 'true') {
   const keepAlive = setInterval(() => {}, 2 ** 31 - 1)
   require('./start-devtools-client')(() => {
     clearInterval(keepAlive)
-    // The not-hit variant only measures the passive cost of installing a probe,
-    // so it exits here instead of running the guarded work loop.
-    if (process.env.INSTALL_ONLY !== 'true') runWork()
+    runWork()
   })
 } else {
   runWork()
@@ -29,12 +28,15 @@ function runWork () {
   for (let i = 0; i < OPERATIONS; i++) {
     doSomeWork(i)
   }
-  guard.done(0.35)
+  guard.done(STARTUP_GUARD_MAX_SHARE)
 }
 
 function doSomeWork (n) {
   const data = getSomeData()
-  data.n = n
+  data.n = n // BREAKPOINT HERE!
+  if (n < 0) {
+    return n // BREAKPOINT HERE!
+  }
   return data.n
 }
 

benchmark/sirun/debugger/meta.json

@@ -1,7 +1,7 @@
 {
   "name": "debugger",
   "cachegrind": false,
-  "iterations": 10,
+  "iterations": 5,
   "instructions": true,
   "variants": {
     "enabled-but-breakpoint-not-hit": {
@@ -10,10 +10,11 @@
       "run": "node app.js",
       "run_with_affinity": "bash -c \"taskset -c $CPU_AFFINITY node app.js\"",
       "env": {
+        "STARTUP_GUARD_MAX_SHARE": "0.1",
+        "OPERATIONS": "220000",
         "DD_DYNAMIC_INSTRUMENTATION_ENABLED": "true",
-        "INSTALL_ONLY": "true",
         "BREAKPOINT_FILE": "app.js",
-        "BREAKPOINT_LINE": "37"
+        "BREAKPOINT_LINE": "38"
       }
     },
     "line-probe-without-snapshot": {
@@ -22,44 +23,45 @@
       "run": "node app.js",
       "run_with_affinity": "bash -c \"taskset -c $CPU_AFFINITY node app.js\"",
       "env": {
+        "STARTUP_GUARD_MAX_SHARE": "0.1",
+        "OPERATIONS": "32000",
         "DD_DYNAMIC_INSTRUMENTATION_ENABLED": "true",
         "BREAKPOINT_FILE": "app.js",
-        "BREAKPOINT_LINE": "37",
-        "OPERATIONS": "3500"
+        "BREAKPOINT_LINE": "36"
       }
     },
     "line-probe-with-snapshot-default": {
       "setup": "bash -c \"nohup node agent.js >/dev/null 2>&1 &\"",
       "setup_with_affinity": "bash -c \"nohup taskset -c $CPU_AFFINITY node agent.js >/dev/null 2>&1 &\"",
       "run": "node app.js",
       "run_with_affinity": "bash -c \"taskset -c $CPU_AFFINITY node app.js\"",
-      "iterations": 5,
       "env": {
+        "STARTUP_GUARD_MAX_SHARE": "0.1",
+        "OPERATIONS": "32000",
         "DD_DYNAMIC_INSTRUMENTATION_ENABLED": "true",
         "BREAKPOINT_FILE": "app.js",
-        "BREAKPOINT_LINE": "37",
+        "BREAKPOINT_LINE": "36",
         "CAPTURE_SNAPSHOT": "true",
-        "MAX_SNAPSHOTS_PER_SECOND_GLOBALLY": "1000000",
-        "OPERATIONS": "2450"
+        "MAX_SNAPSHOTS_PER_SECOND_GLOBALLY": "1000000"
       }
     },
     "line-probe-with-snapshot-minimal": {
       "setup": "bash -c \"nohup node agent.js >/dev/null 2>&1 &\"",
       "setup_with_affinity": "bash -c \"nohup taskset -c $CPU_AFFINITY node agent.js >/dev/null 2>&1 &\"",
       "run": "node app.js",
       "run_with_affinity": "bash -c \"taskset -c $CPU_AFFINITY node app.js\"",
-      "iterations": 7,
       "env": {
+        "STARTUP_GUARD_MAX_SHARE": "0.1",
+        "OPERATIONS": "32000",
         "DD_DYNAMIC_INSTRUMENTATION_ENABLED": "true",
         "BREAKPOINT_FILE": "app.js",
-        "BREAKPOINT_LINE": "37",
+        "BREAKPOINT_LINE": "36",
         "CAPTURE_SNAPSHOT": "true",
         "MAX_SNAPSHOTS_PER_SECOND_GLOBALLY": "1000000",
         "MAX_REFERENCE_DEPTH": "0",
         "MAX_COLLECTION_SIZE": "0",
         "MAX_FIELD_COUNT": "0",
-        "MAX_LENGTH": "9007199254740991",
-        "OPERATIONS": "3500"
+        "MAX_LENGTH": "9007199254740991"
       }
     }
   }

integration-tests/debugger/sampling.spec.js

@@ -28,10 +28,10 @@ describe('Dynamic Instrumentation', function () {
             const duration = timestamp - prev
             clearTimeout(timer)
 
-            // The sampling check inside the worker uses `process.hrtime.bigint()` (monotonic), but the snapshot
-            // `timestamp` is captured via `Date.now()` (wall clock). NTP slewing on CI runners can cause the wall clock
-            // to drift slightly relative to the monotonic clock during the >=1s sampling window, so we allow a 75ms
-            // tolerance on both sides of the expected 1000ms gap.
+            // The sampling check uses `process.hrtime.bigint()` (monotonic), but the snapshot `timestamp` is captured
+            // via `Date.now()` (wall clock). NTP slewing on CI runners can cause the wall clock to drift slightly
+            // relative to the monotonic clock during the >=1s sampling window, so we allow a 75ms tolerance on both
+            // sides of the expected 1000ms gap.
             assert.ok(duration >= 925, `duration (${duration}) should be >= 925`)
             assert.ok(duration < 1075, `duration (${duration}) should be < 1075`)
 
@@ -84,10 +84,10 @@ describe('Dynamic Instrumentation', function () {
             const duration = timestamp - _state.prev
             clearTimeout(_state.timer)
 
-            // The sampling check inside the worker uses `process.hrtime.bigint()` (monotonic), but the snapshot
-            // `timestamp` is captured via `Date.now()` (wall clock). NTP slewing on CI runners can cause the wall clock
-            // to drift slightly relative to the monotonic clock during the >=1s sampling window, so we allow a 75ms
-            // tolerance on both sides of the expected 1000ms gap.
+            // The sampling check uses `process.hrtime.bigint()` (monotonic), but the snapshot `timestamp` is captured
+            // via `Date.now()` (wall clock). NTP slewing on CI runners can cause the wall clock to drift slightly
+            // relative to the monotonic clock during the >=1s sampling window, so we allow a 75ms tolerance on both
+            // sides of the expected 1000ms gap.
             assert.ok(duration >= 925, `duration (${duration}) should be >= 925`)
             assert.ok(duration < 1075, `duration (${duration}) should be < 1075`)
 

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

@@ -5,6 +5,7 @@ const { getGeneratedPosition } = require('./source-maps')
 const session = require('./session')
 const { compile, compileSegments, templateRequiresEvaluation } = require('./condition')
 const { MAX_SNAPSHOTS_PER_SECOND_PER_PROBE, MAX_NON_SNAPSHOTS_PER_SECOND_PER_PROBE } = require('./defaults')
+const { compileBreakpointCondition, getRemoveProbeExpression } = require('./probe_sampler')
 const {
   DEFAULT_MAX_REFERENCE_DEPTH,
   DEFAULT_MAX_COLLECTION_SIZE,
@@ -17,6 +18,7 @@ const {
   locationToBreakpoint,
   breakpointToProbes,
   probeToLocation,
+  samplingIndexToProbe,
 } = require('./state')
 const log = require('./log')
 
@@ -26,6 +28,7 @@ const log = require('./log')
 
 let sessionStarted = false
 const probes = new Map()
+let nextSamplingIndex = 0
 let scriptLoadingStabilizedResolve
 const scriptLoadingStabilized = new Promise((resolve) => { scriptLoadingStabilizedResolve = resolve })
 
@@ -55,6 +58,8 @@ async function addBreakpoint (probe) {
   if (!sessionStarted) await start()
 
   probes.set(probe.id, probe)
+  probe.samplingIndex = nextSamplingIndex++
+  samplingIndexToProbe.set(probe.samplingIndex, probe)
 
   const file = probe.where.sourceFile
   let lineNumber = Number(probe.where.lines[0]) // Tracer doesn't support multiple-line breakpoints
@@ -75,8 +80,6 @@ async function addBreakpoint (probe) {
     ? MAX_SNAPSHOTS_PER_SECOND_PER_PROBE
     : MAX_NON_SNAPSHOTS_PER_SECOND_PER_PROBE)
   probe.nsBetweenSampling = BigInt(Math.trunc(1 / snapshotsPerSecond * 1e9))
-  // Initialize to a large negative value to ensure first probe hit is always captured
-  probe.lastCaptureNs = BigInt(Number.MIN_SAFE_INTEGER)
 
   // Warning: The code below relies on undocumented behavior of the inspector!
   // It expects that `await session.post('Debugger.enable')` will wait for all loaded scripts to be emitted as
@@ -174,7 +177,7 @@ async function addBreakpoint (probe) {
     try {
       result = /** @type {SetBreakpointResponse} */ (await session.post('Debugger.setBreakpoint', {
         location,
-        condition: probe.condition,
+        condition: compileBreakpointCondition([probe]),
       }))
     } catch (err) {
       throw new Error(`Error setting breakpoint for probe ${probe.id} (version: ${probe.version})`, { cause: err })
@@ -195,11 +198,14 @@ async function removeBreakpoint ({ id }) {
   }
 
   probes.delete(id)
+  await removeProbeFromSampler(id)
 
   const locationKey = probeToLocation.get(id)
   const breakpoint = locationToBreakpoint.get(locationKey)
   const probesAtLocation = breakpointToProbes.get(breakpoint.id)
+  const probe = probesAtLocation.get(id)
 
+  samplingIndexToProbe.delete(probe.samplingIndex)
   probesAtLocation.delete(id)
   probeToLocation.delete(id)
 
@@ -229,36 +235,37 @@ async function modifyBreakpoint (probe) {
 
 async function updateBreakpointInternal (breakpoint, probe) {
   const probesAtLocation = breakpointToProbes.get(breakpoint.id)
-  const conditionBeforeNewProbe = compileCompoundCondition([...probesAtLocation.values()])
 
-  // If a probe is provided, add it to the breakpoint. If not, it's because we're removing a probe, but potentially
-  // need to update the condition of the breakpoint.
+  // If a probe is provided, add it to the breakpoint. If not, it's because we're removing a probe. In both cases the
+  // breakpoint condition must be rebuilt to match the remaining probes at the location.
   if (probe) {
     probesAtLocation.set(probe.id, probe)
     probeToLocation.set(probe.id, breakpoint.locationKey)
   }
 
-  const condition = compileCompoundCondition([...probesAtLocation.values()])
-
-  if (condition || conditionBeforeNewProbe !== condition) {
-    try {
-      await session.post('Debugger.removeBreakpoint', { breakpointId: breakpoint.id })
-    } catch (err) {
-      throw new Error(`Error removing breakpoint for probe ${probe.id}`, { cause: err })
-    }
-    breakpointToProbes.delete(breakpoint.id)
-    let result
-    try {
-      result = /** @type {SetBreakpointResponse} */ (await session.post('Debugger.setBreakpoint', {
-        location: breakpoint.location,
-        condition,
-      }))
-    } catch (err) {
-      throw new Error(`Error setting breakpoint for probe ${probe.id} (version: ${probe.version})`, { cause: err })
-    }
-    breakpoint.id = result.breakpointId
-    breakpointToProbes.set(result.breakpointId, probesAtLocation)
+  try {
+    await session.post('Debugger.removeBreakpoint', { breakpointId: breakpoint.id })
+  } catch (err) {
+    const message = probe
+      ? `Error replacing breakpoint while adding probe ${probe.id} (version: ${probe.version})`
+      : `Error replacing breakpoint after removing probe from ${breakpoint.locationKey}`
+    throw new Error(message, { cause: err })
+  }
+  breakpointToProbes.delete(breakpoint.id)
+  let result
+  try {
+    result = /** @type {SetBreakpointResponse} */ (await session.post('Debugger.setBreakpoint', {
+      location: breakpoint.location,
+      condition: compileBreakpointCondition([...probesAtLocation.values()]),
+    }))
+  } catch (err) {
+    const message = probe
+      ? `Error setting breakpoint while adding probe ${probe.id} (version: ${probe.version})`
+      : `Error setting breakpoint after removing probe from ${breakpoint.locationKey}`
+    throw new Error(message, { cause: err })
   }
+  breakpoint.id = result.breakpointId
+  breakpointToProbes.set(result.breakpointId, probesAtLocation)
 }
 
 async function reEvaluateProbe (probe) {
@@ -306,16 +313,20 @@ function lock (fn) {
   }
 }
 
-// Only if all probes have a condition can we use a compound condition.
-// Otherwise, we need to evaluate each probe individually once the breakpoint is hit.
-function compileCompoundCondition (probes) {
-  if (probes.length === 1) return probes[0].condition
-
-  return probes.every(p => p.condition)
-    ? probes
-      .map((p) => `(() => { try { return ${p.condition} } catch { return false } })()`)
-      .join(' || ')
-    : undefined
+/**
+ * Remove cached sampling state for a probe from the runtime sampler.
+ *
+ * @param {string} id - The probe id.
+ * @returns {Promise<void>}
+ */
+async function removeProbeFromSampler (id) {
+  try {
+    await session.post('Runtime.evaluate', {
+      expression: getRemoveProbeExpression(id),
+    })
+  } catch (err) {
+    log.error('[debugger:devtools_client] Error removing probe %s from sampler', id, err)
+  }
 }
 
 function generateLocationKey (scriptId, lineNumber, columnNumber) {