feat: include connectionId for APM with new CMAP connection pool

With the new CMAP pool we have the ability to include a connection
id in all of our APM messages, allowing users to correlate APM
events with CMAP events for greater traceability.

NODE-2419

Author: mbroadst

lib/core/connection/apm.js

@@ -25,6 +25,7 @@ const collectionName = command => command.ns.split('.')[1];
 const generateConnectionId = pool =>
   pool.options ? `${pool.options.host}:${pool.options.port}` : pool.address;
 const maybeRedact = (commandName, result) => (SENSITIVE_COMMANDS.has(commandName) ? {} : result);
+const isLegacyPool = pool => pool.s && pool.queue;
 
 const LEGACY_FIND_QUERY_MAP = {
   $query: 'filter',
@@ -151,6 +152,22 @@ const extractReply = (command, reply) => {
   return reply && reply.result ? reply.result : reply;
 };
 
+const extractConnectionDetails = pool => {
+  if (isLegacyPool(pool)) {
+    return {
+      connectionId: generateConnectionId(pool)
+    };
+  }
+
+  // APM in the modern pool is done at the `Connection` level, so we rename it here for
+  // readability.
+  const connection = pool;
+  return {
+    address: connection.address,
+    connectionId: connection.id
+  };
+};
+
 /** An event indicating the start of a given command */
 class CommandStartedEvent {
   /**
@@ -162,15 +179,15 @@ class CommandStartedEvent {
   constructor(pool, command) {
     const cmd = extractCommand(command);
     const commandName = extractCommandName(cmd);
+    const connectionDetails = extractConnectionDetails(pool);
 
     // NOTE: remove in major revision, this is not spec behavior
     if (SENSITIVE_COMMANDS.has(commandName)) {
       this.commandObj = {};
       this.commandObj[commandName] = true;
     }
 
-    Object.assign(this, {
-      connectionId: generateConnectionId(pool),
+    Object.assign(this, connectionDetails, {
       requestId: command.requestId,
       databaseName: databaseName(command),
       commandName,
@@ -192,9 +209,9 @@ class CommandSucceededEvent {
   constructor(pool, command, reply, started) {
     const cmd = extractCommand(command);
     const commandName = extractCommandName(cmd);
+    const connectionDetails = extractConnectionDetails(pool);
 
-    Object.assign(this, {
-      connectionId: generateConnectionId(pool),
+    Object.assign(this, connectionDetails, {
       requestId: command.requestId,
       commandName,
       duration: calculateDurationInMs(started),
@@ -216,9 +233,9 @@ class CommandFailedEvent {
   constructor(pool, command, error, started) {
     const cmd = extractCommand(command);
     const commandName = extractCommandName(cmd);
+    const connectionDetails = extractConnectionDetails(pool);
 
-    Object.assign(this, {
-      connectionId: generateConnectionId(pool),
+    Object.assign(this, connectionDetails, {
       requestId: command.requestId,
       commandName,
       duration: calculateDurationInMs(started),

test/functional/apm.test.js

@@ -224,8 +224,15 @@ describe('APM', function() {
           .then(() => {
             expect(started).to.have.lengthOf(2);
 
-            // Ensure command was not sent to the primary
-            expect(started[0].connectionId).to.not.equal(started[1].connectionId);
+            if (self.configuration.usingUnifiedTopology()) {
+              expect(started[0])
+                .property('address')
+                .to.not.equal(started[1].address);
+            } else {
+              // Ensure command was not sent to the primary
+              expect(started[0].connectionId).to.not.equal(started[1].connectionId);
+            }
+
             return client.close();
           });
       });
@@ -274,7 +281,14 @@ describe('APM', function() {
             expect(started).to.have.lengthOf(2);
 
             // Ensure command was not sent to the primary
-            expect(started[0].connectionId).to.not.equal(started[1].connectionId);
+            if (self.configuration.usingUnifiedTopology()) {
+              expect(started[0])
+                .property('address')
+                .to.not.equal(started[1].address);
+            } else {
+              expect(started[0].connectionId).to.not.equal(started[1].connectionId);
+            }
+
             return client.close();
           });
       });