Class: Mongo::Cluster

Inherits:
Object
  • Object
show all
Extended by:
Forwardable
Includes:
Event::Subscriber, Loggable, Monitoring::Publishable
Defined in:
lib/mongo/cluster.rb,
lib/mongo/cluster/topology.rb,
lib/mongo/cluster/topology.rb,
lib/mongo/cluster/sdam_flow.rb,
lib/mongo/cluster/topology/base.rb,
lib/mongo/cluster/topology/single.rb,
lib/mongo/cluster/topology/sharded.rb,
lib/mongo/cluster/topology/unknown.rb,
lib/mongo/cluster/periodic_executor.rb,
lib/mongo/cluster/reapers/cursor_reaper.rb,
lib/mongo/cluster/reapers/socket_reaper.rb,
lib/mongo/cluster/topology/no_replica_set_options.rb,
lib/mongo/cluster/topology/replica_set_no_primary.rb,
lib/mongo/cluster/topology/replica_set_with_primary.rb

Overview

Copyright © 2018-2019 MongoDB, Inc.

Licensed under the Apache License, Version 2.0 (the “License”); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an “AS IS” BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Defined Under Namespace

Modules: Topology Classes: CursorReaper, PeriodicExecutor, SdamFlow, SocketReaper

Constant Summary collapse

MAX_READ_RETRIES =

The default number of mongos read retries.

Since:

  • 2.1.1

1
MAX_WRITE_RETRIES =

The default number of mongos write retries.

Since:

  • 2.4.2

1
READ_RETRY_INTERVAL =

The default mongos read retry interval, in seconds.

Since:

  • 2.1.1

5
IDLE_WRITE_PERIOD_SECONDS =

How often an idle primary writes a no-op to the oplog.

Since:

  • 2.4.0

10
CLUSTER_TIME =

The cluster time key in responses from mongos servers.

Since:

  • 2.5.0

'clusterTime'.freeze

Constants included from Loggable

Loggable::PREFIX

Instance Attribute Summary collapse

Attributes included from Event::Subscriber

#event_listeners

Class Method Summary collapse

Instance Method Summary collapse

Methods included from Loggable

#log_debug, #log_error, #log_fatal, #log_info, #log_warn, #logger

Methods included from Event::Subscriber

#subscribe_to

Methods included from Monitoring::Publishable

#publish_event, #publish_sdam_event

Constructor Details

#initialize(seeds, monitoring, options = Options::Redacted.new) ⇒ Cluster

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Note:

Cluster should never be directly instantiated outside of a Client.

Note:

When connecting to a mongodb+srv:// URI, the client expands such a URI into a list of servers and passes that list to the Cluster constructor. When connecting to a standalone mongod, the Cluster constructor receives the corresponding address as an array of one string.

Instantiate the new cluster.

Examples:

Instantiate the cluster.

Mongo::Cluster.new(["127.0.0.1:27017"], monitoring)

Parameters:

  • seeds (Array<String>) —

    The addresses of the configured servers

  • monitoring (Monitoring) —

    The monitoring.

  • options (Hash) (defaults to: Options::Redacted.new) —

    Options. Client constructor forwards its options to Cluster constructor, although Cluster recognizes only a subset of the options recognized by Client.

Options Hash (options):

  • :scan (true, false) —

    Whether to scan all seeds in constructor. The default in driver version 2.x is to do so; driver version 3.x will not scan seeds in constructor. Opt in to the new behavior by setting this option to false. Note: setting this option to nil enables scanning seeds in constructor in driver version 2.x. Driver version 3.x will recognize this option but will ignore it and will never scan seeds in the constructor.

  • :monitoring_io (true, false) —

    For internal driver use only. Set to false to prevent SDAM-related I/O from being done by this cluster or servers under it. Note: setting this option to false will make the cluster non-functional. It is intended for use in tests which manually invoke SDAM state transitions.

Since:

  • 2.0.0



90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
 
# File 'lib/mongo/cluster.rb', line 90

