dogstatsd.js

'use strict'

const dgram = require('dgram')
const isIP = require('net').isIP

const { channel } = require('dc-polyfill')
const { storage } = require('../../datadog-core')
const request = require('./exporters/common/request')
const log = require('./log')
const Histogram = require('./histogram')
const { entityId } = require('./exporters/common/docker')

const legacyStorage = storage('legacy')

const MAX_BUFFER_SIZE = 1024 // limit from the agent

const TYPE_COUNTER = 'c'
const TYPE_GAUGE = 'g'
const TYPE_DISTRIBUTION = 'd'
const TYPE_HISTOGRAM = 'h'

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

/**
 * @import { DogStatsD } from "../../../index.d.ts"
 * @implements {DogStatsD}
 */
class DogStatsDClient {
  #lookup
  #tagsPrefix
  constructor (options) {
    this.#lookup = options.lookup
    if (options.metricsProxyUrl) {
      this._httpOptions = {
        method: 'POST',
        url: options.metricsProxyUrl.toString(),
        path: '/dogstatsd/v2/proxy',
      }
    }

    this._host = options.host
    this._family = isIP(this._host)
    this._port = options.port
    this._tags = options.tags
    this.#tagsPrefix = this._tags.length ? `|#${this._tags.join(',')}` : ''
    this._queue = []
    this._buffer = ''
    this._offset = 0
    this._udp4 = this._socket('udp4')
    this._udp6 = this._socket('udp6')
  }

  /**
   * Recomputes the cached tags and tag-prefix (mirrors the constructor) after a `config.tags`
   * change, e.g. a MicroVM clone resume.
   *
   * Buffered lines have the old prefix baked in, and on a clone resume they were produced during
   * the image build, so every clone holds the same bytes — flushing them would submit one identical
   * copy per clone. Dropping is right here for that reason only: for a tag change on a live process
   * the buffer holds unique data whose old tags are still correct, so that case wants a flush
   * before the swap.
   *
   * @param {string[]} tags - DogStatsD-formatted tags (e.g. `['key:value']`)
   * @returns {boolean} True if the tag prefix actually changed (and buffered lines were dropped)
   */
  updateTags (tags) {
    const tagsPrefix = tags.length ? `|#${tags.join(',')}` : ''

    this._tags = tags

    if (tagsPrefix === this.#tagsPrefix) return false

    this.#tagsPrefix = tagsPrefix
    this._queue = []
    this._buffer = ''
    this._offset = 0

    return true
  }

  increment (stat, value, tags) {
    this._add(stat, value, TYPE_COUNTER, tags)
  }

  decrement (stat, value, tags) {
    this._add(stat, -value, TYPE_COUNTER, tags)
  }

  gauge (stat, value, tags) {
    this._add(stat, value, TYPE_GAUGE, tags)
  }

  distribution (stat, value, tags) {
    this._add(stat, value, TYPE_DISTRIBUTION, tags)
  }

  histogram (stat, value, tags) {
    this._add(stat, value, TYPE_HISTOGRAM, tags)
  }

  flush () {
    const queue = this._enqueue()

    if (queue.length === 0) return

    log.debug('Flushing %s metrics via %s', queue.length, this._httpOptions ? 'HTTP' : 'UDP')

    this._queue = []

    if (this._httpOptions) {
      this._sendHttp(queue)
    } else {
      this._sendUdp(queue)
    }
  }

  _sendHttp (queue) {
    const buffer = Buffer.concat(queue)
    request(buffer, this._httpOptions, (err) => {
      if (err) {
        log.error('DogStatsDClient: HTTP error from agent: %s', err.message, err)
        if (err.status === 404) {
          // Inside this if-block, we have connectivity to the agent, but
          // we're not getting a 200 from the proxy endpoint. If it's a 404,
          // then we know we'll never have the endpoint, so just clear out the
          // options. Either way, we can give UDP a try.
          this._httpOptions = undefined
        }
        this._sendUdp(queue)
      }
    })
  }

  _sendUdp (queue) {
    // dgram resolves the local address via the instrumented dns.lookup when it
    // binds on first send; the noop store keeps that self-traffic off the trace.
    legacyStorage.run({ noop: true }, () => {
      if (this._family === 0) {
        this.#lookup(this._host, (error, address, family) => {
          if (error) return log.error('DogStatsDClient: Host not found', error)
          this._sendUdpFromQueue(queue, address, family)
        })
      } else {
        this._sendUdpFromQueue(queue, this._host, this._family)
      }
    })
  }

  _sendUdpFromQueue (queue, address, family) {
    const socket = family === 6 ? this._udp6 : this._udp4

    for (const buffer of queue) {
      log.debug('Sending to DogStatsD: %s', buffer)
      socket.send(buffer, 0, buffer.length, this._port, address)
    }
  }

  _add (stat, value, type, tags) {
    let message = `${stat}:${value}|${type}`

    if (tags?.length) {
      message += this.#tagsPrefix
        ? `${this.#tagsPrefix},${tags.join(',')}`
        : `|#${tags.join(',')}`
    } else {
      message += this.#tagsPrefix
    }

    if (entityId) {
      message += `|c:${entityId}`
    }

    this._write(`${message}\n`)
  }

  _write (message) {
    const offset = Buffer.byteLength(message)

    if (this._offset + offset > MAX_BUFFER_SIZE) {
      this._enqueue()
    }

    this._offset += offset
    this._buffer += message
  }

  _enqueue () {
    if (this._offset > 0) {
      this._queue.push(Buffer.from(this._buffer))
      this._buffer = ''
      this._offset = 0
    }

    return this._queue
  }

  _socket (type) {
    const socket = dgram.createSocket(type)

    socket.on('error', () => {})
    socket.unref?.()

    return socket
  }

  /**
   * @param {import('./config/config-base')} config - Tracer configuration
   */
  static generateClientConfig (config) {
    const tags = []

    if (config.tags) {
      for (const [key, value] of Object.entries(config.tags)) {
        // Skip runtime-id unless enabled as cardinality may be too high
        if (typeof value === 'string' && (key !== 'runtime-id' || config.runtimeMetricsRuntimeId)) {
          // https://docs.datadoghq.com/tagging/#defining-tags
          const valueStripped = value.replaceAll(/[^a-z0-9_:./-]/ig, '_')

          tags.push(`${key}:${valueStripped}`)
        }
      }
    }

    const clientConfig = {
      host: config.dogstatsd.hostname,
      port: config.dogstatsd.port,
      tags,
      lookup: config.lookup,
    }

    if (config.url) {
      clientConfig.metricsProxyUrl = config.url
    }

    return clientConfig
  }
}

