Merge lp:~mpontillo/maas/try-not-to-spawn-loads-of-observation-processes into lp:~maas-committers/maas/trunk

[Mike Pontillo]

Fix bug that caused a large number of observation processes to spawn in certain situations. (Especially when observation processes die, or interface configuration changes.)

 * Add unit test for runaway process issue.
 * Move responsibility for managing Deferred (used to indicate when the service has stopped) from the ProcessProtocol implementation to the ProcessProtocolService.
 * Refactor unit tests to account for the new design.
 * Fix a typo that could have caused reconfigured interfaces to fail to spawn.

I've been testing this end-to-end all day, and LaMont verified (on an earlier version of this branch) that the code is working much better now.

This branch should ship with the next alpha to increase reliability. Without it, we can spawn processes every minute or so and eventually use up all the resources on a MAAS rack.

[Gavin Panella (allenap)]

All looks good. I have lots of small suggestions for simplifying the code, or to be more idiomatically Twisted, but I'll propose those as a follow-on instead of here, because this code is good to land as it is.

review: Approve

[Mike Pontillo (mpontillo)]

Thanks for taking a look, Gavin!

[Mike Pontillo (mpontillo)]

Setting this to depend on lp:~allenap/maas/drive-bys-2016-09-06 since there is a merge conflict.

[Mike Pontillo (mpontillo)]

Self-approving already-approved branch.

review: Approve

=== modified file 'src/provisioningserver/utils/services.py'
--- src/provisioningserver/utils/services.py	2016-09-12 17:50:38 +0000
+++ src/provisioningserver/utils/services.py	2016-09-12 17:50:38 +0000
@@ -35,7 +35,10 @@
     Deferred,
     maybeDeferred,
 )
-from twisted.internet.error import ProcessDone
+from twisted.internet.error import (
+    ProcessDone,
+    ProcessExitedAlready,
+)
 from twisted.internet.protocol import ProcessProtocol
 from twisted.internet.threads import deferToThread
 from twisted.python import log
@@ -51,22 +54,28 @@
 
     def __init__(self, callback=None):
         super().__init__()
-        self.done = Deferred()
-        self.callback = callback
-        self.outbuf = b''
-        self.errbuf = b''
+        if callback is None:
+            self._callback = lambda result: None
+        else:
+            self._callback = callback
+        self._done = lambda result: None
+        self._outbuf = b''
+        self._errbuf = b''
+
+    def setDoneCallback(self, done):
+        self._done = done
 
     def connectionMade(self):
-        self.outbuf = b''
-        self.errbuf = b''
+        self._outbuf = b''
+        self._errbuf = b''
 
     def outReceived(self, data):
-        lines, self.outbuf = self.splitLines(self.outbuf + data)
+        lines, self._outbuf = self.splitLines(self._outbuf + data)
         for line in lines:
             self.outLineReceived(line)
 
     def errReceived(self, data):
-        lines, self.errbuf = self.splitLines(self.errbuf + data)
+        lines, self._errbuf = self.splitLines(self._errbuf + data)
         for line in lines:
             self.errLineReceived(line)
 
@@ -94,19 +103,21 @@
             self.objectReceived(obj)
 
     def objectReceived(self, obj):
-        self.callback([obj])
+        self._callback([obj])
 
     def errLineReceived(self, line):
         line = line.decode("utf-8")
         log.msg(line.rstrip())
 
     def processEnded(self, reason):
-        if len(self.errbuf) != 0:
-            self.errLineReceived(self.errbuf)
+        if len(self._errbuf) != 0:
+            self.errLineReceived(self._errbuf)
+        # If the process finished normally, call the _done callback with
+        # None. Otherwise, pass the reason through.
         if reason.check(ProcessDone):
-            self.done.callback(None)
+            self._done(None)
         else:
-            self.done.errback(reason)
+            self._done(reason)
 
 
 class NeighbourObservationProtocol(JSONPerLineProtocol):
@@ -126,25 +137,37 @@
 
 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
+    def __init__(self, interval=60.0, reactor_process=None):
+        self.deferredProcessEnded = None
         self.process = None
