Merge lp:~vila/udd/no-shell-true into lp:udd

[Vincent Ladeuil]

As mentioned in previous submissions, using shell=True with subprocesses led to complications for handling kills.

In preparation for a next submission that will allow more than import_package.py to be invoked, I also want to get rid of shell=True for security concerns.

I also added a test to ensure that we can still capture the output/error of a python subprocess if it catch interrupting signals and handled them.

Note that we can't reliably guarantee what return code we will get when a process is killed and I've stopped cheating in this regard ;)

[James Westby (james-w)]

94 + cmd = 'import time; while True: time.sleep(0.001)'

Should this have a timeout?

Other than that this looks good.

Thanks,

James

review: Approve

[John A Meinel (jameinel)]

-----BEGIN PGP SIGNED MESSAGE-----
Hash: SHA1

On 2/23/2011 10:01 AM, James Westby wrote:
> Review: Approve
> 94 + cmd = 'import time; while True: time.sleep(0.001)'
>
> Should this have a timeout?
>
> Other than that this looks good.
>
> Thanks,
>
> James
>

It intentionally doesn't, because it is testing the kill code. But you
have a point, setting it to:

for i in xrange(10000): time.sleep(0.001)

would give us 10s of a process running, and not leave something forever
if the test fails for some reason.

John
=:->

-----BEGIN PGP SIGNATURE-----
Version: GnuPG v1.4.9 (Cygwin)
Comment: Using GnuPG with Mozilla - http://enigmail.mozdev.org/

iEYEARECAAYFAk1lPsMACgkQJdeBCYSNAAN9ewCfZ5GCUl4wYCdZuOvXbHb147+7
BckAoLeQrQTvbLHmtIPzFF/z4LQNo2kW
=a/qV
-----END PGP SIGNATURE-----

[Vincent Ladeuil (vila)]

> -----BEGIN PGP SIGNED MESSAGE-----
> Hash: SHA1
>
> On 2/23/2011 10:01 AM, James Westby wrote:
> > Review: Approve
> > 94 + cmd = 'import time; while True: time.sleep(0.001)'
> >
> > Should this have a timeout?
> It intentionally doesn't, because it is testing the kill code.

Yup.

> But you have a point, setting it to:

Yup.

There are addCleanup(t.kill) that *should* kill them but...

[lp:~vila/udd/no-shell-true]

435. By Vincent Ladeuil on 2011-02-23

Merge 589523-import-timeout into no-shell-true

436. By Vincent Ladeuil on 2011-02-23

Merge 589523-import-timeout into no-shell-true

437. By Vincent Ladeuil on 2011-02-23

Merge 589523-import-timeout into no-shell-true

438. By Vincent Ladeuil on 2011-02-23

Make sure the sleepers die in the end (still peaceful, they sleep ;)

439. By Vincent Ladeuil on 2011-02-24

Oops, we don't rely on the shell to find executables now and we rely on the traceback being available to categorize failures.

440. By Vincent Ladeuil on 2011-02-24

Tests requiring commits are a pita when it comes to typos.

441. By Vincent Ladeuil on 2011-02-24

Fix bad parsing of max_threads files, die untested obvious fixes, die !

=== modified file 'icommon.py'
--- icommon.py	2011-02-24 13:53:11 +0000
+++ icommon.py	2011-02-24 13:53:11 +0000
@@ -1940,10 +1940,10 @@
 
 class SubprocessMonitor(CatchingExceptionThread):
 
