Merge lp:~mpontillo/maas/avahi-service-and-rpc into lp:~maas-committers/maas/trunk

[Mike Pontillo]

Add twisted service and RPC so that mDNS entries can be recorded end-to-end in the MAAS database.

[Gavin Panella (allenap)]

Looks good. I've made a few suggestions, that's all.

review: Approve

[Mike Pontillo (mpontillo)]

Thanks for the review. Some replies below.

=== modified file '.bzrignore'
--- .bzrignore	2016-08-18 17:31:05 +0000
+++ .bzrignore	2016-08-31 18:00:29 +0000
@@ -6,6 +6,7 @@
 ./.failed
 ./.hypothesis
 ./.idea
+./.idea/libraries
 ./.idea/scopes
 ./.idea/workspace.xml
 ./.installed.cfg

=== modified file '.gitignore'
--- .gitignore	2016-08-18 17:31:05 +0000
+++ .gitignore	2016-08-31 18:00:29 +0000
@@ -6,6 +6,7 @@
 /.failed
 /.hypothesis
 /.idea
+/.idea/libraries
 /.idea/scopes
 /.idea/workspace.xml
 /.installed.cfg

=== modified file '.idea/maas.iml'
--- .idea/maas.iml	2016-04-15 16:44:43 +0000
+++ .idea/maas.iml	2016-08-31 18:00:29 +0000
@@ -1,5 +1,12 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <module type="PYTHON_MODULE" version="4">
+  <component name="FacetManager">
+    <facet type="buildout-python" name="Buildout Support">
+      <configuration>
+        <option name="script" value="bin/py" />
+      </configuration>
+    </facet>
+  </component>
   <component name="NewModuleRootManager">
     <content url="file://$MODULE_DIR$">
       <sourceFolder url="file://$MODULE_DIR$/src" isTestSource="false" />
@@ -11,6 +18,7 @@
     </content>
     <orderEntry type="jdk" jdkName="Python 3.5.1+ (/usr/bin/python3.5)" jdkType="Python SDK" />
     <orderEntry type="sourceFolder" forTests="false" />
+    <orderEntry type="library" name="Buildout Eggs" level="project" />
   </component>
   <component name="TemplatesService">
     <option name="TEMPLATE_CONFIGURATION" value="Django" />

=== modified file 'src/maasserver/models/interface.py'
--- src/maasserver/models/interface.py	2016-08-20 12:42:22 +0000
+++ src/maasserver/models/interface.py	2016-08-31 18:00:29 +0000
@@ -1165,7 +1165,7 @@
         from maasserver.models.mdns import MDNS
         if self.mdns_discovery_state is False:
             return None
-        ip = avahi_json['ip']
+        ip = avahi_json['address']
         hostname = avahi_json['hostname']
         deleted = MDNS.objects.delete_and_log_obsolete_mdns_entries(
             hostname, ip, interface=self)
@@ -1177,8 +1177,8 @@
             # If we deleted a previous mDNS entry, then we have already
             # generated a log statement about this mDNS entry.
             if not deleted:
-                maaslog.info("%s: New mDNS hostname observed for %s: '%s'." % (
-                    self.get_log_string(), ip, hostname))
+                maaslog.info("%s: New mDNS entry resolved: '%s' on %s." % (
+                    self.get_log_string(), hostname, ip))
         else:
             binding.count += 1
             binding.save(update_fields=['count', 'updated'])

=== modified file 'src/maasserver/models/mdns.py'
--- src/maasserver/models/mdns.py	2016-08-15 15:33:20 +0000
+++ src/maasserver/models/mdns.py	2016-08-31 18:00:29 +0000
@@ -27,6 +27,7 @@
     get_one,
     UniqueViolation,
 )
+from netaddr import IPAddress
 from provisioningserver.logger import get_maas_logger
 
 