def initialize(seeds, monitoring, options = Options::Redacted.new)
  if options[:monitoring_io] != false && !options[:server_selection_semaphore]
    raise ArgumentError, 'Need server selection semaphore'
  end

  @servers = []
  @monitoring = monitoring
  @event_listeners = Event::Listeners.new
  @options = options.freeze
  @app_metadata = Server::AppMetadata.new(@options)
  @update_lock = Mutex.new
  @sdam_flow_lock = Mutex.new
  @pool_lock = Mutex.new
  @cluster_time = nil
  @cluster_time_lock = Mutex.new
  @topology = Topology.initial(self, monitoring, options)
  Session::SessionPool.create(self)

  # The opening topology is always unknown with no servers.
  # https://github.com/mongodb/specifications/pull/388
  opening_topology = Topology::Unknown.new(options, monitoring, self)

  publish_sdam_event(
    Monitoring::TOPOLOGY_OPENING,
    Monitoring::Event::TopologyOpening.new(opening_topology)
  )

  subscribe_to(Event::DESCRIPTION_CHANGED, Event::DescriptionChanged.new(self))

  @seeds = seeds
  servers = seeds.map do |seed|
    # Server opening events must be sent after topology change events.
    # Therefore separate server addition, done here before topoolgy change
    # event is published, from starting to monitor the server which is
    # done later.
    add(seed, monitor: false)
  end

  if seeds.size >= 1
    # Recreate the topology to get the current server list into it
    @topology = topology.class.new(topology.options, topology.monitoring, self)
    publish_sdam_event(
      Monitoring::TOPOLOGY_CHANGED,
      Monitoring::Event::TopologyChanged.new(opening_topology, @topology)
    )
  end

  servers.each do |server|
    server.start_monitoring
  end

  if options[:monitoring_io] == false
    # Omit periodic executor construction, because without servers
    # no commands can be sent to the cluster and there shouldn't ever
    # be anything that needs to be cleaned up.
    #
    # Also omit legacy single round of SDAM on the main thread,
    # as it would race with tests that mock SDAM responses.
    return
  end

  @cursor_reaper = CursorReaper.new
  @socket_reaper = SocketReaper.new(self)
  @periodic_executor = PeriodicExecutor.new(@cursor_reaper, @socket_reaper)
  @periodic_executor.run!

  ObjectSpace.define_finalizer(self, self.class.finalize(pools, @periodic_executor, @session_pool))

  @connecting = false
  @connected = true

  if options[:scan] != false
    server_selection_timeout = options[:server_selection_timeout] || ServerSelector::SERVER_SELECTION_TIMEOUT
    # The server selection timeout can be very short especially in
    # tests, when the client waits for a synchronous scan before
    # starting server selection. Limiting the scan to server selection time
    # then aborts the scan before it can process even local servers.
    # Therefore, allow at least 3 seconds for the scan here.
    if server_selection_timeout < 3
      server_selection_timeout = 3
    end
    start_time = Time.now
    deadline = start_time + server_selection_timeout
    # Wait for the first scan of each server to complete, for
    # backwards compatibility.
    # If any servers are discovered during this SDAM round we are going to
    # wait for these servers to also be queried, and so on, up to the
    # server selection timeout or the 3 second minimum.
    loop do
      servers = servers_list.dup
      if servers.all? { |server| server.description.last_update_time >= start_time }
        break
      end
      if (time_remaining = deadline - Time.now) <= 0
        break
      end
      options[:server_selection_semaphore].wait(time_remaining)
    end
  end
end

Instance Attribute Details

#app_metadata ⇒ Mongo::Server::AppMetadata (readonly)

Returns The application metadata, used for connection handshakes.

Returns:

Since:

  • 2.4.0



226
227
228
 
# File 'lib/mongo/cluster.rb', line 226

def app_metadata
  @app_metadata
end

#cluster_time ⇒ BSON::Document (readonly)

Returns The latest cluster time seen.

Returns:

  • (BSON::Document) —

    The latest cluster time seen.

Since:

  • 2.5.0



231
232
233
 
# File 'lib/mongo/cluster.rb', line 231

def cluster_time
  @cluster_time
end

#monitoring ⇒ Monitoring (readonly)

Returns monitoring The monitoring.

Returns:

Since:

  • 2.0.0



217
218
219
 
# File 'lib/mongo/cluster.rb', line 217

def monitoring
  @monitoring
end

#options ⇒ Hash (readonly)

Returns The options hash.

Returns:

  • (Hash) —

    The options hash.

Since:

  • 2.0.0



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

def options
  @options
end

#sdam_flow_lock ⇒ Object (readonly)

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Since:

  • 2.0.0



635
636
637
 
# File 'lib/mongo/cluster.rb', line 635

def sdam_flow_lock
  @sdam_flow_lock
end

#seeds ⇒ Array<String> (readonly)

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns The addresses of seed servers. Contains addresses that were given to Cluster when it was instantiated, not current addresses that the cluster is using as a result of SDAM.

