PYTHON-976 - Publish a CommandFailedEvent for arbitrary failures.

Author: Luke Lovett

pymongo/collection.py

@@ -398,19 +398,22 @@ def _legacy_write(
  try:
  result = sock_info.legacy_write(
  rqst_id, msg, max_size, acknowledged)
- except OperationFailure as exc:
+ except Exception as exc:
  if publish:
  dur = (datetime.datetime.now() - start) + duration
- details = exc.details
- # Succeed if GLE was successful and this is a write error.
- if details.get("ok") and "n" in details:
- reply = message._convert_write_result(
- name, cmd, details)
- monitoring.publish_command_success(
- dur, reply, name, rqst_id, sock_info.address, op_id)
+ if isinstance(exc, OperationFailure):
+ details = exc.details
+ # Succeed if GLE was successful and this is a write error.
+ if details.get("ok") and "n" in details:
+ reply = message._convert_write_result(
+ name, cmd, details)
+ monitoring.publish_command_success(
+ dur, reply, name, rqst_id, sock_info.address, op_id)
+ raise
  else:
- monitoring.publish_command_failure(
- dur, details, name, rqst_id, sock_info.address, op_id)
+ details = message._convert_exception(exc)
+ monitoring.publish_command_failure(
+ dur, details, name, rqst_id, sock_info.address, op_id)
  raise
  if publish:
  if result is not None:

pymongo/command_cursor.py

@@ -21,7 +21,7 @@
 from bson.py3compat import integer_types
 from pymongo import helpers, monitoring
 from pymongo.errors import AutoReconnect, NotMasterError, OperationFailure
-from pymongo.message import _CursorAddress, _GetMore
+from pymongo.message import _CursorAddress, _GetMore, _convert_exception
 
 
 class CommandCursor(object):
@@ -132,6 +132,13 @@ def __send_message(self, operation):
 
  client._reset_server_and_request_check(self.address)
  raise
+ except Exception as exc:
+ if publish:
+ duration = (datetime.datetime.now() - start) + cmd_duration
+ monitoring.publish_command_failure(
+ duration, _convert_exception(exc), "getMore", rqst_id,
+ self.__address)
+ raise
 
  if publish:
  duration = (datetime.datetime.now() - start) + cmd_duration

pymongo/cursor.py

@@ -32,7 +32,7 @@
  InvalidOperation,
  NotMasterError,
  OperationFailure)
-from pymongo.message import _CursorAddress, _GetMore, _Query
+from pymongo.message import _CursorAddress, _GetMore, _Query, _convert_exception
 from pymongo.read_preferences import ReadPreference
 
 _QUERY_OPTIONS = {
@@ -850,8 +850,14 @@ def __send_message(self, operation):
  start = datetime.datetime.now()
  try:
  data = self.__exhaust_mgr.sock.receive_message(1, None)
- except ConnectionFailure:
- self.__die()
+ except Exception as exc:
+ if publish:
+ duration = datetime.datetime.now() - start
+ monitoring.publish_command_failure(
+ duration, _convert_exception(exc), cmd_name, rqst_id,
+ self.__address)
+ if isinstance(exc, ConnectionFailure):
+ self.__die()
  raise
  if publish:
  cmd_duration = datetime.datetime.now() - start
@@ -895,6 +901,13 @@ def __send_message(self, operation):
 
  client._reset_server_and_request_check(self.__address)
  raise
+ except Exception as exc:
+ if publish:
+ duration = (datetime.datetime.now() - start) + cmd_duration
+ monitoring.publish_command_failure(
+ duration, _convert_exception(exc), cmd_name, rqst_id,
+ self.__address)
+ raise
 
  if publish:
  duration = (datetime.datetime.now() - start) + cmd_duration

pymongo/helpers.py

@@ -31,7 +31,7 @@
  WriteError,
  WriteConcernError,
  WTimeoutError)
-from pymongo.message import _Query
+from pymongo.message import _Query, _convert_exception
 
 
 _UUNDER = u("_")
