Merge ~blake-rouse/maas:rack-ping-region into maas:master

[Blake Rouse]

Ping the region through RPC every 30 seconds from the rack controller. When ping fails after 10 seconds close the connection.

[MAAS Lander (maas-lander)]

UNIT TESTS
-b rack-ping-region lp:~blake-rouse/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: SUCCESS
COMMIT: a5e703c0140cbbb4f9f59bbf2869a727c0770776

review: Approve

[Andres Rodriguez (andreserl)]

The code looks good to me, it is fairly straightforward.

review: Approve

[MAAS Lander (maas-lander)]

LANDING
-b rack-ping-region lp:~blake-rouse/maas/+git/maas into -b master lp:~maas-committers/maas

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

diff --git a/src/provisioningserver/plugin.py b/src/provisioningserver/plugin.py
index 031d53c..ba46e86 100644
--- a/src/provisioningserver/plugin.py
+++ b/src/provisioningserver/plugin.py
@@ -134,6 +134,14 @@ class ProvisioningServiceMaker:
         rpc_service.setName("rpc")
         return rpc_service
 
+    def _makeRPCPingService(self, rpc_service, clock=reactor):
+        from provisioningserver.rpc.clusterservice import (
+            ClusterClientCheckerService,
+        )
+        service = ClusterClientCheckerService(rpc_service, reactor)
+        service.setName("rpc-ping")
+        return service
+
     def _makeNetworksMonitoringService(self, rpc_service, clock=reactor):
         from provisioningserver.rackdservices.networks_monitoring_service \
             import RackNetworksMonitoringService
@@ -167,6 +175,7 @@ class ProvisioningServiceMaker:
         rpc_service = self._makeRPCService()
         yield rpc_service
         # Other services that make up the MAAS Region Controller.
+        yield self._makeRPCPingService(rpc_service, clock=clock)
         yield self._makeNetworksMonitoringService(rpc_service, clock=clock)
         yield self._makeDHCPProbeService(rpc_service)
         yield self._makeLeaseSocketService(rpc_service)
diff --git a/src/provisioningserver/rpc/clusterservice.py b/src/provisioningserver/rpc/clusterservice.py
index cd631c7..1ab1122 100644
--- a/src/provisioningserver/rpc/clusterservice.py
+++ b/src/provisioningserver/rpc/clusterservice.py
@@ -62,7 +62,10 @@ from provisioningserver.rpc.boot_images import (
     is_import_boot_images_running,
     list_boot_images,
 )
-from provisioningserver.rpc.common import RPCProtocol
+from provisioningserver.rpc.common import (
+    Ping,
+    RPCProtocol,
+)
 from provisioningserver.rpc.interfaces import IConnectionToRegion
 from provisioningserver.rpc.osystems import (
     gen_operating_systems,
@@ -1162,10 +1165,12 @@ class ClusterClientService(TimerService, object):
             # Fully connected to the region; update every so often.
             return self.INTERVAL_HIGH
 
-    def _update_interval(self, num_eventloops, num_connections):
+    def _update_interval(self, num_eventloops, num_connections, reset=False):
         """Change the update interval."""
         self._loop.interval = self.step = self._calculate_interval(
             num_eventloops, num_connections)
+        if reset and self._loop.running:
+            self._loop.reset()
 
     @inlineCallbacks
     def _update_connections(self, eventloops):
@@ -1337,6 +1342,50 @@ class ClusterClientService(TimerService, object):
                     "Lost all connections to region controllers. "
                     "Stopping service(s) %s." % ",".join(stopping_services))
                 service_monitor.ensureServices()