Returns:

  • (Array<String>) —

    The addresses of seed servers. Contains addresses that were given to Cluster when it was instantiated, not current addresses that the cluster is using as a result of SDAM.

Since:

  • 2.7.0



239
240
241
 
# File 'lib/mongo/cluster.rb', line 239

def seeds
  @seeds
end

#server_selection_semaphore ⇒ Object (readonly)

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Since:

  • 2.0.0



344
345
346
 
# File 'lib/mongo/cluster.rb', line 344

def server_selection_semaphore
  @server_selection_semaphore
end

#session_pool ⇒ Object (readonly)

Since:

  • 2.5.1



244
245
246
 
# File 'lib/mongo/cluster.rb', line 244

def session_pool
  @session_pool
end

#topology ⇒ Integer? (readonly)

The logical session timeout value in minutes.

Examples:

Get the logical session timeout in minutes.

cluster.logical_session_timeout

Returns:

  • (Integer, nil) —

    The logical session timeout.

Since:

  • 2.5.0



220
221
222
 
# File 'lib/mongo/cluster.rb', line 220

def topology
  @topology
end

Class Method Details

.create(client) ⇒ Cluster

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Create a cluster for the provided client, for use when we don’t want the client’s original cluster instance to be the same.

Examples:

Create a cluster for the client.

Cluster.create(client)

Parameters:

  • client (Client) —

    The client to create on.

Returns:

Since:

  • 2.0.0



204
205
206
207
208
209
210
211
 
# File 'lib/mongo/cluster.rb', line 204

def self.create(client)
  cluster = Cluster.new(
    client.cluster.addresses.map(&:to_s),
    Monitoring.new,
    client.cluster_options,
  )
  client.instance_variable_set(:@cluster, cluster)
end

.finalize(pools, periodic_executor, session_pool) ⇒ Proc

Finalize the cluster for garbage collection. Disconnects all the scoped connection pools.

Examples:

Finalize the cluster.

Cluster.finalize(pools)

Parameters:

Returns:

  • (Proc) —

    The Finalizer.

Since:

  • 2.2.0



359
360
361
362
363
364
365
366
367
 
# File 'lib/mongo/cluster.rb', line 359

def self.finalize(pools, periodic_executor, session_pool)
  proc do
    session_pool.end_sessions
    periodic_executor.stop!
    pools.values.each do |pool|
      pool.disconnect!
    end
  end
end

Instance Method Details

#==(other) ⇒ true, false

Determine if this cluster of servers is equal to another object. Checks the servers currently in the cluster, not what was configured.

Examples:

Is the cluster equal to the object?

cluster == other

Parameters:

  • other (Object) —

    The object to compare to.

Returns:

  • (true, false) —

    If the objects are equal.

Since:

  • 2.0.0



474
475
476
477
478
479
 
# File 'lib/mongo/cluster.rb', line 474

def ==(other)
  return false unless other.is_a?(Cluster)
  addresses == other.addresses &&
    options.merge(server_selection_semaphore: nil) ==
      other.options.merge(server_selection_semaphore: nil)
end

#add(host, add_options = nil) ⇒ Server

Add a server to the cluster with the provided address. Useful in auto-discovery of new servers when an existing server executes an ismaster and potentially non-configured servers were included.

Examples:

Add the server for the address to the cluster.

cluster.add('127.0.0.1:27018')

Parameters:

  • host (String) —

    The address of the server to add.

  • options (Hash) —

    a customizable set of options

Returns:

  • (Server) —

    The newly added server, if not present already.

Since:

  • 2.0.0



579
580
581
582
583
584
585
586
587
588
589
590
 
# File 'lib/mongo/cluster.rb', line 579

def add(host, add_options=nil)
  address = Address.new(host, options)
  if !addresses.include?(address)
    server = Server.new(address, self, @monitoring, event_listeners, options.merge(
      monitor: false))
    @update_lock.synchronize { @servers.push(server) }
    if add_options.nil? || add_options[:monitor] != false
      server.start_monitoring
    end
    server
  end
end

#addresses ⇒ Array<Mongo::Address>

The addresses in the cluster.

Examples:

Get the addresses in the cluster.

cluster.addresses

Returns:

Since:

  • 2.0.6



307
308
309
 
# File 'lib/mongo/cluster.rb', line 307

def addresses
  servers_list.map(&:address).dup
end

#connected? ⇒ true|false

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Whether the cluster object is connected to its cluster.

Returns:

  • (true|false) —

    Whether the cluster is connected.

Since:

  • 2.7.0



282
283
284
 
# File 'lib/mongo/cluster.rb', line 282

