class Mongo::Cluster::SdamFlow
Handles SDAM flow for a server description changed event.
Updates server descriptions, topology descriptions and publishes SDAM events.
SdamFlow
is meant to be instantiated once for every server description changed event that needs to be processed.
@api private
Attributes
The topology stored in this attribute can change multiple times throughout a single sdam flow (e.g. unknown -> RS no primary -> RS with primary). Events for topology change get sent at the end of flow processing, such that the above example only publishes an unknown -> RS with primary event to the application.
@return Mongo::Cluster::Topology
The current topology.
Public Class Methods
# File lib/mongo/cluster/sdam_flow.rb, line 28 def initialize(cluster, previous_desc, updated_desc) @cluster = cluster @topology = cluster.topology @original_desc = @previous_desc = previous_desc @updated_desc = updated_desc @servers_to_disconnect = [] end
Public Instance Methods
Adds all servers referenced in the given description (which is supposed to have come from a good primary) which are not already in the cluster, to the cluster.
@note Servers are added unmonitored. Monitoring
must be started later separately.
@return [ Array<Server> ] Servers actually added to the cluster.
This is the set of servers on which monitoring should be started.
# File lib/mongo/cluster/sdam_flow.rb, line 374 def add_servers_from_desc(updated_desc) added_servers = [] address_strs = servers_list.map(&:address).map(&:to_s) %w(hosts passives arbiters).each do |m| updated_desc.send(m).each do |address_str| if server = cluster.add(address_str, monitor: false) added_servers << server end end end added_servers end
Returns whether the server whose description this flow processed was not previously unknown, and is now. Used to decide, in particular, whether to clear the server's connection pool.
# File lib/mongo/cluster/sdam_flow.rb, line 584 def became_unknown? updated_desc.unknown? && !original_desc.unknown? end
Checks if the cluster has a primary, and if not, transitions the topology to ReplicaSetNoPrimary. Topology
must be ReplicaSetWithPrimary when invoking this method.
# File lib/mongo/cluster/sdam_flow.rb, line 553 def check_if_has_primary unless topology.replica_set? raise ArgumentError, "check_if_has_primary should only be called when topology is replica set, but it is #{topology.class.name.sub(/.*::/, '')}" end primary = servers_list.detect do |server| # A primary with the wrong set name is not a primary server.primary? && server.description.replica_set_name == topology.replica_set_name end unless primary @topology = Topology::ReplicaSetNoPrimary.new( topology.options, topology.monitoring, self) end end
Publishes server description changed events, updates topology on the cluster and publishes topology changed event, as needed based on operations performed during SDAM flow processing.
# File lib/mongo/cluster/sdam_flow.rb, line 481 def commit_changes # The application-visible sequence of events should be as follows: # # 1. Description change for the server which we are processing; # 2. Topology change, if any; # 3. Description changes for other servers, if any. # # The tricky part here is that the server description changes are # not all processed together. publish_description_change_event start_pool_if_data_bearing topology_changed_event_published = false if topology.object_id != cluster.topology.object_id || @need_topology_changed_event # We are about to publish topology changed event. # Recreate the topology instance to get its server descriptions # up to date. @topology = topology.class.new(topology.options, topology.monitoring, cluster) # This sends the SDAM event cluster.update_topology(topology) topology_changed_event_published = true @need_topology_changed_event = false end # If a server description changed, topology description change event # must be published with the previous and next topologies being of # the same type, unless we already published topology change event if topology_changed_event_published return end if updated_desc.unknown? && previous_desc.unknown? return end if updated_desc.object_id == previous_desc.object_id return end # If we are here, there has been a change in the server descriptions # in our topology, but topology class has not changed. # Publish the topology changed event and recreate the topology to # get the new list of server descriptions into it. @topology = topology.class.new(topology.options, topology.monitoring, cluster) # This sends the SDAM event cluster.update_topology(topology) end
# File lib/mongo/cluster/sdam_flow.rb, line 529 def disconnect_servers while server = @servers_to_disconnect.shift if server.connected? # Do not publish server closed event, as this was already done server.disconnect! end end end
Removes specified server from topology and warns if the topology ends up with an empty server list as a result
# File lib/mongo/cluster/sdam_flow.rb, line 419 def do_remove(address_str) servers = cluster.remove(address_str, disconnect: false) servers.each do |server| # We need to publish server closed event here, but we cannot close # the server because it could be the server owning the monitor in # whose thread this flow is presently executing, in which case closing # the server can terminate the thread and leave SDAM processing # incomplete. Thus we have to remove the server from the cluster, # publish the event, but do not call disconnect on the server until # the very end when all processing has completed. publish_sdam_event( Mongo::Monitoring::SERVER_CLOSED, Mongo::Monitoring::Event::ServerClosed.new(server.address, cluster.topology) ) end @servers_to_disconnect += servers if servers_list.empty? log_warn( "Topology now has no servers - this is likely a misconfiguration of the cluster and/or the application" ) end end
# File lib/mongo/cluster/sdam_flow.rb, line 442 def publish_description_change_event # updated_desc here may not be the description we received from # the server - in case of a stale primary, the server reported itself # as being a primary but updated_desc here will be unknown. # We do not notify on unknown -> unknown changes. # This can also be important for tests which have real i/o # happening against bogus addresses which yield unknown responses # and that also mock responses with the resulting race condition, # though tests should avoid performing real i/o with monitoring_io: false # option. if updated_desc.unknown? && previous_desc.unknown? return end # Avoid dispatching events when updated description is the same as # previous description. This allows this method to be called multiple # times in the flow when the events should be published, without # worrying about whether there are any unpublished changes. if updated_desc.object_id == previous_desc.object_id return end publish_sdam_event( ::Mongo::Monitoring::SERVER_DESCRIPTION_CHANGED, ::Mongo::Monitoring::Event::ServerDescriptionChanged.new( updated_desc.address, topology, previous_desc, updated_desc, ) ) @previous_desc = updated_desc @need_topology_changed_event = true end
Removes the server whose description we are processing from the topology.
# File lib/mongo/cluster/sdam_flow.rb, line 412 def remove publish_description_change_event do_remove(updated_desc.address.to_s) end
Removes servers from the topology which are not present in the given server description (which is supposed to have come from a good primary).
# File lib/mongo/cluster/sdam_flow.rb, line 390 def remove_servers_not_in_desc(updated_desc) updated_desc_address_strs = %w(hosts passives arbiters).map do |m| updated_desc.send(m) end.flatten servers_list.each do |server| unless updated_desc_address_strs.include?(address_str = server.address.to_s) updated_host = updated_desc.address.to_s if updated_desc.me && updated_desc.me != updated_host updated_host += " (self-identified as #{updated_desc.me})" end log_warn( "Removing server #{address_str} because it is not in hosts reported by primary " + "#{updated_host}. Reported hosts are: " + updated_desc.hosts.join(', ') ) do_remove(address_str) end end end
# File lib/mongo/cluster/sdam_flow.rb, line 76 def server_description_changed if updated_desc.me_mismatch? && updated_desc.primary? && (topology.unknown? || topology.replica_set?) then # When the driver receives a description claiming to be a primary, # we are obligated by spec tests to add and remove hosts in that # description even if it also has a me mismatch. The me mismatch # scenario though presents a number of problems: # # 1. Effectively, the server's address changes, meaning we cannot # update the description of the server whose description change we # are processing (instead servers are added and removed), but we # behave to an extent as if we are updating the description, which # causes a bunch of awkwardness. # 2. The server for which we are processing the response will be # removed from topology, which may cause the current thread to terminate # prior to running the entire sdam flow. To deal with this we separate # the removal event publication from actually removing the server # from topology, which again complicates the flow. # Primary-with-me-mismatch response could be the first one we receive # when the topology is still unknown. Change to RS without primary # in this case. if topology.unknown? @topology = Topology::ReplicaSetNoPrimary.new( topology.options.merge(replica_set_name: updated_desc.replica_set_name), topology.monitoring, self) end servers = add_servers_from_desc(updated_desc) # Spec tests require us to remove servers based on data in descrptions # with me mismatches. The driver will be more resilient if it only # removed servers from descriptions with matching mes. remove_servers_not_in_desc(updated_desc) servers.each do |server| server.start_monitoring end # The rest of sdam flow assumes the server being removed is not the one # whose description we are processing, and publishes description update # event. Since we are removing the server whose response we are # processing, do not publish description change event but mark it # published (by assigning to @previous_desc). do_remove(updated_desc.address.to_s) @previous_desc = updated_desc # We may have removed the current primary, check if there is a primary. check_if_has_primary # Publish topology change event. commit_changes disconnect_servers return end unless update_server_descriptions # All of the transitions require that server whose updated_desc we are # processing is still in the cluster (i.e., was not removed as a result # of processing another response, potentially concurrently). # If update_server_descriptions returned false we have no servers # in the topology for the description we are processing, stop. return end case topology when Topology::Single # no changes ever when Topology::Unknown if updated_desc.standalone? update_unknown_with_standalone elsif updated_desc.mongos? @topology = Topology::Sharded.new(topology.options, topology.monitoring, self) elsif updated_desc.primary? @topology = Topology::ReplicaSetWithPrimary.new( topology.options.merge(replica_set_name: updated_desc.replica_set_name), topology.monitoring, self) update_rs_from_primary elsif updated_desc.secondary? || updated_desc.arbiter? || updated_desc.other? @topology = Topology::ReplicaSetNoPrimary.new( topology.options.merge(replica_set_name: updated_desc.replica_set_name), topology.monitoring, self) update_rs_without_primary end when Topology::Sharded unless updated_desc.unknown? || updated_desc.mongos? remove end when Topology::ReplicaSetWithPrimary if updated_desc.standalone? || updated_desc.mongos? remove check_if_has_primary elsif updated_desc.primary? update_rs_from_primary elsif updated_desc.secondary? || updated_desc.arbiter? || updated_desc.other? update_rs_with_primary_from_member else check_if_has_primary end when Topology::ReplicaSetNoPrimary if updated_desc.standalone? || updated_desc.mongos? remove elsif updated_desc.primary? # Here we change topology type to RS with primary, however # while processing updated_desc we may find that its RS name # does not match our existing RS name. For this reason # is is imperative to NOT pass updated_desc's RS name to # topology constructor here. # During processing we may remove the server whose updated_desc # we are be processing (e.g. the RS name mismatch case again), # in which case topoogy type will go back to RS without primary # in the check_if_has_primary step. @topology = Topology::ReplicaSetWithPrimary.new( # Do not pass updated_desc's RS name here topology.options, topology.monitoring, self) update_rs_from_primary elsif updated_desc.secondary? || updated_desc.arbiter? || updated_desc.other? update_rs_without_primary end else raise ArgumentError, "Unknown topology #{topology.class}" end commit_changes disconnect_servers end
Whether updated_desc
is for a stale primary.
# File lib/mongo/cluster/sdam_flow.rb, line 569 def stale_primary? if updated_desc.election_id && updated_desc.set_version if topology.max_set_version && topology.max_election_id && (updated_desc.set_version < topology.max_set_version || (updated_desc.set_version == topology.max_set_version && updated_desc.election_id < topology.max_election_id)) return true end end false end
If the server being processed is identified as data bearing, creates the server's connection pool so it can start populating
# File lib/mongo/cluster/sdam_flow.rb, line 540 def start_pool_if_data_bearing return if !updated_desc.data_bearing? servers_list.each do |server| if server.address == @updated_desc.address server.pool end end end
Updates topology which must be a ReplicaSetWithPrimary with information from the primary's server description.
This method does not change topology type to ReplicaSetWithPrimary - this needs to have been done prior to calling this method.
If the primary whose description is being processed is determined to be stale, this method will change the server description and topology type to unknown.
# File lib/mongo/cluster/sdam_flow.rb, line 226 def update_rs_from_primary if topology.replica_set_name.nil? @topology = Topology::ReplicaSetWithPrimary.new( topology.options.merge(replica_set_name: updated_desc.replica_set_name), topology.monitoring, self) end if topology.replica_set_name != updated_desc.replica_set_name log_warn( "Removing server #{updated_desc.address.to_s} because it has an " + "incorrect replica set name (#{updated_desc.replica_set_name}); " + "current set name is #{topology.replica_set_name}" ) remove check_if_has_primary return end if stale_primary? @updated_desc = ::Mongo::Server::Description.new(updated_desc.address, {}, updated_desc.average_round_trip_time) update_server_descriptions check_if_has_primary return end max_election_id = topology.new_max_election_id(updated_desc) max_set_version = topology.new_max_set_version(updated_desc) if max_election_id != topology.max_election_id || max_set_version != topology.max_set_version then @topology = Topology::ReplicaSetWithPrimary.new( topology.options.merge( max_election_id: max_election_id, max_set_version: max_set_version ), topology.monitoring, self) end # At this point we have accepted the updated server description # and the topology (both are primary). Commit these changes so that # their respective SDAM events are published before SDAM events for # server additions/removals that follow publish_description_change_event servers_list.each do |server| if server.address != updated_desc.address if server.primary? server.update_description(::Mongo::Server::Description.new( server.address, {}, server.description.average_round_trip_time)) end end end servers = add_servers_from_desc(updated_desc) remove_servers_not_in_desc(updated_desc) check_if_has_primary servers.each do |server| server.start_monitoring end end
Updates a ReplicaSetWithPrimary topology from a non-primary member.
# File lib/mongo/cluster/sdam_flow.rb, line 291 def update_rs_with_primary_from_member if topology.replica_set_name != updated_desc.replica_set_name log_warn( "Removing server #{updated_desc.address.to_s} because it has an " + "incorrect replica set name (#{updated_desc.replica_set_name}); " + "current set name is #{topology.replica_set_name}" ) remove check_if_has_primary return end if updated_desc.me_mismatch? log_warn( "Removing server #{updated_desc.address.to_s} because it " + "reported itself as #{updated_desc.me}" ) remove check_if_has_primary return end have_primary = false servers_list.each do |server| if server.primary? have_primary = true break end end unless have_primary @topology = Topology::ReplicaSetNoPrimary.new( topology.options, topology.monitoring, self) end end
Updates a ReplicaSetNoPrimary topology from a non-primary member.
# File lib/mongo/cluster/sdam_flow.rb, line 328 def update_rs_without_primary if topology.replica_set_name.nil? @topology = Topology::ReplicaSetNoPrimary.new( topology.options.merge(replica_set_name: updated_desc.replica_set_name), topology.monitoring, self) end if topology.replica_set_name != updated_desc.replica_set_name log_warn( "Removing server #{updated_desc.address.to_s} because it has an " + "incorrect replica set name (#{updated_desc.replica_set_name}); " + "current set name is #{topology.replica_set_name}" ) remove return end publish_description_change_event servers = add_servers_from_desc(updated_desc) commit_changes servers.each do |server| server.start_monitoring end if updated_desc.me_mismatch? log_warn( "Removing server #{updated_desc.address.to_s} because it " + "reported itself as #{updated_desc.me}" ) remove return end end
Updates descriptions on all servers whose address matches updated_desc
's address.
# File lib/mongo/cluster/sdam_flow.rb, line 58 def update_server_descriptions servers_list.each do |server| if server.address == updated_desc.address changed = server.description != updated_desc # Always update server description, so that fields that do not # affect description equality comparisons but are part of the # description are updated. server.update_description(updated_desc) server.update_last_scan # But return if there was a content difference between # descriptions, and if there wasn't we'll skip the remainder of # sdam flow return changed end end false end
Transitions from unknown to single topology type, when a standalone server is discovered.
# File lib/mongo/cluster/sdam_flow.rb, line 205 def update_unknown_with_standalone if seeds.length == 1 @topology = Topology::Single.new( topology.options, topology.monitoring, self) else log_warn( "Removing server #{updated_desc.address.to_s} because it is a standalone and we have multiple seeds (#{seeds.length})" ) remove end end