feat(plugin): implement redis plugin

.circleci/config.yml

@@ -1,17 +1,22 @@
 version: 2
 
-test_env: &test_env
+node_test_env: &node_test_env
   RUN_POSTGRES_TESTS: 1
   POSTGRES_USER: postgres
   POSTGRES_DB: circle_database
   POSTGRES_HOST: localhost
   POSTGRES_PORT: 5432
+  RUN_REDIS_TESTS: 1
+  OPENTELEMETRY_REDIS_HOST: 'localhost'
+  OPENTELEMETRY_REDIS_PORT: 6379
 
 postgres_service: &postgres_service
   image: circleci/postgres:9.6-alpine
-  environment: # env to pass to CircleCI, specified values must match test_env
+  environment: # env to pass to CircleCI, specified values must match node_test_env
     POSTGRES_USER: postgres
     POSTGRES_DB: circle_database
+redis_service: &redis_service
+  image: redis
 
 node_unit_tests: &node_unit_tests
   steps:
@@ -99,20 +104,23 @@ jobs:
   node8:
     docker:
       - image: node:8
-        environment: *test_env
+        environment: *node_test_env
       - *postgres_service
+      - *redis_service
     <<: *node_unit_tests
   node10:
     docker:
       - image: node:10
-        environment: *test_env
+        environment: *node_test_env
       - *postgres_service
+      - *redis_service
     <<: *node_unit_tests
   node12:
     docker:
       - image: node:12
-        environment: *test_env
+        environment: *node_test_env
       - *postgres_service
+      - *redis_service
     <<: *node_unit_tests
   node12-browsers:
     docker:

packages/opentelemetry-plugin-redis/package.json

@@ -8,11 +8,14 @@
   "repository": "open-telemetry/opentelemetry-js",
   "scripts": {
     "test": "nyc ts-mocha -p tsconfig.json 'test/**/*.ts'",
+    "test:debug": "cross-env RUN_REDIS_TESTS_LOCAL=true ts-mocha --inspect-brk --no-timeouts -p tsconfig.json 'test/**/*.test.ts'",
+    "test:local": "cross-env RUN_REDIS_TESTS_LOCAL=true yarn test",
     "tdd": "yarn test -- --watch-extensions ts --watch",
     "clean": "rimraf build/*",
     "check": "gts check",
     "precompile": "tsc --version",
     "compile": "tsc -p .",
+    "codecov": "nyc report --reporter=json && codecov -f coverage/*.json -p ../../",
     "fix": "gts fix",
     "prepare": "npm run compile"
   },
@@ -42,10 +45,16 @@
   "devDependencies": {
     "@types/mocha": "^5.2.7",
     "@types/node": "^12.6.9",
+    "@types/redis": "^2.8.14",
+    "@types/shimmer": "^1.0.1",
+    "@opentelemetry/node": "^0.2.0",
+    "@opentelemetry/tracing": "^0.2.0",
     "codecov": "^3.5.0",
+    "cross-env": "^6.0.3",
     "gts": "^1.1.0",
     "mocha": "^6.2.0",
     "nyc": "^14.1.1",
+    "redis": "^2.8.0",
     "rimraf": "^3.0.0",
     "ts-mocha": "^6.0.0",
     "ts-node": "^8.3.0",
@@ -55,7 +64,7 @@
   },
   "dependencies": {
     "@opentelemetry/core": "^0.2.0",
-    "@opentelemetry/node": "^0.2.0",
-    "@opentelemetry/types": "^0.2.0"
+    "@opentelemetry/types": "^0.2.0",
+    "shimmer": "^1.2.1"
   }
 }

packages/opentelemetry-plugin-redis/src/enums.ts

