Merge lp:~julian-edwards/maas/periodic-lease-upload-pserv into lp:~maas-committers/maas/trunk

[Julian Edwards]

Add a pserv service that replaces the Celery task that periodically uploads DHCP leases.

This branch contains everything needed to remove the old Celery task to upload leases, and replace it with a pserv-based one.

Roughly, the following changes were made:
 * Removing all trace of the celery config, code and tests
 * Update dev documentation
 * RPC command definition for UpdateLeases call
 * Code to translate between the RPC format and the format needed by the existing code that does the actual updating.
 * New pserv service.
 * Tests. Lots of tests!

Sorry it's on the large side, but it is a complete cut-over. I've tested that it works on my local rig.

[Gavin Panella (allenap)]

Looking good so far :) Lots of comments, but none of them are blockers.

[Julian Edwards (julian-edwards)]

Thanks for the comments!

I tried your fixture suggestion in the last test but I can't get it to work. Can you take a look please?

Take this branch and apply this diff:
http://paste.ubuntu.com/8095590/

I get this output:
======================================================================
FAIL: provisioningserver.tests.test_lease_upload_service.TestPeriodicImageDownloadService.test_upload_is_initiated
----------------------------------------------------------------------
_StringException: Failed expectation: {{{
File "/home/ed/canonical/maas/sandbox/src/maastesting/crochet.py", line 101, in __checkResults
    fail_count, Equals(0), "Unfired and/or unhandled "
  File "/usr/lib/python2.7/dist-packages/testtools/testcase.py", line 447, in expectThat
    postfix_content="MismatchError: " + str(mismatch_error)
MismatchError: 0 != 1: Unfired and/or unhandled EventualResult(s); see test details.
}}}

Unfired/unhandled EventualResult #1: {{{
*** EventualResult has not fired:
<crochet._eventloop.EventualResult object at 0x7f9e58fcfbd0>
*** It was connected to a Deferred:
<Deferred at 0x7f9e58fd35a8>
}}}

traceback-1: {{{
Traceback (most recent call last):
AssertionError: Forced Test Failure
}}}

Traceback (most recent call last):
  File "/home/ed/canonical/maas/sandbox/src/provisioningserver/tests/test_lease_upload_service.py", line 214, in test_upload_is_initiated
    MockCalledOnceWith(uuid=uuid, mappings=mappings))
  File "/usr/lib/python2.7/dist-packages/testtools/testcase.py", line 406, in assertThat
    raise mismatch_error
MismatchError: Expected to be called once. Called 0 times.

-------------------- >> begin captured logging << --------------------
twisted: INFO: AMPTestProtocol#1 connection established (HOST:<twisted.test.iosim.FakeAddress object at 0x7f9e58fc64d0> PEER:<twisted.test.iosim.FakeAddress object at 0x7f9e58fc6510>)
twisted: INFO: ClusterClient connection established (HOST:<twisted.test.iosim.FakeAddress object at 0x7f9e58fc6550> PEER:<twisted.test.iosim.FakeAddress object at 0x7f9e58fc6590>)
--------------------- >> end captured logging << ---------------------

[Gavin Panella (allenap)]

Have a look at https://code.launchpad.net/~allenap/maas/periodic-lease-upload-pserv/+merge/231538.

[Julian Edwards (julian-edwards)]

This is now all working on my test rig. I still can't get your modified tests to pass though, so I'll get a review and land without the changes as your off today and then we can look at it again when you're back.

[Julian Edwards (julian-edwards)]

I made a couple more changes:
 - Catch service errors and log them — the code may look somewhat familiar to you :)
 - moved the code to the new pserv_services dir

[Gavin Panella (allenap)]

Looks grand.

review: Approve

[Julian Edwards (julian-edwards)]

Thanks for reviewing! I'm going to leave the lock for now, see my comment. It's easily changed later if needed.

[Julian Edwards (julian-edwards)]

Tested again on my local rig and it seems fine, so landing now.

=== modified file 'docs/development/lease-scanning-and-dns.rst'
--- docs/development/lease-scanning-and-dns.rst	2013-09-13 04:08:22 +0000
+++ docs/development/lease-scanning-and-dns.rst	2014-08-28 03:19:24 +0000
@@ -20,16 +20,10 @@
 ===============
 
 MAAS will periodically scan the DHCP leases file using the
-``upload_dhcp_leases()`` celery task, which originates via celerybeat.
-
-As leases are discovered, it calls the api function ``update_leases()`` which
-will subsequently generate a new task called ``add_new_dhcp_host_map()`` that
-will use omshell to write a permanent mapping from MAC to IP in the DHCP
-server.
-
-That host map remains until the node is deleted, at which point the
-``remove_dhcp_host_map()`` task is invoked which uses omshell to remove the
-map.
+``PeriodicLeaseUploadService()`` pserv service.
+
+As leases are discovered, it calls the RPC function ``UpdateLeases`` which
+stores the active leases in the DHCPLease table.
 
 
 Updating the DNS zone file

