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

cluster[R]
previous_desc[R]
topology[R]

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.

updated_desc[R]

Public Class Methods

new(cluster, previous_desc, updated_desc) click to toggle source
# File lib/mongo/cluster/sdam_flow.rb, line 28
def initialize(cluster, previous_desc, updated_desc)
  @cluster = cluster
  @topology = cluster.topology
  @previous_desc = previous_desc
  @updated_desc = updated_desc
end

Public Instance Methods

add_servers_from_desc(updated_desc) click to toggle source

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 315
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
check_if_has_primary() click to toggle source

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 453
def check_if_has_primary
  unless topology.replica_set?
    raise ArgumentError, 'check_if_has_primary should only be called when topology is replica set'
  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
commit_changes() click to toggle source

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 403
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

  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
do_remove(address_str) click to toggle source

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 355
def do_remove(address_str)
  cluster.remove(address_str)
  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
publish_description_change_event() click to toggle source
# File lib/mongo/cluster/sdam_flow.rb, line 364
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
remove() click to toggle source

Removes the server whose description we are processing from the topology.

# File lib/mongo/cluster/sdam_flow.rb, line 348
def remove
  publish_description_change_event
  do_remove(updated_desc.address.to_s)
end
remove_servers_not_in_desc(updated_desc) click to toggle source

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 331
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)
      log_warn(
        "Removing server #{address_str} because it is not in hosts reported by primary " +
        "#{updated_desc.address}"
      )
      do_remove(address_str)
    end
  end
end
server_description_changed() click to toggle source
# File lib/mongo/cluster/sdam_flow.rb, line 72
def server_description_changed
  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
end
stale_primary?() click to toggle source

Whether updated_desc is for a stale primary.

# File lib/mongo/cluster/sdam_flow.rb, line 469
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
update_rs_from_primary() click to toggle source

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 167
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
update_rs_with_primary_from_member() click to toggle source

Updates a ReplicaSetWithPrimary topology from a non-primary member.

# File lib/mongo/cluster/sdam_flow.rb, line 232
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
update_rs_without_primary() click to toggle source

Updates a ReplicaSetNoPrimary topology from a non-primary member.

# File lib/mongo/cluster/sdam_flow.rb, line 269
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
update_server_descriptions() click to toggle source

Updates descriptions on all servers whose address matches updated_desc's address.

# File lib/mongo/cluster/sdam_flow.rb, line 56
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 like
      # last_update_time reflect the last ismaster response
      server.update_description(updated_desc)
      # 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
update_unknown_with_standalone() click to toggle source

Transitions from unknown to single topology type, when a standalone server is discovered.

# File lib/mongo/cluster/sdam_flow.rb, line 146
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