Merge lp:~blake-rouse/maas/rpc-client-now-2.1 into lp:maas/2.1

[Blake Rouse]

Backport r5725: Add new getClientNow that ensures a connection to a region controller before returning a client. Use the new getClientNow across the rack controller getClient calls.

Backport: https://code.launchpad.net/~blake-rouse/maas/rpc-client-now/+merge/317226

[MAAS Lander (maas-lander)]

Voting does not meet specified criteria. Required: Approve >= 1, Disapprove == 0. Got: .

[Blake Rouse (blake-rouse)]

Self-approving.

review: Approve

=== modified file 'src/provisioningserver/rackdservices/dhcp_probe_service.py'
--- src/provisioningserver/rackdservices/dhcp_probe_service.py	2016-10-28 15:58:32 +0000
+++ src/provisioningserver/rackdservices/dhcp_probe_service.py	2017-02-14 18:27:19 +0000
@@ -107,7 +107,7 @@
         client = None
         for elapsed, remaining, wait in retries(15, 5, self.clock):
             try:
-                client = self.client_service.getClient()
+                client = yield self.client_service.getClientNow()
                 break
             except NoConnectionsAvailable:
                 yield pause(wait, self.clock)

=== modified file 'src/provisioningserver/rackdservices/image_download_service.py'
--- src/provisioningserver/rackdservices/image_download_service.py	2016-10-28 15:58:32 +0000
+++ src/provisioningserver/rackdservices/image_download_service.py	2017-02-14 18:27:19 +0000
@@ -92,7 +92,7 @@
         # the RPC service.
         for elapsed, remaining, wait in retries(15, 5, self.clock):
             try:
-                client = self.client_service.getClient()
+                client = yield self.client_service.getClientNow()
                 break
             except NoConnectionsAvailable:
                 yield pause(wait, self.clock)

=== modified file 'src/provisioningserver/rackdservices/lease_socket_service.py'
--- src/provisioningserver/rackdservices/lease_socket_service.py	2016-03-28 13:54:47 +0000
+++ src/provisioningserver/rackdservices/lease_socket_service.py	2017-02-14 18:27:19 +0000
@@ -106,7 +106,7 @@
         client = None
         for elapsed, remaining, wait in retries(30, 10, clock):
             try:
-                client = self.client_service.getClient()
+                client = yield self.client_service.getClientNow()
                 break
             except NoConnectionsAvailable:
                 yield pause(wait, self.clock)

=== modified file 'src/provisioningserver/rackdservices/networks_monitoring_service.py'
--- src/provisioningserver/rackdservices/networks_monitoring_service.py	2016-10-25 13:57:02 +0000
+++ src/provisioningserver/rackdservices/networks_monitoring_service.py	2017-02-14 18:27:19 +0000
@@ -30,36 +30,46 @@
 
     def getDiscoveryState(self):
         """Get the discovery state from the region."""
-        client = self.clientService.getClient()
         if self._recorded is None:
             # Wait until the rack has refreshed.
             return {}
         else:
-            d = client(
-                GetDiscoveryState, system_id=client.localIdent)
-            d.addCallback(lambda args: args['interfaces'])
+            def getState(client):
+                d = client(
+                    GetDiscoveryState, system_id=client.localIdent)
+                d.addCallback(lambda args: args['interfaces'])
+                return d
+
+            d = self.clientService.getClientNow()
+            d.addCallback(getState)
             return d
 
     def recordInterfaces(self, interfaces):
         """Record the interfaces information to the region."""
-        client = self.clientService.getClient()
-        # On first run perform a refresh
-        if self._recorded is None:
-            return client(RequestRackRefresh, system_id=client.localIdent)
-        else:
-            return client(
-                UpdateInterfaces, system_id=client.localIdent,
-                interfaces=interfaces)
+        def record(client):
+            # On first run perform a refresh
+            if self._recorded is None:
+                return client(RequestRackRefresh, system_id=client.localIdent)
+            else:
+                return client(
+                    UpdateInterfaces, system_id=client.localIdent,
+                    interfaces=interfaces)
+
+        d = self.clientService.getClientNow()
+        d.addCallback(record)
+        return d
 
     def reportNeighbours(self, neighbours):
         """Report neighbour information to the region."""
-        client = self.clientService.getClient()
-        return client(
+        d = self.clientService.getClientNow()
+        d.addCallback(lambda client: client(
             ReportNeighbours, system_id=client.localIdent,
-            neighbours=neighbours)
+            neighbours=neighbours))
+        return d
 
     def reportMDNSEntries(self, mdns):
         """Report mDNS entries to the region."""
