launchpad-reviewers team mailing list archive

["Jeroen T. Vermeulen" <jtv@xxxxxxxxxxxxx>]

Jeroen T. Vermeulen has proposed merging lp:~jtv/maas/record-worker-shared-secrets into lp:maas.

Requested reviews:
  Launchpad code reviewers (launchpad-reviewers)

For more details, see:
https://code.launchpad.net/~jtv/maas/record-worker-shared-secrets/+merge/118667

Sometimes a worker needs to act independently without a server request.  But it gets API credentials and such from server requests.  We need to store the stuff in worker-side variables, but we'd prefer not to write such sensitive information to the worker filesystem.

This branch provides a generic way of maintaining these items.  I hope the docstring on “refresh” explains it well enough that there's no point repeating it here.  We could now, if we wanted, add an API call for workers to say “hey I just started up, can you send me the credentials etc. that I need please?”  It could be synchronous (get results back from API call, call refresh() directly) or asynchronous (ask the server to send a refresh task).  Having it be a celery task leaves that choice wide open.

One detail worth noting is that in the tests, I don't bother making every call to a task through celery.  Instead I call them directly (this is one of celery's documented selling points, not a hack based on any implementation knowledge) and treat the ability to call the task through celery as an integration matter that I test separately.  I think this keeps the tests more to the point as there's no more need to deal with delay() and the results it returns.  Test failures related to celery integration also become much more informative this way: the it's-a-task test fails but others can still pass — I know because I forgot to add the celery resource to my test initially!


Jeroen
-- 
https://code.launchpad.net/~jtv/maas/record-worker-shared-secrets/+merge/118667
Your team Launchpad code reviewers is requested to review the proposed merge of lp:~jtv/maas/record-worker-shared-secrets into lp:maas.

=== modified file 'src/provisioningserver/dhcp/leases.py'
--- src/provisioningserver/dhcp/leases.py	2012-07-09 14:10:57 +0000
+++ src/provisioningserver/dhcp/leases.py	2012-08-08 04:02:22 +0000
@@ -46,6 +46,18 @@
 # Leases as last parsed.
 recorded_leases = None
 
+# Shared key for use with omshell.  We don't store this key
+# persistently, but when the server sends it, we keep a copy in memory
+# so that celerybeat jobs (which do not originate with the server and
+# therefore don't receive this argument) can make use of it.
+recorded_omapi_shared_key = None
+
+
+def record_omapi_shared_key(shared_key):
+    """Record the OMAPI shared key as received from the server."""
+    global recorded_omapi_shared_key
+    recorded_omapi_shared_key = shared_key
+
 
 def get_leases_timestamp():
     """Return the last modification timestamp of the DHCP leases file."""

=== modified file 'src/provisioningserver/dhcp/tests/test_leases.py'
--- src/provisioningserver/dhcp/tests/test_leases.py	2012-07-09 12:31:50 +0000
+++ src/provisioningserver/dhcp/tests/test_leases.py	2012-08-08 04:02:22 +0000
@@ -30,11 +30,20 @@
     check_lease_changes,
     parse_leases_file,
     record_lease_state,
+    record_omapi_shared_key,
     update_leases,
     upload_leases,
     )
 
 
+class TestHelpers(TestCase):
+
+    def test_record_omapi_shared_key_records_shared_key(self):
+        key = factory.getRandomString()
+        record_omapi_shared_key(key)
+        self.assertEqual(key, leases_module.recorded_omapi_shared_key)
+
+
 class StopExecuting(BaseException):
     """Exception class to stop execution at a desired point.
 

=== modified file 'src/provisioningserver/tasks.py'
--- src/provisioningserver/tasks.py	2012-08-08 02:31:31 +0000
+++ src/provisioningserver/tasks.py	2012-08-08 04:02:22 +0000
@@ -30,6 +30,7 @@
 from celery.task import task
 from celeryconfig import DHCP_CONFIG_FILE
 from provisioningserver.dhcp import config
+from provisioningserver.dhcp.leases import record_omapi_shared_key
 from provisioningserver.dns.config import (
     DNSConfig,
     execute_rndc_command,
@@ -42,6 +43,49 @@
     )
 from provisioningserver.utils import atomic_write
 
+# For each item passed to refresh(), a refresh function to give it to.
+refresh_functions = {
+    'omapi_shared_key': record_omapi_shared_key,
+}
+
+
+@task
+def refresh(**kwargs):
+    """Update the worker's knowledge of various secrets it needs.
+
+    The worker shares some secrets with the MAAS server, such as its
+    omapi key for talking to the DHCP server, and its MAAS API credentials.
+    When the server sends tasks to the worker, the tasks will include these
+    secrets as needed.  But not everything the worker does is initiated by
+    a server request, so it needs copies of these secrets at hand.
+
+    We don't store these secrets in the worker, but we hold copies in
+    memory.  The worker won't perform jobs that require secrets it does
+    not have yet, waiting instead for the next chance to catch up.
+
+    To make sure that the worker does not have to wait too long, the server
+    can send periodic `refresh` messages with the required information.
+    Tasks can also call `refresh` to record information they receive from
+    the server.
+
+    All refreshed items are passed as keyword arguments, to avoid confusion
+    and allow for easy reordering.  All refreshed items are optional.  An
+    item that is not passed will not be refreshed, so it's entirely valid to
+    call this for just a single item.  However `None` is a value like any
+    other, so passing `foo=None` will cause item `foo` to be refreshed with
+    value `None`.
+
+    To help catch simple programming mistakes, passing an unknown argument
+    will result in an assertion failure.
+
+    :param omapi_shared_key: Shared key for working with the worker's
+        DHCP server.
+    """
+    for key, value in kwargs.items():
+        assert key in refresh_functions, "Unknown refresh item: %s" % key
+        refresh_functions[key](value)
+
+
 # =====================================================================
 # Power-related tasks
 # =====================================================================
@@ -182,6 +226,7 @@
         control.
     """
     omshell = Omshell(server_address, shared_key)
