Adding diagnostic-channel (#235)

* Adding diagnostic-channel

By using diagnostic-channel and diagnostic-channel-publishers, we can support context tracking through third-party libraries as well as getting additional telemetry for those dependencies.

* Fixing cyclical reference and supporting multiple AI clients in diagnostic-channel subscribers

* Updating readme with diagnostic-channel information

AutoCollection/ClientRequests.ts

@@ -9,6 +9,12 @@ import RequestResponseHeaders = require("../Library/RequestResponseHeaders");
 import ClientRequestParser = require("./ClientRequestParser");
 import { CorrelationContextManager, CorrelationContext } from "./CorrelationContextManager";
 
+import {enable as enableMongodb} from "./diagnostic-channel/mongodb.sub";
+import {enable as enableMysql} from "./diagnostic-channel/mysql.sub";
+import {enable as enableRedis} from "./diagnostic-channel/redis.sub";
+
+import "./diagnostic-channel/initialization";
+
 class AutoCollectClientRequests {
     public static disableCollectionRequestOption = 'disableAppInsightsAutoCollection';
 
@@ -32,6 +38,9 @@ class AutoCollectClientRequests {
         if (this._isEnabled && !this._isInitialized) {
             this._initialize();
         }
+        enableMongodb(isEnabled, this._client);
+        enableMysql(isEnabled, this._client);
+        enableRedis(isEnabled, this._client);
     }
 
     public isInitialized() {

AutoCollection/Console.ts

@@ -1,6 +1,11 @@
 import Client = require("../Library/Client");
 import Logging = require("../Library/Logging");
 
+import {enable as enableConsole} from "./diagnostic-channel/console.sub";
+import {enable as enableBunyan} from "./diagnostic-channel/bunyan.sub";
+
+import "./diagnostic-channel/initialization";
+
 class AutoCollectConsole {
     public static originalMethods: {[name: string]: (message?: any, ...optionalParams: any[]) => void};
 
@@ -20,7 +25,8 @@ class AutoCollectConsole {
     }
 
     public enable(isEnabled: boolean) {
-        // todo: investigate feasibility/utility of this; does it make sense to have a logging adapter in node?
+        enableConsole(isEnabled, this._client);
+        enableBunyan(isEnabled, this._client);
     }
 
     public isInitialized() {

AutoCollection/CorrelationContextManager.ts

@@ -3,6 +3,8 @@
 import http = require("http");
 import Util = require("../Library/Util");
 
+import {channel} from "diagnostic-channel";
+
 export interface CorrelationContext {
     operation: {
         name: string;
@@ -100,9 +102,11 @@ export class CorrelationContextManager {
         // Run patches for Zone.js
         if (!this.hasEverEnabled) {
             this.hasEverEnabled = true;
+            channel.addContextPreservation((cb) => {
+                return Zone.current.wrap(cb, "AI-ContextPreservation");
+            })
             this.patchError();
             this.patchTimers(["setTimeout", "setInterval"]);
-            this.patchRedis();
         }
 
         this.enabled = true;
@@ -136,33 +140,6 @@ export class CorrelationContextManager {
         return req;
     }
 
-    // A good example of patching a third party library to respect context.
-    // send_command is always used in this library to send data out.
-    // By overwriting the function to capture the callback provided to it,
-    // and wrapping that callback, we ensure that consumers of this library
-    // will have context persisted.
-    private static patchRedis() {
-        var redis = this.requireForPatch("redis");
-
-        if (redis && redis.RedisClient) {
-            var orig = redis.RedisClient.prototype.send_command;
-            redis.RedisClient.prototype.send_command = function() {
-                var args = Array.prototype.slice.call(arguments);
-                var lastArg = args[args.length - 1];
-
-                if (typeof lastArg === "function") {
-                    args[args.length - 1] = Zone.current.wrap(lastArg, "AI.CCM.patchRedis");
-                } else if (lastArg instanceof Array && typeof lastArg[lastArg.length - 1] === "function") {
-                    // The last argument can be an array!
-                    var lastIndexLastArg = lastArg[lastArg.length - 1];
-                    lastArg[lastArg.length - 1] = Zone.current.wrap(lastIndexLastArg, "AI.CCM.patchRedis");
-                }
-
-                return orig.apply(this, args);
-            };
-        }
-    }
-
     // Zone.js breaks concatenation of timer return values.
     // This fixes that.
     private static patchTimers(methodNames: string[]) {

AutoCollection/diagnostic-channel/bunyan.sub.ts

@@ -0,0 +1,40 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for details.
+import Client = require("../../Library/Client");
+import {SeverityLevel} from "../../Declarations/Contracts";
+
+import {channel, IStandardEvent} from "diagnostic-channel";
+
+import {bunyan} from "diagnostic-channel-publishers";
+
+let clients: Client[] = [];
+
+// Mapping from bunyan levels defined at /~https://github.com/trentm/node-bunyan/blob/master/lib/bunyan.js#L256
+const bunyanToAILevelMap = {};
+bunyanToAILevelMap[10] = SeverityLevel.Verbose;
+bunyanToAILevelMap[20] = SeverityLevel.Verbose;
+bunyanToAILevelMap[30] = SeverityLevel.Information;
+bunyanToAILevelMap[40] = SeverityLevel.Warning;
+bunyanToAILevelMap[50] = SeverityLevel.Error;
+bunyanToAILevelMap[60] = SeverityLevel.Critical;
+
+const subscriber = (event: IStandardEvent<bunyan.IBunyanData>) => {
+    clients.forEach((client) => {
+        const AIlevel = bunyanToAILevelMap[event.data.level];
+        client.trackTrace(event.data.result, AIlevel);
+    });
+};
+
+export function enable(enabled: boolean, client: Client) {
+    if (enabled) {
+        if (clients.length === 0) {
+            channel.subscribe<bunyan.IBunyanData>("bunyan", subscriber);
+        };
+        clients.push(client);
+    } else {
+        clients = clients.filter((c) => c != client);
+        if (clients.length === 0) {
+            channel.unsubscribe("bunyan", subscriber);
+        }
+    }
+}

AutoCollection/diagnostic-channel/console.sub.ts

@@ -0,0 +1,30 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for details.
+import Client = require("../../Library/Client");
+import {SeverityLevel} from "../../Declarations/Contracts";
+
+import {channel, IStandardEvent} from "diagnostic-channel";
+
+import {console as consolePub} from "diagnostic-channel-publishers";
+
+let clients: Client[] = [];
+
+const subscriber = (event: IStandardEvent<consolePub.IConsoleData>) => {
+    clients.forEach((client) => {
+        client.trackTrace(event.data.message, event.data.stderr ? SeverityLevel.Warning : SeverityLevel.Information);
+    });
+};
+
+export function enable(enabled: boolean, client: Client) {
+    if (enabled) {
+        if (clients.length === 0) {
+            channel.subscribe<consolePub.IConsoleData>("console", subscriber);
+        };
+        clients.push(client);
+    } else {
+        clients = clients.filter((c) => c != client);
+        if (clients.length === 0) {
+            channel.unsubscribe("console", subscriber);
+        }
+    }
+}

AutoCollection/diagnostic-channel/initialization.ts

@@ -0,0 +1,8 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for details.
+
+import {enable} from "diagnostic-channel-publishers";
+
+if (!process.env["APPLICATION_INSIGHTS_NO_DIAGNOSTIC_CHANNEL"]) {
+    enable();
+}

AutoCollection/diagnostic-channel/mongodb.sub.ts

@@ -0,0 +1,38 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for details.
+import Client = require("../../Library/Client");
+import {channel, IStandardEvent} from "diagnostic-channel";
+
+import {mongodb} from "diagnostic-channel-publishers";
+
+let clients: Client[] = [];
+
+export const subscriber = (event: IStandardEvent<mongodb.IMongoData>) => {
+    clients.forEach((client) => {
+        const dbName = (event.data.startedData && event.data.startedData.databaseName) || "Unknown database";
+        client.trackDependency(
+                dbName,
+                event.data.event.commandName,
+                event.data.event.duration,
+                event.data.succeeded,
+                'mongodb');
+
+        if (!event.data.succeeded) {
+            client.trackException(new Error(event.data.event.failure));
+        }
+    });
+};
+
+export function enable(enabled: boolean, client: Client) {
+    if (enabled) {
+        if (clients.length === 0) {
+            channel.subscribe<mongodb.IMongoData>("mongodb", subscriber);
+        };
+        clients.push(client);
+    } else {
+        clients = clients.filter((c) => c != client);
+        if (clients.length === 0) {
+            channel.unsubscribe("mongodb", subscriber);
+        }
+    }
+}

AutoCollection/diagnostic-channel/mysql.sub.ts

@@ -0,0 +1,40 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for details.
+import Client = require("../../Library/Client");
+import {channel, IStandardEvent} from "diagnostic-channel";
+
+import {mysql} from "diagnostic-channel-publishers";
+
+let clients: Client[] = [];
+
+export const subscriber = (event: IStandardEvent<mysql.IMysqlData>) => {
+    clients.forEach((client) => {
+        const queryObj = event.data.query || {};
+        const sqlString = queryObj.sql || "Unknown query";
+        const success = !event.data.err;
+
+        const connection = queryObj._connection || {};
+        const connectionConfig = connection.config || {};
+        const dbName = connectionConfig.socketPath ? connectionConfig.socketPath : `${connectionConfig.host || "localhost"}:${connectionConfig.port}`;
+        client.trackDependency(
+                dbName,
+                sqlString,
+                event.data.duration | 0,
+                success,
+                "mysql");
+    });
+};
+
+export function enable(enabled: boolean, client: Client) {
+    if (enabled) {
+        if (clients.length === 0) {
+            channel.subscribe<mysql.IMysqlData>("mysql", subscriber);
+        };
+        clients.push(client);
+    } else {
+        clients = clients.filter((c) => c != client);
+        if (clients.length === 0) {
+            channel.unsubscribe("mysql", subscriber);
+        }
+    }
+}

AutoCollection/diagnostic-channel/redis.sub.ts

@@ -0,0 +1,38 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for details.
+import Client = require("../../Library/Client");
+import {channel, IStandardEvent} from "diagnostic-channel";
+
+import {redis} from "diagnostic-channel-publishers";
+
+let clients: Client[] = [];
+
+export const subscriber = (event: IStandardEvent<redis.IRedisData>) => {
+    clients.forEach((client) => {
+        if (event.data.commandObj.command === "info") {
+            // We don't want to report 'info', it's irrelevant
+            return;
+        }
+        client.trackDependency(
+            event.data.address,
+            event.data.commandObj.command,
+            event.data.duration,
+            !event.data.err,
+            "redis"
+        );
+    });
+};
+
+export function enable(enabled: boolean, client: Client) {
+    if (enabled) {
+        if (clients.length === 0) {
+            channel.subscribe<redis.IRedisData>("redis", subscriber);
+        };
+        clients.push(client);
+    } else {
+        clients = clients.filter((c) => c != client);
+        if (clients.length === 0) {
+            channel.unsubscribe("redis", subscriber);
+        }
+    }
+}

README.md

@@ -107,6 +107,14 @@ appInsights.client.config.samplingPercentage = 33; // 33% of all telemetry will
 appInsights.start();
 ```
 
+### Automatic third-party instrumentation
+
+In order to track context across asynchronous calls, some changes are required in third party libraries such as mongodb and redis. By default ApplicationInsights will use `diagnostic-channel-publishers` to monkey-patch some of these libraries. This can be disabled by setting the `APPLICATION_INSIGHTS_NO_DIAGNOSTIC_CHANNEL` environment variable. Note that by setting that environment variable, events may no longer be correctly associated with the right operation.
+
+Currently there are 6 packages which are instrumented: `bunyan`, `console`, `mongodb`, `mongodb-core`, `mysql` and `redis`.
+
+The `bunyan` package and `console` messages will generate Application Insights Trace events based on whether `setAutoCollectConsole` is enabled. The `mongodb`, `mysql` and `redis` packages will generate Application Insights Dependency events based on whether `setAutoCollectDependencies` is enabled.
+
 ## Track custom metrics
 
 You can track any request, event, metric or exception using the Application
@@ -179,7 +187,6 @@ More info on the telemetry API is available in [the docs][].
 
 [the docs]: https://azure.microsoft.com/documentation/articles/app-insights-api-custom-events-metrics/
 
-
 ## Use multiple instrumentation keys
 
 You can create multiple Azure Application Insights resources and send different

package.json

@@ -49,6 +49,8 @@
     "sinon": "1.17.6"
   },
   "dependencies": {
-    "zone.js": "0.7.6"
+    "zone.js": "0.7.6",
+    "diagnostic-channel": "0.1.0",
+    "diagnostic-channel-publishers": "0.1.1"
   }
 }