=== modified file 'etc/celeryconfig_cluster.py'
--- etc/celeryconfig_cluster.py	2014-08-26 23:55:04 +0000
+++ etc/celeryconfig_cluster.py	2014-08-28 03:19:24 +0000
@@ -40,14 +40,6 @@
 PROBE_DHCP_SERVERS_SCHEDULE = timedelta(minutes=1)
 
 CELERYBEAT_SCHEDULE = {
-    'unconditional-dhcp-lease-upload': {
-        'task': 'provisioningserver.tasks.upload_dhcp_leases',
-        'schedule': DHCP_LEASE_UPLOAD_SCHEDULE,
-        'options': {
-            'queue': CLUSTER_UUID,
-            'expires': int(DHCP_LEASE_UPLOAD_SCHEDULE.total_seconds()),
-        },
-    },
     'report-boot-images': {
         'task': 'provisioningserver.tasks.report_boot_images',
         'schedule': REPORT_BOOT_IMAGES_SCHEDULE,

=== added file 'src/maasserver/rpc/leases.py'
--- src/maasserver/rpc/leases.py	1970-01-01 00:00:00 +0000
+++ src/maasserver/rpc/leases.py	2014-08-28 03:19:24 +0000
@@ -0,0 +1,45 @@
+# Copyright 2014 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""RPC helpers relating to DHCP leases."""
+
+from __future__ import (
+    absolute_import,
+    print_function,
+    unicode_literals,
+    )
+
+str = None
+
+__metaclass__ = type
+__all__ = [
+    "update_leases",
+]
+
+from django.shortcuts import get_object_or_404
+from maasserver.models.dhcplease import DHCPLease
+from maasserver.models.nodegroup import NodeGroup
+from maasserver.utils.async import transactional
+from provisioningserver.pserv_services.lease_upload_service import (
+    convert_mappings_to_leases,
+    )
+from provisioningserver.utils.twisted import synchronous
+
+
+@synchronous
+@transactional
+def update_leases(uuid, mappings):
+    """Updates DHCP leases on a cluster given the mappings in UpdateLeases.
+
+    :param uuid: Cluster UUID as found in
+        :py:class`~provisioningserver.rpc.region.UpdateLeases`.
+    :param mappings: List of pairs of (ip, mac) as defined in
+        :py:class`~provisioningserver.rpc.region.UpdateLeases`.
+
+    Converts the mappings format into a dict that
+    DHCPLease.objects.update_leases needs and then calls it.
+    """
+    nodegroup = get_object_or_404(NodeGroup, uuid=uuid)
+    leases = convert_mappings_to_leases(mappings)
+    DHCPLease.objects.update_leases(nodegroup, leases)
+    return {}

=== modified file 'src/maasserver/rpc/regionservice.py'
--- src/maasserver/rpc/regionservice.py	2014-08-26 14:29:13 +0000
+++ src/maasserver/rpc/regionservice.py	2014-08-28 03:19:24 +0000
@@ -34,6 +34,7 @@
     bootsources,
     configuration,
     events,
+    leases,
     nodes,
     )
 from maasserver.utils import synchronised
@@ -99,6 +100,15 @@
         """
         return {}
 
+    @region.UpdateLeases.responder
+    def update_leases(self, uuid, mappings):
+        """update_leases(uuid, mappings)
+
+        Implementation of
+        :py:class`~provisioningserver.rpc.region.UpdateLeases`.
+        """
+        return deferToThread(leases.update_leases, uuid, mappings)
+
     @amp.StartTLS.responder
     def get_tls_parameters(self):
         """get_tls_parameters()

=== modified file 'src/maasserver/rpc/tests/test_regionservice.py'
--- src/maasserver/rpc/tests/test_regionservice.py	2014-08-26 17:32:40 +0000
+++ src/maasserver/rpc/tests/test_regionservice.py	2014-08-28 03:19:24 +0000
@@ -34,6 +34,7 @@
     POWER_STATE,
     )
 from maasserver.models.config import Config
+from maasserver.models.dhcplease import DHCPLease
 from maasserver.models.event import Event
 from maasserver.models.eventtype import EventType
 from maasserver.models.node import Node
@@ -80,6 +81,7 @@
     RegisterEventType,
     ReportBootImages,
     SendEvent,
