Merge lp:~adeuring/launchpad/lp-lazr.jobrunner into lp:launchpad

[Abel Deuring]

use lazr.jobrunner for basic job management

This branch removes some core methods from class JobRunner, which
are now provided by lazr.jobrunner.

It requires lp:~adeuring/launchpad/lazr.jobrunner-more-tests (not yet
merged into lp:lazr.jobrunner/trunk)

lazr.jobrunner does not do everything which the old job runner did.
Most notably, the now removed method BaseJobRunner.runJob() contained
several calls of transaction.commit() and transaction.abort().

The new module lazr.jobrunner does make any assumption where or how
the status of a job is stored, or if a job makes any other changes
to a database, and hence does do any of these calls.

Instead, these calls are now done in Job.start(), Job.complete() etc.

Grepping through the Launchpad source code showed that some of these
methods are called in places that are not related to controlling a
job, but for example to create a job in the status "suspended".

Calling transaction.commit() unconditionally could thus cause unwanted
"intermediate" commits; to avoid this, I added the parameter
manage_transaction to Job.start() etc, which is False by default,
but the calls in lazr.jobrunner use manage_transaction=True. This
is somewhat ugly, but the alternative, keeping transaction management
in lazr.jobrunner, would also look odd: The module is supposed to do
job management, regardless of any possible database related activities
of a job.

test: ./bin/test -vvt lp.services.job

no lint

[Aaron Bentley (abentley)]

Not landable yet because it assumes lazr.jobrunner is a development egg. "../lazr.jobrunner" should not be listed in buildout.cfg, and a version of lazr.jobrunner should be specified in versions.cfg.

runJob may not need to be overridden in BaseJobRunner. If not, please remove it.

job_str may not need to be overridden in BaseJobRunner. If not, please remove it.

review: Needs Fixing

[Abel Deuring (adeuring)]

> Not landable yet because it assumes lazr.jobrunner is a development egg.
> "../lazr.jobrunner" should not be listed in buildout.cfg, and a version
> of lazr.jobrunner should be specified in versions.cfg.

Done.

>
> runJob may not need to be overridden in BaseJobRunner. If not, please
> remove it.

I think that we should keep this method: class runner.BaseRunnableJob
implements some required methods like notifyOops(), but it does not
implement start(), complete() etc. The latter methods are defined in
lp.services.job.model.job.Job -- and all of them are required by
lazr.jobrunner.runner.JobRunner. So, "IRunnableJob(job)" in

    def runJob(self, job):
        super(BaseJobRunner, self).runJob(IRunnableJob(job))

in BaseJobRunner ensures that the Zope machinery adapts the "incomplete"
job to IRunnableJob. OK, IRunnableJob(job) is also called in
JobRunner.runAll() and in TwistedJobRunner.runJobInSubprocess(), so it is
right now not strictly necessary in JobRunner.runJob() -- but keeping
JobRunner.runJob() makes it easier to remove runAll() later.

Alternatively, we could call IRunnableJob(job) in lazr.jobrunner, but I
am not sure if this makes sense.

> job_str may not need to be overridden in BaseJobRunner. If not, please
> remove it.

Right, this is no longer needed. Removed.

[Aaron Bentley (abentley)]

> > runJob may not need to be overridden in BaseJobRunner. If not, please
> > remove it.
>
> I think that we should keep this method: class runner.BaseRunnableJob
> implements some required methods like notifyOops(), but it does not
> implement start(), complete() etc. The latter methods are defined in
> lp.services.job.model.job.Job -- and all of them are required by
> lazr.jobrunner.runner.JobRunner. So, "IRunnableJob(job)" in
>
> def runJob(self, job):
> super(BaseJobRunner, self).runJob(IRunnableJob(job))
>
> in BaseJobRunner ensures that the Zope machinery adapts the "incomplete"
> job to IRunnableJob. OK, IRunnableJob(job) is also called in
> JobRunner.runAll() and in TwistedJobRunner.runJobInSubprocess(), so it is
> right now not strictly necessary in JobRunner.runJob() -- but keeping
> JobRunner.runJob() makes it easier to remove runAll() later.