class MetricsAggregationClient {
  constructor (client) {
    this._client = client

    this.reset()
  }

  /**
   * Recomputes the wrapped client's cached tags (e.g. after a MicroVM clone resume). Pending
   * counters/gauges/histograms were aggregated under the old identity, so they're reset along
   * with the client's buffered lines — but only if the tags actually changed, so a no-op resume
   * doesn't discard in-flight aggregation for nothing.
   * @param {string[]} tags - DogStatsD-formatted tags (e.g. `['key:value']`)
   */
  updateTags (tags) {
    if (this._client.updateTags(tags)) {
      this.reset()
    }
  }

  flush () {
    this._captureCounters()
    this._captureGauges()
    this._captureHistograms()

    this._client.flush()
  }

  reset () {
    this._counters = new Map()
    this._gauges = new Map()
    this._histograms = new Map()
  }

  // TODO: Aggregate with a histogram and send the buckets to the client.
  distribution (name, value, tags) {
    this._client.distribution(name, value, tags)
  }

  boolean (name, value, tags) {
    this.gauge(name, value ? 1 : 0, tags)
  }

  histogram (name, value, tags) {
    const node = this._ensureTree(this._histograms, name, tags, null)

    if (!node.value) {
      node.value = new Histogram()
    }

    node.value.record(value)
  }

  count (name, count, tags = [], monotonic = true) {
    if (typeof tags === 'boolean') {
      monotonic = tags
      tags = []
    }

    const container = monotonic ? this._counters : this._gauges
    const node = this._ensureTree(container, name, tags, 0)

    node.value += count
  }

  gauge (name, value, tags) {
    const node = this._ensureTree(this._gauges, name, tags, 0)

    node.value = value
  }