def connected?
  !!@connected
end

#disconnect!(wait = false) ⇒ true

Note:

Applications should call Client#close to disconnect from

Disconnect all servers.

the cluster rather than calling this method. This method is for internal driver use only.

Examples:

Disconnect the cluster’s servers.

cluster.disconnect!

Parameters:

  • wait (Boolean) (defaults to: false) —

    Whether to wait for background threads to finish running.

Returns:

  • (true) —

    Always true.

Since:

  • 2.1.0



384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
 
# File 'lib/mongo/cluster.rb', line 384

def disconnect!(wait=false)
  unless @connecting || @connected
    return true
  end
  @periodic_executor.stop!
  @servers.each do |server|
    if server.connected?
      server.disconnect!(wait)
      publish_sdam_event(
        Monitoring::SERVER_CLOSED,
        Monitoring::Event::ServerClosed.new(server.address, topology)
      )
    end
  end
  publish_sdam_event(
    Monitoring::TOPOLOGY_CLOSED,
    Monitoring::Event::TopologyClosed.new(topology)
  )
  @connecting = @connected = false
  true
end

#has_readable_server?(server_selector = nil) ⇒ true, false

Determine if the cluster would select a readable server for the provided read preference.

Examples:

Is a readable server present?

topology.has_readable_server?(server_selector)

Parameters:

  • server_selector (ServerSelector) (defaults to: nil) —

    The server selector.

Returns:

  • (true, false) —

    If a readable server is present.

Since:

  • 2.4.0



493
494
495
 
# File 'lib/mongo/cluster.rb', line 493

def has_readable_server?(server_selector = nil)
  topology.has_readable_server?(self, server_selector)
end

#has_writable_server? ⇒ true, false

Determine if the cluster would select a writable server.

Examples:

Is a writable server present?

topology.has_writable_server?

Returns:

  • (true, false) —

    If a writable server is present.

Since:

  • 2.4.0



505
506
507
 
# File 'lib/mongo/cluster.rb', line 505

def has_writable_server?
  topology.has_writable_server?(self)
end

#inspect ⇒ String

Get the nicer formatted string for use in inspection.

Examples:

Inspect the cluster.

cluster.inspect

Returns:

  • (String) —

    The cluster inspection.

Since:

  • 2.0.0



329
330
331
 
# File 'lib/mongo/cluster.rb', line 329

def inspect
  "#<Mongo::Cluster:0x#{object_id} servers=#{servers} topology=#{topology.summary}>"
end

#max_read_retries ⇒ Integer

Get the maximum number of times the cluster can retry a read operation on a mongos.

Examples:

Get the max read retries.

cluster.max_read_retries

Returns:

  • (Integer) —

    The maximum retries.

Since:

  • 2.1.1



259
260
261
 
# File 'lib/mongo/cluster.rb', line 259

def max_read_retries
  options[:max_read_retries] || MAX_READ_RETRIES
end

#next_primary(ping = true) ⇒ Mongo::Server

Get the next primary server we can send an operation to.

Examples:

Get the next primary server.

cluster.next_primary

Parameters:

  • ping (true, false) (defaults to: true) —

    Whether to ping the server before selection. Deprecated, not necessary with the implementation of the Server Selection specification.

Returns:

Since:

  • 2.0.0



521
522
523
524
 
# File 'lib/mongo/cluster.rb', line 521

def next_primary(ping = true)
  @primary_selector ||= ServerSelector.get(ServerSelector::PRIMARY)
  @primary_selector.select_server(self)
end

#pool(server) ⇒ Server::ConnectionPool

Get the scoped connection pool for the server.

Examples:

Get the connection pool.

cluster.pool(server)

Parameters:

  • server (Server) —

    The server.

Returns:

Since:

  • 2.2.0



536
537
538
539
540
 
# File 'lib/mongo/cluster.rb', line 536

def pool(server)
  @pool_lock.synchronize do
    pools[server.address] ||= Server::ConnectionPool.get(server)
  end
end

#read_retry_interval ⇒ Float

Get the interval, in seconds, in which a mongos read operation is retried.

Examples:

Get the read retry interval.

cluster.read_retry_interval

Returns:

  • (Float) —

    The interval.

Since:

  • 2.1.1



272
273
274
 
# File 'lib/mongo/cluster.rb', line 272

def read_retry_interval
  options[:read_retry_interval] || READ_RETRY_INTERVAL
end

#reconnect! ⇒ true

Deprecated.

