feat(opentelemetry): add metrics lifecycle methods

docs/API.md

@@ -517,6 +517,31 @@ cpuGauge.addCallback((result) => {
 })
 ```
 
+Short-lived processes can wait for pending metrics to reach the configured OTLP endpoint before exiting:
+
+Set `OTEL_EXPORTER_OTLP_METRICS_TIMEOUT` to bound each export. The value is in milliseconds.
+
+```javascript
+const meterProvider = metrics.getMeterProvider()
+
+await meterProvider.forceFlush()
+```
+
+Call `shutdown()` when the application is finished recording metrics. It performs one final export and then stops the
+provider. Calls to `forceFlush()` after shutdown have no effect.
+
+```javascript
+await meterProvider.shutdown()
+```
+
+For TypeScript, cast the provider to the Datadog implementation type:
+
+```typescript
+import type { opentelemetry as DatadogOpenTelemetry } from 'dd-trace'
+
+const meterProvider = metrics.getMeterProvider() as DatadogOpenTelemetry.MeterProvider
+```
+
 #### Supported Configuration
 
 The Datadog SDK supports many of the configurations supported by the OpenTelemetry SDK. The following environment variables are supported:

docs/test.ts

@@ -559,6 +559,9 @@ const provider: opentelemetry.TracerProvider = new tracer.TracerProvider();
 provider.register();
 
 const otelTracer: opentelemetry.Tracer = provider.getTracer("name", "version")
+const otelMeterProvider = {} as opentelemetry.MeterProvider
+const otelForceFlush: () => Promise<void> = otelMeterProvider.forceFlush
+const otelShutdown: () => Promise<void> = otelMeterProvider.shutdown
 
 // OTel supports several time input formats
 otelTracer.startSpan("name", { startTime: new Date() })

index.d.ts

@@ -3261,6 +3261,11 @@ declare namespace tracer {
   }
 
   export namespace opentelemetry {
+    export interface MeterProvider extends otel.MeterProvider {
+      forceFlush(): Promise<void>;
+      shutdown(): Promise<void>;
+    }
+
     /**
      * A registry for creating named {@link Tracer}s.
      */

index.d.v5.ts

@@ -3447,6 +3447,11 @@ declare namespace tracer {
   }
 
   export namespace opentelemetry {
+    export interface MeterProvider extends otel.MeterProvider {
+      forceFlush(): Promise<void>;
+      shutdown(): Promise<void>;
+    }
+
     /**
      * A registry for creating named {@link Tracer}s.
      */

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

@@ -79,7 +79,7 @@ function initializeOpenTelemetryMetrics (config) {
   const meterProvider = new MeterProvider({ reader })
   metrics.setGlobalMeterProvider(meterProvider)
   // Include the final metric collection and export in lifecycle retention.
-  registerTelemetryFlusher(done => meterProvider.forceFlush(done))
+  registerTelemetryFlusher(done => reader.forceFlush(done))
 }
 
 /**

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

@@ -50,12 +50,18 @@ class MeterProvider {
     return meter
   }
 
-  /**
-   * @param {Function} [done] Called after the metric export completes
-   */
-  forceFlush (done) {
-    if (this.reader) this.reader.forceFlush(done)
-    else done?.()
+  forceFlush () {
+    return new Promise((resolve, reject) => {
+      if (this.reader) this.reader.forceFlush(error => error ? reject(error) : resolve())
+      else resolve()
+    })
+  }
+
+  shutdown () {
+    return new Promise((resolve, reject) => {
+      if (this.reader) this.reader.shutdown(error => error ? reject(error) : resolve())
+      else resolve()
+    })
   }
 }
 

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

@@ -8,6 +8,15 @@ const {
 const { ObservableInstrument } = require('./instruments')
 const { nowUnixNano } = require('./time')
 
+function invokeLifecycleCallback (callback, error) {
+  if (!callback) return
+  try {
+    callback(error)
+  } catch (callbackError) {
+    log.error('Error completing OTLP metrics lifecycle:', callbackError)
+  }
+}
+
 /**
  * @typedef {import('@opentelemetry/api').Attributes} Attributes
  * @typedef {import('@opentelemetry/core').InstrumentationScope} InstrumentationScope
@@ -105,6 +114,11 @@ class PeriodicMetricReader {
   #exportInterval
   #aggregator
   #batchCallbacks = []
+  #exportQueue = []
+  #isExporting = false
+  #periodicExportPending = false
+  #shutdownCallbacks = []
+  #shutdownComplete = false
 
   /**
    * Creates a new PeriodicMetricReader instance.
@@ -202,42 +216,31 @@ class PeriodicMetricReader {
   forceFlush (done) {
     if (this.#isShutdown) {
       log.warn('PeriodicMetricReader is shutdown. %d measurement(s) were dropped', this.#droppedCount)
-      done?.()
+      invokeLifecycleCallback(done)
       return
     }
-    let pending = 2
-    const complete = () => {
-      if (--pending === 0) done?.()
-    }
-
-    // Snapshot requests already active before starting this flush's export.
-    try {
-      if (typeof this.exporter.flush === 'function') this.exporter.flush(complete)
-      else complete()
-    } catch (error) {
-      log.error('Error flushing OTLP metrics:', error)
-      complete()
-    }
-    try {
-      this.#collectAndExport(complete)
-    } catch (error) {
-      log.error('Error exporting OTLP metrics:', error)
-      complete()
-    }
+    this.#enqueueExport(done)
   }
 
   /**
    * Shuts down the reader and stops periodic collection.
-   * @returns {void}
+   *
+   * @param {Function} [done] Called after the final export and exporter shutdown complete
    */
-  shutdown () {
+  shutdown (done) {
     if (this.#isShutdown) {
-      log.warn('PeriodicMetricReader is already shutdown')
+      if (this.#shutdownComplete) {
+        log.warn('PeriodicMetricReader is already shutdown')
+        invokeLifecycleCallback(done)
+      } else if (done) {
+        this.#shutdownCallbacks.push(done)
+      }
       return
     }
     this.#isShutdown = true
+    if (done) this.#shutdownCallbacks.push(done)
     this.#clearTimer()
-    this.forceFlush()
+    this.#enqueueExport(error => this.#shutdownExporter(error))
   }
 
   /**
@@ -248,7 +251,7 @@ class PeriodicMetricReader {
     if (this.#timer) return
 
     this.#timer = setInterval(() => {
-      this.#collectAndExport()
+      this.#schedulePeriodicExport()
     }, this.#exportInterval)
     this.#timer.unref?.()
   }
@@ -264,6 +267,69 @@ class PeriodicMetricReader {
     }
   }
 
+  #schedulePeriodicExport () {
+    if (this.#periodicExportPending) return
+
+    this.#periodicExportPending = true
+    this.#enqueueExport(error => {
+      this.#periodicExportPending = false
+      if (error) log.error('Error exporting OTLP metrics:', error)
+    })
+  }
+
+  #enqueueExport (callback) {
+    this.#exportQueue.push(callback)
+    this.#drainExportQueue()
+  }
+
+  #drainExportQueue () {
+    if (this.#isExporting || this.#exportQueue.length === 0) return
+
+    this.#isExporting = true
+    const callback = this.#exportQueue.shift()
+    let completed = false
+    const complete = error => {
+      if (completed) return
+      completed = true
+      this.#isExporting = false
+      try {
+        invokeLifecycleCallback(callback, error)
+      } finally {
+        queueMicrotask(() => this.#drainExportQueue())
+      }
+    }
+
+    try {
+      this.#collectAndExport(complete)
+    } catch (error) {
+      log.error('Error exporting OTLP metrics:', error)
+      complete(error)
+    }
+  }
+
+  #shutdownExporter (exportError) {
+    let completed = false
+    const complete = shutdownError => {
+      if (completed) return
+      completed = true
+      this.#shutdownComplete = true
+      const error = exportError || shutdownError
+      const callbacks = this.#shutdownCallbacks
+      this.#shutdownCallbacks = []
+      for (const callback of callbacks) {
+        invokeLifecycleCallback(callback, error)
+      }
+    }
+
+    try {
+      if (typeof this.exporter.shutdown === 'function') this.exporter.shutdown(complete)
+      else complete()
+    } catch (error) {
+      log.error('Error shutting down OTLP metrics exporter:', error)
+      complete(error)
+    }
+  }
+
   /**
    * Collects measurements and exports metrics.
    *
@@ -322,7 +388,13 @@ class PeriodicMetricReader {
       this.#lastExportedState
     )
 
-    this.exporter.export(metrics, callback)
+    this.exporter.export(metrics, result => {
+      if (result?.code === 1) {
+        callback?.(result.error || new Error('OTLP metrics export failed'))
+      } else {
+        callback?.()
+      }
+    })
   }
 }
 

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

@@ -136,9 +136,9 @@ class OtlpHttpExporterBase {
         })
 
         req.once('timeout', () => {
-          req.destroy()
           const error = new Error('Request timeout')
           complete({ code: 1, error })
+          req.destroy()
         })
 
         req.write(payload)
@@ -172,7 +172,9 @@ class OtlpHttpExporterBase {
     this.telemetryTags[0] = `protocol:${this.#transport === https ? 'https' : 'http'}`
   }
 
-  shutdown () {}
+  shutdown (done) {
+    done?.()
+  }
 }
 
 module.exports = OtlpHttpExporterBase