Merge lp:~julian-edwards/maas/cluster-timer-service into lp:~maas-committers/maas/trunk

[Julian Edwards]

Add a timer service on the cluster that the region can invoke over RPC and get an RPC callback if it doesn't cancel the timer before its deadline is reached.

[Raphaël Badin (rvb)]

Looks good.

review: Approve

=== modified file 'src/provisioningserver/rpc/cluster.py'
--- src/provisioningserver/rpc/cluster.py	2014-09-04 04:40:11 +0000
+++ src/provisioningserver/rpc/cluster.py	2014-09-04 09:43:13 +0000
@@ -320,15 +320,14 @@
     errors = []
 
 
-class CancelTimers(amp.Command):
-    """Cancels existing timer(s) on the cluster.
+class CancelTimer(amp.Command):
+    """Cancels an existing timer on the cluster.
 
     :since: 1.7
     """
 
     arguments = [
-        (b"timers", amp.AmpList(
-            (b"id", amp.Unicode()),
-        ))]
+        (b"id", amp.Unicode()),
+        ]
     response = []
     error = []

=== modified file 'src/provisioningserver/rpc/clusterservice.py'
--- src/provisioningserver/rpc/clusterservice.py	2014-09-02 02:48:11 +0000
+++ src/provisioningserver/rpc/clusterservice.py	2014-09-04 09:43:13 +0000
@@ -54,6 +54,10 @@
     change_power_state,
     get_power_state,
     )
+from provisioningserver.rpc.timers import (
+    cancel_timer,
+    start_timers,
+    )
 from twisted.application.internet import TimerService
 from twisted.internet import ssl
 from twisted.internet.defer import inlineCallbacks
@@ -197,6 +201,16 @@
         remove_host_maps(ip_addresses, shared_key)
         return {}
 
+    @cluster.StartTimers.responder
+    def start_timers(self, timers):
+        start_timers(timers)
+        return {}
+
+    @cluster.CancelTimer.responder
+    def cancel_timer(self, id):
+        cancel_timer(id)
+        return {}
+
     @amp.StartTLS.responder
     def get_tls_parameters(self):
         """get_tls_parameters()

=== modified file 'src/provisioningserver/rpc/tests/test_clusterservice.py'
--- src/provisioningserver/rpc/tests/test_clusterservice.py	2014-09-02 09:30:04 +0000
+++ src/provisioningserver/rpc/tests/test_clusterservice.py	2014-09-04 09:43:13 +0000
@@ -14,6 +14,10 @@
 __metaclass__ = type
 __all__ = []
 
+from datetime import (
+    datetime,
+    timedelta,
+    )
 from itertools import product
 import json
 import os.path
@@ -73,9 +77,14 @@
     DummyConnection,
     StubOS,
     )
+from provisioningserver.rpc.timers import (
+    cancel_timer,
+    running_timers,
+    )
 from provisioningserver.testing.config import set_tftp_root
 from testtools import ExpectedException
 from testtools.matchers import (
+    Contains,
     Equals,
     HasLength,
     Is,
@@ -84,6 +93,7 @@
     MatchesAll,
     MatchesListwise,
     MatchesStructure,
+    Not,
     )
 from twisted.application.internet import TimerService
 from twisted.internet import error
@@ -1052,3 +1062,41 @@
         self.assertThat(
             remove_host_maps, MockCalledOnceWith(
                 ip_addresses, shared_key))
+
+
+class TestClusterProtocol_StartTimers(MAASTestCase):
+
+    def test__is_registered(self):
+        protocol = Cluster()
+        responder = protocol.locateResponder(
+            cluster.StartTimers.commandName)
+        self.assertIsNot(responder, None)
+
+    def test__executes_start_timers(self):
+        deadline = datetime.now(amp.utc) + timedelta(seconds=10)
+        timers = [{
+            "deadline": deadline, "context": factory.make_name("ctx"),
+            "id": factory.make_name("id")}]
+        d = call_responder(Cluster(), cluster.StartTimers, {"timers": timers})
+        self.addCleanup(cancel_timer, timers[0]["id"])
+        self.assertTrue(d.called)
+        self.assertThat(running_timers, Contains(timers[0]["id"]))
+
+
+class TestClusterProtocol_CancelTimer(MAASTestCase):
+
+    def test__is_registered(self):
+        protocol = Cluster()
+        responder = protocol.locateResponder(
+            cluster.CancelTimer.commandName)
+        self.assertIsNot(responder, None)
+
+    def test__executes_cancel_timer(self):
+        deadline = datetime.now(amp.utc) + timedelta(seconds=10)
+        timers = [{
+            "deadline": deadline, "context": factory.make_name("ctx"),
+            "id": factory.make_name("id")}]
+        call_responder(Cluster(), cluster.StartTimers, {"timers": timers})
+
+        call_responder(Cluster(), cluster.CancelTimer, {"id": timers[0]["id"]})
+        self.assertThat(running_timers, Not(Contains(timers[0]["id"])))

=== added file 'src/provisioningserver/rpc/tests/test_timers.py'
--- src/provisioningserver/rpc/tests/test_timers.py	1970-01-01 00:00:00 +0000
+++ src/provisioningserver/rpc/tests/test_timers.py	2014-09-04 09:43:13 +0000
@@ -0,0 +1,133 @@
+# Copyright 2014 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Tests for :py:module:`~provisioningserver.rpc.timers`."""
+
+from __future__ import (
+    absolute_import,
+    print_function,
+    unicode_literals,
+    )
+
+str = None
+
+__metaclass__ = type
+__all__ = []
+
+from datetime import (
+    datetime,
+    timedelta,
+    )
+
+from maastesting.factory import factory
+from maastesting.matchers import MockCalledOnceWith
+from mock import Mock
+from provisioningserver.rpc import timers as timers_module
+from provisioningserver.rpc.region import TimerExpired
+from provisioningserver.rpc.timers import (
+    cancel_timer,
+    running_timers,
+    start_timers,
+    )
+from provisioningserver.testing.testcase import PservTestCase
+from testtools.matchers import (
+    Contains,
+    Equals,
+    HasLength,
+    IsInstance,
+    Not,
+    )
+from twisted.internet.base import DelayedCall
+from twisted.internet.task import Clock
+from twisted.protocols import amp
+
+
+def make_timers(time_now=None):
+    """Make some StartTimers, set to go off one second apart starting in
+    one second"""
+    if time_now is None:
+        time_now = datetime.now(amp.utc)
+    timers = []
+    for i in xrange(2):
+        timers.append({
+            "deadline": time_now + timedelta(seconds=i + 1),
+            "context": factory.make_name("context"),
+            "id": factory.make_name("id"),
+            })
+    return timers
+
+
+class TestStartTimers(PservTestCase):
+    """Tests for `~provisioningserver.rpc.timers.start_timers`."""
+
+    def test__sets_up_running_timers(self):
+        clock = Clock()
+        timers = make_timers()
+        start_timers(timers, clock)
+
+        self.expectThat(running_timers, HasLength(len(timers)))
+        for timer in timers:
+            id = timer["id"]
+            self.expectThat(running_timers[id], IsInstance(tuple))
+            delayed_call, context = running_timers[id]
+            self.expectThat(delayed_call, IsInstance(DelayedCall))
+            self.expectThat(context, Equals(timer["context"]))
+
+    def test__removes_from_running_timers_when_timer_expires(self):
+        self.patch(timers_module, "getRegionClient")
+        clock = Clock()
+        timers = make_timers()
+        start_timers(timers, clock)
+
+        # Expire the first timer.
+        clock.advance(1)
+        self.assertThat(running_timers, Not(Contains(timers[0]["id"])))
+        self.assertThat(running_timers, Contains(timers[1]["id"]))
+
+        # Expire the other time.
+        clock.advance(1)
+        self.assertThat(running_timers, Not(Contains(timers[1]["id"])))
+
+    def test__calls_TimerExpired_when_timer_expires(self):
+        getRegionClient = self.patch(timers_module, "getRegionClient")
+        client = Mock()
+        getRegionClient.return_value = client
+        clock = Clock()
+        timers = make_timers()
+        # Just use the first one for this test.
+        timer = timers[0]
+        start_timers([timer], clock)
+        clock.advance(1)
+
+        self.assertThat(
+            client,
+            MockCalledOnceWith(
+                TimerExpired, id=timer["id"],
+                context=timer["context"]))
+
+
+class TestCancelTimer(PservTestCase):
+    """Tests for `~provisioningserver.rpc.timers.cancel_timer`."""
+
+    def test__cancels_running_timer(self):
+        timers = make_timers()
+        clock = Clock()
+        start_timers(timers, clock)
+        dc, _ = running_timers[timers[0]["id"]]
+
+        cancel_timer(timers[0]["id"])
+
+        self.expectThat(running_timers, Not(Contains(timers[0]["id"])))
+        self.expectThat(running_timers, Contains(timers[1]["id"]))
+        self.assertTrue(dc.cancelled)
+
+    def test__silently_ignores_already_cancelled_timer(self):
+        timers = make_timers()
+        clock = Clock()
+        self.addCleanup(running_timers.clear)
+        start_timers(timers, clock)
+
+        cancel_timer(factory.make_string())
+
+        self.expectThat(running_timers, Contains(timers[0]["id"]))
+        self.expectThat(running_timers, Contains(timers[1]["id"]))

