Merge ~cgrabowski/maas:rpc_connection_pool_burst into maas:master

[Christian Grabowski]

move connection lifecycle logic into ConnectionPool

allocate additional connections when busy

always connect max idle connections times

[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/318/consoleText
COMMIT: c19b8616c6acc5cde30fa33916ad40da3d3b04e5

review: Needs Fixing

[Adam Collard (adam-collard)]

jenkins: !test

[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/321/consoleText
COMMIT: c19b8616c6acc5cde30fa33916ad40da3d3b04e5

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: 0c5564670e61e93261925cf91a46850824536a77

review: Approve

[Alexsander de Souza (alexsander-souza)]

LGTM

review: Approve

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/dhcp/tests/test_config.py b/src/provisioningserver/dhcp/tests/test_config.py
index c53e906..8c3f2fe 100644
--- a/src/provisioningserver/dhcp/tests/test_config.py
+++ b/src/provisioningserver/dhcp/tests/test_config.py
@@ -176,7 +176,7 @@ def validate_dhcpd_configuration(test, configuration, ipv6):
                 ),
             ),
         )
-        cmd = (
+        cmd = [
             "dhcpd",
             ("-6" if ipv6 else "-4"),
             "-t",
@@ -184,7 +184,7 @@ def validate_dhcpd_configuration(test, configuration, ipv6):
             conffile.name,
             "-lf",
             leasesfile.name,
-        )
+        ]
         if not running_in_docker():
             # Call `dhcpd` without AppArmor confinement, so that it can read
             # configurations file from /tmp.  This is not needed when running
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..a7205db 100644
--- a/src/provisioningserver/rpc/clusterservice.py
+++ b/src/provisioningserver/rpc/clusterservice.py
@@ -9,7 +9,6 @@ import json
 from operator import itemgetter
 import os
 from os import urandom
-import random
 from socket import AF_INET, AF_INET6, gethostname
 import sys
 from urllib.parse import urlparse
@@ -24,7 +23,6 @@ from twisted.internet.defer import (
     maybeDeferred,
     returnValue,
 )
-from twisted.internet.endpoints import connectProtocol, TCP6ClientEndpoint
 from twisted.internet.error import ConnectError, ConnectionClosed, ProcessDone
 from twisted.internet.threads import deferToThread
 from twisted.protocols import amp
@@ -67,6 +65,7 @@ from provisioningserver.rpc.boot_images import (
     list_boot_images,
 )
 from provisioningserver.rpc.common import Ping, RPCProtocol
+from provisioningserver.rpc.connectionpool import ConnectionPool
 from provisioningserver.rpc.exceptions import CannotConfigureDHCP
 from provisioningserver.rpc.interfaces import IConnectionToRegion
 from provisioningserver.rpc.osystems import (
@@ -999,6 +998,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,13 +1201,15 @@ 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 = {}
         self._previous_work = (None, None)
         self.clock = reactor
 
+        self.connections = ConnectionPool(
+            reactor, self, max_idle_conns, max_conns, keepalive
+        )
+
         # Stored the URL used to connect to the region controller. This will be
         # the URL that was used to get the eventloops.
         self.maas_url = None
@@ -1236,11 +1238,19 @@ class ClusterClientService(TimerService):
         :raises: :py:class:`~.exceptions.NoConnectionsAvailable` when
             there are no open connections to a region controller.
         """
-        conns = list(self.connections.values())
-        if len(conns) == 0:
+        if len(self.connections) == 0:
             raise exceptions.NoConnectionsAvailable()
         else:
-            return common.Client(random.choice(conns))
+            try:
+                return common.Client(
+                    self.connections.get_random_free_connection()
+                )
+            except exceptions.AllConnectionsBusy as e:
+                for endpoint_conns in self.connections.values():
+                    if len(endpoint_conns) < self.connections._max_connections:
+                        raise e
+                # return a busy connection, assume it will free up or timeout
+                return common.Client(self.connections.get_random_connection())
 
     @deferred
     def getClientNow(self):
@@ -1259,10 +1269,17 @@ class ClusterClientService(TimerService):
             return self.getClient()
         except exceptions.NoConnectionsAvailable:
             return self._tryUpdate().addCallback(call, self.getClient)
+        except exceptions.AllConnectionsBusy:
+            return self.connections.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 conn in self.connections.get_all_connections()
+        ]
 
     def _tryUpdate(self):
         """Attempt to refresh outgoing connections.
@@ -1391,7 +1408,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
@@ -1467,8 +1486,8 @@ class ClusterClientService(TimerService):
             # Gather the list of successful responses.
             successful = []
             errors = []
-            for sucess, result in results:
-                if sucess:
+            for success, result in results:
+                if success:
                     body, orig_url = result
                     eventloops = body.get("eventloops")
                     if eventloops is not None:
@@ -1656,12 +1675,15 @@ class ClusterClientService(TimerService):
                 "Dropping connections to event-loops: %s"
                 % (", ".join(drop.keys()))
             )