-        client = self.clientService.getClient()
-        return client(
-            ReportMDNSEntries, system_id=client.localIdent, mdns=mdns)
+        d = self.clientService.getClientNow()
+        d.addCleanup(lambda client: client(
+            ReportMDNSEntries, system_id=client.localIdent, mdns=mdns))
+        return d

=== modified file 'src/provisioningserver/rackdservices/ntp.py'
--- src/provisioningserver/rackdservices/ntp.py	2016-10-28 15:58:32 +0000
+++ src/provisioningserver/rackdservices/ntp.py	2017-02-14 18:27:19 +0000
@@ -66,7 +66,7 @@
         subsequently obtained configuration objects, allowing this service to
         determine whether a change needs to be applied to the NTP server.
         """
-        client = self._rpc_service.getClient()
+        client = yield self._rpc_service.getClientNow()
         time_configuation = yield client(
             GetTimeConfiguration, system_id=client.localIdent)
         controller_type = yield client(

=== modified file 'src/provisioningserver/rackdservices/service_monitor_service.py'
--- src/provisioningserver/rackdservices/service_monitor_service.py	2016-10-28 15:58:32 +0000
+++ src/provisioningserver/rackdservices/service_monitor_service.py	2017-02-14 18:27:19 +0000
@@ -82,7 +82,7 @@
         client = None
         for elapsed, remaining, wait in retries(30, 10, self.clock):
             try:
-                client = self.client_service.getClient()
+                client = yield self.client_service.getClientNow()
                 break
             except NoConnectionsAvailable:
                 yield pause(wait, self.clock)

=== modified file 'src/provisioningserver/rackdservices/tests/test_dhcp_probe_service.py'
--- src/provisioningserver/rackdservices/tests/test_dhcp_probe_service.py	2016-10-21 18:49:40 +0000
+++ src/provisioningserver/rackdservices/tests/test_dhcp_probe_service.py	2017-02-14 18:27:19 +0000
@@ -103,7 +103,8 @@
             region.ReportForeignDHCPServer.commandName]
 
         rpc_service = Mock()
-        rpc_service.getClient.return_value = getRegionClient()
+        rpc_service.getClientNow.return_value = defer.succeed(
+            getRegionClient())
         service = DHCPProbeService(
             rpc_service, clock)
         yield service.startService()
@@ -160,7 +161,7 @@
         service = DHCPProbeService(
             sentinel.service, clock)
         try_get_client = self.patch(service, '_tryGetClient')
-        try_get_client.getClient = Mock()
+        try_get_client.getClientNow = Mock()
         probe_interface = self.patch(dhcp_probe_service, 'probe_interface')
         yield service.startService()
         yield service.stopService()
@@ -182,7 +183,7 @@
         service = DHCPProbeService(
             sentinel.service, clock)
         try_get_client = self.patch(service, '_tryGetClient')
-        try_get_client.getClient = Mock()
+        try_get_client.getClientNow = Mock()
         probe_interface = self.patch(dhcp_probe_service, 'probe_interface')
         yield service.startService()
         yield service.stopService()
@@ -204,7 +205,7 @@
         service = DHCPProbeService(
             sentinel.service, clock)
         try_get_client = self.patch(service, '_tryGetClient')
-        try_get_client.getClient = Mock()
+        try_get_client.getClientNow = Mock()
         probe_interface = self.patch(dhcp_probe_service, 'probe_interface')
         yield service.startService()
         yield service.stopService()
@@ -226,7 +227,7 @@
         service = DHCPProbeService(
             sentinel.service, clock)
         try_get_client = self.patch(service, '_tryGetClient')
-        try_get_client.getClient = Mock()
+        try_get_client.getClientNow = Mock()
         probe_interface = self.patch(dhcp_probe_service, 'probe_interface')
         yield service.startService()
         yield service.stopService()
@@ -257,7 +258,7 @@
         probe_interface.return_value = [foreign_dhcp_ip]
         client = getRegionClient()
         rpc_service = Mock()
-        rpc_service.getClient.return_value = client
+        rpc_service.getClientNow.return_value = defer.succeed(client)
 
         service = DHCPProbeService(
             rpc_service, clock)
@@ -297,7 +298,7 @@
 
         client = getRegionClient()
         rpc_service = Mock()
-        rpc_service.getClient.return_value = client
+        rpc_service.getClientNow.return_value = defer.succeed(client)
         service = DHCPProbeService(
             rpc_service, clock)
         yield service.startService()

=== modified file 'src/provisioningserver/rackdservices/tests/test_image_download_service.py'
--- src/provisioningserver/rackdservices/tests/test_image_download_service.py	2016-10-18 11:55:44 +0000
+++ src/provisioningserver/rackdservices/tests/test_image_download_service.py	2017-02-14 18:27:19 +0000
@@ -145,7 +145,7 @@
                 http=urlparse(http_proxy),
                 https=urlparse(https_proxy))),
             ]
-        rpc_client.getClient.return_value = client_call
+        rpc_client.getClientNow.return_value = defer.succeed(client_call)
 
         # We could patch out 'import_boot_images' instead here but I
         # don't do that for 2 reasons:
@@ -165,7 +165,7 @@
     def test_no_download_if_no_rpc_connections(self):
         rpc_client = Mock()
         failure = NoConnectionsAvailable()
-        rpc_client.getClient.side_effect = failure
+        rpc_client.getClientNow.return_value = defer.fail(failure)
 
         deferToThread = self.patch(boot_images, 'deferToThread')
         service = ImageDownloadService(

=== modified file 'src/provisioningserver/rackdservices/tests/test_lease_socket_service.py'
--- src/provisioningserver/rackdservices/tests/test_lease_socket_service.py	2016-10-18 11:55:44 +0000
+++ src/provisioningserver/rackdservices/tests/test_lease_socket_service.py	2017-02-14 18:27:19 +0000
@@ -203,7 +203,7 @@
 
         client = getRegionClient()
         rpc_service = MagicMock()
-        rpc_service.getClient.return_value = client
+        rpc_service.getClientNow.return_value = defer.succeed(client)
         service = LeaseSocketService(
             rpc_service, reactor)
 

=== modified file 'src/provisioningserver/rackdservices/tests/test_ntp.py'
--- src/provisioningserver/rackdservices/tests/test_ntp.py	2016-10-04 11:50:27 +0000
+++ src/provisioningserver/rackdservices/tests/test_ntp.py	2017-02-14 18:27:19 +0000
@@ -70,7 +70,7 @@
 class StubClusterClientService:
     """A stub `ClusterClientService` service that's never connected."""
 
