PYTHON-1067 - Implement SDAM Monitoring

Author: aherlihy

doc/api/pymongo/monitoring.rst

@@ -7,12 +7,33 @@
  .. autofunction:: register(listener)
  .. autoclass:: CommandListener
  :members:
+ .. autoclass:: ServerListener
+ :members:
+ .. autoclass:: ServerHeartbeatListener
+ :members:
+ .. autoclass:: TopologyListener
+ :members:
  .. autoclass:: CommandStartedEvent
  :members:
- :inherited-members:
  .. autoclass:: CommandSucceededEvent
  :members:
- :inherited-members:
  .. autoclass:: CommandFailedEvent
  :members:
- :inherited-members:
+ .. autoclass:: ServerDescriptionChangedEvent
+ :members:
+ .. autoclass:: ServerOpeningEvent
+ :members:
+ .. autoclass:: ServerClosedEvent
+ :members:
+ .. autoclass:: TopologyDescriptionChangedEvent
+ :members:
+ .. autoclass:: TopologyOpenedEvent
+ :members:
+ .. autoclass:: TopologyClosedEvent
+ :members:
+ .. autoclass:: ServerHeartbeatStartedEvent
+ :members:
+ .. autoclass:: ServerHeartbeatSucceededEvent
+ :members:
+ .. autoclass:: ServerHeartbeatFailedEvent
+ :members:

pymongo/common.py

@@ -49,6 +49,9 @@
 # Frequency to process kill-cursors, in seconds. See MongoClient.close_cursor.
 KILL_CURSOR_FREQUENCY = 1
 
+# Frequency to process events queue, in seconds.
+EVENTS_QUEUE_FREQUENCY = 1
+
 # How long to wait, in seconds, for a suitable server to be found before
 # aborting an operation. For example, if the client attempts an insert
 # during a replica set election, SERVER_SELECTION_TIMEOUT governs the

pymongo/monitor.py

@@ -44,6 +44,9 @@ def __init__(
  self._pool = pool
  self._settings = topology_settings
  self._avg_round_trip_time = MovingAverage()
+ self._listeners = self._settings._pool_options.event_listeners
+ pub = self._listeners is not None
+ self._publish = pub and self._listeners.enabled_for_server_heartbeat
 
  # We strongly reference the executor and it weakly references us via
  # this closure. When the monitor is freed, stop the executor soon.
@@ -61,7 +64,7 @@ def target():
  name="pymongo_server_monitor_thread")
 
  self._executor = executor
- 
+
  # Avoid cycles. When self or topology is freed, stop executor soon.
  self_ref = weakref.ref(self, executor.close)
  self._topology = weakref.proxy(topology, executor.close)
@@ -110,24 +113,34 @@ def _check_with_retry(self):
  address = self._server_description.address
  retry = self._server_description.server_type != SERVER_TYPE.Unknown
 
+ start = _time()
  try:
  return self._check_once()
  except ReferenceError:
  raise
  except Exception as error:
+ error_time = _time() - start
  self._topology.reset_pool(address)
  default = ServerDescription(address, error=error)
  if not retry:
+ if self._publish:
+ self._listeners.publish_server_heartbeat_failed(
+ address, error_time, error)
  self._avg_round_trip_time.reset()
  # Server type defaults to Unknown.
  return default
 
  # Try a second and final time. If it fails return original error.
+ start = _time()
  try:
  return self._check_once()
  except ReferenceError:
  raise
- except Exception:
+ except Exception as error:
+ error_time = _time() - start
+ if self._publish:
+ self._listeners.publish_server_heartbeat_failed(
+ address, error_time, error)
  self._avg_round_trip_time.reset()
  return default
 
@@ -136,13 +149,19 @@ def _check_once(self):
 
  Returns a ServerDescription, or raises an exception.
  """
+ address = self._server_description.address
+ if self._publish:
+ self._listeners.publish_server_heartbeat_started(address)
  with self._pool.get_socket({}) as sock_info:
  response, round_trip_time = self._check_with_socket(sock_info)
  self._avg_round_trip_time.add_sample(round_trip_time)
  sd = ServerDescription(
- address=self._server_description.address,
+ address=address,
  ismaster=response,
  round_trip_time=self._avg_round_trip_time.get())
+ if self._publish:
+ self._listeners.publish_server_heartbeat_succeeded(
+ address, round_trip_time, response)
 
  return sd