Merge ~cgrabowski/maas:rpc_connection_pool_burst into maas:master

[Christian Grabowski]

allocate additional connections when busy

always connect max idle connections times

Allows RPC connections to scale up to a configured amount when busy.

[Alexsander de Souza (alexsander-souza)]

+1

review: Approve

[MAAS Lander (maas-lander)]

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

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas-tester/300/consoleText
COMMIT: ba71e215175a711cbe340ccfa250bc1a1093ccb1

review: Needs Fixing

[MAAS Lander (maas-lander)]

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

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas-tester/301/consoleText
COMMIT: 74329f97f1dd1a3d08ea1f795ebf9c51cacf5b41

review: Needs Fixing

[MAAS Lander (maas-lander)]

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

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas-tester/302/consoleText
COMMIT: 820009b533a8fbbc449ff7c88af733ef8107fae8

review: Needs Fixing

[MAAS Lander (maas-lander)]

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

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas-tester/303/consoleText
COMMIT: 47a4dfedbee9d93a57484e53d0557f5a29c32396

review: Needs Fixing

[MAAS Lander (maas-lander)]

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

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas-tester/304/consoleText
COMMIT: 8f3912b1979ab12d569d49c7086d8cdbe86f55ba

review: Needs Fixing

[MAAS Lander (maas-lander)]

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

STATUS: SUCCESS
COMMIT: 5f3b0b943fd0ca495036f6b21938da58ddeb6bba

review: Approve

[Adam Collard (adam-collard)]

How do the idle connections work with the _update_interval method of ClusterClientService?

That method wants to control how often to poll the region controller to establish connectivity, and has logic to compare the number of connected connections with the expected (which i think will go wrong with idle/bursting connections).

review: Needs Information

[Adam Collard (adam-collard)]

This broke system tests and could do with a little more love.

Reverting in https://code.launchpad.net/~adam-collard/maas/+git/maas/+merge/428258 to retain green tests and give space for the work

diff --git a/src/provisioningserver/config.py b/src/provisioningserver/config.py
index 97b6e68..9511ad6 100644
--- a/src/provisioningserver/config.py
+++ b/src/provisioningserver/config.py
@@ -762,6 +762,23 @@ class ClusterConfiguration(Configuration, metaclass=ClusterConfigurationMeta):
         ),
     )
 
