Class: Mongo::Server::Monitor Private

Inherits:
Object
  • Object
show all
Extended by:
Forwardable
Includes:
BackgroundThread, Event::Publisher, Loggable
Defined in:
lib/mongo/server/monitor.rb,
lib/mongo/server/monitor/connection.rb,
lib/mongo/server/monitor/app_metadata.rb

Overview

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

Responsible for periodically polling a server via hello commands to keep the server's status up to date.

Does all work in a background thread so as to not interfere with other operations performed by the driver.

Since:

  • 2.0.0

Defined Under Namespace

Classes: AppMetadata, Connection

Constant Summary collapse

DEFAULT_HEARTBEAT_INTERVAL =

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

The default interval between server status refreshes is 10 seconds.

Since:

  • 2.0.0

10
MIN_SCAN_INTERVAL =

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

The minimum time between forced server scans. Is minHeartbeatFrequencyMS in the SDAM spec.

Since:

  • 2.0.0

0.5
RTT_WEIGHT_FACTOR =

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

Deprecated.

Will be removed in version 3.0.

The weighting factor (alpha) for calculating the average moving round trip time.

Since:

  • 2.0.0

0.2

Constants included from Loggable

Loggable::PREFIX

Instance Attribute Summary collapse

Attributes included from Event::Publisher

#event_listeners

Instance Method Summary collapse

Methods included from BackgroundThread

#run!, #running?

Methods included from Loggable

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

Methods included from Event::Publisher

#publish

Constructor Details

#initialize(server, event_listeners, monitoring, options = {}) ⇒ Monitor

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:

Monitor must never be directly instantiated outside of a Server.

Create the new server monitor.

Examples:

Create the server monitor.

Mongo::Server::Monitor.new(address, listeners, monitoring)

Parameters:

  • server (Server) —

    The server to monitor.

  • event_listeners (Event::Listeners) —

    The event listeners.

  • monitoring (Monitoring) —

    The monitoring.

  • options (Hash) (defaults to: {}) —

    The options.

Options Hash (options):

  • :connect_timeout (Float) —

    The timeout, in seconds, to use when establishing the monitoring connection.

  • :heartbeat_interval (Float) —

    The interval between regular server checks.

  • :logger (Logger) —

    A custom logger to use.

  • :monitor_app_metadata (Mongo::Server::Monitor::AppMetadata) —

    The metadata to use for regular monitoring connection.

  • :push_monitor_app_metadata (Mongo::Server::Monitor::AppMetadata) —

    The metadata to use for push monitor's connection.

  • :socket_timeout (Float) —

    The timeout, in seconds, to execute operations on the monitoring connection.

Raises:

  • (ArgumentError)

Since:

  • 2.0.0



76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
 
# File 'lib/mongo/server/monitor.rb', line 76

def initialize(server, event_listeners, monitoring, options = {})
  raise ArgumentError, "Wrong monitoring type: #{monitoring.inspect}" unless monitoring.is_a?(Monitoring)
  raise ArgumentError, 'App metadata is required' unless options[:app_metadata]
  raise ArgumentError, 'Push monitor app metadata is required' unless options[:push_monitor_app_metadata]

  @server = server
  @event_listeners = event_listeners
  @monitoring = monitoring
  @options = options.freeze
  @mutex = Mutex.new
  @sdam_mutex = Mutex.new
  @next_earliest_scan = @next_wanted_scan = Time.now
  @update_mutex = Mutex.new
  # Guards reads and writes of @connection so the polling connection can
  # be cancelled from another thread. Per the Server Monitoring spec's
  # cancelCheck pseudocode, the lock is only held long enough to copy or
  # assign the reference - never across the blocking check.
  @connection_lock = Mutex.new
end

Instance Attribute Details

#monitoring ⇒ Monitoring (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 monitoring The monitoring.

Returns:

Since:

  • 2.0.0



127
128
129
 
# File 'lib/mongo/server/monitor.rb', line 127

def monitoring
  @monitoring
end

#options ⇒ Hash (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 options The server options.

Returns:

  • (Hash) —

    options The server options.

Since:

  • 2.0.0



108
109
110
 
