Merge pull request #834 from estolfo/RUBY-1096-sdam-monitoring

RUBY-1096 Implement SDAM Monitoring spec

Author: estolfo

lib/mongo/cluster.rb

@@ -24,6 +24,7 @@ module Mongo
   # @since 2.0.0
   class Cluster
     extend Forwardable
+    include Monitoring::Publishable
     include Event::Subscriber
     include Loggable
 
@@ -45,6 +46,9 @@ class Cluster
     # @return [ Hash ] The options hash.
     attr_reader :options
 
+    # @return [ Monitoring ] monitoring The monitoring.
+    attr_reader :monitoring
+
     # @return [ Object ] The cluster topology.
     attr_reader :topology
 
@@ -54,7 +58,8 @@ class Cluster
     # @since 2.4.0
     attr_reader :app_metadata
 
-    def_delegators :topology, :replica_set?, :replica_set_name, :sharded?, :single?, :unknown?
+    def_delegators :topology, :replica_set?, :replica_set_name, :sharded?,
+                   :single?, :unknown?, :member_discovered
     def_delegators :@cursor_reaper, :register_cursor, :schedule_kill_cursor, :unregister_cursor
 
     # Determine if this cluster of servers is equal to another object. Checks the
@@ -89,7 +94,6 @@ def add(host)
       address = Address.new(host)
       if !addresses.include?(address)
         if addition_allowed?(address)
-          log_debug("Adding #{address.to_s} to the cluster.")
           @update_lock.synchronize { @addresses.push(address) }
           server = Server.new(address, self, @monitoring, event_listeners, options)
           @update_lock.synchronize { @servers.push(server) }
@@ -98,6 +102,34 @@ def add(host)
       end
     end
 
+    # Determine if the cluster would select a readable server for the
+    # provided read preference.
+    #
+    # @example Is a readable server present?
+    #   topology.has_readable_server?(server_selector)
+    #
+    # @param [ ServerSelector ] server_selector The server
+    #   selector.
+    #
+    # @return [ true, false ] If a readable server is present.
+    #
+    # @since 2.4.0
+    def has_readable_server?(server_selector = nil)
+      topology.has_readable_server?(self, server_selector)
+    end
+
+    # Determine if the cluster would select a writable server.
+    #
+    # @example Is a writable server present?
+    #   topology.has_writable_server?
+    #
+    # @return [ true, false ] If a writable server is present.
+    #
+    # @since 2.4.0
+    def has_writable_server?
+      topology.has_writable_server?(self)
+    end
+
     # Instantiate the new cluster.
     #
     # @api private
@@ -119,16 +151,26 @@ def initialize(seeds, monitoring, options = Options::Redacted.new)
       @event_listeners = Event::Listeners.new
       @options = options.freeze
       @app_metadata ||= AppMetadata.new(self)
-      @topology = Topology.initial(seeds, options)
       @update_lock = Mutex.new
       @pool_lock = Mutex.new
+      @topology = Topology.initial(seeds, monitoring, options)
+
+      publish_sdam_event(
+        Monitoring::TOPOLOGY_OPENING,
+        Monitoring::Event::TopologyOpening.new(@topology)
+      )
 
       subscribe_to(Event::STANDALONE_DISCOVERED, Event::StandaloneDiscovered.new(self))
       subscribe_to(Event::DESCRIPTION_CHANGED, Event::DescriptionChanged.new(self))
-      subscribe_to(Event::PRIMARY_ELECTED, Event::PrimaryElected.new(self))
+      subscribe_to(Event::MEMBER_DISCOVERED, Event::MemberDiscovered.new(self))
 
       seeds.each{ |seed| add(seed) }
 
+      publish_sdam_event(
+        Monitoring::TOPOLOGY_CHANGED,
+        Monitoring::Event::TopologyChanged.new(@topology, @topology)
+      ) if @servers.size > 1
+
       @cursor_reaper = CursorReaper.new
       @cursor_reaper.run!
 
@@ -264,11 +306,14 @@ def standalone_discovered
     #
     # @since 2.0.0
     def remove(host)
-      log_debug("#{host} being removed from the cluster.")
       address = Address.new(host)
       removed_servers = @servers.select { |s| s.address == address }
       @update_lock.synchronize { @servers = @servers - removed_servers }
       removed_servers.each{ |server| server.disconnect! } if removed_servers
+      publish_sdam_event(
+        Monitoring::SERVER_CLOSED,
+        Monitoring::Event::ServerClosed.new(address, topology)
+      )
       @update_lock.synchronize { @addresses.reject! { |addr| addr == address } }
     end
 

lib/mongo/cluster/topology.rb

@@ -26,33 +26,34 @@ class Cluster
     module Topology
       extend self
 
-      # The 2 various topologies for server selection.
+      # The various topologies for server selection.
       #
       # @since 2.0.0
       OPTIONS = {
         replica_set: ReplicaSet,
         sharded: Sharded,
         direct: Single
-      }
+      }.freeze
 
       # Get the initial cluster topology for the provided options.
       #
       # @example Get the initial cluster topology.
       #   Topology.initial(topology: :replica_set)
       #
       # @param [ Array<String> ] seeds The addresses of the configured servers.
+      # @param [ Monitoring ] monitoring The monitoring.
       # @param [ Hash ] options The cluster options.
       #
       # @return [ ReplicaSet, Sharded, Single ] The topology.
       #
       # @since 2.0.0
-      def initial(seeds, options)
+      def initial(seeds, monitoring, options)
         if options.has_key?(:connect)
-          OPTIONS.fetch(options[:connect]).new(options, seeds)
+          OPTIONS.fetch(options[:connect]).new(options, monitoring, seeds)
         elsif options.has_key?(:replica_set)
