test: refactor to move knowledge for in-process Agent re-use into Agent methods

Move away from the test/_apm_server.js and test/_agent.js pattern that
encapsulates internal details of Agent and Instrumentation to allow
re-use of an Agent instance in-process in the same test file. The
pattern from these test support files also makes leftover state from a
test case an issue for *subsequent* test cases to deal with.

Instead we add agent._testReset (and ins.testReset) to encapsulate
these internal details.

This also fixes a bug in agent.test.js where a new async-hook was
being created and enabled for each Agent re-use. Cleanup was never
disabling those async-hooks.

Author: trentm

lib/agent.js

@@ -46,6 +46,31 @@ function Agent () {
   this.lambda = elasticApmAwsLambda(this)
 }
 
+// Reset the running state of this agent to a (relatively) clean state, to
+// allow re-`.start()` and re-use of this Agent within the same process for
+// testing. This is not supported outside of the test suite.
+//
+// Limitations: There are untracked async tasks (a) between span.end() and
+// and transport.sendSpan(); and (b) between agent.captureError() and
+// transport.sendError(). Currently there is no way to wait for completion of
+// those tasks in this method. There is code in each of ins.addEndedSpan and
+// agent.captureError to mitigate this.
+Agent.prototype._testReset = function () {
+  this._errorFilters = new Filters()
+  this._transactionFilters = new Filters()
+  this._spanFilters = new Filters()
+  if (this._transport && this._transport.destroy) {
+    this._transport.destroy()
+  }
+  this._transport = null
+  global[symbols.agentInitialized] = null // Mark as not yet started.
+  if (this._uncaughtExceptionListener) {
+    process.removeListener('uncaughtException', this._uncaughtExceptionListener)
+  }
+  this._metrics.stop()
+  this._instrumentation.testReset()
+}
+
 Object.defineProperty(Agent.prototype, 'currentTransaction', {
   get () {
     return this._instrumentation.currentTransaction
@@ -394,6 +419,12 @@ Agent.prototype.captureError = function (err, opts, cb) {
     span._setOutcomeFromErrorCapture(constants.OUTCOME_FAILURE)
   }
 
+  // Ensure an inflight `agent.captureError()` across a call to
+  // `agent._testReset()` does not impact testing of the agent after that call
+  // -- by using *current* values of _errorFilters and _transport.
+  const errorFilters = agent._errorFilters
+  const transport = agent._transport
+
   // Move the remaining captureError processing to a later tick because:
   // 1. This allows the calling code to continue processing. For example, for
   //    Express instrumentation this can significantly improve latency in
@@ -454,7 +485,7 @@ Agent.prototype.captureError = function (err, opts, cb) {
       // _err is always null from createAPMError.
       const id = apmError.id
 
-      apmError = agent._errorFilters.process(apmError)
+      apmError = errorFilters.process(apmError)
       if (!apmError) {
         agent.logger.debug('error ignored by filter %o', { id })
         if (cb) {
@@ -463,9 +494,9 @@ Agent.prototype.captureError = function (err, opts, cb) {
         return
       }
 
-      if (agent._transport) {
+      if (transport) {
         agent.logger.info('Sending error to Elastic APM: %o', { id })
-        agent._transport.sendError(apmError, function () {
+        transport.sendError(apmError, function () {
           agent.flush(function (flushErr) {
             if (cb) {
               cb(flushErr, id)

lib/instrumentation/async-hooks.js

@@ -3,20 +3,12 @@
 const asyncHooks = require('async_hooks')
 const shimmer = require('./shimmer')
 
-// FOR INTERNAL TESTING PURPOSES ONLY!
-const resettable = process.env._ELASTIC_APM_ASYNC_HOOKS_RESETTABLE === 'true'
-let _asyncHook
-
 module.exports = function (ins) {
   const asyncHook = asyncHooks.createHook({ init, before, destroy })
-  const contexts = new WeakMap()
-
-  if (resettable) {
-    if (_asyncHook) _asyncHook.disable()
-    _asyncHook = asyncHook
-  }
+  let activeSpans = new Map()
+  let activeTransactions = new Map()
+  let contexts = new WeakMap()
 
-  const activeTransactions = new Map()
   Object.defineProperty(ins, 'currentTransaction', {
     get () {
       const asyncId = asyncHooks.executionAsyncId()
@@ -32,7 +24,6 @@ module.exports = function (ins) {
     }
   })
 
-  const activeSpans = new Map()
   Object.defineProperty(ins, 'activeSpan', {
     get () {
       const asyncId = asyncHooks.executionAsyncId()
@@ -63,6 +54,18 @@ module.exports = function (ins) {
     }
   })
 
+  shimmer.wrap(ins, 'testReset', function (origTestReset) {
+    return function wrappedTestReset () {
+      asyncHook.disable()
+      activeTransactions = new Map()
+      activeSpans = new Map()
+      contexts = new WeakMap()
+      shimmer.unwrap(ins, 'addEndedTransaction')
+      shimmer.unwrap(ins, 'testReset')
+      return origTestReset.call(this)
+    }
+  })
+
   asyncHook.enable()
 
   function init (asyncId, type, triggerAsyncId, resource) {

lib/instrumentation/index.js

@@ -86,6 +86,27 @@ function Instrumentation (agent) {
   }
 }
 
+// Reset internal context tracking state for (relatively) clean re-use of this
+// Instrumentation in the same process. Used for testing.
+//
+// Limitations: Removing and re-applying 'require-in-the-middle'-based patches
+// has no way to update user or test code that already has references to
+// patched or unpatched exports from those modules. That may mean some
+// automatic instrumentation may not work.
+Instrumentation.prototype.testReset = function () {
+  // Reset context tracking.
+  this.currentTransaction = null
+  this.bindingSpan = null
+  this.activeSpan = null
+
+  // Reset patching.
+  this._started = false
+  if (this._hook) {
+    this._hook.unhook()
+    this._hook = null
+  }
+}
+
 Object.defineProperty(Instrumentation.prototype, 'ids', {
   get () {
     const current = this.currentSpan || this.currentTransaction
@@ -232,21 +253,30 @@ Instrumentation.prototype.addEndedSpan = function (span) {
     // Save effort and logging if disableSend=true.
   } else if (this._started) {
     agent.logger.debug('encoding span %o', { span: span.id, parent: span.parentId, trace: span.traceId, name: span.name, type: span.type })
+
+    // Ensure an inflight `span._encode()` across a call to `agent._testReset()`
+    // does not impact testing of the agent after that call -- by using
+    // *current* values of _spanFilters and _transport.
+    const spanFilters = agent._spanFilters
+    const transport = agent._transport
+
     span._encode(function (err, payload) {
       if (err) {
         agent.logger.error('error encoding span %o', { span: span.id, parent: span.parentId, trace: span.traceId, name: span.name, type: span.type, error: err.message })
         return
       }
 
-      payload = agent._spanFilters.process(payload)
+      payload = spanFilters.process(payload)
 
       if (!payload) {
         agent.logger.debug('span ignored by filter %o', { span: span.id, parent: span.parentId, trace: span.traceId, name: span.name, type: span.type })
         return
       }
 
       agent.logger.debug('sending span %o', { span: span.id, parent: span.parentId, trace: span.traceId, name: span.name, type: span.type })
-      if (agent._transport) agent._transport.sendSpan(payload)
+      if (transport) {
+        transport.sendSpan(payload)
+      }
     })
   } else {
     agent.logger.debug('ignoring span %o', { span: span.id, parent: span.parentId, trace: span.traceId, name: span.name, type: span.type })

test/_mock_apm_server.js

@@ -6,6 +6,8 @@
 //      // Test code using `serverUrl`...
 //      // - Events received on the intake API will be on `server.events`.
 //      // - Raw request data is on `server.requests`.
+//      // - Use `server.clear()` to clear `server.events` and `server.requests`
+//      //   for re-use of the mock server in multiple test cases.
 //      // - Call `server.close()` when done.
 //    })
 
@@ -15,12 +17,16 @@ const zlib = require('zlib')
 
 class MockAPMServer {
   constructor () {
-    this.events = []
-    this.requests = []
+    this.clear()
     this.serverUrl = null // set in .start()
     this._http = http.createServer(this._onRequest.bind(this))
   }
 
+  clear () {
+    this.events = []
+    this.requests = []
+  }
+
   _onRequest (req, res) {
     var parsedUrl = new URL(req.url, this.serverUrl)
     var instream = req