Merge lp:~abentley/launchpad/simplify-twisted-runner-3 into lp:launchpad

[Aaron Bentley]

Inject sys.path into subprocesses.

= Summary =
Fix issues invoking TwistedJobRunner

== Proposed fix ==
Inject the current sys.path into subprocesses as PYTHONPATH

== Pre-implementation notes ==
None

== Implementation details ==
This builds on previous work toward simplifying the TwistedJobRunner.

It fixes issues finding and launching the JobRunnerProcess class by injecting the current sys.path into subprocesses as PYTHONPATH.

It also simplifies the code further by using the inclineCallbacks decorator.

It also improves handling of process startup failures by handling the case when result is None.

It also outputs 'No jobs to run.' if no jobs were found, to clarify output.

It also raises 'Could not acquire lease...' to an INFO message, so that it's at the same level as 'Running ...' messages.

== Tests ==
bin/test test_runner

== Demo and Q/A ==
Create a branch and run cronscripts/scan_branches.py on qastaging. It should complete successfully.

= Launchpad lint =

Checking for conflicts and issues in changed files.

Linting changed files:
  lib/lp/services/job/tests/test_runner.py
  lib/lp/services/job/runner.py

[Brad Crittenden (bac)]

Looks great Aaron.

review: Approve (code)

=== modified file 'lib/lp/services/job/runner.py'
--- lib/lp/services/job/runner.py	2011-09-09 20:05:32 +0000
+++ lib/lp/services/job/runner.py	2011-09-09 20:05:36 +0000
@@ -43,10 +43,14 @@
     reactor,
     )
 from twisted.internet.defer import (
+    inlineCallbacks,
     succeed,
     )
 from twisted.protocols import amp
-from twisted.python import log
+from twisted.python import (
+    failure,
+    log,
+    )
 from zope.component import getUtility
 from zope.security.proxy import removeSecurityProxy
 
@@ -183,7 +187,7 @@
         try:
             job.acquireLease()
         except LeaseHeld:
-            self.logger.debug(
+            self.logger.info(
                 'Could not acquire lease for %s' % self.job_str(job))
             self.incomplete_jobs.append(job)
             return False
@@ -406,11 +410,10 @@
 
     def __init__(self, job_source, dbuser, logger=None, error_utility=None):
         env = {'PATH': os.environ['PATH']}
-        for name in ('PYTHONPATH', 'LPCONFIG'):
-            if name in os.environ:
-                env[name] = os.environ[name]
-        starter = main.ProcessStarter(
-            packages=('_pythonpath', 'twisted', 'ampoule'), env=env)
+        if 'LPCONFIG' in os.environ:
+            env['LPCONFIG'] = os.environ['LPCONFIG']
+        env['PYTHONPATH'] = os.pathsep.join(sys.path)
+        starter = main.ProcessStarter(env=env)
         super(TwistedJobRunner, self).__init__(logger, error_utility)
         self.job_source = job_source
         self.import_name = '%s.%s' % (
@@ -441,6 +444,10 @@
             RunJobCommand, job_id=job_id, _deadline=deadline)
 
         def update(response):
+            if response is None:
+                self.incomplete_jobs.append(job)
+                self.logger.debug('No response for %r', job)
+                return
             if response['success']:
                 self.completed_jobs.append(job)
                 self.logger.debug('Finished %r', job)
@@ -475,18 +482,20 @@
             oops = self._doOops(job, sys.exc_info())
             self._logOopsId(oops['id'])
 
+    @inlineCallbacks
     def runAll(self):
         """Run all ready jobs."""
         self.pool.start()
         try:
-            jobs = list(self.job_source.iterReady())
-            if len(jobs) == 0:
+            try:
+                job = None
+                for job in self.job_source.iterReady():
+                    yield self.runJobInSubprocess(job)
+                if job is None:
+                    self.logger.info('No jobs to run.')
                 self.terminated()
-                return
-            d = self.runJobInSubprocess(jobs[0])
-            for job in jobs[1:]:
-                d.addCallback(lambda ignored: self.runJobInSubprocess(job))
-            d.addCallbacks(self.terminated, self.failed)
+            except:
+                self.failed(failure.Failure())
         except:
             self.terminated()
             raise