-    def getClient(self):
+    def getClientNow(self):
         raise exceptions.NoConnectionsAvailable()
 
 

=== modified file 'src/provisioningserver/rackdservices/tests/test_service_monitor_service.py'
--- src/provisioningserver/rackdservices/tests/test_service_monitor_service.py	2016-10-18 15:51:09 +0000
+++ src/provisioningserver/rackdservices/tests/test_service_monitor_service.py	2017-02-14 18:27:19 +0000
@@ -129,7 +129,7 @@
 
         client = getRegionClient()
         rpc_service = Mock()
-        rpc_service.getClient.return_value = client
+        rpc_service.getClientNow.return_value = succeed(client)
         monitor_service = sms.ServiceMonitorService(
             rpc_service, Clock())
         yield monitor_service.startService()

=== modified file 'src/provisioningserver/rackdservices/tests/test_tftp.py'
--- src/provisioningserver/rackdservices/tests/test_tftp.py	2016-11-01 16:29:24 +0000
+++ src/provisioningserver/rackdservices/tests/test_tftp.py	2017-02-14 18:27:19 +0000
@@ -286,7 +286,7 @@
         client.localIdent = factory.make_name("system_id")
         client.return_value = fail(BootConfigNoResponse())
         client_service = Mock()
-        client_service.getClient.return_value = client
+        client_service.getClientNow.return_value = succeed(client)
         backend = TFTPBackend(
             self.make_dir(), client_service)
 
@@ -301,7 +301,7 @@
         client.localIdent = factory.make_name("system_id")
         client.return_value = fail(exception_type(exception_message))
         client_service = Mock()
-        client_service.getClient.return_value = client
+        client_service.getClientNow.return_value = succeed(client)
         backend = TFTPBackend(
             self.make_dir(), client_service)
 
@@ -405,7 +405,7 @@
         client.localIdent = factory.make_name("system_id")
         client.return_value = succeed(fake_params)
         client_service = Mock()
-        client_service.getClient.return_value = client
+        client_service.getClientNow.return_value = succeed(client)
 
         # get_boot_method_reader() takes a dict() of parameters and returns an
         # `IReader` of a PXE configuration, rendered by
@@ -446,7 +446,7 @@
         client.localIdent = factory.make_name("system_id")
         client.return_value = succeed(fake_params)
         client_service = Mock()