@@ -47,11 +48,16 @@
         """
         # Check if this hostname was previously assigned to another IP address.
         deleted = False
+        incoming_ip_version = IPAddress(ip).version
         previous_bindings = self.filter(
             hostname=hostname, interface=interface).exclude(ip=ip)
+        # Check if this hostname was previously assigned to a different IP.
         for binding in previous_bindings:
+            if incoming_ip_version != IPAddress(binding.ip).version:
+                # Don't move hostnames between address families.
+                continue
             maaslog.info("%s: Hostname '%s' moved from %s to %s." % (
-                interface.get_log_string(), ip, binding.ip, ip))
+                interface.get_log_string(), hostname, binding.ip, ip))
             binding.delete()
             deleted = True
         # Check if this IP address had a different hostname assigned.

=== modified file 'src/maasserver/models/tests/test_interface.py'
--- src/maasserver/models/tests/test_interface.py	2016-08-25 17:06:55 +0000
+++ src/maasserver/models/tests/test_interface.py	2016-08-31 18:00:29 +0000
@@ -951,7 +951,7 @@
         if hostname is None:
             hostname = factory.make_hostname()
         return {
-            'ip': ip,
+            'address': ip,
             'hostname': hostname,
         }
 
@@ -983,7 +983,7 @@
         self.assertThat(MDNS.objects.count(), Equals(1))
         iface.update_mdns_entry(json)
         mdns_entry = reload_object(mdns_entry)
-        self.assertThat(mdns_entry.ip, Equals(json['ip']))
+        self.assertThat(mdns_entry.ip, Equals(json['address']))
         self.assertThat(mdns_entry.hostname, Equals(json['hostname']))
         # This is the second time we saw this entry.
         self.assertThat(mdns_entry.count, Equals(2))
@@ -995,13 +995,13 @@
         json = self.make_mdns_entry_json()
         iface.update_mdns_entry(json)
         # Have a different IP address claim ownership of the hostname.
-        json['ip'] = factory.make_ip_address(ipv6=False)
+        json['address'] = factory.make_ip_address(ipv6=False)
         iface.update_mdns_entry(json)
         self.assertThat(MDNS.objects.count(), Equals(1))
-        self.assertThat(list(MDNS.objects.all())[0].ip, Equals(json['ip']))
+        self.assertThat(MDNS.objects.first().ip, Equals(json['address']))
         # This is the first time we saw this neighbour, because the original
         # binding was deleted.
-        self.assertThat(list(MDNS.objects.all())[0].count, Equals(1))
+        self.assertThat(MDNS.objects.count(), Equals(1))
 
     def test__logs_new_entry(self):
         iface = factory.make_Interface(INTERFACE_TYPE.PHYSICAL)
@@ -1010,7 +1010,7 @@
         with FakeLogger("maas.interface") as maaslog:
             iface.update_mdns_entry(json)
         self.assertDocTestMatches(
-            "...: New mDNS hostname observed for...",
+            "...: New mDNS entry resolved...",
             maaslog.output)
 
     def test__logs_moved_entry(self):
@@ -1019,7 +1019,7 @@
         json = self.make_mdns_entry_json()
         iface.update_mdns_entry(json)
         # Have a different IP address claim ownership of the hostma,e.
-        json['ip'] = factory.make_ip_address(ipv6=False)
+        json['address'] = factory.make_ip_address(ipv6=False)
         with FakeLogger("maas.mDNS") as maaslog:
             iface.update_mdns_entry(json)
         self.assertDocTestMatches(

=== modified file 'src/maasserver/regiondservices/networks_monitoring.py'
--- src/maasserver/regiondservices/networks_monitoring.py	2016-08-18 14:47:07 +0000
+++ src/maasserver/regiondservices/networks_monitoring.py	2016-08-31 18:00:29 +0000
@@ -27,6 +27,10 @@
         """Record the specified list of neighbours."""
         return deferToDatabase(self.recordNeighboursIntoDatabase, neighbours)
 
+    def reportMDNSEntries(self, mdns):
+        """Record the specified list of mDNS entries."""
+        return deferToDatabase(self.recordMDNSEntriesIntoDatabase, mdns)
+
     @transactional
     def recordInterfacesIntoDatabase(self, interfaces):
         """Record the interfaces information."""
@@ -38,3 +42,9 @@
         """Record the interfaces information."""
         region_controller = RegionController.objects.get_running_controller()
         region_controller.report_neighbours(neighbours)
+
+    @transactional
+    def recordMDNSEntriesIntoDatabase(self, mdns):
+        """Record the mDNS entries."""
+        region_controller = RegionController.objects.get_running_controller()
+        region_controller.report_mdns_entries(mdns)

=== modified file 'src/maasserver/rpc/rackcontrollers.py'
--- src/maasserver/rpc/rackcontrollers.py	2016-08-19 13:31:57 +0000
+++ src/maasserver/rpc/rackcontrollers.py	2016-08-31 18:00:29 +0000
@@ -194,6 +194,18 @@
 
 @synchronous
 @transactional
+def report_mdns_entries(system_id, mdns):
+    """Report observed neighbours seen on the rack controller."""
+    try:
+        rack_controller = RackController.objects.get(system_id=system_id)
+    except RackController.DoesNotExist:
+        raise NoSuchNode.from_system_id(system_id)
+    else:
+        rack_controller.report_mdns_entries(mdns)
+
+
+@synchronous
+@transactional
 def report_neighbours(system_id, neighbours):
     """Report observed neighbours seen on the rack controller."""
     try:

=== modified file 'src/maasserver/rpc/regionservice.py'
--- src/maasserver/rpc/regionservice.py	2016-08-19 18:11:11 +0000
+++ src/maasserver/rpc/regionservice.py	2016-08-31 18:00:29 +0000
@@ -408,6 +408,18 @@
         d.addCallback(lambda args: {})
         return d
 
+    @region.ReportMDNSEntries.responder
+    def report_mdns_entries(self, system_id, mdns):
+        """report_neighbours()
+
+        Implementation of
+        :py:class:`~provisioningserver.rpc.region.ReportNeighbours`.
+        """
+        d = deferToDatabase(
+            rackcontrollers.report_mdns_entries, system_id, mdns)
+        d.addCallback(lambda args: {})
+        return d
+
     @region.ReportNeighbours.responder
     def report_neighbours(self, system_id, neighbours):
         """report_neighbours()