-          ReplicaSet.new(options, seeds)
+          ReplicaSet.new(options, monitoring, options)
         else
-          Unknown.new(options, seeds)
+          Unknown.new(options, monitoring, seeds)
         end
       end
     end

lib/mongo/cluster/topology/replica_set.rb

@@ -21,6 +21,7 @@ module Topology
       # @since 2.0.0
       class ReplicaSet
         include Loggable
+        include Monitoring::Publishable
 
         # Constant for the replica set name configuration option.
         #
@@ -30,6 +31,9 @@ class ReplicaSet
         # @return [ Hash ] options The options.
         attr_reader :options
 
+        # @return [ Monitoring ] monitoring The monitoring.
+        attr_reader :monitoring
+
         # The display name for the topology.
         #
         # @since 2.0.0
@@ -61,7 +65,6 @@ def display_name
         def elect_primary(description, servers)
           if description.replica_set_name == replica_set_name
             unless detect_stale_primary!(description)
-              log_debug("Server #{description.address.to_s} elected as primary in #{replica_set_name}.")
               servers.each do |server|
                 if server.primary? && server.address != description.address
                   server.description.unknown!
@@ -79,16 +82,51 @@ def elect_primary(description, servers)
           self
         end
 
+        # Determine if the topology would select a readable server for the
+        # provided candidates and read preference.
+        #
+        # @example Is a readable server present?
+        #   topology.has_readable_server?(cluster, server_selector)
+        #
+        # @param [ Cluster ] cluster The cluster.
+        # @param [ ServerSelector ] server_selector The server
+        #   selector.
+        #
+        # @return [ true, false ] If a readable server is present.
+        #
+        # @since 2.4.0
+        def has_readable_server?(cluster, server_selector = nil)
+          (server_selector || ServerSelector.get(mode: :primary)).candidates(cluster).any?
+        end
+
+        # Determine if the topology would select a writable server for the
+        # provided candidates.
+        #
+        # @example Is a writable server present?
+        #   topology.has_writable_server?(servers)
+        #
+        # @param [ Cluster ] cluster The cluster.
+        #
+        # @return [ true, false ] If a writable server is present.
+        #
+        # @since 2.4.0
+        def has_writable_server?(cluster)
+          cluster.servers.any?{ |server| server.primary? }
+        end
+
         # Initialize the topology with the options.
         #
         # @example Initialize the topology.
         #   ReplicaSet.new(options)
         #
         # @param [ Hash ] options The options.
+        # @param [ Monitoring ] monitoring The monitoring.
+        # @param [ Array<String> ] seeds The seeds.
         #
         # @since 2.0.0
-        def initialize(options, seeds = [])
+        def initialize(options, monitoring, seeds = [])
           @options = options
+          @monitoring = monitoring
           @max_election_id = nil
           @max_set_version = nil
         end
@@ -222,6 +260,14 @@ def unknown?; false; end
         # @since 2.0.6
         def standalone_discovered; self; end
 
+        # Notify the topology that a member was discovered.
+        #
+        # @example Notify the topology that a member was discovered.
+        #   topology.member_discovered
+        #
+        # @since 2.4.0
+        def member_discovered; end;
+
         private
 
         def update_max_election_id(description)

lib/mongo/cluster/topology/sharded.rb

@@ -20,12 +20,19 @@ module Topology
       #
       # @since 2.0.0
       class Sharded
+        include Monitoring::Publishable
 
         # The display name for the topology.
         #
         # @since 2.0.0
         NAME = 'Sharded'.freeze
 
+        # @return [ Hash ] options The options.
+        attr_reader :options
+
+        # @return [ Monitoring ] monitoring The monitoring.
+        attr_reader :monitoring
+
         # Get the display name.
         #
         # @example Get the display name.
@@ -51,16 +58,47 @@ def display_name
         # @return [ Sharded ] The topology.
         def elect_primary(description, servers); self; end
 
+        # Determine if the topology would select a readable server for the
+        # provided candidates and read preference.
+        #
+        # @example Is a readable server present?
+        #   topology.has_readable_server?(cluster, server_selector)
+        #
+        # @param [ Cluster ] cluster The cluster.
+        # @param [ ServerSelector ] server_selector The server
+        #   selector.
+        #
+        # @return [ true ] A Sharded cluster always has a readable server.
+        #
+        # @since 2.4.0
+        def has_readable_server?(cluster, server_selector = nil); true; end
+
+        # Determine if the topology would select a writable server for the
+        # provided candidates.
+        #
+        # @example Is a writable server present?
+        #   topology.has_writable_server?(servers)
+        #
+        # @param [ Cluster ] cluster The cluster.
+        #
+        # @return [ true ] A Sharded cluster always has a writable server.
+        #
+        # @since 2.4.0
+        def has_writable_server?(cluster); true; end
+
         # Initialize the topology with the options.
         #
         # @example Initialize the topology.
         #   Sharded.new(options)
         #
         # @param [ Hash ] options The options.
+        # @param [ Monitoring ] monitoring The monitoring.
+        # @param [ Array<String> ] seeds The seeds.
         #
         # @since 2.0.0
-        def initialize(options, seeds = [])
+        def initialize(options, monitoring, seeds = [])
           @options = options
+          @monitoring = monitoring
         end
 
         # A sharded topology is not a replica set.
@@ -181,6 +219,14 @@ def unknown?; false; end
         # @since 2.0.6
         def standalone_discovered; self; end
 
+        # Notify the topology that a member was discovered.
+        #
+        # @example Notify the cluster that a member was discovered.
+        #   topology.member_discovered
+        #
+        # @since 2.4.0
+        def member_discovered; end;
+
         private
 
         def remove_self?(description, server)