Merge lp:~allenap/maas/rpc-client-now-avoiding-storm--2.1 into lp:maas/2.1

[Gavin Panella]

Backport r5754 from trunk: Concurrent calls to getClientNow will not cause an update storm.

[Gavin Panella (allenap)]

Self-reviewing backport.

review: Approve

=== modified file 'src/provisioningserver/rpc/clusterservice.py'
--- src/provisioningserver/rpc/clusterservice.py	2017-02-14 18:24:19 +0000
+++ src/provisioningserver/rpc/clusterservice.py	2017-02-27 14:32:44 +0000
@@ -95,6 +95,7 @@
     select_c_utf8_bytes_locale,
 )
 from provisioningserver.utils.twisted import (
+    call,
     callOut,
     deferred,
     DeferredValue,
@@ -913,6 +914,13 @@
         else:
             twisted.web.client.URI = PatchedURI
 
+        # When _doUpdate is called we capture it into _updateInProgress so
+        # that concurrent calls can piggyback rather than initiating extra
+        # calls. We start with an already-fired DeferredValue: _tryUpdate
+        # checks if it is set to decide whether or not to call _doUpdate.
+        self._updateInProgress = DeferredValue()
+        self._updateInProgress.set(None)
+
     def startService(self):
         self.time_started = self.clock.seconds()
         super(ClusterClientService, self).startService()
@@ -947,9 +955,7 @@
         try:
             return self.getClient()
         except exceptions.NoConnectionsAvailable:
-            d = self._tryUpdate()
-            d.addCallback(lambda _: self.getClient())
-            return d
+            return self._tryUpdate().addCallback(call, self.getClient)
 
     def getAllClients(self):
         """Return a list of all connected :class:`common.Client`s."""
@@ -958,15 +964,19 @@
     def _tryUpdate(self):
         """Attempt to refresh outgoing connections.
 
-        This simply wraps self.update in a deferred to log errors and keep us
-        from dying if an exception is raised in update.
+        This ensures that calls to `_doUpdate` are deferred, with errors
+        logged but not propagated. It also ensures that `_doUpdate` is never
+        called concurrently.
         """
-        d = maybeDeferred(self.update)
-        d.addErrback(log.err, "Cluster client update failed.")
-        return d
+        if self._updateInProgress.isSet:
+            d = maybeDeferred(self._doUpdate).addErrback(
+                log.err, "Cluster client update failed.")
+            self._updateInProgress = DeferredValue()
+            self._updateInProgress.capture(d)
+        return self._updateInProgress.get()
 
     @inlineCallbacks
-    def update(self):
+    def _doUpdate(self):
         """Refresh outgoing connections.
 
         This obtains a list of endpoints from the region then connects

=== modified file 'src/provisioningserver/rpc/tests/test_clusterservice.py'
--- src/provisioningserver/rpc/tests/test_clusterservice.py	2017-02-14 18:24:19 +0000
+++ src/provisioningserver/rpc/tests/test_clusterservice.py	2017-02-27 14:32:44 +0000
@@ -131,6 +131,7 @@
     reactor,
 )
 from twisted.internet.defer import (
+    Deferred,
     fail,
     inlineCallbacks,
     succeed,
@@ -482,7 +483,7 @@
         observed_rpc_info_url = ClusterClientService._get_rpc_info_url()
         self.assertThat(observed_rpc_info_url, Equals(expected_rpc_info_url))
 
-    def test_update_connect_503_error_is_logged_tersely(self):
+    def test__doUpdate_connect_503_error_is_logged_tersely(self):
         getPage = self.patch(clusterservice, "getPage")
         getPage.return_value = fail(web.error.Error("503"))
 
@@ -505,13 +506,13 @@
         logger = self.useFixture(TwistedLoggerFixture())
 
         service = ClusterClientService(Clock())
-        update = self.patch(service, "update")
-        update.side_effect = error.ConnectionRefusedError()
+        _doUpdate = self.patch(service, "_doUpdate")
+        _doUpdate.side_effect = error.ConnectionRefusedError()
 
         # Starting the service causes the first update to be performed, which
         # will fail because of above.
         service.startService()
-        self.assertThat(update, MockCalledOnceWith())
+        self.assertThat(_doUpdate, MockCalledOnceWith())
 
         dump = logger.dump()
         self.assertIn('Connection was refused by other side.', dump)
@@ -570,7 +571,7 @@
         },
     }).encode("ascii")
 
-    def test_update_calls__update_connections(self):
+    def test__doUpdate_calls__update_connections(self):
         maas_url = "http://localhost/%s/" % factory.make_name("path")
         self.useFixture(ClusterConfigurationFixture(maas_url=maas_url))
         self.patch_autospec(socket, 'getaddrinfo').return_value = (
@@ -850,6 +851,29 @@
                 failure.value, exceptions.NoConnectionsAvailable))
         return d
 
+    def test__tryUpdate_prevents_concurrent_calls_to__doUpdate(self):
+        service = ClusterClientService(Clock())
+
+        d_doUpdate_1, d_doUpdate_2 = Deferred(), Deferred()
+        _doUpdate = self.patch(service, "_doUpdate")
+        _doUpdate.side_effect = [d_doUpdate_1, d_doUpdate_2]
+
+        # Try updating a couple of times concurrently.
+        d_tryUpdate_1 = service._tryUpdate()
+        d_tryUpdate_2 = service._tryUpdate()
+        # _doUpdate completes and returns `done`.
+        d_doUpdate_1.callback(sentinel.done1)
+        # Both _tryUpdate calls yield the same result.
+        self.assertThat(extract_result(d_tryUpdate_1), Is(sentinel.done1))
+        self.assertThat(extract_result(d_tryUpdate_2), Is(sentinel.done1))
+        # _doUpdate was called only once.
+        self.assertThat(_doUpdate, MockCalledOnceWith())
+
+        # The mechanism has reset and is ready to go again.
+        d_tryUpdate_3 = service._tryUpdate()
+        d_doUpdate_2.callback(sentinel.done2)
+        self.assertThat(extract_result(d_tryUpdate_3), Is(sentinel.done2))
+
     def test_getAllClients(self):
         service = ClusterClientService(Clock())
         uuid1 = factory.make_UUID()