+    # RPC Connection Pool options
+    max_idle_rpc_connections = ConfigurationOption(
+        "max_idle_rpc_connections",
+        "The nominal number of connections to have per endpoint",
+        Number(min=1, max=1024, if_missing=1),
+    )
+    max_rpc_connections = ConfigurationOption(
+        "max_rpc_connections",
+        "The maximum number of connections to scale to when under load",
+        Number(min=1, max=1024, if_missing=4),
+    )
+    rpc_keepalive = ConfigurationOption(
+        "rpc_keepalive",
+        "The duration in miliseconds to keep added connections alive",
+        Number(min=1, max=5000, if_missing=1000),
+    )
+
     # TFTP options.
     tftp_port = ConfigurationOption(
         "tftp_port",
diff --git a/src/provisioningserver/plugin.py b/src/provisioningserver/plugin.py
index e987c73..00ff898 100644
--- a/src/provisioningserver/plugin.py
+++ b/src/provisioningserver/plugin.py
@@ -139,7 +139,13 @@ class ProvisioningServiceMaker:
     def _makeRPCService(self):
         from provisioningserver.rpc.clusterservice import ClusterClientService
 
-        rpc_service = ClusterClientService(reactor)
+        with ClusterConfiguration.open() as config:
+            rpc_service = ClusterClientService(
+                reactor,
+                config.max_idle_rpc_connections,
+                config.max_rpc_connections,
+                config.rpc_keepalive,
+            )
         rpc_service.setName("rpc")
         return rpc_service
 
diff --git a/src/provisioningserver/rackdservices/external.py b/src/provisioningserver/rackdservices/external.py
index ccabb74..5b8afe5 100644
--- a/src/provisioningserver/rackdservices/external.py
+++ b/src/provisioningserver/rackdservices/external.py
@@ -68,8 +68,9 @@ class RackOnlyExternalService(metaclass=ABCMeta):
 
         # Filter the connects by region.
         conn_per_region = defaultdict(set)
-        for eventloop, connection in connections.items():
-            conn_per_region[eventloop.split(":")[0]].add(connection)
+        for eventloop, connection_set in connections.items():
+            for connection in connection_set:
+                conn_per_region[eventloop.split(":")[0]].add(connection)
         for eventloop, connections in conn_per_region.items():
             # Sort the connections so the same IP is always picked per
             # region controller. This ensures that the HTTP configuration
diff --git a/src/provisioningserver/rackdservices/http.py b/src/provisioningserver/rackdservices/http.py
index 421e35f..bda9d23 100644
--- a/src/provisioningserver/rackdservices/http.py
+++ b/src/provisioningserver/rackdservices/http.py
@@ -101,8 +101,9 @@ class RackHTTPService(TimerService):
         controller is connected to."""
         # Filter the connects by region.
         conn_per_region = defaultdict(set)
-        for eventloop, connection in self._rpc_service.connections.items():
-            conn_per_region[eventloop.split(":")[0]].add(connection)
+        for eventloop, connection_set in self._rpc_service.connections.items():
+            for connection in connection_set:
+                conn_per_region[eventloop.split(":")[0]].add(connection)
         for _, connections in conn_per_region.items():
             # Sort the connections so the same IP is always picked per
             # region controller. This ensures that the HTTP configuration
diff --git a/src/provisioningserver/rackdservices/tests/test_external.py b/src/provisioningserver/rackdservices/tests/test_external.py
index ad214a1..0cb8601 100644
--- a/src/provisioningserver/rackdservices/tests/test_external.py
+++ b/src/provisioningserver/rackdservices/tests/test_external.py
@@ -430,7 +430,8 @@ class TestRackDNS(MAASTestCase):
         return frozenset(
             {
                 client.address[0]
-                for _, client in rpc_service.connections.items()
+                for _, clients in rpc_service.connections.items()
+                for client in clients
             }
         )
 
@@ -609,7 +610,7 @@ class TestRackDNS(MAASTestCase):
                 ip = factory.make_ip_address()
                 mock_conn = Mock()
                 mock_conn.address = (ip, random.randint(5240, 5250))
-                mock_rpc.connections[eventloop] = mock_conn
+                mock_rpc.connections[eventloop] = {mock_conn}
 
         dns = external.RackDNS()
         region_ips = list(dns._genRegionIps(mock_rpc.connections))
@@ -626,7 +627,7 @@ class TestRackDNS(MAASTestCase):
                 ip = factory.make_ip_address()
                 mock_conn = Mock()
                 mock_conn.address = (ip, random.randint(5240, 5250))
-                mock_rpc.connections[eventloop] = mock_conn
+                mock_rpc.connections[eventloop] = {mock_conn}
 
         dns = external.RackDNS()
         region_ips = frozenset(dns._genRegionIps(mock_rpc.connections))
@@ -659,7 +660,8 @@ class TestRackProxy(MAASTestCase):
         return frozenset(
             {
                 client.address[0]
-                for _, client in rpc_service.connections.items()
+                for _, clients in rpc_service.connections.items()
+                for client in clients
             }
         )
 
@@ -824,7 +826,8 @@ class TestRackSyslog(MAASTestCase):
         return frozenset(
             {
                 (eventloop, client.address[0])
-                for eventloop, client in rpc_service.connections.items()
+                for eventloop, clients in rpc_service.connections.items()
+                for client in clients
             }
         )
 
diff --git a/src/provisioningserver/rackdservices/tests/test_http.py b/src/provisioningserver/rackdservices/tests/test_http.py
index bc43c66..43cb495 100644
--- a/src/provisioningserver/rackdservices/tests/test_http.py
+++ b/src/provisioningserver/rackdservices/tests/test_http.py
@@ -92,7 +92,8 @@ class TestRackHTTPService(MAASTestCase):
         return frozenset(
             {
                 client.address[0]
-                for _, client in rpc_service.connections.items()
+                for _, clients in rpc_service.connections.items()
+                for client in clients
             }
         )
 
@@ -208,7 +209,7 @@ class TestRackHTTPService(MAASTestCase):
                 ip = factory.make_ip_address()
                 mock_conn = Mock()
                 mock_conn.address = (ip, random.randint(5240, 5250))
-                mock_rpc.connections[eventloop] = mock_conn
+                mock_rpc.connections[eventloop] = {mock_conn}
 
         service = http.RackHTTPService(self.make_dir(), mock_rpc, reactor)
         region_ips = list(service._genRegionIps())
@@ -225,7 +226,7 @@ class TestRackHTTPService(MAASTestCase):
                 ip = factory.make_ip_address()
                 mock_conn = Mock()
                 mock_conn.address = (ip, random.randint(5240, 5250))
-                mock_rpc.connections[eventloop] = mock_conn
+                mock_rpc.connections[eventloop] = {mock_conn}
 
         service = http.RackHTTPService(self.make_dir(), mock_rpc, reactor)
         region_ips = frozenset(service._genRegionIps())
@@ -244,7 +245,7 @@ class TestRackHTTPService(MAASTestCase):
             ip_addresses.add("[%s]" % ip)
             mock_conn = Mock()
             mock_conn.address = (ip, random.randint(5240, 5250))
-            mock_rpc.connections[eventloop] = mock_conn
+            mock_rpc.connections[eventloop] = {mock_conn}
 
         service = http.RackHTTPService(self.make_dir(), mock_rpc, reactor)
         region_ips = set(service._genRegionIps())
diff --git a/src/provisioningserver/rpc/clusterservice.py b/src/provisioningserver/rpc/clusterservice.py
index c92d48a..7b94e49 100644
--- a/src/provisioningserver/rpc/clusterservice.py
+++ b/src/provisioningserver/rpc/clusterservice.py
@@ -999,6 +999,7 @@ class ClusterClient(Cluster):
         # Events for this protocol's life-cycle.
         self.authenticated = DeferredValue()
         self.ready = DeferredValue()
+        self.in_use = False
         self.localIdent = None
 
     @property
@@ -1201,7 +1202,7 @@ class ClusterClientService(TimerService):
 
     time_started = None
 
-    def __init__(self, reactor):
+    def __init__(self, reactor, max_idle_conns=1, max_conns=1, keepalive=1000):
         super().__init__(self._calculate_interval(None, None), self._tryUpdate)
         self.connections = {}
         self.try_connections = {}
@@ -1224,10 +1225,40 @@ class ClusterClientService(TimerService):
         self._updateInProgress = DeferredValue()
         self._updateInProgress.set(None)
 
+        # The maximum number of connections to allways allocate per eventloop
+        self._max_idle_connections = max_idle_conns
+        # The maximum number of connections to allocate while under load per eventloop
+        self._max_connections = max_conns
+        # The duration in milliseconds to keep scaled up connections alive
+        self._keepalive = keepalive
+
     def startService(self):
         self.time_started = self.clock.seconds()
         super().startService()
 
+    def _reap_extra_connection(self, eventloop, conn):
+        if not conn.in_use:
+            self._drop_connection(conn)
+            return self._remove_connection(eventloop, conn)
+        return self.clock.callLater(
+            self._keepalive, self._reap_extra_connection, conn
+        )
+
+    @inlineCallbacks
+    def _scale_up_connections(self):
+        for ev, ev_conns in self.connections.items():
+            # pick first group with room for additional conns
+            if len(ev_conns) < self._max_connections:
+                # spawn an extra connection
+                conn_to_clone = random.choice(list(ev_conns))
+                conn = yield self._make_connection(ev, conn_to_clone.address)
+                self.connections[ev].add(conn)
+                self.clock.callLater(
+                    self._keepalive, self._reap_extra_connection, ev, conn
+                )
+                return
+        raise exceptions.MaxConnectionsOpen
+
     def getClient(self):
         """Returns a :class:`common.Client` connected to a region.
 
@@ -1236,11 +1267,22 @@ class ClusterClientService(TimerService):
         :raises: :py:class:`~.exceptions.NoConnectionsAvailable` when
             there are no open connections to a region controller.
         """
-        conns = list(self.connections.values())
+        conns = [
+            conn for conn_set in self.connections.values() for conn in conn_set
+        ]
         if len(conns) == 0:
             raise exceptions.NoConnectionsAvailable()
         else:
-            return common.Client(random.choice(conns))
+            free_conns = [conn for conn in conns if not conn.in_use]
+            if len(free_conns) > 0:
+                return common.Client(random.choice(free_conns))
+            else:
+                for endpoint_conns in self.connections.values():
+                    if len(endpoint_conns) < self._max_connections:
+                        # caller should create a new connection
+                        raise exceptions.AllConnectionsBusy
+                # return a busy connection, assume it will free up or timeout
+                return common.Client(random.choice(conns))
 
     @deferred
     def getClientNow(self):
@@ -1259,10 +1301,18 @@ class ClusterClientService(TimerService):
             return self.getClient()
         except exceptions.NoConnectionsAvailable:
             return self._tryUpdate().addCallback(call, self.getClient)
+        except exceptions.AllConnectionsBusy:
+            return self._scale_up_connections().addCallback(
+                call, self.getClient
+            )
 
     def getAllClients(self):
         """Return a list of all connected :class:`common.Client`s."""
-        return [common.Client(conn) for conn in self.connections.values()]
+        return [
+            common.Client(conn)
+            for conns in self.connections.values()
+            for conn in conns
+        ]
 
     def _tryUpdate(self):
         """Attempt to refresh outgoing connections.
@@ -1391,7 +1441,9 @@ class ClusterClientService(TimerService):
         """Update the saved RPC info state."""
         # Build a list of addresses based on the current connections.
         connected_addr = {
-            conn.address[0] for _, conn in self.connections.items()
+            conn.address[0]
+            for _, conns in self.connections.items()
+            for conn in conns
         }
         if (
             self._rpc_info_state is None
@@ -1761,6 +1813,7 @@ class ClusterClientService(TimerService):
         """Drop the given `connection`."""
         return connection.transport.loseConnection()
 
+    @inlineCallbacks
     def add_connection(self, eventloop, connection):
         """Add the connection to the tracked connections.
 
@@ -1769,7 +1822,16 @@ class ClusterClientService(TimerService):
         """
         if eventloop in self.try_connections:
             del self.try_connections[eventloop]
-        self.connections[eventloop] = connection
+        if not self.connections.get(eventloop):
+            self.connections[eventloop] = set()
+        self.connections[eventloop].add(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._make_connection(
+                    connection.eventloop, connection.address
+                )
+                self.connections[eventloop].add(extra_conn)
         self._update_saved_rpc_info_state()
 
     def remove_connection(self, eventloop, connection):
@@ -1782,8 +1844,10 @@ class ClusterClientService(TimerService):
             if self.try_connections[eventloop] is connection:
                 del self.try_connections[eventloop]
         if eventloop in self.connections:
-            if self.connections[eventloop] is connection:
-                del self.connections[eventloop]
+            if connection in self.connections.get(eventloop, set()):
+                self.connections[eventloop].discard(connection)
+                if len(self.connections[eventloop]) == 0:
+                    del self.connections[eventloop]
         # Disable DHCP when no connections to a region controller.
         if len(self.connections) == 0:
             stopping_services = []
diff --git a/src/provisioningserver/rpc/common.py b/src/provisioningserver/rpc/common.py
index 5d67bba..40e091f 100644
--- a/src/provisioningserver/rpc/common.py
+++ b/src/provisioningserver/rpc/common.py
@@ -14,7 +14,11 @@ from twisted.python.failure import Failure
 from provisioningserver.logger import LegacyLogger
 from provisioningserver.prometheus.metrics import PROMETHEUS_METRICS
 from provisioningserver.rpc.interfaces import IConnection, IConnectionToRegion
-from provisioningserver.utils.twisted import asynchronous, deferWithTimeout
+from provisioningserver.utils.twisted import (
+    asynchronous,
+    callOut,
+    deferWithTimeout,
+)
 
 log = LegacyLogger()
 
@@ -156,6 +160,11 @@ class Client:
         :return: A deferred result.  Call its `wait` method (with a timeout
             in seconds) to block on the call's completion.
         """
+        self._conn.in_use = True
+
+        def _free_conn():
+            self._conn.in_use = False
+
         if len(args) != 0:
             receiver_name = "{}.{}".format(
                 self.__module__,
@@ -171,11 +180,19 @@ class Client:
         if timeout is undefined:
             timeout = 120  # 2 minutes
         if timeout is None or timeout <= 0:
-            return self._conn.callRemote(cmd, **kwargs)
+            d = self._conn.callRemote(cmd, **kwargs)
+            if isinstance(d, Deferred):
+                d.addBoth(lambda x: callOut(x, _free_conn))
+            else:
+                _free_conn()
+            return d
         else:
-            return deferWithTimeout(
-                timeout, self._conn.callRemote, cmd, **kwargs
-            )
+            d = deferWithTimeout(timeout, self._conn.callRemote, cmd, **kwargs)
+            if isinstance(d, Deferred):
+                d.addBoth(lambda x: callOut(x, _free_conn))
+            else:
+                _free_conn()
+            return d
 
     @asynchronous
     def getHostCertificate(self):
diff --git a/src/provisioningserver/rpc/exceptions.py b/src/provisioningserver/rpc/exceptions.py
index 7ee4f3f..136e471 100644
--- a/src/provisioningserver/rpc/exceptions.py
+++ b/src/provisioningserver/rpc/exceptions.py
@@ -12,6 +12,14 @@ class NoConnectionsAvailable(Exception):
         self.uuid = uuid
 
 
+class AllConnectionsBusy(Exception):
+    """The current connection pool is busy"""
+
+
+class MaxConnectionsOpen(Exception):
+    """The maxmimum number of connections are currently open"""
+
+
 class NoSuchEventType(Exception):
     """The specified event type was not found."""
 
diff --git a/src/provisioningserver/rpc/testing/__init__.py b/src/provisioningserver/rpc/testing/__init__.py
index ee4a9e2..1b2f94f 100644
--- a/src/provisioningserver/rpc/testing/__init__.py
+++ b/src/provisioningserver/rpc/testing/__init__.py
@@ -262,7 +262,8 @@ class MockClusterToRegionRPCFixtureBase(fixtures.Fixture, metaclass=ABCMeta):
             {
                 "eventloops": {
                     eventloop: [client.address]
-                    for eventloop, client in connections
+                    for eventloop, clients in connections
+                    for client in clients
                 }
             },
             orig_url,
diff --git a/src/provisioningserver/rpc/testing/doubles.py b/src/provisioningserver/rpc/testing/doubles.py
index cb9f27f..0785859 100644
--- a/src/provisioningserver/rpc/testing/doubles.py
+++ b/src/provisioningserver/rpc/testing/doubles.py
@@ -30,6 +30,7 @@ class FakeConnection:
     ident = attr.ib(default=sentinel.ident)
     hostCertificate = attr.ib(default=sentinel.hostCertificate)
     peerCertificate = attr.ib(default=sentinel.peerCertificate)
+    in_use = attr.ib(default=False)
 
     def callRemote(self, cmd, **arguments):
         return succeed(sentinel.response)
@@ -48,6 +49,7 @@ class FakeConnectionToRegion:
     address = attr.ib(default=(sentinel.host, sentinel.port))
     hostCertificate = attr.ib(default=sentinel.hostCertificate)
     peerCertificate = attr.ib(default=sentinel.peerCertificate)
+    in_use = attr.ib(default=False)
 
     def callRemote(self, cmd, **arguments):
         return succeed(sentinel.response)
@@ -56,6 +58,22 @@ class FakeConnectionToRegion:
 verifyObject(IConnectionToRegion, FakeConnectionToRegion())
 
 
+@attr.s(eq=False, order=False)
+@implementer(IConnectionToRegion)
+class FakeBusyConnectionToRegion:
+    "A fake `IConnectionToRegion` that appears busy." ""
+
+    ident = attr.ib(default=sentinel.ident)
+    localIdent = attr.ib(default=sentinel.localIdent)
+    address = attr.ib(default=(sentinel.host, sentinel.port))
+    hostCertificate = attr.ib(default=sentinel.hostCertificate)
+    peerCertificate = attr.ib(default=sentinel.peerCertificate)
+    in_use = attr.ib(default=True)
+
+    def callRemote(self, cmd, **arguments):
+        return succeed(sentinel.response)
+
+
 class StubOS(OperatingSystem):
     """An :py:class:`OperatingSystem` subclass that has canned answers.
 
diff --git a/src/provisioningserver/rpc/tests/test_clusterservice.py b/src/provisioningserver/rpc/tests/test_clusterservice.py
index b50311d..fab455a 100644
--- a/src/provisioningserver/rpc/tests/test_clusterservice.py
+++ b/src/provisioningserver/rpc/tests/test_clusterservice.py
@@ -117,7 +117,11 @@ from provisioningserver.rpc.testing import (
     call_responder,
     MockLiveClusterToRegionRPCFixture,
 )
-from provisioningserver.rpc.testing.doubles import DummyConnection, StubOS
+from provisioningserver.rpc.testing.doubles import (
+    FakeBusyConnectionToRegion,
+    FakeConnection,
+    StubOS,
+)
 from provisioningserver.security import set_shared_secret_on_filesystem
 from provisioningserver.service_monitor import service_monitor
 from provisioningserver.testing.config import ClusterConfigurationFixture
@@ -444,8 +448,10 @@ class TestClusterProtocol_DescribePowerTypes(MAASTestCase):
         )
 
 
-def make_inert_client_service():
-    service = ClusterClientService(Clock())
+def make_inert_client_service(max_idle_conns=1, max_conns=1, keepalive=1):
+    service = ClusterClientService(
+        Clock(), max_idle_conns, max_conns, keepalive
+    )
     # ClusterClientService's superclass, TimerService, creates a
     # LoopingCall with now=True. We neuter it here to allow
     # observation of the behaviour of _update_interval() for
@@ -499,10 +505,10 @@ class TestClusterClientService(MAASTestCase):
 
         # Fake some connections.
         service.connections = {
-            ipv4client.eventloop: ipv4client,
-            ipv6client.eventloop: ipv6client,
-            ipv6mapped.eventloop: ipv6mapped,
-            hostclient.eventloop: hostclient,
+            ipv4client.eventloop: {ipv4client},
+            ipv6client.eventloop: {ipv6client},
+            ipv6mapped.eventloop: {ipv6mapped},
+            hostclient.eventloop: {hostclient},
         }
 
         # Update the RPC state to the filesystem and info cache.
@@ -515,7 +521,8 @@ class TestClusterClientService(MAASTestCase):
             Equals(
                 {
                     client.address[0]
-                    for _, client in service.connections.items()
+                    for _, clients in service.connections.items()
+                    for client in clients
                 }
             ),
         )
@@ -1234,7 +1241,7 @@ class TestClusterClientService(MAASTestCase):
         connection = Mock()
         connection.address = (":::ffff", 2222)
         service.add_connection(endpoint, connection)
-        self.assertThat(service.connections, Equals({endpoint: connection}))
+        self.assertEqual(service.connections, {endpoint: {connection}})
 
     def test_add_connection_calls__update_saved_rpc_info_state(self):
         service = make_inert_client_service()
@@ -1248,6 +1255,30 @@ class TestClusterClientService(MAASTestCase):
             service._update_saved_rpc_info_state, MockCalledOnceWith()
         )
 