Use Client#reconnect to reconnect to the cluster instead of calling this method. This method does not send SDAM events.

Reconnect all servers.

Examples:

Reconnect the cluster’s servers.

cluster.reconnect!

Returns:

  • (true) —

    Always true.

Since:

  • 2.1.0



416
417
418
419
420
421
422
423
424
425
 
# File 'lib/mongo/cluster.rb', line 416

def reconnect!
  @connecting = true
  scan!
  servers.each do |server|
    server.reconnect!
  end
  @periodic_executor.restart!
  @connecting = false
  @connected = true
end

#remove(host) ⇒ true|false

Remove the server from the cluster for the provided address, if it exists.

Examples:

Remove the server from the cluster.

server.remove('127.0.0.1:27017')

Parameters:

  • host (String) —

    The host/port or socket address.

Returns:

  • (true|false) —

    Whether any servers were removed.

Since:

  • 2.0.0, return value added in 2.7.0



603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
 
# File 'lib/mongo/cluster.rb', line 603

def remove(host)
  address = Address.new(host)
  removed_servers = @servers.select { |s| s.address == address }
  @update_lock.synchronize { @servers = @servers - removed_servers }
  removed_servers.each do |server|
    if server.connected?
      server.disconnect!
      publish_sdam_event(
        Monitoring::SERVER_CLOSED,
        Monitoring::Event::ServerClosed.new(address, topology)
      )
    end
  end
  removed_servers.any?
end

#scan!(sync = true) ⇒ true

Note:

In both synchronous and asynchronous scans, each monitor thread maintains a minimum interval between scans, meaning calling this method may not initiate a scan on a particular server the very next instant.

Force a scan of all known servers in the cluster.

If the sync parameter is true which is the default, the scan is performed synchronously in the thread which called this method. Each server in the cluster is checked sequentially. If there are many servers in the cluster or they are slow to respond, this can be a long running operation.

If the sync parameter is false, this method instructs all server monitor threads to perform an immediate scan and returns without waiting for scan results.

Examples:

Force a full cluster scan.

cluster.scan!

Returns:

  • (true) —

    Always true.

Since:

  • 2.0.0



450
451
452
453
454
455
456
457
458
459
460
461
 
# File 'lib/mongo/cluster.rb', line 450

def scan!(sync=true)
  if sync
    servers_list.each do |server|
      server.scan!
    end
  else
    servers_list.each do |server|
      server.monitor.scan_semaphore.signal
    end
  end
  true
end

#servers ⇒ Array<Server>

Get a list of server candidates from the cluster that can have operations executed on them.

Examples:

Get the server candidates for an operation.

cluster.servers

Returns:

  • (Array<Server>) —

    The candidate servers.

Since:

  • 2.0.0



295
296
297
 
# File 'lib/mongo/cluster.rb', line 295

def servers
  topology.servers(servers_list.compact).compact
end

#servers_list ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Since:

  • 2.0.0



630
631
632
 
# File 'lib/mongo/cluster.rb', line 630

def servers_list
  @update_lock.synchronize { @servers.dup }
end

#summary ⇒ Object

Note:

This method is experimental and subject to change.

Since:

  • 2.7.0



337
338
339
340
341
 
# File 'lib/mongo/cluster.rb', line 337

def summary
  "#<Cluster " +
  "topology=#{topology.summary} "+
  "servers=[#{servers.map(&:summary).join(',')}]>"
end

#update_cluster_time(result) ⇒ Object

Update the max cluster time seen in a response.

Examples:

Update the cluster time.

cluster.update_cluster_time(result)

Parameters:

Returns:

  • (Object) —

    The cluster time.

Since:

  • 2.5.0



552
553
554
555
556
557
558
559
560
561
562
 
# File 'lib/mongo/cluster.rb', line 552

def update_cluster_time(result)
  if cluster_time_doc = result.cluster_time
    @cluster_time_lock.synchronize do
      if @cluster_time.nil?
        @cluster_time = cluster_time_doc
      elsif cluster_time_doc[CLUSTER_TIME] > @cluster_time[CLUSTER_TIME]
        @cluster_time = cluster_time_doc
      end
    end
  end
end

#update_topology(new_topology) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Since:

  • 2.0.0



620
621
622
623
624
625
626
627
 
# File 'lib/mongo/cluster.rb', line 620

def update_topology(new_topology)
  old_topology = topology
  @topology = new_topology
  publish_sdam_event(
    Monitoring::TOPOLOGY_CHANGED,
    Monitoring::Event::TopologyChanged.new(old_topology, topology)
  )
end