PYTHON-2634 Only update pools for data-bearing servers (#590)

Fixes a noisy OperationFailure: Authentication failed error.
Do not attempt to create unneeded connections to arbiters, ghosts,
hidden members, or unknown members.

(cherry picked from commit 4c7718e)

 Conflicts:
	pymongo/topology.py
	test/test_client.py
	test/test_cmap.py

Author: ShaneHarvey

pymongo/topology.py

@@ -430,15 +430,26 @@ def handle_getlasterror(self, address, error_msg):
                     ServerDescription(address, error=error), True)
                 server.request_check()
 
+    def data_bearing_servers(self):
+        """Return a list of all data-bearing servers.
+
+        This includes any server that might be selected for an operation.
+        """
+        if self._description.topology_type == TOPOLOGY_TYPE.Single:
+            return self._description.known_servers
+        return self._description.readable_servers
+
     def update_pool(self, all_credentials):
         # Remove any stale sockets and add new sockets if pool is too small.
         servers = []
         with self._lock:
-            for server in self._servers.values():
-                servers.append((server, server._pool.generation))
+            # Only update pools for data-bearing servers.
+            for sd in self.data_bearing_servers():
+                server = self._servers[sd.address]
+                servers.append((server, server.pool.generation))
 
         for server, generation in servers:
-            server._pool.remove_stale_sockets(generation, all_credentials)
+            server.pool.remove_stale_sockets(generation, all_credentials)
 
     def close(self):
         """Clear pools and terminate monitors. Topology reopens on demand."""

test/pymongo_mocks.py

@@ -85,13 +85,13 @@ def _check_once(self):
 class MockClient(MongoClient):
     def __init__(
             self, standalones, members, mongoses, ismaster_hosts=None,
-            *args, **kwargs):
+            arbiters=None, down_hosts=None, *args, **kwargs):
         """A MongoClient connected to the default server, with a mock topology.
 
-        standalones, members, mongoses determine the configuration of the
-        topology. They are formatted like ['a:1', 'b:2']. ismaster_hosts
-        provides an alternative host list for the server's mocked ismaster
-        response; see test_connect_with_internal_ips.
+        standalones, members, mongoses, arbiters, and down_hosts determine the
+        configuration of the topology. They are formatted like ['a:1', 'b:2'].
+        ismaster_hosts provides an alternative host list for the server's
+        mocked ismaster response; see test_connect_with_internal_ips.
         """
         self.mock_standalones = standalones[:]
         self.mock_members = members[:]