+    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.address = (":::ffff", 2222)
+        connection2 = Mock()
+        connection.address = (":::ffff", 2222)
+        self.patch(service, "_make_connection").return_value = succeed(
+            connection2
+        )
+        self.patch_autospec(service, "_update_saved_rpc_info_state")
+        service.add_connection(endpoint, connection)
+        self.assertEqual(
+            len(
+                [
+                    conn
+                    for conns in service.connections.values()
+                    for conn in conns
+                ]
+            ),
+            service._max_idle_connections,
+        )
+
     def test_remove_connection_removes_from_try_connections(self):
         service = make_inert_client_service()
         service.startService()
@@ -1262,7 +1293,7 @@ class TestClusterClientService(MAASTestCase):
         service.startService()
         endpoint = Mock()
         connection = Mock()
-        service.connections[endpoint] = connection
+        service.connections[endpoint] = {connection}
         service.remove_connection(endpoint, connection)
         self.assertThat(service.connections, Equals({}))
 
@@ -1271,7 +1302,7 @@ class TestClusterClientService(MAASTestCase):
         service.startService()
         endpoint = Mock()
         connection = Mock()
-        service.connections[endpoint] = connection
+        service.connections[endpoint] = {connection}
         service.remove_connection(endpoint, connection)
         self.assertEqual(service.step, service.INTERVAL_LOW)
 