+        if reactor_process is not None:
+            self.reactor_process = reactor_process
+        else:
+            self.reactor_process = reactor
         super().__init__(interval, self.startProcess)
 
     @deferred
     def startProcess(self):
         env = select_c_utf8_bytes_locale()
-        log.msg("%s started." % self.description)
+        log.msg("%s started." % self.getDescription())
         args = self.getProcessParameters()
+        self.deferredProcessEnded = Deferred()
+        protocol = self.createProcessProtocol()
+        protocol.setDoneCallback(self.processEnded)
         assert all(isinstance(arg, bytes) for arg in args), (
             "Process arguments must all be bytes, got: %s" % repr(args))
-        self.process = reactor.spawnProcess(
-            self.protocol, args[0], args, env=env)
-        return self.protocol.done.addErrback(
-            log.err, "%s failed." % self.description)
+        self.process = self.reactor_process.spawnProcess(
+            protocol, args[0], args, env=env)
+        return self.deferredProcessEnded
+
+    def processEnded(self, failure):
+        if failure is None:
+            log.msg("%s ended normally." % self.getDescription())
+        else:
+            log.err(failure, "%s failed." % self.getDescription())
+        # We don't need to call the errback, because we don't want the service
+        # to crash with an unhandled error.
+        self.deferredProcessEnded.callback(None)
 
     @abstractmethod
     def getProcessParameters(self):
@@ -153,21 +176,51 @@
         This MUST be overridden in subclasses.
         """
 
+    @abstractmethod
+    def getDescription(self):
+        """Return the description of this process, suitable to use in verbose
+        logging.
+
+        This MUST be overridden in subclasses.
+        """
+
+    @abstractmethod
+    def createProcessProtocol(self):
+        """
+        Creates and returns the ProcessProtocol that will be used to
+        communicate with the process.
+
+        This MUST be overridden in subclasses.
+        """
+
+    def stopProcess(self):
+        try:
+            self.process.signalProcess("INT")
+        except ProcessExitedAlready:
+            pass
+
     def stopService(self):
         """Stops the neighbour observation service."""
         if self.process is not None:
             self.process.loseConnection()
+            # We don't care about the result; we just want to make sure the
+            # process is dead.
+            # XXX this causes us to log errors such as:
+            #     mDNS resolver process failed.
+            # reactor.callFromThread(self.stopProcess)
         return super().stopService()
 
 
 class NeighbourDiscoveryService(ProcessProtocolService):
     """Service to spawn the per-interface device discovery subprocess."""
 
-    def __init__(self, ifname: str, callback):
+    def __init__(self, ifname: str, callback: callable):
         self.ifname = ifname
-        description = "Neighbour observation process for %s" % ifname
-        protocol = NeighbourObservationProtocol(ifname, callback=callback)
-        super().__init__(description=description, protocol=protocol)
+        self.callback = callback
+        super().__init__()
+
+    def getDescription(self) -> str:
+        return "Neighbour observation process for %s" % self.ifname
 
     def getProcessParameters(self):
         maas_rack_cmd = get_maas_provision_command().encode("utf-8")
@@ -177,14 +230,20 @@
             self.ifname.encode("utf-8")
         ]
 
+    def createProcessProtocol(self):
+        return NeighbourObservationProtocol(
+            self.ifname, callback=self.callback)
+
 
 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)
+        self.callback = callback
+        super().__init__()
+
+    def getDescription(self):
+        return "mDNS observation process"
 
     def getProcessParameters(self):
         maas_rack_cmd = get_maas_provision_command().encode("utf-8")
@@ -193,6 +252,9 @@
             b"observe-mdns",
         ]
 
+    def createProcessProtocol(self):
+        return JSONPerLineProtocol(callback=self.callback)
+
 
 class NetworksMonitoringService(MultiService, metaclass=ABCMeta):
     """Service to monitor network interfaces for configuration changes.
@@ -379,7 +441,7 @@
         for ifname in new_interfaces:
             # Sanity check to ensure the service isn't already started.
             try:
-                self.getServiceNamed("neighbour_disovery:" + ifname)
+                self.getServiceNamed("neighbour_discovery:" + ifname)
             except KeyError:
                 # This is an expected exception. (The call inside the `try`
                 # is only necessary to ensure the service doesn't exist.)
