Merge ~blake-rouse/maas:ha-fixes-2.2 into maas:2.2

[Blake Rouse]

Fixes LP: #1705594, #1724677, #1727073.

Backports of ad6874821711208d0e420ee2168862b59cf16cef, 98d01e01e0064fa62e1cda75363805c50398220d, d388d7ad9e5660973cddd81e09cc52ec4e8e2595, b2f6b7759e6345b1a81f9e9b1b3b6fedfefd84ce from master.

Backports from:

https://code.launchpad.net/~blake-rouse/maas/+git/maas/+merge/332717
https://code.launchpad.net/~blake-rouse/maas/+git/maas/+merge/332727
https://code.launchpad.net/~blake-rouse/maas/+git/maas/+merge/332741
https://code.launchpad.net/~blake-rouse/maas/+git/maas/+merge/332814

[Blake Rouse (blake-rouse)]

Currently running in the CI to verify a good backport.

http://162.213.35.104:8080/job/maas-git-manual/78/console

[Blake Rouse (blake-rouse)]

Last CI failed because of bad import, fixed and new CI run:

http://162.213.35.104:8080/job/maas-git-manual/79/console

[Blake Rouse (blake-rouse)]

Passed CI

Self-approving backport.

review: Approve

[MAAS Lander (maas-lander)]

LANDING
-b ha-fixes-2.2 lp:~blake-rouse/maas into -b 2.2 lp:~maas-committers/maas

STATUS: FAILED BUILD
LOG: http://maas-ci-jenkins.internal:8080/job/maas/job/branch-tester/503/consoleText

[MAAS Lander (maas-lander)]

LANDING
-b ha-fixes-2.2 lp:~blake-rouse/maas into -b 2.2 lp:~maas-committers/maas

STATUS: FAILED BUILD
LOG: http://maas-ci-jenkins.internal:8080/job/maas/job/branch-tester/505/consoleText

[MAAS Lander (maas-lander)]

LANDING
-b ha-fixes-2.2 lp:~blake-rouse/maas into -b 2.2 lp:~maas-committers/maas

STATUS: FAILED MERGE
LOG: http://maas-ci-jenkins.internal:8080/job/maas/job/branch-lander/86532/console

diff --git a/src/maasserver/locks.py b/src/maasserver/locks.py
index b482f5c..f68539e 100644
--- a/src/maasserver/locks.py
+++ b/src/maasserver/locks.py
@@ -9,7 +9,6 @@ __all__ = [
     "eventloop",
     "import_images",
     "node_acquire",
-    "rack_registration",
     "security",
     "startup",
 ]
@@ -19,7 +18,11 @@ from maasserver.utils.dblocks import (
     DatabaseXactLock,
 )
 
-# Lock around starting-up a MAAS region.
+# Lock around starting-up a MAAS region and connection of rack controllers.
+# This can be a problem where a region controller and a rack controller try
+# to create there node objects at the same time. Rack registration also
+# involves populating fabrics, VLANs, and other information that may overlap
+# between rack controller.
 startup = DatabaseLock(1)
 
 # Lock around performing critical security-related operations, like
@@ -41,10 +44,10 @@ node_acquire = DatabaseXactLock(7)
 # Lock to help with concurrent allocation of IP addresses.
 address_allocation = DatabaseLock(8)
 
-# Lock to prevent concurrent registration of rack controllers. This can be a
-# problem because registration involves populating fabrics, VLANs, and other
-# information that may overlap between rack controller.
-rack_registration = DatabaseLock(9)
+# Lock used to be used just for rack registration. Because of lp:1705594 this
+# was consolidated into the startup lock with the region controller.
+# DO NOT USE '9' AGAIN, it is reserved so it doesn't break upgrades.
+# rack_registration = DatabaseLock(9)
 
 # Lock to prevent concurrent network scanning.
 try_active_discovery = DatabaseLock(10).TRY
diff --git a/src/maasserver/models/node.py b/src/maasserver/models/node.py
index f6ee6cc..e0c95fe 100644
--- a/src/maasserver/models/node.py
+++ b/src/maasserver/models/node.py
@@ -54,7 +54,10 @@ from django.db.models import (
 )
 from django.db.models.query import QuerySet
 from django.shortcuts import get_object_or_404