  increment (name, count = 1, tags) {
    this.count(name, count, tags)
  }

  decrement (name, count = 1, tags) {
    this.count(name, -count, tags)
  }

  _captureGauges () {
    this._captureTree(this._gauges, (node, name, tags) => {
      this._client.gauge(name, node.value, tags)
    })

    this._gauges.clear()
  }

  _captureCounters () {
    this._captureTree(this._counters, (node, name, tags) => {
      this._client.increment(name, node.value, tags)
    })

    this._counters.clear()
  }

  _captureHistograms () {
    this._captureTree(this._histograms, (node, name, tags) => {
      const stats = node.value

      this._client.gauge(`${name}.min`, stats.min, tags)
      this._client.gauge(`${name}.max`, stats.max, tags)
      this._client.increment(`${name}.sum`, stats.sum, tags)
      this._client.increment(`${name}.total`, stats.sum, tags)
      this._client.gauge(`${name}.avg`, stats.avg, tags)
      this._client.increment(`${name}.count`, stats.count, tags)
      this._client.gauge(`${name}.median`, stats.median, tags)
      this._client.gauge(`${name}.95percentile`, stats.p95, tags)
    })

    this._histograms.clear()
  }

  _captureTree (tree, fn) {
    for (const [name, root] of tree) {
      this._captureNode(root, name, [], fn)
    }
  }

  _captureNode (node, name, tags, fn) {
    if (node.touched) {
      fn(node, name, tags)
    }

    for (const [tag, next] of node.nodes) {
      tags.push(tag)
      this._captureNode(next, name, tags, fn)
      tags.pop()
    }
  }

  _ensureTree (tree, name, tags = [], value) {
    if (!Array.isArray(tags)) {
      tags = [tags]
    }

    let node = this._ensureNode(tree, name, value)

    for (const tag of tags) {
      node = this._ensureNode(node.nodes, tag, value)
    }

    node.touched = true

    return node
  }

  _ensureNode (container, key, value) {
    let node = container.get(key)

    if (!node) {
      node = { nodes: new Map(), touched: false, value }

      if (typeof key === 'string') {
        container.set(key, node)
      }
    }

    return node
  }
}

/**
 * This is a simplified user-facing proxy to the underlying DogStatsDClient instance
 *
 * @implements {DogStatsD}
 */
class CustomMetrics {
  #client
  constructor (config) {
    const clientConfig = DogStatsDClient.generateClientConfig(config)
    this.#client = new MetricsAggregationClient(new DogStatsDClient(clientConfig))

    // CustomMetrics has process-lifetime flush handlers and no stop hook, so this shares that lifetime.
    identityRefreshChannel.subscribe(() => {
      this.#client.updateTags(DogStatsDClient.generateClientConfig(config).tags)
    })

    const flush = this.flush.bind(this)

    // TODO(bengl) this magic number should be configurable
    setInterval(flush, 10 * 1000).unref?.()

    globalThis[Symbol.for('dd-trace')].beforeExitHandlers.add(flush)
  }

  increment (stat, value = 1, tags) {
    this.#client.increment(stat, value, CustomMetrics.tagTranslator(tags))
  }

  decrement (stat, value = 1, tags) {
    this.#client.decrement(stat, value, CustomMetrics.tagTranslator(tags))
  }

  gauge (stat, value, tags) {
    this.#client.gauge(stat, value, CustomMetrics.tagTranslator(tags))
  }

  distribution (stat, value, tags) {
    this.#client.distribution(stat, value, CustomMetrics.tagTranslator(tags))
  }

  histogram (stat, value, tags) {
    this.#client.histogram(stat, value, CustomMetrics.tagTranslator(tags))
  }

  flush () {
    return this.#client.flush()
  }

  /**
   * Exposing { tagName: 'tagValue' } to the end user
   * These are translated into [ 'tagName:tagValue' ] for internal use
   */
  static tagTranslator (objTags) {
    if (Array.isArray(objTags)) return objTags

    const arrTags = []

    if (!objTags) return arrTags

    for (const [key, value] of Object.entries(objTags)) {
      arrTags.push(`${key}:${value}`)
    }

    return arrTags
  }
}

module.exports = {
  DogStatsDClient,
  CustomMetrics,
  MetricsAggregationClient,
}