RUBY-1560 Finalize CMAP event API

docs/tutorials/ruby-driver-monitoring.txt

@@ -13,10 +13,11 @@ Monitoring
 The driver allows the application to be notified when certain events happen.
 These events are organized into the following categories:
 
-- Command monitoring;
-- Topology lifecycle;
-- Server lifecycle;
-- Server heartbeats.
+- Command monitoring
+- Topology lifecycle
+- Server lifecycle
+- Server heartbeats
+- Connection pools and connections
 
 Topology and server events are part of Server Discovery and Monitoring (SDAM).
 
@@ -331,6 +332,60 @@ Sample output:
   D, [2018-09-23T13:44:10.707018 #1739] DEBUG -- : HEARTBEAT | 127.0.0.1:27027 | STARTED
   D, [2018-09-23T13:44:10.707778 #1739] DEBUG -- : HEARTBEAT | 127.0.0.1:27027 | SUCCEEDED | 0.000772381s
 
+Connection Pool And Connection Monitoring
+-----------------------------------------
+
+Each client maintains a connection pool for each server in the deployment that
+it is aware of, and publishes events for both connection pools and individual
+connections. To subscribe to these events, define a subscriber class implementing
+the method ``pubished`` which takes a single parameter for the event that
+is being published. Note that future versions of the driver may introduce
+additional events published through this mechanism.
+
+The following events are currently implemented by the driver, following
+the `CMAP specification <https://github.com/mongodb/specifications/blob/master/source/connection-monitoring-and-pooling/connection-monitoring-and-pooling.rst>`_:
+
+- PoolCreated
+- PoolCleared
+- PoolClosed
+- ConnectionCreated
+- ConnectionReady
+- ConnectionClosed
+- ConnectionCheckOutStarted
+- ConnectionCheckOutFailed
+- ConnectionCheckOutSucceeded
+- ConnectionCheckedIn
+
+The driver provides a logging subscriber which may be used to log all
+connection pool and connection-related events. This subscriber is not enabled
+by default because it will create log entries for each operation performed
+by the application. To enable this subscriber globally or per client:
+
+.. code-block:: ruby
+
+  Mongo::Monitoring::Global.subscribe(
+    Mongo::Monitoring::CONNECTION_POOL,
+    Mongo::Monitoring::CmapLogSubscriber.new)
+
+  client = Mongo::Client.new([ '127.0.0.1:27017' ], :database => 'test' )
+  subscriber = Mongo::Monitoring::CmapLogSubscriber.new
+  client.subscribe( Mongo::Monitoring::CONNECTION_POOL, subscriber )
+
+Sample output:
+
+.. code-block:: none
+
+  D, [2019-05-06T17:23:21.595412 #8576] DEBUG -- : MONGODB | EVENT: #<PoolCreated address=127.0.0.1:27741 options={...}>
+  D, [2019-05-06T17:23:21.595584 #8576] DEBUG -- : MONGODB | EVENT: #<PoolCleared address=127.0.0.1:27741>
+  D, [2019-05-06T17:23:21.603549 #8576] DEBUG -- : MONGODB | EVENT: #<PoolCreated address=localhost:27741 options={...}>
+  D, [2019-05-06T17:23:21.603616 #8576] DEBUG -- : MONGODB | EVENT: #<ConnectionCheckOutStarted address=localhost:27741>
+  D, [2019-05-06T17:23:21.603684 #8576] DEBUG -- : MONGODB | EVENT: #<ConnectionCreated address=localhost:27741 connection_id=1>
+  D, [2019-05-06T17:23:21.604079 #8576] DEBUG -- : MONGODB | EVENT: #<ConnectionCheckedOut address=localhost:27741 connection_id=1>
+  D, [2019-05-06T17:23:21.605759 #8576] DEBUG -- : MONGODB | EVENT: #<ConnectionReady address=localhost:27741 connection_id=1>
+  D, [2019-05-06T17:23:21.605784 #8576] DEBUG -- : MONGODB | EVENT: #<ConnectionCheckedIn address=localhost:27741 connection_id=1>
+  D, [2019-05-06T17:23:21.605817 #8576] DEBUG -- : MONGODB | EVENT: #<PoolCleared address=localhost:27741>
+  D, [2019-05-06T17:23:21.605852 #8576] DEBUG -- : MONGODB | EVENT: #<ConnectionClosed address=localhost:27741 connection_id=1 reason=stale>
+
 
 Disabling Monitoring
 --------------------

lib/mongo/monitoring.rb

@@ -223,6 +223,9 @@ def initialize(options = {})
           end
         end
         subscribe(COMMAND, CommandLogSubscriber.new(options))
+        # CMAP events are not logged by default because this will create
+        # log entries for every operation performed by the driver.
+        #subscribe(CONNECTION_POOL, CmapLogSubscriber.new(options))
         subscribe(SERVER_OPENING, ServerOpeningLogSubscriber.new(options))
         subscribe(SERVER_CLOSED, ServerClosedLogSubscriber.new(options))
         subscribe(SERVER_DESCRIPTION_CHANGED, ServerDescriptionChangedLogSubscriber.new(options))
@@ -240,8 +243,24 @@ def monitoring?
       options[:monitoring] != false
     end
 
+    # Publish an event.
+    #
+    # This method is used for event types which only have a single event
+    # in them.
+    #
+    # @param [ String ] topic The event topic.
+    # @param [ Event ] event The event to publish.
+    #
+    # @since 2.9.0
+    def published(topic, event)
+      subscribers_for(topic).each{ |subscriber| subscriber.published(event) }
+    end
+
     # Publish a started event.
     #
+    # This method is used for event types which have the started/succeeded/failed
+    # events in them, such as command and heartbeat events.
+    #
     # @example Publish a started event.
     #   monitoring.started(COMMAND, event)
     #
@@ -255,6 +274,9 @@ def started(topic, event)
 
     # Publish a succeeded event.
     #
+    # This method is used for event types which have the started/succeeded/failed
+    # events in them, such as command and heartbeat events.
+    #
     # @example Publish a succeeded event.
     #   monitoring.succeeded(COMMAND, event)
     #
@@ -268,6 +290,9 @@ def succeeded(topic, event)
 
     # Publish a failed event.
     #
+    # This method is used for event types which have the started/succeeded/failed
+    # events in them, such as command and heartbeat events.
+    #
     # @example Publish a failed event.
     #   monitoring.failed(COMMAND, event)
     #
@@ -293,6 +318,7 @@ def initialize_copy(original)
 require 'mongo/monitoring/event'
 require 'mongo/monitoring/publishable'
 require 'mongo/monitoring/command_log_subscriber'
+require 'mongo/monitoring/cmap_log_subscriber'
 require 'mongo/monitoring/sdam_log_subscriber'
 require 'mongo/monitoring/server_description_changed_log_subscriber'
 require 'mongo/monitoring/server_closed_log_subscriber'

lib/mongo/monitoring/cmap_log_subscriber.rb

@@ -0,0 +1,53 @@
+# Copyright (C) 2019 MongoDB, Inc.
+#
+# 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
+#
+#   http://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.
+
+module Mongo
+  class Monitoring
+
+    # Subscribes to CMAP events and logs them.
+    #
+    # @since 2.9.0
+    class CmapLogSubscriber
+      include Loggable
+
+      # @return [ Hash ] options The options.
+      #
+      # @since 2.9.0
+      attr_reader :options
+
+      # Create the new log subscriber.
+      #
+      # @example Create the log subscriber.
+      #   CmapLogSubscriber.new
+      #
+      # @param [ Hash ] options The options.
+      #
+      # @option options [ Logger ] :logger An optional custom logger.
+      #
+      # @since 2.9.0
+      def initialize(options = {})
+        @options = options
+      end
+
+      # Handle a CMAP event.
+      #
+      # @param [ Event ] event The event.
+      #
+      # @since 2.9.0
+      def published(event)
+        log_debug("EVENT: #{event.summary}") if logger.debug?
+      end
+    end
+  end
+end

lib/mongo/monitoring/event/cmap/connection_checked_in.rb

@@ -60,4 +60,4 @@ def summary
       end
     end
   end
-end
+end

lib/mongo/monitoring/publishable.rb

@@ -24,6 +24,7 @@ module Publishable
       # @return [ Monitoring ] monitoring The monitoring.
       attr_reader :monitoring
 
+      # @deprecated
       def publish_event(topic, event)
         monitoring.succeeded(topic, event)
       end
@@ -38,7 +39,7 @@ def publish_sdam_event(topic, event)
       def publish_cmap_event(event)
         return unless monitoring?
 
-        publish_event(Monitoring::CONNECTION_POOL, event)
+        monitoring.published(Monitoring::CONNECTION_POOL, event)
       end
 
       private

lib/mongo/server/connection_pool.rb

@@ -518,7 +518,7 @@ def create_connection
       #
       # @raise [ Error::PoolClosedError ] If the pool has been closed.
       #
-      # @since 2.8.0
+      # @since 2.9.0
       def raise_if_closed!
         if closed?
           raise Error::PoolClosedError.new(@server.address)

spec/integration/step_down_spec.rb

@@ -64,7 +64,7 @@
       item = enum.next
       expect(item['test']).to eq(1)
 
-      connection_created_events = EventSubscriber.succeeded_events.select do |event|
+      connection_created_events = EventSubscriber.published_events.select do |event|
         event.is_a?(Mongo::Monitoring::Event::Cmap::ConnectionCreated)
       end
       expect(connection_created_events).not_to be_empty
@@ -90,7 +90,7 @@
       expect(get_more_events.length).to eq(1)
 
       # getMore should have been sent on the same connection as find
-      connection_created_events = EventSubscriber.succeeded_events.select do |event|
+      connection_created_events = EventSubscriber.published_events.select do |event|
         event.is_a?(Mongo::Monitoring::Event::Cmap::ConnectionCreated)
       end
       expect(connection_created_events).to be_empty
@@ -154,7 +154,7 @@
           collection.insert_one(test: 1)
         end.to raise_error(Mongo::Error::OperationFailure, /10107/)
 
-        expect(event_subscriber.select_cmap_events(Mongo::Monitoring::Event::Cmap::PoolCleared).count).to eq(0)
+        expect(event_subscriber.select_published_events(Mongo::Monitoring::Event::Cmap::PoolCleared).count).to eq(0)
       end
     end
 
@@ -173,7 +173,7 @@
           collection.insert_one(test: 1)
         end.to raise_error(Mongo::Error::OperationFailure, /10107/)
 
-        expect(event_subscriber.select_cmap_events(Mongo::Monitoring::Event::Cmap::PoolCleared).count).to eq(1)
+        expect(event_subscriber.select_published_events(Mongo::Monitoring::Event::Cmap::PoolCleared).count).to eq(1)
       end
     end
 
@@ -190,7 +190,7 @@
           collection.insert_one(test: 1)
         end.to raise_error(Mongo::Error::OperationFailure, /11600/)
 
-        expect(event_subscriber.select_cmap_events(Mongo::Monitoring::Event::Cmap::PoolCleared).count).to eq(1)
+        expect(event_subscriber.select_published_events(Mongo::Monitoring::Event::Cmap::PoolCleared).count).to eq(1)
       end
     end
   end

spec/support/cmap.rb

@@ -84,7 +84,7 @@ def run
           end
 
           result['error'] ||= nil
-          result['events'] = subscriber.succeeded_events.reduce([]) do |events, event|
+          result['events'] = subscriber.published_events.reduce([]) do |events, event|
             next events unless event.is_a?(Mongo::Monitoring::Event::Cmap::Base)
 
             event = case event
@@ -322,18 +322,22 @@ def run_wait_for_thread_op(state)
 
       def run_wait_for_event_op(state)
         subscriber = @spec.subscriber
-        looped = false
+        looped = 0
+        deadline = Time.now + 3
         loop do
-          actual_events = @spec.subscriber.succeeded_events.select do |e|
+          actual_events = @spec.subscriber.published_events.select do |e|
             e.class.name.sub(/.*::/, '').sub(/^ConnectionPool/, 'Pool') == @event.sub(/^ConnectionPool/, 'Pool')
           end
           if actual_events.length >= @count
             break
           end
-          if looped
+          if looped == 1
             puts("Waiting for #{@count} #{@event} events (have #{actual_events.length}): #{@spec.description}")
           end
-          looped = true
+          if Time.now > deadline
+            raise "Did not receive #{@count} #{@event} events in time (have #{actual_events.length}): #{@spec.description}"
+          end
+          looped += 1
           sleep 0.1
         end
       end

spec/support/event_subscriber.rb

@@ -20,6 +20,8 @@ module Impl
     # @since 2.5.0
     attr_reader :failed_events
 
+    attr_reader :published_events
+
     # Cache the succeeded event.
     #
     # @param [ Event ] event The event.
@@ -53,15 +55,15 @@ def failed(event)
       end
     end
 
-    def select_cmap_events(cls)
-      succeeded_events.select do |event|
+    def select_published_events(cls)
+      @published_events.select do |event|
         event.is_a?(cls)
       end
     end
 
-    def cmap_event(event)
+    def published(event)
       @mutex.synchronize do
-        @cmap_events << event
+        @published_events << event
       end
     end
 
@@ -72,7 +74,7 @@ def clear_events!
       @started_events = []
       @succeeded_events = []
       @failed_events = []
-      @cmap_events = []
+      @published_events = []
       self
     end