@@ -1280,7 +1311,7 @@ class TestClusterClientService(MAASTestCase):
         service.startService()
         endpoint = Mock()
         connection = Mock()
-        service.connections[endpoint] = connection
+        service.connections[endpoint] = {connection}
 
         # Enable both dhcpd and dhcpd6.
         service_monitor.getServiceByName("dhcpd").on()
@@ -1295,13 +1326,17 @@ class TestClusterClientService(MAASTestCase):
     def test_getClient(self):
         service = ClusterClientService(Clock())
         service.connections = {
-            sentinel.eventloop01: DummyConnection(),
-            sentinel.eventloop02: DummyConnection(),
-            sentinel.eventloop03: DummyConnection(),
+            sentinel.eventloop01: {FakeConnection()},
+            sentinel.eventloop02: {FakeConnection()},
+            sentinel.eventloop03: {FakeConnection()},
         }
         self.assertIn(
             service.getClient(),
-            {common.Client(conn) for conn in service.connections.values()},
+            {
+                common.Client(conn)
+                for conns in service.connections.values()
+                for conn in conns
+            },
         )
 
     def test_getClient_when_there_are_no_connections(self):
@@ -1310,17 +1345,65 @@ class TestClusterClientService(MAASTestCase):
         self.assertRaises(exceptions.NoConnectionsAvailable, service.getClient)
 
     @inlineCallbacks