-    def __init__(self, cmd, max_duration=None):
+    def __init__(self, args, max_duration=None):
         """Create a SubprocessMonitor.
 
-        :param cmd: The command string to be run in the subprocess.
+        :param args: A list of arguments to give to python.
 
         :param max_duration: The max number of seconds the command is expected
             to be run (elapsed time).
@@ -1952,26 +1952,20 @@
         self.stopped = threading.Event()
         super(SubprocessMonitor, self).__init__(target=self.spawn,
                                                 sync_event=self.started)
-        self.cmd = cmd
+        self.args = args
         self.proc_pid = None
         self.retcode = None
+        self.out = self.err = None
         self.started_at = self.ended_at = None
         self.max_duration = max_duration
 
     def spawn(self):
-        """Spawn the command in a subprocess."""
-        def child_setup():
-            # python installs a SIGPIPE handler by default and signal
-            # dispositions set to SIG_IGN (used in that case) are preserved
-            # across fork and exceve. We don't want to propagate this to
-            # non-python subprocesses.
-            signal.signal(signal.SIGPIPE, signal.SIG_DFL)
-        proc = subprocess.Popen(self.cmd,
-                                shell=True,
+        """Spawn the python command in a subprocess."""
+        proc = subprocess.Popen(['python'] + self.args,
+                                executable=sys.executable,
                                 stdout=subprocess.PIPE,
                                 stderr=subprocess.PIPE,
-                                stdin=subprocess.PIPE,
-                                preexec_fn=child_setup)
+                                stdin=subprocess.PIPE)
         self.proc_pid = proc.pid
         self.started_at = time.time()
         self.switch_and_set(self.stopped)
@@ -1981,9 +1975,8 @@
             self.retcode = proc.returncode
         except OSError, e:
             if e.errno in (errno.ECHILD, errno.ESRCH):
-                # The process doesn't exist anymore, pretend is has been
-                # terminated
-                self.retcode = -signal.SIGTERM
+                # The process doesn't exist anymore
+                pass
             else:
                 raise
         self.stopped.set()
@@ -2011,7 +2004,6 @@
                 if signum is None:
                     signum = signal.SIGTERM
                 os.kill(self.proc_pid, signum)
-                self.retcode = -signum
                 killed = True
             except OSError, e:
                 if e.errno in (errno.ECHILD, errno.ESRCH):
 
=== modified file 'mass_import.py'
--- mass_import.py	2011-02-24 13:53:11 +0000
+++ mass_import.py	2011-02-24 13:53:11 +0000
@@ -103,6 +103,9 @@
 logger.addHandler(progress_handler)
 
 logger.info("Starting up")
+# Some context to ease debug
+logger.info("PATH: %s" % os.environ.get('PATH', None))
+logger.info("CWD: %s" % os.getcwd())
 
 
 logger.debug("Getting Launchpad object")
@@ -115,13 +118,13 @@
 
 class Importer(icommon.SubprocessMonitor):
 
-    def __init__(self, cmd, package_name, job_id):
+    def __init__(self, args, package_name, job_id):
         # FIXME: We should put the max_duration in a config file when it
         # becomes available. Until then, during the wheezy import the longest
         # import took more than 2 hours, so 24 is generous enough. On the other
         # hand, we had evidence of imports hanging for weeks (at least a month
         # even in one case, so 24 hours will get rid of them)
-        super(Importer, self).__init__(cmd % (package_name,),
+        super(Importer, self).__init__(args,
                                        # No more than a day
                                        max_duration=24*60*60)
         self.package_name = package_name
@@ -133,7 +136,9 @@
         if self.retcode == icommon.no_lock_returncode:
             logger.info("Couldn't lock %s, skipping" % self.package_name)
         else:
-            unicode_output = self.out.decode("utf-8", "replace")
+            # We mostly care about self.err which contains the traceback here
+            output = self.out + self.err
+            unicode_output = output.decode("utf-8", "replace")
             ascii_output = unicode_output.encode("ascii", "replace")
             success = self.retcode == 0
             if success:
@@ -174,7 +179,8 @@
         except Queue.Empty:
             self.unfinished_tasks += 1
             job_id, package_name = self.next_job()
-        return Importer('import_package.py %s',
+        return Importer(['scripts/import_package.py',
+                         package_name],
                         package_name, job_id)
 
     def next_job(self):
@@ -268,10 +274,13 @@
                 # Then check the graceful stop file
                 if os.path.exists(icommon.stop_file):
                     self.driver.queue_closed.set()
+                    logger.info('Driver would not process new requests')
                     continue
                 # And the number of threads
                 nb_threads = self.get_max_threads()
                 if self.driver.max_threads != nb_threads:
+                    logger.info('Read %d in %s'
+                                % (nb_threads, icommon.max_threads_file))
                     self.driver.max_threads = nb_threads
             # Catch driver exception if any
             self.report_driver_exception()
@@ -286,10 +295,12 @@
         if os.path.exists(icommon.max_threads_file):
             f = open(icommon.max_threads_file)
             try:
-                nb_threads = int(f.readlines()[0].strip())
+                line = f.readline().strip()
+                if line:
+                   nb_threads = int(line)
             except Exception, e:
-                logger.warning('Error reading max threads file %s: %s',
-                               (icommon.max_threads_file, str(e)))
+                logger.warning('Error reading max threads file %s: %s'
+                               % (icommon.max_threads_file, str(e)))
             finally:
                 f.close()
         return nb_threads
 
=== modified file 'tests.py'
--- tests.py	2011-02-24 13:53:11 +0000
+++ tests.py	2011-02-24 13:53:11 +0000
@@ -381,50 +381,56 @@
 
 class TestSubprocessMonitor(tests.TestCase):
 
-    def run_in_subprocess(self, cmd, klass=None):
+    def run_in_subprocess(self, args, klass=None):
         if klass is None:
             klass = icommon.SubprocessMonitor
-        sub = klass(cmd)
+        sub = klass(args)
         self.addCleanup(sub.join, 0)
         sub.start()
         return sub
 
+    def run_sleeper_in_subprocess(self, klass=None):
+        """Run a sleeper in a subprocess.
+
+        This is used to simulate a never-ending process without consuming too
+        much CPU cycles.
+        """
+        cmd = 'import time; while True: time.sleep(0.001)'
+        return self.run_in_subprocess(['-c', cmd], klass=klass)
+
     def test_broken_command(self):
-        sub = self.run_in_subprocess('this-is-not-a-program')
+        sub = self.run_in_subprocess(['this-is-not-a-python-script'])
         sub.stopped.wait()
         self.assertEquals(True, sub.started.isSet())
-        self.assertEquals('/bin/sh: this-is-not-a-program: not found\n',
-                          sub.err)
+        self.assertEquals(
+            "python: can't open file 'this-is-not-a-python-script':"
+            " [Errno 2] No such file or directory\n",
+            sub.err)
 
     def test_success(self):
-        sub = self.run_in_subprocess('echo toto')
+        sub = self.run_in_subprocess(['-c', 'print "toto"'])
         sub.stopped.wait()
         self.assertEquals('toto\n', sub.out)
         self.assertEquals('', sub.err)
         self.assertEquals(0, sub.retcode)
 
     def test_failure(self):
-        sub = self.run_in_subprocess('false')
+        sub = self.run_in_subprocess(['-c', 'import sys; sys.exit(1)'])
         sub.stopped.wait()
         self.assertEquals('', sub.out)
         self.assertEquals('', sub.err)
         self.assertEquals(1, sub.retcode)
 
     def test_kill(self):
-        sub = self.run_in_subprocess('yes')
+        sub = self.run_sleeper_in_subprocess()
         sub.started.wait()
         self.assertEquals(True, sub.kill())
         sub.stopped.wait()
         self.assertEquals(-signal.SIGTERM, sub.retcode)
 
     def test_kill_SIGKILL(self):
-        sub = self.run_in_subprocess('yes')
+        sub = self.run_sleeper_in_subprocess()
         sub.started.wait()
-        # For oscure reasons, using SIGABRT instead of SIGKILL and stopping
-        # below left the 'yes' process alive and the SubprocessMonitor happily
-        # consuming memroy (8GB mark reached while sitting under pdb, which is
-        # good enough to consider bug #589532 addressed.
-        # import pdb; pdb.set_trace()
         self.assertEquals(True, sub.kill(signal.SIGKILL))
         sub.stopped.wait()
         self.assertEquals(-signal.SIGKILL, sub.retcode)
@@ -432,17 +438,17 @@
     def test_kill_catch_zombie(self):
         kill_me = threading.Event()
         resume = threading.Event()
-        class TestSubprocessMonitor(icommon.SubprocessMonitor):
+        class TestMonitor(icommon.SubprocessMonitor):
 
             def switch_and_set(self, new):
-                super(TestSubprocessMonitor, self).switch_and_set(new)
+                super(TestMonitor, self).switch_and_set(new)
                 if new is self.stopped:
                     # The process is running but we haven't called
                     # proc.communicate yet.
                     kill_me.set()
                     resume.wait()
 
-        sub = self.run_in_subprocess('yes', TestSubprocessMonitor)
+        sub = self.run_sleeper_in_subprocess(klass=TestMonitor)
         self.addCleanup(sub.join, 0)
         sub.started.wait()
         kill_me.wait()
@@ -450,25 +456,90 @@
         os.kill(sub.proc_pid, signal.SIGTERM)
         pid, st = os.waitpid(sub.proc_pid, 0)
         self.assertEquals(sub.proc_pid, pid)
+        # Only the low byte contains the signal and the high bit if for core
+        # dumps.
+        self.assertEquals(signal.SIGTERM, st & 0x7F)
         # Release the thread so it can call proc.communicate
         resume.set()
         # Now pretend we don't know the subprocess is dead
         self.assertEquals(False, sub.kill())
         sub.join()
-        self.assertEquals(-signal.SIGTERM, sub.retcode)
-
+        # No useful retcode available
+        self.assertEquals(None, sub.retcode)
+        # Nothing either in stdout/stderr. This means that subprocesses should
+        # handled SIGTERM to allow us to collect their data.
+        self.assertIs(None, sub.out)
+        self.assertIs(None, sub.err)
+
+
+class TestSubprocessMonitorWithScript(tests.TestCaseInTempDir):
+
+    def run_in_subprocess(self, args, klass=None):
+        if klass is None:
+            klass = icommon.SubprocessMonitor
+        sub = klass(args)
+        self.addCleanup(sub.join, 0)
+        sub.start()
+        return sub
+
+    def test_kill_caught(self):
+        self.build_tree_contents([('test.py',
+                                   """\
+class Killed(Exception):
+
+    def __str__(self):
+
+        return "I'm so dead"
+
+def die(signum, frame):
+    raise Killed
+import signal
+signal.signal(signal.SIGTERM, die)
+f = open('ready_to_die', 'w')
+try:
+    f.write('Yes I am\\n')
+finally:
+    f.close()
+import time
+slice = 0.001 # seconds
+max = 120.0 # seconds
+# We want to die long after the test fail (it it succeeds we're dead already)
+while xrange(max/slice):
+    time.sleep(0.001)
+""")])
+        sub = self.run_in_subprocess(['test.py'])
+        sub.started.wait()
+        # Gross hack to ensure the script ran long enough to catch the signal
+        ready = False
+        while not ready:
+            try:
+                f = open('ready_to_die')
+                ready = True
+            except:
+                pass
+        self.assertEquals(True, sub.kill())
+        sub.stopped.wait()
+        # The signal has been caught, we just get a failure return code, not
+        # funky stuff with the signal in the low byte and so on
+        self.assertEquals(1, sub.retcode)
+        # after the split, the last line is empty
+        self.assertEndsWith(sub.err.split('\n')[-2], "I'm so dead")
 
 class Sleeper(icommon.SubprocessMonitor):
 
     def __init__(self, sleep_time=0.001, max_duration=None):
         # sleep can't be killed (or interrupted) so we don't sleep for long but
         # keep sleeping until killed
-        super(Sleeper, self).__init__('while true ; do sleep %f ; done'
-                                      % (sleep_time,),
-                                      max_duration=max_duration)
-
-
-class TestDriverFeatures(tests.TestCase):
+        super(Sleeper, self).__init__(
+            # Make sure we die long after the test finish (to avoid leaks) but
+            # still use a value that allow *some* debug (you want more for
+            # *heavy* debug :)
+            ['-c', 'import time; while xrange(120.0/%s): time.sleep(%s)'
+             % (sleep_time, sleep_time)],
+            max_duration=max_duration)
+
+
+class TestThreadDriver(tests.TestCase):
 
     def get_driver(self, *args, **kwargs):
         klass = kwargs.pop('klass', None)
@@ -557,10 +628,9 @@
 
             def collect(thread):
                 self.assertIs(self.thread, thread)
-                # false returns 1
                 self.assertEquals(1, thread.retcode)
                 self.terminated.set()
-        self.thread = TestThread('false')
+        self.thread = TestThread(['-c', 'import sys; sys.exit(1)'])
         self.addCleanup(self.thread.join, 0)
         q = Queue.Queue(1)
         q.put(self.thread)
@@ -583,7 +653,8 @@
         self.addCleanup(driver.stop.set)
         # Wait for the subprocess thread to start
         self.started.wait()
-        # the thread is waiting but not yet deleted from the driver
+        # the thread is started and at most wait for collection on
+        # self.terminated so it's still there
         self.assertLength(1, driver.threads)
         self.assertEquals(self.thread, driver.threads[0])
         self.resumed.set()
@@ -596,11 +667,10 @@
         class TestThread(icommon.SubprocessMonitor):
 
             def collect(thread):
-                # false returns 1
                 self.assertEquals(1, thread.retcode)
                 raise ExceptionForTests()
         q = Queue.Queue(1)
-        t = TestThread('false')
+        t = TestThread(['-c', 'import sys; sys.exit(1)'])
         self.addCleanup(t.join, 0)
         q.put(t)
         driver = self.get_started_driver(queue=q, max_threads=1)