feat(libdatadog): accept a borrowed transport agent

packages/libdatadog/README.md

@@ -17,6 +17,9 @@ The package accepts Datadog v0.4 MessagePack payloads and exports them to an
 agentless intake. `sendV04()` reports completion through a callback and sends
 delivery failures to the supplied logger. It does not return a promise.
 
+`createAgentlessExporter(options, { agent })` accepts an optional borrowed
+Node.js HTTP agent. The caller owns the agent and its lifetime.
+
 The package uses a wasm-bindgen backend with the WebAssembly bytes embedded in
 JavaScript. The canonical inlined output is published as the regular
 `@datadog/libdatadog-wasm` dependency from the `wasm` workspace. The WASM

packages/libdatadog/index.d.ts

@@ -80,6 +80,15 @@ export interface AgentlessExporterOptions {
   obfuscation?: ObfuscationConfig
 }
 
+interface AgentlessTransportAgent {
+  addRequest(request: object, options: object): void
+}
+
+export interface AgentlessTransportOptions {
+  /** Borrowed Node.js HTTP agent. The exporter does not destroy it. */
+  agent?: AgentlessTransportAgent
+}
+
 export interface AgentlessExporter {
   sendV04(payload: Uint8Array, done: () => void, log: AgentlessLogger): void
   close(): void
@@ -89,7 +98,10 @@ export interface AgentlessLogger {
   error(message: string, ...args: unknown[]): void
 }
 
-export function createAgentlessExporter(options: AgentlessExporterOptions): AgentlessExporter
+export function createAgentlessExporter(
+  options: AgentlessExporterOptions,
+  transportOptions?: AgentlessTransportOptions
+): AgentlessExporter
 export function backend(): 'wasm'
 
 export function zstd_compress(data: Uint8Array, level: number): Uint8Array

packages/libdatadog/lib/agentless-transport.js

@@ -4,13 +4,16 @@
 /** @typedef {{ id: number, url: string, method: string, headers: Header[], body: Uint8Array }} RequestPlan */
 /** @typedef {{ status: number | undefined, body: Buffer }} Response */
 /** @typedef {(error?: Error, response?: Response) => void} RequestCallback */
+/** @typedef {import('../index').AgentlessTransportOptions} AgentlessTransportOptions */
 
 const maxActiveBufferSize = 16 * 1024 * 1024
 const discardedResponse = { status: 200, body: Buffer.alloc(0) }
 
 let activeBufferSize = 0
 
-function createHostTransport () {
+/** @param {AgentlessTransportOptions} [options] */
+function createHostTransport ({ agent } = {}) {
+  const requestAgent = agent ?? false
   const requests = new Map()
   const timers = new Map()
 
@@ -28,6 +31,7 @@ function createHostTransport () {
     const target = new URL(url)
     const client = target.protocol === 'https:' ? require('node:https') : require('node:http')
     const headers = Object.fromEntries(headerList.map(({ name, value }) => [name, value]))
+    if (requestAgent === false) headers.connection = 'close'
 
     let settled = false
     /**
@@ -44,8 +48,8 @@ function createHostTransport () {
       return true
     }
     const outgoing = client.request(target, {
-      agent: false,
-      headers: { ...headers, connection: 'close' },
+      agent: requestAgent,
+      headers,
       method,
     }, (response) => {
       const chunks = []

packages/libdatadog/lib/agentless.js

@@ -5,6 +5,7 @@ const { randomUUID } = require('node:crypto')
 const { createHostTransport } = require('./agentless-transport')
 
 /** @typedef {import('../index').AgentlessExporterOptions} AgentlessExporterOptions */
+/** @typedef {import('../index').AgentlessTransportOptions} AgentlessTransportOptions */
 /** @typedef {import('../index').AgentlessLogger} AgentlessLogger */
 /** @typedef {typeof import('@datadog/libdatadog-wasm')} AgentlessBinding */
 
@@ -17,13 +18,14 @@ class AgentlessExporter {
   /**
    * @param {AgentlessBinding} binding
    * @param {AgentlessExporterOptions} options
+   * @param {AgentlessTransportOptions} [transportOptions]
    */
-  constructor (binding, options) {
+  constructor (binding, options, transportOptions) {
     const { runtimeId } = options
     const bindingOptions = runtimeId === undefined || runtimeId === null
       ? { ...options, runtimeId: randomUUID() }
       : options
-    const transport = createHostTransport()
+    const transport = createHostTransport(transportOptions)
     this.#binding = new binding.AgentlessExporter(
       bindingOptions,
       transport.request,
@@ -81,9 +83,10 @@ function errorMessage (error) {
 /**
  * @param {AgentlessBinding} binding
  * @param {AgentlessExporterOptions} options
+ * @param {AgentlessTransportOptions} [transportOptions]
  */
-function createAgentlessExporter (binding, options) {
-  return new AgentlessExporter(binding, options)
+function createAgentlessExporter (binding, options, transportOptions) {
+  return new AgentlessExporter(binding, options, transportOptions)
 }
 
 module.exports = { createAgentlessExporter }

packages/libdatadog/lib/wasm.js

@@ -9,7 +9,10 @@ module.exports = {
   zstd_compress: binding.zstd_compress,
 }
 
-/** @param {import('../index').AgentlessExporterOptions} options */
-function createAgentlessExporter (options) {
-  return require('./agentless').createAgentlessExporter(binding, options)
+/**
+ * @param {import('../index').AgentlessExporterOptions} options
+ * @param {import('../index').AgentlessTransportOptions} [transportOptions]
+ */
+function createAgentlessExporter (options, transportOptions) {
+  return require('./agentless').createAgentlessExporter(binding, options, transportOptions)
 }

packages/libdatadog/test/exporter.test.js

@@ -18,6 +18,39 @@ const wasmArtifact = path.join(packageRoot, 'wasm', 'dist', 'libdatadog_wasm.js'
 
 /** @typedef {(error?: unknown) => void} BindingDone */
 
+class TrackingAgent extends http.Agent {
+  connections = 0
+  requests = 0
+  destroyed = false
+
+  constructor () {
+    super({ keepAlive: true })
+  }
+
+  /**
+   * @param {import('node:http').ClientRequestArgs} options
+   * @param {(error: Error | null, stream: import('node:stream').Duplex) => void} [callback]
+   */
+  createConnection (options, callback) {
+    this.connections++
+    return super.createConnection(options, callback)
+  }
+
+  /**
+   * @param {import('node:http').ClientRequest} request
+   * @param {import('node:http').RequestOptions} options
+   */
+  addRequest (request, options) {
+    this.requests++
+    super.addRequest(request, options)
+  }
+
+  destroy () {
+    this.destroyed = true
+    super.destroy()
+  }
+}
+
 test('package entry points defer unused agentless modules', {
   skip: !fs.existsSync(wasmArtifact),
 }, () => {
@@ -188,6 +221,21 @@ test('package entry point compresses agentless v0.4 exports with Zstandard', {
   await assertExport(require('..'))
 })
 
+test('package entry point uses a borrowed transport agent', {
+  skip: !fs.existsSync(wasmArtifact),
+}, async () => {
+  const agent = new TrackingAgent()
+
+  try {
+    await assertExport(require('..'), { agent }, 2)
+    assert.strictEqual(agent.requests, 2)
+    assert.strictEqual(agent.connections, 1)
+    assert.strictEqual(agent.destroyed, false)
+  } finally {
+    agent.destroy()
+  }
+})
+
 test('inline-WASM backend validates optional values', {
   skip: !fs.existsSync(wasmArtifact),
 }, async () => {
@@ -400,8 +448,10 @@ test('agentless exporter close cancels retry backoff', {
 
 /**
  * @param {typeof import('..')} pipeline
+ * @param {import('../index').AgentlessTransportOptions} [transportOptions]
+ * @param {number} [count]
  */
-async function assertExport (pipeline) {
+async function assertExport (pipeline, transportOptions, count = 1) {
   const received = await withIntake(async (endpoint) => {
     const exporter = pipeline.createAgentlessExporter({
       endpoint,
@@ -412,10 +462,12 @@ async function assertExport (pipeline) {
       runtimeId: 'runtime-id',
       service: 'service',
       containerId: 'container-id',
-    })
+    }, transportOptions)
 
     try {
-      await sendExport(exporter)
+      for (let i = 0; i < count; i++) {
+        await sendExport(exporter)
+      }
     } finally {
       exporter.close()
     }

packages/libdatadog/test/types.test.ts

@@ -1,5 +1,6 @@
 import {
   type AgentlessLogger,
+  type AgentlessTransportOptions,
   backend,
   createAgentlessExporter,
   DDSketch,
@@ -20,7 +21,15 @@ const agentlessExporter = createAgentlessExporter({
   tracerVersion: '1.2.3',
   languageVersion: '22.0.0',
   languageInterpreter: 'v8',
+}, {
+  agent: {
+    addRequest () {},
+  },
 })
+const invalidTransportOptions: AgentlessTransportOptions = {
+  // @ts-expect-error Transport agents must implement addRequest.
+  agent: {},
+}
 const logger: AgentlessLogger = {
   error () {},
 }