Class: Mongo::Cluster::SdamFlow Private

# File 'lib/mongo/cluster/sdam_flow.rb', line 611

def became_unknown?
  updated_desc.unknown? && !original_desc.unknown?
end

# File 'lib/mongo/cluster/sdam_flow.rb', line 39

attr_reader :cluster, :previous_desc, :updated_desc, :original_desc

# File 'lib/mongo/cluster/sdam_flow.rb', line 39

attr_reader :cluster, :previous_desc, :updated_desc, :original_desc

# File 'lib/mongo/cluster/sdam_flow.rb', line 39

attr_reader :cluster, :previous_desc, :updated_desc, :original_desc

# File 'lib/mongo/cluster/sdam_flow.rb', line 584

def stale_primary?
  if updated_desc.max_wire_version >= 17
    return true if updated_desc.election_id.nil? && !topology.max_election_id.nil?
    if updated_desc.election_id && topology.max_election_id && updated_desc.election_id < topology.max_election_id
      return true
    end

    if updated_desc.election_id == topology.max_election_id
      return true if updated_desc.set_version.nil? && !topology.max_set_version.nil?
      if updated_desc.set_version && topology.max_set_version && updated_desc.set_version < topology.max_set_version
        return true
      end
    end
  elsif 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

# File 'lib/mongo/cluster/sdam_flow.rb', line 51

attr_reader :topology

# File 'lib/mongo/cluster/sdam_flow.rb', line 624

def topology_effectively_changed?
  return true unless topology.equal?(cluster.topology)

  server_descriptions = servers_list.map do |server|
    [ server.address.to_s, server.description ]
  end

  @previous_server_descriptions != server_descriptions
end

# File 'lib/mongo/cluster/sdam_flow.rb', line 39

attr_reader :cluster, :previous_desc, :updated_desc, :original_desc

# File 'lib/mongo/cluster/sdam_flow.rb', line 565

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
  return if primary

  @topology = Topology::ReplicaSetNoPrimary.new(
    topology.options, topology.monitoring, self
  )
end

# File 'lib/mongo/cluster/sdam_flow.rb', line 499

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.equal?(cluster.topology) || @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
  return if topology_changed_event_published

  return if updated_desc.unknown? && previous_desc.unknown?
  return if updated_desc.equal?(previous_desc)

  return unless topology_effectively_changed?

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

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

# File 'lib/mongo/cluster/sdam_flow.rb', line 430

def do_remove(address_str)
  servers = cluster.remove(address_str, disconnect: false)
  servers.each do |server|
    # Clear the description so that the server is marked as unknown.
    server.clear_description

    # 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
  return unless servers_list.empty?

  log_warn(
    'Topology now has no servers - this is likely a misconfiguration of the cluster and/or the application'
  )
end

# File 'lib/mongo/cluster/sdam_flow.rb', line 456

def publish_description_change_event
  # This method may be invoked when server description definitely changed
  # but prior to the topology getting updated. Therefore we check both
  # server description changes and overall topology changes. When this
  # method is called at the end of SDAM flow as part of "commit changes"
  # step, server description change is incorporated into the topology
  # change.
  return unless @server_description_changed || topology_effectively_changed?

  # 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 used to not notify on Unknown -> Unknown server changes.
  # Technically these are valid state changes (or at least as valid as
  # other server description changes when the description has not
  # changed meaningfully but the events are still published).
  # The current version of the driver notifies on Unknown -> Unknown
  # transitions.

  # 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.
  return if updated_desc.equal?(previous_desc)

  publish_sdam_event(
    ::Mongo::Monitoring::SERVER_DESCRIPTION_CHANGED,
    ::Mongo::Monitoring::Event::ServerDescriptionChanged.new(
      updated_desc.address,
      topology,
      previous_desc,
      updated_desc,
      awaited: awaited?
    )
  )
  @previous_desc = updated_desc
  @need_topology_changed_event = true
end

# File 'lib/mongo/cluster/sdam_flow.rb', line 423

def remove
  publish_description_change_event
  do_remove(updated_desc.address.to_s)
end

# File 'lib/mongo/cluster/sdam_flow.rb', line 403

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|
    next if updated_desc_address_strs.include?(address_str = server.address.to_s)

    updated_host = updated_desc.address.to_s
    updated_host += " (self-identified as #{updated_desc.me})" if updated_desc.me && updated_desc.me != updated_host
    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

# File 'lib/mongo/cluster/sdam_flow.rb', line 85