=== modified file 'src/provisioningserver/rackdservices/networks_monitoring_service.py'
--- src/provisioningserver/rackdservices/networks_monitoring_service.py	2016-08-18 14:47:07 +0000
+++ src/provisioningserver/rackdservices/networks_monitoring_service.py	2016-08-31 18:00:29 +0000
@@ -9,6 +9,7 @@
 
 from provisioningserver.logger.log import get_maas_logger
 from provisioningserver.rpc.region import (
+    ReportMDNSEntries,
     ReportNeighbours,
     RequestRackRefresh,
     UpdateInterfaces,
@@ -27,7 +28,7 @@
         self.clientService = clientService
 
     def recordInterfaces(self, interfaces):
-        """Record the interfaces information."""
+        """Record the interfaces information to the region."""
         client = self.clientService.getClient()
         # On first run perform a refresh
         if self._recorded is None:
@@ -38,8 +39,14 @@
                 interfaces=interfaces)
 
     def reportNeighbours(self, neighbours):
-        """Report neighbour information."""
+        """Report neighbour information to the region."""
         client = self.clientService.getClient()
         return client(
             ReportNeighbours, system_id=client.localIdent,
             neighbours=neighbours)
+
+    def reportMDNSEntries(self, mdns):
+        """Report mDNS entries to the region."""
+        client = self.clientService.getClient()
+        return client(
+            ReportMDNSEntries, system_id=client.localIdent, mdns=mdns)

=== modified file 'src/provisioningserver/rpc/region.py'
--- src/provisioningserver/rpc/region.py	2016-08-19 18:11:11 +0000
+++ src/provisioningserver/rpc/region.py	2016-08-31 18:00:29 +0000
@@ -23,6 +23,7 @@
     "RegisterRackController",
     "ReportBootImages",
     "ReportForeignDHCPServer",
+    "ReportMDNSEntries",
     "ReportNeighbours",
     "RequestNodeInfoByMACAddress",
     "SendEvent",
@@ -367,6 +368,22 @@
     errors = []
 
 
