start of apm code

Author: christkv

index.js

@@ -1,5 +1,7 @@
 // Core module
-var core = require('mongodb-core');
+var core = require('mongodb-core'),
+  EventEmitter = require('events').EventEmitter,
+  inherits = require('util').inherits;
 
 // Set up the connect function
 var connect = require('./lib/mongo_client').connect;
@@ -35,5 +37,86 @@ connect.Timestamp = core.BSON.Timestamp;
 
 // Add connect method
 connect.connect = connect;
+// Operation id
+var operationId = 1;
+
+var Instrumentation = function() {
+  EventEmitter.call(this);
+  // Names of methods we need to wrap
+  var methods = ['command'];
+  // Prototype
+  var proto = core.Server.prototype;
+  // Reference
+  var self = this;
+  // Core server method we are going to wrap
+  methods.forEach(function(x) {
+    var func = proto[x];
+
+    // The actual prototype
+    proto[x] = function() {
+      var requestId = core.Query.nextRequestId();
+      var ourOpId = operationId++;
+      // Get the aruments
+      var args = Array.prototype.slice.call(arguments, 0);
+      var ns = args[0];
+      var commandObj = args[1];
+      var keys = Object.keys(commandObj);
+      var commandName = keys[0];
+      var db = ns.split('.')[0];
+
+      // Get a connection reference for this server instance
+      var connection = this.s.pool.get()
+      // Emit the start event for the command
+      var command = {type: 'started', operationId: ourOpId};
+      command.command = commandObj;
+      command.databaseName = db;
+      command.commandName = commandName
+      command.connectionId = connection;
+      command.requestId = requestId;
+      self.emit('command', command)
+
+      // Start time
+      var startTime = new Date().getTime();
+
+      // Get the callback
+      var callback = args.pop();
+      args.push(function(err, r) {
+        var endTime = new Date().getTime();
+        var command = {type: 'succeeded',
+          duration: (endTime - startTime),
+          requestId: requestId,
+          operationId: ourOpId,
+          connectionId: connection};
+
+        // If we have an error
+        if(err) {
+          command.type = 'failed'
+          command.failure = err;
+        } else {
+          command.reply = r;
+        }
+
+        // Emit the command
+        self.emit('command', command)
+
+        // Return to caller
+        callback(err, r);
+      });
+
+      // Apply the call
+      func.apply(this, args);
+    }
+  });
+}
+
+inherits(Instrumentation, EventEmitter);
+
+var instrumentation = new Instrumentation();
+
+// Set up the instrumentation method
+connect.instrument = function() {
+  return instrumentation;
+}
+
 // Set our exports to be the connect function
-module.exports = connect;
+module.exports = connect;

test/functional/apm_tests.js

@@ -0,0 +1,23 @@
+"use strict";
+
+exports['Correctly receive the APM events'] = {
+  metadata: { requires: { topology: ['single'] } },
+
+  // The actual test we wish to run
+  test: function(configuration, test) {
+    var listener = require('../..').instrument();
+    listener.on('command', function(event) {
+      console.log("===================================== received command")
+      console.log(JSON.stringify(event, null, 2))
+    });
+
+    var db = configuration.newDbInstance({w:1}, {poolSize:1, auto_reconnect:false});
+    db.open(function(err, db) {
+      db.collection('apm_test').insertOne({a:1}).then(function(r) {
+        test.equal(1, r.insertedCount);
+        db.close();
+        test.done();
+      });
+    });
+  }
+}

test/runner.js

@@ -254,6 +254,9 @@ var testFiles =[
   , '/test/functional/crud_api_tests.js'
   , '/test/functional/reconnect_tests.js'
 
+  // APM tests
+  , '/test/functional/apm_tests.js'
+
   // Logging tests
   , '/test/functional/logger_tests.js'