def server_description_changed
  @previous_server_descriptions = servers_list.map do |server|
    [ server.address.to_s, server.description ]
  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::LoadBalanced
    @updated_desc = ::Mongo::Server::Description::LoadBalancer.new(
      updated_desc.address
    )
    update_server_descriptions
  when Topology::Single
    if topology.replica_set_name && (updated_desc.replica_set_name != topology.replica_set_name)
      log_warn(
        "Server #{updated_desc.address} has an incorrect replica set name '#{updated_desc.replica_set_name}'; expected '#{topology.replica_set_name}'"
      )
      @updated_desc = ::Mongo::Server::Description.new(
        updated_desc.address,
        {},
        average_round_trip_time: updated_desc.average_round_trip_time,
        minimum_round_trip_time: updated_desc.minimum_round_trip_time
      )
      update_server_descriptions
    end
  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?
      log_warn(
        "Removing server #{updated_desc.address} because it is of the wrong type (#{updated_desc.server_type.to_s.upcase}) - expected SHARDED"
      )
      remove
    end
  when Topology::ReplicaSetWithPrimary
    if updated_desc.standalone? || updated_desc.mongos?
      log_warn(
        "Removing server #{updated_desc.address} because it is of the wrong type (#{updated_desc.server_type.to_s.upcase}) - expected a replica set member"
      )
      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?
      log_warn(
        "Removing server #{updated_desc.address} because it is of the wrong type (#{updated_desc.server_type.to_s.upcase}) - expected a replica set member"
      )
      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

  verify_invariants
  commit_changes
  disconnect_servers
end

# File 'lib/mongo/cluster/sdam_flow.rb', line 554

def start_pool_if_data_bearing
  return unless updated_desc.data_bearing?

  servers_list.each do |server|
    server.pool if server.address == @updated_desc.address
  end
end

# File 'lib/mongo/cluster/sdam_flow.rb', line 215

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} because it has an " +
      "incorrect replica set name '#{updated_desc.replica_set_name}'; " +
      "expected '#{topology.replica_set_name}'"
    )
    remove
    check_if_has_primary
    return
  end

  if stale_primary?
    @updated_desc = ::Mongo::Server::Description.new(
      updated_desc.address,
      {},
      average_round_trip_time: updated_desc.average_round_trip_time,
      minimum_round_trip_time: updated_desc.minimum_round_trip_time
    )
    update_server_descriptions
    check_if_has_primary
    return
  end

  if updated_desc.max_wire_version >= 17
    @topology = Topology::ReplicaSetWithPrimary.new(
      topology.options.merge(
        max_election_id: updated_desc.election_id,
        max_set_version: updated_desc.set_version
      ), topology.monitoring, self
    )
  else
    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
      @topology = Topology::ReplicaSetWithPrimary.new(
        topology.options.merge(
          max_election_id: max_election_id,
          max_set_version: max_set_version
        ), topology.monitoring, self
      )
    end
  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|
    next unless server.address != updated_desc.address

    next unless server.primary?

    server.update_description(
      ::Mongo::Server::Description.new(
        server.address,
        {},
        average_round_trip_time: server.description.average_round_trip_time,
        minimum_round_trip_time: updated_desc.minimum_round_trip_time
      )
    )
  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

# File 'lib/mongo/cluster/sdam_flow.rb', line 300

def update_rs_with_primary_from_member
  if topology.replica_set_name != updated_desc.replica_set_name
    log_warn(
      "Removing server #{updated_desc.address} 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} 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

  return if have_primary

  @topology = Topology::ReplicaSetNoPrimary.new(
    topology.options, topology.monitoring, self
  )
end

# File 'lib/mongo/cluster/sdam_flow.rb', line 338

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

  return unless updated_desc.me_mismatch?

  log_warn(
    "Removing server #{updated_desc.address} because it " +
    "reported itself as #{updated_desc.me}"
  )
  remove
  nil
end

# File 'lib/mongo/cluster/sdam_flow.rb', line 61

def update_server_descriptions
  servers_list.each do |server|
    next unless server.address == updated_desc.address
    # SDAM flow must be run when topology version in the new description
    # is equal to the current topology version, per the example in
    # https://github.com/mongodb/specifications/blob/master/source/server-discovery-and-monitoring/server-discovery-and-monitoring.md#what-is-the-purpose-of-topologyversion
    return false unless updated_desc.topology_version_gte?(server.description)

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

    # If there was no content difference between descriptions, we
    # still need to run sdam flow, but if the flow produces no change
    # in topology we will omit sending events.
    return true
  end
  false
end

# File 'lib/mongo/cluster/sdam_flow.rb', line 193

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} because it is a standalone and we have multiple seeds (#{seeds.length})"
    )
    remove
  end
end

# File 'lib/mongo/cluster/sdam_flow.rb', line 634

def verify_invariants
  return unless Mongo::Lint.enabled?
  return unless cluster.topology.single?
  return unless cluster.servers_list.length > 1

  raise Mongo::Error::LintError,
        "Trying to create a single topology with multiple servers: #{cluster.servers_list}"
end