fix(opentelemetry): wait for trace delivery in forceFlush

packages/dd-trace/src/exporters/agent/index.js

@@ -1,6 +1,7 @@
 'use strict'
 
 const { URL } = require('url')
+const getFlushError = require('../../flush-error')
 const log = require('../../log')
 const { createServerlessDeliveryTracker } = require('../../serverless')
 const Writer = require('./writer')
@@ -59,26 +60,42 @@ class AgentExporter {
     }
   }
 
-  flush (done) {
+  /**
+   * @param {(error?: Error) => void} [done]
+   * @param {{ reportErrors?: boolean }} [options]
+   */
+  flush (done, options) {
     clearTimeout(this.#timer)
     this.#timer = undefined
 
     if (!this.#serverlessDeliveryTracker) {
       try {
-        return this._writer.flush(done)
+        return this._writer.flush(done, options)
       } catch (error) {
         log.error('Failed to flush traces: %s', error.message)
-        done?.()
+        done?.(options?.reportErrors ? error : undefined)
         return
       }
     }
 
+    let boundaryError
+    let waiting = false
+    const captureError = error => {
+      if (!waiting) boundaryError = error
+    }
     try {
-      this._writer.flush()
+      this._writer.flush(captureError, options)
     } catch (error) {
       log.error('Failed to flush traces: %s', error.message)
+      boundaryError = error
     }
-    this.#serverlessDeliveryTracker.waitForIdle(done)
+    waiting = true
+    if (!done) return
+
+    this.#serverlessDeliveryTracker.waitForIdle(error => {
+      if (!options?.reportErrors || !boundaryError) return done(error)
+      done(getFlushError(error ? [boundaryError, error] : [boundaryError]))
+    }, options)
   }
 }
 

packages/dd-trace/src/exporters/agent/writer.js

