RUBY-1955 Log connection ids in command logger (#1552)

* RUBY-1955 Log connection ids in command logger

* RUBY-1955 Also log request id to be able to easily correlate started & succeeded/failed entries

* RUBY-1955 Pending connections also need ids

Author: p-mongo

lib/mongo/monitoring/command_log_subscriber.rb

@@ -54,7 +54,8 @@ def initialize(options = {})
       # @since 2.1.0
       def started(event)
         if logger.debug?
-          log_debug("#{prefix(event)} | STARTED | #{format_command(event.command)}")
+          log_debug("#{prefix(event, connection_id: event.connection_id)} | " +
+            "STARTED | #{format_command(event.command)}")
         end
       end
 
@@ -96,8 +97,9 @@ def format_command(args)
         end
       end
 
-      def prefix(event)
-        "#{event.address.to_s} | #{event.database_name}.#{event.command_name}"
+      def prefix(event, connection_id: nil)
+        "[#{event.request_id}] #{event.address.to_s}#{connection_id && " ##{connection_id}"} | " +
+          "#{event.database_name}.#{event.command_name}"
       end
 
       def truncate(command)

lib/mongo/monitoring/event/command_started.rb

@@ -45,6 +45,12 @@ class CommandStarted < Mongo::Event::Base
         # @api private
         attr_reader :socket_object_id
 
+        # @return [ Integer ] The ID for the connection over which the command
+        #   is sent.
+        #
+        # @api private
+        attr_reader :connection_id
+
         # Create the new event.
         #
         # @example Create the event.
@@ -59,7 +65,7 @@ class CommandStarted < Mongo::Event::Base
         # @since 2.1.0
         # @api private
         def initialize(command_name, database_name, address, request_id,
-          operation_id, command, socket_object_id = nil
+          operation_id, command, socket_object_id: nil, connection_id: nil
         )
           @command_name = command_name.to_s
           @database_name = database_name
@@ -68,6 +74,7 @@ def initialize(command_name, database_name, address, request_id,
           @operation_id = operation_id
           @command = redacted(command_name, command)
           @socket_object_id = socket_object_id
+          @connection_id = connection_id
         end
 
         # Create the event from a wire protocol message payload.
@@ -83,15 +90,18 @@ def initialize(command_name, database_name, address, request_id,
         #
         # @since 2.1.0
         # @api private
-        def self.generate(address, operation_id, payload, socket_object_id = nil)
+        def self.generate(address, operation_id, payload,
+          socket_object_id: nil, connection_id: nil
+        )
           new(
             payload[:command_name],
             payload[:database_name],
             address,
             payload[:request_id],
             operation_id,
             payload[:command],
-            socket_object_id,
+            socket_object_id: socket_object_id,
+            connection_id: connection_id,
           )
         end
 

lib/mongo/monitoring/publishable.rb

@@ -44,11 +44,13 @@ def publish_cmap_event(event)
 
       private
 
-      def command_started(address, operation_id, payload, socket_object_id = nil)
+      def command_started(address, operation_id, payload,
+        socket_object_id: nil, connection_id: nil
+      )
         monitoring.started(
           Monitoring::COMMAND,
           Event::CommandStarted.generate(address, operation_id, payload,
-            socket_object_id)
+            socket_object_id: socket_object_id, connection_id: connection_id)
         )
       end
 

lib/mongo/server/connection.rb

@@ -186,7 +186,7 @@ def do_connect
 
         begin
           handshake!(socket)
-          pending_connection = PendingConnection.new(socket, @server, monitoring, options)
+          pending_connection = PendingConnection.new(socket, @server, monitoring, options.merge(id: id))
           authenticate!(pending_connection)
         rescue Exception
           socket.close

lib/mongo/server/connection_base.rb

@@ -96,7 +96,8 @@ def deliver(message)
         buffer = serialize(message)
         ensure_connected do |socket|
           operation_id = Monitoring.next_operation_id
-          command_started(address, operation_id, message.payload, socket.object_id)
+          command_started(address, operation_id, message.payload,
+            socket_object_id: socket.object_id, connection_id: id)
           start = Time.now
           result = nil
           begin

lib/mongo/server/pending_connection.rb

@@ -26,8 +26,14 @@ def initialize(socket, server, monitoring, options = {})
         @options = options
         @server = server
         @monitoring = monitoring
+        @id = options[:id]
       end
 
+      # @return [ Integer ] The ID for the connection. This is the same ID
+      #   as that of the regular Connection object for which this
+      #   PendingConnection instance was created.
+      attr_reader :id
+
       def ensure_connected
         yield @socket
       end