+class ReportMDNSEntries(amp.Command):
+    """Called by a rack controller to report observed mDNS entries.
+
+    :since: 2.1
+    """
+
+    arguments = [
+        (b'system_id', amp.Unicode()),
+        (b'mdns', StructureAsJSON()),
+    ]
+    response = []
+    errors = {
+        NoSuchNode: b"NoSuchNode",
+    }
+
+
 class ReportNeighbours(amp.Command):
     """Called by a rack controller to report observed neighbor devices.
 

=== modified file 'src/provisioningserver/utils/services.py'
--- src/provisioningserver/utils/services.py	2016-08-19 15:34:46 +0000
+++ src/provisioningserver/utils/services.py	2016-08-31 18:00:29 +0000
@@ -13,6 +13,7 @@
 )
 from datetime import timedelta
 import json
+from json.decoder import JSONDecodeError
 import os
 
 from provisioningserver.config import is_dev_environment
@@ -34,35 +35,27 @@
     Deferred,
     maybeDeferred,
 )
-from twisted.internet.error import (
-    ProcessDone,
-    ProcessExitedAlready,
-)
-from twisted.internet.interfaces import ILoggingContext
+from twisted.internet.error import ProcessDone
 from twisted.internet.protocol import ProcessProtocol
 from twisted.internet.threads import deferToThread
 from twisted.python import log
-from zope.interface import implementer
 
 
 maaslog = get_maas_logger("networks.monitor")
 
 
-@implementer(ILoggingContext)
-class NeighbourObservationProtocol(ProcessProtocol):
+class JSONPerLineProtocol(ProcessProtocol):
+    """ProcessProtocol which allows easy parsing of a single JSON object per
+    line of text.
+    """
 
-    def __init__(self, interface, prefix=None, callback=None):
+    def __init__(self, callback=None):
         super().__init__()
         self.done = Deferred()
-        self.interface = interface
-        self.prefix = prefix
         self.callback = callback
         self.outbuf = b''
         self.errbuf = b''
 
-    def logPrefix(self):
-        return "-" if self.prefix is None else self.prefix
-
     def connectionMade(self):
         self.outbuf = b''
         self.errbuf = b''
@@ -93,10 +86,14 @@
 
     def outLineReceived(self, line):
         line = line.decode("utf-8")
-        obj = json.loads(line)
-        obj['interface'] = self.interface
-        # Report one observation at a time for now. (The API is designed to
-        # support multiple, but that will be done in a future branch.)
+        try:
+            obj = json.loads(line)
+        except JSONDecodeError:
+            log.msg("Failed to parse JSON: %r" % line)
+        else:
+            self.objectReceived(obj)
+
+    def objectReceived(self, obj):
         self.callback([obj])
 
     def errLineReceived(self, line):
@@ -112,52 +109,87 @@
             self.done.errback(reason)
 
 
-class NeighbourDiscoveryService(TimerService):
-    """Service to spawn the per-interface device discovery subprocess."""
-
-    def __init__(self, ifname):
-        super().__init__(60.0, self.observeInterface, ifname)
-        self._ifname = ifname
-        self._process = None
+class NeighbourObservationProtocol(JSONPerLineProtocol):
+
+    def __init__(self, interface, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.interface = interface
+
+    def objectReceived(self, obj):
+        # The only difference between the JSONPerLineProtocol and the
+        # NeighbourObservationProtocol is that the neighbour observation
+        # protocol needs to insert the interface metadata into the resultant
+        # object before the callback.
+        obj['interface'] = self.interface
+        super().objectReceived(obj)
+
+
+class ProcessProtocolService(TimerService, metaclass=ABCMeta):
+
+    def __init__(self, description, protocol, interval=60.0):
+        assert protocol is not None
+        assert description is not None
+        self.description = description
+        self.protocol = protocol
+        self.process = None
+        super().__init__(interval, self.startProcess)
 
     @deferred
-    def observeInterface(self, ifname):
-        """Start a network observation process for the specified interface."""
-        observer_protocol = NeighbourObservationProtocol(
-            ifname, callback=self.parent.reportNeighbours)
+    def startProcess(self):
         env = select_c_utf8_bytes_locale()
-        maas_rack_cmd = get_maas_provision_command().encode("utf-8")
-        self._process = reactor.spawnProcess(
-            observer_protocol, maas_rack_cmd,
-            [maas_rack_cmd, b"observe-arp", ifname.encode("utf-8")],
-            env=env)
-        log.msg("Started neighbour observation on interface: %s" % ifname)
-        return observer_protocol.done.addErrback(
-            log.err,
-            "Neighbour observation process failed for interface: %s" % (
-                self._ifname
-            ))
-
-    def stopObservationProcess(self):
-        """Attempts to kill the neighbour observation process gracefully."""
-        if self._process is not None:
-            try:
-                self._process.signalProcess("INT")
-            except ProcessExitedAlready:
-                pass  # Our work here is done.
-            except OSError:
-                log.err(
-                    None,
-                    "Neighbour observation process for %s failed to "
-                    "shut down." % self._ifname)
-            else:
-                return self._process.done
+        log.msg("%s started." % self.description)
+        args = self.getProcessParameters()
+        self.process = reactor.spawnProcess(
+            self.protocol, args[0], args, env=env)
+        return self.protocol.done.addErrback(
+            log.err, "%s failed." % self.description)
+
+    @abstractmethod
+    def getProcessParameters(self):
+        """Return the parameters for the subprocess to launch.
+
+        This MUST be overridden in subclasses.
+        """
 
     def stopService(self):
         """Stops the neighbour observation service."""
-        d = super().stopService()
-        d.addCallback(callOut, self.stopObservationProcess)
-        return d
+        if self.process is not None:
+            self.process.loseConnection()
+        return super().stopService()
+
+
+class NeighbourDiscoveryService(ProcessProtocolService):
+    """Service to spawn the per-interface device discovery subprocess."""
+
+    def __init__(self, ifname: str, callback):
+        self.ifname = ifname
+        description = "Neighbour observation process for %s" % ifname
+        protocol = NeighbourObservationProtocol(ifname, callback=callback)
+        super().__init__(description=description, protocol=protocol)
+
+    def getProcessParameters(self):
+        maas_rack_cmd = get_maas_provision_command().encode("utf-8")
+        return [
+            maas_rack_cmd,
+            b"observe-arp",
+            self.ifname.encode("utf-8")
+        ]
+
+
+class MDNSResolverService(ProcessProtocolService):
+    """Service to spawn the per-interface device discovery subprocess."""
+
+    def __init__(self, callback):
+        protocol = JSONPerLineProtocol(callback=callback)
+        super().__init__(
+            description="mDNS resolver process", protocol=protocol)
+
+    def getProcessParameters(self):
+        maas_rack_cmd = get_maas_provision_command().encode("utf-8")
+        return [
+            maas_rack_cmd,
+            b"observe-mdns",
+        ]
 
 
 class NetworksMonitoringService(MultiService, metaclass=ABCMeta):
@@ -231,6 +263,13 @@
         This MUST be overridden in subclasses.
         """
 