+    UpdateLeases,
     UpdateNodePowerState,
     )
 from provisioningserver.rpc.testing import (
@@ -234,6 +236,44 @@
         return d.addCallback(check)
 
 
+class TestRegionProtocol_UpdateLeases(TransactionTestCase):
+
+    def test_update_leases_is_registered(self):
+        protocol = Region()
+        responder = protocol.locateResponder(UpdateLeases.commandName)
+        self.assertIsNot(responder, None)
+
+    @transactional
+    def make_node_group(self, uuid):
+        return factory.make_node_group(uuid=uuid)
+
+    @transactional
+    def get_leases_for(self, nodegroup):
+        return [
+            (ng.ip, ng.mac)
+            for ng in DHCPLease.objects.filter(nodegroup=nodegroup)]
+
+    @wait_for_reactor
+    @inlineCallbacks
+    def test__stores_leases(self):
+        uuid = factory.make_name("uuid")
+        nodegroup = yield deferToThread(self.make_node_group, uuid)
+        mapping = {
+            "ip": factory.getRandomIPAddress(),
+            "mac": factory.getRandomMACAddress()
+        }
+
+        response = yield call_responder(Region(), UpdateLeases, {
+            b"uuid": uuid, b"mappings": [mapping]})
+
+        self.assertThat(response, Equals({}))
+
+        [(ip, mac)] = yield deferToThread(
+            self.get_leases_for, nodegroup=nodegroup)
+        self.expectThat(ip, Equals(mapping["ip"]))
+        self.expectThat(mac, Equals(mapping["mac"]))
+
+
 class TestRegionProtocol_GetBootSources(TransactionTestCase):
 
     def test_get_boot_sources_is_registered(self):

=== modified file 'src/provisioningserver/dhcp/leases.py'
--- src/provisioningserver/dhcp/leases.py	2014-08-13 21:49:35 +0000
+++ src/provisioningserver/dhcp/leases.py	2014-08-28 03:19:24 +0000
@@ -34,6 +34,7 @@
     ]
 
 
+from collections import defaultdict
 import errno
 import json
 from os import (
@@ -46,8 +47,6 @@
     MAASDispatcher,
     MAASOAuth,
     )