@@ -0,0 +1,32 @@
+/*!
+ * Copyright 2019, OpenTelemetry Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+export enum AttributeNames {
+  // required by https://github.com/open-telemetry/opentelemetry-specification/blob/master/specification/data-semantic-conventions.md#databases-client-calls
+  COMPONENT = 'component',
+  DB_TYPE = 'db.type',
+  DB_INSTANCE = 'db.instance',
+  DB_STATEMENT = 'db.statement',
+  PEER_ADDRESS = 'peer.address',
+  PEER_HOSTNAME = 'peer.host',
+
+  // optional
+  DB_USER = 'db.user',
+  PEER_PORT = 'peer.port',
+  PEER_IPV4 = 'peer.ipv4',
+  PEER_IPV6 = 'peer.ipv6',
+  PEER_SERVICE = 'peer.service',
+}

packages/opentelemetry-plugin-redis/src/index.ts

@@ -13,3 +13,5 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
+export * from './redis';

packages/opentelemetry-plugin-redis/src/redis.ts

@@ -0,0 +1,189 @@
+/*!
+ * Copyright 2019, OpenTelemetry Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { BasePlugin } from '@opentelemetry/core';
+import * as redisTypes from 'redis';
+import * as shimmer from 'shimmer';
+import { SpanKind, CanonicalCode } from '@opentelemetry/types';
+import { AttributeNames } from './enums';
+import {
+  RedisPluginStreamTypes,
+  RedisPluginClientTypes,
+  RedisCommand,
+} from './types';
+import { EventEmitter } from 'events';
+
+export class RedisPlugin extends BasePlugin<typeof redisTypes> {
+  static readonly COMPONENT = 'redis';
+  readonly supportedVersions = ['^2.6.0']; // should be >= 2.6.0
+
+  constructor(readonly moduleName: string) {
+    super();
+  }
+
+  protected patch() {
+    if (this._moduleExports.RedisClient) {
+      this._logger.debug(
+        'Patching redis.RedisClient.prototype.internal_send_command'
+      );
+      shimmer.wrap(
+        this._moduleExports.RedisClient.prototype,
+        'internal_send_command',
+        this._getPatchInternalSendCommand()
+      );
+
+      this._logger.debug('patching redis.create_stream');
+      shimmer.wrap(
+        this._moduleExports.RedisClient.prototype,
+        'create_stream',
+        this._getPatchCreateStream()
+      );
+
+      this._logger.debug('patching redis.createClient');
+      shimmer.wrap(
+        this._moduleExports,
+        'createClient',
+        this._getPatchCreateClient()
+      );
+    }
+    return this._moduleExports;
+  }
+
+  protected unpatch(): void {
+    if (this._moduleExports) {
+      shimmer.unwrap(
+        this._moduleExports.RedisClient.prototype,
+        'internal_send_command'
+      );
+      shimmer.unwrap(
+        this._moduleExports.RedisClient.prototype,
+        'create_stream'
+      );
+      shimmer.unwrap(this._moduleExports, 'createClient');
+    }
+  }
+
+  /**
+   * Patch internal_send_command(...) to trace requests
+   */
+  private _getPatchInternalSendCommand() {
+    const plugin = this;
+    return function internal_send_command(original: Function) {
+      return function internal_send_command_trace(
+        this: redisTypes.RedisClient & RedisPluginClientTypes,
+        cmd?: RedisCommand
+      ) {
+        const parentSpan = plugin._tracer.getCurrentSpan();
+
+        // New versions of redis (2.4+) use a single options object
+        // instead of named arguments
+        if (arguments.length === 1 && typeof cmd === 'object') {
+          const span = plugin._tracer.startSpan(`redis-${cmd.command}`, {
+            kind: SpanKind.CLIENT,
+            parent: parentSpan || undefined,
+            attributes: {
+              [AttributeNames.COMPONENT]: RedisPlugin.COMPONENT,
+              [AttributeNames.DB_STATEMENT]: cmd.command,
+            },
+          });
+
+          // Set attributes for not explicitly typed RedisPluginClientTypes
+          if (this.options) {
+            span.setAttributes({
+              [AttributeNames.PEER_HOSTNAME]: this.options.host,
+              [AttributeNames.PEER_PORT]: this.options.port,
+            });
+          }
+          if (this.address) {
+            span.setAttribute(
+              AttributeNames.PEER_ADDRESS,
+              `redis://${this.address}`
+            );
+          }
+
+          let spanEnded = false;
+          const endSpan = (err?: Error | null) => {
+            if (spanEnded) return; // never
+            if (err) {
+              span.setStatus({
+                code: CanonicalCode.UNKNOWN,
+                message: err.message,
+              });
+            } else {
+              span.setStatus({ code: CanonicalCode.OK });
+            }
+            span.end();
+            spanEnded = true;
+          };
+
+          const originalCallback = arguments[0].callback;
+          if (originalCallback) {
+            (arguments[0] as RedisCommand).callback = function callback<T>(
+              this: unknown,
+              err: Error | null,
+              _reply: T
+            ) {
+              endSpan(err);
+              return originalCallback.apply(this, arguments);
+            };
+          }
+          try {
+            // Span will be ended in callback
+            return original.apply(this, arguments);
+          } catch (rethrow) {
+            endSpan(rethrow);
+            throw rethrow; // rethrow after ending span
+          }
+        }
+
+        // We don't know how to trace this call, so don't start/stop a span
+        return original.apply(this, arguments);
+      };
+    };
+  }
+
+  private _getPatchCreateClient() {
+    const plugin = this;
+    return function createClient(original: Function) {
+      return function createClientTrace(this: redisTypes.RedisClient) {
+        const client: redisTypes.RedisClient = original.apply(this, arguments);
+        return plugin._tracer.bind(client);
+      };
+    };
+  }
+
+  private _getPatchCreateStream() {
+    const plugin = this;
+    return function createReadStream(original: Function) {
+      return function create_stream_trace(this: RedisPluginStreamTypes) {
+        if (!this.stream) {
+          Object.defineProperty(this, 'stream', {
+            get() {
+              return this._patched_redis_stream;
+            },
+            set(val: EventEmitter) {
+              plugin._tracer.bind(val);
+              this._patched_redis_stream = val;
+            },
+          });
+        }
+        return original.apply(this, arguments);
+      };
+    };
+  }
+}
+
+export const plugin = new RedisPlugin(RedisPlugin.COMPONENT);

packages/opentelemetry-plugin-redis/src/types.ts

@@ -0,0 +1,41 @@
+/*!
+ * Copyright 2019, OpenTelemetry Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import * as redisTypes from 'redis';
+import { EventEmitter } from 'events';
+
+// exported from
+// https://github.com/NodeRedis/node_redis/blob/master/lib/command.js
+export interface RedisCommand {
+  command: string;
+  args: string[];
+  buffer_args: boolean;
+  callback: redisTypes.Callback<unknown>;
+  call_on_write: boolean;
+}
+
+export interface RedisPluginClientTypes {
+  options?: {
+    host: string;
+    port: string;
+  };
+
+  address?: string;
+}
+
+export interface RedisPluginStreamTypes {
+  stream?: { get(): EventEmitter; set(val: EventEmitter): void };
+}