feat(pipeline): add OTLP trace export config to the wasm binding

libdatadog's TraceExporter can export traces over OTLP HTTP (JSON or
protobuf) instead of to the Datadog agent, mapping its internal traces to
OTLP directly. Expose that on WasmSpanState so dd-trace-js can honour
OTEL_TRACES_EXPORTER=otlp without resurrecting a JS-side OTLP exporter.

- setOtlpEndpoint(url): route export to an OTLP HTTP endpoint (e.g. an OTel
  Collector) instead of the agent.
- setOtlpProtocol('http/json'|'http/protobuf'): select the wire format;
  rejects unsupported values (grpc) at the parse boundary.
- setOtlpHeaders([k, v, ...]): extra headers (e.g. collector auth).

All three apply at lazy build time and only when an OTLP endpoint is set.

Author: bengl

crates/pipeline/src/lib.rs

@@ -3,6 +3,7 @@ use libdd_data_pipeline::trace_exporter::agent_response::AgentResponse;
 use libdd_data_pipeline::trace_exporter::{
     TraceExporter, TraceExporterBuilder, TraceExporterOutputFormat,
 };
+use libdd_data_pipeline::OtlpProtocol;
 use libdd_shared_runtime::LocalRuntime;
 use std::cell::{Cell, RefCell, UnsafeCell};
 use std::ffi::CStr;
@@ -180,6 +181,18 @@ pub struct WasmSpanState {
     /// fixed once the exporter is built on the first send, so `setUseV05` only
     /// takes effect if called before then.
     use_v05: Cell<bool>,
+    /// When set, the lazily-built exporter is configured to export traces via
+    /// OTLP HTTP to this endpoint (e.g. an OTel Collector) INSTEAD of the
+    /// Datadog agent. libdatadog maps its internal traces to OTLP, so no
+    /// JS-formatted spans are involved. Like `use_v05`, only takes effect if
+    /// set before the first send (when the exporter is built).
+    otlp_endpoint: RefCell<Option<String>>,
+    /// OTLP wire protocol (`http/json` default, or `http/protobuf`). Only
+    /// applied when `otlp_endpoint` is set.
+    otlp_protocol: Cell<Option<OtlpProtocol>>,
+    /// Extra HTTP headers for OTLP export (e.g. collector auth), as key/value
+    /// pairs. Only applied when `otlp_endpoint` is set.
+    otlp_headers: RefCell<Vec<(String, String)>>,
 }
 
 /// Clears an in-flight flag on drop, so an early return or a dropped future
@@ -269,6 +282,9 @@ impl WasmSpanState {
             prepared_spans: RefCell::new(None),
             sending: Cell::new(false),
             use_v05: Cell::new(false),
+            otlp_endpoint: RefCell::new(None),
+            otlp_protocol: Cell::new(None),
+            otlp_headers: RefCell::new(Vec::new()),
         })
     }
 
@@ -284,6 +300,36 @@ impl WasmSpanState {
         self.use_v05.set(v);
     }
 