-from celery.app import app_or_default
-from provisioningserver import cache
 from provisioningserver.auth import (
     get_recorded_api_credentials,
     get_recorded_nodegroup_uuid,
@@ -59,6 +58,12 @@
 
 maaslog = get_maas_logger("dhcp.leases")
 
+# This used to be the cache in provisioningserver.cache, but that
+# unfortunately makes Twisted fail in ways we can't work out, probably
+# because of its use of multiprocessing.  Instead it's now using a
+# simple dict, because there are no plans to make pserv multi-process.
+cache = defaultdict()
+
 
 # Cache key for the modification time on last-processed leases file.
 LEASES_TIME_CACHE_KEY = 'leases_time'
@@ -69,8 +74,13 @@
 
 
 def get_leases_file():
-    """Get the location of the DHCP leases file from the config."""
-    return app_or_default().conf.DHCP_LEASES_FILE
+    """Return the location of the DHCP leases file."""
+    # This used to be celery config-based so that the development env could
+    # have a different location.  However, nobody seems to be
+    # provisioning from a dev environment so it's hard-coded until that
+    # need arises, as converting to the pserv config would be wasted
+    # work right now.
+    return "/var/lib/maas/dhcp/dhcpd.leases"
 
 
 def get_leases_timestamp():
@@ -115,8 +125,8 @@
     # These variables are shared between worker threads/processes.
     # A bit of inconsistency due to concurrent updates is not a problem,
     # but read them both at once here to reduce the scope for trouble.
-    previous_leases = cache.cache.get(LEASES_CACHE_KEY)
-    previous_leases_time = cache.cache.get(LEASES_TIME_CACHE_KEY)
+    previous_leases = cache.get(LEASES_CACHE_KEY)
+    previous_leases_time = cache.get(LEASES_TIME_CACHE_KEY)
 
     if get_leases_timestamp() == previous_leases_time:
         return None
@@ -139,8 +149,8 @@
     :param leases: A dict mapping each leased IP address to the MAC address
         that it has been assigned to.
     """
-    cache.cache.set(LEASES_TIME_CACHE_KEY, last_change)
-    cache.cache.set(LEASES_CACHE_KEY, leases)
+    cache[LEASES_TIME_CACHE_KEY] = last_change
+    cache[LEASES_CACHE_KEY] = leases
 
 
 def list_missing_items(knowledge):

=== modified file 'src/provisioningserver/dhcp/tests/test_leases.py'
--- src/provisioningserver/dhcp/tests/test_leases.py	2014-07-18 17:05:57 +0000
+++ src/provisioningserver/dhcp/tests/test_leases.py	2014-08-28 03:19:24 +0000
@@ -30,10 +30,10 @@
     get_write_time,
     )
 from mock import Mock
-from provisioningserver import cache
 from provisioningserver.auth import NODEGROUP_UUID_CACHE_KEY
 from provisioningserver.dhcp import leases as leases_module
 from provisioningserver.dhcp.leases import (
+    cache,
     check_lease_changes,
     LEASES_CACHE_KEY,
     LEASES_TIME_CACHE_KEY,
@@ -56,8 +56,8 @@
         record_lease_state(time, leases)
         self.assertEqual(
             (time, leases), (
-                cache.cache.get(LEASES_TIME_CACHE_KEY),
-                cache.cache.get(LEASES_CACHE_KEY),
+                cache.get(LEASES_TIME_CACHE_KEY),
+                cache.get(LEASES_CACHE_KEY),
                 ))
 
 
@@ -115,12 +115,12 @@
     def set_nodegroup_uuid(self):
         """Set the recorded nodegroup uuid for the duration of this test."""
         uuid = factory.make_UUID()
-        cache.cache.set(NODEGROUP_UUID_CACHE_KEY, uuid)
+        cache[NODEGROUP_UUID_CACHE_KEY] = uuid
         return uuid
 
     def clear_nodegroup_uuid(self):
         """Clear the recorded nodegroup uuid."""
-        cache.cache.set(NODEGROUP_UUID_CACHE_KEY, None)
+        cache[NODEGROUP_UUID_CACHE_KEY] = None
 
     def set_maas_url(self):
         """Set the recorded MAAS URL for the duration of this test."""
@@ -138,11 +138,11 @@
         """Set recorded API credentials for the duration of this test."""
         creds_string = ':'.join(
             factory.make_string() for counter in range(3))
-        cache.cache.set('api_credentials', creds_string)
+        cache['api_credentials'] = creds_string
 
     def clear_api_credentials(self):
         """Clear recorded API credentials."""
-        cache.cache.set('api_credentials', None)
+        cache['api_credentials'] = None
 
     def set_items_needed_for_lease_update(self):
         """Set the recorded items required by `update_leases`."""
@@ -157,8 +157,8 @@
         state so that it gets reset at the end of the test.  Using this will
         prevent recorded lease state from leaking into other tests.
         """
-        cache.cache.set(LEASES_TIME_CACHE_KEY, time)
-        cache.cache.set(LEASES_CACHE_KEY, leases)
+        cache[LEASES_TIME_CACHE_KEY] = time
+        cache[LEASES_CACHE_KEY] = leases
 
     def test_record_lease_state_sets_leases_and_timestamp(self):
         time = datetime.utcnow()
@@ -167,8 +167,8 @@
         record_lease_state(time, leases)
         self.assertEqual(
             (time, leases), (
-                cache.cache.get(LEASES_TIME_CACHE_KEY),
-                cache.cache.get(LEASES_CACHE_KEY),
+                cache.get(LEASES_TIME_CACHE_KEY),
+                cache.get(LEASES_CACHE_KEY),
                 ))
 
     def test_check_lease_changes_returns_tuple_if_no_state_cached(self):

=== modified file 'src/provisioningserver/plugin.py'
--- src/provisioningserver/plugin.py	2014-08-27 03:38:28 +0000
+++ src/provisioningserver/plugin.py	2014-08-28 03:19:24 +0000
@@ -34,6 +34,9 @@
 from provisioningserver.pserv_services.image_download_service import (
     PeriodicImageDownloadService,
     )
+from provisioningserver.pserv_services.lease_upload_service import (
+    PeriodicLeaseUploadService,
+    )
 from provisioningserver.pserv_services.node_power_monitor_service import (
     NodePowerMonitorService,
     )
@@ -232,6 +235,12 @@
         image_download_service.setName("image_download")
         return image_download_service
 
+    def _makePeriodicLeaseUploadService(self, rpc_service):
+        lease_upload_service = PeriodicLeaseUploadService(
+            rpc_service, reactor, get_cluster_uuid())
+        lease_upload_service.setName("lease_upload")
+        return lease_upload_service
+
     def _makeNodePowerMonitorService(self, rpc_service):
         node_monitor = NodePowerMonitorService(
             rpc_service, reactor, get_cluster_uuid())
@@ -267,4 +276,8 @@
             rpc_service)
         image_download_service.setServiceParent(services)
 
+        lease_upload_service = self._makePeriodicLeaseUploadService(
+            rpc_service)
+        lease_upload_service.setServiceParent(services)
+
         return services

=== added file 'src/provisioningserver/pserv_services/lease_upload_service.py'
--- src/provisioningserver/pserv_services/lease_upload_service.py	1970-01-01 00:00:00 +0000
+++ src/provisioningserver/pserv_services/lease_upload_service.py	2014-08-28 03:19:24 +0000
@@ -0,0 +1,145 @@
+# Copyright 2014 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Twisted service that periodically uploads DHCP leases to the region."""
+
+from __future__ import (
+    absolute_import,
+    print_function,
+    unicode_literals,
+    )
+
+str = None
+
+__metaclass__ = type
+__all__ = [
+    "convert_leases_to_mappings",
+    "convert_mappings_to_leases",
+    "PeriodicLeaseUploadService",
+    ]
+
+
+from provisioningserver.dhcp.leases import (
+    check_lease_changes,
+    record_lease_state,
+    )
+from provisioningserver.logger import get_maas_logger
+from provisioningserver.rpc.exceptions import NoConnectionsAvailable
+from provisioningserver.rpc.region import UpdateLeases
+from provisioningserver.utils.twisted import (
+    pause,
+    retries,
+    )
+from twisted.application.internet import TimerService
+from twisted.internet.defer import (
+    DeferredLock,
+    inlineCallbacks,
+    )
+from twisted.internet.threads import deferToThread
+from twisted.python import log
+
+
+maaslog = get_maas_logger("lease_upload_service")
+service_lock = DeferredLock()
+
+
+def convert_mappings_to_leases(mappings):
+    """Convert AMP mappings to record_lease_state() leases.
+
+    Take mappings, as used by UpdateLeases, and turn into leases
+    as used by record_lease_state().
+    """
+    return {
+        mapping["ip"]: mapping["mac"]
+        for mapping in mappings
+    }
+
+
+def convert_leases_to_mappings(leases):
+    """Convert record_lease_state() leases into UpdateLeases mappings.
+
+    Take the leases dict, as returned by record_lease_state(), and
+    turn it into a mappings list suitable for transportation in
+    the UpdateLeases AMP command.
+    """
+    return [
+        {"ip": ip, "mac": leases[ip]}
+        for ip in leases
+        ]
+
+
+class PeriodicLeaseUploadService(TimerService, object):
+    """Twisted service to periodically upload DHCP leases to the region.
+
+    :param client_service: A `ClusterClientService` instance for talking
+        to the region controller.
+    :param reactor: An `IReactor` instance.
+    """
+
+    check_interval = 60  # In seconds.
+
+    def __init__(self, client_service, reactor, cluster_uuid):
+        # Call self.maybe_start_upload() every self.check_interval.
+        super(PeriodicLeaseUploadService, self).__init__(
+            self.check_interval, self.try_upload)
+        self.clock = reactor
+        self.client_service = client_service
+        self.uuid = cluster_uuid
+        maaslog.info("PeriodicLeaseUploadService starting.")
+
+    def try_upload(self):
+        """Wrap upload attempts in something that catches Failures.
+
+        Log the full error to the Twisted log, and a concise error to
+        the maas log.
+        """
+        def upload_failure(failure):
+            log.err(failure)
+            maaslog.error(
+                "Failed to upload leases: %s", failure.getErrorMessage())
+
+        return self.maybe_start_upload().addErrback(upload_failure)
+
+    @inlineCallbacks
+    def maybe_start_upload(self):
+        """Initiate a new upload if one is not already in progress."""
+        # Use a DeferredLock to prevent simultaneous uploads.
+        if service_lock.locked:
+            # Don't want to block on lock release.
+            return
+        yield service_lock.acquire()
+        try:
+            yield self._get_client_and_start_upload()
+        finally:
+            service_lock.release()
+
+    @inlineCallbacks
+    def _get_client_and_start_upload(self):
+        # Retry a few times, since this service usually comes up before
+        # the RPC service.
+        for elapsed, remaining, wait in retries(15, 5, self.clock):
+            try:
+                client = self.client_service.getClient()
+                break
+            except NoConnectionsAvailable:
+                yield pause(wait, clock=self.clock)
+        else:
+            maaslog.error(
+                "Failed to connect to region controller, cannot upload leases")
+            return
+        yield self._start_upload(client)
+
+    @inlineCallbacks
+    def _start_upload(self, client):
+        maaslog.info("Scanning DHCP leases...")
+        updated_lease_info = yield deferToThread(check_lease_changes)
+        if updated_lease_info is None:
+            maaslog.info("No leases changed since last scan")
+        else:
+            timestamp, leases = updated_lease_info
+            record_lease_state(timestamp, leases)
+            mappings = convert_leases_to_mappings(leases)
+            maaslog.info("Uploading DHCP leases to region controller.")
+            yield client(
+                UpdateLeases, uuid=self.uuid, mappings=mappings)
+            maaslog.info("Lease upload complete.")

=== added file 'src/provisioningserver/pserv_services/tests/test_lease_upload_service.py'
--- src/provisioningserver/pserv_services/tests/test_lease_upload_service.py	1970-01-01 00:00:00 +0000
+++ src/provisioningserver/pserv_services/tests/test_lease_upload_service.py	2014-08-28 03:19:24 +0000
@@ -0,0 +1,235 @@
+# Copyright 2014 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Tests for src/provisioningserver/pserv_services/lease_upload_service.py"""
+
+from __future__ import (
+    absolute_import,
+    print_function,
+    unicode_literals,
+    )
+
+str = None
+
+__metaclass__ = type
+__all__ = []
+
+from datetime import datetime
+
+from fixtures import FakeLogger
+from maastesting.factory import factory
+from maastesting.matchers import (
+    MockCalledOnceWith,
+    MockCallsMatch,
+    MockNotCalled,
+    )
+from mock import (
+    ANY,
+    call,
+    Mock,
+    sentinel,
+    )
+from provisioningserver import services
+from provisioningserver.dhcp.leases import check_lease_changes
+from provisioningserver.pserv_services import lease_upload_service
+from provisioningserver.pserv_services.lease_upload_service import (
+    convert_leases_to_mappings,
+    convert_mappings_to_leases,
+    PeriodicLeaseUploadService,
+    service_lock,
+    )
+from provisioningserver.rpc.exceptions import NoConnectionsAvailable
+from provisioningserver.rpc.region import UpdateLeases
+from provisioningserver.rpc.testing import (
+    MockClusterToRegionRPCFixture,
+    TwistedLoggerFixture,
+    )
+from provisioningserver.testing.testcase import PservTestCase
+from provisioningserver.utils.twisted import pause
+from testtools.deferredruntest import extract_result
+from twisted.application.internet import TimerService
+from twisted.internet import defer
+from twisted.internet.task import Clock
+
+
+def make_random_lease():
+    ip = factory.getRandomIPAddress()
+    mac = factory.getRandomMACAddress()
+    return {ip: mac}
+
+
+def make_random_mapping():
+    ip = factory.getRandomIPAddress()
+    mac = factory.getRandomMACAddress()
+    mapping = {"ip": ip, "mac": mac}
+    return mapping
+
+
+class TestHelperFunctions(PservTestCase):
+
+    def test_convert_leases_to_mappings_maps_correctly(self):
+        mappings = list()
+        for _ in xrange(3):
+            mappings.append(make_random_mapping())
+
+        # Convert to leases.
+        leases = convert_mappings_to_leases(mappings)
+        # Convert back and test against our original mappings.
+        observed = convert_leases_to_mappings(leases)
+        self.assertItemsEqual(mappings, observed)
+
+    def test_convert_leases_to_mappings_converts_correctly(self):
+        leases = dict()
+        for _ in xrange(3):
+            leases.update(make_random_lease())
+
+        # Convert to mappings.
+        mappings = convert_leases_to_mappings(leases)
+        # Convert back and test against our original leases.
+        observed = convert_mappings_to_leases(mappings)
+        self.assertEqual(observed, leases)
+
+
+class TestPeriodicImageDownloadService(PservTestCase):
+
+    def test_init(self):
+        service = PeriodicLeaseUploadService(
+            sentinel.service, sentinel.clock, sentinel.uuid)
+        self.assertIsInstance(service, TimerService)
+        self.assertIs(service.clock, sentinel.clock)
+        self.assertIs(service.uuid, sentinel.uuid)
+        self.assertIs(service.client_service, sentinel.service)
+
+    def patch_upload(self, service, return_value=None):
+        patched = self.patch(service, '_get_client_and_start_upload')
+        patched.return_value = defer.succeed(return_value)
+        return patched
+
+    def test_is_called_every_interval(self):
+        clock = Clock()
+        service = PeriodicLeaseUploadService(
+            sentinel.service, clock, sentinel.uuid)
+        # Avoid actual uploads:
+        start_upload = self.patch_upload(service)
+
+        # There are no calls before the service is started.
+        self.assertThat(start_upload, MockNotCalled())
+
+        service.startService()
+
+        # The first call is issued at startup.
+        self.assertThat(start_upload, MockCalledOnceWith())
+
+        # Wind clock forward one second less than the desired interval.
+        clock.advance(service.check_interval - 1)
+        # No more periodic calls made.
+        self.assertThat(start_upload, MockCalledOnceWith())
+
+        # Wind clock forward one second, past the interval.
+        clock.advance(1)
+
+        # Now there were two calls.
+        self.assertThat(start_upload, MockCallsMatch(call(), call()))
+
+        # Forward another interval, should be three calls.
+        clock.advance(service.check_interval)
+        self.assertThat(
+            start_upload, MockCallsMatch(call(), call(), call()))
+
+    @defer.inlineCallbacks
+    def test_does_not_run_if_lock_taken(self):
+        service = PeriodicLeaseUploadService(
+            sentinel.rpc, Clock(), sentinel.uuid)
+        start_upload = self.patch_upload(service)
+        yield service_lock.acquire()
+        self.addCleanup(service_lock.release)
+        service.startService()
+        self.assertThat(start_upload, MockNotCalled())
+
+    def test_takes_lock_when_running(self):
+        clock = Clock()
+        service = PeriodicLeaseUploadService(
+            sentinel.rpc, clock, sentinel.uuid)
+
+        # Patch the upload func so it's just a Deferred that waits for
+        # one second.
+        _start_upload = self.patch(service, '_get_client_and_start_upload')
+        _start_upload.return_value = pause(1, clock)
+
+        # Lock is acquired for the first download after startup.
+        self.assertFalse(service_lock.locked)
+        service.startService()
+        self.assertTrue(service_lock.locked)
+
+        # Lock is released once the download is done.
+        clock.advance(1)
+        self.assertFalse(service_lock.locked)
+
+    def test_no_upload_if_no_rpc_connections(self):
+        rpc_client = Mock()
+        rpc_client.getClient.side_effect = NoConnectionsAvailable()
+
+        clock = Clock()
+        service = PeriodicLeaseUploadService(
+            rpc_client, clock, sentinel.uuid)
+        start_upload = self.patch(service, '_start_upload')
+        service.startService()
+        # Wind clock past all the retries.  You can't do this in one big
+        # lump, it seems.  The test looks like it passes, but the
+        # maybe_start_upload() method never returns properly which
+        # leaves the service_lock locked.
+        clock.pump((5, 5, 5))
+        self.assertThat(start_upload, MockNotCalled())
+
+    def test_upload_is_initiated(self):
+        # We're pretending to be the reactor in this thread. To ensure correct
+        # operation from things like the @asynchronous decorators we need to
+        # register as the IO thread.
+        self.register_as_io_thread()
+
+        # Create a fixture for the region side of the RPC.
+        rpc_fixture = self.useFixture(MockClusterToRegionRPCFixture())
+        rpc_service = services.getServiceNamed('rpc')
+        server, io = rpc_fixture.makeEventLoop(UpdateLeases)
+        server.UpdateLeases.return_value = defer.succeed({})
+
+        # Create a mock response to "check_lease_changes()"
+        fake_lease = make_random_lease()
+        deferToThread = self.patch(lease_upload_service, 'deferToThread')
+        deferToThread.return_value = defer.succeed(
+            (datetime.now(), fake_lease),)
+        mappings = convert_leases_to_mappings(fake_lease)
+
+        # Start the service.
+        uuid = factory.make_UUID()
+        service = PeriodicLeaseUploadService(rpc_service, Clock(), uuid)
+        service.startService()
+
+        # Gavin says that I need to pump my IO.  I don't know what this
+        # means but it sounds important!
+        io.pump()
+
+        # Ensure it called out to a new thread to get and parse the leases.
+        self.assertThat(deferToThread, MockCalledOnceWith(check_lease_changes))
+
+        # Ensure it sent them to the region using RPC.
+        self.assertThat(
+            server.UpdateLeases,
+            MockCalledOnceWith(ANY, uuid=uuid, mappings=mappings))
+
+    def test_logs_other_errors(self):
+        service = PeriodicLeaseUploadService(
+            sentinel.rpc, Clock(), sentinel.uuid)
+
+        maybe_start_upload = self.patch(service, "maybe_start_upload")
+        maybe_start_upload.return_value = defer.fail(
+            ZeroDivisionError("Such a shame I can't divide by zero"))
+
+        with FakeLogger("maas") as maaslog, TwistedLoggerFixture():
+            d = service.try_upload()
+
+        self.assertEqual(None, extract_result(d))
+        self.assertDocTestMatches(
+            "Failed to upload leases: "
+            "Such a shame I can't divide by zero",
+            maaslog.output)

=== modified file 'src/provisioningserver/rpc/region.py'
--- src/provisioningserver/rpc/region.py	2014-08-25 17:19:56 +0000
+++ src/provisioningserver/rpc/region.py	2014-08-28 03:19:24 +0000
@@ -104,6 +104,22 @@
     errors = []
 
 
+class UpdateLeases(amp.Command):
+    """Report DHCP leases on the invoking cluster controller.
+
+    :since: 1.7
+    """
+    arguments = [
+        # The cluster UUID.
+        (b"uuid", amp.Unicode()),
+        (b"mappings", amp.AmpList(
+            [(b"ip", amp.Unicode()),
+             (b"mac", amp.Unicode())]))
+    ]
+    response = []
+    errors = []
+
+
 class GetProxies(amp.Command):
     """Return the HTTP and HTTPS proxies to use.
 

=== modified file 'src/provisioningserver/tasks.py'
--- src/provisioningserver/tasks.py	2014-08-26 23:55:04 +0000
+++ src/provisioningserver/tasks.py	2014-08-28 03:19:24 +0000
@@ -51,7 +51,6 @@
     restart_dhcpv4,
     stop_dhcpv4,
     )
-from provisioningserver.dhcp.leases import upload_leases
 from provisioningserver.dns.config import (
     DNSConfig,
     execute_rndc_command,
@@ -321,18 +320,6 @@
 
 
 @task
-@log_task_events(level=logging.DEBUG)
-@log_exception_text
-def upload_dhcp_leases():
-    """Upload DHCP leases.
-
-    Uploads leases to the MAAS API, using cached credentials -- the task
-    originates with celerybeat, not with a server request.
-    """
-    upload_leases()
-
-
-@task
 @log_task_events()
 @log_exception_text
 def add_new_dhcp_host_map(mappings, server_address, shared_key):

=== modified file 'src/provisioningserver/testing/testcase.py'
--- src/provisioningserver/testing/testcase.py	2013-10-07 09:12:40 +0000
+++ src/provisioningserver/testing/testcase.py	2014-08-28 03:19:24 +0000
@@ -25,6 +25,8 @@
     record_nodegroup_uuid,
     )
 from provisioningserver.testing.worker_cache import WorkerCacheFixture
+from twisted.internet import reactor
+from twisted.python import threadable
 
 
 class PservTestCase(testcase.MAASTestCase):
@@ -52,3 +54,19 @@
         self.set_maas_url()
         self.set_api_credentials()
         self.set_node_group_uuid()
+
+    def register_as_io_thread(self):
+        """Make the current thread the IO thread.
+
+        When pretending to be the reactor, by using clocks and suchlike,
+        register the current thread as the reactor thread, a.k.a. the IO
+        thread, to ensure correct operation from things like the `synchronous`
+        and `asynchronous` decorators.
+
+        Do not use this when the reactor is running.
+        """
+        self.assertFalse(
+            reactor.running, "Do not use this to change the IO thread "
+            "while the reactor is running.")
+        self.addCleanup(setattr, threadable, "ioThread", threadable.ioThread)
+        threadable.ioThread = threadable.getThreadID()

=== modified file 'src/provisioningserver/tests/test_plugin.py'
--- src/provisioningserver/tests/test_plugin.py	2014-08-27 03:38:28 +0000
+++ src/provisioningserver/tests/test_plugin.py	2014-08-28 03:19:24 +0000
@@ -114,7 +114,8 @@
         service = service_maker.makeService(options)
         self.assertIsInstance(service, MultiService)
         self.assertSequenceEqual(
-            ["image_download", "log", "node_monitor", "oops", "rpc", "tftp"],
+            ["image_download", "lease_upload", "log", "node_monitor", "oops",
+             "rpc", "tftp"],
             sorted(service.namedServices))
         self.assertEqual(
             len(service.namedServices), len(service.services),

=== modified file 'src/provisioningserver/tests/test_tasks.py'
--- src/provisioningserver/tests/test_tasks.py	2014-08-25 03:14:30 +0000
+++ src/provisioningserver/tests/test_tasks.py	2014-08-28 03:19:24 +0000
@@ -14,7 +14,6 @@
 __metaclass__ = type
 __all__ = []
 
-from datetime import datetime
 import json
 import os
 import random
@@ -50,10 +49,7 @@
     tasks,
     )
 from provisioningserver.boot import tftppath
-from provisioningserver.dhcp import (
-    config,
-    leases,
-    )
+from provisioningserver.dhcp import config
 from provisioningserver.dhcp.testing.config import make_subnet_config
 from provisioningserver.dns.config import (
     celery_conf,
@@ -195,14 +191,6 @@
             'dhcp_subnets': [make_subnet_config()],
             }
 
-    def test_upload_dhcp_leases(self):
-        self.patch(
-            leases, 'parse_leases_file',
-            Mock(return_value=(datetime.utcnow(), {})))
-        self.patch(leases, 'process_leases', Mock())
-        tasks.upload_dhcp_leases.delay()
-        self.assertEqual(1, leases.process_leases.call_count)
-
     def test_add_new_dhcp_host_map(self):
         # We don't want to actually run omshell in the task, so we stub
         # out the wrapper class's _run method and record what it would