PYTHON-1651 Publish CommandFailedEvent when bulk write fails with a network error

Author: ShaneHarvey

pymongo/message.py

@@ -117,7 +117,7 @@ def _convert_exception(exception):
 
 
 def _convert_write_result(operation, command, result):
-    """Convert a legacy write result to write commmand format."""
+    """Convert a legacy write result to write command format."""
 
     # Based on _merge_legacy from bulk.py
     affected = result.get("n", 0)
@@ -971,14 +971,17 @@ def legacy_write(self, request_id, msg, max_doc_size, acknowledged, docs):
                     # Comply with APM spec.
                     reply = {'ok': 1}
                 self._succeed(request_id, reply, duration)
-        except OperationFailure as exc:
+        except Exception as exc:
             if self.publish:
                 duration = (datetime.datetime.now() - start) + duration
-                self._fail(
-                    request_id,
-                    _convert_write_result(
-                        self.name, cmd, exc.details),
-                    duration)
+                if isinstance(exc, OperationFailure):
+                    failure = _convert_write_result(
+                        self.name, cmd, exc.details)
+                elif isinstance(exc, NotMasterError):
+                    failure = exc.details
+                else:
+                    failure = _convert_exception(exc)
+                self._fail(request_id, failure, duration)
             raise
         finally:
             self.start_time = datetime.datetime.now()
@@ -996,10 +999,14 @@ def write_command(self, request_id, msg, docs):
             if self.publish:
                 duration = (datetime.datetime.now() - start) + duration
                 self._succeed(request_id, reply, duration)
-        except OperationFailure as exc:
+        except Exception as exc:
             if self.publish:
                 duration = (datetime.datetime.now() - start) + duration
-                self._fail(request_id, exc.details, duration)
+                if isinstance(exc, (NotMasterError, OperationFailure)):
+                    failure = exc.details
+                else:
+                    failure = _convert_exception(exc)
+                self._fail(request_id, failure, duration)
             raise
         finally:
             self.start_time = datetime.datetime.now()

test/__init__.py

@@ -734,6 +734,17 @@ def assertEqualCommand(self, expected, actual, msg=None):
     def assertEqualReply(self, expected, actual, msg=None):
         self.assertEqual(sanitize_reply(expected), sanitize_reply(actual), msg)
 
+    @contextmanager
+    def fail_point(self, command_args):
+        cmd_on = SON([('configureFailPoint', 'failCommand')])
+        cmd_on.update(command_args)
+        client_context.client.admin.command(cmd_on)
+        try:
+            yield
+        finally:
+            client_context.client.admin.command(
+                'configureFailPoint', cmd_on['configureFailPoint'], mode='off')
+
 
 class IntegrationTest(PyMongoTestCase):
     """Base class for TestCases that need a connection to MongoDB to pass."""
@@ -748,16 +759,6 @@ def setUpClass(cls):
         else:
             cls.credentials = {}
 
-    @contextmanager
-    def fail_point(self, command_args):
-        cmd_on = SON([('configureFailPoint', 'failCommand')])
-        cmd_on.update(command_args)
-        self.client.admin.command(cmd_on)
-        try:
-            yield
-        finally:
-            self.client.admin.command(
-                'configureFailPoint', cmd_on['configureFailPoint'], mode='off')
 
 # Use assertRaisesRegex if available, otherwise use Python 2.7's
 # deprecated assertRaisesRegexp, with a 'p'.

test/test_monitoring.py

@@ -25,7 +25,9 @@
 from bson.son import SON
 from pymongo import CursorType, monitoring, InsertOne, UpdateOne, DeleteOne
 from pymongo.command_cursor import CommandCursor
-from pymongo.errors import NotMasterError, OperationFailure
+from pymongo.errors import (AutoReconnect,
+                            NotMasterError,
+                            OperationFailure)
 from pymongo.read_preferences import ReadPreference
 from pymongo.write_concern import WriteConcern
 from test import (client_context,
@@ -34,6 +36,7 @@
                   sanitize_cmd,
                   unittest)
 from test.utils import (EventListener,
+                        get_pool,
                         rs_or_single_client,
                         single_client,
                         wait_until)
@@ -1172,6 +1175,78 @@ def test_bulk_write(self):
                                           ('limit', 1)])])])
         self.assertEqualCommand(expected, started[2].command)
 
+    @client_context.require_failCommand_fail_point
+    def test_bulk_write_command_network_error(self):
+        coll = self.client.pymongo_test.test
+        self.listener.results.clear()
+
+        insert_network_error = {
+            'configureFailPoint': 'failCommand',
+            'mode': {'times': 1},
+            'data': {
+                'failCommands': ['insert'],
+                'closeConnection': True,
+            },
+        }
+        with self.fail_point(insert_network_error):
+            with self.assertRaises(AutoReconnect):
+                coll.bulk_write([InsertOne({'_id': 1})])
+        failed = self.listener.results['failed']
+        self.assertEqual(1, len(failed))
+        event = failed[0]
+        self.assertEqual(event.command_name, 'insert')
+        self.assertIsInstance(event.failure, dict)
+        self.assertEqual(event.failure['errtype'], 'AutoReconnect')
+        self.assertTrue(event.failure['errmsg'])
+
+    @client_context.require_failCommand_fail_point
+    def test_bulk_write_command_error(self):
+        coll = self.client.pymongo_test.test
+        self.listener.results.clear()
+
+        insert_command_error = {
+            'configureFailPoint': 'failCommand',
+            'mode': {'times': 1},
+            'data': {
+                'failCommands': ['insert'],
+                'closeConnection': False,
+                'errorCode': 10107,  # NotMaster
+            },
+        }
+        with self.fail_point(insert_command_error):
+            with self.assertRaises(NotMasterError):
+                coll.bulk_write([InsertOne({'_id': 1})])
+        failed = self.listener.results['failed']
+        self.assertEqual(1, len(failed))
+        event = failed[0]
+        self.assertEqual(event.command_name, 'insert')
+        self.assertIsInstance(event.failure, dict)
+        self.assertEqual(event.failure['code'], 10107)
+        self.assertTrue(event.failure['errmsg'])
+
+    @client_context.require_version_max(3, 4, 99)
+    def test_bulk_write_legacy_network_error(self):
+        self.listener.results.clear()
+
+        # Make the delete operation run on a closed connection.
+        self.client.admin.command('ping')
+        pool = get_pool(self.client)
+        sock_info = pool.sockets[0]
+        sock_info.sock.close()
+
+        # Test legacy unacknowledged write network error.
+        coll = self.client.pymongo_test.get_collection(
+            'test', write_concern=WriteConcern(w=0))
+        with self.assertRaises(AutoReconnect):
+            coll.bulk_write([InsertOne({'_id': 1})], ordered=False)
+        failed = self.listener.results['failed']
+        self.assertEqual(1, len(failed))
+        event = failed[0]
+        self.assertEqual(event.command_name, 'insert')
+        self.assertIsInstance(event.failure, dict)
+        self.assertEqual(event.failure['errtype'], 'AutoReconnect')
+        self.assertTrue(event.failure['errmsg'])
+
     def test_write_errors(self):
         coll = self.client.pymongo_test.test
         coll.drop()