Ensure advancement of sampling instant never takes much time (#6172)

szegedi · nsavoire · BridgeAR · commit bec72512dffa · 2025-08-06T18:56:44.000+02:00
* Ensure advancement of sampling instant never takes much time

* Extract the poisson filter so it is unit-testable

* Update packages/dd-trace/src/profiling/profilers/events.js

Co-authored-by: Nicolas Savoire &lt;nicolas.savoire@datadoghq.com&gt;

* Fix another place where there's typo

---------

Co-authored-by: Nicolas Savoire &lt;nicolas.savoire@datadoghq.com&gt;
diff --git a/packages/dd-trace/src/profiling/profilers/events.js b/packages/dd-trace/src/profiling/profilers/events.js
@@ -3,9 +3,12 @@
 const { performance, constants, PerformanceObserver } = require('perf_hooks')
 const { END_TIMESTAMP_LABEL, SPAN_ID_LABEL, LOCAL_ROOT_SPAN_ID_LABEL, encodeProfileAsync } = require('./shared')
 const { Function, Label, Line, Location, Profile, Sample, StringTable, ValueType } = require('pprof-format')
+const PoissonProcessSamplingFilter = require('./poisson')
 const { availableParallelism, effectiveLibuvThreadCount } = require('../libuv-size')
 // perf_hooks uses millis, with fractional part representing nanos. We emit nanos into the pprof file.
 const MS_TO_NS = 1_000_000
+// The number of sampling intervals that need to pass before we reset the Poisson process sampling instant.
+const POISSON_RESET_FACTOR = 2
 
 // While this is an "events profiler", meaning it emits a pprof file based on events observed as
 // perf_hooks events, the emitted pprof file uses the type "timeline".
@@ -371,35 +374,12 @@ class DatadogInstrumentationEventSource {
 }
 
 function createPoissonProcessSamplingFilter (samplingIntervalMillis) {
-  let nextSamplingInstant = performance.now()
-  let currentSamplingInstant = 0
-  setNextSamplingInstant()
-
-  return event => {
-    const endTime = event.startTime + event.duration
-    while (endTime >= nextSamplingInstant) {
-      setNextSamplingInstant()
-    }
-    // An event is sampled if it started before, and ended on or after a sampling instant. The above
-    // while loop will ensure that the ending invariant is always true for the current sampling
-    // instant so we don't have to test for it below. Across calls, the invariant also holds as long
-    // as the events arrive in endTime order. This is true for events coming from
-    // DatadogInstrumentationEventSource; they will be ordered by endTime by virtue of this method
-    // being invoked synchronously with the plugins' finish() handler which evaluates
-    // performance.now(). OTOH, events coming from NodeAPIEventSource (GC in typical setup) might be
-    // somewhat delayed as they are queued by Node, so they can arrive out of order with regard to
-    // events coming from the non-queued source. By omitting the endTime check, we will pass through
-    // some short events that started and ended before the current sampling instant. OTOH, if we
-    // were to check for this.currentSamplingInstant <= endTime, we would discard some long events
-    // that also ended before the current sampling instant. We'd rather err on the side of including
-    // some short events than excluding some long events.
-    return event.startTime < currentSamplingInstant
-  }
-
-  function setNextSamplingInstant () {
-    currentSamplingInstant = nextSamplingInstant
-    nextSamplingInstant -= Math.log(1 - Math.random()) * samplingIntervalMillis
-  }
+  const poissonFilter = new PoissonProcessSamplingFilter({
+    samplingInterval: samplingIntervalMillis,
+    resetInterval: samplingIntervalMillis * POISSON_RESET_FACTOR,
+    now: performance.now.bind(performance)
+  })
+  return poissonFilter.filter.bind(poissonFilter)
 }
 
 /**
diff --git a/packages/dd-trace/src/profiling/profilers/poisson.js b/packages/dd-trace/src/profiling/profilers/poisson.js
@@ -0,0 +1,105 @@
+'use strict'
+
+class PoissonProcessSamplingFilter {
+  #currentSamplingInstant = 0
+  #nextSamplingInstant
+  #samplingInterval
+  #resetInterval
+  #now
+  #lastNow = Number.NEGATIVE_INFINITY
+  #samplingInstantCount = 0
+
+  constructor ({ samplingInterval, now, resetInterval }) {
+    if (samplingInterval <= 0) {
+      throw new RangeError(`samplingInterval (${samplingInterval}) must be greater than 0`)
+    }
+    if (resetInterval < samplingInterval) {
+      throw new RangeError(
+        `resetInterval (${resetInterval}) must be greater than samplingInterval (${samplingInterval})`
+      )
+    }
+    if (typeof now !== 'function') {
+      throw new TypeError('now must be a function')
+    }
+    this.#samplingInterval = samplingInterval
+    this.#resetInterval = resetInterval
+    this.#now = now
+    this.#nextSamplingInstant = this.#callNow()
+    this.#setNextSamplingInstant()
+  }
+
+  get currentSamplingInstant () {
+    return this.#currentSamplingInstant
+  }
+
+  get nextSamplingInstant () {
+    return this.#nextSamplingInstant
+  }
+
+  get samplingInstantCount () {
+    return this.#samplingInstantCount
+  }
+
+  #callNow () {
+    const nowValue = this.#now()
+    if (typeof nowValue !== 'number') {
+      throw new TypeError('now() must return a number')
+    }
+    if (nowValue < this.#lastNow) {
+      throw new RangeError('now() must return a value greater than or equal to the last returned value')
+    }
+    this.#lastNow = nowValue
+    return nowValue
+  }
+
+  filter (event) {
+    const endTime = event.startTime + event.duration
+    // We're using the end times of events as an approximation of current time as events are
+    // expected to be reported close to where they ended. If the end time (and thus, presumably, the
+    // current time) is past the next sampling instant, we make it the current sampling instant and
+    // compute the next sampling instant in its future.
+    if (endTime >= this.#nextSamplingInstant) {
+      // All observed events are supposed to have happened in the past. For purposes of advancing
+      // the next sampling instant, we cap endTime to now(). This protects us from advancing it far
+      // into future if we receive an event with erroneously long duration, which would also take
+      // many iterations of the below "while" loop.
+      const cappedEndTime = Math.min(endTime, this.#callNow())
+
+      // If nextSamplingInstant is far in cappedEndTime's past, first advance it close to it. This
+      // can happen if we didn't receive any events for a while. Since a Poisson process has no
+      // memory, we can reset it anytime. This will ensure that the "while" loop below runs at most
+      // few iterations.
+      const earliestContinuousPast = cappedEndTime - this.#resetInterval
+      if (this.#nextSamplingInstant < earliestContinuousPast) {
+        this.#nextSamplingInstant = earliestContinuousPast
+      }
+
+      // Advance the next sampling instant until it is in cappedEndTime's future.
+      while (cappedEndTime >= this.#nextSamplingInstant) {
+        this.#setNextSamplingInstant()
+      }
+    }
+    // An event is sampled if it started before, and ended on or after a sampling instant. The above
+    // while loop will ensure that the ending invariant is always true for the current sampling
+    // instant so we don't have to test for it below. Across calls, the invariant also holds as long
+    // as the events arrive in endTime order. This is true for events coming from
+    // DatadogInstrumentationEventSource; they will be ordered by endTime by virtue of this method
+    // being invoked synchronously with the plugins' finish() handler which evaluates
+    // performance.now(). OTOH, events coming from NodeAPIEventSource (GC in typical setup) might be
+    // somewhat delayed as they are queued by Node, so they can arrive out of order with regard to
+    // events coming from the non-queued source. By omitting the endTime check, we will pass through
+    // some short events that started and ended before the current sampling instant. OTOH, if we
+    // were to check for this.currentSamplingInstant <= endTime, we would discard some long events
+    // that also ended before the current sampling instant. We'd rather err on the side of including
+    // some short events than excluding some long events.
+    return event.startTime < this.#currentSamplingInstant
+  }
+
+  #setNextSamplingInstant () {
+    this.#currentSamplingInstant = this.#nextSamplingInstant
+    this.#nextSamplingInstant -= Math.log(1 - Math.random()) * this.#samplingInterval
+    this.#samplingInstantCount++
+  }
+}
+
+module.exports = PoissonProcessSamplingFilter
diff --git a/packages/dd-trace/test/profiling/profilers/poisson.spec.js b/packages/dd-trace/test/profiling/profilers/poisson.spec.js
@@ -0,0 +1,183 @@
+'use strict'
+
+require('../../setup/tap')
+const assert = require('assert')
+
+const PoissonProcessSamplingFilter = require('../../../src/profiling/profilers/poisson')
+
+describe('PoissonProcessSamplingFilter', () => {
+  let nowValue
+  const now = () => nowValue
+
+  beforeEach(() => {
+    nowValue = 0
+  })
+
+  it('should throw if resetInterval < samplingInterval', () => {
+    assert.throws(() => new PoissonProcessSamplingFilter({
+      samplingInterval: 100,
+      resetInterval: 50,
+      now
+    }), RangeError)
+  })
+
+  it('should throw if now is not a function', () => {
+    assert.throws(() => new PoissonProcessSamplingFilter({
+      samplingInterval: 100,
+      resetInterval: 200,
+      now: 123
+    }), (err) => err instanceof TypeError && err.message === 'now must be a function')
+  })
+
+  it('should throw if now() returns a non-number', () => {
+    const badNow = () => 'not-a-number'
+    assert.throws(() => new PoissonProcessSamplingFilter({
+      samplingInterval: 100,
+      resetInterval: 200,
+      now: badNow
+    }), (err) => err instanceof TypeError && err.message === 'now() must return a number')
+  })
+
+  it('should throw if now() returns a decreasing value', () => {
+    let callCount = 0
+    const decreasingNow = () => {
+      callCount++
+      return callCount === 1 ? 100 : 50
+    }
+    const filter = new PoissonProcessSamplingFilter({
+      samplingInterval: 10,
+      resetInterval: 20,
+      now: decreasingNow
+    })
+    const event = { startTime: 0, duration: filter.nextSamplingInstant + 1 }
+    assert.throws(() => filter.filter(event), (err) => err instanceof RangeError &&
+      err.message === 'now() must return a value greater than or equal to the last returned value')
+  })
+
+  it('should allow now() to return the same value as last time', () => {
+    let callCount = 0
+    const constantNow = () => {
+      callCount++
+      return 42
+    }
+    const filter = new PoissonProcessSamplingFilter({
+      samplingInterval: 10,
+      resetInterval: 20,
+      now: constantNow
+    })
+    const event = { startTime: 0, duration: 100 }
+    assert.doesNotThrow(() => filter.filter(event))
+    assert.strictEqual(callCount, 2)
+  })
+
+  it('should initialize with correct properties', () => {
+    const filter = new PoissonProcessSamplingFilter({
+      samplingInterval: 100,
+      resetInterval: 200,
+      now
+    })
+    assert.strictEqual(typeof filter.currentSamplingInstant, 'number')
+    assert.strictEqual(filter.currentSamplingInstant, 0)
+    assert.strictEqual(typeof filter.nextSamplingInstant, 'number')
+    assert.ok(filter.nextSamplingInstant > 0)
+    assert.strictEqual(filter.samplingInstantCount, 1)
+  })
+
+  it('should advance sampling instant when event endTime >= nextSamplingInstant', () => {
+    const filter = new PoissonProcessSamplingFilter({
+      samplingInterval: 100,
+      resetInterval: 200,
+      now
+    })
+    const prevNextSamplingInstant = filter.nextSamplingInstant
+    const event = {
+      startTime: -10,
+      duration: prevNextSamplingInstant + 15
+    }
+    assert.strictEqual(filter.currentSamplingInstant, 0)
+    nowValue = prevNextSamplingInstant + 15
+    filter.filter(event)
+    assert.ok(filter.nextSamplingInstant > prevNextSamplingInstant)
+    assert.ok(filter.currentSamplingInstant > 0)
+    assert.ok(filter.samplingInstantCount > 1)
+  })
+
+  it('should not advance sampling instant if event endTime < nextSamplingInstant', () => {
+    const filter = new PoissonProcessSamplingFilter({
+      samplingInterval: 100,
+      resetInterval: 200,
+      now
+    })
+    const prevSamplingInstant = filter.currentSamplingInstant
+    const prevNextSamplingInstant = filter.nextSamplingInstant
+    const event = { startTime: prevSamplingInstant - 10, duration: 1 }
+    filter.filter(event)
+    assert.strictEqual(filter.currentSamplingInstant, prevSamplingInstant)
+    assert.strictEqual(filter.nextSamplingInstant, prevNextSamplingInstant)
+    assert.strictEqual(filter.samplingInstantCount, 1)
+  })
+
+  it('should cap endTime to now() if event endTime is in the future', () => {
+    const filter = new PoissonProcessSamplingFilter({
+      samplingInterval: 100,
+      resetInterval: 200,
+      now
+    })
+    const prevNextSamplingInstant = filter.nextSamplingInstant
+    nowValue = 1000
+    const event = { startTime: 0, duration: 1e6 }
+    filter.filter(event)
+    assert.ok(filter.currentSamplingInstant >= prevNextSamplingInstant)
+    assert.strictEqual(typeof filter.nextSamplingInstant, 'number')
+    assert.ok(filter.nextSamplingInstant < 500000)
+    assert.ok(filter.samplingInstantCount < 10)
+  })
+
+  it('should reset nextSamplingInstant if it is too far in the past', () => {
+    const filter = new PoissonProcessSamplingFilter({
+      samplingInterval: 100,
+      resetInterval: 200,
+      now
+    })
+    const event = { startTime: 100000, duration: 100 }
+    nowValue = event.startTime + event.duration
+    filter.filter(event)
+    assert.ok(filter.nextSamplingInstant > nowValue)
+    assert.ok(filter.samplingInstantCount < 10)
+  })
+
+  it('should return true if event.startTime < currentSamplingInstant', () => {
+    const filter = new PoissonProcessSamplingFilter({
+      samplingInterval: 100,
+      resetInterval: 200,
+      now
+    })
+    const event = { startTime: filter.currentSamplingInstant - 1, duration: 1 }
+    assert.strictEqual(filter.filter(event), true)
+  })
+
+  it('should return false if event.startTime >= currentSamplingInstant', () => {
+    const filter = new PoissonProcessSamplingFilter({
+      samplingInterval: 100,
+      resetInterval: 200,
+      now
+    })
+    const event = { startTime: filter.currentSamplingInstant, duration: 1 }
+    assert.strictEqual(filter.filter(event), false)
+  })
+
+  it('should increment samplingInstantCount on each sampling instant', () => {
+    const filter = new PoissonProcessSamplingFilter({
+      samplingInterval: 10,
+      resetInterval: 100,
+      now
+    })
+    const initialCount = filter.samplingInstantCount
+    for (let i = 0; i < 5; i++) {
+      nowValue += 20
+      const event = { startTime: 0, duration: filter.nextSamplingInstant - 0 }
+      filter.filter(event)
+    }
+    assert.ok(filter.samplingInstantCount > initialCount)
+  })
+})