+    /// Route trace export through libdatadog's OTLP HTTP exporter to `url`
+    /// instead of the Datadog agent. Must be called before the first send.
+    #[wasm_bindgen(js_name = "setOtlpEndpoint")]
+    pub fn set_otlp_endpoint(&self, url: String) {
+        *self.otlp_endpoint.borrow_mut() = Some(url);
+    }
+
+    /// Select the OTLP wire protocol: `http/json` (default) or `http/protobuf`.
+    /// Rejects unsupported values (e.g. `grpc`). Only takes effect with an OTLP
+    /// endpoint set, before the first send.
+    #[wasm_bindgen(js_name = "setOtlpProtocol")]
+    pub fn set_otlp_protocol(&self, protocol: String) -> Result<(), JsValue> {
+        let parsed = protocol
+            .parse::<OtlpProtocol>()
+            .map_err(|e| JsValue::from_str(&format!("setOtlpProtocol: {e}")))?;
+        self.otlp_protocol.set(Some(parsed));
+        Ok(())
+    }
+
+    /// Set extra HTTP headers for OTLP export as a flat `[key, value, ...]`
+    /// array. Only takes effect with an OTLP endpoint set, before the first send.
+    #[wasm_bindgen(js_name = "setOtlpHeaders")]
+    pub fn set_otlp_headers(&self, kv: Vec<String>) {
+        let headers = kv
+            .chunks_exact(2)
+            .map(|pair| (pair[0].clone(), pair[1].clone()))
+            .collect();
+        *self.otlp_headers.borrow_mut() = headers;
+    }
+
     #[wasm_bindgen]
     pub fn change_queue_ptr(&self) -> *const u8 {
         self.change_queue.as_ptr()
@@ -408,6 +454,19 @@ impl WasmSpanState {
             if self.use_v05.get() {
                 builder.set_output_format(TraceExporterOutputFormat::V05);
             }
+            // When an OTLP endpoint is configured, libdatadog exports traces via
+            // OTLP HTTP to that endpoint instead of the Datadog agent (mutually
+            // exclusive with the agent v0.4/v0.5 path).
+            if let Some(url) = self.otlp_endpoint.borrow().as_deref() {
+                builder.set_otlp_endpoint(url);
+                if let Some(protocol) = self.otlp_protocol.get() {
+                    builder.set_otlp_protocol(protocol);
+                }
+                let headers = self.otlp_headers.borrow();
+                if !headers.is_empty() {
+                    builder.set_otlp_headers(headers.clone());
+                }
+            }
             let built = builder
                 .build_async::<WasmCapabilities>()
                 .await

test/pipeline.js

@@ -827,6 +827,93 @@ describe('pipeline', () => {
       assert.ok(req, 'agent received a POST')
       assert.strictEqual(req.url, '/v0.5/traces')
     })
+
+    it('exports via OTLP HTTP after setOtlpEndpoint(url)', async () => {
+      // libdatadog maps its internal traces to OTLP and POSTs them to the
+      // configured endpoint instead of the Datadog agent. Confirms the OTLP
+      // path runs end-to-end over the wasm HTTP transport.
+      const http = require('node:http')
+      const seen = []
+      const server = http.createServer((req, res) => {
+        const chunks = []
+        req.on('data', c => chunks.push(c))
+        req.on('end', () => {
+          seen.push({
+            method: req.method,
+            url: req.url,
+            ct: req.headers['content-type'],
+            len: Buffer.concat(chunks).length
+          })
+          res.writeHead(200, { 'content-type': 'application/json' })
+          res.end('{}')
+        })
+      })
+      await new Promise(resolve => server.listen(0, '127.0.0.1', resolve))
+      const { port } = server.address()
+      const ns = new NativeSpansInterface({ agentUrl: `http://127.0.0.1:${port}` })
+      ns.state.setOtlpEndpoint(`http://127.0.0.1:${port}/v1/traces`)
+      const span = ns.createSpan()
+      span.name = 'otlp-span'
+      span.service = 'test-service'
+      span.resource = 'test-resource'
+      span.type = 'web'
+      span.duration = 1000000n
+      try {
+        await ns.flushSpans(span)
+        const req = seen.find(r => r.method === 'POST')
+        assert.ok(req, 'OTLP endpoint received a POST')
+        assert.strictEqual(req.url, '/v1/traces')
+        assert.match(req.ct || '', /json|protobuf/)
+        assert.ok(req.len > 0, 'OTLP body is non-empty')
+      } finally {
+        server.closeAllConnections?.()
+        server.close()
+      }
+    })
+
+    it('honors setOtlpProtocol(http/protobuf) and setOtlpHeaders', async () => {
+      const http = require('node:http')
+      let captured
+      const server = http.createServer((req, res) => {
+        req.on('data', () => {})
+        req.on('end', () => {
+          if (req.method === 'POST') {
+            captured = {
+              ct: req.headers['content-type'],
+              auth: req.headers.authorization
+            }
+          }
+          res.writeHead(200, { 'content-type': 'application/json' })
+          res.end('{}')
+        })
+      })
+      await new Promise(resolve => server.listen(0, '127.0.0.1', resolve))
+      const { port } = server.address()
+      const ns = new NativeSpansInterface({ agentUrl: `http://127.0.0.1:${port}` })
+      ns.state.setOtlpEndpoint(`http://127.0.0.1:${port}/v1/traces`)
+      ns.state.setOtlpProtocol('http/protobuf')
+      ns.state.setOtlpHeaders(['authorization', 'Bearer test-token'])
+      const span = ns.createSpan()
+      span.name = 'otlp-span'
+      span.service = 'test-service'
+      span.resource = 'test-resource'
+      span.type = 'web'
+      span.duration = 1000000n
+      try {
+        await ns.flushSpans(span)
+        assert.ok(captured, 'OTLP endpoint received a POST')
+        assert.match(captured.ct || '', /protobuf/)
+        assert.strictEqual(captured.auth, 'Bearer test-token')
+      } finally {
+        server.closeAllConnections?.()
+        server.close()
+      }
+    })
+
+    it('rejects unsupported OTLP protocols (e.g. grpc)', () => {
+      const ns = new NativeSpansInterface({ agentUrl: 'http://127.0.0.1:8126' })
+      assert.throws(() => ns.state.setOtlpProtocol('grpc'), /setOtlpProtocol|not supported/)
+    })
   })
 
   describe('client-side stats', () => {