+    def test_getClientNow_scales_connections_when_busy(self):
+        service = ClusterClientService(Clock(), max_conns=2)
+        service.connections = {
+            sentinel.eventloop01: {FakeBusyConnectionToRegion()},
+            sentinel.eventloop02: {FakeBusyConnectionToRegion()},
+            sentinel.eventloop03: {FakeBusyConnectionToRegion()},
+        }
+        self.patch(service, "_make_connection").return_value = succeed(
+            FakeConnection()
+        )
+        original_conns = [
+            conn for conns in service.connections.values() for conn in conns
+        ]
+        new_client = yield service.getClientNow()
+        new_conn = new_client._conn
+        self.assertIsNotNone(new_conn)
+        self.assertNotIn(new_conn, original_conns)
+        self.assertIn(
+            new_conn,
+            [conn for conns in service.connections.values() for conn in conns],
+        )
+
+    @inlineCallbacks
+    def test_getClientNow_returns_an_existing_connection_when_max_are_open(
+        self,
+    ):
+        service = ClusterClientService(Clock(), max_conns=1)
+        service.connections = {
+            sentinel.eventloop01: {FakeBusyConnectionToRegion()},
+            sentinel.eventloop02: {FakeBusyConnectionToRegion()},
+            sentinel.eventloop03: {FakeBusyConnectionToRegion()},
+        }
+        self.patch(service, "_make_connection").return_value = succeed(
+            FakeConnection()
+        )
+        original_conns = [
+            conn for conns in service.connections.values() for conn in conns
+        ]
+        new_client = yield service.getClientNow()
+        new_conn = new_client._conn
+        self.assertIsNotNone(new_conn)
+        self.assertIn(new_conn, original_conns)
+
+    @inlineCallbacks
     def test_getClientNow_returns_current_connection(self):
         service = ClusterClientService(Clock())
         service.connections = {
-            sentinel.eventloop01: DummyConnection(),
-            sentinel.eventloop02: DummyConnection(),
-            sentinel.eventloop03: DummyConnection(),
+            sentinel.eventloop01: {FakeConnection()},
+            sentinel.eventloop02: {FakeConnection()},
+            sentinel.eventloop03: {FakeConnection()},
         }
         client = yield service.getClientNow()
         self.assertIn(
             client,
-            {common.Client(conn) for conn in service.connections.values()},
+            {
+                common.Client(conn)
+                for conns in service.connections.values()
+                for conn in conns
+            },
         )
 
     @inlineCallbacks
