Merge lp:~allenap/maas/terminate-process into lp:~maas-committers/maas/trunk

[Gavin Panella]

New function terminateProcess that helps with clean termination of processes spawned via Twisted.

Also includes a new process protocol, ProcessGroupLeaderMixin, which helps ensure that spawned processes cooperate with the termination protocol defined by terminateProcess.

[Mike Pontillo (mpontillo)]

This looks nice. I'm going to take a closer look.

[Mike Pontillo (mpontillo)]

I didn't get a chance to dig into this branch as much as I wanted to, but the code looks good and I tested it some last week. So I'll go ahead and approve. Thanks for the help with this, Gavin.

review: Approve

[Gavin Panella (allenap)]

Thanks!

=== modified file 'src/maasserver/static/js/angular/services/pollingmanager.js'
--- src/maasserver/static/js/angular/services/pollingmanager.js	2016-09-22 15:05:38 +0000
+++ src/maasserver/static/js/angular/services/pollingmanager.js	2016-09-26 06:36:24 +0000
@@ -1,4 +1,4 @@
-/* Copyright 2015-2016 Canonical Ltd.  This software is licensed under the
+/* Copyright 2016 Canonical Ltd.  This software is licensed under the
  * GNU Affero General Public License version 3 (see the file LICENSE).
  *
  * MAAS Space Manager

=== modified file 'src/maasserver/tests/test_auth.py'
--- src/maasserver/tests/test_auth.py	2016-09-22 02:53:33 +0000
+++ src/maasserver/tests/test_auth.py	2016-09-26 06:36:24 +0000
@@ -1,4 +1,4 @@
-# Copyright 2012-2015 Canonical Ltd.  This software is licensed under the
+# Copyright 2012-2016 Canonical Ltd.  This software is licensed under the
 # GNU Affero General Public License version 3 (see the file LICENSE).
 
 """Test permissions."""

=== modified file 'src/maasserver/views/__init__.py'
--- src/maasserver/views/__init__.py	2016-09-23 16:22:10 +0000
+++ src/maasserver/views/__init__.py	2016-09-26 06:36:24 +0000
@@ -1,4 +1,4 @@
-# Copyright 2012-2015 Canonical Ltd.  This software is licensed under the
+# Copyright 2012-2016 Canonical Ltd.  This software is licensed under the
 # GNU Affero General Public License version 3 (see the file LICENSE).
 
 """Views."""

=== modified file 'src/maasserver/views/prefs.py'
--- src/maasserver/views/prefs.py	2016-09-22 13:16:12 +0000
+++ src/maasserver/views/prefs.py	2016-09-26 06:36:24 +0000
@@ -1,4 +1,4 @@
-# Copyright 2012-2015 Canonical Ltd.  This software is licensed under the
+# Copyright 2012-2016 Canonical Ltd.  This software is licensed under the
 # GNU Affero General Public License version 3 (see the file LICENSE).
 
 """Preferences views."""

=== modified file 'src/maasserver/views/tests/test_account.py'
--- src/maasserver/views/tests/test_account.py	2016-09-22 02:53:33 +0000
+++ src/maasserver/views/tests/test_account.py	2016-09-26 06:36:24 +0000
@@ -1,4 +1,4 @@
-# Copyright 2012-2015 Canonical Ltd.  This software is licensed under the
+# Copyright 2012-2016 Canonical Ltd.  This software is licensed under the
 # GNU Affero General Public License version 3 (see the file LICENSE).
 
 """Test maasserver account views."""

=== modified file 'src/maasserver/views/tests/test_prefs.py'
--- src/maasserver/views/tests/test_prefs.py	2016-09-22 13:16:12 +0000
+++ src/maasserver/views/tests/test_prefs.py	2016-09-26 06:36:24 +0000
@@ -1,4 +1,4 @@
-# Copyright 2012-2015 Canonical Ltd.  This software is licensed under the
+# Copyright 2012-2016 Canonical Ltd.  This software is licensed under the
 # GNU Affero General Public License version 3 (see the file LICENSE).
 
 """Test maasserver preferences views."""

=== modified file 'src/maasserver/websockets/handlers/user.py'
--- src/maasserver/websockets/handlers/user.py	2016-09-22 02:53:33 +0000
+++ src/maasserver/websockets/handlers/user.py	2016-09-26 06:36:24 +0000
@@ -1,4 +1,4 @@
-# Copyright 2015 Canonical Ltd.  This software is licensed under the
+# Copyright 2015-2016 Canonical Ltd.  This software is licensed under the
 # GNU Affero General Public License version 3 (see the file LICENSE).
 
 """The user handler for the WebSocket connection."""

=== modified file 'src/provisioningserver/utils/tests/test_twisted.py'
--- src/provisioningserver/utils/tests/test_twisted.py	2016-09-17 06:15:12 +0000
+++ src/provisioningserver/utils/tests/test_twisted.py	2016-09-26 06:36:24 +0000
@@ -6,15 +6,21 @@
 __all__ = []
 
 from functools import partial
+import io
 from itertools import cycle
 import operator
 from operator import attrgetter
+import os
 from random import (
     randint,
     random,
+    randrange,
 )
 import re
+import signal
+import sys
 import threading
+from unittest import mock
 from unittest.mock import (
     ANY,
     Mock,
@@ -29,6 +35,8 @@
     IsFiredDeferred,
     IsUnfiredDeferred,
     MockCalledOnceWith,
+    MockCallsMatch,
+    MockNotCalled,
     Provides,
 )
 from maastesting.testcase import (
@@ -56,13 +64,16 @@
     LONGTIME,
     makeDeferredWithProcessProtocol,
     pause,
+    ProcessGroupLeaderMixin,
     retries,
     RPCFetcher,
     synchronous,
+    terminateProcess,
     ThreadPool,
     ThreadPoolLimiter,
     ThreadUnpool,
 )
+from testtools.content import content_from_stream
 from testtools.deferredruntest import assert_fails_with
 from testtools.matchers import (
     AfterPreprocessing,
@@ -88,7 +99,11 @@
     inlineCallbacks,
     succeed,
 )
-from twisted.internet.error import ProcessDone
+from twisted.internet.error import (
+    ProcessDone,
+    ProcessTerminated,
+)
+from twisted.internet.protocol import ProcessProtocol
 from twisted.internet.task import Clock
 from twisted.internet.threads import (
     deferToThread,
@@ -1677,3 +1692,163 @@
         protocol.processEnded(Failure(exception))
         with ExpectedException(type(exception)):
             yield d
+
+
+# A script that prints the signals it receives, as long as they're SIGTERM or
+# SIGQUIT. It prints "Ready." on stderr once it's registered signal handlers.
+signal_printer = """\
+from signal import Signals, signal
+from sys import stderr
+from time import sleep
+from traceback import print_exc
+
+def print_signal(signum, frame):
+    print(Signals(signum).name, flush=True)
+
+signal(Signals.SIGTERM, print_signal)
+signal(Signals.SIGQUIT, print_signal)
+
+print("Ready.", file=stderr, flush=True)
+
+sleep(5.0)
+"""
+
+
+class SignalPrinterProtocol(ProcessProtocol):
+    """Process protocol for use with `signal_printer`."""
+
+    def __init__(self):
+        super(SignalPrinterProtocol, self).__init__()
+        self.ready = Deferred()
+        self.done = Deferred()
+        self.out = io.BytesIO()
+        self.err = io.BytesIO()
+
+    def outReceived(self, data):
+        self.out.write(data)
+
+    def errReceived(self, data):
+        self.err.write(data)
+        if not self.ready.called:
+            self.ready.callback(data)
+
+    def processEnded(self, reason):
+        self.done.callback(reason)
+
+
+class SignalPrinterProtocolAsGroupLeader(
+        ProcessGroupLeaderMixin, SignalPrinterProtocol):
+    """Process protocol for use with `signal_printer` that ...
+
+    ...configures itself as a process group leader.
+    """
+
+
+class TestTerminateProcess(MAASTestCase):
+    """Tests for `terminateProcess`."""
+
+    run_tests_with = MAASTwistedRunTest.make_factory(timeout=10)
+
+    def setUp(self):
+        super(TestTerminateProcess, self).setUp()
+        self.sigprint = self.make_file("sigprint.py", signal_printer)
+        # Allow spying on calls to os.kill and os.killpg by terminateProcess.
+        self.assertThat(twisted_module._os_kill, Is(os.kill))
+        self.patch(
+            twisted_module, "_os_kill",
+            Mock(wraps=twisted_module._os_kill))
+        self.assertThat(twisted_module._os_killpg, Is(os.killpg))
+        self.patch(
+            twisted_module, "_os_killpg",
+            Mock(wraps=twisted_module._os_killpg))
+
+    def startSignalPrinter(self, protocol):
+        self.assertThat(protocol, IsInstance(SignalPrinterProtocol))
+        self.addDetail("out", content_from_stream(protocol.out, seek_offset=0))
+        self.addDetail("err", content_from_stream(protocol.err, seek_offset=0))
+        python = sys.executable.encode("utf-8")
+        process = reactor.spawnProcess(
+            protocol, python, (python, self.sigprint.encode("utf-8")))
+
+        # Wait for the spawned subprocess to tell us that it's ready.
+        def cbReady(message):
+            self.assertThat(message.decode("utf-8"), Equals("Ready.\n"))
+            return process
+
+        return protocol.ready.addCallback(cbReady)
+
+    def terminateSignalPrinter(self, process, protocol):
+        # Terminate with some short timings; no point waiting long in a test,
+        # and we need to do it before the subprocess finishes sleeping.
+        terminateProcess(
+            process.pid, protocol.done, quit_after=0.1,
+            kill_after=0.2)
+
+    @inlineCallbacks
+    def test__terminates_process_with_TERM_QUIT_then_KILL(self):
+        protocol = SignalPrinterProtocol()
+        process = yield self.startSignalPrinter(protocol)
+        self.terminateSignalPrinter(process, protocol)
+        # The subprocess is terminated with SIGKILL but received SIGTERM then
+        # SIGQUIT prior to that.
+        try:
+            yield protocol.done
+        except ProcessTerminated as reason:
+            self.assertThat(reason, IsInstance(ProcessTerminated))
+            self.assertThat(reason.signal, Equals(signal.SIGKILL))
+            self.assertThat(
+                protocol.out.getvalue().decode("utf-8").split(),
+                Equals(["SIGTERM", "SIGQUIT"]))
+
+    @inlineCallbacks
+    def test__terminates_with_kill_and_killpg(self):
+        protocol = SignalPrinterProtocolAsGroupLeader()
+        process = yield self.startSignalPrinter(protocol)
+        # Capture the pid now; it gets cleared when the process exits.
+        pid = process.pid
+        # Terminate and wait for it to exit.
+        self.terminateSignalPrinter(process, protocol)
+        yield protocol.done.addErrback(Failure.trap, ProcessTerminated)
+        # os.kill was called once then os.killpg was called twice because the
+        # subprocess made itself a process group leader.
+        self.assertThat(
+            twisted_module._os_kill, MockCallsMatch(
+                mock.call(pid, signal.SIGTERM),
+            ))
+        self.assertThat(
+            twisted_module._os_killpg, MockCallsMatch(
+                mock.call(pid, signal.SIGQUIT),
+                mock.call(pid, signal.SIGKILL),
+            ))
+
+    @inlineCallbacks
+    def test__terminates_with_kill_if_not_in_separate_process_group(self):
+        protocol = SignalPrinterProtocol()
+        process = yield self.startSignalPrinter(protocol)
+        # Capture the pid now; it gets cleared when the process exits.
+        pid = process.pid
+        # Terminate and wait for it to exit.
+        self.terminateSignalPrinter(process, protocol)
+        yield protocol.done.addErrback(Failure.trap, ProcessTerminated)
+        # os.kill was called 3 times because the subprocess did not make
+        # itself a process group leader.
+        self.assertThat(
+            twisted_module._os_kill, MockCallsMatch(
+                mock.call(pid, signal.SIGTERM),
+                mock.call(pid, signal.SIGQUIT),
+                mock.call(pid, signal.SIGKILL),
+            ))
+        self.assertThat(
+            twisted_module._os_killpg, MockNotCalled())
+
+
+class TestProcessGroupLeaderMixin(MAASTestCase):
+    """Tests for `ProcessGroupLeaderMixin`."""
+
+    def test__calls_setpgid_on_child_process(self):
+        self.assertThat(twisted_module._os_setpgid, Is(os.setpgid))
+        setpgid = self.patch(twisted_module, "_os_setpgid")
+        protocol = ProcessGroupLeaderMixin()
+        transport = Mock(pid=randrange(99999, 9999999))
+        protocol.makeConnection(transport)
+        self.assertThat(setpgid, MockCalledOnceWith(transport.pid, 0))

=== modified file 'src/provisioningserver/utils/twisted.py'
--- src/provisioningserver/utils/twisted.py	2016-09-17 05:25:56 +0000
+++ src/provisioningserver/utils/twisted.py	2016-09-26 06:36:24 +0000
@@ -2,9 +2,6 @@
 # GNU Affero General Public License version 3 (see the file LICENSE).
 
 """Utilities related to the Twisted/Crochet execution environment."""
-from twisted.internet.error import ProcessDone
-from twisted.internet.protocol import ProcessProtocol
-
 
 __all__ = [
     'asynchronous',
@@ -23,6 +20,7 @@
     'LONGTIME',
     'makeDeferredWithProcessProtocol',
     'pause',
+    'ProcessGroupLeaderMixin',
     'retries',
     'synchronous',
     'ThreadPool',
@@ -40,6 +38,13 @@
 )
 from itertools import repeat
 from operator import attrgetter
+import os
+from os import (
+    kill as _os_kill,
+    killpg as _os_killpg,
+    setpgid as _os_setpgid,
+)
+import signal
 import threading
 
 from crochet import run_in_reactor
@@ -51,6 +56,8 @@
     maybeDeferred,
     succeed,
 )
+from twisted.internet.error import ProcessDone
+from twisted.internet.protocol import ProcessProtocol
 from twisted.internet.threads import deferToThread
 from twisted.logger import Logger
 from twisted.python import (
@@ -956,3 +963,74 @@
         done.errback(reason) if (reason and not reason.check(ProcessDone))
         else done.callback(None))
     return done, protocol
+
+
+def terminateProcess(pid, done, *, quit_after=5.0, kill_after=10.0):
+    """Terminate the given process.
+
+    A "sensible" way to terminate a process. Does the following:
+
+      1. Sends SIGTERM to the process identified by `pid`.
+      2. Waits for up to 5 seconds.
+      3. Sends SIGQUIT to the process *group* of process `pid`.
+      4. Waits for up to an additional 5 seconds.
+      5. Sends SIGKILL to the process *group* of process `pid`.
+
+    Steps #3 and #5 have a safeguard: if the process identified by `pid` has
+    the same process group as the invoking process the signal is sent only to
+    the process and not to the process group. This prevents the caller from
+    inadvertently killing itself. You may want to use this in conjunction with
+    `ProcessGroupLeaderMixin` to ensure that the child process becomes a
+    process group leader soon after spawning.
+
+    :param pid: The PID to terminate.
+    :param done: A `Deferred` that fires when the process exits.
+    """
+    ppgid = os.getpgrp()
+
+    def kill(sig):
+        """Attempt to send `signal` to the given `pid`."""
+        try:
+            _os_kill(pid, sig)
+        except ProcessLookupError:
+            pass  # Already exited.
+
+    def killpg(sig):
+        """Attempt to send `signal` to the progress group of `pid`.
+
+        If `pid` is running in the same process group as the invoking process,
+        this falls back to using kill(2) instead of killpg(2).
+        """
+        try:
+            pgid = os.getpgid(pid)
+            if pgid == ppgid:
+                _os_kill(pid, sig)
+            else:
+                _os_killpg(pgid, sig)
+        except ProcessLookupError:
+            pass  # Already exited.
+
+    killers = (
+        reactor.callLater(0.0, kill, signal.SIGTERM),
+        reactor.callLater(quit_after, killpg, signal.SIGQUIT),
+        reactor.callLater(kill_after, killpg, signal.SIGKILL),
+    )
+
+    def ended():
+        for killer in killers:
+            if killer.active():
+                killer.cancel()
+
+    done.addBoth(callOut, ended)
+
+
+class ProcessGroupLeaderMixin(ProcessProtocol):
+    """Mix-in to ensure that the spawned process is a process group leader."""
+
+    def connectionMade(self):
+        super().connectionMade()
+        if self.transport.pid is not None:
+            try:
+                _os_setpgid(self.transport.pid, 0)
+            except ProcessLookupError:
+                pass  # Already exited.