SPEC-222: SDAM Monitoring

lib/mongo/cluster.rb

@@ -22,6 +22,7 @@ module Mongo
   # @since 2.0.0
   class Cluster
     extend Forwardable
+    include Monitoring::Publishable
     include Event::Subscriber
     include Loggable
 
@@ -38,6 +39,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
 
@@ -75,7 +79,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) }
@@ -84,6 +87,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.3.0
+    def has_readable_server?(server_selector)
+      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.3.0
+    def has_writable_server?
+      topology.has_writable_server?(self)
+    end
+
     # Instantiate the new cluster.
     #
     # @api private
@@ -104,7 +135,7 @@ def initialize(seeds, monitoring, options = Options::Redacted.new)
       @monitoring = monitoring
       @event_listeners = Event::Listeners.new
       @options = options.freeze
-      @topology = Topology.initial(seeds, options)
+      @topology = Topology.initial(seeds, monitoring, options)
       @update_lock = Mutex.new
       @pool_lock = Mutex.new
 
@@ -242,11 +273,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

@@ -33,27 +33,29 @@ module Topology
         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)
-        if options.has_key?(:connect)
-          OPTIONS.fetch(options[:connect]).new(options, seeds)
+      def initial(seeds, monitoring, options)
+        topology = if options.has_key?(:connect)
+          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
+        topology
       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
@@ -79,18 +83,57 @@ 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.3.0
+        def has_readable_server?(cluster, server_selector)
+          server_selector.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.3.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
+          publish_sdam_event(
+            Monitoring::TOPOLOGY_OPENING,
+            Monitoring::Event::TopologyOpening.new(self)
+          )
         end
 
         # A replica set topology is a replica set.

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,51 @@ 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, false ] If a readable server is present.
+        #
+        # @since 2.3.0
+        def has_readable_server?(cluster, server_selector); 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, false ] If a writable server is present.
+        #
+        # @since 2.3.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
+          publish_sdam_event(
+            Monitoring::TOPOLOGY_OPENING,
+            Monitoring::Event::TopologyOpening.new(self)
+          )
         end
 
         # A sharded topology is not a replica set.

lib/mongo/cluster/topology/single.rb

@@ -20,15 +20,22 @@ module Topology
       #
       # @since 2.0.0
       class Single
-
-        # @return [ String ] seed The seed address.
-        attr_reader :seed
+        include Monitoring::Publishable
 
         # The display name for the topology.
         #
         # @since 2.0.0
         NAME = 'Single'.freeze
 
+        # @return [ Hash ] options The options.
+        attr_reader :options
+
+        # @return [ String ] seed The seed address.
+        attr_reader :seed
+
+        # @return [ monitoring ] monitoring the monitoring.
+        attr_reader :monitoring
+
         # Get the display name.
         #
         # @example Get the display name.
@@ -54,17 +61,56 @@ def display_name
         # @return [ Single ] 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, false ] If a readable server is present.
+        #
+        # @since 2.3.0
+        def has_readable_server?(cluster, server_selector)
+          server_selector.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.3.0
+        def has_writable_server?(cluster)
+          cluster.servers.any?{ |server| server.primary? }
+        end
+
         # Initialize the topology with the options.
         #
         # @example Initialize the topology.
         #   Single.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
           @seed = seeds.first
+          publish_sdam_event(
+            Monitoring::TOPOLOGY_OPENING,
+            Monitoring::Event::TopologyOpening.new(self)
+          )
         end
 
         # A single topology is not a replica set.