Add support for OpenTelemetry Logs API (#6465)

* first attempt at otel logs support

* lint files and centralize configs

* add tests

* clean up yarn.lock

* clean up logs

* add back log exporter logic

* clean up docs

* clean up docs and fix otlp protocol

* revert readme docs

* add new otel files

* working version

* fmy

* fmt

* Update packages/dd-trace/src/config_defaults.js

* fix tests

* allow any version of logs api, let opentelemetry api determine the version

* add otlp payload tests

* add telemetry metrics

* some other clean ups

* simplify tests

* use agent hostname to resolve otlp endpoints

* clean up initalization

* parse additional otlp headers

* clean up component args

* clean up docs

* clean up component args

* remove addLogProcessor, init provider with a processor

* support trace-log correlation

* clean up registering provider in tests, and rename exporter arg

* first round of clean ups from PR review

* clean ups part 2

* make things private and clean up tests

* clean up tests

* clean up yarn file

* remove unused configs

* fix context issues

* nother round of clean ups

* group payloads by instrumentation scope

* fix typing

* address review comments

* add better typing, and better support for sending schemaurl

* revert instrumentationScope change to span

* review comments

* clean up protobuf loader file

* lint

* move protos to same dir, this will set up metrics work

* clean up throws

* update protos

* disable log injection when otel logs support is enabled

* update configurations to pass telemetry system tests

* remove useless import

* add more tests

* provide fix for failing system test

* add test case for noop logger and fix mocking for remote config

* add test case for noop logger and fix mocking for remote config

* fix encoding for doubles, remove unused shutdown code, test getLogger params

* remove unused timer and shutdown logic, simplify loggerprovider register and improve tests

* fix comment

* clean up how otel endpoint configs are loaded

* address Ayans comments

* add integration tests

Author: mabdinur

.github/workflows/platform.yml

@@ -285,6 +285,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@08c6903cd8c0fde910a37f88322edcfb5dd907a8 # v5.0.0
+      - uses: ./.github/actions/testagent/start
       - uses: ./.github/actions/node
         with:
           version: ${{ matrix.version }}

LICENSE-3rdparty.csv

@@ -7,7 +7,9 @@ require,@datadog/pprof,Apache license 2.0,Copyright 2019 Google Inc.
 require,@datadog/sketches-js,Apache license 2.0,Copyright 2020 Datadog Inc.
 require,@datadog/wasm-js-rewriter,Apache license 2.0,Copyright 2018 Datadog Inc.
 require,@opentelemetry/api,Apache license 2.0,Copyright OpenTelemetry Authors
+require,@opentelemetry/api-logs,Apache license 2.0,Copyright OpenTelemetry Authors
 require,@opentelemetry/core,Apache license 2.0,Copyright OpenTelemetry Authors
+require,@opentelemetry/resources,Apache license 2.0,Copyright OpenTelemetry Authors
 require,@isaacs/ttlcache,ISC,Copyright (c) 2022-2023 - Isaac Z. Schlueter and Contributors
 require,crypto-randomuuid,MIT,Copyright 2021 Node.js Foundation and contributors
 require,dc-polyfill,MIT,Copyright 2023 Datadog Inc.

docker-compose.yml

@@ -180,6 +180,7 @@ services:
     image: ghcr.io/datadog/dd-apm-test-agent/ddapm-test-agent:v1.33.1
     ports:
       - "127.0.0.1:9126:9126"
+      - "127.0.0.1:4318:4318"
     environment:
       - LOG_LEVEL=DEBUG
       - TRACE_LANGUAGE=javascript

docs/API.md

@@ -379,6 +379,44 @@ The following attributes are available to override Datadog-specific options:
 * `resource.name`: The resource name to be used for this span. The operation name will be used if this is not provided.
 * `span.type`: The span type to be used for this span. Will fallback to `custom` if not provided.
 
+<h3 id="opentelemetry-logs">OpenTelemetry Logs</h3>
+
+dd-trace-js includes experimental support for OpenTelemetry logs, designed as a drop-in replacement for the OpenTelemetry SDK. This support is primarily intended for logging libraries rather than direct user configuration. Enable it by setting `DD_LOGS_OTEL_ENABLED=true` and use the [OpenTelemetry Logs API](https://open-telemetry.github.io/opentelemetry-js/modules/_opentelemetry_api-logs.html) to emit structured log data:
+
+```javascript
+require('dd-trace').init()
+const { logs } = require('@opentelemetry/api-logs')
+const express = require('express')
+
+const app = express()
+const logger = logs.getLogger('my-service', '1.0.0')
+
+app.get('/api/users/:id', (req, res) => {
+  logger.emit({
+    severityText: 'INFO',
+    severityNumber: 9,
+    body: `Processing user request for ID: ${req.params.id}`,
+  })
+  res.json({ id: req.params.id, name: 'John Doe' })
+})
+
+app.listen(3000)
+```
+
+#### Supported Configuration
+
+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_BSP_MAX_EXPORT_BATCH_SIZE` - Maximum logs per batch (default: `512`)
+
+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/).
+
 <h2 id="advanced-configuration">Advanced Configuration</h2>
 
 <h3 id="tracer-settings">Tracer settings</h3>

integration-tests/opentelemetry-logs.spec.js

@@ -0,0 +1,60 @@
+'use strict'
+
+const { assert } = require('chai')
+const { createSandbox } = require('./helpers')
+const http = require('http')
+
+describe('OpenTelemetry Logs Integration', () => {
+  let sandbox
+
+  beforeEach(async () => {
+    sandbox = await createSandbox()
+  })
+
+  afterEach(async () => {
+    await sandbox.remove()
+  })
+
+  it('should send OTLP logs to test agent and receive 200', (done) => {
+    const payload = JSON.stringify({
+      resourceLogs: [{
+        scopeLogs: [{ logRecords: [{ body: { stringValue: 'test' }, timeUnixNano: String(Date.now() * 1000000) }] }]
+      }]
+    })
+
+    const req = http.request({
+      hostname: '127.0.0.1',
+      port: 4318,
+      path: '/v1/logs',
+      method: 'POST',
+      headers: { 'Content-Type': 'application/json', 'Content-Length': payload.length }
+    }, (res) => {
+      assert.strictEqual(res.statusCode, 200)
+      done()
+    })
+
+    req.on('error', done)
+    req.write(payload)
+    req.end()
+  })
+
+  it('should receive 400 when sending protobuf to JSON endpoint', (done) => {
+    const protobufPayload = Buffer.from([0x0a, 0x04, 0x08, 0x01, 0x12, 0x00])
+
+    const req = http.request({
+      hostname: '127.0.0.1',
+      port: 4318,
+      path: '/v3/logs',
+      method: 'POST',
+      headers: { 'Content-Type': 'application/x-protobuf', 'Content-Length': protobufPayload.length }
+    }, (res) => {
+      // 404 Not Found - wrong path
+      assert.strictEqual(res.statusCode, 404)
+      done()
+    })
+
+    req.on('error', done)
+    req.write(protobufPayload)
+    req.end()
+  })
+})

package.json

@@ -126,7 +126,9 @@
     "@datadog/wasm-js-rewriter": "4.0.1",
     "@isaacs/ttlcache": "^1.4.1",
     "@opentelemetry/api": ">=1.0.0 <1.10.0",
+    "@opentelemetry/api-logs": "<1.0.0",
     "@opentelemetry/core": ">=1.14.0 <1.31.0",
+    "@opentelemetry/resources": ">=1.0.0 <1.10.0",
     "crypto-randomuuid": "^1.0.0",
     "dc-polyfill": "^0.1.10",
     "ignore": "^7.0.5",

packages/dd-trace/src/config.js

@@ -67,6 +67,8 @@ const VALID_PROPAGATION_BEHAVIOR_EXTRACT = new Set(['continue', 'restart', 'igno
 
 const VALID_LOG_LEVELS = new Set(['debug', 'info', 'warn', 'error'])
 
+const DEFAULT_OTLP_PORT = 4318
+
 function getFromOtelSamplerMap (otelTracesSampler, otelTracesSamplerArg) {
   const OTEL_TRACES_SAMPLER_MAPPING = {
     always_on: '1.0',
@@ -554,6 +556,7 @@ class Config {
       DD_INSTRUMENTATION_TELEMETRY_ENABLED,
       DD_INSTRUMENTATION_CONFIG_ID,
       DD_LOGS_INJECTION,
+      DD_LOGS_OTEL_ENABLED,
       DD_LANGCHAIN_SPAN_CHAR_LIMIT,
       DD_LANGCHAIN_SPAN_PROMPT_COMPLETION_SAMPLE_RATE,
       DD_LLMOBS_AGENTLESS_ENABLED,
@@ -635,7 +638,17 @@ class Config {
       OTEL_RESOURCE_ATTRIBUTES,
       OTEL_SERVICE_NAME,
       OTEL_TRACES_SAMPLER,
-      OTEL_TRACES_SAMPLER_ARG
+      OTEL_TRACES_SAMPLER_ARG,
+      OTEL_EXPORTER_OTLP_LOGS_ENDPOINT,
+      OTEL_EXPORTER_OTLP_LOGS_HEADERS,
+      OTEL_EXPORTER_OTLP_LOGS_PROTOCOL,
+      OTEL_EXPORTER_OTLP_LOGS_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
     } = getEnvironmentVariables()
 
     const tags = {}
@@ -649,6 +662,23 @@ class Config {
     tagger.add(tags, DD_TRACE_TAGS)
     tagger.add(tags, DD_TRACE_GLOBAL_TAGS)
 
+    this._setBoolean(env, 'otelLogsEnabled', isTrue(DD_LOGS_OTEL_ENABLED))
+    // Set OpenTelemetry logs configuration with specific _LOGS_ vars taking precedence over generic _EXPORTERS_ vars
+    if (OTEL_EXPORTER_OTLP_ENDPOINT) {
+      // Only set if there's a custom URL, otherwise let calc phase handle the default
+      this._setString(env, 'otelUrl', OTEL_EXPORTER_OTLP_ENDPOINT)
+    }
+    if (OTEL_EXPORTER_OTLP_ENDPOINT || OTEL_EXPORTER_OTLP_LOGS_ENDPOINT) {
+      this._setString(env, 'otelLogsUrl', OTEL_EXPORTER_OTLP_LOGS_ENDPOINT || env.otelUrl)
+    }
+    this._setString(env, 'otelHeaders', OTEL_EXPORTER_OTLP_HEADERS)
+    this._setString(env, 'otelLogsHeaders', OTEL_EXPORTER_OTLP_LOGS_HEADERS || env.otelHeaders)
+    this._setString(env, 'otelProtocol', OTEL_EXPORTER_OTLP_PROTOCOL)
+    this._setString(env, 'otelLogsProtocol', OTEL_EXPORTER_OTLP_LOGS_PROTOCOL || env.otelProtocol)
+    env.otelTimeout = maybeInt(OTEL_EXPORTER_OTLP_TIMEOUT)
+    env.otelLogsTimeout = maybeInt(OTEL_EXPORTER_OTLP_LOGS_TIMEOUT) || env.otelTimeout
+    env.otelLogsBatchTimeout = maybeInt(OTEL_BSP_SCHEDULE_DELAY)
+    env.otelLogsMaxExportBatchSize = maybeInt(OTEL_BSP_MAX_EXPORT_BATCH_SIZE)
     this._setBoolean(
       env,
       'apmTracingEnabled',
@@ -1156,7 +1186,20 @@ class Config {
       calc.testManagementAttemptToFixRetries = maybeInt(DD_TEST_MANAGEMENT_ATTEMPT_TO_FIX_RETRIES) ?? 20
       this._setBoolean(calc, 'isImpactedTestsEnabled', !isFalse(DD_CIVISIBILITY_IMPACTED_TESTS_DETECTION_ENABLED))
     }
+
+    // Disable log injection when OTEL logs are enabled
+    // OTEL logs and DD log injection are mutually exclusive
+    if (this._env.otelLogsEnabled) {
+      this._setBoolean(calc, 'logInjection', false)
+    }
+
     calc['dogstatsd.hostname'] = this._getHostname()
+
+    // Compute OTLP logs URL to send payloads to the active Datadog Agent
+    const agentHostname = this._getHostname()
+    calc.otelLogsUrl = `http://${agentHostname}:${DEFAULT_OTLP_PORT}`
+    calc.otelUrl = `http://${agentHostname}:${DEFAULT_OTLP_PORT}`
+
     this._setBoolean(calc, 'isGitUploadEnabled',
       calc.isIntelligentTestRunnerEnabled && !isFalse(this._isCiVisibilityGitUploadEnabled()))
     this._setBoolean(calc, 'spanComputePeerService', this._getSpanComputePeerService())

packages/dd-trace/src/config_defaults.js

@@ -126,6 +126,17 @@ module.exports = {
   isTestManagementEnabled: false,
   isImpactedTestsEnabled: false,
   logInjection: true,
+  otelLogsEnabled: false,
+  otelUrl: undefined,
+  otelLogsUrl: undefined, // Will be computed using agent host
+  otelHeaders: undefined,
+  otelLogsHeaders: '',
+  otelProtocol: 'http/protobuf',
+  otelLogsProtocol: 'http/protobuf',
+  otelLogsTimeout: 10_000,
+  otelTimeout: 10_000,
+  otelLogsBatchTimeout: 5000,
+  otelLogsMaxExportBatchSize: 512,
   lookup: undefined,
   inferredProxyServicesEnabled: false,
   memcachedCommandEnabled: false,

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

@@ -0,0 +1,100 @@
+'use strict'
+
+/**
+ * @typedef {import('@opentelemetry/api-logs').LogRecord} LogRecord
+ * @typedef {import('@opentelemetry/core').InstrumentationScope} InstrumentationScope
+ */
+
+/**
+ * BatchLogRecordProcessor processes log records in batches for efficient export to Datadog Agent.
+ *
+ * This implementation follows the OpenTelemetry JavaScript SDK BatchLogRecordProcessor:
+ * https://open-telemetry.github.io/opentelemetry-js/classes/_opentelemetry_sdk-logs.BatchLogRecordProcessor.html
+ *
+ * @class BatchLogRecordProcessor
+ */
+class BatchLogRecordProcessor {
+  #logRecords
+  #timer
+  #batchTimeout
+  #maxExportBatchSize
+
+  /**
+   * Creates a new BatchLogRecordProcessor instance.
+   *
+   * @param {OtlpHttpLogExporter} exporter - Log processor for exporting batches to Datadog Agent
+   * @param {number} batchTimeout - Timeout in milliseconds for batch processing
+   * @param {number} maxExportBatchSize - Maximum number of log records per batch
+   */
+  constructor (exporter, batchTimeout, maxExportBatchSize) {
+    this.exporter = exporter
+    this.#batchTimeout = batchTimeout
+    this.#maxExportBatchSize = maxExportBatchSize
+    this.#logRecords = []
+    this.#timer = null
+  }
+
+  /**
+   * Processes a single log record.
+   *
+   * @param {LogRecord} logRecord - The enriched log record with trace correlation and metadata
+   * @param {InstrumentationScope} instrumentationScope - The instrumentation library
+   */
+  onEmit (logRecord, instrumentationScope) {
+    // Store the log record (already enriched by Logger.emit)
+    this.#logRecords.push({ ...logRecord, instrumentationScope })
+
+    if (this.#logRecords.length >= this.#maxExportBatchSize) {
+      this.#export()
+    } else if (this.#logRecords.length === 1) {
+      this.#startTimer()
+    }
+  }
+
+  /**
+   * Forces an immediate flush of all pending log records.
+   * @returns {undefined} Promise that resolves when flush is complete
+   */
+  forceFlush () {
+    this.#export()
+  }
+
+  /**
+   * Starts the batch timeout timer.
+   * @private
+   */
+  #startTimer () {
+    if (this.#timer) {
+      return
+    }
+
+    this.#timer = setTimeout(() => {
+      this.#export()
+    }, this.#batchTimeout)
+  }
+
+  /**
+   * Exports the current batch of log records.
+   * @private
+   */
+  #export () {
+    const logRecords = this.#logRecords.slice(0, this.#maxExportBatchSize)
+    this.#logRecords = this.#logRecords.slice(this.#maxExportBatchSize)
+
+    this.#clearTimer()
+    this.exporter.export(logRecords, () => {})
+  }
+
+  /**
+   * Clears the batch timeout timer.
+   * @private
+   */
+  #clearTimer () {
+    if (this.#timer) {
+      clearTimeout(this.#timer)
+      this.#timer = null
+    }
+  }
+}
+
+module.exports = BatchLogRecordProcessor