@@ -259,11 +259,15 @@ def _first_batch(sock_info, namespace, query,
  response = sock_info.receive_message(1, request_id)
  try:
  result = _unpack_response(response, None, codec_options)
- except (NotMasterError, OperationFailure) as exc:
+ except Exception as exc:
  if publish:
  duration = (datetime.datetime.now() - start) + encoding_duration
+ if isinstance(exc, (NotMasterError, OperationFailure)):
+ failure = exc.details
+ else:
+ failure = _convert_exception(exc)
  monitoring.publish_command_failure(
- duration, exc.details, name, request_id, sock_info.address)
+ duration, failure, name, request_id, sock_info.address)
  raise
  if publish:
  duration = (datetime.datetime.now() - start) + encoding_duration

pymongo/message.py

@@ -82,6 +82,12 @@ def _maybe_add_read_preference(spec, read_preference):
  return spec
 
 
+def _convert_exception(exception):
+ """Convert an Exception into a failure document for publishing."""
+ return {'errmsg': exception.message,
+ 'errtype': exception.__class__.__name__}
+
+
 def _convert_write_result(operation, command, result):
  """Convert a legacy write result to write commmand format."""
 

pymongo/mongo_client.py

@@ -938,7 +938,15 @@ def _process_kill_cursors_queue(self):
  monitoring.publish_command_start(
  command, dbname, data[0], address)
  start = datetime.datetime.now()
- server.send_message(data, self.__all_credentials)
+ try:
+ server.send_message(data, self.__all_credentials)
+ except Exception as exc:
+ if publish:
+ dur = (datetime.datetime.now() - start) + duration
+ monitoring.publish_command_failure(
+ dur, message._convert_exception(exc),
+ 'killCursors', data[0], address)
+ raise
  if publish:
  duration = (datetime.datetime.now() - start) + duration
  # OP_KILL_CURSORS returns no reply, fake one.

pymongo/monitoring.py

@@ -198,7 +198,7 @@ def publish_command_failure(
 
  :Parameters:
  - `duration`: The command duration as a datetime.timedelta.
- - `failure`: The server reply document.
+ - `failure`: The server reply document or failure description document.
  - `command_name`: The command name.
  - `request_id`: The request id for this operation.
  - `connection_id`: The address (host, port) of the server this command

pymongo/network.py

@@ -75,9 +75,9 @@ def command(sock, dbname, spec, slave_ok, is_mongos,
  monitoring.publish_command_start(orig, dbname, request_id, address)
  start = datetime.datetime.now()
 
- sock.sendall(msg)
- response = receive_message(sock, 1, request_id)
  try:
+ sock.sendall(msg)
+ response = receive_message(sock, 1, request_id)
  unpacked = helpers._unpack_response(
  response, codec_options=codec_options)
 
@@ -86,11 +86,15 @@ def command(sock, dbname, spec, slave_ok, is_mongos,
  msg = "command %s on namespace %s failed: %%s" % (
  repr(spec).replace("%", "%%"), ns)
  helpers._check_command_response(response_doc, msg, allowable_errors)
- except (NotMasterError, OperationFailure) as exc:
+ except Exception as exc:
  if publish:
  duration = (datetime.datetime.now() - start) + encoding_duration
+ if isinstance(exc, (NotMasterError, OperationFailure)):
+ failure = exc.details
+ else:
+ failure = message._convert_exception(exc)
  monitoring.publish_command_failure(
- duration, exc.details, name, request_id, address)
+ duration, failure, name, request_id, address)
  raise
  if publish:
  duration = (datetime.datetime.now() - start) + encoding_duration

pymongo/server.py

@@ -19,6 +19,7 @@
 from datetime import datetime
 
 from pymongo import monitoring
+from pymongo.message import _convert_exception
 from pymongo.response import Response, ExhaustResponse
 from pymongo.server_type import SERVER_TYPE
 
@@ -101,8 +102,17 @@ def send_message_with_response(
  cmd, dbn, request_id, sock_info.address)
  start = datetime.now()
 
- sock_info.send_message(data, max_doc_size)
- response_data = sock_info.receive_message(1, request_id)
+ try:
+ sock_info.send_message(data, max_doc_size)
+ response_data = sock_info.receive_message(1, request_id)
+ except Exception as exc:
+ if publish:
+ duration = (datetime.now() - start) + encoding_duration
+ failure = _convert_exception(exc)
+ monitoring.publish_command_failure(
+ duration, failure, next(iter(cmd)), request_id,
+ sock_info.address)
+ raise
 
  if publish:
  duration = (datetime.now() - start) + encoding_duration