feat(debugger): report guardrail telemetry metrics

benchmark/sirun/debugger/benchmark-worker.js

@@ -92,15 +92,17 @@ function markProbeHandled () {
  * @param {Record<string, unknown> | undefined} dd
  * @param {DebuggerSnapshot} snapshot
  * @param {string | undefined} processTags
+ * @param {number} eventType
+ * @param {number} incompleteReasons
  * @returns {void}
  */
-function sendAndCount (message, logger, dd, snapshot, processTags) {
+function sendAndCount (message, logger, dd, snapshot, processTags, eventType, incompleteReasons) {
   const captureKind = getCaptureKind(snapshot)
   if (preflightPending) {
     preflightPending = false
     Atomics.store(probeCounts, CAPTURE_KIND_INDEX, captureKind)
   } else {
-    send(message, logger, dd, snapshot, processTags)
+    send(message, logger, dd, snapshot, processTags, eventType, incompleteReasons)
     if (captureKind === EXPECTED_CAPTURE_KIND) Atomics.add(probeCounts, MATCHED_CAPTURE_KIND_INDEX, 1)
   }
 

integration-tests/debugger/guardrail-metrics.spec.js

@@ -0,0 +1,103 @@
+'use strict'
+
+const assert = require('node:assert/strict')
+const { inspect } = require('node:util')
+
+const { setup } = require('./utils')
+
+// The guardrail counters are converted into telemetry metrics every 10 seconds, which are then sent on the next
+// telemetry heartbeat
+const GUARDRAIL_METRICS_FLUSH_INTERVAL_MS = 10_000
+const TELEMETRY_HEARTBEAT_INTERVAL_SECONDS = 1
+
+describe('Dynamic Instrumentation', function () {
+  const t = setup({
+    testApp: 'target-app/basic.js',
+    dependencies: ['fastify'],
+    env: {
+      DD_TELEMETRY_HEARTBEAT_INTERVAL: String(TELEMETRY_HEARTBEAT_INTERVAL_SECONDS),
+      // The app-started event is sent before the test can listen for it, but the extended heartbeat repeats its
+      // application payload
+      DD_TELEMETRY_EXTENDED_HEARTBEAT_INTERVAL: String(TELEMETRY_HEARTBEAT_INTERVAL_SECONDS),
+    },
+  })
+
+  describe('guardrail telemetry', function () {
+    this.timeout(GUARDRAIL_METRICS_FLUSH_INTERVAL_MS * 3)
+
+    it('should report Dynamic Instrumentation as an enabled product', async function () {
+      await t.agent.assertTelemetryReceived({
+        fn: ({ payload }) => {
+          assert.deepStrictEqual(payload.payload.products.dynamic_instrumentation, { enabled: true })
+        },
+        requestType: 'app-extended-heartbeat',
+      })
+    })
+
+    it('should report skipped events and incomplete captures', async function () {
+      // A log probe that is hit far more often than its per-probe rate limit allows
+      const rateLimitedProbe = t.breakpoints[0].generateRemoteConfig({ sampling: { snapshotsPerSecond: 1 } })
+      // A snapshot probe that cannot capture the request object within its reference depth limit
+      const shallowSnapshotProbe = t.breakpoints[1].generateRemoteConfig({
+        captureSnapshot: true,
+        capture: { maxReferenceDepth: 0 },
+      })
+
+      const installed = new Set()
+      const allInstalled = new Promise((/** @type {(value?: void) => void} */ resolve) => {
+        t.agent.on('debugger-diagnostics', ({ payload }) => {
+          for (const { debugger: { diagnostics: { probeId, status } } } of payload) {
+            if (status === 'INSTALLED') installed.add(probeId)
+          }
+          if (installed.size === 2) resolve()
+        })
+      })
+
+      const checkMetrics = t.agent.assertTelemetryReceived({
+        fn: ({ payload }) => {
+          const { series } = payload.payload
+
+          const skipped = findMetric(series, 'events.skipped', ['event_type:log', 'reason:rateLimitProbe'])
+          assert.strictEqual(skipped.type, 'count')
+          assert.strictEqual(skipped.common, true)
+          assert.strictEqual(skipped.points.length, 1)
+          assert.ok(skipped.points[0][1] >= 1, `Expected ${skipped.points[0][1]} >= 1`)
+
+          const incomplete = findMetric(series, 'capture.incomplete', ['event_type:snapshot', 'reason:depth'])
+          assert.strictEqual(incomplete.type, 'count')
+          assert.strictEqual(incomplete.common, true)
+          assert.strictEqual(incomplete.points.length, 1)
+          assert.strictEqual(incomplete.points[0][1], 1)
+        },
+        requestType: 'generate-metrics',
+        timeout: GUARDRAIL_METRICS_FLUSH_INTERVAL_MS * 2,
+        resolveAtFirstSuccess: true,
+        namespace: 'live_debugger',
+      })
+
+      t.agent.addRemoteConfig(rateLimitedProbe)
+      t.agent.addRemoteConfig(shallowSnapshotProbe)
+      await allInstalled
+
+      // Trigger the rate limited probe well within a single second, so all but the first hit are skipped
+      await Promise.all(Array.from({ length: 5 }, () => t.request(t.breakpoints[0].url)))
+      await t.request(t.breakpoints[1].url)
+
+      await checkMetrics
+    })
+  })
+})
+
+/**
+ * @param {Array<{ metric: string, tags: string[] }>} series - The telemetry metric series
+ * @param {string} metric - The metric name to find
+ * @param {string[]} tags - The exact tags the metric must carry
+ * @returns {object} The matching series entry
+ */
+function findMetric (series, metric, tags) {
+  const match = series.find((entry) => {
+    return entry.metric === metric && entry.tags.length === tags.length && tags.every((tag) => entry.tags.includes(tag))
+  })
+  assert.ok(match, `Expected metric ${metric} with tags ${inspect(tags)} in ${inspect(series)}`)
+  return match
+}

packages/dd-trace/src/debugger/devtools_client/guardrail-metrics.js

@@ -0,0 +1,15 @@
+'use strict'
+
+const { workerData } = require('node:worker_threads')
+
+const { GuardrailMetrics } = require('../guardrail-metrics')
+
+// For testing purposes, we allow `workerData` to be undefined and fallback to counters that are never drained
+const buffer = workerData?.guardrailMetricsBuffer ?? GuardrailMetrics.createBuffer()
+
+/**
+ * The worker's view of the guardrail counters shared with the main thread, which drains them into telemetry.
+ *
+ * @type {GuardrailMetrics}
+ */
+module.exports = new GuardrailMetrics(buffer)

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

@@ -5,6 +5,7 @@ const { workerData: { probeSamplerBuffer } } = require('worker_threads')
 const { version } = require('../../../../../package.json')
 const processTags = require('../../process-tags')
 const { INSPECT_SEGMENT_GLOBAL_PROPERTY } = require('../constants')
+const { EVENT_TYPE, INCOMPLETE_REASON } = require('../guardrail-metrics')
 const {
   MAX_SAMPLED_PROBES_PER_PAUSE,
   SAMPLED_PROBE_COUNT_INDEX,
@@ -140,17 +141,14 @@ session.on('Debugger.paused', async ({ params }) => {
   }
 
   // TODO: Create unique states for each affected probe based on that probes unique `capture` settings (DEBUG-2863)
-  let processLocalState
-  /** @type {Error[] | undefined} */
-  let fatalSnapshotErrors
+  /** @type {Awaited<ReturnType<typeof getLocalStateForCallFrame>> | undefined} */
+  let localState
   if (numberOfProbesWithSnapshots !== 0) {
-    const result = await getLocalStateForCallFrame(
+    localState = await getLocalStateForCallFrame(
       params.callFrames[0],
       { maxReferenceDepth, maxCollectionSize, maxFieldCount, maxLength },
       start + config.dynamicInstrumentation.captureTimeoutNs
     )
-    processLocalState = result.processLocalState
-    fatalSnapshotErrors = result.fatalErrors
   }
 
   // Evaluate capture expressions for probes that have them
@@ -206,19 +204,30 @@ session.on('Debugger.paused', async ({ params }) => {
       language: 'javascript',
     }
 
+    // Which guardrail bucket the event belongs to, and which capture limits were enforced while producing it. The
+    // snapshot module records the reasons, including runtime errors: a fatal error does not necessarily mean one, as
+    // the collector also raises a fatal error to disable capture when it hits its large object safety threshold.
+    /** @type {number} */
+    let eventType = EVENT_TYPE.LOG
+    let incompleteReasons = 0
+
     if (probe.captureSnapshot) {
-      if (fatalSnapshotErrors && fatalSnapshotErrors.length > 0) {
+      eventType = EVENT_TYPE.SNAPSHOT
+      const { processLocalState, fatalErrors, incomplete } = /** @type {NonNullable<typeof localState>} */ (localState)
+      if (fatalErrors.length > 0) {
         // There was an error collecting the snapshot for this probe, let's not try again
         probe.captureSnapshot = false
-        probe.permanentEvaluationErrors = fatalSnapshotErrors.map(error => ({
+        probe.permanentEvaluationErrors = fatalErrors.map(error => ({
           expr: '',
           message: error.message,
         }))
       }
       snapshot.captures = {
-        lines: { [probe.location.lines[0]]: { locals: /** @type {Function} */ (processLocalState)() } },
+        lines: { [probe.location.lines[0]]: { locals: processLocalState() } },
       }
+      incompleteReasons |= incomplete.reasons
     } else if (probe.compiledCaptureExpressions !== undefined) {
+      eventType = EVENT_TYPE.SNAPSHOT
       const expressionResult = /** @type {Map} */ (captureExpressionResults).get(probe.id)
       if (expressionResult) {
         // Handle fatal capture errors - disable capture expressions for this probe permanently
@@ -233,6 +242,7 @@ session.on('Debugger.paused', async ({ params }) => {
         snapshot.captures = {
           lines: { [probe.location.lines[0]]: { captureExpressions: expressionResult.processCaptureExpressions() } },
         }
+        incompleteReasons |= expressionResult.incomplete.reasons
 
         // Handle transient evaluation errors - include in snapshot for this capture
         if (expressionResult.evaluationErrors?.length > 0) {
@@ -249,6 +259,7 @@ session.on('Debugger.paused', async ({ params }) => {
           expr: '',
           message: 'Internal error: capture expression results not found',
         }]
+        incompleteReasons |= INCOMPLETE_REASON.RUNTIME_ERROR
       }
     }
 
@@ -283,7 +294,8 @@ session.on('Debugger.paused', async ({ params }) => {
     ackEmitting(probe)
 
     send(message, logger, dd, snapshot,
-      config.propagateProcessTags.enabled ? processTags.serialized : undefined)
+      config.propagateProcessTags.enabled ? processTags.serialized : undefined,
+      eventType, incompleteReasons)
   }
 })
 

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

@@ -6,9 +6,11 @@ const { stringify } = require('querystring')
 const { version } = require('../../../../../package.json')
 const request = require('../../exporters/common/request')
 const { DEBUGGER_DIAGNOSTICS_V1, DEBUGGER_INPUT_V2 } = require('../constants')
+const { INCOMPLETE_REASON } = require('../guardrail-metrics')
 const log = require('./log')
 const JSONBuffer = require('./json-buffer')
 const config = require('./config')
+const guardrailMetrics = require('./guardrail-metrics')
 const getRequestOptions = require('./request-options')
 const { pruneSnapshot } = require('./snapshot-pruner')
 const buildTags = require('./tags')
@@ -34,14 +36,28 @@ const jsonBuffer = new JSONBuffer({
   onFlush,
 })
 
-function send (message, logger, dd, snapshot, processTags) {
+/**
+ * Queue a probe result for upload.
+ *
+ * @param {string} message - The evaluated log message
+ * @param {object} logger - The logger metadata
+ * @param {object | undefined} dd - The trace and span ids of the active trace, if any
+ * @param {object} snapshot - The snapshot payload
+ * @param {string | undefined} processTags - The serialized process tags, if enabled
+ * @param {number} eventType - The guardrail event type, one of `EVENT_TYPE`
+ * @param {number} incompleteReasons - Bitmask of `INCOMPLETE_REASON` flags enforced while capturing the snapshot
+ */
+function send (message, logger, dd, snapshot, processTags, eventType, incompleteReasons) {
+  if (message?.length > MAX_MESSAGE_LENGTH) {
+    message = message.slice(0, MAX_MESSAGE_LENGTH) + '…'
+    incompleteReasons |= INCOMPLETE_REASON.STRING_LENGTH
+  }
+
   const payload = {
     ddsource,
     hostname,
     service,
-    message: message?.length > MAX_MESSAGE_LENGTH
-      ? message.slice(0, MAX_MESSAGE_LENGTH) + '…'
-      : message,
+    message,
     logger,
     dd,
     process_tags: processTags,
@@ -52,6 +68,7 @@ function send (message, logger, dd, snapshot, processTags) {
   let size = Buffer.byteLength(json)
 
   if (size > MAX_LOG_PAYLOAD_SIZE_BYTES) {
+    incompleteReasons |= INCOMPLETE_REASON.PAYLOAD_TOO_LARGE
     let pruned
     try {
       pruned = pruneSnapshot(json, size, MAX_LOG_PAYLOAD_SIZE_BYTES)
@@ -71,6 +88,8 @@ function send (message, logger, dd, snapshot, processTags) {
   }
 
   jsonBuffer.write(json, size)
+
+  if (incompleteReasons !== 0) guardrailMetrics.captureIncomplete(incompleteReasons, eventType)
 }
 
 /**