@@ -1330,9 +1413,9 @@ class TestClusterClientService(MAASTestCase):
 
         def addConnections():
             service.connections = {
-                sentinel.eventloop01: DummyConnection(),
-                sentinel.eventloop02: DummyConnection(),
-                sentinel.eventloop03: DummyConnection(),
+                sentinel.eventloop01: {FakeConnection()},
+                sentinel.eventloop02: {FakeConnection()},
+                sentinel.eventloop03: {FakeConnection()},
             }
             return succeed(None)
 
@@ -1340,7 +1423,11 @@ class TestClusterClientService(MAASTestCase):
         client = yield service.getClientNow()
         self.assertIn(
             client,
-            {common.Client(conn) for conn in service.connections.values()},
+            {
+                common.Client(conn)
+                for conns in service.connections.values()
+                for conn in conns
+            },
         )
 
     def test_getClientNow_raises_exception_when_no_clients(self):
@@ -1383,11 +1470,11 @@ class TestClusterClientService(MAASTestCase):
     def test_getAllClients(self):
         service = ClusterClientService(Clock())
         uuid1 = factory.make_UUID()
-        c1 = DummyConnection()
-        service.connections[uuid1] = c1
+        c1 = FakeConnection()
+        service.connections[uuid1] = {c1}
         uuid2 = factory.make_UUID()
