feat: add support for Distributed Tracing (#538)

Author: Qard

docs/agent-api.asciidoc

@@ -1027,7 +1027,7 @@ NOTE: `apm.middleware.connect` _must_ be added to the middleware stack _before_
 
 [source,js]
 ----
-var transaction = apm.startTransaction([name][, type])
+var transaction = apm.startTransaction([name][, type][, traceparent])
 ----
 
 Start a new transaction.
@@ -1042,6 +1042,8 @@ Defaults to `unnamed`
 You can always set this later via <<transaction-type,`transaction.type`>>.
 Defaults to `custom`
 
+* `traceparent` - The traceparent header received from a remote service (string).
+
 Use this function to create a custom transaction.
 Note that the agent will do this for you automatically when ever your application receives an incoming HTTP request.
 You only need to use this function to create custom transactions.

lib/agent.js

@@ -54,6 +54,12 @@ Object.defineProperty(Agent.prototype, 'currentTransaction', {
   }
 })
 
+Object.defineProperty(Agent.prototype, 'currentSpan', {
+  get () {
+    return this._instrumentation.currentSpan
+  }
+})
+
 Agent.prototype.destroy = function () {
   if (this._apmServer) this._apmServer.destroy()
 }
@@ -242,8 +248,9 @@ Agent.prototype.captureError = function (err, opts, cb) {
 
   var agent = this
   var trans = this.currentTransaction
+  var span = this.currentSpan
   var timestamp = Date.now() * 1000
-  var id = crypto.randomBytes(128 / 8).toString('hex')
+  var context = (span || trans || {}).context || {}
   var req = opts && opts.request instanceof IncomingMessage
     ? opts.request
     : trans && trans.req
@@ -271,7 +278,9 @@ Agent.prototype.captureError = function (err, opts, cb) {
   }
 
   function prepareError (error) {
-    error.id = id
+    error.id = crypto.randomBytes(16).toString('hex')
+    error.parent_id = context.id
+    error.trace_id = context.traceId
     error.timestamp = timestamp
     error.context = {
       user: Object.assign(
@@ -291,7 +300,8 @@ Agent.prototype.captureError = function (err, opts, cb) {
         opts && opts.custom
       )
     }
-    if (trans) error.transaction = { id: trans.id }
+
+    if (trans) error.transaction_id = trans.id
 
     if (error.exception) {
       error.exception.handled = !opts || opts.handled
@@ -345,21 +355,21 @@ Agent.prototype.captureError = function (err, opts, cb) {
     error = agent._errorFilters.process(error)
 
     if (!error) {
-      agent.logger.debug('error ignored by filter %o', { id: id })
-      if (cb) cb(null, id)
+      agent.logger.debug('error ignored by filter %o', { id: error.id })
+      if (cb) cb(null, error.id)
       return
     }
 
     if (agent._apmServer) {
-      agent.logger.info(`Sending error ${id} to Elastic APM`)
+      agent.logger.info(`Sending error to Elastic APM`, { id: error.id })
       agent._apmServer.sendError(error, function () {
         agent._apmServer.flush(function (err) {
-          if (cb) cb(err, id)
+          if (cb) cb(err, error.id)
         })
       })
     } else if (cb) {
       // TODO: Swallow this error just as it's done in agent.flush()?
-      process.nextTick(cb.bind(null, new Error('cannot capture error before agent is started'), id))
+      process.nextTick(cb.bind(null, new Error('cannot capture error before agent is started'), error.id))
     }
   }
 }

lib/instrumentation/async-hooks.js

@@ -3,17 +3,29 @@
 const asyncHooks = require('async_hooks')
 
 module.exports = function (ins) {
-  const asyncHook = asyncHooks.createHook({ init, destroy })
-  const transactions = new Map()
+  const asyncHook = asyncHooks.createHook({ init, before, destroy })
 
+  const activeTransactions = new Map()
   Object.defineProperty(ins, 'currentTransaction', {
     get () {
       const asyncId = asyncHooks.executionAsyncId()
-      return transactions.has(asyncId) ? transactions.get(asyncId) : null
+      return activeTransactions.has(asyncId) ? activeTransactions.get(asyncId) : null
     },
     set (trans) {
       const asyncId = asyncHooks.executionAsyncId()
-      transactions.set(asyncId, trans)
+      activeTransactions.set(asyncId, trans)
+    }
+  })
+
+  const activeSpans = new Map()
+  Object.defineProperty(ins, 'activeSpan', {
+    get () {
+      const asyncId = asyncHooks.executionAsyncId()
+      return activeSpans.has(asyncId) ? activeSpans.get(asyncId) : null
+    },
+    set (span) {
+      const asyncId = asyncHooks.executionAsyncId()
+      activeSpans.set(asyncId, span)
     }
   })
 
@@ -25,11 +37,21 @@ module.exports = function (ins) {
     // type, which will init for each scheduled timer.
     if (type === 'TIMERWRAP') return
 
-    transactions.set(asyncId, ins.currentTransaction)
+    activeTransactions.set(asyncId, ins.currentTransaction)
+    activeSpans.set(asyncId, ins.bindingSpan || ins.activeSpan)
+  }
+
+  function before (asyncId) {
+    ins.bindingSpan = null
   }
 
   function destroy (asyncId) {
-    if (!transactions.has(asyncId)) return // in case type === TIMERWRAP
-    transactions.delete(asyncId)
+    // in case type === TIMERWRAP
+    if (activeTransactions.has(asyncId)) {
+      activeTransactions.delete(asyncId)
+    }
+    if (activeSpans.has(asyncId)) {
+      activeSpans.delete(asyncId)
+    }
   }
 }

lib/instrumentation/http-shared.js

@@ -18,7 +18,8 @@ exports.instrumentRequest = function (agent, moduleName) {
           // don't leak previous transaction
           agent._instrumentation.currentTransaction = null
         } else {
-          var trans = agent.startTransaction()
+          var traceparent = req.headers['elastic-apm-traceparent']
+          var trans = agent.startTransaction(null, null, traceparent)
           trans.type = 'request'
           trans.req = req
           trans.res = res
@@ -78,19 +79,52 @@ function isRequestBlacklisted (agent, req) {
   return false
 }
 
+// NOTE: This will also stringify and parse URL instances
+// to a format which can be mixed into the options object.
+function ensureUrl (v) {
+  if (typeof v === 'string' || v instanceof url.URL) {
+    return url.parse(String(v))
+  } else {
+    return v
+  }
+}
+
 exports.traceOutgoingRequest = function (agent, moduleName) {
   var spanType = 'ext.' + moduleName + '.http'
   var ins = agent._instrumentation
 
   return function (orig) {
-    return function () {
+    return function (...args) {
       var span = agent.buildSpan()
       var id = span && span.transaction.id
 
       agent.logger.debug('intercepted call to %s.request %o', moduleName, { id: id })
 
-      var req = orig.apply(this, arguments)
+      var options = {}
+      var newArgs = [ options ]
+      for (let arg of args) {
+        if (typeof arg === 'function') {
+          newArgs.push(arg)
+        } else {
+          Object.assign(options, ensureUrl(arg))
+        }
+      }
+
+      if (!options.headers) options.headers = {}
+
+      // Attempt to use the span context as a traceparent header.
+      // If the transaction is unsampled the span will not exist,
+      // however a traceparent header must still be propagated
+      // to indicate requested services should not be sampled.
+      // Use the transaction context as the parent, in this case.
+      var parent = span || agent.currentTransaction
+      if (parent && parent.context) {
+        options.headers['elastic-apm-traceparent'] = parent.context.toString()
+      }
+
+      var req = orig.apply(this, newArgs)
       if (!span) return req
+
       if (req._headers.host === agent._conf.serverHost) {
         agent.logger.debug('ignore %s request to intake API %o', moduleName, { id: id })
         return req

lib/instrumentation/index.js

@@ -46,6 +46,18 @@ function Instrumentation (agent) {
   this._hook = null // this._hook is only exposed for testing purposes
   this._started = false
   this.currentTransaction = null
+
+  // Span for binding callbacks
+  this.bindingSpan = null
+
+  // Span which is actively bound
+  this.activeSpan = null
+
+  Object.defineProperty(this, 'currentSpan', {
+    get () {
+      return this.bindingSpan || this.activeSpan
+    }
+  })
 }
 
 Instrumentation.modules = Object.freeze(MODULES)
@@ -130,8 +142,8 @@ Instrumentation.prototype.addEndedSpan = function (span) {
   }
 }
 
-Instrumentation.prototype.startTransaction = function (name, type) {
-  return new Transaction(this._agent, name, type)
+Instrumentation.prototype.startTransaction = function (name, type, traceparent) {
+  return new Transaction(this._agent, name, type, traceparent)
 }
 
 Instrumentation.prototype.endTransaction = function (result) {
@@ -174,17 +186,20 @@ Instrumentation.prototype.bindFunction = function (original) {
 
   var ins = this
   var trans = this.currentTransaction
+  var span = this.currentSpan
   if (trans && !trans.sampled) {
     return original
   }
 
   return elasticAPMCallbackWrapper
 
   function elasticAPMCallbackWrapper () {
-    var prev = ins.currentTransaction
+    var prevTrans = ins.currentTransaction
     ins.currentTransaction = trans
+    ins.bindingSpan = null
+    ins.activeSpan = span
     var result = original.apply(this, arguments)
-    ins.currentTransaction = prev
+    ins.currentTransaction = prevTrans
     return result
   }
 }

lib/instrumentation/span.js

@@ -1,7 +1,5 @@
 'use strict'
 
-var crypto = require('crypto')
-
 var afterAll = require('after-all-results')
 var Value = require('async-value-promise')
 
@@ -14,7 +12,6 @@ const TEST = process.env.ELASTIC_APM_TEST
 module.exports = Span
 
 function Span (transaction) {
-  this.id = crypto.randomBytes(64 / 8).toString('hex') // TODO: Replace with correct id once OT is ready
   this.transaction = transaction
   this.started = false
   this.ended = false
@@ -26,9 +23,18 @@ function Span (transaction) {
   this._stackObj = null
   this._agent = transaction._agent
 
+  var current = this._agent._instrumentation.activeSpan || transaction
+  this.context = current.context.child()
+
   this._agent.logger.debug('init span %o', { id: this.transaction.id })
 }
 
+Object.defineProperty(Span.prototype, 'id', {
+  get () {
+    return this.context.id
+  }
+})
+
 Span.prototype.start = function (name, type) {
   if (this.started) {
     this._agent.logger.debug('tried to call span.start() on already started span %o', { id: this.transaction.id, name: this.name, type: this.type })
@@ -39,6 +45,8 @@ Span.prototype.start = function (name, type) {
   this.name = name || this.name || 'unnamed'
   this.type = type || this.type || 'custom'
 
+  this._agent._instrumentation.bindingSpan = this
+
   if (this._agent._conf.captureSpanStackTraces && !this._stackObj) {
     this._recordStackTrace()
   }
@@ -140,10 +148,10 @@ Span.prototype._encode = function (cb) {
     }
 
     var payload = {
-      id: self.id,
+      id: self.context.id,
       transaction_id: self.transaction.id,
-      parent_id: self.transaction.id,
-      trace_id: self.transaction.traceId,
+      parent_id: self.context.parentId,
+      trace_id: self.context.traceId,
       name: self.name,
       type: self.type,
       timestamp: self.timestamp,