always use master for commands in MasterSlaveConnection instances

Author: Mike Dirolf

pymongo/collection.py

@@ -261,7 +261,7 @@ def remove(self, spec_or_object_id):
         self._send_message(2006, _ZERO + bson.BSON.from_dict(spec))
 
     def find_one(self, spec_or_object_id=None, fields=None, slave_okay=None,
-                 _sock=None):
+                 _sock=None, _must_use_master=False):
         """Get a single object from the database.
 
         Raises TypeError if the argument is of an improper type. Returns a
@@ -286,7 +286,8 @@ def find_one(self, spec_or_object_id=None, fields=None, slave_okay=None,
             spec = SON({"_id": spec})
 
         for result in self.find(spec, limit=-1, fields=fields,
-                                slave_okay=slave_okay, _sock=_sock):
+                                slave_okay=slave_okay, _sock=_sock,
+                                _must_use_master=_must_use_master):
             return result
         return None
 
@@ -309,7 +310,7 @@ def _fields_list_to_dict(self, fields):
 
     def find(self, spec=None, fields=None, skip=0, limit=0,
              slave_okay=None, timeout=True, snapshot=False,
-             _sock=None):
+             _sock=None, _must_use_master=False):
         """Query the database.
 
         The `spec` argument is a prototype document that all results must
@@ -376,7 +377,7 @@ def find(self, spec=None, fields=None, skip=0, limit=0,
             fields = self._fields_list_to_dict(fields)
 
         return Cursor(self, spec, fields, skip, limit, slave_okay, timeout, snapshot,
-                      _sock=_sock)
+                      _sock=_sock, _must_use_master=_must_use_master)
 
     def count(self):
         """Get the number of documents in this collection.

pymongo/connection.py

@@ -509,7 +509,9 @@ def receive(length):
 
     __hack_socket_lock = threading.Lock()
 
-    def _receive_message(self, operation, data, _sock=None):
+    # we just ignore _must_use_master here: it's only relavant for
+    # MasterSlaveConnection instances.
+    def _receive_message(self, operation, data, _sock=None, _must_use_master=False):
         """Receive a message from Mongo.
 
         Sends the given message and returns the response.

pymongo/cursor.py

@@ -37,7 +37,7 @@ class Cursor(object):
     """
 
     def __init__(self, collection, spec, fields, skip, limit, slave_okay,
-                 timeout, snapshot=False, _sock=None):
+                 timeout, snapshot=False, _sock=None, _must_use_master=False):
         """Create a new cursor.
 
         Should not be called directly by application developers.
@@ -54,6 +54,7 @@ def __init__(self, collection, spec, fields, skip, limit, slave_okay,
         self.__explain = False
         self.__hint = None
         self.__socket = _sock
+        self.__must_use_master = _must_use_master
 
         self.__data = []
         self.__id = None
@@ -290,7 +291,8 @@ def _refresh(self):
 
         def send_message(operation, message):
             db = self.__collection.database()
-            kwargs = {"_sock": self.__socket}
+            kwargs = {"_sock": self.__socket,
+                      "_must_use_master": self.__must_use_master}
             if self.__connection_id is not None:
                 kwargs["_connection_to_use"] = self.__connection_id
 

pymongo/database.py

@@ -182,7 +182,9 @@ def _fix_outgoing(self, son, collection):
     def _command(self, command, allowable_errors=[], check=True, sock=None):
         """Issue a DB command.
         """
-        result = self["$cmd"].find_one(command, _sock=sock)
+        result = self["$cmd"].find_one(command, _sock=sock,
+                                       _must_use_master=True)
+
         if check and result["ok"] != 1:
             if result["errmsg"] in allowable_errors:
                 return result
@@ -193,7 +195,7 @@ def _command(self, command, allowable_errors=[], check=True, sock=None):
     def collection_names(self):
         """Get a list of all the collection names in this database.
         """
-        results = self["system.namespaces"].find()
+        results = self["system.namespaces"].find(_must_use_master=True)
         names = [r["name"] for r in results]
         names = [n[len(self.__name) + 1:] for n in names
                  if n.startswith(self.__name + ".")]

pymongo/master_slave_connection.py

@@ -114,7 +114,7 @@ def _send_message(self, operation, data, _connection_to_use=None):
     # that getmore operations can be sent to the same instance on which
     # the cursor actually resides...
     def _receive_message(self, operation, data,
-                         _sock=None, _connection_to_use=None):
+                         _sock=None, _connection_to_use=None, _must_use_master=False):
         """Receive a message from Mongo.
 
         Sends the given message and returns a (connection_id, response) pair.
@@ -135,7 +135,10 @@ def _receive_message(self, operation, data,
         # for now just load-balance randomly among slaves only...
         connection_id = random.randrange(0, len(self.__slaves))
 
-        if self.__in_request or connection_id == -1:
+        # _must_use_master is set for commands, which must be sent to the
+        # master instance. any queries in a request must be sent to the
+        # master since that is where writes go.
+        if _must_use_master or self.__in_request or connection_id == -1:
             return (-1, self.__master._receive_message(operation, data, _sock))
 
         return (connection_id,

test/test_master_slave_connection.py

@@ -23,8 +23,10 @@
 from nose.plugins.skip import SkipTest
 
 from pymongo.errors import ConnectionFailure, InvalidName
+from pymongo.errors import CollectionInvalid, OperationFailure
 from pymongo.database import Database
 from pymongo.connection import Connection
+from pymongo.collection import Collection
 from pymongo.master_slave_connection import MasterSlaveConnection
 
 
@@ -145,6 +147,23 @@ def test_insert_find_one_no_slaves(self):
                 count += 1
         self.failIf(count)
 
+    # This was failing because commands were being sent to the slaves
+    def test_create_collection(self):
+        self.connection.drop_database('pymongo_test')
+
+        collection = self.db.create_collection('test')
+        self.assert_(isinstance(collection, Collection))
+
+        self.assertRaises(CollectionInvalid, self.db.create_collection, 'test')
+
+    # Believe this was failing for the same reason...
+    def test_unique_index(self):
+        self.connection.drop_database('pymongo_test')
+        self.db.test.create_index('username', unique=True)
+
+        self.db.test.save({'username': 'mike'}, safe=True)
+        self.assertRaises(OperationFailure, self.db.test.save, {'username': 'mike'}, safe=True)
+
     # NOTE this test is non-deterministic, but I expect
     # some failures unless the db is pulling instantaneously...
     def test_insert_find_one_with_slaves(self):
@@ -170,7 +189,7 @@ def test_insert_find_one_with_pause(self):
 
         self.db.test.remove({})
         self.db.test.insert({"x": 5586})
-        time.sleep(6)
+        time.sleep(7)
         for _ in range(10):
             try:
                 if 5586 != self.db.test.find_one()["x"]: