Merge lp:~blake-rouse/maas/queue-status-messages into lp:~maas-committers/maas/trunk

[Blake Rouse]

Status message from curtin and cloud-init are now handled by twisted. Each message is placed in a queue to be processed every second. This helps with the load on the regiond processes by dropping load by more than 50%. This branch also allows a Twisted API endpoint to overlay a Django endpoint.

[Andres Rodriguez (andreserl)]

Overall this looks good. Just a few comments inline.

review: Approve

[MAAS Lander (maas-lander)]

The attempt to merge lp:~blake-rouse/maas/queue-status-messages into lp:maas failed. Below is the output from the failed tests.

Hit:1 http://prodstack-zone-2.clouds.archive.ubuntu.com/ubuntu xenial InRelease
Get:2 http://prodstack-zone-2.clouds.archive.ubuntu.com/ubuntu xenial-updates InRelease [102 kB]
Get:3 http://prodstack-zone-2.clouds.archive.ubuntu.com/ubuntu xenial-backports InRelease [102 kB]
Get:4 http://security.ubuntu.com/ubuntu xenial-security InRelease [102 kB]
Get:5 http://prodstack-zone-2.clouds.archive.ubuntu.com/ubuntu xenial-updates/main Sources [233 kB]
Get:6 http://prodstack-zone-2.clouds.archive.ubuntu.com/ubuntu xenial-updates/main amd64 Packages [483 kB]
Get:7 http://prodstack-zone-2.clouds.archive.ubuntu.com/ubuntu xenial-updates/universe amd64 Packages [411 kB]
Get:8 http://security.ubuntu.com/ubuntu xenial-security/main Sources [61.1 kB]
Get:9 http://security.ubuntu.com/ubuntu xenial-security/main amd64 Packages [219 kB]
Get:10 http://security.ubuntu.com/ubuntu xenial-security/universe amd64 Packages [79.1 kB]
Fetched 1,793 kB in 0s (2,724 kB/s)
Reading package lists...
sudo DEBIAN_FRONTEND=noninteractive apt-get -y \
    --no-install-recommends install apache2 archdetect-deb authbind avahi-utils bash bind9 bind9utils build-essential bzr bzr-builddeb chromium-browser chromium-chromedriver curl daemontools debhelper dh-apport dh-systemd distro-info dnsutils firefox freeipmi-tools git gjs ipython isc-dhcp-common isc-dhcp-server libjs-angularjs libjs-jquery libjs-jquery-hotkeys libjs-yui3-full libjs-yui3-min libnss-wrapper libpq-dev make nodejs-legacy npm postgresql psmisc pxelinux python3-all python3-apt python3-attr python3-bson python3-convoy python3-crochet python3-cssselect python3-curtin python3-dev python3-distro-info python3-django python3-django-nose python3-django-piston3 python3-dnspython python3-docutils python3-formencode python3-hivex python3-httplib2 python3-jinja2 python3-jsonschema python3-lxml python3-netaddr python3-netifaces python3-novaclient python3-oauth python3-oauthlib python3-openssl python3-paramiko python3-petname python3-pexpect python3-psycopg2 python3-pyinotify python3-pyparsing python3-pyvmomi python3-requests python3-seamicroclient python3-setuptools python3-simplestreams python3-sphinx python3-tempita python3-twisted python3-txtftp python3-tz python3-yaml python3-zope.interface python-bson python-crochet python-django python-django-piston python-djorm-ext-pgarray python-formencode python-lxml python-netaddr python-netifaces python-pocket-lint python-psycopg2 python-simplejson python-tempita python-twisted python-yaml socat syslinux-common tgt ubuntu-cloudimage-keyring wget xvfb
Reading package lists...
Building dependency tree...
Reading state information...
authbind is already the newest version (2.1.1+nmu1).
avahi-utils is already the newest version (0.6.32~rc+dfsg-1ubuntu2).
build-essential is already the newest version (12.1ubuntu2).
debhelper is already the newest version (9.20160115ubuntu3).
distro-info is already the newest version (0.14build1).
git is already the newest version (1:2.7.4-0ubuntu1).
libjs-angularjs is already the newest version (1.2.28-1ubuntu2).
libjs-jquery is ...

=== modified file 'src/maasserver/eventloop.py'
--- src/maasserver/eventloop.py	2016-12-05 09:38:41 +0000
+++ src/maasserver/eventloop.py	2017-02-28 20:14:47 +0000
@@ -120,6 +120,11 @@
     return RackControllerService(postgresListener, advertisingService)
 
 
+def make_StatusWorkerService(dbtasks):
+    from metadataserver.api_twisted import StatusWorkerService
+    return StatusWorkerService(dbtasks)
+
+
 def make_ServiceMonitorService(advertisingService):
     from maasserver.regiondservices import service_monitor_service
     return service_monitor_service.ServiceMonitorService(advertisingService)
@@ -151,7 +156,7 @@
     return ntp.RegionNetworkTimeProtocolService(reactor)
 
 
-def make_WebApplicationService(postgresListener):
+def make_WebApplicationService(postgresListener, statusWorker):
     from maasserver.webapp import WebApplicationService
     site_port = DEFAULT_PORT  # config["port"]
     # Make a socket with SO_REUSEPORT set so that we can run multiple web
@@ -169,7 +174,8 @@
     site_endpoint = AdoptedStreamServerEndpoint(reactor, s.fileno(), s.family)
     site_endpoint.port = site_port  # Make it easy to get the port number.
     site_endpoint.socket = s  # Prevent garbage collection.
-    site_service = WebApplicationService(site_endpoint, postgresListener)
+    site_service = WebApplicationService(
+        site_endpoint, postgresListener, statusWorker)
     return site_service
 
 
@@ -245,13 +251,18 @@
         "web": {
             "only_on_master": False,
             "factory": make_WebApplicationService,
-            "requires": ["postgres-listener"],
+            "requires": ["postgres-listener", "status-worker"],
         },
         "service-monitor": {
             "only_on_master": True,
             "factory": make_ServiceMonitorService,
             "requires": ["rpc-advertise"],
         },
+        "status-worker": {
+            "only_on_master": False,
+            "factory": make_StatusWorkerService,
+            "requires": ["database-tasks"],
+        },
         "networks-monitor": {
             "only_on_master": False,
             "factory": make_NetworksMonitoringService,

=== modified file 'src/maasserver/models/event.py'
--- src/maasserver/models/event.py	2016-12-07 12:46:14 +0000
+++ src/maasserver/models/event.py	2017-02-28 20:14:47 +0000
@@ -32,7 +32,8 @@
 
     def register_event_and_event_type(
             self, system_id, type_name, type_description='',
-            type_level=logging.INFO, event_action='', event_description=''):
+            type_level=logging.INFO, event_action='', event_description='',
+            created=None):
         """Register EventType if it does not exist, then register the Event."""
         node = Node.objects.get(system_id=system_id)
         try:
@@ -42,9 +43,9 @@
             # We didn't find it so register it.
             event_type = EventType.objects.register(
                 type_name, type_description, type_level)
-        Event.objects.create(
+        return Event.objects.create(
             node=node, type=event_type, action=event_action,
-            description=event_description)
+            description=event_description, created=created)
 
     def create_node_event(
             self, system_id, event_type, event_action='',

=== modified file 'src/maasserver/models/node.py'
--- src/maasserver/models/node.py	2017-02-17 21:27:46 +0000
+++ src/maasserver/models/node.py	2017-02-28 20:14:47 +0000
@@ -1037,7 +1037,7 @@
 
     def __str__(self):
         if self.hostname:
-            return "%s (%s)" % (self.system_id, self.fqdn)
+            return "%s (%s)" % (self.system_id, self.hostname)
         else:
             return self.system_id
 

=== modified file 'src/maasserver/models/tests/test_event.py'
--- src/maasserver/models/tests/test_event.py	2016-12-07 12:46:14 +0000
+++ src/maasserver/models/tests/test_event.py	2017-02-28 20:14:47 +0000
@@ -33,6 +33,16 @@
             system_id=node.system_id, type_name=event_type.name)
         self.assertIsNotNone(Event.objects.get(node=node))
 
+    def test_register_event_and_event_type_registers_event_with_datetime(self):
+        # EvenType exists
+        node = factory.make_Node()
+        event_type = factory.make_EventType()
+        created = factory.make_date()
+        event = Event.objects.register_event_and_event_type(
+            system_id=node.system_id, type_name=event_type.name,
+            created=created)
+        self.assertEqual(created, event.created)
+
     def test_register_event_and_event_type_registers_event_for_new_type(self):
         # EventType does not exist
         node = factory.make_Node()

=== modified file 'src/maasserver/preseed.py'
--- src/maasserver/preseed.py	2017-02-22 14:55:09 +0000
+++ src/maasserver/preseed.py	2017-02-28 20:14:47 +0000
@@ -191,6 +191,7 @@
                     'consumer_key': token.consumer.key,
                     'token_key': token.key,
                     'token_secret': token.secret,
+                    'level': 'INFO',
                 },
             },
             'install': {

=== modified file 'src/maasserver/tests/test_eventloop.py'
--- src/maasserver/tests/test_eventloop.py	2017-01-28 00:51:47 +0000
+++ src/maasserver/tests/test_eventloop.py	2017-02-28 20:14:47 +0000
@@ -32,6 +32,7 @@
 )
 from maastesting.factory import factory
 from maastesting.testcase import MAASTestCase
+from metadataserver import api_twisted
 from testtools.matchers import (
     Equals,
     IsInstance,
@@ -254,7 +255,7 @@
 
     def test_make_WebApplicationService(self):
         service = eventloop.make_WebApplicationService(
-            FakePostgresListenerService())
+            FakePostgresListenerService(), sentinel.status_worker)
         self.assertThat(service, IsInstance(webapp.WebApplicationService))
         # The endpoint is set to port 5243 on localhost.
         self.assertThat(service.endpoint, MatchesStructure.byEquality(
@@ -271,7 +272,7 @@
             eventloop.loop.factories["web"]["factory"])
         # Has a dependency of postgres-listener.
         self.assertEquals(
-            ["postgres-listener"],
+            ["postgres-listener", "status-worker"],
             eventloop.loop.factories["web"]["requires"])
         self.assertFalse(
             eventloop.loop.factories["web"]["only_on_master"])
@@ -308,6 +309,22 @@
         self.assertTrue(
             eventloop.loop.factories["service-monitor"]["only_on_master"])
 
+    def test_make_StatusWorkerService(self):
+        service = eventloop.make_StatusWorkerService(
+            sentinel.dbtasks)
+        self.assertThat(service, IsInstance(
+            api_twisted.StatusWorkerService))
+        # It is registered as a factory in RegionEventLoop.
+        self.assertIs(
+            eventloop.make_StatusWorkerService,
+            eventloop.loop.factories["status-worker"]["factory"])
+        # Has a dependency of database-tasks.
+        self.assertEquals(
+            ["database-tasks"],
+            eventloop.loop.factories["status-worker"]["requires"])
+        self.assertFalse(
+            eventloop.loop.factories["status-worker"]["only_on_master"])
+
 
 class TestDisablingDatabaseConnections(MAASServerTestCase):
 

=== modified file 'src/maasserver/tests/test_plugin.py'
--- src/maasserver/tests/test_plugin.py	2017-01-28 00:51:47 +0000
+++ src/maasserver/tests/test_plugin.py	2017-02-28 20:14:47 +0000
@@ -108,6 +108,7 @@
             "rpc-advertise",
             "service-monitor",
             "status-monitor",
+            "status-worker",
             "web",
         ]
         self.assertItemsEqual(expected_services, service.namedServices.keys())

=== modified file 'src/maasserver/tests/test_preseed.py'
--- src/maasserver/tests/test_preseed.py	2017-02-24 15:17:17 +0000
+++ src/maasserver/tests/test_preseed.py	2017-02-28 20:14:47 +0000
@@ -646,6 +646,9 @@
         self.assertEqual(
             token.secret,
             reporter['reporting']['maas']['token_secret'])
+        self.assertEqual(
+            'INFO',
+            reporter['reporting']['maas']['level'])
 
     def test__curtin_maas_reporter_without_events_support(self):
         node = factory.make_Node_with_Interface_on_Subnet()

=== modified file 'src/maasserver/tests/test_webapp.py'
--- src/maasserver/tests/test_webapp.py	2017-01-28 00:51:47 +0000
+++ src/maasserver/tests/test_webapp.py	2017-02-28 20:14:47 +0000
@@ -17,6 +17,7 @@
     webapp,
 )
 from maasserver.testing.listener import FakePostgresListenerService
+from maasserver.webapp import OverlaySite
 from maasserver.websockets.protocol import WebSocketFactory
 from maastesting.factory import factory
 from maastesting.matchers import MockCalledOnceWith
@@ -27,10 +28,15 @@
     Is,
     IsInstance,
     MatchesStructure,
+    Not,
 )
 from twisted.internet import reactor
 from twisted.internet.endpoints import TCP4ServerEndpoint
-from twisted.web.resource import Resource
+from twisted.web.error import UnsupportedMethod
+from twisted.web.resource import (
+    NoResource,
+    Resource,
+)
 from twisted.web.server import Site
 from twisted.web.test.requesthelper import (
     DummyChannel,
@@ -76,6 +82,70 @@
                 sentinel.command, single_path, sentinel.version))
 
 
+class TestOverlaySite(MAASTestCase):
+
+    def test__init__(self):
+        root = Resource()
+        site = OverlaySite(root)
+        self.assertThat(site, IsInstance(Site))
+
+    def test_getResourceFor_returns_no_resource_wo_underlay(self):
+        root = Resource()
+        site = OverlaySite(root)
+        request = DummyRequest([b'MAAS'])
+        resource = site.getResourceFor(request)
+        self.assertThat(resource, IsInstance(NoResource))
+
+    def test_getResourceFor_wraps_render_wo_underlay(self):
+        root = Resource()
+        maas = Resource()
+        mock_render = self.patch(maas, 'render')
+        root.putChild(b'MAAS', maas)
+        site = OverlaySite(root)
+        request = DummyRequest([b'MAAS'])
+        resource = site.getResourceFor(request)
+        self.assertThat(resource, Is(maas))
+        self.assertThat(resource.render, Not(Is(mock_render)))
+        resource.render(request)
+        self.assertThat(mock_render, MockCalledOnceWith(request))
+
+    def test_getResourceFor_wraps_render_wo_underlay_raises_no_method(self):
+        root = Resource()
+        maas = Resource()
+        root.putChild(b'MAAS', maas)
+        site = OverlaySite(root)
+        request = DummyRequest([b'MAAS'])
+        resource = site.getResourceFor(request)
+        self.assertThat(resource, Is(maas))
+        self.assertRaises(UnsupportedMethod, resource.render, request)
+
+    def test_getResourceFor_returns_resource_from_underlay(self):
+        underlay_root = Resource()
+        underlay_maas = Resource()
+        underlay_root.putChild(b'MAAS', underlay_maas)
+        overlay_root = Resource()
+        site = OverlaySite(overlay_root)
+        site.underlay = Site(underlay_root)
+        request = DummyRequest([b'MAAS'])
+        resource = site.getResourceFor(request)
+        self.assertThat(resource, Is(underlay_maas))
+
+    def test_getResourceFor_calls_render_on_underlay_when_no_method(self):
+        underlay_root = Resource()
+        underlay_maas = Resource()
+        mock_underlay_maas_render = self.patch(underlay_maas, 'render')
+        underlay_root.putChild(b'MAAS', underlay_maas)
+        overlay_root = Resource()
+        overlay_maas = Resource()
+        overlay_root.putChild(b'MAAS', overlay_maas)
+        site = OverlaySite(overlay_root)
+        site.underlay = Site(underlay_root)
+        request = DummyRequest([b'MAAS'])
+        resource = site.getResourceFor(request)
+        resource.render(request)
+        self.assertThat(mock_underlay_maas_render, MockCalledOnceWith(request))
+
+
 class TestResourceOverlay(MAASTestCase):
 
     def make_resourceoverlay(self):