+            drop_defers = []
+            for eventloop, connections in drop.items():
+                for connection in connections:
+                    drop_defers.append(
+                        maybeDeferred(self.connections.disconnect, connection)
+                    )
+                    self.connections.remove_connection(eventloop, connection)
             yield DeferredList(
-                [
-                    maybeDeferred(self._drop_connection, connection)
-                    for eventloop, connections in drop.items()
-                    for connection in connections
-                ],
+                drop_defers,
                 consumeErrors=True,
             )
 
@@ -1692,11 +1714,12 @@ class ClusterClientService(TimerService):
         # between consenting adults.
         for eventloop, addresses in eventloops.items():
             if eventloop in self.connections:
-                connection = self.connections[eventloop]
-                if connection.address not in addresses:
-                    drop[eventloop] = [connection]
-            if eventloop in self.try_connections:
-                connection = self.try_connections[eventloop]
+                connection_list = self.connections[eventloop]
+                for connection in connection_list:
+                    if connection.address not in addresses:
+                        drop[eventloop] = [connection]
+            if self.connections.is_staged(eventloop):
+                connection = self.connections.get_staged_connection(eventloop)
                 if connection.address not in addresses:
                     drop[eventloop] = [connection]
 
@@ -1705,7 +1728,7 @@ class ClusterClientService(TimerService):
         for eventloop, addresses in eventloops.items():
             if (
                 eventloop not in self.connections
-                and eventloop not in self.try_connections
+                and not self.connections.is_staged(eventloop)
             ) or eventloop in drop:
                 connect[eventloop] = addresses
 
@@ -1714,13 +1737,13 @@ class ClusterClientService(TimerService):
         # the process in which the event-loop is no longer running, but
         # it could be an indicator of a heavily loaded machine, or a
         # fault. In any case, it seems to make sense to disconnect.
-        for eventloop in self.connections:
+        for eventloop in self.connections.keys():
             if eventloop not in eventloops:
-                connection = self.connections[eventloop]
-                drop[eventloop] = [connection]
-        for eventloop in self.try_connections:
+                connection_list = self.connections[eventloop]
+                drop[eventloop] = connection_list
+        for eventloop in self.connections.get_staged_connections():
             if eventloop not in eventloops:
-                connection = self.try_connections[eventloop]
+                connection = self.connections.get_staged_connection(eventloop)
                 drop[eventloop] = [connection]
 
         return drop, connect
@@ -1730,7 +1753,7 @@ class ClusterClientService(TimerService):
         """Connect to `eventloop` using all `addresses`."""
         for address in addresses:
             try:
-                connection = yield self._make_connection(eventloop, address)
+                connection = yield self.connections.connect(eventloop, address)
             except ConnectError as error:
                 host, port = address
                 log.msg(
@@ -1747,29 +1770,17 @@ class ClusterClientService(TimerService):
                     ),
                 )
             else:
-                self.try_connections[eventloop] = connection
+                self.connections.stage_connection(eventloop, connection)
                 break
 
-    def _make_connection(self, eventloop, address):
-        """Connect to `eventloop` at `address`."""
-        # Force everything to use AF_INET6 sockets.
-        endpoint = TCP6ClientEndpoint(self.clock, *address)
-        protocol = ClusterClient(address, eventloop, self)
-        return connectProtocol(endpoint, protocol)
-
-    def _drop_connection(self, connection):
-        """Drop the given `connection`."""
-        return connection.transport.loseConnection()
-
+    @inlineCallbacks
     def add_connection(self, eventloop, connection):
         """Add the connection to the tracked connections.
 
         Update the saved RPC info state information based on the new
         connection.
         """
-        if eventloop in self.try_connections:
-            del self.try_connections[eventloop]
-        self.connections[eventloop] = connection
+        yield self.connections.add_connection(eventloop, connection)
         self._update_saved_rpc_info_state()
 
     def remove_connection(self, eventloop, connection):