@@ -389,7 +451,7 @@
         """Stop monitoring services for the specified set of interfaces."""
         for ifname in deleted_interfaces:
             try:
-                service = self.getServiceNamed("neighbour_disovery:" + ifname)
+                service = self.getServiceNamed("neighbour_discovery:" + ifname)
             except KeyError:
                 # Service doesn't exist, so no need to stop it.
                 pass
 
=== modified file 'src/provisioningserver/utils/tests/test_services.py'
--- src/provisioningserver/utils/tests/test_services.py	2016-09-12 17:50:38 +0000
+++ src/provisioningserver/utils/tests/test_services.py	2016-09-12 17:50:38 +0000
@@ -47,6 +47,7 @@
 from twisted.application.service import MultiService
 from twisted.internet import reactor
 from twisted.internet.defer import (
+    Deferred,
     DeferredQueue,
     inlineCallbacks,
     succeed,
@@ -56,6 +57,7 @@
     ProcessExitedAlready,
     ProcessTerminated,
 )
+from twisted.internet.task import Clock
 from twisted.python import threadable
 from twisted.python.failure import Failure
 
@@ -282,12 +284,25 @@
 
     run_tests_with = MAASTwistedRunTest.make_factory(timeout=5)
 
+    def make_deferred_callback(self):
+        d = Deferred()
+
+        def callback(result):
+            if result is None:
+                d.callback(None)
+            else:
+                d.errback(result)
+
+        return d, callback
+
     @inlineCallbacks
     def test__propagates_exit_errors(self):
         proto = JSONPerLineProtocol(callback=lambda json: None)
+        d, callback = self.make_deferred_callback()
+        proto.setDoneCallback(callback)
         reactor.spawnProcess(proto, b"false", (b"false",))
         with ExpectedException(ProcessTerminated, ".* exit code 1"):
-            yield proto.done
+            yield d
 
     def test__parses_only_full_lines(self):
         callback = Mock()
@@ -357,13 +372,16 @@
             proto.processEnded(Failure(ProcessDone(0)))
         self.assertThat(logger.output, Equals(message))
 
+    @inlineCallbacks
     def test__propagates_errors_from_command(self):
-        callback = Mock()
-        proto = JSONPerLineProtocol(callback=callback)
+        d, callback = self.make_deferred_callback()
+        proto = JSONPerLineProtocol()
+        proto.setDoneCallback(callback)
         proto.connectionMade()
         reason = Failure(ProcessTerminated(1))
         proto.processEnded(reason)
-        self.assertRaises(ProcessTerminated, extract_result, proto.done)
+        with ExpectedException(ProcessTerminated):
+            yield d
 
 
 class TestNeighbourObservationProtocol(MAASTestCase):
@@ -381,25 +399,38 @@
             callback, MockCallsMatch(call([{"interface": ifname}])))
 
 
-class TrueProcessProtocolService(ProcessProtocolService):
+class MockProcessProtocolService(ProcessProtocolService):
+
+    def __init__(self):
+        super().__init__()
+        self._callback = Mock()
+
+    def getDescription(self):
+        return "%s mock ProcessProtocolService" % self.__class__.__name__
+
+    def createProcessProtocol(self):
+        return JSONPerLineProtocol(callback=self._callback)
+
+
+class TrueProcessProtocolService(MockProcessProtocolService):
 
     def getProcessParameters(self):
         return [b"/bin/true"]
 
 
-class FalseProcessProtocolService(ProcessProtocolService):
+class FalseProcessProtocolService(MockProcessProtocolService):
 
     def getProcessParameters(self):
         return [b"/bin/false"]
 
 
-class CatProcessProtocolService(ProcessProtocolService):
+class CatProcessProtocolService(MockProcessProtocolService):
 
     def getProcessParameters(self):
         return [b"/bin/cat"]
 
 
-class EchoProcessProtocolService(ProcessProtocolService):
+class EchoProcessProtocolService(MockProcessProtocolService):
 
     def getProcessParameters(self):
         return [b"/bin/echo", b"{}\n"]
@@ -416,70 +447,59 @@
 
     def test__base_class_cannot_be_used(self):
         with ExpectedException(TypeError):
