Merge ~cgrabowski/maas:fix_connection_pool_handshake_race_condition into maas:master

[Christian Grabowski]

only allow the clusterservice to call add_connection to avoid race condition with rpc handshake

[MAAS Lander (maas-lander)]

UNIT TESTS
-b fix_connection_pool_handshake_race_condition lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas-tester/439/consoleText
COMMIT: 3833aa2744122bff1b69b0cf0b00776f4571a666

review: Needs Fixing

[Alexsander de Souza (alexsander-souza)]

+1

review: Approve

[MAAS Lander (maas-lander)]

UNIT TESTS
-b fix_connection_pool_handshake_race_condition lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: SUCCESS
COMMIT: cdc506149c60aeed840bf0dd951e41fdf0860a53

review: Approve

diff --git a/src/provisioningserver/rpc/clusterservice.py b/src/provisioningserver/rpc/clusterservice.py
index 32d5ad4..3a5045f 100644
--- a/src/provisioningserver/rpc/clusterservice.py
+++ b/src/provisioningserver/rpc/clusterservice.py
@@ -1084,7 +1084,7 @@ class ClusterClient(Cluster):
             log.msg("Event-loop '%s' authenticated." % self.ident)
             registered = yield self.registerRackWithRegion()
             if registered:
-                self.service.add_connection(self.eventloop, self)
+                yield self.service.add_connection(self.eventloop, self)
                 self.ready.set(self.eventloop)
             else:
                 self.transport.loseConnection()
diff --git a/src/provisioningserver/rpc/connectionpool.py b/src/provisioningserver/rpc/connectionpool.py
index 8023f80..3acaf08 100644
--- a/src/provisioningserver/rpc/connectionpool.py
+++ b/src/provisioningserver/rpc/connectionpool.py
@@ -88,7 +88,6 @@ class ConnectionPool:
                 # spawn an extra connection
                 conn_to_clone = random.choice(list(ev_conns))
                 conn = yield self.connect(ev, conn_to_clone.address)
-                self.connections[ev].append(conn)
                 self.clock.callLater(
                     self._keepalive, self._reap_extra_connection, ev, conn
                 )
@@ -144,14 +143,15 @@ class ConnectionPool:
             self.connections[eventloop] = []
 
         self.connections[eventloop].append(connection)
-
         # clone connection to equal num idle connections
-        if self._max_idle_connections - 1 > 0:
-            for _ in range(self._max_idle_connections - 1):
-                extra_conn = yield self.connect(
-                    connection.eventloop, connection.address
-                )
-                self.connections[eventloop].append(extra_conn)
+        idle_limit = self._max_idle_connections - len(
+            self.connections[eventloop]
+        )
+        # if there's room for more and first conn, create more idle conns
+        if idle_limit > 0 and len(self.connections[eventloop]) == 1:
+            for _ in range(idle_limit):
+                # calls to service to add self when handshake is finished
+                yield self.connect(connection.eventloop, connection.address)
 
     def remove_connection(self, eventloop, connection):
         if self.is_staged(eventloop):
diff --git a/src/provisioningserver/rpc/tests/test_clusterservice.py b/src/provisioningserver/rpc/tests/test_clusterservice.py
index cd15e04..ddb0f1c 100644
--- a/src/provisioningserver/rpc/tests/test_clusterservice.py
+++ b/src/provisioningserver/rpc/tests/test_clusterservice.py
@@ -1211,19 +1211,26 @@ class TestClusterClientService(MAASTestCase):
             service._update_saved_rpc_info_state, MockCalledOnceWith()
         )
 
+    @inlineCallbacks
     def test_add_connection_creates_max_idle_connections(self):
         service = make_inert_client_service(max_idle_conns=2)
         service.startService()
         endpoint = Mock()
         connection = Mock()
+        connection.eventloop = endpoint
         connection.address = (":::ffff", 2222)
-        connection2 = Mock()
-        connection.address = (":::ffff", 2222)
-        self.patch(service.connections, "connect").return_value = succeed(
-            connection2
-        )
+
+        @inlineCallbacks
+        def mock_connect(ev, addr):
+            new_conn = Mock()
+            new_conn.eventloop = ev
+            new_conn.address = addr
+            yield service.add_connection(ev, new_conn)
+            return new_conn
+
+        self.patch(service.connections, "connect").side_effect = mock_connect
         self.patch_autospec(service, "_update_saved_rpc_info_state")
-        service.add_connection(endpoint, connection)
+        yield service.add_connection(endpoint, connection)
         self.assertEqual(
             len(
                 [
@@ -1329,11 +1336,25 @@ class TestClusterClientService(MAASTestCase):
         self.patch(service.connections, "connect").return_value = succeed(
             FakeConnection()
         )
+
+        # skip the connectionMade logic for this test
+        @inlineCallbacks
+        def mock_scale_up_connections():
+            for ev, conns in service.connections.items():
+                if len(conns) < service.connections._max_connections:
+                    conn = yield service.connections.connect()
+                    service.connections[ev].append(conn)
+                    break
+
+        scale_up = self.patch(service.connections, "scale_up_connections")
+        scale_up.side_effect = mock_scale_up_connections
+
         original_conns = [
             conn for conns in service.connections.values() for conn in conns
         ]
         new_client = yield service.getClientNow()
         new_conn = new_client._conn
+        scale_up.assert_called_once()
         self.assertIsNotNone(new_conn)
         self.assertNotIn(new_conn, original_conns)
         self.assertIn(
diff --git a/src/provisioningserver/rpc/tests/test_connectionpool.py b/src/provisioningserver/rpc/tests/test_connectionpool.py
index dc06bd4..2f0096c 100644
--- a/src/provisioningserver/rpc/tests/test_connectionpool.py
+++ b/src/provisioningserver/rpc/tests/test_connectionpool.py
@@ -117,10 +117,35 @@ class TestConnectionPool(MAASTestCase):
     def test_scale_up_connections_adds_a_connection(self):
         cp = ConnectionPool(Clock(), Mock(), max_conns=2)
         eventloop = Mock()
-        connection1 = Mock()
-        connection2 = Mock()
+        address = (factory.make_ip_address(), 5240)
+        service = Mock()
+
+        @inlineCallbacks
+        def mock_service_add_connection(ev, conn):
+            yield cp.add_connection(ev, conn)
+
+        service.add_connection = mock_service_add_connection
+
+        connection1 = ClusterClient(address, eventloop, service)
+        connection2 = ClusterClient(address, eventloop, service)
         connect = self.patch(cp, "connect")
-        connect.return_value = succeed(connection2)
+
+        @inlineCallbacks
+        def call_connectionMade(*args, **kwargs):
+            yield connection2.connectionMade()
+            return connection2
+
+        connect.side_effect = call_connectionMade
+
+        authRegion = self.patch(connection2, "authenticateRegion")
+        authRegion.return_value = succeed(True)
+        register = self.patch(connection2, "registerRackWithRegion")
+
+        def set_ident(*args, **kwargs):
+            connection2.localIdent = factory.make_name()
+            return succeed(True)
+
+        register.side_effect = set_ident
         cp[eventloop] = [connection1]
         cp.scale_up_connections()
         self.assertCountEqual(cp[eventloop], [connection1, connection2])