@@ -100,7 +170,7 @@
         listener = FakePostgresListenerService()
         service_endpoint = self.make_endpoint()
         service = webapp.WebApplicationService(
-            service_endpoint, listener)
+            service_endpoint, listener, sentinel.status_worker)
         # Patch the getServiceNamed so the WebSocketFactory does not
         # error trying to register for events from the RPC service. In this
         # test the RPC service is not started.
@@ -197,11 +267,24 @@
 
         service.startService()
 
+        # Overlay
+        site = service.site
+        self.assertThat(site, IsInstance(OverlaySite))
         resource = service.site.resource
         self.assertThat(resource, IsInstance(Resource))
         overlay_resource = resource.getChildWithDefault(b"MAAS", request=None)
-        self.assertThat(overlay_resource, IsInstance(webapp.ResourceOverlay))
-        self.assertThat(overlay_resource.basis, MatchesStructure(
+        self.assertThat(overlay_resource, IsInstance(Resource))
+
+        # Underlay
+        site = service.site.underlay
+        self.assertThat(site, IsInstance(Site))
+        underlay_resource = site.resource
+        self.assertThat(underlay_resource, IsInstance(Resource))
+        underlay_maas_resource = underlay_resource.getChildWithDefault(
+            b"MAAS", request=None)
+        self.assertThat(
+            underlay_maas_resource, IsInstance(webapp.ResourceOverlay))
+        self.assertThat(underlay_maas_resource.basis, MatchesStructure(
             _reactor=Is(reactor), _threadpool=Is(service.threadpool),
             _application=IsInstance(WSGIHandler)))
 

=== modified file 'src/maasserver/webapp.py'
--- src/maasserver/webapp.py	2016-10-28 15:58:32 +0000
+++ src/maasserver/webapp.py	2017-02-28 20:14:47 +0000
@@ -1,4 +1,4 @@
-# Copyright 2014-2016 Canonical Ltd.  This software is licensed under the
+# Copyright 2014-2017 Canonical Ltd.  This software is licensed under the
 # GNU Affero General Public License version 3 (see the file LICENSE).
 
 """The MAAS Web Application."""
@@ -7,6 +7,8 @@
     "WebApplicationService",
 ]
 
+import copy
+from functools import partial
 from http.client import SERVICE_UNAVAILABLE
 import re
 
@@ -19,6 +21,7 @@
     lookupProtocolForFactory,
     WebSocketsResource,
 )
+from metadataserver.api_twisted import StatusHandlerResource
 from provisioningserver.logger import LegacyLogger
 from provisioningserver.utils.twisted import (
     asynchronous,
@@ -30,8 +33,10 @@
     reactor,
 )
 from twisted.python import failure
+from twisted.web.error import UnsupportedMethod
 from twisted.web.resource import (
     ErrorPage,
+    NoResource,
     Resource,
 )
 from twisted.web.server import (
@@ -81,6 +86,49 @@
             command, path, version)
 
 
+class OverlaySite(Site):
+    """A site that is over another site.
+
+    If this site cannot resolve a valid resource to handle the request, then
+    the underlay site gets passed the request to process.
+    """
+
+    underlay = None
+
+    def getResourceFor(self, request):
+        """Override to support an underlay site.
+
+        If this site cannot return a valid resource to request is passed to
+        the underlay site to resolve the request.
+        """
+        def call_underlay(request):
+            # Reset the paths and forward to the underlay site.
+            request.prepath = []
+            request.postpath = postpath
+            return self.underlay.getResourceFor(request)
+
+        def wrap_render(orig_render, request):
+            # Wrap the render call of the resource, catching any
+            # UnsupportedMethod exceptions and forwarding those onto
+            # the underlay site.
+            try:
+                return orig_render(request)
+            except UnsupportedMethod:
+                if self.underlay is not None:
+                    resource = call_underlay(request)
+                    return resource.render(request)
+                else:
+                    raise
+
+        postpath = copy.copy(request.postpath)
+        result = super(OverlaySite, self).getResourceFor(request)
+        if isinstance(result, NoResource) and self.underlay is not None:
+            return call_underlay(request)
+        else:
+            result.render = partial(wrap_render, result.render)
+            return result
+
+
 class ResourceOverlay(Resource, object):
     """A resource that can fall-back to a basis resource.
 
@@ -126,13 +174,14 @@
         the web application.
     """
 
-    def __init__(self, endpoint, listener):
-        self.site = Site(StartPage())
+    def __init__(self, endpoint, listener, status_worker):
+        self.site = OverlaySite(StartPage())
         self.site.requestFactory = CleanPathRequest
         super(WebApplicationService, self).__init__(endpoint, self.site)
         self.websocket = WebSocketFactory(listener)
         self.threadpool = ThreadPoolLimiter(
             reactor.threadpoolForDatabase, concurrency.webapp)
+        self.status_worker = status_worker
 
     def prepareApplication(self):
         """Return the WSGI application.
@@ -157,16 +206,33 @@
         with RegionConfiguration.open() as config:
             static_root = File(config.static_root)
 
+        # Setup resources to process paths that twisted handles.
+        metadata = Resource()
+        metadata.putChild(b'status', StatusHandlerResource(self.status_worker))
+
+        maas = Resource()
+        maas.putChild(b'metadata', metadata)
+        maas.putChild(b'static', static_root)
+        maas.putChild(
+            b'ws',
+            WebSocketsResource(lookupProtocolForFactory(self.websocket)))
+
         root = Resource()
-        webapp = ResourceOverlay(
+        root.putChild(b'', Redirect(b"MAAS/"))
+        root.putChild(b'MAAS', maas)
+
+        # Setup the resources to process paths that django handles.
+        underlay_maas = ResourceOverlay(
             WSGIResource(reactor, self.threadpool, application))
-        root.putChild(b"", Redirect(b"MAAS/"))
-        root.putChild(b"MAAS", webapp)
-        webapp.putChild(
-            b'ws',
-            WebSocketsResource(lookupProtocolForFactory(self.websocket)))
-        webapp.putChild(b'static', static_root)
+        underlay_root = Resource()
+        underlay_root.putChild(b'MAAS', underlay_maas)
+        underlay_site = Site(underlay_root)
+        underlay_site.requestFactory = CleanPathRequest
+
+        # Setup the main resource as the twisted handler and the underlay
+        # resource as the django handler.
         self.site.resource = root
+        self.site.underlay = underlay_site
 
     def installFailed(self, failure):
         """Display a page explaining why the web app could not start."""

=== modified file 'src/metadataserver/api.py'
--- src/metadataserver/api.py	2017-02-17 14:23:04 +0000
+++ src/metadataserver/api.py	2017-02-28 20:14:47 +0000
@@ -176,7 +176,7 @@
 
 
 def add_event_to_node_event_log(
-        node, origin, action, description, result=None):
+        node, origin, action, description, result=None, created=None):
     """Add an entry to the node's event log."""
     if node.status == NODE_STATUS.COMMISSIONING:
         if result in ['SUCCESS', None]:
