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]
original_desc[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
  @original_desc = @previous_desc = previous_desc
  @updated_desc = updated_desc
  @servers_to_disconnect = []
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 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
became_unknown?() click to toggle source

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
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 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
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 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
disconnect_servers() click to toggle source
# 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
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 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
publish_description_change_event() click to toggle source
# 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
remove() click to toggle source

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
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 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
server_description_changed() click to toggle source
# 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
stale_primary?() click to toggle source

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

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