Introduction

The Job System allows jobs to be run asynchronously from web requests.

Where a request needs to trigger a long-running job that would take too long for processing during a single web request the job system can be used to queue the job for later processing.

It is currently used for things such as:

• Generating merge proposal diffs
• Calculating bug heat

Architecture

Each job that will go through the system has a specific type, which stores the arguments specific to that job type. For instance the merge proposal diff generation links to the merge proposal for which the diff should be generated.

Each type of job also shares some common fields that are used by the job system to ensure that the job is processed. These include the status of the job, timestamps at which it was queued, completed, etc.

Each job is backed by the database, providing durability and error tolerance.

Each job type also has an associated script which processes jobs of that type as needed, usually run from cron. The script selects any outstanding jobs and runs them. In some cases, one script runs several types of jobs.

Code

The code for this lives in lp.services.job.

You can see implementations of job types in

• lib/lp/bugs/interfaces/bugjob.py, lib/lp/bugs/model/bugjob.py, lib/lp/bugs/model/bugheat.py and cronscripts/calculate-bug-heat.py
• lib/lp/code/interfaces/branchjob.py, lib/lp/code/model/branchjob.py, lib/lp/code/interfaces/branchmergeproposal.py, lib/lp/code/model/branchmergeproposal.py and cronscripts/merge-proposal-jobs.py

In lp.services.job.interfaces.job are the basic interfaces for jobs, which each job type will build upon.

• JobStatus is an enum of the statuses a job can be in, waiting, running, completed, etc.

• IJob is the interface implemented by each job type, and hold the attributes common to all types, as well as standard methods that the job system can use to run the jobs and manipulate them in other ways.
• IRunnableJob is a job that can be run right now, and has a run() method that will do the work.
• IJobSource is the interface for an object which can provide jobs that are ready to run right now.

Implementing a job type

At minimum, you need an object with a .job attribute, where the object implements IRunnableJob, and the .job attribute implements IJob.

You may choose to implement this using single-table polymorphism.

Implementing the behaviour of the specific job type

Now we move on to implementing the behaviour of our specific type, DO_FOO.

For this example we will put the tests in lib/lp/component/tests/test_dofoo.py

First we will test something static.

from canonical.testing import DatabaseFunctionalLayer

from lp.testing import TestCaseWithFactory


class DoFooJobTests(TestCaseWithFactory):
    """Test case for DoFooJob."""
    
    layer = DatabaseFunctionalLayer

    def test_getOopsVars(self):
        foo = self.factory.makeFoo()
        job = DoFooJob.create(foo)
        vars = job.getOopsVars()
        self.assertIn(('foo_id', foo.id), vars)
        self.assertIn(('foo_job_id', job.context.id), vars)
        self.assertIn(('foo_job_type', job.context.job_type.title), vars)

This doesn't actually depend on the behaviour of the job, and in fact the variables are applicable to any FooJob, so getOopsVars on FooJobDerived could implement that. That is in fact what we want, but we can't test FooJobDerived directly as we can't create instances of it, therefore we test it via DoFooJob.

Firstly then, add a method to FooJobDerived.

    def getOopsVars(self):
        """See `IRunnableJob`."""
        vars =  BaseRunnableJob.getOopsVars(self)
        vars.extend([
            ('foo_id', self.context.foo.id),
            ('foo_job_id', self.context.id),
            ('foo_job_type', self.context.job_type.title),
            ])
        return vars

This won't let the test pass though, as we still need DoFooJob.

In ./lib/lp/component/interfaces/foojob.py put:

from lp.services.job.interfaces.job import IRunnableJob

class IDoFooJob(IRunnableJob):
    """A Job to do Foo."""


class IDoFooJobSource(IFooJobSource):
    """An interface for acquiring DoFooJobs."""

In ./lib/lp/component/model/dofoojob.py put:

from zope.interface import classProvides, implements

from lp.component.interfaces.foojob import (
    FooJobType, IDoFooJob, IDoFooJobSource)
from lp.component.model.foojob import FooJobDerived


class DoFooJob(FooJobDerived):

    implements(IDoFooJob)

    class_job_type = FooJobType.DO_FOO
    classProvides(IDoFooJobSource)

    @classmethod
    def create(cls, foo):
        """See `IDoFooJobSource`."""
        return super(DoFooJob, cls).create(foo)

Now your test should pass.

Unique jobs per-target

