Merge lp:~blake-rouse/maas/power-query-multiple into lp:~maas-committers/maas/trunk

[Blake Rouse]

Run 5 queries in parallel, when querying the power status of nodes.

[Julian Edwards (julian-edwards)]

Looking good so far. I have a suggestion to make the parallelism better since you're blocking on arbitrary Deferreds rather than the ones that might have fired.

[Blake Rouse (blake-rouse)]

I have updated the branch to stop the blocking on just the oldest defer, if you could give it another test that would be great.

review: Approve

[Blake Rouse (blake-rouse)]

Haha. Sorry about the "Approve" I must press that to much. ;-)

[Blake Rouse (blake-rouse)]

review: Abstain

[Julian Edwards (julian-edwards)]

On Friday 10 Oct 2014 12:34:47 you wrote:
> Haha. Sorry about the "Approve" I must press that to much. ;-)

I laughed :)

[Julian Edwards (julian-edwards)]

On Friday 10 Oct 2014 12:34:07 you wrote:
> I think I will keep it the way it is, because it allows me to make the log
> message type different per exception and the message different for each. I

You could still do that with the single callback, FWIW.

> choose not to catch general exceptions, do you think this is a good idea?
> Or should I just catch all exceptions and maaslog as an unknown error
> printing the exception message.

I think the service has a general catch-all at the top doesn't it? Remember
that the service stops completely if there's any exception, so make sure
that's the case. I think it's fine to omit the catch-all here if there's one
at the top of the service.

Right, I'll go and test this now, the code looks much nicer now with the
semaphore.

[Julian Edwards (julian-edwards)]

Ok tested and it doesn't work. I manually powered up a node and it was not reflected in the UI.

[Julian Edwards (julian-edwards)]

(And I have a good test rig here, I intentionally left power disconnected from some of the BMCs so I have a mix of Microservers with power, some without and some NUCs)

[Blake Rouse (blake-rouse)]

There is an issue using semaphore.run so I went back to using acquire and release.

I was able to test this, and it is now working correctly.

[Gavin Panella (allenap)]

Not a full review yet, but some questions.

review: Needs Information

[Gavin Panella (allenap)]

Comment about handling failures.

[Gavin Panella (allenap)]

Info about cooperate/coiterate.

[Gavin Panella (allenap)]

Looks good.

review: Approve

[Gavin Panella (allenap)]

You may also need http://paste.ubuntu.com/8578497/ to get test_node_power_monitor_service to run successfully. The tests need to run in the reactor thread, otherwise the behaviour of @asynchronous becomes blocking.

[Gavin Panella (allenap)]

Applying http://paste.ubuntu.com/8578519/ (to use semaphore.run) in addition to the above also appears to work fine.

=== modified file 'src/provisioningserver/pserv_services/node_power_monitor_service.py'
--- src/provisioningserver/pserv_services/node_power_monitor_service.py	2014-08-27 09:41:37 +0000
+++ src/provisioningserver/pserv_services/node_power_monitor_service.py	2014-10-20 21:07:30 +0000
@@ -43,6 +43,7 @@
     """Service to monitor the power status of all nodes in this cluster."""
 
     check_interval = timedelta(minutes=5).total_seconds()
+    max_nodes_at_once = 5
 
     def __init__(self, cluster_uuid, clock=None):
         # Call self.query_nodes() every self.check_interval.
@@ -92,4 +93,6 @@
                 uuid)
         else:
             node_power_parameters = response['nodes']
-            yield query_all_nodes(node_power_parameters, clock=self.clock)
+            yield query_all_nodes(
+                node_power_parameters,
+                max_concurrency=self.max_nodes_at_once, clock=self.clock)