# File 'lib/mongo/server/monitor.rb', line 108

def options
  @options
end

#server ⇒ Server (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 server The server that this monitor is monitoring.

Returns:

  • (Server) —

    server The server that this monitor is monitoring.

Since:

  • 2.0.0



98
99
100
 
# File 'lib/mongo/server/monitor.rb', line 98

def server
  @server
end

Instance Method Details

#cancel_check! ⇒ 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.

Cancel the in-progress check and close the monitoring connection.

Called when the server is marked Unknown from a network error, per the Server Monitoring spec ("hello or legacy hello Cancellation"). Stops the streaming PushMonitor (interrupting its awaited hello read) and closes the polling connection, so the next check must establish a fresh one rather than re-validating the server over a possibly-dead socket.

Since:

  • 2.0.0



210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
 
# File 'lib/mongo/server/monitor.rb', line 210

def cancel_check!
  stop_push_monitor!

  # Copy the connection reference under the lock, then interrupt and close
  # it outside the lock. Closing the socket interrupts any in-progress
  # read on the monitor thread; nil-ing the reference forces the next
  # check to reconnect. The monitor thread is the only writer of a new
  # connection, so it is safe for this thread to clear it.
  connection = @connection_lock.synchronize do
    conn = @connection
    @connection = nil
    conn
  end
  connection&.disconnect!
end

#connection ⇒ Mongo::Server::Monitor::Connection | nil

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 connection to use, read under @connection_lock so callers never observe a stale reference after a concurrent cancel_check! clears it.

Returns:

  • (Mongo::Server::Monitor::Connection | nil) —

    The connection to use, read under @connection_lock so callers never observe a stale reference after a concurrent cancel_check! clears it.

Since:

  • 2.0.0



103
104
105
 
# File 'lib/mongo/server/monitor.rb', line 103

def connection
  @connection_lock.synchronize { @connection }
end

#create_push_monitor!(topology_version, connection) ⇒ 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.

Parameters:

  • connection (Mongo::Server::Monitor::Connection) —

    The freshly established monitoring connection. Passed explicitly (rather than read from @connection) so a concurrent cancel_check! cannot nil it out from under us between establishing it and building the PushMonitor.

Since:

  • 2.0.0



175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
 
# File 'lib/mongo/server/monitor.rb', line 175

def create_push_monitor!(topology_version, connection)
  @update_mutex.synchronize do
    @push_monitor = nil if @push_monitor && !@push_monitor.running?

    @push_monitor ||= PushMonitor.new(
      self,
      topology_version,
      monitoring,
      **Utils.shallow_symbolize_keys(options.merge(
                                       socket_timeout: heartbeat_interval + connection.socket_timeout,
                                       app_metadata: options[:push_monitor_app_metadata],
                                       check_document: connection.check_document
                                     ))
    )
  end
end

#do_work ⇒ 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.

Perform a check of the server.

Since:

  • 2.0.0



140
141
142
143
144
145
146
147
148
149
150
151
152
 
# File 'lib/mongo/server/monitor.rb', line 140

def do_work
  scan!
  # @next_wanted_scan may be updated by the push monitor.
  # However we need to check for termination flag so that the monitor
  # thread exits when requested.
  loop do
    delta = @next_wanted_scan - Time.now
    break unless delta > 0

    signaled = server.scan_semaphore.wait(delta)
    break if signaled || @stop_requested
  end
end

#heartbeat_interval ⇒ Float

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.

The interval between regular server checks.

Returns:

  • (Float) —

    The heartbeat interval, in seconds.

Since:

  • 2.0.0



113
114
115
 
# File 'lib/mongo/server/monitor.rb', line 113

def heartbeat_interval
  options[:heartbeat_interval] || DEFAULT_HEARTBEAT_INTERVAL
end

#push_monitor ⇒ Server::PushMonitor | nil

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 push monitor, if one is being used.

Returns:

Since:

  • 2.0.0



131
132
133
134
135
 
# File 'lib/mongo/server/monitor.rb', line 131

def push_monitor
  @update_mutex.synchronize do
    @push_monitor
  end
end

#restart! ⇒ Thread

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.

Restarts the server monitor unless the current thread is alive.