+    refresh(omapi_shared_key=shared_key)
     try:
         for ip_address, mac_address in mappings.items():
             omshell.create(ip_address, mac_address)
@@ -203,6 +248,7 @@
         control.
     """
     omshell = Omshell(server_address, shared_key)
+    refresh(omapi_shared_key=shared_key)
     try:
         omshell.remove(ip_address)
     except CalledProcessError:

=== modified file 'src/provisioningserver/tests/test_tasks.py'
--- src/provisioningserver/tests/test_tasks.py	2012-08-07 10:57:35 +0000
+++ src/provisioningserver/tests/test_tasks.py	2012-08-08 04:02:22 +0000
@@ -23,6 +23,7 @@
 from maastesting.testcase import TestCase
 from netaddr import IPNetwork
 from provisioningserver import tasks
+from provisioningserver.dhcp import leases
 from provisioningserver.dns.config import (
     conf,
     DNSZoneConfig,
@@ -37,6 +38,7 @@
     Omshell,
     power_off,
     power_on,
+    refresh,
     remove_dhcp_host_map,
     restart_dhcp_server,
     rndc_command,
@@ -60,6 +62,50 @@
 arbitrary_mac = "AA:BB:CC:DD:EE:FF"
 
 
+class TestRefresh(TestCase):
+    """Tests for the `refresh` task."""
+
+    resources = (
+        ("celery", FixtureResource(CeleryFixture())),
+        )
+
+    def test_does_not_require_arguments(self):
+        refresh()
+        # Nothing is refreshed, but there is no error either.
+        pass
+
+    def test_calls_refresh_function(self):
+        value = factory.make_name('new-value')
+        refresh_function = FakeMethod()
+        self.patch(tasks, 'refresh_functions', {'my_item': refresh_function})
+        refresh(my_item=value)
+        self.assertEqual([(value, )], refresh_function.extract_args())
+
+    def test_refreshes_even_if_None(self):
+        refresh_function = FakeMethod()
+        self.patch(tasks, 'refresh_functions', {'my_item': refresh_function})
+        refresh(my_item=None)
+        self.assertEqual([(None, )], refresh_function.extract_args())
+
+    def test_does_not_refresh_if_omitted(self):
+        refresh_function = FakeMethod()
+        self.patch(tasks, 'refresh_functions', {'my_item': refresh_function})
+        refresh()
+        self.assertEqual([], refresh_function.extract_args())
+
+    def test_breaks_on_unknown_item(self):
+        self.assertRaises(AssertionError, refresh, not_an_item=None)
+
+    def test_updates_omapi_shared_key(self):
+        self.patch(leases, 'recorded_omapi_shared_key', None)
+        key = factory.make_name('omapi-shared-key')
+        refresh(omapi_shared_key=key)
+        self.assertEqual(key, leases.recorded_omapi_shared_key)
+
+    def test_is_a_task(self):
+        self.assertTrue(refresh.delay().successful())
+
+
 class TestPowerTasks(TestCase):
 
     resources = (
@@ -124,6 +170,12 @@
             CalledProcessError, add_new_dhcp_host_map.delay,
             {mac: ip}, server_address, key)
 
+    def test_add_new_dhcp_host_map_records_shared_key(self):
+        key = factory.getRandomString()
+        self.patch(Omshell, '_run', FakeMethod())
+        add_new_dhcp_host_map({}, factory.make_name('server'), key)
+        self.assertEqual(key, leases.recorded_omapi_shared_key)
+
     def test_remove_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
@@ -148,6 +200,13 @@
             CalledProcessError, remove_dhcp_host_map.delay,
             ip, server_address, key)
 
+    def test_remove_dhcp_host_map_records_shared_key(self):
+        key = factory.getRandomString()
+        self.patch(Omshell, '_run', FakeMethod((0, "obj: <null>")))
+        remove_dhcp_host_map(
+            factory.getRandomIPAddress(), factory.make_name('server'), key)
+        self.assertEqual(key, leases.recorded_omapi_shared_key)
+
     def test_write_dhcp_config_writes_config(self):
         conf_file = self.make_file(contents=factory.getRandomString())
         self.patch(tasks, 'DHCP_CONFIG_FILE', conf_file)