Some job types can be unique for the target, such as calculating bug heat, or generating a merge proposal diff. If you want to have only a single job of each type for each foo then you need a bit more code. Add to DoFooJobTests:

    def _getJobs(self):
        """Return the pending DoFooJobs as a list."""
        return list(DoFooJob.iterReady())
        
    def _getJobCount(self):
        """Return the number of DoFooJobs in the queue."""
        return len(self._getJobs())
        
    def test_create_only_creates_one(self):
        archive = self.factory.makeFoo()
        # If there's already a DoFooJob for a foo,
        # DoFooJob.create() won't create a new one.
        job = DoFooJob.create(foo)
    
        # There will now be one job in the queue.
        self.assertEqual(1, self._getJobCount())

        new_job = DoFooJob.create(foo)
        
        # The two jobs will in fact be the same job.
        self.assertEqual(job, new_job)
    
        # And the queue will still have a length of 1.
        self.assertEqual(1, self._getJobCount())

To make this test pass we need to make DoFooJob.create look like:

    @classmethod                                                              
    def create(cls, foo):                                                 
        """See `IDoFooJobSource`."""
        # If there's already a job for the foo, don't create a new one.   
        store = getUtility(IStoreSelector).get(MAIN_STORE, DEFAULT_FLAVOR)    
        job_for_foo = store.find(                                             
            FooJob,                                                       
            FooJob.foo == foo,                                        
            FooJob.job_type == cls.class_job_type,                        
            FooJob.job == Job.id,                                         
            Job.id.is_in(Job.ready_jobs)                                      
            ).any()                                                           
                                                                              
        if job_for_foo is not None:                                           
            return cls(job_for_foo)                                           
        else:                                                                 
            return super(DoFooJob, cls).create(foo)

Doing the actual work

Now lets look at implementing the actual work. A new method on the test case:

    def test_run(self):                                                       
        """Test that DoFooJob.run() actually frozzles the foo."""     
        foo = self.factory.makeFoo()
        job = DoFooJob.create(foo)                                  
         
        job.run()

        # Put checks here that the job actually ran
        self.assertEqual("frozzled", foo.some_attribute)

To implement this we have to add DoFooJob.run

    def run(self):
        """See `IRunnableJob`."""
        # Put here the code that acts on self.foo to do the work

The job processing script

Here's a test that the cronscript does what it should:

import transaction
from canonical.launchpad.scripts.tests import run_script


    def test_cronscript_succeeds(self):
        # The do-foo cronscript will run all pending
        # DoFooJobs.
        foo = self.factory.makeFoo()
        DoFooJob.create(foo)
        transaction.commit()

        retcode, stdout, stderr = run_script(
            'cronscripts/calculate-bug-heat.py', [],
            expect_returncode=0)
        self.assertEqual('', stdout)
        self.assertIn(
            'INFO    Ran 1 DoFooJob jobs.\n', stderr)

The content of that cronscript should be something like

#
# Copyright 2010 Canonical Ltd.  This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).

# pylint: disable-msg=W0403

"""Do Foo."""

__metaclass__ = type

import _pythonpath
        
from canonical.launchpad.webapp import errorlog

from lp.services.job.runner import JobCronScript
from lp.component.interfaces.foojob import IDoFooJobSource
        
        
class RunDoFoo(JobCronScript):
    """Run DoFooJob jobs."""
        
    config_name = 'do_foo'
    source_interface = IDoFooJobSource
            
    def main(self):
        errorlog.globalErrorUtility.configure(self.config_name)
        return super(RunDoFoo, self).main()
        
        
if __name__ == '__main__':
    script = RunDoFoo()
    script.lock_and_run()

Attaching metdata

We defined a JSON metadata field in the schema, but we haven't made any use of it yet.

To do that first change FooJobDerived.create to be

    @classmethod
    def create(cls, foo, metadata=None):
        """See `IFooJob`."""
        if metadata is None:
            metadata = {}
        job = FooJob(foo, cls.class_job_type, metadata)
        return cls(job)

Thereby allowing us to create FooJobs with metadata.

In the subclasses, e.g. DoFooJob, you can have the create method pass a json-compatible object to the up-called create classmethod. This can either be passed directly from an argument, or more likely be a dict or similar formed from specific parameters, e.g.

    @classmethod
    def create(cls, foo, bar):
        ...
        ...
        return super(DoFooJob, cls).create(foo, {'bar': bar})

The run method can then access self.metadata when running the job.

Notes

If creating your Foo object adds one of the jobs that you are testing for, see lp.bugs.tests.test_bugheat.CalculateBugHeatJobTestCase.setUp() for how to avoid this messing up your tests.