@@ -43,7 +43,7 @@ class AgentWriter extends BaseWriter {
    * Performs the writer flush without registering a serverless delivery.
    * Test Optimization owns its own request lifecycle tracking.
    * @param {(error?: Error) => void} [done]
-   * @param {{ deadline?: number }} [options]
+   * @param {{ deadline?: number, reportErrors?: boolean }} [options]
    * @returns {void}
    */
   flushDirect (done, options) {
@@ -73,7 +73,8 @@ class AgentWriter extends BaseWriter {
 
       if (err) {
         log.errorWithoutTelemetry('Error sending payload to the agent (status code: %s)', err.status, err)
-        done(flushOptions?.deadline === undefined ? undefined : err)
+        const reportError = flushOptions?.reportErrors || flushOptions?.deadline !== undefined
+        done(reportError ? err : undefined)
         return
       }
 

packages/dd-trace/src/exporters/common/writer.js

@@ -26,7 +26,7 @@ class Writer {
   /**
    * Flushes queued telemetry, retaining delivery on supported serverless platforms.
    * @param {(error?: Error) => void} [done]
-   * @param {{ deadline?: number }} [options]
+   * @param {{ deadline?: number, reportErrors?: boolean }} [options]
    * @returns {void}
    */
   flush (done, options) {
@@ -39,15 +39,21 @@ class Writer {
   /**
    * Flushes queued telemetry without registering serverless delivery retention.
    * @param {(error?: Error) => void} [done]
-   * @param {{ deadline?: number }} [options]
+   * @param {{ deadline?: number, reportErrors?: boolean }} [options]
    * @returns {void}
    */
   flushDirect (done = noop, options) {
     const count = this._encoder.count()
 
     if (!request.writable && options?.deadline === undefined && !this.#retainOnBackpressure) {
       this._encoder.reset()
-      done()
+      if (options?.reportErrors) {
+        const error = new log.NoTransmitError('Maximum active request buffer size reached: payload is discarded.')
+        error.code = 'ERR_DD_REQUEST_BUFFER_FULL'
+        done(error)
+      } else {
+        done()
+      }
     } else if (count > 0) {
       if (this.#isFirstFlush && firstFlushChannel.hasSubscribers && this._beforeFirstFlush) {
         this.#isFirstFlush = false
@@ -66,7 +72,7 @@ class Writer {
         // the oversized payload at the network boundary anyway.
         this._encoder.reset()
         log.error('Writer dropped %d trace(s) that exceeded the %d byte chunk cap', count, MAX_CHUNK_SIZE)
-        done()
+        done(options?.reportErrors ? error : undefined)
         return
       }
       if (options === undefined) {

packages/dd-trace/src/flush-error.js

@@ -0,0 +1,34 @@
+'use strict'
+
+/**
+ * @param {unknown} reason
+ * @param {unknown[]} reasons
+ */
+function collectReason (reason, reasons) {
+  if (reason instanceof AggregateError) {
+    for (const error of reason.errors) collectReason(error, reasons)
+    return
+  }
+
+  reasons.push(reason)
+}
+
+/**
+ * Preserves a single rejection reason and aggregates independent failures.
+ * @param {unknown[]} flushReasons
+ * @returns {unknown}
+ */
+function getFlushError (flushReasons) {
+  let flushError
+  if (flushReasons.length > 0) {
+    const reasons = []
+    for (const reason of flushReasons) collectReason(reason, reasons)
+
+    flushError = reasons.length === 1
+      ? reasons[0]
+      : new AggregateError(reasons, 'Multiple errors occurred while flushing')
+  }
+  return flushError
+}
+
+module.exports = getFlushError

packages/dd-trace/src/opentelemetry/otlp/otlp_http_exporter_base.js

@@ -5,7 +5,7 @@ const https = require('node:https')
 const { URL } = require('node:url')
 const { storage } = require('../../../../datadog-core')
 const log = require('../../log')
-const { createServerlessDeliveryTracker } = require('../../serverless')
+const TelemetryDeliveryTracker = require('../../serverless/telemetry-delivery-tracker')
 const telemetryMetrics = require('../../telemetry/metrics')
 
 const tracerMetrics = telemetryMetrics.manager.namespace('tracers')
@@ -20,8 +20,8 @@ const legacyStorage = storage('legacy')
  * @class OtlpHttpExporterBase
  */
 class OtlpHttpExporterBase {
+  #deliveryTracker = new TelemetryDeliveryTracker()
   #transport = https
-  #serverlessDeliveryTracker
 
   /**
    * Creates a new OtlpHttpExporterBase instance.
@@ -34,7 +34,6 @@ class OtlpHttpExporterBase {
    * @param {string} signalType - Signal type for error messages (e.g., 'logs', 'metrics')
    */
   constructor (url, headers, timeout, protocol, signalType) {
-    this.#serverlessDeliveryTracker = createServerlessDeliveryTracker()
     this.protocol = protocol
     this.signalType = signalType
 
@@ -83,10 +82,7 @@ class OtlpHttpExporterBase {
    * @protected
    */
   sendPayload (payload, resultCallback) {
-    if (this.#serverlessDeliveryTracker) {
-      return this.#serverlessDeliveryTracker.track(done => this.#sendPayload(payload, resultCallback, done))
-    }
-    this.#sendPayload(payload, resultCallback)
+    this.#deliveryTracker.track(done => this.#sendPayload(payload, resultCallback, done))
   }
 
   #sendPayload (payload, resultCallback, done) {
@@ -103,7 +99,7 @@ class OtlpHttpExporterBase {
       if (completed) return
       completed = true
       resultCallback(result)
-      done?.()
+      done?.(result.error)
     }
 
     try {
@@ -151,12 +147,12 @@ class OtlpHttpExporterBase {
   }
 
   /**
-   * Calls back once Vercel-tracked requests active at the flush boundary complete.
-   * @param {Function} [done]
+   * Calls back once requests active at the flush boundary complete.
+   * @param {(error?: Error) => void} [done]
+   * @param {{ reportErrors?: boolean }} [options]
    */
-  flush (done) {
-    if (this.#serverlessDeliveryTracker) return this.#serverlessDeliveryTracker.waitForIdle(done)
-    done?.()
+  flush (done, options) {
+    this.#deliveryTracker.waitForIdle(done, options)
   }
 
   /**

packages/dd-trace/src/opentelemetry/span_processor.js

@@ -1,5 +1,33 @@
 'use strict'
 
+const getFlushError = require('../flush-error')
+
+/**
+ * @typedef {{ status: 'fulfilled', value: unknown } | { status: 'rejected', reason: unknown }} FlushResult
+ */
+
+/**
+ * @param {FlushResult[]} results
+ * @returns {Promise<void> | undefined}
+ */
+function collectFlushErrors (results) {
+  const reasons = []
+  for (const result of results) {
+    if (result.status !== 'rejected') continue
+    reasons.push(result.reason)
+  }
+
+  if (reasons.length > 0) return Promise.reject(getFlushError(reasons))
+}
+
+/**
+ * @param {Promise<unknown>[]} flushes
+ * @returns {Promise<void>}
+ */
+function settleAllFlushes (flushes) {
+  return Promise.allSettled(flushes).then(collectFlushErrors)
+}
+
 class NoopSpanProcessor {
   forceFlush () {
     return Promise.resolve()
@@ -22,9 +50,9 @@ class MultiSpanProcessor extends NoopSpanProcessor {
   }
 
   forceFlush () {
-    return Promise.all(
-      this.#processors.map(p => p.forceFlush())
-    )
+    const flushes = []
+    for (const processor of this.#processors) flushes.push(processor.forceFlush())
+    return settleAllFlushes(flushes)
   }
 
   onStart (span, context) {
@@ -49,4 +77,5 @@ class MultiSpanProcessor extends NoopSpanProcessor {
 module.exports = {
   MultiSpanProcessor,
   NoopSpanProcessor,
+  settleAllFlushes,
 }

packages/dd-trace/src/opentelemetry/tracer_provider.js

@@ -6,12 +6,45 @@ const { W3CTraceContextPropagator } = require('../../../../vendor/dist/@opentele
 const tracer = require('../../')
 
 const ContextManager = require('./context_manager')
-const { MultiSpanProcessor, NoopSpanProcessor } = require('./span_processor')
+const { MultiSpanProcessor, NoopSpanProcessor, settleAllFlushes } = require('./span_processor')
 const Tracer = require('./tracer')
 
+/**
+ * @typedef {{
+ *   flush?: (done?: (error?: Error) => void, options?: { reportErrors?: boolean }) => void
+ * }} TraceExporter
+ */
+
+/**
+ * @param {TraceExporter} exporter
+ * @returns {Promise<void>}
+ */
+function flushExporter (exporter) {
+  if (typeof exporter.flush !== 'function') return Promise.resolve()
+
+  /**
+   * @param {() => void} resolve
+   * @param {(reason?: unknown) => void} reject
+   */
+  function flush (resolve, reject) {
+    /**
+     * @param {Error} [error]
+     */
+    function done (error) {
+      if (error) reject(error)
+      else resolve()
+    }
+
+    exporter.flush(done, { reportErrors: true })
+  }
+
+  return new Promise(flush)
+}
+
 class TracerProvider {
   #activeProcessor = new NoopSpanProcessor()
   #contextManager = new ContextManager()
+  #flush
   #processors = []
   #tracers = new Map()
 
@@ -84,8 +117,19 @@ class TracerProvider {
       return Promise.reject(new Error('Not started'))
     }
 
-    exporter._writer?.flush()
-    return this.#activeProcessor.forceFlush()
+    const flush = () => settleAllFlushes([
+      flushExporter(exporter),
+      this.#activeProcessor.forceFlush(),
+    ])
+    const pending = this.#flush ? this.#flush.then(flush, flush) : flush()
+    this.#flush = pending
+
+    const clear = () => {
+      if (this.#flush === pending) this.#flush = undefined
+    }
+    pending.then(clear, clear)
+
+    return pending
   }
 
   shutdown () {