+    @abstractmethod
+    def reportMDNSEntries(self, mdns):
+        """Report on new or refreshed neighbours.
+
+        This MUST be overridden in subclasses.
+        """
+
     def stopService(self):
         """Stop the service.
 
@@ -307,10 +346,32 @@
 
     def _startNeighbourDiscovery(self, ifname):
         """"Start neighbour discovery service on the specified interface."""
-        service = NeighbourDiscoveryService(ifname)
+        service = NeighbourDiscoveryService(ifname, self.reportNeighbours)
         service.setName("neighbour_discovery:" + ifname)
         service.setServiceParent(self)
 
+    def _startMDNSDiscoveryService(self):
+        """Start resolving mDNS entries on attached networks."""
+        try:
+            self.getServiceNamed("mdns_resolver")
+        except KeyError:
+            # This is an expected exception. (The call inside the `try`
+            # is only necessary to ensure the service doesn't exist.)
+            service = MDNSResolverService(self.reportMDNSEntries)
+            service.setName("mdns_resolver")
+            service.setServiceParent(self)
+
+    def _stopMDNSDiscoveryService(self):
+        """Stop resolving mDNS entries on attached networks."""
+        try:
+            service = self.getServiceNamed("mdns_resolver")
+        except KeyError:
+            # Service doesn't exist, so no need to stop it.
+            pass
+        else:
+            service.disownServiceParent()
+            maaslog.info("Stopped mDNS resolver service.")
+
     def _startNeighbourDiscoveryServices(self, new_interfaces):
         """Start monitoring services for the specified set of interfaces."""
         for ifname in new_interfaces:
@@ -319,7 +380,7 @@
                 self.getServiceNamed("neighbour_disovery:" + ifname)
             except KeyError:
                 # This is an expected exception. (The call inside the `try`