@@ -1778,12 +1789,7 @@ class ClusterClientService(TimerService):
         If this is the last connection that was keeping rackd connected to
         a regiond then dhcpd and dhcpd6 services will be turned off.
         """
-        if eventloop in self.try_connections:
-            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]
+        self.connections.remove_connection(eventloop, connection)
         # 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/connectionpool.py b/src/provisioningserver/rpc/connectionpool.py
new file mode 100644
index 0000000..8023f80
--- /dev/null
+++ b/src/provisioningserver/rpc/connectionpool.py
@@ -0,0 +1,163 @@
+# Copyright 2022 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+""" RPC Connection Pooling and Lifecycle """
+
+import random
+
+from twisted.internet.defer import inlineCallbacks
+from twisted.internet.endpoints import connectProtocol, TCP6ClientEndpoint
+
+from provisioningserver.rpc import exceptions
+
+
+class ConnectionPool:
+    def __init__(
+        self, reactor, service, max_idle_conns=1, max_conns=1, keepalive=1000
+    ):
+        # 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
+
+        self.connections = {}
+        self.try_connections = {}
+        self.clock = reactor
+        self._service = service
+
+    def __setitem__(self, key, item):
+        self.connections[key] = item
+
+    def __getitem__(self, key):
+        return self.connections.get(key)
+
+    def __repr__(self):
+        return repr(self.connections)
+
+    def __len__(self):
+        return len(self.get_all_connections())
+
+    def __delitem__(self, key):
+        del self.connections[key]
+
+    def __contains__(self, item):
+        return item in self.connections
+
+    def __cmp__(self, value):
+        return self.connections.__cmp__(value)
+
+    def __eq__(self, value):
+        return self.connections.__eq__(value)
+
+    def keys(self):
+        return self.connections.keys()
+
+    def values(self):
+        return self.connections.values()
+
+    def items(self):
+        return self.connections.items()
+
+    def _reap_extra_connection(self, eventloop, conn):
+        if not conn.in_use:
+            self.disconnect(conn)
+            return self.remove_connection(eventloop, conn)
+        return self.clock.callLater(
+            self._keepalive, self._reap_extra_connection, eventloop, conn
+        )
+
+    def is_staged(self, eventloop):
+        return eventloop in self.try_connections
+
+    def get_staged_connection(self, eventloop):
+        return self.try_connections.get(eventloop)
+
+    def get_staged_connections(self):
+        return self.try_connections
+
+    def stage_connection(self, eventloop, connection):
+        self.try_connections[eventloop] = connection
+
+    @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.connect(ev, conn_to_clone.address)
+                self.connections[ev].append(conn)
+                self.clock.callLater(
+                    self._keepalive, self._reap_extra_connection, ev, conn
+                )
+                return
+        raise exceptions.MaxConnectionsOpen()
+
+    def get_connection(self, eventloop):
+        return random.choice(self.connections[eventloop])
+
+    def get_random_connection(self):
+        return random.choice(self.get_all_connections())
+
+    def get_random_free_connection(self):
+        free_conns = self.get_all_free_connections()
+        if len(free_conns) == 0:
+            # caller should create a new connection
+            raise exceptions.AllConnectionsBusy()
+        return random.choice(free_conns)
+
+    def get_all_connections(self):
+        return [
+            conn
+            for conn_list in self.connections.values()
+            for conn in conn_list
+        ]
+
+    def get_all_free_connections(self):
+        return [
+            conn
+            for conn_list in self.connections.values()
+            for conn in conn_list
+            if not conn.in_use
+        ]
+
+    @inlineCallbacks
+    def connect(self, eventloop, address):
+        from provisioningserver.rpc.clusterservice import ClusterClient
+
+        # Force everything to use AF_INET6 sockets.
+        endpoint = TCP6ClientEndpoint(self.clock, *address)
+        protocol = ClusterClient(address, eventloop, self._service)
+        conn = yield connectProtocol(endpoint, protocol)
+        return conn
+
+    def disconnect(self, connection):
+        return connection.transport.loseConnection()
+
+    @inlineCallbacks
+    def add_connection(self, eventloop, connection):
+        if self.is_staged(eventloop):
+            del self.try_connections[eventloop]
+        if eventloop not in self.connections:
+            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)
+
+    def remove_connection(self, eventloop, connection):
+        if self.is_staged(eventloop):
+            if self.try_connections[eventloop] is connection:
+                del self.try_connections[eventloop]
+        if connection in self.connections.get(eventloop, []):
+            self.connections[eventloop].remove(connection)
+            if len(self.connections[eventloop]) == 0:
+                del self.connections[eventloop]
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..6f3e4f9 100644
--- a/src/provisioningserver/rpc/tests/test_clusterservice.py
+++ b/src/provisioningserver/rpc/tests/test_clusterservice.py
@@ -23,7 +23,6 @@ from testtools.matchers import (
     Is,
     IsInstance,
     KeysEqual,
-    MatchesAll,
     MatchesDict,
     MatchesListwise,
     MatchesStructure,
@@ -32,7 +31,6 @@ from twisted import web
 from twisted.application.internet import TimerService
 from twisted.internet import error, reactor
 from twisted.internet.defer import Deferred, fail, inlineCallbacks, succeed
-from twisted.internet.endpoints import TCP6ClientEndpoint
 from twisted.internet.error import ConnectionClosed
 from twisted.internet.task import Clock
 from twisted.internet.testing import StringTransportWithDisconnection
@@ -117,7 +115,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 +446,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
@@ -498,11 +502,11 @@ class TestClusterClientService(MAASTestCase):
         )
 
         # Fake some connections.
-        service.connections = {
-            ipv4client.eventloop: ipv4client,
-            ipv6client.eventloop: ipv6client,
-            ipv6mapped.eventloop: ipv6mapped,
-            hostclient.eventloop: hostclient,
+        service.connections.connections = {
+            ipv4client.eventloop: [ipv4client],
+            ipv6client.eventloop: [ipv6client],
+            ipv6mapped.eventloop: [ipv6mapped],
+            hostclient.eventloop: [hostclient],
         }
 
         # Update the RPC state to the filesystem and info cache.
@@ -515,7 +519,8 @@ class TestClusterClientService(MAASTestCase):
             Equals(
                 {
                     client.address[0]
-                    for _, client in service.connections.items()
+                    for _, clients in service.connections.items()
+                    for client in clients
                 }
             ),
         )
@@ -999,9 +1004,9 @@ class TestClusterClientService(MAASTestCase):
     def test_update_connections_initially(self):
         service = ClusterClientService(Clock())
         mock_client = Mock()
-        _make_connection = self.patch(service, "_make_connection")
+        _make_connection = self.patch(service.connections, "connect")
         _make_connection.side_effect = lambda *args: succeed(mock_client)
-        _drop_connection = self.patch(service, "_drop_connection")
+        _drop_connection = self.patch(service.connections, "disconnect")
 
         info = json.loads(self.example_rpc_info_view_response.decode("ascii"))
         yield service._update_connections(info["eventloops"])
@@ -1020,7 +1025,7 @@ class TestClusterClientService(MAASTestCase):
                 "host1:pid=2002": mock_client,
                 "host2:pid=3003": mock_client,
             },
-            service.try_connections,
+            service.connections.try_connections,
         )
 
         self.assertEqual([], _drop_connection.mock_calls)
@@ -1038,7 +1043,7 @@ class TestClusterClientService(MAASTestCase):
             for address in addresses:
                 client = Mock()
                 client.address = address
-                service.connections[eventloop] = client
+                service.connections.connections[eventloop] = [client]
 
         logger = self.useFixture(TwistedLoggerFixture())
 
@@ -1055,7 +1060,7 @@ class TestClusterClientService(MAASTestCase):
     @inlineCallbacks
     def test_update_connections_connect_error_is_logged_tersely(self):
         service = ClusterClientService(Clock())
-        _make_connection = self.patch(service, "_make_connection")
+        _make_connection = self.patch(service.connections, "connect")
         _make_connection.side_effect = error.ConnectionRefusedError()
 
         logger = self.useFixture(TwistedLoggerFixture())
@@ -1079,7 +1084,7 @@ class TestClusterClientService(MAASTestCase):
     @inlineCallbacks
     def test_update_connections_unknown_error_is_logged_with_stack(self):
         service = ClusterClientService(Clock())
-        _make_connection = self.patch(service, "_make_connection")
+        _make_connection = self.patch(service.connections, "connect")
         _make_connection.side_effect = RuntimeError("Something went wrong.")
 
         logger = self.useFixture(TwistedLoggerFixture())
@@ -1106,8 +1111,8 @@ class TestClusterClientService(MAASTestCase):
 
     def test_update_connections_when_there_are_existing_connections(self):
         service = ClusterClientService(Clock())
-        _make_connection = self.patch(service, "_make_connection")
-        _drop_connection = self.patch(service, "_drop_connection")
+        _connect = self.patch(service.connections, "connect")
+        _disconnect = self.patch(service.connections, "disconnect")
 
         host1client = ClusterClient(
             ("::ffff:1.1.1.1", 1111), "host1:pid=1", service
@@ -1120,9 +1125,9 @@ class TestClusterClientService(MAASTestCase):
         )
 
         # Fake some connections.
-        service.connections = {
-            host1client.eventloop: host1client,
-            host2client.eventloop: host2client,
+        service.connections.connections = {
+            host1client.eventloop: [host1client],
+            host2client.eventloop: [host2client],
         }
 
         # Request a new set of connections that overlaps with the
@@ -1137,10 +1142,10 @@ class TestClusterClientService(MAASTestCase):
         # A connection is made for host3's event-loop, and the
         # connection to host2's event-loop is dropped.
         self.assertThat(
-            _make_connection,
+            _connect,
             MockCalledOnceWith(host3client.eventloop, host3client.address),
         )
-        self.assertThat(_drop_connection, MockCalledWith(host2client))
+        self.assertThat(_disconnect, MockCalledWith(host2client))
 
     @inlineCallbacks
     def test_update_only_updates_interval_when_eventloops_are_unknown(self):
@@ -1175,57 +1180,15 @@ class TestClusterClientService(MAASTestCase):
             logger.dump(),
         )
 
-    def test_make_connection(self):
-        service = ClusterClientService(Clock())
-        connectProtocol = self.patch(clusterservice, "connectProtocol")
-        service._make_connection("an-event-loop", ("a.example.com", 1111))
-        self.assertThat(connectProtocol.call_args_list, HasLength(1))
-        self.assertThat(
-            connectProtocol.call_args_list[0][0],
-            MatchesListwise(
-                (
-                    # First argument is an IPv4 TCP client endpoint
-                    # specification.
-                    MatchesAll(
-                        IsInstance(TCP6ClientEndpoint),
-                        MatchesStructure.byEquality(
-                            _reactor=service.clock,
-                            _host="a.example.com",
-                            _port=1111,
-                        ),
-                    ),
-                    # Second argument is a ClusterClient instance, the
-                    # protocol to use for the connection.
-                    MatchesAll(
-                        IsInstance(clusterservice.ClusterClient),
-                        MatchesStructure.byEquality(
-                            address=("a.example.com", 1111),
-                            eventloop="an-event-loop",
-                            service=service,
-                        ),
-                    ),
-                )
-            ),
-        )
-
-    def test_drop_connection(self):
-        connection = Mock()
-        service = make_inert_client_service()
-        service.startService()
-        service._drop_connection(connection)
-        self.assertThat(
-            connection.transport.loseConnection, MockCalledOnceWith()
-        )
-
     def test_add_connection_removes_from_try_connections(self):
         service = make_inert_client_service()
         service.startService()
         endpoint = Mock()
         connection = Mock()
         connection.address = (":::ffff", 2222)
-        service.try_connections[endpoint] = connection
+        service.connections.try_connections[endpoint] = connection
         service.add_connection(endpoint, connection)
-        self.assertThat(service.try_connections, Equals({}))
+        self.assertThat(service.connections.try_connections, Equals({}))
 
     def test_add_connection_adds_to_connections(self):
         service = make_inert_client_service()
@@ -1234,7 +1197,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,21 +1211,45 @@ 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.connections, "connect").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.connections._max_idle_connections,
+        )
+
     def test_remove_connection_removes_from_try_connections(self):
         service = make_inert_client_service()
         service.startService()
         endpoint = Mock()
         connection = Mock()
-        service.try_connections[endpoint] = connection
+        service.connections.try_connections[endpoint] = connection
         service.remove_connection(endpoint, connection)
-        self.assertThat(service.try_connections, Equals({}))
+        self.assertEqual(service.connections.try_connections, {})
 
     def test_remove_connection_removes_from_connections(self):
         service = make_inert_client_service()
         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 +1258,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 +1267,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()
@@ -1294,45 +1281,96 @@ class TestClusterClientService(MAASTestCase):
 
     def test_getClient(self):
         service = ClusterClientService(Clock())
-        service.connections = {
-            sentinel.eventloop01: DummyConnection(),
-            sentinel.eventloop02: DummyConnection(),
-            sentinel.eventloop03: DummyConnection(),
+        service.connections.connections = {
+            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):
         service = ClusterClientService(Clock())
-        service.connections = {}
+        service.connections.connections = {}
         self.assertRaises(exceptions.NoConnectionsAvailable, service.getClient)
 
     @inlineCallbacks
+    def test_getClientNow_scales_connections_when_busy(self):
+        service = ClusterClientService(Clock(), max_conns=2)
+        service.connections.connections = {
+            sentinel.eventloop01: [FakeBusyConnectionToRegion()],
+            sentinel.eventloop02: [FakeBusyConnectionToRegion()],
+            sentinel.eventloop03: [FakeBusyConnectionToRegion()],
+        }
+        self.patch(service.connections, "connect").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.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(),
+        service.connections.connections = {
+            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
     def test_getClientNow_calls__tryUpdate_when_there_are_no_connections(self):
         service = ClusterClientService(Clock())
-        service.connections = {}
 
         def addConnections():
-            service.connections = {
-                sentinel.eventloop01: DummyConnection(),
-                sentinel.eventloop02: DummyConnection(),
-                sentinel.eventloop03: DummyConnection(),
+            service.connections.connections = {
+                sentinel.eventloop01: [FakeConnection()],
+                sentinel.eventloop02: [FakeConnection()],
+                sentinel.eventloop03: [FakeConnection()],
             }
             return succeed(None)
 
@@ -1340,12 +1378,15 @@ 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):
         service = ClusterClientService(Clock())
-        service.connections = {}
 
         self.patch(service, "_tryUpdate").return_value = succeed(None)
         d = service.getClientNow()
@@ -1383,17 +1424,16 @@ 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)])
 
     def test_getAllClients_when_there_are_no_connections(self):
         service = ClusterClientService(Clock())
-        service.connections = {}
         self.assertThat(service.getAllClients(), Equals([]))
 
 
@@ -1546,7 +1586,7 @@ class TestClusterClient(MAASTestCase):
 
     def test_connecting(self):
         client = self.make_running_client()
-        client.service.try_connections[client.eventloop] = client
+        client.service.connections.try_connections[client.eventloop] = client
         self.patch_authenticate_for_success(client)
         self.patch_register_for_success(client)
         self.assertEqual(client.service.connections, {})
@@ -1560,16 +1600,19 @@ class TestClusterClient(MAASTestCase):
         self.assertTrue(extract_result(wait_for_authenticated))
         # ready has been set with the name of the event-loop.
         self.assertEqual(client.eventloop, extract_result(wait_for_ready))
-        self.assertEqual(client.service.try_connections, {})
+        self.assertEqual(len(client.service.connections.try_connections), 0)
         self.assertEqual(
-            client.service.connections, {client.eventloop: client}
+            client.service.connections.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.connections[client.eventloop] = [
+            sentinel.connection
+        ]
 
         # Connect via an in-memory transport.
         transport = StringTransportWithDisconnection()
@@ -1586,7 +1629,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)
@@ -1631,7 +1675,7 @@ class TestClusterClient(MAASTestCase):
 
         # The connections list is unchanged because the new connection
         # immediately disconnects.
-        self.assertEqual(client.service.connections, {})
+        self.assertEqual(client.service.connections.connections, {})
         self.assertFalse(client.connected)
 
     def test_disconnects_when_authentication_errors(self):
diff --git a/src/provisioningserver/rpc/tests/test_connectionpool.py b/src/provisioningserver/rpc/tests/test_connectionpool.py
new file mode 100644
index 0000000..692d5e6
--- /dev/null
+++ b/src/provisioningserver/rpc/tests/test_connectionpool.py
@@ -0,0 +1,280 @@
+# Copyright 2022 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+from unittest.mock import Mock
+
+from twisted.internet.defer import inlineCallbacks, succeed
+from twisted.internet.endpoints import TCP6ClientEndpoint
+from twisted.internet.task import Clock
+
+from maastesting import get_testing_timeout
+from maastesting.testcase import MAASTestCase, MAASTwistedRunTest
+from maastesting.twisted import extract_result
+from provisioningserver.rpc import connectionpool as connectionpoolModule
+from provisioningserver.rpc import exceptions
+from provisioningserver.rpc.clusterservice import ClusterClient
+from provisioningserver.rpc.connectionpool import ConnectionPool
+
+TIMEOUT = get_testing_timeout()
+
+
+class TestConnectionPool(MAASTestCase):
+
+    run_tests_with = MAASTwistedRunTest.make_factory(timeout=TIMEOUT)
+
+    def test_setitem_sets_item_in_connections(self):
+        cp = ConnectionPool(Clock(), Mock())
+        key = Mock()
+        val = Mock()
+        cp[key] = val
+        self.assertEqual(cp.connections, {key: val})
+
+    def test_getitem_gets_item_in_connections(self):
+        cp = ConnectionPool(Clock(), Mock())
+        key = Mock()
+        val = Mock()
+        cp[key] = val
+        self.assertEqual(cp.connections[key], cp[key])
+
+    def test_len_gets_length_of_connections(self):
+        cp = ConnectionPool(Clock(), Mock())
+        key = Mock()
+        val = Mock()
+        cp[key] = [val]
+        self.assertEqual(len(cp), len(cp.get_all_connections()))
+
+    def test_delitem_removes_item_from_connections(self):
+        cp = ConnectionPool(Clock(), Mock())
+        key = Mock()
+        val = Mock()
+        cp[key] = val
+        self.assertEqual(cp.connections[key], val)
+        del cp[key]
+        self.assertIsNone(cp.connections.get(key))
+
+    def test_contains_returns_if_key_in_connections(self):
+        cp = ConnectionPool(Clock(), Mock())
+        key = Mock()
+        val = Mock()
+        cp[key] = val
+        self.assertEqual(key in cp, key in cp.connections)
+
+    def test_compare_ConnectionPool_equal_to_compare_connections(self):
+        cp = ConnectionPool(Clock(), Mock())
+        self.assertEqual(cp, cp.connections)
+        self.assertEqual(cp, {})
+
+    def test__reap_extra_connection_reaps_a_non_busy_connection(self):
+        cp = ConnectionPool(Clock(), Mock())
+        eventloop = Mock()
+        connection = Mock()
+        connection.in_use = False
+        cp[eventloop] = [connection]
+        disconnect = self.patch(cp, "disconnect")
+        cp._reap_extra_connection(eventloop, connection)
+        self.assertEqual(len(cp), 0)
+        disconnect.assert_called_once_with(connection)
+
+    def test__reap_extra_connection_waits_for_a_busy_connection(self):
+        clock = Clock()
+        cp = ConnectionPool(clock, Mock())
+        eventloop = Mock()
+        connection = Mock()
+        connection.in_use = True
+        cp[eventloop] = [connection]
+        self.patch(cp, "disconnect")
+        cp._reap_extra_connection(eventloop, connection)
+        self.assertIn(eventloop, clock.calls[0].args)
+        self.assertIn(connection, clock.calls[0].args)
+        self.assertEqual(
+            "_reap_extra_connection", clock.calls[0].func.__name__
+        )
+        self.assertEqual(cp._keepalive, clock.calls[0].time)
+
+    def test_is_staged(self):
+        cp = ConnectionPool(Clock(), Mock())
+        eventloop1 = Mock()
+        eventloop2 = Mock()
+        cp.try_connections[eventloop1] = Mock()
+        self.assertTrue(cp.is_staged(eventloop1))
+        self.assertFalse(cp.is_staged(eventloop2))
+
+    def test_get_staged_connection(self):
+        cp = ConnectionPool(Clock(), Mock())
+        eventloop = Mock()
+        connection = Mock()
+        cp.try_connections[eventloop] = connection
+        self.assertEqual(cp.get_staged_connection(eventloop), connection)
+
+    def test_get_staged_connections(self):
+        cp = ConnectionPool(Clock(), Mock())
+        eventloop = Mock()
+        connection = Mock()
+        cp.try_connections[eventloop] = connection
+        self.assertEqual(cp.get_staged_connections(), {eventloop: connection})
+
+    def test_scale_up_connections_adds_a_connection(self):
+        cp = ConnectionPool(Clock(), Mock(), max_conns=2)
+        eventloop = Mock()
+        connection1 = Mock()
+        connection2 = Mock()
+        connect = self.patch(cp, "connect")
+        connect.return_value = succeed(connection2)
+        cp[eventloop] = [connection1]
+        cp.scale_up_connections()
+        self.assertCountEqual(cp[eventloop], [connection1, connection2])
+
+    def test_scale_up_connections_raises_MaxConnectionsOpen_when_cannot_create_another(
+        self,
+    ):
+        cp = ConnectionPool(Clock(), Mock())
+        eventloop = Mock()
+        connection1 = Mock()
+        connection2 = Mock()
+        connect = self.patch(cp, "connect")
+        connect.return_value = succeed(connection2)
+        cp[eventloop] = [connection1]
+        self.assertRaises(
+            exceptions.MaxConnectionsOpen,
+            extract_result,
+            cp.scale_up_connections(),
+        )
+
+    def test_get_connection(self):
+        cp = ConnectionPool(Clock(), Mock(), max_idle_conns=2, max_conns=2)
+        eventloops = [Mock() for _ in range(3)]
+        cp.connections = {
+            eventloop: [Mock() for _ in range(2)] for eventloop in eventloops
+        }
+        self.assertIn(cp.get_connection(eventloops[0]), cp[eventloops[0]])
+
+    def test_get_random_connection(self):
+        cp = ConnectionPool(Clock(), Mock(), max_idle_conns=2, max_conns=2)
+        eventloops = [Mock() for _ in range(3)]
+        cp.connections = {
+            eventloop: [Mock() for _ in range(2)] for eventloop in eventloops
+        }
+        self.assertIn(
+            cp.get_connection(eventloops[0]),
+            [conn for conn_list in cp.values() for conn in conn_list],
+        )
+
+    def test_get_random_free_connection_returns_a_free_connection(self):
+        cp = ConnectionPool(Clock(), Mock())
+        eventloops = [Mock() for _ in range(3)]
+
+        def _create_conn(in_use):
+            conn = Mock()
+            conn.in_use = in_use
+            return conn
+
+        cp.connections = {
+            eventloops[0]: [_create_conn(True)],
+            eventloops[1]: [_create_conn(False)],
+            eventloops[2]: [_create_conn(True)],
+        }
+        conn = cp.get_random_free_connection()
+        self.assertIn(conn, cp[eventloops[1]])
+
+    def test_get_random_free_connection_raises_AllConnectionsBusy_when_there_are_no_free_connections(
+        self,
+    ):
+        cp = ConnectionPool(Clock(), Mock())
+        eventloops = [Mock() for _ in range(3)]
+
+        def _create_conn(in_use):
+            conn = Mock()
+            conn.in_use = in_use
+            return conn
+
+        cp.connections = {
+            eventloops[0]: [_create_conn(True)],
+            eventloops[1]: [_create_conn(True)],
+            eventloops[2]: [_create_conn(True)],
+        }
+
+        self.assertRaises(
+            exceptions.AllConnectionsBusy, cp.get_random_free_connection
+        )
+
+    def test_get_all_connections(self):
+        cp = ConnectionPool(Clock(), Mock())
+        eventloops = [Mock() for _ in range(3)]
+        cp.connections = {
+            eventloops[0]: [Mock()],
+            eventloops[1]: [Mock()],
+            eventloops[2]: [Mock()],
+        }
+
+        self.assertCountEqual(
+            cp.get_all_connections(),
+            [conn for conn_list in cp.values() for conn in conn_list],
+        )
+
+    def test_get_all_free_connections(self):
+        cp = ConnectionPool(Clock(), Mock(), max_conns=2)
+        eventloops = [Mock() for _ in range(3)]
+
+        def _create_conn(in_use):
+            conn = Mock()
+            conn.in_use = in_use
+            return conn
+
+        cp.connections = {
+            eventloops[0]: [_create_conn(True), _create_conn(False)],
+            eventloops[1]: [_create_conn(True)],
+            eventloops[2]: [_create_conn(False)],
+        }
+
+        self.assertCountEqual(
+            cp.get_all_free_connections(),
+            [
+                conn
+                for conn_list in cp.values()
+                for conn in conn_list
+                if not conn.in_use
+            ],
+        )
+
+    @inlineCallbacks
+    def test_connect(self):
+        clock = Clock()
+        connection = Mock()
+        service = Mock()
+        cp = ConnectionPool(clock, service)
+        connectProtocol = self.patch(connectionpoolModule, "connectProtocol")
+        connectProtocol.return_value = connection
+        result = yield cp.connect("an-event-loop", ("a.example.com", 1111))
+        self.assertEqual(len(connectProtocol.call_args_list), 1)
+        connectProtocol.called_once_with(
+            TCP6ClientEndpoint(reactor=clock, host="a.example.com", port=1111),
+            ClusterClient(
+                address=("a.example.com", 1111),
+                eventloop="an-event-loop",
+                service=service,
+            ),
+        )
+        self.assertEqual(result, connection)
+
+    def test_drop_connection(self):
+        connection = Mock()
+        cp = ConnectionPool(Clock(), Mock())
+        cp.disconnect(connection)
+        connection.transport.loseConnection.assert_called_once_with()
+
+    @inlineCallbacks
+    def test_add_connection_adds_the_staged_connection(self):
+        eventloop = Mock()
+        connection = Mock()
+        cp = ConnectionPool(Clock(), Mock())
+        cp.try_connections = {eventloop: connection}
+        yield cp.add_connection(eventloop, connection)
+        self.assertIn(connection, cp.get_all_connections())
+
+    def test_remove_connection_removes_connection_from_pool(self):
+        eventloop = Mock()
+        connection = Mock()
+        cp = ConnectionPool(Clock(), Mock())
+        cp.connections[eventloop] = [connection]
+        cp.remove_connection(eventloop, connection)
+        self.assertEqual(cp.connections, {})