lib/instrumentation/modules/redis.js

/*
 * Copyright Elasticsearch B.V. and other contributors where applicable.
 * Licensed under the BSD 2-Clause License; you may not use this file except in
 * compliance with the BSD 2-Clause License.
 */

'use strict';

var semver = require('semver');

const constants = require('../../constants');
var shimmer = require('../shimmer');
var { getDBDestination } = require('../context');

const isWrappedRedisCbSym = Symbol('ElasticAPMIsWrappedRedisCb');

const TYPE = 'db';
const SUBTYPE = 'redis';
const ACTION = 'query';

module.exports = function (redis, agent, { version, enabled }) {
  if (!enabled) {
    return redis;
  }
  if (!semver.satisfies(version, '>=2.0.0 <4.0.0')) {
    // Explicitly do not log.debug here, because the message is misleading for
    // redis@4 and later that is being handled by @redis/client instrumentation.
    return redis;
  }

  const ins = agent._instrumentation;

  // The undocumented field on a RedisClient instance on which connection
  // options are stored has changed a few times.
  //
  // - >=2.4.0: `client.connection_options.{host,port}`, commit eae5596a
  // - >=2.3.0, <2.4.0: `client.connection_option.{host,port}`, commit d454e402
  // - >=0.12.0, <2.3.0: `client.connectionOption.{host,port}`, commit 064260d1
  // - <0.12.0: *maybe* `client.{host,port}`
  const connOptsFromRedisClient = (rc) =>
    rc.connection_options || rc.connection_option || rc.connectionOption || {};

  var proto = redis.RedisClient && redis.RedisClient.prototype;
  if (semver.satisfies(version, '>2.5.3')) {
    agent.logger.debug(
      'shimming redis.RedisClient.prototype.internal_send_command',
    );
    shimmer.wrap(proto, 'internal_send_command', wrapInternalSendCommand);
  } else {
    agent.logger.debug('shimming redis.RedisClient.prototype.send_command');
    shimmer.wrap(proto, 'send_command', wrapSendCommand);
  }

  return redis;

  function makeWrappedCallback(spanRunContext, span, origCb) {
    const wrappedCallback = ins.bindFunctionToRunContext(
      spanRunContext,
      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;
    return wrappedCallback;
  }

  function wrapInternalSendCommand(original) {
    return function wrappedInternalSendCommand(commandObj) {
      if (!commandObj || typeof commandObj.command !== 'string') {
        // Unexpected usage. Skip instrumenting this call.
        return original.apply(this, arguments);
      }

      if (commandObj.callback && commandObj.callback[isWrappedRedisCbSym]) {
        // 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 },
        'intercepted call to RedisClient.prototype.internal_send_command',
      );
      const span = ins.createSpan(
        command.toUpperCase(),
        TYPE,
        SUBTYPE,
        ACTION,
        { exitSpan: true },
      );
      if (!span) {
        return original.apply(this, arguments);
      }

      const connOpts = connOptsFromRedisClient(this);
      span._setDestinationContext(
        getDBDestination(connOpts.host, connOpts.port),
      );
      span.setDbContext({ type: 'redis' });

      const spanRunContext = ins.currRunContext().enterSpan(span);
      commandObj.callback = makeWrappedCallback(
        spanRunContext,
        span,
        commandObj.callback,
      );
      return ins.withRunContext(spanRunContext, original, this, ...arguments);
    };
  }

  function wrapSendCommand(original) {
    return function wrappedSendCommand(command, args, cb) {
      if (typeof command !== 'string') {
        // Unexpected usage. Skip instrumenting this call.
        return original.apply(this, arguments);
      }

      let origCb = cb;
      if (
        !origCb &&
        Array.isArray(args) &&
        typeof args[args.length - 1] === 'function'
      ) {
        origCb = args[args.length - 1];
      }
      if (origCb && origCb[isWrappedRedisCbSym]) {
        // Avoid re-wrapping send_command called *again* for commands queued
        // before the client was "ready".
        return original.apply(this, arguments);
      }

      agent.logger.debug(
        { command },
        'intercepted call to RedisClient.prototype.send_command',
      );
      var span = ins.createSpan(command.toUpperCase(), TYPE, SUBTYPE, ACTION, {
        exitSpan: true,
      });
      if (!span) {
        return original.apply(this, arguments);
      }

      const connOpts = connOptsFromRedisClient(this);
      span._setDestinationContext(
        getDBDestination(connOpts.host, connOpts.port),
      );
      span.setDbContext({ type: 'redis' });

      const spanRunContext = ins.currRunContext().enterSpan(span);
      const wrappedCb = makeWrappedCallback(spanRunContext, span, origCb);
      if (cb) {
        cb = wrappedCb;
      } else if (origCb) {
        args[args.length - 1] = wrappedCb;
      } else {
        cb = wrappedCb;
      }
      return ins.withRunContext(
        spanRunContext,
        original,
        this,
        command,
        args,
        cb,
      );
    };
  }
};