-            ProcessProtocolService(
-                description="Mock process", protocol=Mock())
+            ProcessProtocolService()
 
     @inlineCallbacks
     def test__starts_and_stops_process(self):
-        protocol = MockJSONProtocol()
-        service = CatProcessProtocolService(
-            description="Unit test process", protocol=protocol)
+        service = CatProcessProtocolService()
         mock_callback = Mock()
-        protocol.done.addCallback(mock_callback)
         service.startService()
+        self.assertThat(service.deferredProcessEnded, Not(IsFiredDeferred()))
+        service.deferredProcessEnded.addCallback(mock_callback)
         yield service.stopService()
+        yield service.deferredProcessEnded
+        self.assertThat(service.deferredProcessEnded, IsFiredDeferred())
         self.assertTrue(mock_callback.called)
         yield pause(0.0)
-        self.assertThat(protocol.done, IsFiredDeferred())
-        self.assertThat(extract_result(protocol.done), Is(None))
+        self.assertThat(service.deferredProcessEnded, IsFiredDeferred())
+        self.assertThat(extract_result(service.deferredProcessEnded), 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)
+        service = TrueProcessProtocolService()
         mock_callback = Mock()
-        protocol.done.addCallback(mock_callback)
         service.startService()
+        service.deferredProcessEnded.addCallback(mock_callback)
         yield service.stopService()
+        yield service.deferredProcessEnded
         self.assertTrue(mock_callback.called)
         yield pause(0.0)
-        self.assertThat(protocol.done, IsFiredDeferred())
-        self.assertThat(extract_result(protocol.done), Is(None))
+        self.assertThat(service.deferredProcessEnded, IsFiredDeferred())
+        self.assertThat(extract_result(service.deferredProcessEnded), 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))
+        service = FalseProcessProtocolService()
+        with TwistedLoggerFixture() as logger:
+            service.startService()
+            d = service.deferredProcessEnded
+            yield service.stopService()
+            self.assertThat(d, IsFiredDeferred())
+        self.assertThat(logger.output, DocTestMatches("... failed..."))
 
     @inlineCallbacks
     def test__calls_protocol_callback(self):
-        callback = Mock()
-        protocol = MockJSONProtocol(callback=callback)
-        service = EchoProcessProtocolService(
-            description="Unit test process", protocol=protocol)
+        service = EchoProcessProtocolService()
         service.startService()
         # Wait for the protocol to finish. (the echo process will stop)
-        yield protocol.done
-        self.assertThat(callback, MockCalledOnceWith([{}]))
+        d = service.deferredProcessEnded
+        result = yield d
+        self.assertThat(service._callback, MockCalledOnceWith([{}]))
         yield service.stopService()
-        yield pause(0.0)
-        self.assertThat(protocol.done, IsFiredDeferred())
-        self.assertThat(extract_result(protocol.done), Equals(None))
+        self.assertThat(d, IsFiredDeferred())
+        self.assertThat(result, Equals(None))
 
 
 class TestNeighbourDiscoveryService(MAASTestCase):
@@ -496,6 +516,26 @@
         self.assertTrue(args[1], Equals(b"observe-arp"))
         self.assertTrue(args[2], Equals(ifname.encode('utf-8')))
 
+    @inlineCallbacks
+    def test__restarts_process_after_finishing(self):
+        ifname = factory.make_name('eth')
+        service = NeighbourDiscoveryService(ifname, Mock())
+        mock_process_params = self.patch(service, 'getProcessParameters')
+        mock_process_params.return_value = [b'/bin/echo', b'{}']
+        service.clock = Clock()
+        service.startService()
+        # Wait for the protocol to finish
+        service.clock.advance(0.0)
+        yield service.deferredProcessEnded
+        self.assertThat(service.deferredProcessEnded, IsFiredDeferred())
+        # Advance the clock (should start the service again)
+        interval = service.step
+        service.clock.advance(interval)
+        # The Deferred should have been recreated.
+        self.assertThat(service.deferredProcessEnded, Not(IsFiredDeferred()))
+        yield service.deferredProcessEnded
+        service.stopService()
+
 
 class TestMDNSResolverService(MAASTestCase):
     """Tests for `MDNSResolverService`."""