-from maasserver import DefaultMeta
+from maasserver import (
+    DefaultMeta,
+    locks,
+)
 from maasserver.clusterrpc.pods import decompose_machine
 from maasserver.clusterrpc.power import (
     power_cycle,
@@ -143,6 +146,7 @@ from maasserver.storage_layouts import (
     StorageLayoutError,
     StorageLayoutMissingBootDiskError,
 )
+from maasserver.utils import synchronised
 from maasserver.utils.dns import validate_hostname
 from maasserver.utils.mac import get_vendor_for_mac
 from maasserver.utils.orm import (
@@ -151,6 +155,7 @@ from maasserver.utils.orm import (
     post_commit,
     post_commit_do,
     transactional,
+    with_connection,
 )
 from maasserver.utils.threads import (
     callOutToDatabase,
@@ -208,6 +213,7 @@ from provisioningserver.utils.twisted import (
     asynchronous,
     callOut,
     deferWithTimeout,
+    synchronous,
 )
 from twisted.internet.defer import (
     Deferred,
@@ -4595,6 +4601,10 @@ class Controller(Node):
             for interface in interfaces
         }
 
+    @synchronous
+    @with_connection
+    @synchronised(locks.startup)
+    @transactional
     def update_interfaces(self, interfaces):
         """Update the interfaces attached to the controller.
 
@@ -4872,7 +4882,7 @@ class RackController(Controller):
                 Service.objects.update_service_for(
                     self, "rackd", SERVICE_STATUS.DEGRADED,
                     "{:.0%} connected to region controllers.".format(
-                        percentage))
+                        1.0 - percentage))
 
     def get_image_sync_status(self, boot_images=None):
         """Return the status of the boot image import process."""
diff --git a/src/maasserver/models/tests/test_node.py b/src/maasserver/models/tests/test_node.py
index 0467bba..e8ece11 100644
--- a/src/maasserver/models/tests/test_node.py
+++ b/src/maasserver/models/tests/test_node.py
@@ -9495,7 +9495,7 @@ class TestRackController(MAASTransactionServerTestCase):
             MatchesStructure.byEquality(
                 status=SERVICE_STATUS.DEGRADED, status_info=(
                     "{:.0%} connected to region controllers.".format(
-                        percentage))))
+                        1.0 - percentage))))
 
     fake_images = [
         {
diff --git a/src/maasserver/rpc/rackcontrollers.py b/src/maasserver/rpc/rackcontrollers.py
index 9c49532..7c82548 100644
--- a/src/maasserver/rpc/rackcontrollers.py
+++ b/src/maasserver/rpc/rackcontrollers.py
@@ -72,7 +72,7 @@ def handle_upgrade(rack_controller, nodegroup_uuid):
 
 @synchronous
 @with_connection
-@synchronised(locks.rack_registration)
+@synchronised(locks.startup)
 @transactional
 def register(
         system_id=None, hostname='', interfaces=None,
diff --git a/src/maasserver/rpc/regionservice.py b/src/maasserver/rpc/regionservice.py
index 3ddf50c..534002b 100644
--- a/src/maasserver/rpc/regionservice.py
+++ b/src/maasserver/rpc/regionservice.py
@@ -1214,15 +1214,31 @@ class RegionAdvertising:
         Each tuple corresponds to somewhere an event-loop is listening
         within the whole region. The `name` is the event-loop name.
         """
+        # Each regiond might be running a local bridge that duplicates the
+        # same IP address across region controllers. Each region controller
+        # must output a set of unique of IP addresses, to prevent the rack
+        # controller from connecting to a different region controller then
+        # the rack controller was expecting to be connecting to.
+        def _unique_to_region(address, region, regions):
+            for region_obj in regions:
+                if region_obj != region:
+                    for process in region_obj.processes.all():
+                        for endpoint in process.endpoints.all():
+                            if endpoint.address == address:
+                                return False
+            return True
+
         regions = RegionController.objects.all()
         regions = regions.prefetch_related("processes", "processes__endpoints")
         all_endpoints = []
         for region_obj in regions:
             for process in region_obj.processes.all():
                 for endpoint in process.endpoints.all():
-                    all_endpoints.append((
-                        "%s:pid=%d" % (region_obj.hostname, process.pid),
-                        endpoint.address, endpoint.port))
+                    if _unique_to_region(
+                            endpoint.address, region_obj, regions):
+                        all_endpoints.append((
+                            "%s:pid=%d" % (region_obj.hostname, process.pid),
+                            endpoint.address, endpoint.port))
         return all_endpoints
 
     @classmethod
diff --git a/src/maasserver/rpc/tests/test_rackcontrollers.py b/src/maasserver/rpc/tests/test_rackcontrollers.py
index 8c62f80..e5b075b 100644
--- a/src/maasserver/rpc/tests/test_rackcontrollers.py
+++ b/src/maasserver/rpc/tests/test_rackcontrollers.py
@@ -288,11 +288,11 @@ class TestRegisterRackController(MAASServerTestCase):
                 for name, interface in interfaces.items()
             )))
 