=== added file 'src/provisioningserver/rpc/timers.py'
--- src/provisioningserver/rpc/timers.py	1970-01-01 00:00:00 +0000
+++ src/provisioningserver/rpc/timers.py	2014-09-04 09:43:13 +0000
@@ -0,0 +1,77 @@
+# Copyright 2014 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""RPC helpers for timers."""
+
+from __future__ import (
+    absolute_import,
+    print_function,
+    unicode_literals,
+    )
+
+str = None
+
+__metaclass__ = type
+__all__ = [
+    "cancel_timer",
+    "start_timers",
+]
+
+from datetime import datetime
+
+from provisioningserver.logger import get_maas_logger
+from provisioningserver.rpc import getRegionClient
+from provisioningserver.rpc.exceptions import NoConnectionsAvailable
+from provisioningserver.rpc.region import TimerExpired
+from twisted.internet import reactor
+from twisted.protocols import amp
+
+
+maaslog = get_maas_logger("timers")
+
+
+# Currently running timers; contains dict with keys of ID mapping to a
+# (delayed_call, context) pair.
+running_timers = dict()
+
+
+def start_timers(timers, clock=reactor):
+    """RPC responder to start timers as specified.
+
+    :param timers: a `StartTimers` message.
+
+    Will create one delayed callback for each of the timers and if it
+    reaches its deadline, call `TimerExpired` in the region passing back the
+    timer ID.
+    """
+    for timer in timers:
+        delay = timer["deadline"] - datetime.now(amp.utc)
+        timer_id = timer["id"]
+        call = clock.callLater(delay.total_seconds(), timer_expired, timer_id)
+        running_timers[timer_id] = (call, timer["context"])
+
+
+def timer_expired(timer_id):
+    """Called when a timer hits its deadline.
+
+    Call TimerExpired with the context for the timer.
+    """
+    _, context = running_timers.pop(timer_id)
+    try:
+        client = getRegionClient()
+    except NoConnectionsAvailable:
+        maaslog.error(
+            "Lost connection to the region, unable to fire timer with ID: %s",
+            timer_id)
+        return None
+
+    return client(TimerExpired, id=timer_id, context=context)
+
+
+def cancel_timer(timer_id):
+    """Called from the region to cancel a running timer."""
+    try:
+        dc, _ = running_timers.pop(timer_id)
+    except KeyError:
+        return
+    dc.cancel()