-        c2 = DummyConnection()
-        service.connections[uuid2] = c2
+        c2 = FakeConnection()
+        service.connections[uuid2] = {c2}
         clients = service.getAllClients()
         self.assertEqual(clients, [common.Client(c1), common.Client(c2)])
 
@@ -1396,6 +1483,26 @@ class TestClusterClientService(MAASTestCase):
         service.connections = {}
         self.assertThat(service.getAllClients(), Equals([]))
 
+    @inlineCallbacks
+    def test__reap_extra_connection_reaps_a_scaled_up_connection(self):
+        clock = Clock()
+        service = ClusterClientService(clock, max_conns=2, keepalive=0)
+        service.connections = {
+            sentinel.eventloop01: {FakeBusyConnectionToRegion()},
+            sentinel.eventloop02: {FakeBusyConnectionToRegion()},
+            sentinel.eventloop03: {FakeBusyConnectionToRegion()},
+        }
+        self.patch(service, "_make_connection").return_value = succeed(
+            FakeConnection()
+        )
+        reap_call = self.patch(service, "_reap_extra_connection")
+        new_client = yield service.getClientNow()
+        delayed_calls = clock.getDelayedCalls()
+        self.assertEqual(len(delayed_calls), 1)
+        delayed_call = delayed_calls[0]
+        self.assertIn(new_client._conn, delayed_call.args)
+        self.assertEqual(reap_call.__name__, delayed_call.func.__name__)
+
 
 class TestClusterClientServiceIntervals(MAASTestCase):
 
@@ -1562,14 +1669,14 @@ class TestClusterClient(MAASTestCase):
         self.assertEqual(client.eventloop, extract_result(wait_for_ready))
         self.assertEqual(client.service.try_connections, {})
         self.assertEqual(
-            client.service.connections, {client.eventloop: client}
+            client.service.connections, {client.eventloop: {client}}
         )
 
     def test_disconnects_when_there_is_an_existing_connection(self):
         client = self.make_running_client()
 
         # Pretend that a connection already exists for this address.
-        client.service.connections[client.eventloop] = sentinel.connection
+        client.service.connections[client.eventloop] = {sentinel.connection}
 
         # Connect via an in-memory transport.
         transport = StringTransportWithDisconnection()
@@ -1586,7 +1693,8 @@ class TestClusterClient(MAASTestCase):
         # The connections list is unchanged because the new connection
         # immediately disconnects.
         self.assertEqual(
-            client.service.connections, {client.eventloop: sentinel.connection}
+            client.service.connections,
+            {client.eventloop: {sentinel.connection}},
         )
         self.assertFalse(client.connected)
         self.assertIsNone(client.transport)