-        client_service.getClient.return_value = client
+        client_service.getClientNow.return_value = succeed(client)
 
         # get_boot_method_reader() takes a dict() of parameters and returns an
         # `IReader` of a PXE configuration, rendered by
@@ -517,9 +517,11 @@
         params_all = params_okay.copy()
         params_all.update(params_other)
 
-        client_service = Mock()
-        client = client_service.getClient.return_value
+        client = Mock()
         client.localIdent = params_okay["system_id"]
+        client_service = Mock()
+        client_service.getClientNow.return_value = succeed(client)
+
         backend = TFTPBackend(self.make_dir(), client_service)
         backend.fetcher = Mock()
 

=== modified file 'src/provisioningserver/rackdservices/tftp.py'
--- src/provisioningserver/rackdservices/tftp.py	2016-11-01 16:29:24 +0000
+++ src/provisioningserver/rackdservices/tftp.py	2017-02-14 18:27:19 +0000
@@ -235,11 +235,16 @@
             name: params[name] for name in arguments
             if name in params
         }
-        client = self.client_service.getClient()
-        params["system_id"] = client.localIdent
-        d = self.fetcher(client, GetBootConfig, **params)
-        d.addCallback(self.get_boot_image, client)
-        d.addCallback(lambda data: KernelParameters(**data))
+
+        def fetch(client, params):
+            params["system_id"] = client.localIdent
+            d = self.fetcher(client, GetBootConfig, **params)
+            d.addCallback(self.get_boot_image, client)
+            d.addCallback(lambda data: KernelParameters(**data))
+            return d
+
+        d = self.client_service.getClientNow()
+        d.addCallback(fetch, params)
         return d
 
     @deferred

=== modified file 'src/provisioningserver/rpc/clusterservice.py'
--- src/provisioningserver/rpc/clusterservice.py	2016-10-28 15:58:32 +0000
+++ src/provisioningserver/rpc/clusterservice.py	2017-02-14 18:27:19 +0000
@@ -96,6 +96,7 @@
 )
 from provisioningserver.utils.twisted import (
     callOut,
+    deferred,
     DeferredValue,
     makeDeferredWithProcessProtocol,
 )
@@ -930,6 +931,26 @@
         else:
             return common.Client(random.choice(conns))
 
+    @deferred
+    def getClientNow(self):
+        """Returns a `Defer` that resolves to a :class:`common.Client`
+        connected to a region.
+
+        If a connection already exists to the region then this method
+        will just return that current connection. If no connections exists
+        this method will try its best to make a connection before returning
+        the client.
+
+        :raises: :py:class:`~.exceptions.NoConnectionsAvailable` when
+            there no connections can be made to a region controller.
+        """
+        try:
+            return self.getClient()
+        except exceptions.NoConnectionsAvailable:
+            d = self._tryUpdate()
+            d.addCallback(lambda _: self.getClient())
+            return d
+
     def getAllClients(self):
         """Return a list of all connected :class:`common.Client`s."""
         return [common.Client(conn) for conn in self.connections.values()]

=== modified file 'src/provisioningserver/rpc/tests/test_clusterservice.py'
--- src/provisioningserver/rpc/tests/test_clusterservice.py	2016-10-28 15:58:32 +0000
+++ src/provisioningserver/rpc/tests/test_clusterservice.py	2017-02-14 18:27:19 +0000
@@ -801,6 +801,55 @@
             exceptions.NoConnectionsAvailable,
             service.getClient)
 
+    @inlineCallbacks
+    def test_getClientNow_returns_current_connection(self):
+        service = ClusterClientService(Clock())
+        service.connections = {
+            sentinel.eventloop01: DummyConnection(),
+            sentinel.eventloop02: DummyConnection(),
+            sentinel.eventloop03: DummyConnection(),
+        }
+        client = yield service.getClientNow()
+        self.assertIn(
+            client, {
+                common.Client(conn)
+                for conn in service.connections.values()
+            })
+
+    @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(),
+            }
+            return succeed(None)
+
+        self.patch(service, "_tryUpdate").side_effect = addConnections
+        client = yield service.getClientNow()
+        self.assertIn(
+            client, {
+                common.Client(conn)
+                for conn in service.connections.values()
+            })
+
+    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()
+        d.addCallback(
+            lambda _: self.fail("Errback should have been called."))
+        d.addErrback(
+            lambda failure: self.assertIsInstance(
+                failure.value, exceptions.NoConnectionsAvailable))
+        return d
+
     def test_getAllClients(self):
         service = ClusterClientService(Clock())
         uuid1 = factory.make_UUID()