feat(otel): add support for otel metrics api via protobuf and json (#6783)

* add otlp metrics protos

* feat: add OTLP metrics proto definitions and reorganize directory

- Add metrics.proto and metrics_service.proto (OTLP v1 spec)
- Update protobuf_loader to support metrics protos
- Rename protos/ -> otlp/ directory for better organization

* refactor: extract common OTLP logic into base classes

- Create OtlpHttpExporterBase for shared HTTP export logic
- Create OtlpTransformerBase for shared transformation logic
- Refactor logs exporter/transformer to extend base classes
- Update test mocking paths
- Eliminates ~400 lines of duplication

* fix logs

* increase test coverage

* feat(metrics): add support for otel metrics provider

* feat(metrics): add support for otlp configurations

* updates to pass system tests

* add temperolality support and clean up implementation

* add support for encoding attributes

* simplify tests

* use real values in tests

* do not encode numbers as strings

* use enum in transformation

* remove unneed fields

* add better temporality support and include encoding for async metric types

* improve test coverage for scope attributes

* simplify stubs in tests

* validate scope attributes

* ruben comments

* fix broken number test

* clean up tests

* clean up instruments and update tests

* simplify meter

* use constants and implement missing callbacks

* move otlp_transformer changes over

* clean up private/public fields

* avoid redefining constant

* clean up tests

* update typing

* linting clean ups

* first round of changes from cr

* limit number of metrics in each batch

* round 3 changes

* address more comments part 4

* update doc strings and typing

* update max batch size to operate on the aggregate metrics

* avoid converting aggregated metrics to arrays, perf

* fix regression in logs implementation

* revert closure

* cleanup max measurement queue, jdocs and typing

* log warning if value is invalid

* final set of changes

* update tests to be compatible with master branch

* remove register from meter provider and simplify export

* clean up stable stringify and update limit of max queue size

* remove @Private from metrics docs

* only implement the meterprovider api, renove shutdown and forceflush operations

* remove observableInstruments

* lint

* fix linting failures

* fix linting failures

* fix linting 3

Author: mabdinur

docs/API.md

@@ -408,14 +408,74 @@ app.listen(3000)
 The Datadog SDK supports many of the configurations supported by the OpenTelemetry SDK. The following environment variables are supported:
 
 - `DD_LOGS_OTEL_ENABLED` - Enable OpenTelemetry logs (default: `false`)
-- `OTEL_EXPORTER_OTLP_LOGS_ENDPOINT` - OTLP endpoint URL for logs (default: `http://localhost:4318`)
-- `OTEL_EXPORTER_OTLP_LOGS_HEADERS` - Optional headers in JSON format for logs (default: `{}`)
-- `OTEL_EXPORTER_OTLP_LOGS_PROTOCOL` - OTLP protocol for logs (default: `http/protobuf`)
-- `OTEL_EXPORTER_OTLP_LOGS_TIMEOUT` - Request timeout in milliseconds for logs (default: `10000`)
-- `OTEL_BSP_SCHEDULE_DELAY` - Batch timeout in milliseconds (default: `5000`)
+- `OTEL_EXPORTER_OTLP_LOGS_ENDPOINT` - OTLP endpoint URL for logs. Falls back to `OTEL_EXPORTER_OTLP_ENDPOINT` with `/v1/logs` appended (default: `http://localhost:4318/v1/logs`)
+- `OTEL_EXPORTER_OTLP_LOGS_HEADERS` - Optional headers for logs in JSON format. Falls back to `OTEL_EXPORTER_OTLP_HEADERS` (default: `{}`)
+- `OTEL_EXPORTER_OTLP_LOGS_PROTOCOL` - OTLP protocol for logs. Options: `http/protobuf`, `http/json`. Falls back to `OTEL_EXPORTER_OTLP_PROTOCOL` (default: `http/protobuf`)
+- `OTEL_EXPORTER_OTLP_LOGS_TIMEOUT` - Request timeout in milliseconds for logs. Falls back to `OTEL_EXPORTER_OTLP_TIMEOUT` (default: `10000`)
+- `OTEL_BSP_SCHEDULE_DELAY` - Batch export delay in milliseconds (default: `5000`)
 - `OTEL_BSP_MAX_EXPORT_BATCH_SIZE` - Maximum logs per batch (default: `512`)
+- `OTEL_BSP_MAX_QUEUE_SIZE` - Maximum logs to queue before dropping (default: `2048`)
 
-Logs are exported via OTLP over HTTP. The protocol can be configured using `OTEL_EXPORTER_OTLP_LOGS_PROTOCOL` or `OTEL_EXPORTER_OTLP_PROTOCOL` environment variables. Supported protocols are `http/protobuf` (default) and `http/json`. For complete OTLP exporter configuration options, see the [OpenTelemetry OTLP Exporter documentation](https://opentelemetry.io/docs/languages/sdk-configuration/otlp-exporter/).
+For complete OTLP exporter configuration options, see the [OpenTelemetry OTLP Exporter documentation](https://opentelemetry.io/docs/languages/sdk-configuration/otlp-exporter/).
+
+<h3 id="opentelemetry-metrics">OpenTelemetry Metrics</h3>
+
+dd-trace-js includes experimental support for OpenTelemetry metrics, designed as a drop-in replacement for the OpenTelemetry Metrics SDK. This lightweight implementation is fully compliant with the OpenTelemetry Metrics API and integrates with the existing OTLP export infrastructure. Enable it by setting `DD_METRICS_OTEL_ENABLED=true` and use the [OpenTelemetry Metrics API](https://open-telemetry.github.io/opentelemetry-js/modules/_opentelemetry_api.html) to record metric data:
+
+```javascript
+require('dd-trace').init()
+const { metrics } = require('@opentelemetry/api')
+
+const meter = metrics.getMeter('my-service', '1.0.0')
+
+// Counter - monotonically increasing values
+const requestCounter = meter.createCounter('http.requests', {
+  description: 'Total HTTP requests',
+  unit: 'requests'
+})
+requestCounter.add(1, { method: 'GET', status: 200 })
+
+// Histogram - distribution of values
+const durationHistogram = meter.createHistogram('http.duration', {
+  description: 'HTTP request duration',
+  unit: 'ms'
+})
+durationHistogram.record(145, { route: '/api/users' })
+
+// UpDownCounter - can increase and decrease
+const connectionCounter = meter.createUpDownCounter('active.connections', {
+  description: 'Active connections',
+  unit: 'connections'
+})
+connectionCounter.add(1)  // New connection
+connectionCounter.add(-1) // Connection closed
+
+// ObservableGauge - asynchronous observations
+const cpuGauge = meter.createObservableGauge('system.cpu.usage', {
+  description: 'CPU usage percentage',
+  unit: 'percent'
+})
+cpuGauge.addCallback((result) => {
+  const cpuUsage = process.cpuUsage()
+  result.observe(cpuUsage.system / 1000000, { core: '0' })
+})
+```
+
+#### Supported Configuration
+
+The Datadog SDK supports many of the configurations supported by the OpenTelemetry SDK. The following environment variables are supported:
+
+- `DD_METRICS_OTEL_ENABLED` - Enable OpenTelemetry metrics (default: `false`)
+- `OTEL_EXPORTER_OTLP_METRICS_ENDPOINT` - OTLP endpoint URL for metrics. Falls back to `OTEL_EXPORTER_OTLP_ENDPOINT` with `/v1/metrics` appended (default: `http://localhost:4318/v1/metrics`)
+- `OTEL_EXPORTER_OTLP_METRICS_HEADERS` - Optional headers for metrics in JSON format. Falls back to `OTEL_EXPORTER_OTLP_HEADERS` (default: `{}`)
+- `OTEL_EXPORTER_OTLP_METRICS_PROTOCOL` - OTLP protocol for metrics. Options: `http/protobuf`, `http/json`. Falls back to `OTEL_EXPORTER_OTLP_PROTOCOL` (default: `http/protobuf`)
+- `OTEL_EXPORTER_OTLP_METRICS_TIMEOUT` - Request timeout in milliseconds for metrics. Falls back to `OTEL_EXPORTER_OTLP_TIMEOUT` (default: `10000`)
+- `OTEL_METRIC_EXPORT_INTERVAL` - Metric export interval in milliseconds (default: `10000`)
+- `OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE` - Aggregation temporality preference. Options: `CUMULATIVE`, `DELTA`, `LOWMEMORY` (default: `DELTA`). See [OpenTelemetry spec](https://opentelemetry.io/docs/specs/otel/metrics/sdk_exporters/otlp/#additional-environment-variable-configuration) for details
+- `OTEL_BSP_MAX_QUEUE_SIZE` - Maximum metrics to queue before dropping (default: `2048`)
+- `OTEL_METRIC_EXPORT_TIMEOUT` - [NOT YET SUPPORTED] Time to export metrics including retries
+
+For complete OTLP exporter configuration options, see the [OpenTelemetry OTLP Exporter documentation](https://opentelemetry.io/docs/languages/sdk-configuration/otlp-exporter/).
 
 <h2 id="advanced-configuration">Advanced Configuration</h2>
 

packages/dd-trace/src/config.js

@@ -478,6 +478,7 @@ class Config {
       DD_INSTRUMENTATION_CONFIG_ID,
       DD_LOGS_INJECTION,
       DD_LOGS_OTEL_ENABLED,
+      DD_METRICS_OTEL_ENABLED,
       DD_LANGCHAIN_SPAN_CHAR_LIMIT,
       DD_LANGCHAIN_SPAN_PROMPT_COMPLETION_SAMPLE_RATE,
       DD_LLMOBS_AGENTLESS_ENABLED,
@@ -572,12 +573,20 @@ class Config {
       OTEL_EXPORTER_OTLP_LOGS_HEADERS,
       OTEL_EXPORTER_OTLP_LOGS_PROTOCOL,
       OTEL_EXPORTER_OTLP_LOGS_TIMEOUT,
+      OTEL_EXPORTER_OTLP_METRICS_ENDPOINT,
+      OTEL_EXPORTER_OTLP_METRICS_HEADERS,
+      OTEL_EXPORTER_OTLP_METRICS_PROTOCOL,
+      OTEL_EXPORTER_OTLP_METRICS_TIMEOUT,
+      OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE,
+      OTEL_METRIC_EXPORT_TIMEOUT,
       OTEL_EXPORTER_OTLP_PROTOCOL,
       OTEL_EXPORTER_OTLP_ENDPOINT,
       OTEL_EXPORTER_OTLP_HEADERS,
       OTEL_EXPORTER_OTLP_TIMEOUT,
       OTEL_BSP_SCHEDULE_DELAY,
-      OTEL_BSP_MAX_EXPORT_BATCH_SIZE
+      OTEL_BSP_MAX_EXPORT_BATCH_SIZE,
+      OTEL_BSP_MAX_QUEUE_SIZE,
+      OTEL_METRIC_EXPORT_INTERVAL
     } = source
 
     const tags = {}
@@ -604,10 +613,40 @@ class Config {
     this.#setString(target, 'otelLogsHeaders', OTEL_EXPORTER_OTLP_LOGS_HEADERS || target.otelHeaders)
     this.#setString(target, 'otelProtocol', OTEL_EXPORTER_OTLP_PROTOCOL)
     this.#setString(target, 'otelLogsProtocol', OTEL_EXPORTER_OTLP_LOGS_PROTOCOL || target.otelProtocol)
-    target.otelTimeout = maybeInt(OTEL_EXPORTER_OTLP_TIMEOUT)
-    target.otelLogsTimeout = maybeInt(OTEL_EXPORTER_OTLP_LOGS_TIMEOUT) || target.otelTimeout
-    target.otelLogsBatchTimeout = maybeInt(OTEL_BSP_SCHEDULE_DELAY)
-    target.otelLogsMaxExportBatchSize = maybeInt(OTEL_BSP_MAX_EXPORT_BATCH_SIZE)
+    const otelTimeout = nonNegInt(OTEL_EXPORTER_OTLP_TIMEOUT, 'OTEL_EXPORTER_OTLP_TIMEOUT')
+    if (otelTimeout !== undefined) {
+      target.otelTimeout = otelTimeout
+    }
+    const otelLogsTimeout = nonNegInt(OTEL_EXPORTER_OTLP_LOGS_TIMEOUT, 'OTEL_EXPORTER_OTLP_LOGS_TIMEOUT')
+    target.otelLogsTimeout = otelLogsTimeout === undefined ? target.otelTimeout : otelLogsTimeout
+    const otelBatchTimeout = nonNegInt(OTEL_BSP_SCHEDULE_DELAY, 'OTEL_BSP_SCHEDULE_DELAY', false)
+    if (otelBatchTimeout !== undefined) {
+      target.otelBatchTimeout = otelBatchTimeout
+    }
+    target.otelMaxExportBatchSize = nonNegInt(OTEL_BSP_MAX_EXPORT_BATCH_SIZE, 'OTEL_BSP_MAX_EXPORT_BATCH_SIZE', false)
+    target.otelMaxQueueSize = nonNegInt(OTEL_BSP_MAX_QUEUE_SIZE, 'OTEL_BSP_MAX_QUEUE_SIZE', false)
+
+    const otelMetricsExporter = !OTEL_METRICS_EXPORTER || OTEL_METRICS_EXPORTER.toLowerCase() !== 'none'
+    this.#setBoolean(target, 'otelMetricsEnabled', DD_METRICS_OTEL_ENABLED && otelMetricsExporter)
+    // Set OpenTelemetry metrics configuration with specific _METRICS_ vars
+    // taking precedence over generic _EXPORTERS_ vars
+    if (OTEL_EXPORTER_OTLP_ENDPOINT || OTEL_EXPORTER_OTLP_METRICS_ENDPOINT) {
+      this.#setString(target, 'otelMetricsUrl', OTEL_EXPORTER_OTLP_METRICS_ENDPOINT || target.otelUrl)
+    }
+    this.#setString(target, 'otelMetricsHeaders', OTEL_EXPORTER_OTLP_METRICS_HEADERS || target.otelHeaders)
+    this.#setString(target, 'otelMetricsProtocol', OTEL_EXPORTER_OTLP_METRICS_PROTOCOL || target.otelProtocol)
+    const otelMetricsTimeout = nonNegInt(OTEL_EXPORTER_OTLP_METRICS_TIMEOUT, 'OTEL_EXPORTER_OTLP_METRICS_TIMEOUT')
+    target.otelMetricsTimeout = otelMetricsTimeout === undefined ? target.otelTimeout : otelMetricsTimeout
+    target.otelMetricsExportTimeout = nonNegInt(OTEL_METRIC_EXPORT_TIMEOUT, 'OTEL_METRIC_EXPORT_TIMEOUT')
+    target.otelMetricsExportInterval = nonNegInt(OTEL_METRIC_EXPORT_INTERVAL, 'OTEL_METRIC_EXPORT_INTERVAL', false)
+
+    // Parse temporality preference (default to DELTA for Datadog)
+    if (OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE) {
+      const temporalityPref = OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE.toUpperCase()
+      if (['DELTA', 'CUMULATIVE', 'LOWMEMORY'].includes(temporalityPref)) {
+        this.#setString(target, 'otelMetricsTemporalityPreference', temporalityPref)
+      }
+    }
     this.#setBoolean(
       target,
       'apmTracingEnabled',
@@ -1206,9 +1245,10 @@ class Config {
 
     calc['dogstatsd.hostname'] = this.#getHostname()
 
-    // Compute OTLP logs URL to send payloads to the active Datadog Agent
+    // Compute OTLP logs and metrics URLs to send payloads to the active Datadog Agent
     const agentHostname = this.#getHostname()
     calc.otelLogsUrl = `http://${agentHostname}:${DEFAULT_OTLP_PORT}`
+    calc.otelMetricsUrl = `http://${agentHostname}:${DEFAULT_OTLP_PORT}/v1/metrics`
     calc.otelUrl = `http://${agentHostname}:${DEFAULT_OTLP_PORT}`
 
     this.#setBoolean(calc, 'isGitUploadEnabled',
@@ -1497,6 +1537,16 @@ function maybeFloat (number) {
   return Number.isNaN(parsed) ? undefined : parsed
 }
 
+function nonNegInt (value, envVarName, allowZero = true) {
+  if (value === undefined) return
+  const parsed = Number.parseInt(value)
+  if (Number.isNaN(parsed) || parsed < 0 || (parsed === 0 && !allowZero)) {
+    log.warn(`Invalid value ${parsed} for ${envVarName}. Using default value.`)
+    return
+  }
+  return parsed
+}
+
 function getAgentUrl (url, options) {
   if (url) return new URL(url)
 

packages/dd-trace/src/config_defaults.js

@@ -140,8 +140,17 @@ module.exports = {
   otelLogsProtocol: 'http/protobuf',
   otelLogsTimeout: 10_000,
   otelTimeout: 10_000,
-  otelLogsBatchTimeout: 5000,
-  otelLogsMaxExportBatchSize: 512,
+  otelBatchTimeout: 5000,
+  otelMaxExportBatchSize: 512,
+  otelMaxQueueSize: 2048,
+  otelMetricsEnabled: false,
+  otelMetricsUrl: undefined, // Will be computed using agent host
+  otelMetricsHeaders: '',
+  otelMetricsProtocol: 'http/protobuf',
+  otelMetricsTimeout: 10_000,
+  otelMetricsExportTimeout: 7500,
+  otelMetricsExportInterval: 10_000,
+  otelMetricsTemporalityPreference: 'DELTA', // DELTA, CUMULATIVE, or LOWMEMORY
   lookup: undefined,
   inferredProxyServicesEnabled: false,
   memcachedCommandEnabled: false,

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

@@ -70,8 +70,8 @@ function initializeOpenTelemetryLogs (config) {
   // Create batch processor for exporting logs to Datadog Agent
   const processor = new BatchLogRecordProcessor(
     exporter,
-    config.otelLogsBatchTimeout,
-    config.otelLogsMaxExportBatchSize
+    config.otelBatchTimeout,
+    config.otelMaxExportBatchSize
   )
 
   // Create logger provider with processor for Datadog Agent export

packages/dd-trace/src/opentelemetry/logs/logger.js

@@ -2,7 +2,8 @@
 
 const { sanitizeAttributes } = require('@opentelemetry/core')
 const { context } = require('@opentelemetry/api')
-const packageVersion = require('../../../../../package.json').version
+const { VERSION: packageVersion } = require('../../../../../version')
+
 /**
  * @typedef {import('@opentelemetry/api-logs').LogRecord} LogRecord
  * @typedef {import('@opentelemetry/api-logs').LoggerProvider} LoggerProvider

packages/dd-trace/src/opentelemetry/logs/otlp_http_log_exporter.js

@@ -11,7 +11,7 @@ const OtlpTransformer = require('./otlp_transformer')
 /**
  * OtlpHttpLogExporter exports log records via OTLP over HTTP.
  *
- * This implementation follows the OTLP HTTP specification:
+ * This implementation follows the OTLP HTTP v1.7.0 specification:
  * https://opentelemetry.io/docs/specs/otlp/#otlphttp
  *
  * @class OtlpHttpLogExporter
@@ -37,6 +37,8 @@ class OtlpHttpLogExporter extends OtlpHttpExporterBase {
    *
    * @param {LogRecord[]} logRecords - Array of enriched log records to export
    * @param {Function} resultCallback - Callback function for export result
+   *
+   * @returns {void}
    */
   export (logRecords, resultCallback) {
     if (logRecords.length === 0) {
@@ -45,8 +47,8 @@ class OtlpHttpLogExporter extends OtlpHttpExporterBase {
     }
 
     const payload = this.transformer.transformLogRecords(logRecords)
-    this._sendPayload(payload, resultCallback)
-    this._recordTelemetry('otel.log_records', logRecords.length)
+    this.sendPayload(payload, resultCallback)
+    this.recordTelemetry('otel.log_records', logRecords.length)
   }
 }
 

packages/dd-trace/src/opentelemetry/logs/otlp_transformer.js

@@ -40,7 +40,7 @@ const SEVERITY_MAP = {
 /**
  * OtlpTransformer transforms log records to OTLP format.
  *
- * This implementation follows the OTLP Logs Data Model specification:
+ * This implementation follows the OTLP Logs v1.7.0 Data Model specification:
  * https://opentelemetry.io/docs/specs/otlp/#log-data-model
  *
  * @class OtlpTransformer
@@ -79,12 +79,12 @@ class OtlpTransformer extends OtlpTransformerBase {
 
     const logsData = {
       resourceLogs: [{
-        resource: this._transformResource(),
+        resource: this.transformResource(),
         scopeLogs: this.#transformScope(logRecords),
       }]
     }
 
-    return this._serializeToProtobuf(protoLogsService, logsData)
+    return this.serializeToProtobuf(protoLogsService, logsData)
   }
 
   /**
@@ -95,11 +95,11 @@ class OtlpTransformer extends OtlpTransformerBase {
   #transformToJson (logRecords) {
     const logsData = {
       resourceLogs: [{
-        resource: this._transformResource(),
+        resource: this.transformResource(),
         scopeLogs: this.#transformScope(logRecords)
       }]
     }
-    return this._serializeToJson(logsData)
+    return this.serializeToJson(logsData)
   }
 
   /**
@@ -108,7 +108,7 @@ class OtlpTransformer extends OtlpTransformerBase {
    * @returns {Object[]} Array of scope log objects
    */
   #transformScope (logRecords) {
-    const groupedRecords = this._groupByInstrumentationScope(logRecords)
+    const groupedRecords = this.groupByInstrumentationScope(logRecords)
     const scopeLogs = []
 
     for (const records of groupedRecords.values()) {
@@ -155,7 +155,7 @@ class OtlpTransformer extends OtlpTransformerBase {
     }
 
     if (logRecord.attributes) {
-      result.attributes = this._transformAttributes(logRecord.attributes)
+      result.attributes = this.transformAttributes(logRecord.attributes)
     }
 
     if (spanContext?.traceFlags !== undefined) {
@@ -232,7 +232,7 @@ class OtlpTransformer extends OtlpTransformerBase {
         kvlistValue: {
           values: Object.entries(body).map(([key, value]) => ({
             key,
-            value: this._transformAnyValue(value)
+            value: this.transformAnyValue(value)
           }))
         }
       }

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

@@ -0,0 +1,34 @@
+'use strict'
+
+// Metric type constants
+const METRIC_TYPES = {
+  HISTOGRAM: 'histogram',
+  COUNTER: 'counter',
+  UPDOWNCOUNTER: 'updowncounter',
+  OBSERVABLECOUNTER: 'observable-counter',
+  OBSERVABLEUPDOWNCOUNTER: 'observable-updowncounter',
+  GAUGE: 'gauge'
+}
+
+// Temporality constants
+const TEMPORALITY = {
+  DELTA: 'DELTA',
+  CUMULATIVE: 'CUMULATIVE',
+  GAUGE: 'GAUGE',
+  LOWMEMORY: 'LOWMEMORY'
+}
+
+// Default histogram bucket boundaries (in milliseconds for latency metrics)
+const DEFAULT_HISTOGRAM_BUCKETS = [0, 5, 10, 25, 50, 75, 100, 250, 500, 750, 1000, 2500, 5000, 7500, 10_000]
+
+// Maximum number of measurements to queue before dropping
+// This limit corresponds to the maximum number of items
+// that be reliably added to a single Array.
+const DEFAULT_MAX_MEASUREMENT_QUEUE_SIZE = 65_536
+
+module.exports = {
+  METRIC_TYPES,
+  TEMPORALITY,
+  DEFAULT_HISTOGRAM_BUCKETS,
+  DEFAULT_MAX_MEASUREMENT_QUEUE_SIZE
+}