index.js

'use strict'

const { URL } = require('node:url')
const os = require('node:os')

const { channel } = require('dc-polyfill')

const log = require('../../log')
const { entityId } = require('../common/docker')
const tracerVersion = require('../../../../../package.json').version
const Writer = require('./writer')
const { computeIntakeUrl } = require('./intake')

const identityRefreshChannel = channel('datadog:identity:refresh')

// Only one AgentlessExporter is ever live in a real process, so replacing the subscription on
// construction is safe - it just keeps tests (which build several) from piling up listeners.
let unsubscribeBatchReset = null

/**
 * Agentless exporter for APM trace intake.
 * Sends traces directly to the Datadog intake without requiring a local agent.
 * Batches multiple traces per request using timer-based flushing.
 */
class AgentlessExporter {
  #timer
  #config

  /**
   * @param {object} config - Configuration object
   * @param {string} [config.site] - The Datadog site. Defaults to 'datadoghq.com'.
   * @param {number} [config.flushInterval] - Batch flush interval in ms
   * @param {string} [config.env] - Environment name
   * @param {object} config.tags - Tags including runtime-id
   */
  constructor (config) {
    this.#config = config
    const site = config.site ?? 'datadoghq.com'

    try {
      // Agentless traffic carries the Datadog API key, so the intake is always an https endpoint
      // derived from the site; never config.url (the agent's cleartext http) or the key leaks.
      this._url = new URL(computeIntakeUrl(site))
    } catch (err) {
      log.error('Invalid site for agentless exporter. site=%s. Error: %s', site, err.message)
      this._url = null
    }

    const metadata = {
      hostname: os.hostname(),
      languageName: 'nodejs',
      languageVersion: process.version,
      tracerVersion,
      // Read live off `config` (instead of copying the value) so a later change
      // (e.g. a MicroVM clone resume) is picked up by the next `JSON.stringify` in the encoder.
      get env () { return config.env },
      get runtimeID () { return config.tags['runtime-id'] },
      ...(entityId ? { containerID: entityId } : {}),
    }

    this._writer = new Writer({
      url: this._url,
      site,
      metadata,
    })

    // A clone resume shouldn't flush spans buffered before the snapshot under its own identity.
    unsubscribeBatchReset?.()
    const onIdentityRefresh = () => this._writer.resetPendingBatch()
    identityRefreshChannel.subscribe(onIdentityRefresh)
    unsubscribeBatchReset = () => identityRefreshChannel.unsubscribe(onIdentityRefresh)

    const ddTrace = globalThis[Symbol.for('dd-trace')]
    if (ddTrace?.beforeExitHandlers) {
      ddTrace.beforeExitHandlers.add(this.flush.bind(this))
    } else {
      log.error('dd-trace global not properly initialized. beforeExit handler not registered for agentless exporter.')
    }
  }

  /**
   * Sets the intake URL.
   * @param {string} urlString - The new intake URL
   * @returns {boolean} True if URL was set successfully
   */
  setUrl (urlString) {
    try {
      const url = new URL(urlString)
      this._url = url
      this._writer.setUrl(url)
      return true
    } catch (err) {
      log.error(
        'Invalid URL for agentless exporter: %s. Using previous URL: %s. Error: %s',
        urlString,
        this._url?.href || 'none',
        err.message
      )
      return false
    }
  }

  /**
   * Exports a trace. Traces are batched and flushed on a timer.
   * @param {object[]} spans - Array of spans (all from the same trace)
   */
  export (spans) {
    this._writer.append(spans)

    const { flushInterval } = this.#config

    if (flushInterval === 0) {
      try {
        this._writer.flush()
      } catch (err) {
        log.error('Failed to flush traces: %s', err.message)
      }
    } else if (this.#timer === undefined) {
      this.#timer = setTimeout(() => {
        try {
          this._writer.flush()
        } catch (err) {
          log.error('Failed to flush traces on timer: %s', err.message)
        }
        this.#timer = undefined
      }, flushInterval)
      this.#timer.unref?.()
    }
  }

  /**
   * Flushes any pending traces immediately. Clears the batch timer.
   * @param {Function} [done] - Callback when flush is complete
   */
  flush (done = () => {}) {
    clearTimeout(this.#timer)
    this.#timer = undefined
    try {
      this._writer.flush(done)
    } catch (err) {
      log.error('Failed to flush traces: %s', err.message)
      done()
    }
  }
}

module.exports = AgentlessExporter