-                # is only necessary to ensure the service doesn't exist.
+                # is only necessary to ensure the service doesn't exist.)
                 self._startNeighbourDiscovery(ifname)
 
     def _stopNeighbourDiscoveryServices(self, deleted_interfaces):
@@ -356,6 +417,22 @@
         deleted_interfaces = self._monitored.difference(monitored_interfaces)
         self._startNeighbourDiscoveryServices(new_interfaces)
         self._stopNeighbourDiscoveryServices(deleted_interfaces)
+        # Determine if we went from monitoring zero interfaces to monitoring
+        # at least one. If so, we need to start mDNS discovery.
+        # XXX Need to get this setting from the region.
+        if len(self._monitored) == 0 and len(monitored_interfaces) > 0:
+            # We weren't currently monitoring any interfaces, but we have been
+            # requested to monitor at least one.
+            self._startMDNSDiscoveryService()
+        elif len(self._monitored) > 0 and len(monitored_interfaces) == 0:
+            # We are currently monitoring at least one interface, but we have
+            # been requested to stop monitoring them all.
+            self._stopMDNSDiscoveryService()
+        else:
+            # No state change. We either still AREN'T monitoring any
+            # interfaces, or we still ARE monitoring them. (Either way, it
+            # doesn't matter for mDNS discovery purposes.)
+            pass
         self._monitored = monitored_interfaces
 
     def _interfacesRecorded(self, interfaces):

=== modified file 'src/provisioningserver/utils/tests/test_services.py'
--- src/provisioningserver/utils/tests/test_services.py	2016-08-20 12:42:22 +0000
+++ src/provisioningserver/utils/tests/test_services.py	2016-08-31 18:00:29 +0000
@@ -8,6 +8,7 @@
 import threading
 from unittest.mock import (
     call,
+    Mock,
     sentinel,
 )
 
@@ -15,6 +16,7 @@
 from maastesting.matchers import (
     DocTestMatches,
     HasLength,
+    IsFiredDeferred,
     MockCalledOnceWith,
     MockCallsMatch,
     MockNotCalled,
@@ -25,12 +27,23 @@
 )
 from maastesting.twisted import TwistedLoggerFixture
 from provisioningserver.utils import services
-from provisioningserver.utils.services import NetworksMonitoringService
+from provisioningserver.utils.services import (
+    JSONPerLineProtocol,
+    MDNSResolverService,
+    NeighbourDiscoveryService,
+    NeighbourObservationProtocol,
+    NetworksMonitoringService,
+    ProcessProtocolService,
+)
+from provisioningserver.utils.twisted import pause
+from testtools import ExpectedException
 from testtools.matchers import (
     Equals,
+    Is,
     IsInstance,
     Not,
 )
+from testtools.tests.twistedsupport.test_deferred import extract_result
 from twisted.application.service import MultiService
 from twisted.internet import reactor
 from twisted.internet.defer import (
@@ -38,7 +51,13 @@
     inlineCallbacks,
     succeed,
 )
+from twisted.internet.error import (
+    ProcessDone,
+    ProcessExitedAlready,
+    ProcessTerminated,
+)
 from twisted.python import threadable
+from twisted.python.failure import Failure
 
 
 class StubNetworksMonitoringService(NetworksMonitoringService):
@@ -60,6 +79,9 @@
     def reportNeighbours(self, neighbours):
         pass
 
+    def reportMDNSEntries(self, neighbours):
+        pass
+
 
 class TestNetworksMonitoringService(MAASTestCase):
     """Tests of `NetworksMonitoringService`."""
@@ -253,3 +275,236 @@
         yield service.updateInterfaces()
         # ... interfaces ARE recorded.
         self.assertThat(service.interfaces, Not(Equals([])))