I think this is fine for now. In the future, I think we might want to make the job source responsible for ensuring that the job is runnable, and have the runner assume the job is runnable.

> Alternatively, we could call IRunnableJob(job) in lazr.jobrunner, but I
> am not sure if this makes sense.

I'd like to keep zope.component out of lazr.jobrunner if we can.

review: Approve

[Abel Deuring (adeuring)]

Some additional changes are necessary to use lazr.jobrunner (see also https://code.launchpad.net/~adeuring/lazr.jobrunner/use_job_repr_in_logging/+merge/98821):

The change in lib/lp/registry/tests/test_process_job_sources_cronjob.py reflects the change from logger.debug() in the original BaseJobRunner.runJob() method to logger.info() in lazr.jobrunner.

The second change: I added the method Job._doOops() again to BasJobRunner. The reason:

The implementation of _doOops() in lazr.jobrunner calls oops_report.publish(). The makes sense because the module does not make any asssumptions about a setup for some sort automatic OOPS generation: It simply publishes the OOPS report and the logs an error message.

This conflicts with the way how Launchpad generates an OOPS in scripts: lp.services.scripts.logger has a class OopsHandler(logging.Handler), which automatically generates an OOPS report, when logger.error() is called. So, a call of _doOops() from lazr.jobrunner first publishes an OOPS, then logs the error -- and a log handler publishes the same OPS report again.

tests:

bin/test -vv \
-t lp.registry.tests.test_process_job_sources_cronjob.ProcessJobSourceGroupsTest.test_processed \
-t lp.registry.tests.test_process_job_sources_cronjob.ProcessJobSourceTest.test_processed \
-t lp.code.scripts.tests.test_create_merge_proposals.TestCreateMergeProposals.test_oops \
-t lp.code.scripts.tests.test_reclaim_branch_space.TestReclaimBranchSpaceScript.test_reclaimbranchspace_script_logs_oops \
-t lp.code.scripts.tests.test_sendbranchmail.TestSendbranchmail.test_sendbranchmail_handles_oops \
-t lp.translations.tests.test_rosetta_branches_script.TestRosettaBranchesScript.test_rosetta_branches_script_oops \
-t lp.registry.tests.test_membership_notification_job.MembershipNotificationJobTest.test_smoke_admining_team

no lint

[Abel Deuring (adeuring)]

review: Resubmit

[Aaron Bentley (abentley)]

I believe this would introduce a regression with database error handling. If there is a database error (like a permission failure) runJobHandleError will call Job.fail. But that will cause a database lookup. Which fails due to the previous error. e.g. http://pastebin.ubuntu.com/892304/

review: Needs Fixing

[Abel Deuring (adeuring)]

The problem you mentioned is fixed in lazr.jobrunner (including tests).

review: Resubmit

[Aaron Bentley (abentley)]

review: Approve

=== modified file 'lib/lp/code/scripts/tests/test_create_merge_proposals.py'
--- lib/lp/code/scripts/tests/test_create_merge_proposals.py	2012-01-01 02:58:52 +0000
+++ lib/lp/code/scripts/tests/test_create_merge_proposals.py	2012-03-28 11:45:27 +0000
@@ -36,11 +36,11 @@
         retcode, stdout, stderr = run_script(
             'cronscripts/create_merge_proposals.py', [])
         self.assertEqual(0, retcode)
-        self.assertEqual(
+        self.assertTextMatchesExpressionIgnoreWhitespace(
             "INFO    Creating lockfile: "
             "/var/lock/launchpad-create_merge_proposals.lock\n"
-            "INFO    "
-            "Running CreateMergeProposalJob (ID %d) in status Waiting\n"
+            "INFO    Running <.*CreateMergeProposalJob object at .*?> "
+            "\(ID %s\) in status Waiting\n"
             "INFO    Ran 1 CreateMergeProposalJobs.\n" % job.job.id, stderr)
         self.assertEqual('', stdout)
         self.assertEqual(1, source.landing_targets.count())

=== modified file 'lib/lp/code/scripts/tests/test_merge_proposal_jobs.py'
--- lib/lp/code/scripts/tests/test_merge_proposal_jobs.py	2012-01-01 02:58:52 +0000
+++ lib/lp/code/scripts/tests/test_merge_proposal_jobs.py	2012-03-28 11:45:27 +0000
@@ -40,7 +40,8 @@
             '\tworkers: 0\n'
             'INFO    \tworkers: 0\n'
             '(.|\n)*'
-            'INFO    Running GenerateIncrementalDiffJob \(ID %d\).\n'
+            'INFO    Running '
+            '<GENERATE_INCREMENTAL_DIFF job for merge .*?> \(ID %d\).\n'
             '(.|\n)*'
             'INFO    STOPPING: \'\'\n'
             'Main loop terminated.\n'

=== modified file 'lib/lp/code/scripts/tests/test_reclaim_branch_space.py'
--- lib/lp/code/scripts/tests/test_reclaim_branch_space.py	2012-01-01 02:58:52 +0000
+++ lib/lp/code/scripts/tests/test_reclaim_branch_space.py	2012-03-28 11:45:27 +0000
@@ -46,7 +46,8 @@
             'cronscripts/reclaimbranchspace.py', [])
         self.assertEqual('', stdout)
         self.assertEqual(
-            'INFO    Creating lockfile: /var/lock/launchpad-reclaimbranchspace.lock\n'
+            'INFO    '
+            'Creating lockfile: /var/lock/launchpad-reclaimbranchspace.lock\n'
             'INFO    Reclaimed space for 0 branches.\n', stderr)
         self.assertEqual(0, retcode)
         self.assertTrue(
@@ -63,16 +64,17 @@
         retcode, stdout, stderr = run_script(
             'cronscripts/reclaimbranchspace.py', [])
         self.assertEqual('', stdout)
-        self.assertEqual(
-            'INFO    Creating lockfile: /var/lock/launchpad-reclaimbranchspace.lock\n'
-            'INFO    Running ReclaimBranchSpaceJob (ID %d) in status Waiting\n'
+        self.assertTextMatchesExpressionIgnoreWhitespace(
+            'INFO    '
+            'Creating lockfile: /var/lock/launchpad-reclaimbranchspace.lock\n'
+            'INFO    Running <RECLAIM_BRANCH_SPACE branch job \(\d+\) for '
+            '\d+> \(ID %s\) in status Waiting\n'
             'INFO    Reclaimed space for 1 branches.\n' % reclaim_job.job.id,
             stderr)
         self.assertEqual(0, retcode)
         self.assertFalse(
             os.path.exists(mirrored_path))
 
-
     def test_reclaimbranchspace_script_logs_oops(self):
         # If the job fails, an oops is logged.
         db_branch = self.factory.makeAnyBranch()

=== modified file 'lib/lp/code/scripts/tests/test_sendbranchmail.py'
--- lib/lp/code/scripts/tests/test_sendbranchmail.py	2012-01-01 02:58:52 +0000
+++ lib/lp/code/scripts/tests/test_sendbranchmail.py	2012-03-28 11:45:27 +0000
@@ -50,9 +50,11 @@
         transaction.commit()
         retcode, stdout, stderr = run_script(
             'cronscripts/sendbranchmail.py', [])
-        self.assertEqual(
-            'INFO    Creating lockfile: /var/lock/launchpad-sendbranchmail.lock\n'
-            'INFO    Running RevisionMailJob (ID %d) in status Waiting\n'
+        self.assertTextMatchesExpressionIgnoreWhitespace(
+            'INFO    '
+            'Creating lockfile: /var/lock/launchpad-sendbranchmail.lock\n'
+            'INFO    Running <REVISION_MAIL branch job \(\d+\) for .*?> '
+            '\(ID %d\) in status Waiting\n'
             'INFO    Ran 1 RevisionMailJobs.\n' % mail_job.job.id, stderr)
         self.assertEqual('', stdout)
         self.assertEqual(0, retcode)
@@ -67,7 +69,8 @@
         retcode, stdout, stderr = run_script(
             'cronscripts/sendbranchmail.py', [])
         self.assertIn(
-            'INFO    Creating lockfile: /var/lock/launchpad-sendbranchmail.lock\n',
+            'INFO    '
+            'Creating lockfile: /var/lock/launchpad-sendbranchmail.lock\n',
             stderr)
         self.assertIn('INFO    Job resulted in OOPS:', stderr)
         self.assertIn('INFO    Ran 0 RevisionMailJobs.\n', stderr)
@@ -88,9 +91,11 @@
         transaction.commit()
         retcode, stdout, stderr = run_script(
             'cronscripts/sendbranchmail.py', [])
-        self.assertEqual(
-            'INFO    Creating lockfile: /var/lock/launchpad-sendbranchmail.lock\n'
-            'INFO    Running RevisionsAddedJob (ID %d) in status Waiting\n'
+        self.assertTextMatchesExpressionIgnoreWhitespace(
+            'INFO    '
+            'Creating lockfile: /var/lock/launchpad-sendbranchmail.lock\n'
+            'INFO    Running <REVISIONS_ADDED_MAIL branch job \(\d+\) '
+            'for .*?> \(ID %d\) in status Waiting\n'
             'INFO    Ran 1 RevisionMailJobs.\n' % job.job.id,
             stderr)
         self.assertEqual('', stdout)

=== modified file 'lib/lp/registry/tests/test_process_job_sources_cronjob.py'
--- lib/lp/registry/tests/test_process_job_sources_cronjob.py	2012-01-01 02:58:52 +0000
+++ lib/lp/registry/tests/test_process_job_sources_cronjob.py	2012-03-28 11:45:27 +0000
@@ -62,7 +62,7 @@
         returncode, output, error = run_script(
             self.script, ['-v', 'IMembershipNotificationJobSource'])
         self.assertIn(
-            ('DEBUG   Running <MembershipNotificationJob '
+            ('INFO    Running <MembershipNotificationJob '
              'about ~murdock in ~a-team; status=Waiting>'),
             error)
         self.assertIn('DEBUG   MembershipNotificationJob sent email', error)
@@ -126,7 +126,7 @@
         returncode, output, error = run_script(
             self.script, ['-v', '--wait', 'MAIN'])
         self.assertTextMatchesExpressionIgnoreWhitespace(
-            ('DEBUG Running <MembershipNotificationJob '
+            ('INFO Running <MembershipNotificationJob '
              'about ~murdock in ~a-team; status=Waiting>'),
             error)
         self.assertIn('DEBUG   MembershipNotificationJob sent email', error)

=== modified file 'lib/lp/services/job/interfaces/job.py'
--- lib/lp/services/job/interfaces/job.py	2011-12-24 16:54:44 +0000
+++ lib/lp/services/job/interfaces/job.py	2012-03-28 11:45:27 +0000
@@ -14,7 +14,6 @@
     'ITwistedJobSource',
     'JobStatus',
     'LeaseHeld',
-    'SuspendJobException',
     ]
 
 
@@ -39,11 +38,6 @@
 from lp.registry.interfaces.person import IPerson
 
 
-class SuspendJobException(Exception):
-    """Raised when a running job wants to suspend itself."""
-    pass
-
-
 class LeaseHeld(Exception):
     """Raised when attempting to acquire a list that is already held."""
 
@@ -88,6 +82,9 @@
 class IJob(Interface):
     """Basic attributes of a job."""
 
+    job_id = Int(title=_(
+        'A unique identifier for this job.'))
+
     scheduled_start = Datetime(
         title=_('Time when the IJob was scheduled to start.'))
 

=== modified file 'lib/lp/services/job/model/job.py'
--- lib/lp/services/job/model/job.py	2012-02-24 03:47:44 +0000
+++ lib/lp/services/job/model/job.py	2012-03-28 11:45:27 +0000
@@ -25,6 +25,7 @@
     Int,
     Reference,
     )
+import transaction
 from zope.interface import implements
 
 from lp.services.database import bulk
@@ -56,6 +57,10 @@
 
     implements(IJob)
 
+    @property
+    def job_id(self):
+        return self.id
+
     scheduled_start = UtcDateTimeCol()
 
     date_created = UtcDateTimeCol()
@@ -144,31 +149,49 @@
         expiry = timegm(self.lease_expires.timetuple())
         return max(0, expiry - time.time())
 
-    def start(self):
+    def start(self, manage_transaction=False):
         """See `IJob`."""
         self._set_status(JobStatus.RUNNING)
         self.date_started = datetime.datetime.now(UTC)
         self.date_finished = None
         self.attempt_count += 1
+        if manage_transaction:
+            transaction.commit()
 
-    def complete(self):
+    def complete(self, manage_transaction=False):
         """See `IJob`."""
+        # Commit the transaction to update the DB time.
+        if manage_transaction:
+            transaction.commit()
         self._set_status(JobStatus.COMPLETED)
         self.date_finished = datetime.datetime.now(UTC)
+        if manage_transaction:
+            transaction.commit()
 
-    def fail(self):
+    def fail(self, manage_transaction=False):
         """See `IJob`."""
+        if manage_transaction:
+            transaction.abort()
         self._set_status(JobStatus.FAILED)
         self.date_finished = datetime.datetime.now(UTC)
+        if manage_transaction:
+            transaction.commit()
 
-    def queue(self):
+    def queue(self, manage_transaction=False):
         """See `IJob`."""
+        # Commit the transaction to update the DB time.
+        if manage_transaction:
+            transaction.commit()
         self._set_status(JobStatus.WAITING)
         self.date_finished = datetime.datetime.now(UTC)
+        if manage_transaction:
+            transaction.commit()
 
-    def suspend(self):
+    def suspend(self, manage_transaction=False):
         """See `IJob`."""
         self._set_status(JobStatus.SUSPENDED)
+        if manage_transaction:
+            transaction.commit()
 
     def resume(self):
         """See `IJob`."""

=== modified file 'lib/lp/services/job/runner.py'
--- lib/lp/services/job/runner.py	2012-01-01 02:58:52 +0000
+++ lib/lp/services/job/runner.py	2012-03-28 11:45:27 +0000
@@ -38,6 +38,7 @@
     pool,
     )
 from lazr.delegates import delegates
+from lazr.jobrunner.jobrunner import JobRunner as LazrJobRunner
 import transaction
 from twisted.internet import reactor
 from twisted.internet.defer import (
@@ -61,7 +62,6 @@
     IJob,
     IRunnableJob,
     LeaseHeld,
-    SuspendJobException,
     )
 from lp.services.mail.sendmail import (
     MailController,
@@ -176,20 +176,24 @@
             return
         ctrl.send()
 
-
-class BaseJobRunner(object):
+    def makeOopsReport(self, oops_config, info):
+        """Generate an OOPS report using the given OOPS configuration."""
+        return oops_config.create(
+            context=dict(exc_info=info))
+
+
+class BaseJobRunner(LazrJobRunner):
     """Runner of Jobs."""
 
     def __init__(self, logger=None, error_utility=None):
-        self.completed_jobs = []
-        self.incomplete_jobs = []
-        if logger is None:
-            logger = logging.getLogger()
-        self.logger = logger
-        self.error_utility = error_utility
         self.oops_ids = []
-        if self.error_utility is None:
+        if error_utility is None:
             self.error_utility = errorlog.globalErrorUtility
+        else:
+            self.error_utility = error_utility
+        super(BaseJobRunner, self).__init__(
+            logger, oops_config=self.error_utility._oops_config,
+            oopsMessage=self.error_utility.oopsMessage)
 
     def acquireLease(self, job):
         self.logger.debug(
@@ -204,83 +208,8 @@
             return False
         return True
 
-    @staticmethod
-    def job_str(job):
-        class_name = job.__class__.__name__
-        ijob_id = removeSecurityProxy(job).job.id
-        return '%s (ID %d)' % (class_name, ijob_id)
-
     def runJob(self, job):
-        """Attempt to run a job, updating its status as appropriate."""
-        job = IRunnableJob(job)
-
-        self.logger.info(
-            'Running %s in status %s' % (
-                self.job_str(job), job.status.title))
-        job.start()
-        transaction.commit()
-        do_retry = False
-        try:
-            try:
-                job.run()
-            except job.retry_error_types, e:
-                if job.attempt_count > job.max_retries:
-                    raise
-                self.logger.exception(
-                    "Scheduling retry due to %s.", e.__class__.__name__)
-                do_retry = True
-        except SuspendJobException:
-            self.logger.debug("Job suspended itself")
-            job.suspend()
-            self.incomplete_jobs.append(job)
-        except Exception:
-            transaction.abort()
-            job.fail()
-            # Record the failure.
-            transaction.commit()
-            self.incomplete_jobs.append(job)
-            raise
-        else:
-            # Commit transaction to update the DB time.
-            transaction.commit()
-            if do_retry:
-                job.queue()
-                self.incomplete_jobs.append(job)
-            else:
-                job.complete()
-                self.completed_jobs.append(job)
-        # Commit transaction to update job status.
-        transaction.commit()
-
-    def runJobHandleError(self, job):
-        """Run the specified job, handling errors.
-
-        Most errors will be logged as Oopses.  Jobs in user_error_types won't.
-        The list of complete or incomplete jobs will be updated.
-        """
-        job = IRunnableJob(job)
-        with self.error_utility.oopsMessage(
-            dict(job.getOopsVars())):
-            try:
-                try:
-                    self.logger.debug('Running %r', job)
-                    self.runJob(job)
-                except job.user_error_types, e:
-                    self.logger.info('Job %r failed with user error %r.' %
-                        (job, e))
-                    job.notifyUserError(e)
-                except Exception:
-                    info = sys.exc_info()
-                    return self._doOops(job, info)
-            except Exception:
-                # This only happens if sending attempting to notify users
-                # about errors fails for some reason (like a misconfigured
-                # email server).
-                self.logger.exception(
-                    "Failed to notify users about a failure.")
-                info = sys.exc_info()
-                # Returning the oops says something went wrong.
-                return self.error_utility.raising(info)
+        super(BaseJobRunner, self).runJob(IRunnableJob(job))
 
     def _doOops(self, job, info):
         """Report an OOPS for the provided job and info.
@@ -291,6 +220,7 @@
         """
         oops = self.error_utility.raising(info)
         job.notifyOops(oops)
+        self._logOopsId(oops['id'])
         return oops
 
     def _logOopsId(self, oops_id):
@@ -331,9 +261,7 @@
                 continue
             # Commit transaction to clear the row lock.
             transaction.commit()
-            oops = self.runJobHandleError(job)
-            if oops is not None:
-                self._logOopsId(oops['id'])
+            self.runJobHandleError(job)
 
 
 class RunJobCommand(amp.Command):

=== modified file 'lib/lp/services/job/tests/test_runner.py'
--- lib/lp/services/job/tests/test_runner.py	2012-01-01 02:58:52 +0000
+++ lib/lp/services/job/tests/test_runner.py	2012-03-28 11:45:27 +0000
@@ -8,6 +8,7 @@
 from textwrap import dedent
 from time import sleep
 
+from lazr.jobrunner.jobrunner import SuspendJobException
 from testtools.matchers import MatchesRegex
 from testtools.testcase import ExpectedException
 import transaction
@@ -19,7 +20,6 @@
     IRunnableJob,
     JobStatus,
     LeaseHeld,
-    SuspendJobException,
     )
 from lp.services.job.model.job import Job
 from lp.services.job.runner import (
@@ -411,8 +411,8 @@
         self.job = Job()
 
     def __repr__(self):
-        return '<StuckJob(%r, lease_length=%s, delay=%s)>' % (
-            self.id, self.lease_length, self.delay)
+        return '<%s(%r, lease_length=%s, delay=%s)>' % (
+            self.__class__.__name__, self.id, self.lease_length, self.delay)
 
     def acquireLease(self):
         return self.job.acquireLease(self.lease_length)
@@ -550,15 +550,14 @@
             (1, 1), (len(runner.completed_jobs), len(runner.incomplete_jobs)))
         self.oops_capture.sync()
         oops = self.oopses[0]
-        self.assertEqual(
-            ('TimeoutError', 'Job ran too long.'),
-            (oops['type'], oops['value']))
+        expected_exception = ('TimeoutError', 'Job ran too long.')
+        self.assertEqual(expected_exception, (oops['type'], oops['value']))
         self.assertThat(logger.getLogBuffer(), MatchesRegex(
             dedent("""\
-            INFO Running through Twisted.
-            INFO Running StuckJob \(ID .*\).
-            INFO Running StuckJob \(ID .*\).
-            INFO Job resulted in OOPS: .*
+                INFO Running through Twisted.
+                INFO Running <StuckJob.*?> \(ID .*?\).
+                INFO Running <StuckJob.*?> \(ID .*?\).
+                INFO Job resulted in OOPS: .*
             """)))
 
     def test_timeout_short(self):
@@ -582,8 +581,8 @@
             logger.getLogBuffer(), MatchesRegex(
                 dedent("""\
                 INFO Running through Twisted.
-                INFO Running ShorterStuckJob \(ID .*\).
-                INFO Running ShorterStuckJob \(ID .*\).
+                INFO Running <ShorterStuckJob.*?> \(ID .*?\).
+                INFO Running <ShorterStuckJob.*?> \(ID .*?\).
                 INFO Job resulted in OOPS: %s
                 """) % oops['id']))
         self.assertEqual(('TimeoutError', 'Job ran too long.'),

=== modified file 'lib/lp/soyuz/model/packagecopyjob.py'
--- lib/lp/soyuz/model/packagecopyjob.py	2012-02-24 03:47:44 +0000
+++ lib/lp/soyuz/model/packagecopyjob.py	2012-03-28 11:45:27 +0000
@@ -11,6 +11,7 @@
 import logging
 
 from lazr.delegates import delegates
+from lazr.jobrunner.jobrunner import SuspendJobException
 from storm.locals import (
     And,
     Int,
@@ -49,7 +50,6 @@
 from lp.services.database.stormbase import StormBase
 from lp.services.job.interfaces.job import (
     JobStatus,
-    SuspendJobException,
     )
 from lp.services.job.model.job import Job
 from lp.services.job.runner import BaseRunnableJob

=== modified file 'lib/lp/soyuz/tests/test_packagecopyjob.py'
--- lib/lp/soyuz/tests/test_packagecopyjob.py	2012-02-01 11:31:20 +0000
+++ lib/lp/soyuz/tests/test_packagecopyjob.py	2012-03-28 11:45:27 +0000
@@ -6,6 +6,7 @@
 import operator
 from textwrap import dedent
 
+from lazr.jobrunner.jobrunner import SuspendJobException
 from storm.store import Store
 from testtools.content import text_content
 from testtools.matchers import MatchesStructure
@@ -25,7 +26,6 @@
 from lp.services.features.testing import FeatureFixture
 from lp.services.job.interfaces.job import (
     JobStatus,
-    SuspendJobException,
     )
 from lp.services.webapp.testing import verifyObject
 from lp.soyuz.adapters.overrides import SourceOverride

=== modified file 'lib/lp/testing/factory.py'
--- lib/lp/testing/factory.py	2012-03-26 14:23:39 +0000
+++ lib/lp/testing/factory.py	2012-03-28 11:45:27 +0000
@@ -48,6 +48,7 @@
 from bzrlib.merge_directive import MergeDirective2
 from bzrlib.plugins.builder.recipe import BaseRecipeBranch
 from bzrlib.revision import Revision as BzrRevision
+from lazr.jobrunner.jobrunner import SuspendJobException
 import pytz
 from pytz import UTC
 import simplejson
@@ -233,7 +234,6 @@
     IEmailAddressSet,
     )
 from lp.services.identity.model.account import Account
-from lp.services.job.interfaces.job import SuspendJobException
 from lp.services.librarian.interfaces import ILibraryFileAliasSet
 from lp.services.mail.signedmessage import SignedMessage
 from lp.services.messages.model.message import (

=== modified file 'lib/lp/translations/scripts/tests/test_packaging_translations.py'
--- lib/lp/translations/scripts/tests/test_packaging_translations.py	2012-01-01 02:58:52 +0000
+++ lib/lp/translations/scripts/tests/test_packaging_translations.py	2012-03-28 11:45:27 +0000
@@ -33,10 +33,10 @@
         matcher = MatchesRegex(dedent("""\
             INFO    Creating lockfile: /var/lock/launchpad-jobcronscript.lock
             INFO    Running synchronously.
-            INFO    Running TranslationMergeJob \(ID .*\) in status Waiting
+            INFO    Running <.*?TranslationMergeJob.*?> \(ID .*\) in status Waiting
             INFO    Merging .* and .* in Ubuntu Distroseries.*
             INFO    Deleted POTMsgSets: 1.  TranslationMessages: 1.
-            INFO    Running TranslationSplitJob \(ID .*\) in status Waiting
+            INFO    Running <.*?TranslationSplitJob.*?> \(ID .*\) in status Waiting
             INFO    Splitting .* and .* in Ubuntu Distroseries.*
             INFO    1 entries split.
             INFO    Ran 1 TranslationMergeJob jobs.

=== modified file 'setup.py'
--- setup.py	2012-03-09 04:33:37 +0000
+++ setup.py	2012-03-28 11:45:27 +0000
@@ -47,6 +47,7 @@
         'lazr.enum',
         'lazr.lifecycle',
         'lazr.restful',
+        'lazr.jobrunner',
         'lazr.smtptest',
         'lazr.testing',
         'lazr.uri',

=== modified file 'versions.cfg'
--- versions.cfg	2012-03-26 22:45:24 +0000
+++ versions.cfg	2012-03-28 11:45:27 +0000
@@ -5,11 +5,13 @@
 # Alphabetical, case-insensitive, please! :-)
 
 ampoule = 0.2.0
-amqplib = 0.6.1
+amqplib = 1.0.2
+anyjson = 0.3.1
 argparse = 1.2.1
 BeautifulSoup = 3.1.0.1
 bson = 0.3.2
 bzr = 2.5.0dev2-r6152
+celery = 2.5.1
 Chameleon = 2.4.0
 ClientForm = 0.2.10
 cssutils = 0.9.7
@@ -26,11 +28,13 @@
 grokcore.component = 1.6
 html5browser = 0.0.9
 httplib2 = 0.6.0
+importlib = 1.0.2
 ipython = 0.9.1
 iso8601 = 0.1.4
 jsautobuild = 0.2
 Jinja2 = 2.2
 keyring = 0.6.2
+kombu = 2.1.1
 launchpadlib = 1.9.12
 lazr.amqp = 0.1
 lazr.authentication = 0.1.1
@@ -38,6 +42,7 @@
 lazr.config = 1.1.3
 lazr.delegates = 1.2.0
 lazr.enum = 1.1.3
+lazr.jobrunner = 0.2
 lazr.lifecycle = 1.1
 lazr.restful = 0.19.6
 lazr.restfulclient = 0.12.0
@@ -78,6 +83,7 @@
 pymongo = 2.1.1
 pyOpenSSL = 0.10
 pystache = 0.3.1
+python-dateutil = 1.5
 python-memcached = 1.48
 # 2.2.1 with the one-liner Expect: 100-continue fix from
 # lp:~wgrant/python-openid/python-openid-2.2.1-fix676372.