@@ -205,7 +205,7 @@
         node.system_id, type_name, type_level=event_details.level,
         type_description=event_details.description,
         event_action=action,
-        event_description="'%s' %s" % (origin, description))
+        event_description="'%s' %s" % (origin, description), created=created)
 
 
 def process_file(

=== added file 'src/metadataserver/api_twisted.py'
--- src/metadataserver/api_twisted.py	1970-01-01 00:00:00 +0000
+++ src/metadataserver/api_twisted.py	2017-02-28 20:14:47 +0000
@@ -0,0 +1,290 @@
+# Copyright 2017 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Metadata API that runs in the Twisted reactor."""
+
+import base64
+import bz2
+from collections import defaultdict
+from datetime import datetime
+import json
+
+from maasserver.api.utils import extract_oauth_key_from_auth_header
+from maasserver.enum import (
+    NODE_STATUS,
+    NODE_TYPE,
+)
+from maasserver.utils.orm import (
+    in_transaction,
+    transactional,
+    TransactionManagementError,
+)
+from maasserver.utils.threads import deferToDatabase
+from metadataserver import logger
+from metadataserver.api import (
+    add_event_to_node_event_log,
+    process_file,
+)
+from metadataserver.models import NodeKey
+from provisioningserver.logger import LegacyLogger
+from twisted.application.internet import TimerService
+from twisted.internet import reactor
+from twisted.web.resource import Resource
+
+
+log = LegacyLogger()
+
+
+class StatusHandlerResource(Resource):
+
+    # Has no children, so getChild will not be called.
+    isLeaf = True
+
+    # Only POST operations are allowed.
+    allowedMethods = [b'POST']
+
+    # Required keys in the message.
+    requiredMessageKeys = ['event_type', 'origin', 'name', 'description']
+
+    def __init__(self, status_worker):
+        self.worker = status_worker
+
+    def render_POST(self, request):
+        # Extract the authorization from request. This only does a basic
+        # check that its provided. The status worker will do the authorization,
+        # the negative to this is that the calling client will no know. To
+        # them the message was accepted. This overall is okay since they are
+        # just status messages.
+        authorization = request.getHeader(b'authorization')
+        if not authorization:
+            request.setResponseCode(401)
+            return b""
+        authorization = extract_oauth_key_from_auth_header(
+            authorization.decode('utf-8'))
+        if authorization is None:
+            request.setResponseCode(401)
+            return b""
+
+        # Load the content to ensure that its atleast correct before placing
+        # it into the status worker.
+        payload = request.content.read()
+        try:
+            payload = payload.decode("ascii")
+        except UnicodeDecodeError as error:
+            request.setResponseCode(400)
+            error_msg = "Status payload must be ASCII-only: %s" % error
+            logger.error(error_msg)
+            return error_msg.encode('ascii')
+
+        try:
+            message = json.loads(payload)
+        except ValueError:
+            request.setResponseCode(400)
+            error_msg = "Status payload is not valid JSON:\n%s\n\n" % payload
+            logger.error(error_msg)
+            return error_msg.encode('ascii')
+
+        # Filter the level early so less messages need to be processed.
+        level = message.get('level', None)
+        if level is not None and level == "DEBUG":
+            # Ignore all debug messages.
+            request.setResponseCode(204)
+            return b""
+
+        # Ensure the other required keys exist.
+        missing_keys = [
+            key
+            for key in self.requiredMessageKeys
+            if key not in message
+        ]
+        if len(missing_keys) > 0:
+            request.setResponseCode(400)
+            error_msg = (
+                'Missing parameter(s) %s in '
+                'status message.' % ', '.join(missing_keys))
+            logger.error(error_msg)
+            return error_msg.encode('ascii')
+
+        # Queue the message with its authorization in the status worker.
+        self.worker.queueMessage(authorization, message)
+        request.setResponseCode(204)
+        return b""
+
+
+class StatusWorkerService(TimerService, object):
+    """Service to update nodes from recieved status messages."""
+
+    check_interval = 60  # Every second.
+
+    def __init__(self, dbtasks, clock=reactor):
+        # Call self._tryUpdateNodes() every self.check_interval.
+        super(StatusWorkerService, self).__init__(
+            self.check_interval, self._tryUpdateNodes)
+        self.dbtasks = dbtasks
+        self.clock = clock
+        self.queue = defaultdict(list)
+
+    def _tryUpdateNodes(self):
+        if len(self.queue) != 0:
+            queue, self.queue = self.queue, defaultdict(list)
+            d = deferToDatabase(self._preProcessQueue, queue)
+            d.addCallback(self._processMessagesLater)
+            d.addErrback(log.err, "Failed to process node status messages.")
+            return d
+
+    @transactional
+    def _preProcessQueue(self, queue):
+        """Check authorizations.
+
+        Return a list of (node, messages) tuples, where each node is found
+        from its authorisation.
+        """
+        keys = NodeKey.objects.filter(
+            key__in=list(queue.keys())).select_related('node')
+        return [
+            (key.node, queue[key.key])
+            for key in keys
+        ]
+
+    def _processMessagesLater(self, tasks):
+        # Move all messages on the queue off onto the database tasks queue.
+        # We're not going to wait for them to be processed because we can't /
+        # don't apply back-pressure to those systems that are producing these
+        # messages anyway.
+        for node, messages in tasks:
+            self.dbtasks.addTask(self._processMessages, node, messages)
+
+    def _processMessages(self, node, messages):
+        # Push the messages into the database, recording them for this node.
+        # This should be called in a non-reactor thread with a pre-existing
+        # connection (e.g. via deferToDatabase).
+        if in_transaction():
+            raise TransactionManagementError(
+                "_processMessages must be called from "
+                "outside of a transaction.")
+        else:
+            # Here we're in a database thread, with a database connection.
+            for message in messages:
+                try:
+                    self._processMessage(node, message)
+                except:
+                    log.err(
+                        None,
+                        "Failed to process message "
+                        "for node: %s" % node.hostname)
+
+    @transactional
+    def _processMessage(self, node, message):
+        event_type = message['event_type']
+        origin = message['origin']
+        activity_name = message['name']
+        description = message['description']
+        result = message.get('result', None)
+
+        # Add this event to the node event log.
+        add_event_to_node_event_log(
+            node, origin, activity_name, description, result,
+            message['timestamp'])
+
+        # Group files together with the ScriptResult they belong.
+        results = {}
+        for sent_file in message.get('files', []):
+            # Set the result type according to the node's status.
+            if node.status == NODE_STATUS.TESTING:
+                script_set = node.current_testing_script_set
+            elif (node.status == NODE_STATUS.COMMISSIONING or
+                    node.node_type != NODE_TYPE.MACHINE):
+                script_set = node.current_commissioning_script_set
+            elif node.status == NODE_STATUS.DEPLOYING:
+                script_set = node.current_installation_script_set
+            else:
+                raise ValueError(
+                    "Invalid status for saving files: %d" % node.status)
+
+            script_name = sent_file['path']
+            content = self._retrieve_content(
+                compression=sent_file.get('compression', None),
+                encoding=sent_file['encoding'],
+                content=sent_file['content'])
+            process_file(results, script_set, script_name, content, sent_file)
+
+        # Commit results to the database.
+        for script_result, args in results.items():
+            script_result.store_result(**args)
+
+        # Update the last ping in any status which uses a script_set whenever a
+        # node in that status contacts us.
+        script_set_statuses = {
+            NODE_STATUS.COMMISSIONING: node.current_commissioning_script_set,
+            NODE_STATUS.TESTING: node.current_testing_script_set,
+            NODE_STATUS.DEPLOYING: node.current_installation_script_set,
+        }
+        script_set = script_set_statuses.get(node.status)
+        if script_set is not None:
+            script_set.last_ping = message['timestamp']
+            script_set.save()
+
+        # At the end of a top-level event, we change the node status.
+        save_node = False
+        if self._is_top_level(activity_name) and event_type == 'finish':
+            if node.status == NODE_STATUS.COMMISSIONING:
+                if result in ['FAIL', 'FAILURE']:
+                    node.status = NODE_STATUS.FAILED_COMMISSIONING
+                    save_node = True
+            elif node.status == NODE_STATUS.DEPLOYING:
+                if result in ['FAIL', 'FAILURE']:
+                    node.mark_failed(
+                        comment="Installation failed (refer to the "
+                                "installation log for more information).")
+                    save_node = True
+            elif node.status == NODE_STATUS.DISK_ERASING:
+                if result in ['FAIL', 'FAILURE']:
+                    node.mark_failed(comment="Failed to erase disks.")
+                    save_node = True
+
+            # Deallocate the node if we enter any terminal state.
+            if node.node_type == NODE_TYPE.MACHINE and node.status in [
+                    NODE_STATUS.READY,
+                    NODE_STATUS.FAILED_COMMISSIONING]:
+                node.status_expires = None
+                node.owner = None
+                node.error = 'failed: %s' % description
+                save_node = True
+
+        if save_node:
+            node.save()
+
+    def _retrieve_content(self, compression, encoding, content):
+        """Extract the content of the sent file."""
+        # Select the appropriate decompressor.
+        if compression is None:
+            decompress = lambda s: s
+        elif compression == 'bzip2':
+            decompress = bz2.decompress
+        else:
+            raise ValueError('Invalid compression: %s' % compression)
+
+        # Select the appropriate decoder.
+        if encoding == 'base64':
+            decode = base64.decodebytes
+        else:
+            raise ValueError('Invalid encoding: %s' % encoding)
+
+        return decompress(decode(content.encode("ascii")))
+
+    def _is_top_level(self, activity_name):
+        """Top-level events do not have slashes in their names."""
+        return '/' not in activity_name
+
+    def queueMessage(self, authorization, message):
+        """Queue message for processing."""
+        # Ensure a timestamp exists in the message and convert it to a
+        # datetime object. This is used to update the `last_ping` and the
+        # time for the event message.
+        timestamp = message.get('timestamp', None)
+        if timestamp is not None:
+            message['timestamp'] = datetime.utcfromtimestamp(
+                message['timestamp'])
+        else:
+            message['timestamp'] = datetime.utcnow()
+        self.queue[authorization].append(message)

=== added file 'src/metadataserver/tests/test_api_twisted.py'
--- src/metadataserver/tests/test_api_twisted.py	1970-01-01 00:00:00 +0000
+++ src/metadataserver/tests/test_api_twisted.py	2017-02-28 20:14:47 +0000
@@ -0,0 +1,766 @@
+# Copyright 2017 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Tests for the twisted metadata API."""
+
+__all__ = []
+
+import base64
+import bz2
+from datetime import datetime
+from io import BytesIO
+import json
+from unittest.mock import (
+    Mock,
+    sentinel,
+)
+
+from crochet import wait_for
+from maasserver.enum import NODE_STATUS
+from maasserver.models import (
+    Event,
+    Tag,
+)
+from maasserver.models.signals.testing import SignalsDisabled
+from maasserver.testing.factory import factory
+from maasserver.testing.testcase import (
+    MAASServerTestCase,
+    MAASTransactionServerTestCase,
+)
+from maasserver.utils.orm import (
+    reload_object,
+    transactional,
+    TransactionManagementError,
+)
+from maasserver.utils.threads import deferToDatabase
+from maastesting.matchers import (
+    MockCalledOnceWith,
+    MockNotCalled,
+)
+from maastesting.testcase import MAASTestCase
+from metadataserver import api
+from metadataserver.api_twisted import (
+    StatusHandlerResource,
+    StatusWorkerService,
+)
+from metadataserver.enum import SCRIPT_STATUS
+from metadataserver.models import NodeKey
+from testtools import ExpectedException
+from testtools.matchers import (
+    Equals,
+    MatchesListwise,
+    MatchesSetwise,
+)
+from twisted.internet.defer import inlineCallbacks
+from twisted.web.test.requesthelper import DummyRequest
+
+
+wait_for_reactor = wait_for(30)
+
+
+class TestStatusHandlerResource(MAASTestCase):
+
+    def make_request(self, content=None, token=None):
+        request = DummyRequest([])
+        if token is None:
+            token = factory.make_name('token')
+        request.requestHeaders.addRawHeader(
+            b'authorization', 'oauth_token=%s' % token)
+        if content is not None:
+            request.content = BytesIO(content)
+        return request
+
+    def test__init__(self):
+        resource = StatusHandlerResource(sentinel.status_worker)
+        self.assertIs(sentinel.status_worker, resource.worker)
+        self.assertTrue(resource.isLeaf)
+        self.assertEquals([b'POST'], resource.allowedMethods)
+        self.assertEquals(
+            ['event_type', 'origin', 'name', 'description'],
+            resource.requiredMessageKeys)
+
+    def test__render_POST_missing_authorization(self):
+        resource = StatusHandlerResource(sentinel.status_worker)
+        request = DummyRequest([])
+        output = resource.render_POST(request)
+        self.assertEquals(b'', output)
+        self.assertEquals(401, request.responseCode)
+
+    def test__render_POST_empty_authorization(self):
+        resource = StatusHandlerResource(sentinel.status_worker)
+        request = DummyRequest([])
+        request.requestHeaders.addRawHeader(b'authorization', '')
+        output = resource.render_POST(request)
+        self.assertEquals(b'', output)
+        self.assertEquals(401, request.responseCode)
+
+    def test__render_POST_bad_authorization(self):
+        resource = StatusHandlerResource(sentinel.status_worker)
+        request = DummyRequest([])
+        request.requestHeaders.addRawHeader(
+            b'authorization', factory.make_name('auth'))
+        output = resource.render_POST(request)
+        self.assertEquals(b'', output)
+        self.assertEquals(401, request.responseCode)
+
+    def test__render_POST_body_must_be_ascii(self):
+        resource = StatusHandlerResource(sentinel.status_worker)
+        request = self.make_request(content=b'\xe9')
+        output = resource.render_POST(request)
+        self.assertEquals(
+            b"Status payload must be ASCII-only: 'ascii' codec can't "
+            b"decode byte 0xe9 in position 0: ordinal not in range(128)",
+            output)
+        self.assertEquals(400, request.responseCode)
+
+    def test__render_POST_body_must_be_valid_json(self):
+        resource = StatusHandlerResource(sentinel.status_worker)
+        request = self.make_request(content=b'testing not json')
+        output = resource.render_POST(request)
+        self.assertEquals(
+            b"Status payload is not valid JSON:\ntesting not json\n\n",
+            output)
+        self.assertEquals(400, request.responseCode)
+
+    def test__render_POST_ignores_debug_messages(self):
+        resource = StatusHandlerResource(sentinel.status_worker)
+        request = self.make_request(content=json.dumps({
+            'level': 'DEBUG',
+        }).encode('ascii'))
+        output = resource.render_POST(request)
+        self.assertEquals(b'', output)
+        self.assertEquals(204, request.responseCode)
+
+    def test__render_POST_validates_required_keys(self):
+        resource = StatusHandlerResource(sentinel.status_worker)
+        request = self.make_request(content=json.dumps({}).encode('ascii'))
+        output = resource.render_POST(request)
+        self.assertEquals(
+            b'Missing parameter(s) event_type, origin, name, description '
+            b'in status message.', output)
+        self.assertEquals(400, request.responseCode)
+
+    def test__render_POST_queue_messages(self):
+        status_worker = Mock()
+        status_worker.queueMessage = Mock()
+        resource = StatusHandlerResource(status_worker)
+        message = {
+            'event_type': factory.make_name('type'),
+            'origin': factory.make_name('origin'),
+            'name': factory.make_name('name'),
+            'description': factory.make_name('description'),
+        }
+        token = factory.make_name('token')
+        request = self.make_request(
+            content=json.dumps(message).encode('ascii'), token=token)
+        output = resource.render_POST(request)
+        self.assertEquals(b'', output)
+        self.assertEquals(204, request.responseCode)
+        self.assertThat(
+            status_worker.queueMessage, MockCalledOnceWith(token, message))
+
+
+class TestStatusWorkerServiceTransactional(MAASTransactionServerTestCase):
+
+    @transactional
+    def make_nodes_with_tokens(self):
+        nodes = [
+            factory.make_Node()
+            for _ in range(3)
+        ]
+        return [
+            (node, NodeKey.objects.get_token_for_node(node))
+            for node in nodes
+        ]
+
+    def make_message(self):
+        return {
+            'event_type': factory.make_name('type'),
+            'origin': factory.make_name('origin'),
+            'name': factory.make_name('name'),
+            'description': factory.make_name('description'),
+            'timestamp': datetime.utcnow().timestamp(),
+        }
+
+    def test__init__(self):
+        worker = StatusWorkerService(sentinel.dbtasks, clock=sentinel.reactor)
+        self.assertEqual(sentinel.dbtasks, worker.dbtasks)
+        self.assertEqual(sentinel.reactor, worker.clock)
+        self.assertEqual(60, worker.step)
+        self.assertEqual((worker._tryUpdateNodes, tuple(), {}), worker.call)
+
+    def test__tryUpdateNodes_returns_None_when_empty_queue(self):
+        worker = StatusWorkerService(sentinel.dbtasks)
+        self.assertIsNone(worker._tryUpdateNodes())
+
+    @wait_for_reactor
+    @inlineCallbacks
+    def test__tryUpdateNodes_sends_work_to_dbtasks(self):
+        nodes_with_tokens = yield deferToDatabase(self.make_nodes_with_tokens)
+        node_messages = {
+            node: [
+                self.make_message()
+                for _ in range(3)
+            ]
+            for node, _ in nodes_with_tokens
+        }
+        dbtasks = Mock()
+        dbtasks.addTask = Mock()
+        worker = StatusWorkerService(dbtasks)
+        for node, token in nodes_with_tokens:
+            for message in node_messages[node]:
+                worker.queueMessage(token.key, message)
+        yield worker._tryUpdateNodes()
+        call_args = [
+            (call_arg[0][1], call_arg[0][2])
+            for call_arg in dbtasks.addTask.call_args_list
+        ]
+        self.assertThat(call_args, MatchesSetwise(*[
+            MatchesListwise([Equals(node), Equals(messages)])
+            for node, messages in node_messages.items()
+        ]))
+
+    @wait_for_reactor
+    @inlineCallbacks
+    def test__processMessages_fails_when_in_transaction(self):
+        worker = StatusWorkerService(sentinel.dbtasks)
+        with ExpectedException(TransactionManagementError):
+            yield deferToDatabase(
+                transactional(worker._processMessages),
+                sentinel.node, [sentinel.message])
+
+    @wait_for_reactor
+    @inlineCallbacks
+    def test__processMessages_calls_processMessage(self):
+        worker = StatusWorkerService(sentinel.dbtasks)
+        mock_processMessage = self.patch(worker, "_processMessage")
+        yield deferToDatabase(
+            worker._processMessages, sentinel.node, [sentinel.message])
+        self.assertThat(
+            mock_processMessage,
+            MockCalledOnceWith(sentinel.node, sentinel.message))
+
+
+def encode_as_base64(content):
+    return base64.encodebytes(content).decode("ascii")
+
+
+class TestStatusWorkerService(MAASServerTestCase):
+
+    def setUp(self):
+        super(TestStatusWorkerService, self).setUp()
+        self.useFixture(SignalsDisabled("power"))
+
+    def processMessage(self, node, payload):
+        worker = StatusWorkerService(sentinel.dbtasks)
+        worker._processMessage(node, payload)
+
+    def test_status_installation_result_does_not_affect_other_node(self):
+        node1 = factory.make_Node(status=NODE_STATUS.DEPLOYING)
+        node2 = factory.make_Node(status=NODE_STATUS.DEPLOYING)
+        payload = {
+            'event_type': 'finish',
+            'result': 'SUCCESS',
+            'origin': 'curtin',
+            'name': 'cmd-install',
+            'description': 'Command Install',
+            'timestamp': datetime.utcnow(),
+        }
+        self.processMessage(node1, payload)
+        self.assertEqual(
+            NODE_STATUS.DEPLOYING, reload_object(node2).status)
+        # Check last node1 event.
+        self.assertEqual(
+            "'curtin' Command Install",
+            Event.objects.filter(node=node1).last().description)
+        # There must me no events for node2.
+        self.assertFalse(Event.objects.filter(node=node2).exists())
+
+    def test_status_installation_success_leaves_node_deploying(self):
+        node = factory.make_Node(interface=True, status=NODE_STATUS.DEPLOYING)
+        payload = {
+            'event_type': 'finish',
+            'result': 'SUCCESS',
+            'origin': 'curtin',
+            'name': 'cmd-install',
+            'description': 'Command Install',
+            'timestamp': datetime.utcnow(),
+        }
+        self.processMessage(node, payload)
+        self.assertEqual(NODE_STATUS.DEPLOYING, reload_object(node).status)
+        # Check last node event.
+        self.assertEqual(
+            "'curtin' Command Install",
+            Event.objects.filter(node=node).last().description)
+
+    def test_status_commissioning_failure_leaves_node_failed(self):
+        node = factory.make_Node(
+            interface=True, status=NODE_STATUS.COMMISSIONING)
+        payload = {
+            'event_type': 'finish',
+            'result': 'FAILURE',
+            'origin': 'curtin',
+            'name': 'commissioning',
+            'description': 'Commissioning',
+            'timestamp': datetime.utcnow(),
+        }
+        self.processMessage(node, payload)
+        self.assertEqual(
+            NODE_STATUS.FAILED_COMMISSIONING, reload_object(node).status)
+        # Check last node event.
+        self.assertEqual(
+            "'curtin' Commissioning",
+            Event.objects.filter(node=node).last().description)
+
+    def test_status_commissioning_failure_clears_owner(self):
+        user = factory.make_User()
+        node = factory.make_Node(
+            interface=True, status=NODE_STATUS.COMMISSIONING, owner=user)
+        payload = {
+            'event_type': 'finish',
+            'result': 'FAILURE',
+            'origin': 'curtin',
+            'name': 'commissioning',
+            'description': 'Commissioning',
+            'timestamp': datetime.utcnow(),
+        }
+        self.assertEqual(user, node.owner)  # Node has an owner
+        self.processMessage(node, payload)
+        self.assertEqual(
+            NODE_STATUS.FAILED_COMMISSIONING, reload_object(node).status)
+        self.assertIsNone(reload_object(node).owner)
+
+    def test_status_installation_failure_leaves_node_failed(self):
+        node = factory.make_Node(interface=True, status=NODE_STATUS.DEPLOYING)
+        payload = {
+            'event_type': 'finish',
+            'result': 'FAILURE',
+            'origin': 'curtin',
+            'name': 'cmd-install',
+            'description': 'Command Install',
+            'timestamp': datetime.utcnow(),
+        }
+        self.processMessage(node, payload)
+        self.assertEqual(
+            NODE_STATUS.FAILED_DEPLOYMENT, reload_object(node).status)
+        # Check last node event.
+        self.assertEqual(
+            "Installation failed (refer to the installation"
+            " log for more information).",
+            Event.objects.filter(node=node).last().description)
+
+    def test_status_installation_fail_leaves_node_failed(self):
+        node = factory.make_Node(interface=True, status=NODE_STATUS.DEPLOYING)
+        payload = {
+            'event_type': 'finish',
+            'result': 'FAIL',
+            'origin': 'curtin',
+            'name': 'cmd-install',
+            'description': 'Command Install',
+            'timestamp': datetime.utcnow(),
+        }
+        self.processMessage(node, payload)
+        self.assertEqual(
+            NODE_STATUS.FAILED_DEPLOYMENT, reload_object(node).status)
+        # Check last node event.
+        self.assertEqual(
+            "Installation failed (refer to the installation"
+            " log for more information).",
+            Event.objects.filter(node=node).last().description)
+
+    def test_status_installation_failure_doesnt_clear_owner(self):
+        user = factory.make_User()
+        node = factory.make_Node(
+            interface=True, status=NODE_STATUS.DEPLOYING, owner=user)
+        payload = {
+            'event_type': 'finish',
+            'result': 'FAILURE',
+            'origin': 'curtin',
+            'name': 'cmd-install',
+            'description': 'Command Install',
+            'timestamp': datetime.utcnow(),
+        }
+        self.assertEqual(user, node.owner)  # Node has an owner
+        self.processMessage(node, payload)
+        self.assertEqual(
+            NODE_STATUS.FAILED_DEPLOYMENT, reload_object(node).status)
+        self.assertIsNotNone(reload_object(node).owner)
+
+    def test_status_commissioning_failure_does_not_populate_tags(self):
+        populate_tags_for_single_node = self.patch(
+            api, "populate_tags_for_single_node")
+        node = factory.make_Node(
+            interface=True, status=NODE_STATUS.COMMISSIONING)
+        payload = {
+            'event_type': 'finish',
+            'result': 'FAILURE',
+            'origin': 'curtin',
+            'name': 'commissioning',
+            'description': 'Commissioning',
+            'timestamp': datetime.utcnow(),
+        }
+        self.processMessage(node, payload)
+        self.assertEqual(
+            NODE_STATUS.FAILED_COMMISSIONING, reload_object(node).status)
+        self.assertThat(populate_tags_for_single_node, MockNotCalled())
+
+    def test_status_erasure_failure_leaves_node_failed(self):
+        node = factory.make_Node(
+            interface=True, status=NODE_STATUS.DISK_ERASING)
+        payload = {
+            'event_type': 'finish',
+            'result': 'FAILURE',
+            'origin': 'curtin',
+            'name': 'cmd-erase',
+            'description': 'Erasing disk',
+            'timestamp': datetime.utcnow(),
+        }
+        self.processMessage(node, payload)
+        self.assertEqual(
+            NODE_STATUS.FAILED_DISK_ERASING, reload_object(node).status)
+        # Check last node event.
+        self.assertEqual(
+            "Failed to erase disks.",
+            Event.objects.filter(node=node).last().description)
+
+    def test_status_erasure_failure_does_not_populate_tags(self):
+        populate_tags_for_single_node = self.patch(
+            api, "populate_tags_for_single_node")
+        node = factory.make_Node(
+            interface=True, status=NODE_STATUS.DISK_ERASING)
+        payload = {
+            'event_type': 'finish',
+            'result': 'FAILURE',
+            'origin': 'curtin',
+            'name': 'cmd-erase',
+            'description': 'Erasing disk',
+            'timestamp': datetime.utcnow(),
+        }
+        self.processMessage(node, payload)
+        self.assertEqual(
+            NODE_STATUS.FAILED_DISK_ERASING, reload_object(node).status)
+        self.assertThat(populate_tags_for_single_node, MockNotCalled())
+
+    def test_status_erasure_failure_doesnt_clear_owner(self):
+        user = factory.make_User()
+        node = factory.make_Node(
+            interface=True, status=NODE_STATUS.DISK_ERASING, owner=user)
+        payload = {
+            'event_type': 'finish',
+            'result': 'FAILURE',
+            'origin': 'curtin',
+            'name': 'cmd-erase',
+            'description': 'Erasing disk',
+            'timestamp': datetime.utcnow(),
+        }
+        self.processMessage(node, payload)
+        self.assertEqual(
+            NODE_STATUS.FAILED_DISK_ERASING, reload_object(node).status)
+        self.assertEqual(user, node.owner)
+
+    def test_status_with_file_bad_encoder_fails(self):
+        node = factory.make_Node(
+            interface=True, status=NODE_STATUS.COMMISSIONING)
+        contents = b'These are the contents of the file.'
+        encoded_content = encode_as_base64(bz2.compress(contents))
+        payload = {
+            'event_type': 'finish',
+            'result': 'FAILURE',
+            'origin': 'curtin',
+            'name': 'commissioning',
+            'description': 'Commissioning',
+            'timestamp': datetime.utcnow(),
+            'files': [
+                {
+                    "path": "sample.txt",
+                    "encoding": "uuencode",
+                    "compression": "bzip2",
+                    "content": encoded_content
+                }
+            ]
+        }
+        with ExpectedException(ValueError):
+            self.processMessage(node, payload)
+
+    def test_status_with_file_bad_compression_fails(self):
+        node = factory.make_Node(
+            interface=True, status=NODE_STATUS.COMMISSIONING)
+        contents = b'These are the contents of the file.'
+        encoded_content = encode_as_base64(bz2.compress(contents))
+        payload = {
+            'event_type': 'finish',
+            'result': 'FAILURE',
+            'origin': 'curtin',
+            'name': 'commissioning',
+            'description': 'Commissioning',
+            'timestamp': datetime.utcnow(),
+            'files': [
+                {
+                    "path": "sample.txt",
+                    "encoding": "base64",
+                    "compression": "jpeg",
+                    "content": encoded_content
+                }
+            ]
+        }
+        with ExpectedException(ValueError):
+            self.processMessage(node, payload)
+
+    def test_status_with_file_no_compression_succeeds(self):
+        node = factory.make_Node(
+            interface=True, status=NODE_STATUS.COMMISSIONING,
+            with_empty_script_sets=True)
+        script_result = (
+            node.current_commissioning_script_set.scriptresult_set.first())
+        script_result.status = SCRIPT_STATUS.RUNNING
+        script_result.save()
+        contents = b'These are the contents of the file.'
+        encoded_content = encode_as_base64(contents)
+        payload = {
+            'event_type': 'finish',
+            'result': 'FAILURE',
+            'origin': 'curtin',
+            'name': 'commissioning',
+            'description': 'Commissioning',
+            'timestamp': datetime.utcnow(),
+            'files': [
+                {
+                    "path": script_result.name,
+                    "encoding": "base64",
+                    "content": encoded_content
+                }
+            ]
+        }
+        self.processMessage(node, payload)
+        self.assertEqual(contents, reload_object(script_result).stdout)
+
+    def test_status_with_file_invalid_statuses_fails(self):
+        """Adding files should fail for every status that's neither
+        COMMISSIONING nor DEPLOYING"""
+        for node_status in [
+                NODE_STATUS.DEFAULT,
+                NODE_STATUS.NEW,
+                NODE_STATUS.FAILED_COMMISSIONING,
+                NODE_STATUS.MISSING,
+                NODE_STATUS.READY,
+                NODE_STATUS.RESERVED,
+                NODE_STATUS.DEPLOYED,
+                NODE_STATUS.RETIRED,
+                NODE_STATUS.BROKEN,
+                NODE_STATUS.ALLOCATED,
+                NODE_STATUS.FAILED_DEPLOYMENT,
+                NODE_STATUS.RELEASING,
+                NODE_STATUS.FAILED_RELEASING,
+                NODE_STATUS.DISK_ERASING,
+                NODE_STATUS.FAILED_DISK_ERASING]:
+            node = factory.make_Node(interface=True, status=node_status)
+            contents = b'These are the contents of the file.'
+            encoded_content = encode_as_base64(bz2.compress(contents))
+            payload = {
+                'event_type': 'finish',
+                'result': 'FAILURE',
+                'origin': 'curtin',
+                'name': 'commissioning',
+                'description': 'Commissioning',
+                'timestamp': datetime.utcnow(),
+                'files': [
+                    {
+                        "path": "sample.txt",
+                        "encoding": "base64",
+                        "compression": "bzip2",
+                        "content": encoded_content
+                    }
+                ]
+            }
+            with ExpectedException(ValueError):
+                self.processMessage(node, payload)
+
+    def test_status_with_file_succeeds(self):
+        """Adding files should succeed for every status that's either
+        COMMISSIONING or DEPLOYING"""
+        for node_status, target_status in [
+                (NODE_STATUS.COMMISSIONING, NODE_STATUS.FAILED_COMMISSIONING),
+                (NODE_STATUS.DEPLOYING, NODE_STATUS.FAILED_DEPLOYMENT)]:
+            node = factory.make_Node(
+                interface=True, status=node_status,
+                with_empty_script_sets=True)
+            if node_status == NODE_STATUS.COMMISSIONING:
+                script_set = node.current_commissioning_script_set
+            elif node_status == NODE_STATUS.DEPLOYING:
+                script_set = node.current_installation_script_set
+            script_result = script_set.scriptresult_set.first()
+            script_result.status = SCRIPT_STATUS.RUNNING
+            script_result.save()
+            contents = b'These are the contents of the file.'
+            encoded_content = encode_as_base64(bz2.compress(contents))
+            payload = {
+                'event_type': 'finish',
+                'result': 'FAILURE',
+                'origin': 'curtin',
+                'name': 'commissioning',
+                'description': 'Commissioning',
+                'timestamp': datetime.utcnow(),
+                'files': [
+                    {
+                        "path": script_result.name,
+                        "encoding": "base64",
+                        "compression": "bzip2",
+                        "content": encoded_content
+                    }
+                ]
+            }
+            self.processMessage(node, payload)
+            self.assertEqual(
+                target_status, reload_object(node).status)
+            # Check the node result.
+            self.assertEqual(contents, reload_object(script_result).stdout)
+
+    def test_status_with_results_succeeds(self):
+        """Adding a script result should succeed"""
+        node = factory.make_Node(
+            interface=True, status=NODE_STATUS.COMMISSIONING,
+            with_empty_script_sets=True)
+        script_result = (
+            node.current_commissioning_script_set.scriptresult_set.first())
+        script_result.status = SCRIPT_STATUS.RUNNING
+        script_result.save()
+        contents = b'These are the contents of the file.'
+        encoded_content = encode_as_base64(bz2.compress(contents))
+        payload = {
+            'event_type': 'finish',
+            'result': 'FAILURE',
+            'origin': 'curtin',
+            'name': 'commissioning',
+            'description': 'Commissioning',
+            'timestamp': datetime.utcnow(),
+            'files': [
+                {
+                    "path": script_result.name,
+                    "encoding": "base64",
+                    "compression": "bzip2",
+                    "content": encoded_content,
+                    "result": -42
+                }
+            ]
+        }
+        self.processMessage(node, payload)
+        script_result = reload_object(script_result)
+        self.assertEqual(contents, script_result.stdout)
+        self.assertEqual(-42, script_result.exit_status)
+
+    def test_status_with_results_no_exit_status_defaults_to_zero(self):
+        """Adding a script result should succeed without a return code defaults
+        it to zero."""
+        node = factory.make_Node(
+            interface=True, status=NODE_STATUS.COMMISSIONING,
+            with_empty_script_sets=True)
+        script_result = (
+            node.current_commissioning_script_set.scriptresult_set.first())
+        script_result.status = SCRIPT_STATUS.RUNNING
+        script_result.save()
+        contents = b'These are the contents of the file.'
+        encoded_content = encode_as_base64(bz2.compress(contents))
+        payload = {
+            'event_type': 'finish',
+            'result': 'FAILURE',
+            'origin': 'curtin',
+            'name': 'commissioning',
+            'description': 'Commissioning',
+            'timestamp': datetime.utcnow(),
+            'files': [
+                {
+                    "path": script_result.name,
+                    "encoding": "base64",
+                    "compression": "bzip2",
+                    "content": encoded_content,
+                }
+            ]
+        }
+        self.processMessage(node, payload)
+        self.assertEqual(0, reload_object(script_result).exit_status)
+
+    def test_status_stores_virtual_tag_on_node_if_virtual(self):
+        node = factory.make_Node(
+            status=NODE_STATUS.COMMISSIONING, with_empty_script_sets=True)
+        content = 'virtual'.encode('utf-8')
+        payload = {
+            'event_type': 'finish',
+            'result': 'SUCCESS',
+            'origin': 'curtin',
+            'name': 'commissioning',
+            'description': 'Commissioning',
+            'timestamp': datetime.utcnow(),
+            'files': [
+                {
+                    "path": "00-maas-02-virtuality",
+                    "encoding": "base64",
+                    "content": encode_as_base64(content),
+                }
+            ]
+        }
+        self.processMessage(node, payload)
+        node = reload_object(node)
+        self.assertEqual(
+            ["virtual"], [each_tag.name for each_tag in node.tags.all()])
+        for script_result in node.current_commissioning_script_set:
+            if script_result.name == "00-maas-02-virtuality":
+                break
+        self.assertEqual(content, script_result.stdout)
+
+    def test_status_removes_virtual_tag_on_node_if_not_virtual(self):
+        node = factory.make_Node(
+            status=NODE_STATUS.COMMISSIONING, with_empty_script_sets=True)
+        tag, _ = Tag.objects.get_or_create(name='virtual')
+        node.tags.add(tag)
+        content = 'none'.encode('utf-8')
+        payload = {
+            'event_type': 'finish',
+            'result': 'SUCCESS',
+            'origin': 'curtin',
+            'name': 'commissioning',
+            'description': 'Commissioning',
+            'timestamp': datetime.utcnow(),
+            'files': [
+                {
+                    "path": "00-maas-02-virtuality",
+                    "encoding": "base64",
+                    "content": encode_as_base64(content),
+                }
+            ]
+        }
+        self.processMessage(node, payload)
+        node = reload_object(node)
+        self.assertEqual(
+            [], [each_tag.name for each_tag in node.tags.all()])
+        for script_result in node.current_commissioning_script_set:
+            if script_result.name == "00-maas-02-virtuality":
+                break
+        self.assertEqual(content, script_result.stdout)
+
+    def test_status_updates_script_status_last_ping(self):
+        nodes = {
+            status: factory.make_Node(
+                status=status, with_empty_script_sets=True)
+            for status in (
+                NODE_STATUS.COMMISSIONING,
+                NODE_STATUS.TESTING,
+                NODE_STATUS.DEPLOYING)
+        }
+
+        for status, node in nodes.items():
+            payload = {
+                'event_type': 'progress',
+                'origin': 'curtin',
+                'name': 'test',
+                'description': 'testing',
+                'timestamp': datetime.utcnow(),
+            }
+            self.processMessage(node, payload)
+            script_set_statuses = {
+                NODE_STATUS.COMMISSIONING: (
+                    node.current_commissioning_script_set),
+                NODE_STATUS.TESTING: node.current_testing_script_set,
+                NODE_STATUS.DEPLOYING: node.current_installation_script_set,
+            }
+            script_set = script_set_statuses.get(node.status)
+            self.assertIsNotNone(script_set.last_ping)