+
+
+class TestJSONPerLineProtocol(MAASTestCase):
+    """Tests for `JSONPerLineProtocol`."""
+
+    run_tests_with = MAASTwistedRunTest.make_factory(timeout=5)
+
+    @inlineCallbacks
+    def test__propagates_exit_errors(self):
+        proto = JSONPerLineProtocol(callback=lambda json: None)
+        reactor.spawnProcess(proto, b"false", (b"false",))
+        with ExpectedException(ProcessTerminated, ".* exit code 1"):
+            yield proto.done
+
+    def test__parses_only_full_lines(self):
+        callback = Mock()
+        proto = JSONPerLineProtocol(callback=callback)
+        proto.connectionMade()
+        # Send an empty JSON dictionary using 3 separate writes.
+        proto.outReceived(b"{")
+        # No callback yet...
+        self.expectThat(callback, MockCallsMatch())
+        proto.outReceived(b"}")
+        # Still no callback...
+        self.expectThat(callback, MockCallsMatch())
+        proto.outReceived(b"\n")
+        # After a newline, we expect the JSON to be parsed and the callback
+        # to receive an empty Python dictionary (which corresponds to the JSON
+        # that was sent.)
+        self.expectThat(callback, MockCallsMatch(call([{}])))
+
+    def test__ignores_interspersed_zero_length_writes(self):
+        callback = Mock()
+        proto = JSONPerLineProtocol(callback=callback)
+        proto.connectionMade()
+        proto.outReceived(b"")
+        self.expectThat(callback, MockCallsMatch())
+        proto.outReceived(b"{}\n")
+        self.expectThat(callback, MockCallsMatch(call([{}])))
+        proto.outReceived(b"")
+        self.expectThat(callback, MockCallsMatch(call([{}])))
+        proto.outReceived(b"{}\n")
+        self.expectThat(callback, MockCallsMatch(call([{}]), call([{}])))
+
+    def test__logs_non_json_output(self):
+        callback = Mock()
+        proto = JSONPerLineProtocol(callback=callback)
+        proto.connectionMade()
+        with TwistedLoggerFixture() as logger:
+            proto.outReceived(b"{\n")
+        self.assertThat(
+            logger.output, DocTestMatches("Failed to parse JSON: ..."))
+
+    def test__logs_stderr(self):
+        message = factory.make_name("message")
+        callback = Mock()
+        proto = JSONPerLineProtocol(callback=callback)
+        proto.connectionMade()
+        with TwistedLoggerFixture() as logger:
+            proto.errReceived((message + "\n").encode("ascii"))
+        self.assertThat(logger.output, Equals(message))
+
+    def test__logs_only_full_lines_from_stderr(self):
+        message = factory.make_name("message")
+        callback = Mock()
+        proto = JSONPerLineProtocol(callback=callback)
+        proto.connectionMade()
+        with TwistedLoggerFixture() as logger:
+            proto.errReceived(message.encode("ascii"))
+        self.assertThat(logger.output, Equals(""))
+
+    def test__logs_stderr_at_process_end(self):
+        message = factory.make_name("message")
+        callback = Mock()
+        proto = JSONPerLineProtocol(callback=callback)
+        proto.connectionMade()
+        with TwistedLoggerFixture() as logger:
+            proto.errReceived(message.encode("ascii"))
+            self.assertThat(logger.output, Equals(""))
+            proto.processEnded(Failure(ProcessDone(0)))
+        self.assertThat(logger.output, Equals(message))
+
+    def test__propagates_errors_from_command(self):
+        callback = Mock()
+        proto = JSONPerLineProtocol(callback=callback)
+        proto.connectionMade()
+        reason = Failure(ProcessTerminated(1))
+        proto.processEnded(reason)
+        self.assertRaises(ProcessTerminated, extract_result, proto.done)
+
+
+class TestNeighbourObservationProtocol(MAASTestCase):
+    """Tests for `NeighbourObservationProtocol`."""
+
+    run_tests_with = MAASTwistedRunTest.make_factory(timeout=5)
+
+    def test_adds_interface(self):
+        callback = Mock()
+        ifname = factory.make_name('eth')
+        proto = NeighbourObservationProtocol(ifname, callback=callback)
+        proto.connectionMade()
+        proto.outReceived(b"{}\n")
+        self.expectThat(
+            callback, MockCallsMatch(call([{"interface": ifname}])))
+
+
+class TrueProcessProtocolService(ProcessProtocolService):
+
+    def getProcessParameters(self):
+        return ["/bin/true"]
+
+
+class FalseProcessProtocolService(ProcessProtocolService):
+
+    def getProcessParameters(self):
+        return ["/bin/false"]
+
+
+class CatProcessProtocolService(ProcessProtocolService):
+
+    def getProcessParameters(self):
+        return ["/bin/cat"]
+
+
+class EchoProcessProtocolService(ProcessProtocolService):
+
+    def getProcessParameters(self):
+        return ["/bin/echo", "{}\n"]
+
+
+class MockJSONProtocol(JSONPerLineProtocol):
+    pass
+
+
+class TestProcessProtocolService(MAASTestCase):
+    """Tests for `JSONPerLineProtocol`."""
+
+    run_tests_with = MAASTwistedRunTest.make_factory(debug=True, timeout=5)
+
+    def test__base_class_cannot_be_used(self):
+        with ExpectedException(TypeError):
+            ProcessProtocolService(
+                description="Mock process", protocol=Mock())
+
+    @inlineCallbacks
+    def test__starts_and_stops_process(self):
+        protocol = MockJSONProtocol()
+        service = CatProcessProtocolService(
+            description="Unit test process", protocol=protocol)
+        mock_callback = Mock()
+        protocol.done.addCallback(mock_callback)
+        service.startService()
+        yield service.stopService()
+        self.assertTrue(mock_callback.called)
+        yield pause(0.0)
+        self.assertThat(protocol.done, IsFiredDeferred())
+        self.assertThat(extract_result(protocol.done), Is(None))
+        with ExpectedException(ProcessExitedAlready):
+            service.process.signalProcess("INT")
+
+    @inlineCallbacks
+    def test__handles_normal_process_exit(self):
+        protocol = MockJSONProtocol()
+        service = TrueProcessProtocolService(
+            description="Unit test process", protocol=protocol)
+        mock_callback = Mock()
+        protocol.done.addCallback(mock_callback)
+        service.startService()
+        yield service.stopService()
+        self.assertTrue(mock_callback.called)
+        yield pause(0.0)
+        self.assertThat(protocol.done, IsFiredDeferred())
+        self.assertThat(extract_result(protocol.done), Is(None))
+
+    @inlineCallbacks
+    def test__handles_abnormal_process_exit(self):
+        protocol = MockJSONProtocol()
+        service = FalseProcessProtocolService(
+            description="Unit test process", protocol=protocol)
+        mock_errback = Mock()
+        mock_callback = Mock()
+        protocol.done.addCallback(mock_callback)
+        protocol.done.addErrback(mock_errback)
+        service.startService()
+        yield service.stopService()
+        self.assertTrue(mock_errback.called)
+        self.assertFalse(mock_callback.called)
+        yield pause(0.0)
+        self.assertThat(protocol.done, IsFiredDeferred())
+        self.assertThat(extract_result(protocol.done), Equals(None))
+
+    @inlineCallbacks
+    def test__calls_protocol_callback(self):
+        callback = Mock()
+        protocol = MockJSONProtocol(callback=callback)
+        service = EchoProcessProtocolService(
+            description="Unit test process", protocol=protocol)
+        service.startService()
+        # Wait for the protocol to finish. (the echo process will stop)
+        yield protocol.done
+        self.assertThat(callback, MockCalledOnceWith([{}]))
+        yield service.stopService()
+        yield pause(0.0)
+        self.assertThat(protocol.done, IsFiredDeferred())
+        self.assertThat(extract_result(protocol.done), Equals(None))
+
+
+class TestNeighbourDiscoveryService(MAASTestCase):
+    """Tests for `NeighbourDiscoveryService`."""
+
+    run_tests_with = MAASTwistedRunTest.make_factory(timeout=5)
+
+    def test__returns_expected_arguments(self):
+        ifname = factory.make_name('eth')
+        service = NeighbourDiscoveryService(ifname, Mock())
+        args = service.getProcessParameters()
+        self.assertThat(args, HasLength(3))
+        self.assertTrue(args[0].endswith(b'maas-rack'))
+        self.assertTrue(args[1], Equals(b"observe-arp"))
+        self.assertTrue(args[2], Equals(ifname.encode('utf-8')))
+
+
+class TestMDNSResolverService(MAASTestCase):
+    """Tests for `MDNSResolverService`."""
+
+    run_tests_with = MAASTwistedRunTest.make_factory(timeout=5)
+
+    def test__returns_expected_arguments(self):
+        service = MDNSResolverService(Mock())
+        args = service.getProcessParameters()
+        self.assertThat(args, HasLength(2))
+        self.assertTrue(args[0].endswith(b"maas-rack"))
+        self.assertTrue(args[1], Equals(b"observe-mdns"))