fix: redis run-context handling, error capture, span.outcome

Refs: #2430

CHANGELOG.asciidoc

@@ -40,15 +40,20 @@ Notes:
 [float]
 ===== Bug fixes
 
-- Fix 'redis' instrumentation to avoid a rare possible double-instrumentation
-  of redis commands internally-queued before the RedisClient is "ready".
-  ({pull}2446[#2446])
-
-- Avoid setting the `tracestate` header for outgoing HTTP requests to the empty
+* Fixes for 'redis' instrumentations:
++
+**  Fix run-context so that a span created in the same tick as a redis client
+    command will no longer be a child of the redis span. ({issues}2430[#2430])
+**  Capture an APM error and correctly set span.outcome to "failure" when
+    a redis client command calls back with an error.
+**  Avoid a rare possible double-instrumentation of redis commands
+    internally-queued before the RedisClient is "ready". ({pull}2446[#2446])
+
+* Avoid setting the `tracestate` header for outgoing HTTP requests to the empty
   string. This can happen for non-trace-root transactions. While the HTTP spec
   allows empty header values, some servers do not. ({issues}2405[#2405])
 
-- Deprecate `transaction.subtype` and `transaction.action`. These fields
+* Deprecate `transaction.subtype` and `transaction.action`. These fields
   were never used by APM server. This also deprecates the
   `apm.startTransaction(...)` call signatures that take `subtype` and `action`
   arguments. In the next major version these two fields will be removed.

examples/trace-redis.js

@@ -0,0 +1,40 @@
+#!/usr/bin/env node --unhandled-rejections=strict
+
+// A small example showing Elastic APM tracing the 'redis' package.
+//
+// This assumes a Redis server running on localhost. You can use:
+//    npm run docker:start
+// to start an Redis docker container (and other containers used for
+// testing of this project). Then `npm run docker:stop` to stop them.
+
+const apm = require('../').start({ // elastic-apm-node
+  serviceName: 'example-trace-redis'
+})
+
+const redis = require('redis')
+
+const client = redis.createClient()
+
+// For tracing spans to be created, there must be an active transaction.
+// Typically, a transaction is automatically started for incoming HTTP
+// requests to a Node.js server. However, because this script is not running
+// an HTTP server, we manually start a transaction. More details at:
+// https://www.elastic.co/guide/en/apm/agent/nodejs/current/custom-transactions.html
+const t1 = apm.startTransaction('t1')
+
+client.set('key1', 'val1')
+client.get('key1', function (err, reply) {
+  console.log('GET key1: %s', err ? `${err.name}: ${err.message}` : reply)
+  t1.end()
+  client.quit()
+})
+
+// Simulate a redis client error with `enable_offline_queue: false` and a
+// quick `.set()` before the client connection ready.
+const clientSimErr = redis.createClient({ enable_offline_queue: false })
+const t2 = apm.startTransaction('t2')
+clientSimErr.set('key2', 'val2', function (err, reply) {
+  console.log('SET key2: %s', err ? `${err.name}: ${err.message}` : reply)
+  t2.end()
+  clientSimErr.quit()
+})

lib/instrumentation/modules/redis.js

@@ -2,17 +2,23 @@
 
 var semver = require('semver')
 
+const constants = require('../../constants')
 var shimmer = require('../shimmer')
 var { getDBDestination } = require('../context')
 
 const isWrappedRedisCbSym = Symbol('ElasticAPMIsWrappedRedisCb')
 
+const TYPE = 'cache'
+const SUBTYPE = 'redis'
+
 module.exports = function (redis, agent, { version, enabled }) {
   if (!semver.satisfies(version, '>=2.0.0 <4.0.0')) {
     agent.logger.debug('redis version %s not supported - aborting...', version)
     return redis
   }
 
+  const ins = agent._instrumentation
+
   // The undocumented field on a RedisClient instance on which connection
   // options are stored has changed a few times.
   //
@@ -34,11 +40,15 @@ module.exports = function (redis, agent, { version, enabled }) {
 
   return redis
 
-  function makeWrappedCallback (span, cb) {
-    const wrappedCallback = agent._instrumentation.bindFunction(function () {
-      if (span) span.end()
-      if (cb) {
-        return cb.apply(this, arguments)
+  function makeWrappedCallback (span, origCb) {
+    const wrappedCallback = ins.bindFunction(function (err, _reply) {
+      if (err) {
+        span._setOutcomeFromErrorCapture(constants.OUTCOME_FAILURE)
+        agent.captureError(err, { skipOutcome: true })
+      }
+      span.end()
+      if (origCb) {
+        return origCb.apply(this, arguments)
       }
     })
     wrappedCallback[isWrappedRedisCbSym] = true
@@ -53,22 +63,24 @@ module.exports = function (redis, agent, { version, enabled }) {
       }
 
       if (commandObj.callback && commandObj.callback[isWrappedRedisCbSym]) {
-        // Avoid re-wrapping send_command called *again* for commands queued
-        // before the client was "ready".
+        // Avoid re-wrapping internal_send_command called *again* for commands
+        // queued before the client was "ready".
         return original.apply(this, arguments)
       }
 
       const command = commandObj.command
       agent.logger.debug({ command: command }, 'intercepted call to RedisClient.prototype.internal_send_command')
-      const spanName = command.toUpperCase()
-      var span = enabled && agent.startSpan(spanName, 'cache', 'redis')
-      if (span) {
-        const connOpts = connOptsFromRedisClient(this)
-        span.setDestinationContext(getDBDestination(span, connOpts.host, connOpts.port))
-        commandObj.callback = makeWrappedCallback(span, commandObj.callback)
+      const span = enabled && ins.createSpan(command.toUpperCase(), TYPE, SUBTYPE)
+      if (!span) {
+        return original.apply(this, arguments)
       }
 
-      return original.apply(this, arguments)
+      const connOpts = connOptsFromRedisClient(this)
+      span.setDestinationContext(getDBDestination(span, connOpts.host, connOpts.port))
+
+      commandObj.callback = makeWrappedCallback(span, commandObj.callback)
+      const spanRunContext = ins.currRunContext().enterSpan(span)
+      return ins.withRunContext(spanRunContext, original, this, ...arguments)
     }
   }
 
@@ -89,32 +101,25 @@ module.exports = function (redis, agent, { version, enabled }) {
         return original.apply(this, arguments)
       }
 
-      var span = enabled && agent.startSpan(null, 'cache', 'redis')
-      var id = span && span.transaction.id
-
-      agent.logger.debug('intercepted call to RedisClient.prototype.send_command %o', { id: id, command: command })
-
-      if (span) {
-        let host, port
-        const connOpts = connOptsFromRedisClient(this)
-        if (connOpts) {
-          host = connOpts.host
-          port = connOpts.port
-        }
-        span.setDestinationContext(getDBDestination(span, host, port))
-        span.name = String(command).toUpperCase()
-
-        const wrappedCb = makeWrappedCallback(span, origCb)
-        if (cb) {
-          cb = wrappedCb
-        } else if (origCb) {
-          args[args.length - 1] = wrappedCb
-        } else {
-          cb = wrappedCb
-        }
+      agent.logger.debug({ command: command }, 'intercepted call to RedisClient.prototype.send_command')
+      var span = enabled && ins.createSpan(command.toUpperCase(), TYPE, SUBTYPE)
+      if (!span) {
+        return original.apply(this, arguments)
       }
 
-      return original.call(this, command, args, cb)
+      const connOpts = connOptsFromRedisClient(this)
+      span.setDestinationContext(getDBDestination(span, connOpts.host, connOpts.port))
+
+      const wrappedCb = makeWrappedCallback(span, origCb)
+      if (cb) {
+        cb = wrappedCb
+      } else if (origCb) {
+        args[args.length - 1] = wrappedCb
+      } else {
+        cb = wrappedCb
+      }
+      const spanRunContext = ins.currRunContext().enterSpan(span)
+      return ins.withRunContext(spanRunContext, original, this, command, args, cb)
     }
   }
 }

test/instrumentation/modules/redis.test.js

@@ -59,7 +59,7 @@ test('redis', function (t) {
         address: process.env.REDIS_HOST || '127.0.0.1',
         port: 6379
       }, 'span.context.destination')
-      // t.strictEqual(span.parent_id, trans.id, 'span is a child of the transaction')
+      t.strictEqual(span.parent_id, trans.id, 'span is a child of the transaction')
 
       var offset = span.timestamp - trans.timestamp
       t.ok(offset + span.duration * 1000 < trans.duration * 1000,
@@ -125,6 +125,42 @@ test('redis', function (t) {
   })
 })
 
+// Skip testing error capture with redis 2.x. It works, but there are behaviour
+// differences (e.g. `client.quit()` throws with `enable_offline_queue: false`)
+// such that testing is a pain. Redis 2.x is too old to bother.
+if (semver.satisfies(redisVersion, '>=3.0.0')) {
+  test('redis client error', function (t) {
+    resetAgent(function (data) {
+      t.equal(data.transactions.length, 1, 'got 1 transaction')
+      t.equal(data.spans.length, 1, 'got 1 span')
+      t.equal(data.errors.length, 1, 'got 1 error')
+      t.equal(data.spans[0].name, 'SET', 'span.name')
+      t.equal(data.spans[0].parent_id, data.transactions[0].id, 'span.parent_id')
+      t.equal(data.spans[0].outcome, 'failure', 'span.outcome')
+      t.equal(data.errors[0].transaction_id, data.transactions[0].id, 'error.transaction_id')
+      t.equal(data.errors[0].exception.type, 'AbortError', 'error.exception.type')
+      t.end()
+    })
+
+    // Simulate a redis client error with `enable_offline_queue: false` and a
+    // quick `.set()` before the client connection ready.
+    var client = redis.createClient({
+      host: process.env.REDIS_HOST,
+      port: '6379',
+      enable_offline_queue: false
+    })
+    var t0 = agent.startTransaction('t0')
+    client.set('k', 'v', function (err, reply) {
+      t.ok(err, 'got error from client.set')
+      t.equal(err.name, 'AbortError', 'error.name')
+      t.ok(reply === undefined, 'no reply')
+      t0.end()
+      client.quit()
+      agent.flush()
+    })
+  })
+}
+
 test('client.cmd(...) call signatures', function (t) {
   let nCbCalled = 0
   function myCb () {
@@ -193,5 +229,4 @@ if (semver.satisfies(redisVersion, '<=2.4.2')) {
 function resetAgent (cb) {
   agent._instrumentation.testReset()
   agent._transport = mockClient(cb)
-  agent.captureError = function (err) { throw err }
 }