Examples:

Restart the monitor.

monitor.restart!

Returns:

  • (Thread) —

    The thread the monitor runs on.

Since:

  • 2.1.0



308
309
310
311
312
313
314
 
# File 'lib/mongo/server/monitor.rb', line 308

def restart!
  if @thread && @thread.alive?
    @thread
  else
    run!
  end
end

#run_sdam_flow(result, awaited: false, scan_error: nil, rtt_only: false) ⇒ 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



268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
 
# File 'lib/mongo/server/monitor.rb', line 268

def run_sdam_flow(result, awaited: false, scan_error: nil, rtt_only: false)
  @sdam_mutex.synchronize do
    old_description = server.description

    # An RTT-only measurement (streaming protocol active) must not update
    # the topology or publish SDAM events. The RTT it gathered is
    # incorporated into the next streaming-hello description via the
    # shared RTT calculator. The scheduling below still runs so the
    # monitor keeps pacing its checks.
    unless rtt_only
      new_description = Description.new(
        server.address,
        result,
        average_round_trip_time: server.round_trip_time_calculator.average_round_trip_time,
        minimum_round_trip_time: server.round_trip_time_calculator.minimum_round_trip_time
      )
      server.cluster.run_sdam_flow(server.description, new_description, awaited: awaited, scan_error: scan_error)
    end

    server.description.tap do |new_description|
      unless awaited
        if new_description.unknown? && !old_description.unknown?
          @next_earliest_scan = @next_wanted_scan = Time.now
        else
          @next_earliest_scan = Time.now + MIN_SCAN_INTERVAL
          @next_wanted_scan = Time.now + heartbeat_interval
        end
      end
    end
  end
end

#scan! ⇒ Description

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:

If the system clock moves backwards, this method can sleep for a very long time.

Note:

The return value of this method is deprecated. In version 3.0.0 this method will not have a return value.

Perform a check of the server with throttling, and update the server's description and average round trip time.

If the server was checked less than MIN_SCAN_INTERVAL seconds ago, sleep until MIN_SCAN_INTERVAL seconds have passed since the last check. Then perform the check which involves running hello on the server being monitored and updating the server description as a result.

Returns:

Since:

  • 2.0.0



244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
 
# File 'lib/mongo/server/monitor.rb', line 244

def scan!
  # Ordinarily the background thread would invoke this method.
  # But it is also possible to invoke scan! directly on a monitor.
  # Allow only one scan to be performed at a time.
  @mutex.synchronize do
    throttle_scan_frequency!

    # When the streaming protocol is active the PushMonitor is the
    # authoritative SDAM source and this scan only measures RTT on the
    # dedicated connection. Per the Server Monitoring spec, an RTT
    # command MUST NOT publish events or update the topology. Compute
    # this before do_scan, which may (re)connect and change the state.
    rtt_only = rtt_measurement_only?

    begin
      result = do_scan(publish_heartbeat: !rtt_only)
    rescue StandardError => e
      run_sdam_flow({}, scan_error: e, rtt_only: rtt_only)
    else
      run_sdam_flow(result, rtt_only: rtt_only)
    end
  end
end

#stop! ⇒ true | false

Stop the background thread and wait for it to terminate for a reasonable amount of time.

Returns:

  • (true | false) —

    Whether the thread was terminated.

Since:

  • 2.0.0



160
161
162
163
164
165
166
167
168
169
 
# File 'lib/mongo/server/monitor.rb', line 160

def stop!
  stop_push_monitor!

  # Forward super's return value
  super.tap do
    # Important: disconnect should happen after the background thread
    # terminates.
    connection&.disconnect!
  end
end

#stop_push_monitor! ⇒ 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



192
193
194
195
196
197
198
199
 
# File 'lib/mongo/server/monitor.rb', line 192

def stop_push_monitor!
  @update_mutex.synchronize do
    if @push_monitor
      @push_monitor.stop!
      @push_monitor = nil
    end
  end
end

#to_s ⇒ 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



316
317
318
 
# File 'lib/mongo/server/monitor.rb', line 316

def to_s
  "#<#{self.class.name}:#{object_id} #{server.address}>"
end