-    def test_registers_with_rack_registration_lock_held(self):
+    def test_registers_with_startup_lock_held(self):
         lock_status = []
 
         def record_lock_status(*args):
-            lock_status.append(locks.rack_registration.is_locked())
+            lock_status.append(locks.startup.is_locked())
             return None  # Simulate that no rack found.
 
         find = self.patch(rackcontrollers, "find")
diff --git a/src/maasserver/rpc/tests/test_regionservice.py b/src/maasserver/rpc/tests/test_regionservice.py
index 0a8016a..12678c5 100644
--- a/src/maasserver/rpc/tests/test_regionservice.py
+++ b/src/maasserver/rpc/tests/test_regionservice.py
@@ -33,6 +33,7 @@ from maasserver.models import (
     RackController,
     RegionController,
     RegionControllerProcess,
+    RegionControllerProcessEndpoint,
     RegionRackRPCConnection,
     Service as ServiceModel,
     timestampedmodel,
@@ -1604,6 +1605,42 @@ class TestRegionAdvertising(MAASServerTestCase):
         ]
         self.assertItemsEqual(expected, advertising.dump())
 
+    def test_dump_doesnt_duplicate_ips_across_region_controllers(self):
+        duplicate_address = factory.make_ipv4_address()
+        addresses = {
+            (factory.make_ipv4_address(), factory.pick_port()),
+            (factory.make_ipv4_address(), factory.pick_port()),
+        }
+        advertising = RegionAdvertising.promote()
+        advertising.update(
+            addresses.union({(duplicate_address, factory.pick_port()), }))
+
+        other_addresses = {
+            (factory.make_ipv4_address(), factory.pick_port()),
+            (factory.make_ipv4_address(), factory.pick_port()),
+        }
+        other_region = factory.make_Node(node_type=NODE_TYPE.REGION_CONTROLLER)
+        other_process = RegionControllerProcess.objects.create(
+            region=other_region, pid=randint(1, 1000))
+        for address, port in other_addresses:
+            RegionControllerProcessEndpoint.objects.create(
+                process=other_process, address=address, port=port)
+        RegionControllerProcessEndpoint.objects.create(
+            process=other_process,
+            address=duplicate_address,
+            port=factory.pick_port())
+
+        expected = [
+            ("%s:pid=%d" % (gethostname(), os.getpid()), addr, port)
+            for (addr, port) in addresses
+        ] + [
+            ("%s:pid=%d" % (
+                other_region.hostname, other_process.pid),
+             addr, port)
+            for (addr, port) in other_addresses
+        ]
+        self.assertItemsEqual(expected, advertising.dump())
+
     def test__adds_connection_and_removes_connection(self):
         advertising = RegionAdvertising.promote()
         process = advertising.getRegionProcess()
diff --git a/src/maasserver/utils/orm.py b/src/maasserver/utils/orm.py
index 2ec0590..40f8a3f 100644
--- a/src/maasserver/utils/orm.py
+++ b/src/maasserver/utils/orm.py
@@ -616,14 +616,25 @@ def connected():
 
     If there is not yet a connection to the database, this will connect on
     entry and disconnect on exit. Preexisting connections will be left alone.