=== modified file 'src/provisioningserver/pserv_services/tests/test_node_power_monitor_service.py'
--- src/provisioningserver/pserv_services/tests/test_node_power_monitor_service.py	2014-08-27 09:41:37 +0000
+++ src/provisioningserver/pserv_services/tests/test_node_power_monitor_service.py	2014-10-20 21:07:30 +0000
@@ -21,7 +21,10 @@
     MockCalledOnceWith,
     MockCallsMatch,
     )
-from maastesting.testcase import MAASTestCase
+from maastesting.testcase import (
+    MAASTestCase,
+    MAASTwistedRunTest,
+    )
 from mock import (
     ANY,
     call,
@@ -48,6 +51,8 @@
 
 class TestNodePowerMonitorService(MAASTestCase):
 
+    run_tests_with = MAASTwistedRunTest.make_factory(timeout=5)
+
     def test_init_sets_up_timer_correctly(self):
         cluster_uuid = factory.make_UUID()
         service = npms.NodePowerMonitorService(cluster_uuid)
@@ -118,7 +123,9 @@
         self.assertEqual(None, extract_result(d))
         self.assertThat(
             query_all_nodes,
-            MockCalledOnceWith([], clock=service.clock))
+            MockCalledOnceWith(
+                [], max_concurrency=service.max_nodes_at_once,
+                clock=service.clock))
 
     def test_query_nodes_copes_with_NoSuchCluster(self):
         cluster_uuid, service = self.make_monitor_service()

=== modified file 'src/provisioningserver/rpc/power.py'
--- src/provisioningserver/rpc/power.py	2014-10-06 11:02:29 +0000
+++ src/provisioningserver/rpc/power.py	2014-10-20 21:07:30 +0000
@@ -18,6 +18,7 @@
 ]
 
 from datetime import timedelta
+from functools import partial
 
 from provisioningserver.events import (
     EVENT_TYPES,
@@ -46,6 +47,8 @@
     )
 from twisted.internet import reactor
 from twisted.internet.defer import (
+    DeferredList,
+    DeferredSemaphore,
     inlineCallbacks,
     returnValue,
     )
@@ -314,29 +317,49 @@
     raise PowerActionFail(error)
 
 
-@asynchronous
-@inlineCallbacks
-def query_all_nodes(nodes, clock=reactor):
+def maaslog_query_failure(node, failure):
+    """Log failure to query node."""
+    if failure.check(PowerActionFail):
+        maaslog.error(
+            "%s: Failed to query power state: %s.",
+            node['hostname'], failure.getErrorMessage())
+    elif failure.check(NoSuchNode):
+        maaslog.debug(
+            "%s: Could not update power status; "
+            "no such node.", node['hostname'])
+    else:
+        maaslog.error(
+            "%s: Failed to query power state, unknown error: %s",
+            node['hostname'], failure.getErrorMessage())
+
+
+def maaslog_query(node, power_state):
+    """Log change in power state for node."""
+    if node['power_state'] != power_state:
+        maaslog.info(
+            "%s: Power state has changed from %s to %s.", node['hostname'],
+            node['power_state'], power_state)
+    return power_state
+
+
+def _query_node(node, clock):
+    """Performs `power_query` on the given node."""
+    # Start the query of the power state for the given node, logging to
+    # maaslog as errors and power states change.
+    d = get_power_state(
+        node['system_id'], node['hostname'],
+        node['power_type'], node['context'],
+        clock=clock)
+    d.addCallbacks(
+        partial(maaslog_query, node),
+        partial(maaslog_query_failure, node))
+    return d
+
+
+def query_all_nodes(nodes, max_concurrency=5, clock=reactor):
     """Performs `power_query` on all nodes. If the nodes state has changed,
     then that is sent back to the region."""
