feat: improved filtering function API (elastic#579)

Changes the `agent.addFilter()` function to be called once per event
instead of with an array of events.

Besides the existing filtering function, three new custom filtering
functions have been added:

- `agent.addErrorFilter()`
- `agent.addTransactionFilter()`
- `agent.addSpanFilter()`

BREAKING CHANGE: The `agent.addFilter()` API have changed.

docs/agent-api.asciidoc

@@ -613,7 +613,7 @@ Currently the following HTTP headers are anonymized by default:
   See the https://github.com/watson/is-secret[is-secret] module for details about which patterns are considered sensitive.
 
 If you wish to filter or santitize other data,
-use the <<apm-add-filter,`apm.addFilter()`>> function.
+use one of the <<apm-add-filter,filtering>> functions.
 
 [[disable-instrumentations]]
 ===== `disableInstrumentations`
@@ -676,7 +676,8 @@ Each filter function will be called in the order they were added,
 and will receive a `payload` object as the only argument,
 containing the data about to be sent to the APM Server.
 
-For details on the format of the payload,
+The format of the payload depends on the event type being sent.
+For details about the different formats,
 see the {apm-server-ref}/intake-api.html[APM Server intake API documentation].
 
 The filter function is synchronous and should return the manipulated payload object.
@@ -688,17 +689,10 @@ Example usage:
 [source,js]
 ----
 apm.addFilter(function (payload) {
-  // the payload can either contain an array of transactions or errors
-  var items = payload.transactions || payload.errors || []
-
-  // loop over each item in the array to redact any secrets we don't
-  // want sent to the APM Server
-  items.forEach(function (item) {
-    if (item.context.request && item.context.request.headers) {
-      // redact sensitive data
-      payload.context.request.headers['x-secret'] = '[REDACTED]'
-    }
-  })
+  if (payload.context.request && payload.context.request.headers) {
+    // redact sensitive data
+    payload.context.request.headers['x-secret'] = '[REDACTED]'
+  }
 
   // remember to return the modified payload
   return payload
@@ -711,6 +705,39 @@ See <<filter-http-headers,`filterHttpHeaders`>> for details.
 Though you can also use filter functions to add new contextual information to the `user` and `custom` properties,
 it's recommended that you use <<apm-set-user-context,`apm.setUserContext()`>> and <<apm-set-custom-context,`apm.setCustomContext()`>> for that purpose.
 
+[[apm-add-error-filter]]
+==== `apm.addErrorFilter()`
+
+[source,js]
+----
+apm.addErrorFilter(callback)
+----
+
+Similar to <<apm-add-filter,`apm.addFilter()`>>,
+but the `callback` will only be called with error payloads.
+
+[[apm-add-transaction-filter]]
+==== `apm.addTransactionFilter()`
+
+[source,js]
+----
+apm.addTransactionFilter(callback)
+----
+
+Similar to <<apm-add-filter,`apm.addFilter()`>>,
+but the `callback` will only be called with transaction payloads.
+
+[[apm-add-span-filter]]
+==== `apm.addSpanFilter()`
+
+[source,js]
+----
+apm.addSpanFilter(callback)
+----
+
+Similar to <<apm-add-filter,`apm.addFilter()`>>,
+but the `callback` will only be called with span payloads.
+
 [[apm-set-user-context]]
 ==== `apm.setUserContext()`
 

docs/custom-stack.asciidoc

@@ -252,7 +252,7 @@ use the <<capture-body,`captureBody`>> config option
 To disable this,
 use the <<filter-http-headers,`filterHttpHeaders`>> config option
 * To apply custom filters,
-use the <<apm-add-filter,`apm.addFilter()`>> function
+use one of the <<apm-add-filter,filtering>> functions
 
 [float]
 [[custom-stack-add-your-own-data]]

docs/express.asciidoc

@@ -164,7 +164,7 @@ use the <<capture-body,`captureBody`>> config option
 To disable this,
 use the <<filter-http-headers,`filterHttpHeaders`>> config option
 * To apply custom filters,
-use the <<apm-add-filter,`apm.addFilter()`>> function
+use one of the <<apm-add-filter,filtering>> functions
 
 [float]
 [[express-add-your-own-data]]

docs/hapi.asciidoc

@@ -175,7 +175,7 @@ use the <<capture-body,`captureBody`>> config option
 To disable this,
 use the <<filter-http-headers,`filterHttpHeaders`>> config option
 * To apply custom filters,
-use the <<apm-add-filter,`apm.addFilter()`>> function
+use one of the <<apm-add-filter,filtering>> functions
 
 [float]
 [[hapi-add-your-own-data]]

docs/koa.asciidoc

@@ -177,7 +177,7 @@ use the <<capture-body,`captureBody`>> config option
 To disable this,
 use the <<filter-http-headers,`filterHttpHeaders`>> config option
 * To apply custom filters,
-use the <<apm-add-filter,`apm.addFilter()`>> function
+use one of the <<apm-add-filter,filtering>> functions
 
 [float]
 [[koa-add-your-own-data]]

docs/lambda.asciidoc

@@ -113,7 +113,7 @@ use the <<capture-body,`logBody`>> config option
 To disable this,
 use the <<filter-http-headers,`filterHttpHeaders`>> config option
 * To apply custom filters,
-use the <<apm-add-filter,`apm.addFilter()`>> function
+use one of the <<apm-add-filter,filtering>> functions
 
 [float]
 [[lambda-add-your-own-data]]

docs/restify.asciidoc

@@ -167,7 +167,7 @@ use the <<capture-body,`captureBody`>> config option
 To disable this,
 use the <<filter-http-headers,`filterHttpHeaders`>> config option
 * To apply custom filters,
-use the <<apm-add-filter,`apm.addFilter()`>> function
+use one of the <<apm-add-filter,filtering>> functions
 
 [float]
 [[restify-add-your-own-data]]

lib/agent.js

@@ -30,7 +30,9 @@ function Agent () {
   this.middleware = { connect: connect.bind(this) }
 
   this._instrumentation = new Instrumentation(this)
-  this._filters = new Filters()
+  this._errorFilters = new Filters()
+  this._transactionFilters = new Filters()
+  this._spanFilters = new Filters()
   this._apmServer = null
 
   this._conf = null
@@ -95,7 +97,10 @@ Agent.prototype.start = function (opts) {
   global[symbols.agentInitialized] = true
 
   this._config(opts)
-  this._filters.config(this._conf)
+
+  if (this._conf.filterHttpHeaders) {
+    this.addFilter(require('./filters/http-headers'))
+  }
 
   if (!this._conf.active) {
     this.logger.info('Elastic APM agent is inactive due to configuration')
@@ -200,12 +205,36 @@ Agent.prototype.addTags = function (tags) {
 }
 
 Agent.prototype.addFilter = function (fn) {
+  this.addErrorFilter(fn)
+  this.addTransactionFilter(fn)
+  this.addSpanFilter(fn)
+}
+
+Agent.prototype.addErrorFilter = function (fn) {
+  if (typeof fn !== 'function') {
+    this.logger.error('Can\'t add filter of type %s', typeof fn)
+    return
+  }
+
+  this._errorFilters.add(fn)
+}
+
+Agent.prototype.addTransactionFilter = function (fn) {
+  if (typeof fn !== 'function') {
+    this.logger.error('Can\'t add filter of type %s', typeof fn)
+    return
+  }
+
+  this._transactionFilters.add(fn)
+}
+
+Agent.prototype.addSpanFilter = function (fn) {
   if (typeof fn !== 'function') {
     this.logger.error('Can\'t add filter of type %s', typeof fn)
     return
   }
 
-  this._filters.add(fn)
+  this._spanFilters.add(fn)
 }
 
 Agent.prototype.captureError = function (err, opts, cb) {
@@ -313,7 +342,7 @@ Agent.prototype.captureError = function (err, opts, cb) {
   }
 
   function send (error) {
-    error = agent._filters.process(error) // TODO: Update filter to expect this format
+    error = agent._errorFilters.process(error)
 
     if (!error) {
       agent.logger.debug('error ignored by filter %o', { id: id })

lib/filters/http-headers.js

@@ -0,0 +1,44 @@
+'use strict'
+
+const REDACTED = require('./').REDACTED
+
+const cookie = require('cookie')
+const redact = require('redact-secrets')(REDACTED)
+const SetCookie = require('set-cookie-serde')
+
+module.exports = httpHeaders
+
+function httpHeaders (obj) {
+  const headers = obj.context && obj.context.request && obj.context.request.headers
+
+  if (!headers) return obj
+
+  if (headers.authorization) headers.authorization = REDACTED
+
+  if (typeof headers.cookie === 'string') {
+    var cookies = cookie.parse(headers.cookie)
+    redact.forEach(cookies)
+    headers.cookie = Object.keys(cookies)
+      .map(function (k) { return k + '=' + cookies[k] })
+      .join('; ')
+  }
+
+  if (typeof headers['set-cookie'] !== 'undefined') {
+    try {
+      var setCookies = new SetCookie(headers['set-cookie'])
+      redact.forEach(setCookies)
+      headers['set-cookie'] = stringify(setCookies)
+    } catch (err) {
+      // Ignore error
+      headers['set-cookie'] = '[malformed set-cookie header]'
+    }
+  }
+
+  return obj
+}
+
+function stringify (value) {
+  return Array.isArray(value)
+    ? value.map(value => value.toString())
+    : value.toString()
+}

lib/filters/index.js

@@ -0,0 +1,26 @@
+'use strict'
+
+module.exports = Filters
+
+function Filters () {
+  if (!(this instanceof Filters)) return new Filters()
+  this._filters = []
+}
+
+Filters.REDACTED = '[REDACTED]'
+
+Filters.prototype.add = function (fn) {
+  this._filters.push(fn)
+}
+
+Filters.prototype.process = function (payload) {
+  var result = payload
+
+  // abort if a filter function doesn't return an object
+  this._filters.some(function (filter) {
+    result = filter(result)
+    return !result
+  })
+
+  return result
+}

lib/instrumentation/index.js

@@ -93,7 +93,7 @@ Instrumentation.prototype.addEndedTransaction = function (transaction) {
   var agent = this._agent
 
   if (this._started) {
-    var payload = agent._filters.process(transaction._encode()) // TODO: Update filter to expect this format
+    var payload = agent._transactionFilters.process(transaction._encode())
     if (!payload) return agent.logger.debug('transaction ignored by filter %o', { id: transaction.id })
     agent.logger.debug('sending transaction %o', { id: transaction.id })
     agent._apmServer.sendTransaction(payload)
@@ -113,7 +113,7 @@ Instrumentation.prototype.addEndedSpan = function (span) {
         return
       }
 
-      payload = agent._filters.process(payload) // TODO: Update filter to expect this format
+      payload = agent._spanFilters.process(payload)
 
       if (!payload) {
         agent.logger.debug('span ignored by filter %o', { trans: span.transaction.id, name: span.name, type: span.type })

test/_agent.js

@@ -21,7 +21,9 @@ function clean () {
   global[symbols.agentInitialized] = null
   process._events.uncaughtException = uncaughtExceptionListeners
   if (agent) {
-    agent._filters = new Filters()
+    agent._errorFilters = new Filters()
+    agent._transactionFilters = new Filters()
+    agent._spanFilters = new Filters()
     if (agent._instrumentation && agent._instrumentation._hook) {
       agent._instrumentation._hook.unhook()
     }