+
+    If the preexisting connection is not usable it is closed and a new
+    connection is made.
     """
     if connection.connection is None:
+        connection.close_if_unusable_or_obsolete()
         connection.ensure_connection()
         try:
             yield
         finally:
             connection.close()
+    elif connection.is_usable():
+        yield
     else:
+        # Connection is not usable, so we disconnect and reconnect. Since
+        # the connection was previously connected we do not disconnect this
+        # new connection.
+        connection.close_if_unusable_or_obsolete()
+        connection.ensure_connection()
         yield
 
 
diff --git a/src/maasserver/utils/tests/test_orm.py b/src/maasserver/utils/tests/test_orm.py
index 064f1b0..94ee817 100644
--- a/src/maasserver/utils/tests/test_orm.py
+++ b/src/maasserver/utils/tests/test_orm.py
@@ -1004,6 +1004,22 @@ class TestConnected(MAASTransactionServerTestCase):
             self.assertThat(connection.connection, Is(preexisting_connection))
         self.assertThat(connection.connection, Is(preexisting_connection))
 
+    def test__disconnects_and_reconnects_if_not_usable(self):
+        connection.ensure_connection()
+        preexisting_connection = connection.connection
+
+        connection.errors_occurred = True
+        self.patch(connection, "is_usable").return_value = False
+
+        self.assertThat(connection.connection, Not(Is(None)))
+        with orm.connected():
+            self.assertThat(
+                connection.connection, Not(Is(preexisting_connection)))
+            self.assertThat(connection.connection, Not(Is(None)))
+
+        self.assertThat(connection.connection, Not(Is(preexisting_connection)))
+        self.assertThat(connection.connection, Not(Is(None)))
+
 
 class TestWithConnection(MAASTransactionServerTestCase):
     """Tests for the `orm.with_connection` decorator."""
diff --git a/src/provisioningserver/rackdservices/tests/test_tftp.py b/src/provisioningserver/rackdservices/tests/test_tftp.py
index 1082d6a..14510d9 100644
--- a/src/provisioningserver/rackdservices/tests/test_tftp.py
+++ b/src/provisioningserver/rackdservices/tests/test_tftp.py
@@ -379,6 +379,229 @@ class TestTFTPBackend(MAASTestCase):
         )
 
     @inlineCallbacks
+    def test_get_boot_method_reader_uses_same_client(self):
+        # Fake configuration parameters, as discovered from the file path.
+        fake_params = {"mac": factory.make_mac_address("-")}
+        # Fake kernel configuration parameters, as returned from the RPC call.
+        fake_kernel_params = make_kernel_parameters()
+        fake_params = fake_kernel_params._asdict()
+
+        # Stub the output of list_boot_images so the label is set in the
+        # kernel parameters.
+        boot_image = {
+            "osystem": fake_params["osystem"],
+            "release": fake_params["release"],
+            "architecture": fake_params["arch"],
+            "subarchitecture": fake_params["subarch"],
+            "purpose": fake_params["purpose"],
+            "supported_subarches": "",
+            "label": fake_params["label"],
+        }
+        self.patch(tftp_module, "list_boot_images").return_value = [boot_image]
+        del fake_params["label"]
+
+        # Stub RPC call to return the fake configuration parameters.
+        clients = []
+        for _ in range(10):
+            client = Mock()
+            client.localIdent = factory.make_name("system_id")
+            client.side_effect = lambda *args, **kwargs: (
+                succeed(dict(fake_params)))
+            clients.append(client)
+        client_service = Mock()
+        client_service.getClientNow.side_effect = [
+            succeed(client)
+            for client in clients
+        ]
+        client_service.getAllClients.return_value = clients
+
+        # get_boot_method_reader() takes a dict() of parameters and returns an
+        # `IReader` of a PXE configuration, rendered by
+        # `PXEBootMethod.get_reader`.
+        backend = TFTPBackend(
+            self.make_dir(), client_service)
+
+        # Stub get_reader to return the render parameters.
+        method = PXEBootMethod()
+        fake_render_result = factory.make_name("render").encode("utf-8")
+        render_patch = self.patch(method, "get_reader")
+        render_patch.return_value = BytesReader(fake_render_result)
+
+        # Get the reader once.
+        remote_ip = factory.make_ipv4_address()
+        params_with_ip = dict(fake_params)
+        params_with_ip['remote_ip'] = remote_ip
+        reader = yield backend.get_boot_method_reader(method, params_with_ip)
+        self.addCleanup(reader.finish)
+
+        # Get the reader twice.
+        params_with_ip = dict(fake_params)
+        params_with_ip['remote_ip'] = remote_ip
+        reader = yield backend.get_boot_method_reader(method, params_with_ip)
+        self.addCleanup(reader.finish)
+
+        # Only one client is saved.
+        self.assertEquals(clients[0], backend.client_to_remote[remote_ip])
+
+        # Only the first client should have been called twice, and all the
+        # other clients should not have been called.
+        self.assertEquals(2, clients[0].call_count)
+        for idx in range(1, 10):
+            self.assertThat(clients[idx], MockNotCalled())
+
+    @inlineCallbacks
+    def test_get_boot_method_reader_uses_different_clients(self):
+        # Fake configuration parameters, as discovered from the file path.
+        fake_params = {"mac": factory.make_mac_address("-")}
+        # Fake kernel configuration parameters, as returned from the RPC call.
+        fake_kernel_params = make_kernel_parameters()
+        fake_params = fake_kernel_params._asdict()
+
+        # Stub the output of list_boot_images so the label is set in the
+        # kernel parameters.
+        boot_image = {
+            "osystem": fake_params["osystem"],
+            "release": fake_params["release"],
+            "architecture": fake_params["arch"],
+            "subarchitecture": fake_params["subarch"],
+            "purpose": fake_params["purpose"],
+            "supported_subarches": "",
+            "label": fake_params["label"],
+        }
+        self.patch(tftp_module, "list_boot_images").return_value = [boot_image]
+        del fake_params["label"]
+
+        # Stub RPC call to return the fake configuration parameters.
+        clients = []
+        for _ in range(10):
+            client = Mock()
+            client.localIdent = factory.make_name("system_id")
+            client.side_effect = lambda *args, **kwargs: (
+                succeed(dict(fake_params)))
+            clients.append(client)
+        client_service = Mock()
+        client_service.getClientNow.side_effect = [
+            succeed(client)
+            for client in clients
+        ]
+        client_service.getAllClients.return_value = clients
+
+        # get_boot_method_reader() takes a dict() of parameters and returns an
+        # `IReader` of a PXE configuration, rendered by
+        # `PXEBootMethod.get_reader`.
+        backend = TFTPBackend(
+            self.make_dir(), client_service)
+
+        # Stub get_reader to return the render parameters.
+        method = PXEBootMethod()
+        fake_render_result = factory.make_name("render").encode("utf-8")
+        render_patch = self.patch(method, "get_reader")
+        render_patch.return_value = BytesReader(fake_render_result)
+
+        # Get the reader once.
+        remote_ip_one = factory.make_ipv4_address()
+        params_with_ip = dict(fake_params)
+        params_with_ip['remote_ip'] = remote_ip_one
+        reader = yield backend.get_boot_method_reader(method, params_with_ip)
+        self.addCleanup(reader.finish)
+
+        # Get the reader twice.
+        remote_ip_two = factory.make_ipv4_address()
+        params_with_ip = dict(fake_params)
+        params_with_ip['remote_ip'] = remote_ip_two
+        reader = yield backend.get_boot_method_reader(method, params_with_ip)
+        self.addCleanup(reader.finish)
+
+        # The both clients are saved.
+        self.assertEquals(clients[0], backend.client_to_remote[remote_ip_one])
+        self.assertEquals(clients[1], backend.client_to_remote[remote_ip_two])
+
+        # Only the first and second client should have been called once, and
+        # all the other clients should not have been called.
+        self.assertEquals(1, clients[0].call_count)
+        self.assertEquals(1, clients[1].call_count)
+        for idx in range(2, 10):
+            self.assertThat(clients[idx], MockNotCalled())
+
+    @inlineCallbacks
+    def test_get_boot_method_reader_grabs_new_client_on_lost_conn(self):
+        # Fake configuration parameters, as discovered from the file path.
+        fake_params = {"mac": factory.make_mac_address("-")}
+        # Fake kernel configuration parameters, as returned from the RPC call.
+        fake_kernel_params = make_kernel_parameters()
+        fake_params = fake_kernel_params._asdict()
+
+        # Stub the output of list_boot_images so the label is set in the
+        # kernel parameters.
+        boot_image = {
+            "osystem": fake_params["osystem"],
+            "release": fake_params["release"],
+            "architecture": fake_params["arch"],
+            "subarchitecture": fake_params["subarch"],
+            "purpose": fake_params["purpose"],
+            "supported_subarches": "",
+            "label": fake_params["label"],
+        }
+        self.patch(tftp_module, "list_boot_images").return_value = [boot_image]
+        del fake_params["label"]
+
+        # Stub RPC call to return the fake configuration parameters.
+        clients = []
+        for _ in range(10):
+            client = Mock()
+            client.localIdent = factory.make_name("system_id")
+            client.side_effect = lambda *args, **kwargs: (
+                succeed(dict(fake_params)))
+            clients.append(client)
+        client_service = Mock()
+        client_service.getClientNow.side_effect = [
+            succeed(client)
+            for client in clients
+        ]
+        client_service.getAllClients.side_effect = [
+            clients[1:],
+            clients[2:],
+        ]
+
+        # get_boot_method_reader() takes a dict() of parameters and returns an
+        # `IReader` of a PXE configuration, rendered by
+        # `PXEBootMethod.get_reader`.
+        backend = TFTPBackend(
+            self.make_dir(), client_service)
+
+        # Stub get_reader to return the render parameters.
+        method = PXEBootMethod()
+        fake_render_result = factory.make_name("render").encode("utf-8")
+        render_patch = self.patch(method, "get_reader")
+        render_patch.return_value = BytesReader(fake_render_result)
+
+        # Get the reader once.
+        remote_ip = factory.make_ipv4_address()
+        params_with_ip = dict(fake_params)
+        params_with_ip['remote_ip'] = remote_ip
+        reader = yield backend.get_boot_method_reader(method, params_with_ip)
+        self.addCleanup(reader.finish)
+
+        # The first client is now saved.
+        self.assertEquals(clients[0], backend.client_to_remote[remote_ip])
+
+        # Get the reader twice.
+        params_with_ip = dict(fake_params)
+        params_with_ip['remote_ip'] = remote_ip
+        reader = yield backend.get_boot_method_reader(method, params_with_ip)
+        self.addCleanup(reader.finish)
+
+        # The second client is now saved.
+        self.assertEquals(clients[1], backend.client_to_remote[remote_ip])
+
+        # Only the first and second client should have been called once, and
+        # all the other clients should not have been called.
+        self.assertEquals(1, clients[0].call_count)
+        self.assertEquals(1, clients[1].call_count)
+        for idx in range(2, 10):
+            self.assertThat(clients[idx], MockNotCalled())
+
+    @inlineCallbacks
     def test_get_boot_method_reader_returns_rendered_params(self):
         # Fake configuration parameters, as discovered from the file path.
         fake_params = {"mac": factory.make_mac_address("-")}
diff --git a/src/provisioningserver/rackdservices/tftp.py b/src/provisioningserver/rackdservices/tftp.py
index b56d4b3..5dc1547 100644
--- a/src/provisioningserver/rackdservices/tftp.py
+++ b/src/provisioningserver/rackdservices/tftp.py
@@ -66,6 +66,7 @@ from twisted.internet.defer import (
     inlineCallbacks,
     maybeDeferred,
     returnValue,
+    succeed,
 )
 from twisted.internet.task import deferLater
 from twisted.python.filepath import FilePath
@@ -163,9 +164,48 @@ class TFTPBackend(FilesystemSynchronousBackend):
             base_path = FilePath(base_path)
         super(TFTPBackend, self).__init__(
             base_path, can_read=True, can_write=False)
+        self.client_to_remote = {}
         self.client_service = client_service
         self.fetcher = RPCFetcher()
 
+    def _get_new_client_for_remote(self, remote_ip):
+        """Return a new client for the `remote_ip`.
+
+        Don't use directly called from `get_client_for`.
+        """
+        def store_client(client):
+            self.client_to_remote[remote_ip] = client
+            return client
+
+        d = self.client_service.getClientNow()
+        d.addCallback(store_client)
+        return d
+
+    def get_client_for(self, params):
+        """Always gets the same client based on `params`.
+
+        This is done so that all TFTP requests from the same remote client go
+        to the same regiond process. `RPCFetcher` only duplciate on the client
+        and arguments, so if the client is not the same the duplicate effort
+        is not consolidated.
+        """
+        remote_ip = params.get('remote_ip')
+        if remote_ip:
+            client = self.client_to_remote.get(remote_ip, None)
+            if client is None:
+                # Get a new client for the remote_ip.
+                return self._get_new_client_for_remote(remote_ip)
+            else:
+                # Check that the existing client is still valid.
+                clients = self.client_service.getAllClients()
+                if client in clients:
+                    return succeed(client)
+                else:
+                    del self.client_to_remote[remote_ip]
+                    return self._get_new_client_for_remote(remote_ip)
+        else:
+            return self.client_service.getClientNow()
+
     @inlineCallbacks
     @typed
     def get_boot_method(self, file_name: TFTPPath):
@@ -243,7 +283,7 @@ class TFTPBackend(FilesystemSynchronousBackend):
             d.addCallback(lambda data: KernelParameters(**data))
             return d
 
-        d = self.client_service.getClientNow()
+        d = self.get_client_for(params)
         d.addCallback(fetch, params)
         return d
 
diff --git a/src/provisioningserver/rpc/clusterservice.py b/src/provisioningserver/rpc/clusterservice.py
index 26e2172..a4ab2ba 100644
--- a/src/provisioningserver/rpc/clusterservice.py
+++ b/src/provisioningserver/rpc/clusterservice.py
@@ -106,6 +106,7 @@ from twisted import web
 from twisted.application.internet import TimerService
 from twisted.internet import reactor
 from twisted.internet.defer import (
+    DeferredList,
     inlineCallbacks,
     maybeDeferred,
     returnValue,
@@ -784,6 +785,8 @@ class ClusterClient(Cluster):
             log.msg("Event-loop '%s' authenticated." % self.ident)
             registered = yield self.registerRackWithRegion()
             if registered:
+                if self.eventloop in self.service.try_connections:
+                    del self.service.try_connections[self.eventloop]
                 self.service.connections[self.eventloop] = self
                 self.ready.set(self.eventloop)
             else:
@@ -935,6 +938,8 @@ class ClusterClientService(TimerService, object):
         super(ClusterClientService, self).__init__(
             self._calculate_interval(None, None), self._tryUpdate)
         self.connections = {}
+        self.try_connections = {}
+        self._previous_work = (None, None)
         self.clock = reactor
 
         # XXX jtv 2014-09-23, bug=1372767: Fix
@@ -1172,6 +1177,48 @@ class ClusterClientService(TimerService, object):
             ]
             for name, addresses in eventloops.items()
         }
+
+        drop, connect = self._calculate_work(eventloops)
+
+        # Log fully connected only once. If that state changes then log
+        # it again. This prevents flooding the log with the same message when
+        # the state of the connections has not changed.
+        prev_work, self._previous_work = self._previous_work, (drop, connect)
+        if len(drop) == 0 and len(connect) == 0:
+            if prev_work != (drop, connect) and len(eventloops) > 0:
+                controllers = {
+                    eventloop.split(':')[0]
+                    for eventloop, _ in eventloops.items()
+                }
+                log.msg(
+                    "Fully connected to all %d event-loops on all %d "
+                    "region controllers (%s)." % (
+                        len(eventloops), len(controllers),
+                        ', '.join(sorted(controllers))))
+
+        # Drop all connections at once, as the are no longer required.
+        if len(drop) > 0:
+            log.msg("Dropping connections to event-loops: %s" % (
+                ', '.join(drop.keys())))
+            yield DeferredList([
+                maybeDeferred(self._drop_connection, connection)
+                for eventloop, connections in drop.items()
+                for connection in connections
+            ], consumeErrors=True)
+
+        # Make all the new connections to each endpoint at the same time.
+        if len(connect) > 0:
+            log.msg("Making connections to event-loops: %s" % (
+                ', '.join(connect.keys())))
+            yield DeferredList([
+                self._make_connections(eventloop, addresses)
+                for eventloop, addresses in connect.items()
+            ], consumeErrors=True)
+
+    def _calculate_work(self, eventloops):
+        """Calculate the work that needs to be performed for reconnection."""
+        drop, connect = {}, {}
+
         # Drop connections to event-loops that no longer include one of
         # this cluster's established connections among its advertised
         # endpoints. This is most likely to have happened because of
@@ -1183,23 +1230,20 @@ class ClusterClientService(TimerService, object):
             if eventloop in self.connections:
                 connection = self.connections[eventloop]
                 if connection.address not in addresses:
-                    yield self._drop_connection(connection)
+                    drop[eventloop] = [connection]
+            if eventloop in self.try_connections:
+                connection = self.try_connections[eventloop]
+                if connection.address not in addresses:
+                    drop[eventloop] = [connection]
+
         # Create new connections to event-loops that the cluster does
-        # not yet have a connection to. Try each advertised endpoint
-        # (address) in turn until one of them bites.
+        # not yet have a connection to.
         for eventloop, addresses in eventloops.items():
-            if eventloop not in self.connections:
-                for address in addresses:
-                    try:
-                        yield self._make_connection(eventloop, address)
-                    except ConnectError as error:
-                        host, port = address
-                        log.msg("Event-loop %s (%s:%d): %s" % (
-                            eventloop, host, port, error))
-                    except:
-                        log.err(None, "Failure making new RPC connection.")
-                    else:
-                        break
+            if ((eventloop not in self.connections and
+                    eventloop not in self.try_connections) or
+                    eventloop in drop):
+                connect[eventloop] = addresses
+
         # Remove connections to event-loops that are no longer
         # advertised by the RPC info view. Most likely this means that
         # the process in which the event-loop is no longer running, but
@@ -1208,7 +1252,32 @@ class ClusterClientService(TimerService, object):
         for eventloop in self.connections:
             if eventloop not in eventloops:
                 connection = self.connections[eventloop]
-                yield self._drop_connection(connection)
+                drop[eventloop] = [connection]
+        for eventloop in self.try_connections:
+            if eventloop not in eventloops:
+                connection = self.try_connections[eventloop]
+                drop[eventloop] = [connection]
+
+        return drop, connect
+
+    @inlineCallbacks
+    def _make_connections(self, eventloop, addresses):
+        """Connect to `eventloop` using all `addresses`."""
+        for address in addresses:
+            try:
+                connection = yield self._make_connection(eventloop, address)
+            except ConnectError as error:
+                host, port = address
+                log.msg("Event-loop %s (%s:%d): %s" % (
+                    eventloop, host, port, error))
+            except:
+                host, port = address
+                log.err(None, (
+                    "Failure with event-loop %s (%s:%d)" % (
+                        eventloop, host, port)))
+            else:
+                self.try_connections[eventloop] = connection
+                break
 
     def _make_connection(self, eventloop, address):
         """Connect to `eventloop` at `address`."""
@@ -1227,6 +1296,9 @@ class ClusterClientService(TimerService, object):
         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]
@@ -1242,7 +1314,7 @@ class ClusterClientService(TimerService, object):
                 dhcp_v6.off()
                 stopping_services.append("dhcpd6")
             if len(stopping_services) > 0:
-                maaslog.error(
+                log.msg(
                     "Lost all connections to region controllers. "
                     "Stopping service(s) %s." % ",".join(stopping_services))
                 service_monitor.ensureServices()
diff --git a/src/provisioningserver/rpc/tests/test_clusterservice.py b/src/provisioningserver/rpc/tests/test_clusterservice.py
index e12a107..f8177f4 100644
--- a/src/provisioningserver/rpc/tests/test_clusterservice.py
+++ b/src/provisioningserver/rpc/tests/test_clusterservice.py
@@ -613,7 +613,9 @@ class TestClusterClientService(MAASTestCase):
     @inlineCallbacks
     def test__update_connections_initially(self):
         service = ClusterClientService(Clock())
+        mock_client = Mock()
         _make_connection = self.patch(service, "_make_connection")
+        _make_connection.side_effect = lambda *args: succeed(mock_client)
         _drop_connection = self.patch(service, "_drop_connection")
 
         info = json.loads(self.example_rpc_info_view_response.decode("ascii"))
@@ -627,10 +629,40 @@ class TestClusterClientService(MAASTestCase):
         self.assertItemsEqual(
             _make_connection_expected,
             _make_connection.call_args_list)
+        self.assertEquals({
+            "host1:pid=1001": mock_client,
+            "host1:pid=2002": mock_client,
+            "host2:pid=3003": mock_client,
+        }, service.try_connections)
 
         self.assertEqual([], _drop_connection.mock_calls)
 
     @inlineCallbacks
+    def test__update_connections_logs_fully_connected(self):
+        service = ClusterClientService(Clock())
+        eventloops = {
+            "region1:123": [("::ffff:127.0.0.1", 1234)],
+            "region1:124": [("::ffff:127.0.0.1", 1235)],
+            "region2:123": [("::ffff:127.0.0.2", 1234)],
+            "region2:124": [("::ffff:127.0.0.2", 1235)],
+        }
+        for eventloop, addresses in eventloops.items():
+            for address in addresses:
+                client = Mock()
+                client.address = address
+                service.connections[eventloop] = client
+
+        logger = self.useFixture(TwistedLoggerFixture())
+
+        yield service._update_connections(eventloops)
+        # Second call should not add it to the log.
+        yield service._update_connections(eventloops)
+
+        self.assertEqual(
+            "Fully connected to all 4 event-loops on all 2 region "
+            "controllers (region1, region2).", logger.dump())
+
+    @inlineCallbacks
     def test__update_connections_connect_error_is_logged_tersely(self):
         service = ClusterClientService(Clock())
         _make_connection = self.patch(service, "_make_connection")
@@ -646,6 +678,8 @@ class TestClusterClientService(MAASTestCase):
             MockCalledOnceWith("an-event-loop", ("::ffff:127.0.0.1", 1234)))
 
         self.assertEqual(
+            "Making connections to event-loops: an-event-loop\n"
+            "---\n"
             "Event-loop an-event-loop (::ffff:127.0.0.1:1234): Connection "
             "was refused by other side.", logger.dump())
 
@@ -666,7 +700,9 @@ class TestClusterClientService(MAASTestCase):
 
         self.assertDocTestMatches(
             """\
-            Failure making new RPC connection.
+            Making connections to event-loops: an-event-loop
+            ---
+            Failure with event-loop an-event-loop (::ffff:127.0.0.1:1234)
             Traceback (most recent call last):
             ...
             builtins.RuntimeError: Something went wrong.
@@ -774,6 +810,15 @@ class TestClusterClientService(MAASTestCase):
             connection.transport.loseConnection,
             MockCalledOnceWith())
 
+    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.remove_connection(endpoint, connection)
+        self.assertThat(service.try_connections, Equals({}))
+
     def test__remove_connection_removes_from_connections(self):
         service = make_inert_client_service()
         service.startService()
@@ -1040,6 +1085,7 @@ class TestClusterClient(MAASTestCase):
 
     def test_connecting(self):
         client = self.make_running_client()
+        client.service.try_connections[client.eventloop] = client
         self.patch_authenticate_for_success(client)
         self.patch_register_for_success(client)
         self.assertEqual(client.service.connections, {})
@@ -1053,6 +1099,7 @@ 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(
             client.service.connections,
             {client.eventloop: client})