-    for node in nodes:
-        system_id = node['system_id']
-        hostname = node['hostname']
-        power_state_recorded = node['power_state']
-        try:
-            power_state_observed = yield get_power_state(
-                system_id, hostname, node['power_type'], node['context'],
-                clock=clock)
-        except PowerActionFail as e:
-            maaslog.error(
-                "%s: Failed to query power state: %s.", hostname, e)
-            continue
-        except NoSuchNode:
-            maaslog.debug(
-                "%s: Could not update power status; "
-                "no such node.", hostname)
-            continue
-        if power_state_observed != power_state_recorded:
-            maaslog.info(
-                "%s: Power state has changed from %s to %s.", hostname,
-                power_state_recorded, power_state_observed)
+    semaphore = DeferredSemaphore(tokens=max_concurrency)
+    return DeferredList(
+        semaphore.run(_query_node, node, clock)
+        for node in nodes)

=== modified file 'src/provisioningserver/rpc/tests/test_power.py'
--- src/provisioningserver/rpc/tests/test_power.py	2014-10-06 11:02:29 +0000
+++ src/provisioningserver/rpc/tests/test_power.py	2014-10-20 21:07:30 +0000
@@ -58,9 +58,11 @@
 from twisted.internet import reactor
 from twisted.internet.defer import (
     Deferred,
+    fail,
     inlineCallbacks,
     maybeDeferred,
     returnValue,
+    succeed,
     )
 from twisted.internet.task import Clock
 
@@ -620,7 +622,10 @@
         # Report back that all nodes' power states are as recorded.
         power_states = [node['power_state'] for node in nodes]
         get_power_state = self.patch(power, 'get_power_state')
-        get_power_state.side_effect = power_states
+        get_power_state.side_effect = [
+            succeed(power_state)
+            for power_state in power_states
+            ]
 
         yield power.query_all_nodes(nodes)
         self.assertThat(get_power_state, MockCallsMatch(*(
@@ -638,7 +643,7 @@
         get_power_state = self.patch(power, 'get_power_state')
         error_msg = factory.make_name("error")
         get_power_state.side_effect = [
-            PowerActionFail(error_msg), new_state_2]
+            fail(PowerActionFail(error_msg)), succeed(new_state_2)]
 
         with FakeLogger("maas.power", level=logging.DEBUG) as maaslog:
             yield power.query_all_nodes([node1, node2])
@@ -656,7 +661,7 @@
         new_state_2 = self.pick_alternate_state(node2['power_state'])
         get_power_state = self.patch(power, 'get_power_state')
         get_power_state.side_effect = [
-            exceptions.NoSuchNode(), new_state_2]
+            fail(exceptions.NoSuchNode()), succeed(new_state_2)]
 
         with FakeLogger("maas.power", level=logging.DEBUG) as maaslog:
             yield power.query_all_nodes([node1, node2])
@@ -668,6 +673,36 @@
             """,
             maaslog.output)
 
+    @inlineCallbacks
+    def test_query_all_nodes_swallows_Exception(self):
+        node1, node2 = self.make_nodes(2)
+        new_state_2 = self.pick_alternate_state(node2['power_state'])
+        get_power_state = self.patch(power, 'get_power_state')
+        get_power_state.side_effect = [
+            fail(Exception('unknown')), succeed(new_state_2)]
+
+        with FakeLogger("maas.power", level=logging.DEBUG) as maaslog:
+            yield power.query_all_nodes([node1, node2])
+
+        self.assertDocTestMatches(
+            """\
+            hostname-...: Failed to query power state, unknown error: unknown
+            hostname-...: Power state has changed from ... to ...
+            """,
+            maaslog.output)
+
+    @inlineCallbacks
+    def test_query_all_nodes_returns_deferredlist_of_number_of_nodes(self):
+        node1, node2 = self.make_nodes(2)
+        get_power_state = self.patch(power, 'get_power_state')
+        get_power_state.side_effect = [
+            succeed(node1['power_state']), succeed(node2['power_state'])]
+
+        results = yield power.query_all_nodes([node1, node2])
+        self.assertEqual(
+            [(True, node1['power_state']), (True, node2['power_state'])],
+            results)
+
 
 class TestMaybeChangePowerState(MAASTestCase):