-        # Lower the interval so a re-check happens sooner instead of its
-        # currently set interval.
-        self._update_interval(0, 0)
+        # Lower and reset the interval so a reconnection happens.
+        self._update_interval(0, 0, reset=True)
+
+
+class ClusterClientCheckerService(TimerService, object):
+    """A cluster controller RPC client checker service.
+
+    This is a service - in the Twisted sense - that cordinates with the
+    `ClusterClientService` to ensure that all RPC connections are functional.
+    A ping is performed over each current connection to ensure that the
+    connection is working properly. If connection is not operational then it
+    is dropped allowing the `ClusterClientService` to make a new connection.
+
+    :ivar client_service: The `ClusterClientService` instance.
+    """
+
+    def __init__(self, client_service, reactor):
+        super(ClusterClientCheckerService, self).__init__(30, self.tryLoop)
+        self.client_service = client_service
+        self.clock = reactor
+
+    def tryLoop(self):
+        d = self.loop()
+        d.addErrback(
+            log.err, "Failure while performing ping on RPC connections.")
+        return d
+
+    def loop(self):
+        return DeferredList([
+            self._ping(client)
+            for client in self.client_service.getAllClients()
+        ], consumeErrors=True)
+
+    def _ping(self, client):
+        """Ping the client to ensure it works."""
+
+        def _onFailure(failure):
+            log.msg(
+                "Failure on ping dropping connection to event-loop: %s" % (
+                    client.ident))
+            # The protocol will call `remove_connection on the
+            # `ClusterClientService` that will perform the reconnection.
+            client._conn.transport.loseConnection()
+
+        d = client(Ping, _timeout=10)
+        d.addErrback(_onFailure)
+        return d
diff --git a/src/provisioningserver/rpc/common.py b/src/provisioningserver/rpc/common.py
index be168cf..6f74366 100644
--- a/src/provisioningserver/rpc/common.py
+++ b/src/provisioningserver/rpc/common.py
@@ -79,6 +79,17 @@ class Authenticate(amp.Command):
     errors = []
 
 