@@ -101,6 +101,9 @@ def __init__(
         else:
             self.mock_primary = None
 
+        # Hosts that should be considered an arbiter.
+        self.mock_arbiters = arbiters[:] if arbiters else []
+
         if ismaster_hosts is not None:
             self.mock_ismaster_hosts = ismaster_hosts
         else:
@@ -109,7 +112,7 @@ def __init__(
         self.mock_mongoses = mongoses[:]
 
         # Hosts that should raise socket errors.
-        self.mock_down_hosts = []
+        self.mock_down_hosts = down_hosts[:] if down_hosts else []
 
         # Hostname -> (min wire version, max wire version)
         self.mock_wire_versions = {}
@@ -182,6 +185,10 @@ def mock_is_master(self, host):
 
             if self.mock_primary:
                 response['primary'] = self.mock_primary
+
+            if host in self.mock_arbiters:
+                response['arbiterOnly'] = True
+                response['secondary'] = False
         elif host in self.mock_mongoses:
             response = {
                 'ok': 1,

test/test_client.py

@@ -35,7 +35,7 @@
 from bson.son import SON
 from bson.tz_util import utc
 import pymongo
-from pymongo import auth, message
+from pymongo import auth, message, monitoring
 from pymongo.common import CONNECT_TIMEOUT, _UUID_REPRESENTATIONS
 from pymongo.command_cursor import CommandCursor
 from pymongo.compression_support import _HAVE_SNAPPY, _HAVE_ZSTD
@@ -58,7 +58,7 @@
 from pymongo.pool import SocketInfo, _METADATA
 from pymongo.read_preferences import ReadPreference
 from pymongo.server_description import ServerDescription
-from pymongo.server_selectors import (any_server_selector,
+from pymongo.server_selectors import (readable_server_selector,
                                       writable_server_selector)
 from pymongo.server_type import SERVER_TYPE
 from pymongo.settings import TOPOLOGY_TYPE
@@ -76,6 +76,7 @@
 from test.pymongo_mocks import MockClient
 from test.utils import (assertRaisesExactly,
                         connected,
+                        CMAPListener,
                         delay,
                         FunctionCallRecorder,
                         get_pool,
@@ -452,21 +453,25 @@ def test_uri_security_options(self):
 
 class TestClient(IntegrationTest):
 
-    def test_max_idle_time_reaper(self):
+    def test_max_idle_time_reaper_default(self):
         with client_knobs(kill_cursor_frequency=0.1):
             # Assert reaper doesn't remove sockets when maxIdleTimeMS not set
             client = rs_or_single_client()
-            server = client._get_topology().select_server(any_server_selector)
+            server = client._get_topology().select_server(
+                readable_server_selector)
             with server._pool.get_socket({}) as sock_info:
                 pass
             self.assertEqual(1, len(server._pool.sockets))
             self.assertTrue(sock_info in server._pool.sockets)
             client.close()
 
+    def test_max_idle_time_reaper_removes_stale_minPoolSize(self):
+        with client_knobs(kill_cursor_frequency=0.1):
             # Assert reaper removes idle socket and replaces it with a new one
             client = rs_or_single_client(maxIdleTimeMS=500,
                                          minPoolSize=1)
-            server = client._get_topology().select_server(any_server_selector)
+            server = client._get_topology().select_server(
+                readable_server_selector)
             with server._pool.get_socket({}) as sock_info:
                 pass
             # When the reaper runs at the same time as the get_socket, two
@@ -478,11 +483,14 @@ def test_max_idle_time_reaper(self):
                        "replace stale socket")
             client.close()
 
+    def test_max_idle_time_reaper_does_not_exceed_maxPoolSize(self):
+        with client_knobs(kill_cursor_frequency=0.1):
             # Assert reaper respects maxPoolSize when adding new sockets.
             client = rs_or_single_client(maxIdleTimeMS=500,
                                          minPoolSize=1,
                                          maxPoolSize=1)
-            server = client._get_topology().select_server(any_server_selector)
+            server = client._get_topology().select_server(
+                readable_server_selector)
             with server._pool.get_socket({}) as sock_info:
                 pass
             # When the reaper runs at the same time as the get_socket,
@@ -494,9 +502,12 @@ def test_max_idle_time_reaper(self):
                        "replace stale socket")
             client.close()
 
+    def test_max_idle_time_reaper_removes_stale(self):
+        with client_knobs(kill_cursor_frequency=0.1):
             # Assert reaper has removed idle socket and NOT replaced it
             client = rs_or_single_client(maxIdleTimeMS=500)
-            server = client._get_topology().select_server(any_server_selector)
+            server = client._get_topology().select_server(
+                readable_server_selector)
             with server._pool.get_socket({}) as sock_info_one:
                 pass
             # Assert that the pool does not close sockets prematurely.
@@ -512,12 +523,14 @@ def test_max_idle_time_reaper(self):
     def test_min_pool_size(self):
         with client_knobs(kill_cursor_frequency=.1):
             client = rs_or_single_client()
-            server = client._get_topology().select_server(any_server_selector)
+            server = client._get_topology().select_server(
+                readable_server_selector)
             self.assertEqual(0, len(server._pool.sockets))
 
             # Assert that pool started up at minPoolSize
             client = rs_or_single_client(minPoolSize=10)
-            server = client._get_topology().select_server(any_server_selector)
+            server = client._get_topology().select_server(
+                readable_server_selector)
             wait_until(lambda: 10 == len(server._pool.sockets),
                        "pool initialized with 10 sockets")
 
@@ -532,7 +545,8 @@ def test_max_idle_time_checkout(self):
         # Use high frequency to test _get_socket_no_auth.
         with client_knobs(kill_cursor_frequency=99999999):
             client = rs_or_single_client(maxIdleTimeMS=500)
-            server = client._get_topology().select_server(any_server_selector)
+            server = client._get_topology().select_server(
+                readable_server_selector)
             with server._pool.get_socket({}) as sock_info:
                 pass
             self.assertEqual(1, len(server._pool.sockets))
@@ -546,7 +560,8 @@ def test_max_idle_time_checkout(self):
 
             # Test that sockets are reused if maxIdleTimeMS is not set.
             client = rs_or_single_client()
-            server = client._get_topology().select_server(any_server_selector)
+            server = client._get_topology().select_server(
+                readable_server_selector)
             with server._pool.get_socket({}) as sock_info:
                 pass
             self.assertEqual(1, len(server._pool.sockets))
@@ -2008,5 +2023,60 @@ def timeout_task():
         self.assertIsNone(ct.get())
 
 
+class TestClientPool(MockClientTest):
+
+    def test_rs_client_does_not_maintain_pool_to_arbiters(self):
+        listener = CMAPListener()
+        c = MockClient(
+            standalones=[],
+            members=['a:1', 'b:2', 'c:3', 'd:4'],
+            mongoses=[],
+            arbiters=['c:3'],  # c:3 is an arbiter.
+            down_hosts=['d:4'],  # d:4 is unreachable.
+            host=['a:1', 'b:2', 'c:3', 'd:4'],
+            replicaSet='rs',
+            minPoolSize=1,  # minPoolSize
+            event_listeners=[listener],
+        )
+        self.addCleanup(c.close)
+
+        wait_until(lambda: len(c.nodes) == 3, 'connect')
+        self.assertEqual(c.address, ('a', 1))
+        self.assertEqual(c.arbiters, set([('c', 3)]))
+        # Assert that we create 2 and only 2 pooled connections.
+        listener.wait_for_event(monitoring.ConnectionReadyEvent, 2)
+        self.assertEqual(
+            listener.event_count(monitoring.ConnectionCreatedEvent), 2)
+        # Assert that we do not create connections to arbiters.
+        arbiter = c._topology.get_server_by_address(('c', 3))
+        self.assertFalse(arbiter.pool.sockets)
+        # Assert that we do not create connections to unknown servers.
+        arbiter = c._topology.get_server_by_address(('d', 4))
+        self.assertFalse(arbiter.pool.sockets)
+
+    def test_direct_client_maintains_pool_to_arbiter(self):
+        listener = CMAPListener()
+        c = MockClient(
+            standalones=[],
+            members=['a:1', 'b:2', 'c:3'],
+            mongoses=[],
+            arbiters=['c:3'],  # c:3 is an arbiter.
+            host='c:3',
+            directConnection=True,
+            minPoolSize=1,  # minPoolSize
+            event_listeners=[listener],
+        )
+        self.addCleanup(c.close)
+
+        wait_until(lambda: len(c.nodes) == 1, 'connect')
+        self.assertEqual(c.address, ('c', 3))
+        # Assert that we create 1 pooled connection.
+        listener.wait_for_event(monitoring.ConnectionReadyEvent, 1)
+        self.assertEqual(
+            listener.event_count(monitoring.ConnectionCreatedEvent), 1)
+        arbiter = c._topology.get_server_by_address(('c', 3))
+        self.assertEqual(len(arbiter.pool.sockets), 1)
+
+
 if __name__ == "__main__":
     unittest.main()

test/test_heartbeat_monitoring.py

@@ -51,12 +51,12 @@ def _check_with_socket(self, *args, **kwargs):
             # monitor thread may run multiple times during the execution
             # of this test.
             wait_until(
-                lambda: len(listener.results) >= expected_len,
+                lambda: len(listener.events) >= expected_len,
                 "publish all events")
 
         try:
             # zip gives us len(expected_results) pairs.
-            for expected, actual in zip(expected_results, listener.results):
+            for expected, actual in zip(expected_results, listener.events):
                 self.assertEqual(expected,
                                  actual.__class__.__name__)
                 self.assertEqual(actual.connection_id,

test/test_streaming_protocol.py

@@ -215,7 +215,7 @@ def hb_failed(event):
         self.assertTrue(hb_failed_events[0].awaited)
         # Depending on thread scheduling, the failed heartbeat could occur on
         # the second or third check.
-        events = [type(e) for e in hb_listener.results[:4]]
+        events = [type(e) for e in hb_listener.events[:4]]
         if events == [monitoring.ServerHeartbeatStartedEvent,
                       monitoring.ServerHeartbeatSucceededEvent,
                       monitoring.ServerHeartbeatStartedEvent,

test/utils.py

@@ -36,7 +36,7 @@
 from pymongo import (MongoClient,
                      monitoring, read_preferences)
 from pymongo.errors import ConfigurationError, OperationFailure
-from pymongo.monitoring import _SENSITIVE_COMMANDS, ConnectionPoolListener
+from pymongo.monitoring import _SENSITIVE_COMMANDS
 from pymongo.pool import (_CancellationContext,
                           PoolOptions)
 from pymongo.read_concern import ReadConcern
@@ -60,7 +60,7 @@
 IMPOSSIBLE_WRITE_CONCERN = WriteConcern(w=50)
 
 
-class CMAPListener(ConnectionPoolListener):
+class BaseListener(object):
     def __init__(self):
         self.events = []
 
@@ -71,8 +71,26 @@ def add_event(self, event):
         self.events.append(event)
 
     def event_count(self, event_type):
-        return len([event for event in self.events[:]
-                    if isinstance(event, event_type)])
+        return len(self.events_by_type(event_type))
+
+    def events_by_type(self, event_type):
+        """Return the matching events by event class.
+
+        event_type can be a single class or a tuple of classes.
+        """
+        return self.matching(lambda e: isinstance(e, event_type))
+
+    def matching(self, matcher):
+        """Return the matching events."""
+        return [event for event in self.events[:] if matcher(event)]
+
+    def wait_for_event(self, event, count):
+        """Wait for a number of events to be published, or fail."""
+        wait_until(lambda: self.event_count(event) >= count,
+                   'find %s %s event(s)' % (count, event))
+
+
+class CMAPListener(BaseListener, monitoring.ConnectionPoolListener):
 
     def connection_created(self, event):
         self.add_event(event)
@@ -196,25 +214,17 @@ class ServerAndTopologyEventListener(ServerEventListener,
     """Listens to Server and Topology events."""
 
 
-class HeartbeatEventListener(monitoring.ServerHeartbeatListener):
+class HeartbeatEventListener(BaseListener, monitoring.ServerHeartbeatListener):
     """Listens to only server heartbeat events."""
 
-    def __init__(self):
-        self.results = []
-
     def started(self, event):
-        self.results.append(event)
+        self.add_event(event)
 
     def succeeded(self, event):
-        self.results.append(event)
+        self.add_event(event)
 
     def failed(self, event):
-        self.results.append(event)
-
-    def matching(self, matcher):
-        """Return the matching events."""
-        results = self.results[:]
-        return [event for event in results if matcher(event)]
+        self.add_event(event)
 
 
 class MockSocketInfo(object):