fix(otlp): emit non-overlapping delta metric windows

packages/dd-trace/src/opentelemetry/metrics/otlp_span_stats_transformer.js

@@ -68,7 +68,7 @@ class OtlpStatsTransformer extends OtlpTransformerBase {
   }
 
   /**
-   * @param {Array<{timeNs: number, bucket: import('../../span_stats').SpanBuckets}>} drained
+   * @param {Array<{timeNs: number, durationNs?: number, bucket: import('../../span_stats').SpanBuckets}>} drained
    * @param {number} bucketSizeNs
    */
   transform (drained, bucketSizeNs) {
@@ -89,9 +89,9 @@ class OtlpStatsTransformer extends OtlpTransformerBase {
 
     const dataPoints = []
 
-    for (const { timeNs, bucket } of drained) {
+    for (const { timeNs, durationNs = bucketSizeNs, bucket } of drained) {
       const distributions = new Map()
-      const endTimeNs = timeNs + bucketSizeNs
+      const endTimeNs = timeNs + durationNs
       const startNano = isJson ? String(timeNs) : timeNs
       const endNano = isJson ? String(endTimeNs) : endTimeNs
 

packages/dd-trace/src/opentelemetry/metrics/periodic_metric_reader.js

@@ -41,15 +41,18 @@ function callDone (done, error) {
  *
  * @typedef {SumCumulativeState | HistogramCumulativeState} CumulativeStateValue
  *
+ * @typedef {{ value: number, timeUnixNano: number }} SumLastExportedState
+ *
  * @typedef {{
  *   count: number,
  *   sum: number,
  *   min?: number,
  *   max?: number,
- *   bucketCounts: number[]
+ *   bucketCounts: number[],
+ *   timeUnixNano: number
  * }} HistogramLastExportedState
  *
- * @typedef {number | HistogramLastExportedState} LastExportedStateValue
+ * @typedef {SumLastExportedState | HistogramLastExportedState} LastExportedStateValue
  */
 
 /**
@@ -506,7 +509,7 @@ class MetricAggregator {
       }
     }
 
-    this.#applyDeltaTemporality(metricsMap.values(), lastExportedState)
+    this.#applyDeltaTemporality(metricsMap.values(), lastExportedState, nowUnixNano())
     return metricsMap
   }
 
@@ -547,33 +550,40 @@ class MetricAggregator {
    *
    * @param {Iterable<AggregatedMetric>} metrics - The metrics to apply delta temporality to
    * @param {Map<string, LastExportedStateValue>} lastExportedState - The last exported state of the metrics
+   * @param {number} collectionTime - The collection timestamp in nanoseconds
    */
-  #applyDeltaTemporality (metrics, lastExportedState) {
+  #applyDeltaTemporality (metrics, lastExportedState, collectionTime) {
     for (const metric of metrics) {
       if (metric.temporality === TEMPORALITY.DELTA && this.#isDeltaType(metric.type)) {
         const scopeKey = this.#getScopeKey(metric.instrumentationScope)
 
         for (const dataPoint of metric.dataPointMap.values()) {
           const stateKey = this.#getStateKey(scopeKey, metric.name, metric.type, dataPoint.attrKey)
+          dataPoint.timeUnixNano = collectionTime
 
           if (metric.type === METRIC_TYPES.COUNTER || metric.type === METRIC_TYPES.OBSERVABLECOUNTER) {
-            const lastValue = lastExportedState.get(stateKey) || 0
+            const lastState = lastExportedState.get(stateKey)
             const currentValue = dataPoint.value
-            dataPoint.value = currentValue - lastValue
-            lastExportedState.set(stateKey, currentValue)
+            dataPoint.startTimeUnixNano = lastState?.timeUnixNano ??
+              dataPoint.startTimeUnixNano ?? dataPoint.timeUnixNano
+            dataPoint.value = currentValue - (lastState?.value ?? 0)
+            lastExportedState.set(stateKey, { value: currentValue, timeUnixNano: dataPoint.timeUnixNano })
           } else if (metric.type === METRIC_TYPES.HISTOGRAM) {
             const lastState = lastExportedState.get(stateKey) || {
               count: 0,
               sum: 0,
               bucketCounts: new Array(dataPoint.bucketCounts.length).fill(0),
+              timeUnixNano: dataPoint.startTimeUnixNano,
             }
             const currentState = {
               count: dataPoint.count,
               sum: dataPoint.sum,
               min: dataPoint.min,
               max: dataPoint.max,
               bucketCounts: [...dataPoint.bucketCounts],
+              timeUnixNano: dataPoint.timeUnixNano,
             }
+            dataPoint.startTimeUnixNano = lastState.timeUnixNano
             dataPoint.count = currentState.count - lastState.count
             dataPoint.sum = currentState.sum - lastState.sum
             dataPoint.bucketCounts = currentState.bucketCounts.map(

packages/dd-trace/src/span_stats.js

@@ -16,6 +16,7 @@ const {
 const { ORIGIN_KEY, TOP_LEVEL_KEY, SVC_SRC_KEY, GRPC_STATUS_NAMES } = require('./constants')
 const id = require('./id')
 const log = require('./log')
+const { nowUnixNano } = require('./opentelemetry/metrics/time')
 
 const GRPC_STATUS_CODE_MAP = Object.fromEntries(GRPC_STATUS_NAMES.map((name, i) => [name, String(i)]))
 const ZERO_ID = id('0')
@@ -191,6 +192,7 @@ class TimeBuckets extends Map {
 
 class SpanStatsProcessor {
   #config
+  #otlpStartTimeNs
 
   /**
    * @param {import('./config/config-base')} config
@@ -220,6 +222,7 @@ class SpanStatsProcessor {
     this.hostname = os.hostname()
     this.enabled = enabled
     this.otlpExporter = otlpExporter || null
+    this.#otlpStartTimeNs = otlpExporter ? nowUnixNano() : undefined
     this.env = env
     this.#config = config
     this.sequence = 0
@@ -241,10 +244,11 @@ class SpanStatsProcessor {
    */
   forceFlush (done) {
     this.#flush(done)
+    if (this.otlpExporter) this.timer.refresh()
   }
 
   #flush (done) {
-    const drained = this.#drainBuckets()
+    const drained = this.otlpExporter ? this.#drainOtlpBucket() : this.#drainBuckets()
 
     if (this.enabled && !this.otlpExporter) {
       this.exporter.export({
@@ -282,7 +286,7 @@ class SpanStatsProcessor {
         this.otlpExporter.export(drained, this.bucketSizeNs, done)
       }
     } else if (this.otlpExporter) {
-      if (typeof this.otlpExporter.flush === 'function') this.otlpExporter.flush(done)
+      if (done && typeof this.otlpExporter.flush === 'function') this.otlpExporter.flush(done)
       else done?.()
     } else done?.()
   }
@@ -291,8 +295,11 @@ class SpanStatsProcessor {
     if (!this.enabled && !this.otlpExporter) return
     if (!span.metrics[TOP_LEVEL_KEY] && !span.metrics[MEASURED]) return
 
-    const spanEndNs = span.start + span.duration
-    const bucketTime = spanEndNs - (spanEndNs % this.bucketSizeNs)
+    let bucketTime = this.#otlpStartTimeNs
+    if (!this.otlpExporter) {
+      const spanEndNs = span.start + span.duration
+      bucketTime = spanEndNs - (spanEndNs % this.bucketSizeNs)
+    }
 
     this.buckets.forTime(bucketTime)
       .forSpan(span)
@@ -308,6 +315,15 @@ class SpanStatsProcessor {
     return drained
   }
 
+  #drainOtlpBucket () {
+    const startTimeNs = this.#otlpStartTimeNs
+    const endTimeNs = Math.max(nowUnixNano(), startTimeNs + 1e3)
+    const bucket = this.buckets.get(startTimeNs)
+    this.buckets = new TimeBuckets(true)
+    this.#otlpStartTimeNs = endTimeNs
+    return bucket ? [{ timeNs: startTimeNs, durationNs: endTimeNs - startTimeNs, bucket }] : []
+  }
+
   #toV06Payload (drained) {
     const { bucketSizeNs } = this
     return drained.map(({ timeNs, bucket }) => ({

packages/dd-trace/test/opentelemetry/metrics.spec.js

@@ -522,6 +522,37 @@ describe('OpenTelemetry Meter Provider', () => {
       }, 120)
     })
 
+    it('advances DELTA start times between exports', () => {
+      const clock = sinon.useFakeTimers()
+      const exported = []
+      mockOtlpExport((decoded) => {
+        exported.push(decoded.resourceMetrics[0].scopeMetrics[0].metrics)
+      })
+
+      setupMetrics({ OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE: 'DELTA' })
+      const meter = metrics.getMeter('app')
+      const counter = meter.createCounter('requests')
+      const histogram = meter.createHistogram('latency')
+
+      counter.add(5)
+      histogram.record(10)
+      clock.tick(100)
+      counter.add(3)
+      histogram.record(20)
+      clock.tick(100)
+
+      assert.strictEqual(exported.length, 2)
+      for (const name of ['requests', 'latency']) {
+        const metric = exported.map(metrics => metrics.find(metric => metric.name === name))
+        const points = metric.map(metric => (metric.sum || metric.histogram).dataPoints[0])
+        assert(points[0].timeUnixNano > points[0].startTimeUnixNano)
+        assert.strictEqual(points[1].startTimeUnixNano, points[0].timeUnixNano)
+        assert(points[1].timeUnixNano > points[1].startTimeUnixNano)
+      }
+      assert.strictEqual(exported[1].find(metric => metric.name === 'requests').sum.dataPoints[0].asInt, 3)
+      assert.strictEqual(exported[1].find(metric => metric.name === 'latency').histogram.dataPoints[0].count, 1)
+    })
+
     it('LOWMEMORY uses DELTA for sync counters', (done) => {
       const validator = mockOtlpExport((decoded) => {
         const counter = decoded.resourceMetrics[0].scopeMetrics[0].metrics[0]

packages/dd-trace/test/opentelemetry/metrics/otlp_span_stats_transformer.spec.js

@@ -341,13 +341,15 @@ describe('OtlpStatsTransformer', () => {
       assert.strictEqual(serviceByResource['GET /bar'], 'svc-other')
     })
 
-    it('sets timestamps from the bucket time and size', () => {
+    it('sets timestamps from the collection window', () => {
       const timeNs = 12340000000000
-      const dp = dataPointsOf(JSON.parse(transformer.transform(makeDrained(timeNs, [makeSpan()]), BUCKET_SIZE_NS)))[0]
+      const drained = makeDrained(timeNs, [makeSpan()])
+      drained[0].durationNs = 123456
+      const dp = dataPointsOf(JSON.parse(transformer.transform(drained, BUCKET_SIZE_NS)))[0]
 
       assert.deepStrictEqual(
         { start: dp.startTimeUnixNano, end: dp.timeUnixNano },
-        { start: String(timeNs), end: String(timeNs + BUCKET_SIZE_NS) }
+        { start: String(timeNs), end: String(timeNs + drained[0].durationNs) }
       )
     })
 

packages/dd-trace/test/span_stats.spec.js

@@ -634,6 +634,35 @@ describe('SpanStatsProcessor', () => {
     assert.strictEqual(bucketSizeNs, p.bucketSizeNs)
   })
 
+  it('uses continuous OTLP windows and restarts the interval after force flush', () => {
+    const clock = sinon.useFakeTimers({ now: 12_345_000 })
+    try {
+      const localExporter = {
+        export: sinon.stub().callsFake((_drained, _bucketSizeNs, done) => done?.()),
+        flush: sinon.stub().callsFake(done => done?.()),
+      }
+      const p = new SpanStatsProcessor(config, localExporter)
+
+      p.onSpanFinished(topLevelSpan)
+      clock.tick(3_000)
+      p.forceFlush(() => {})
+      p.onSpanFinished(topLevelSpan)
+
+      clock.tick(7_000)
+      assert.ok(localExporter.export.calledOnce)
+
+      clock.tick(3_000)
+      assert.ok(localExporter.export.calledTwice)
+      const [first] = localExporter.export.firstCall.args[0]
+      const [second] = localExporter.export.secondCall.args[0]
+
+      assert.strictEqual(second.timeNs, first.timeNs + first.durationNs)
+      assert.strictEqual(second.durationNs, 10_000 * 1e6)
+    } finally {
+      clock.restore()
+    }
+  })
+
   it('should split OTLP trace roots when their attribute is exported', () => {
     const childSpan = { ...topLevelSpan, parent_id: { equals: () => false } }
     const processor = new SpanStatsProcessor(config, otlpExporter)