+class Ping(amp.Command):
+    """Ensure the connection is still good.
+
+    :since: 2.4
+    """
+
+    arguments = []
+    response = []
+    errors = []
+
+
 class Client:
     """Wrapper around an :class:`amp.AMP` instance.
 
@@ -283,3 +294,12 @@ class RPCProtocol(amp.AMP, object):
             "Unhandled failure during AMP request. This is probably a bug. "
             "Please ensure that this error is handled within application "
             "code."))
+
+    @Ping.responder
+    def ping(self):
+        """ping()
+
+        Implementation of
+        :py:class:`~provisioningserver.rpc.common.Ping`.
+        """
+        return {}
diff --git a/src/provisioningserver/rpc/tests/test_clusterservice.py b/src/provisioningserver/rpc/tests/test_clusterservice.py
index 1c9eb0f..2989a2c 100644
--- a/src/provisioningserver/rpc/tests/test_clusterservice.py
+++ b/src/provisioningserver/rpc/tests/test_clusterservice.py
@@ -90,6 +90,7 @@ from provisioningserver.rpc import (
 from provisioningserver.rpc.clusterservice import (
     Cluster,
     ClusterClient,
+    ClusterClientCheckerService,
     ClusterClientService,
     executeScanNetworksSubprocess,
     get_scan_all_networks_args,
@@ -131,7 +132,10 @@ from testtools.matchers import (
 )
 from twisted import web
 from twisted.application.internet import TimerService
-from twisted.internet import error
+from twisted.internet import (
+    error,
+    reactor,
+)
 from twisted.internet.defer import (
     Deferred,
     fail,
@@ -1534,6 +1538,68 @@ class TestClusterClient(MAASTestCase):
                 version=get_maas_version()))
 
 
+class TestClusterClientCheckerService(MAASTestCase):
+
+    run_tests_with = MAASTwistedRunTest.make_factory(timeout=5)
+
+    def make_client(self):
+        client = Mock()
+        client.return_value = succeed(None)
+        return client
+
+    def test_init_sets_up_timer_correctly(self):
+        service = ClusterClientCheckerService(
+            sentinel.client_service, sentinel.clock)
+        self.assertThat(service, MatchesStructure.byEquality(
+            call=(service.tryLoop, (), {}),
+            step=(30), client_service=sentinel.client_service,
+            clock=sentinel.clock))
+
+    def test_tryLoop_calls_loop(self):
+        service = ClusterClientCheckerService(
+            sentinel.client_service, sentinel.clock)
+        mock_loop = self.patch(service, "loop")
+        mock_loop.return_value = succeed(None)
+        service.tryLoop()
+        self.assertThat(
+            mock_loop,
+            MockCalledOnceWith())
+
+    def test_loop_does_nothing_with_no_clients(self):
+        mock_client_service = MagicMock()
+        mock_client_service.getAllClients.return_value = []
+        service = ClusterClientCheckerService(
+            mock_client_service, reactor)
+        # Test will timeout if this blocks longer than 5 seconds.
+        return service.loop()
+
+    @inlineCallbacks
+    def test_loop_calls_ping_for_each_client(self):
+        clients = [
+            self.make_client()
+            for _ in range(3)
+        ]
+        mock_client_service = MagicMock()
+        mock_client_service.getAllClients.return_value = clients
+        service = ClusterClientCheckerService(
+            mock_client_service, reactor)
+        yield service.loop()
+        for client in clients:
+            self.expectThat(
+                client, MockCalledOnceWith(common.Ping, _timeout=10))
+
+    @inlineCallbacks
+    def test_ping_calls_loseConnection_on_failure(self):
+        client = MagicMock()
+        client.return_value = fail(factory.make_exception())
+        mock_client_service = MagicMock()
+        service = ClusterClientCheckerService(
+            mock_client_service, reactor)
+        yield service._ping(client)
+        self.assertThat(
+            client._conn.transport.loseConnection, MockCalledOnceWith())
+
+
 class TestClusterProtocol_ListSupportedArchitectures(MAASTestCase):
 
     run_tests_with = MAASTwistedRunTest.make_factory(timeout=5)
diff --git a/src/provisioningserver/tests/test_plugin.py b/src/provisioningserver/tests/test_plugin.py
index d62b765..caa9409 100644
--- a/src/provisioningserver/tests/test_plugin.py
+++ b/src/provisioningserver/tests/test_plugin.py
@@ -45,6 +45,7 @@ from provisioningserver.rackdservices.tftp import (
     TFTPService,
 )
 from provisioningserver.rackdservices.tftp_offload import TFTPOffloadService
+from provisioningserver.rpc.clusterservice import ClusterClientCheckerService
 from provisioningserver.testing.config import ClusterConfigurationFixture
 from provisioningserver.utils.twisted import reducedWebLogFormatter
 from testtools.matchers import (
@@ -105,8 +106,8 @@ class TestProvisioningServiceMaker(MAASTestCase):
         self.assertIsInstance(service, MultiService)
         expected_services = [
             "dhcp_probe", "networks_monitor", "image_download",
-            "lease_socket_service", "node_monitor", "ntp", "rpc", "tftp",
-            "image_service", "service_monitor",
+            "lease_socket_service", "node_monitor", "ntp", "rpc", "rpc-ping",
+            "tftp", "image_service", "service_monitor",
             ]
         self.assertThat(service.namedServices, KeysEqual(*expected_services))
         self.assertEqual(
@@ -177,6 +178,13 @@ class TestProvisioningServiceMaker(MAASTestCase):
         service_monitor = service.getServiceNamed("service_monitor")
         self.assertIsInstance(service_monitor, ServiceMonitorService)
 
+    def test_rpc_ping_service(self):
+        options = Options()
+        service_maker = ProvisioningServiceMaker("Harry", "Hill")
+        service = service_maker.makeService(options, clock=None)
+        rpc_ping = service.getServiceNamed("rpc-ping")
+        self.assertIsInstance(rpc_ping, ClusterClientCheckerService)
+
     def test_tftp_service(self):
         # A TFTP service is configured and added to the top-level service.
         options = Options()