Merge lp:~edwin-grubbs/launchpad/bug-615654-jobqueue-cron-script into lp:launchpad/db-devel

[Edwin Grubbs]

Summary
-------

This branch adds a cron script to process the
IMembershipNotificationJobs. The cronscript has been abstracted so that
new job sources can be processed by just adding configs to
schema-lazr.conf, as long as it uses a crontab_group that already has a
crontab entry configured in production.

Implementation details
----------------------

New db users and configs for cronjobs:
    database/schema/security.cfg
    lib/canonical/config/schema-lazr.conf

Cronscripts and tests.
    cronscripts/process-job-source-groups.py
    cronscripts/process-job-source.py
    lib/lp/registry/tests/test_membership_notification_job_creation.py
    lib/lp/registry/tests/test_process_job_sources_cronjob.py

Improved error message.
    lib/canonical/launchpad/webapp/errorlog.py

Added get() to IJobSource definition, since TwistedJobRunner needs it.
Moved the get() method into the class that actually provides IJobSource.
    lib/lp/services/job/interfaces/job.py
    lib/lp/registry/model/persontransferjob.py

Cruft from previous branch.
    lib/lp/registry/model/teammembership.py

Eliminate the need for TwistedJobRunner cron scripts to be run with
bin/py by importing _pythonpath in ampoule child processes like normal
cron scripts. Eliminated the need to specify the error_dir, oops_prefix,
and copy_to_zlog when the defaults in [error_reports] are fine.
    lib/lp/services/job/runner.py

Tests
-----

./bin/test -vv -t 'test_membership_notification_job_creation|test_process_job_sources_cronjob'

Demo and Q/A
------------

* Add a member to a team.
  * Change the member status to ADMIN.
  * Run ./cronscripts/process-job-source-groups.py -v MAIN
  * Check email.

[Gavin Panella (allenap)]

Hi Edwin,

This branch looks good.

I know you're adding ProcessJobSourceGroups in here as well as the
membership stuff, but it still seems like a lot of work to do
something asynchronously. Moving something out of the app server takes
a lot of plumbing work, and it's not trivial plumbing work either.

I'd really like a way to create, in the app server, something like a
persistent closure that can be run later, there or somewhere
else. When run it would have the same user and database access as in
the app server thread that spawned it. Creating a new job now seems to
involve creating tables, database users, job sources, and so on, and
it's getting in the way of the actual work that needs doing.

I'm rambling on, and it's not a fault in this branch, but after doing
this recently, do you have any thoughts on that?

Gavin.

[1]

+ for job_source in selected_job_sources:
+ child = subprocess.Popen(child_args + [job_source])
+ children.append(child)
+ for child in children:
+ child.wait()

Is selected_job_sources ever likely to be large? If not now, but could
be one day, consider filing a bug to limit the number of concurrency
processes.

[2]

+ for child in children:
+ child.wait()

Is there any value in checking and doing something with the exit code
from the child?

[3]

+ packages=('_pythonpath', 'twisted', 'ampoule'), env=env)

I just looked again at _pythonpath.py and now I can't see.

review: Approve

[Aaron Bentley (abentley)]

Gavin, radical changes in the way we create and run tasks are the purview of the NewTaskSystem. Not having to create new database users is already listed in https://dev.launchpad.net/Foundations/NewTaskSystem/Requirements and you may wish to add your other wishes under "nice to have".

[Edwin Grubbs (edwin-grubbs)]

> Hi Edwin,
>
> This branch looks good.

Thanks for the review. Unfortunately, abentley had already started reviewing it, but it was all done on IRC up to this point. I should have just listed him as the requested reviewer. Sorry for the duplicate effort. I'll try to answer your questions below.

> I know you're adding ProcessJobSourceGroups in here as well as the
> membership stuff, but it still seems like a lot of work to do
> something asynchronously. Moving something out of the app server takes
> a lot of plumbing work, and it's not trivial plumbing work either.
>
> I'd really like a way to create, in the app server, something like a
> persistent closure that can be run later, there or somewhere
> else. When run it would have the same user and database access as in
> the app server thread that spawned it. Creating a new job now seems to
> involve creating tables, database users, job sources, and so on, and
> it's getting in the way of the actual work that needs doing.
>
> I'm rambling on, and it's not a fault in this branch, but after doing
> this recently, do you have any thoughts on that?
>
> Gavin.

It seems that the reason we don't support a closure method for processing the jobs is that there are a lot of security restrictions placed on us. For example, each job class is required to have its own dbuser, so that the losas can figure out which job class is causing problems on the db. Of course, the job system seems much better suited for really slow processes, and it might be worthwhile to have a secondary solution that just moves the processing after the request response.

> [1]
>
> + for job_source in selected_job_sources:
> + child = subprocess.Popen(child_args + [job_source])
> + children.append(child)
> + for child in children:
> + child.wait()
>
> Is selected_job_sources ever likely to be large? If not now, but could
> be one day, consider filing a bug to limit the number of concurrency
> processes.

Currently, each job source class is given its own process. If we run into a problem with too many concurrent jobs being processed, the job source classes can be reorganized into different crontab_groups, and then different crontab entries can process the jobs at different times. We are not really interested in a long term solution for this, since the job system will most likely be moved to some kind of message queue, so the cron scripts for job processing will disappear.

> [2]
>
> + for child in children:
> + child.wait()
>
> Is there any value in checking and doing something with the exit code
> from the child?

No, waiting for the children to finish is really only helpful for testing, so that all the output from the child processes can be gathered. The cron scripts log to files, and I am actually making the default behavior not to wait on the child processes, so that one really long running child process does not cause the other job sources not to run the next time the crontab entry calls it, because the parent process still has its lock file.

> [3]
>
> + packages=('_pythonpath', 'twisted', 'ampoule'), env=env)
>
> I just looked again at ...

[Aaron Bentley (abentley)]

Waiting for children to exit will mean that no new jobs can be processed until all the jobs complete. This may severely impair responsiveness. I suggest starting the subprocesses and then exiting.

The NoErrorOptionParser should not exist. Option handling should be taken care of in the ProcessJobSource class. If the layering is getting in our way, we should fix it, not work around it.

Unless you are specifically testing stringification, you should not stringify objects in tests, so that your tests are not fragile.

In test_setstatus_admin, what's the purpose of transaction.commit?

Since you do person_set.getByEmail('<email address hidden>'), you can use login_person or with person_logged_in instead of login. Also, lifeless requested that you use the constant for <email address hidden>' instead of spelling it out.

Please use lp.testing.run_script instead of rolling your own.

In test_processed, I don't think you really care what the database id of the job is, so use assertTextMatchesExpressionIgnoreWhitespace. (You'll need to move it to TestCaseWithFactory.)

get isn't required to run jobs under the non-twisted job runner, so it seems like an unnecessary requirement on IJobSource.

Perhaps ProcessJobSource should go in lp.services.job.runner ?

test_empty_queue looks fragile because it includes the directory of the lock file. Perhaps another case for assertTextMatchesExpressionIgnoreWhitespace?

Perhaps PYTHONPATH and LPCONFIG should be handled by common code, eg:
for name in ['PYTHONPATH', 'LPCONFIG']:
    if name in os.environ:
        env[name] = os.environ[name]

review: Needs Fixing

[Gavin Panella (allenap)]

> Thanks for the review. Unfortunately, abentley had already started reviewing
> it, but it was all done on IRC up to this point. I should have just listed him
> as the requested reviewer. Sorry for the duplicate effort. I'll try to answer
> your questions below.

No worries. It's good that Aaron also reviewed this. He spotted a lot
of stuff that I skimmed over; I didn't take as long on this review as
I ought to have done.

>
> > I know you're adding ProcessJobSourceGroups in here as well as the
> > membership stuff, but it still seems like a lot of work to do
> > something asynchronously. Moving something out of the app server takes
> > a lot of plumbing work, and it's not trivial plumbing work either.
> >
> > I'd really like a way to create, in the app server, something like a
> > persistent closure that can be run later, there or somewhere
> > else. When run it would have the same user and database access as in
> > the app server thread that spawned it. Creating a new job now seems to
> > involve creating tables, database users, job sources, and so on, and
> > it's getting in the way of the actual work that needs doing.
> >
> > I'm rambling on, and it's not a fault in this branch, but after doing
> > this recently, do you have any thoughts on that?
> >
> > Gavin.
>
>
> It seems that the reason we don't support a closure method for processing the
> jobs is that there are a lot of security restrictions placed on us. For
> example, each job class is required to have its own dbuser, so that the losas
> can figure out which job class is causing problems on the db. Of course, the
> job system seems much better suited for really slow processes, and it might be
> worthwhile to have a secondary solution that just moves the processing after
> the request response.

Yes, that sounds good. As suggested by Aaron, I'll add it to the
wishlist.

>
> > [1]
> >
> > + for job_source in selected_job_sources:
> > + child = subprocess.Popen(child_args + [job_source])
> > + children.append(child)
> > + for child in children:
> > + child.wait()
> >
> > Is selected_job_sources ever likely to be large? If not now, but could
> > be one day, consider filing a bug to limit the number of concurrency
> > processes.
>
>
> Currently, each job source class is given its own process. If we run into a
> problem with too many concurrent jobs being processed, the job source classes
> can be reorganized into different crontab_groups, and then different crontab
> entries can process the jobs at different times. We are not really interested
> in a long term solution for this, since the job system will most likely be
> moved to some kind of message queue, so the cron scripts for job processing
> will disappear.

Okay, fair enough.

>
> > [2]
> >
> > + for child in children:
> > + child.wait()
> >
> > Is there any value in checking and doing something with the exit code
> > from the child?
>
>
> No, waiting for the children to finish is really only helpful for testing, so
> that all the output from the child processes can be gathered. The cron scripts
> log to files, and I am actually making the default behavi...

[Edwin Grubbs (edwin-grubbs)]

On Thu, Sep 30, 2010 at 11:02 AM, Aaron Bentley <email address hidden> wrote:
> Review: Needs Fixing
> Waiting for children to exit will mean that no new jobs can be processed until all the jobs complete.  This may severely impair responsiveness.  I suggest starting the subprocesses and then exiting.

It no longer waits for child processes by default. I added an option
to wait, which is convenient when running it from the command line
when experimenting.

> The NoErrorOptionParser should not exist.  Option handling should be taken care of in the ProcessJobSource class.  If the layering is getting in our way, we should fix it, not work around it.

I moved most of the setup code to the run() methods in LaunchpadScript
and LaunchpadCronScript so __init__ does very little now besides
setting up the parser attribute.

> Unless you are specifically testing stringification, you should not stringify objects in tests, so that your tests are not fragile.

Fixed.

> In test_setstatus_admin, what's the purpose of transaction.commit?

I'm not sure why I was thinking I would need that. Removed.

> Since you do person_set.getByEmail('<email address hidden>'), you can use login_person or with person_logged_in instead of login.  Also, lifeless requested that you use the constant for <email address hidden>' instead of spelling it out.

Fixed.

> Please use lp.testing.run_script instead of rolling your own.

Fixed.

> In test_processed, I don't think you really care what the database id of the job is, so use  assertTextMatchesExpressionIgnoreWhitespace.  (You'll need to move it to TestCaseWithFactory.)

Fixed.

> get isn't required to run jobs under the non-twisted job runner, so it seems like an unnecessary requirement on IJobSource.

ok, I've moved get() to a new ITwistedJobSource interface.

> Perhaps ProcessJobSource should go in lp.services.job.runner ?

I'd rather wait until we have a reason for subclassing from it, which
I hope won't happen before we stop using cron for job processing.

> test_empty_queue looks fragile because it includes the directory of the lock file.  Perhaps another case for assertTextMatchesExpressionIgnoreWhitespace?

Fixed.

> Perhaps PYTHONPATH and LPCONFIG should be handled by common code, eg:
> for name in ['PYTHONPATH', 'LPCONFIG']:
>    if name in os.environ:
>        env[name] = os.environ[name]

Fixed.

Incremental diff:

=== modified file 'cronscripts/process-job-source-groups.py'
--- cronscripts/process-job-source-groups.py 2010-09-28 15:50:05 +0000
+++ cronscripts/process-job-source-groups.py 2010-09-30 20:33:49 +0000
@@ -47,6 +47,13 @@
             '-e', '--exclude', dest='excluded_job_sources',
             metavar="JOB_SOURCE", default=[], action='append',
             help="Exclude specific job sources.")
+ self.parser.add_option(
+ '--wait', dest='do_wait', default=False, action='store_true',
+ help="Wait for the child processes to finish. This is useful "
+ "for testing, but it shouldn't be used in a cronjob, since "
+ "it would prevent the cronjob from processing new jobs "
+ "if just one of the child processes is still processing, "
...

[Edwin Grubbs (edwin-grubbs)]

I realized that I could make process-job-source.py simpler. Here is the diff.

=== modified file 'cronscripts/process-job-source.py'
--- cronscripts/process-job-source.py 2010-09-30 20:39:02 +0000
+++ cronscripts/process-job-source.py 2010-10-01 13:39:54 +0000
@@ -24,12 +24,8 @@
         "For more help, run:\n"
         " cronscripts/process-job-source-groups.py --help")

- def configure(self):
- """Override configs main() is called by run().
-
- It is done here instead of __init__(), so that we get to use its
- option parser.
- """
+ def __init__(self):
+ super(ProcessJobSource, self).__init__()
         if len(self.args) != 1:
             self.parser.print_help()
             sys.exit(1)
@@ -54,5 +50,4 @@

 if __name__ == '__main__':
     script = ProcessJobSource()
- script.configure()
     script.lock_and_run()

[Edwin Grubbs (edwin-grubbs)]

I ran into some problems merging in db-devel due to some changes in LaunchpadCronScript. I switched to using properties to make it easy for subclasses to override the values without interfering with __init__() logic.

Instead of an incremental diff, here is a full diff of the affected files.

=== added file 'cronscripts/process-job-source.py'
--- cronscripts/process-job-source.py 1970-01-01 00:00:00 +0000
+++ cronscripts/process-job-source.py 2010-10-04 04:01:36 +0000
@@ -0,0 +1,69 @@
+#!/usr/bin/python -S
+#
+# Copyright 2009, 2010 Canonical Ltd. This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Handle jobs for a specified job source class."""
+
+__metaclass__ = type
+
+import sys
+
+import _pythonpath
+from twisted.python import log
+
+from canonical.config import config
+from lp.services.job import runner
+from lp.services.job.runner import JobCronScript
+
+
+class ProcessJobSource(JobCronScript):
+ """Run jobs for a specified job source class."""
+ usage = (
+ "Usage: %prog [options] JOB_SOURCE\n\n"
+ "For more help, run:\n"
+ " cronscripts/process-job-source-groups.py --help")
+
+ def __init__(self):
+ super(ProcessJobSource, self).__init__()
+ # The fromlist argument is necessary so that __import__()
+ # returns the bottom submodule instead of the top one.
+ module = __import__(self.config_section.module,
+ fromlist=[self.job_source_name])
+ self.source_interface = getattr(module, self.job_source_name)
+
+ @property
+ def config_name(self):
+ return self.job_source_name
+
+ @property
+ def name(self):
+ return 'process-job-source-%s' % self.job_source_name
+
+ @property
+ def dbuser(self):
+ return self.config_section.dbuser
+
+ @property
+ def runner_class(self):
+ runner_class_name = getattr(
+ self.config_section, 'runner_class', 'JobRunner')
+ # Override attributes that are normally set in __init__().
+ return getattr(runner, runner_class_name)
+
+ def handle_options(self):
+ if len(self.args) != 1:
+ self.parser.print_help()
+ sys.exit(1)
+ self.job_source_name = self.args[0]
+ super(ProcessJobSource, self).handle_options()
+
+ def main(self):
+ if self.options.verbose:
+ log.startLogging(sys.stdout)
+ super(ProcessJobSource, self).main()
+
+
+if __name__ == '__main__':
+ script = ProcessJobSource()
+ script.lock_and_run()

=== modified file 'lib/lp/services/job/runner.py'
--- lib/lp/services/job/runner.py 2010-09-20 09:48:58 +0000
+++ lib/lp/services/job/runner.py 2010-10-04 04:01:36 +0000
@@ -77,7 +77,6 @@

     # We redefine __eq__ and __ne__ here to prevent the security proxy
     # from mucking up our comparisons in tests and elsewhere.
-
     def __eq__(self, job):
         return (
             self.__class__ is removeSecurityProxy(job.__class__)
@@ -337,13 +336,12 @@
     """Run Jobs via twisted."""

     def __init__(self, job_source, dbuser, logger=None, error_utility=None):
- env = {'PYTHONPATH': os.e...

[Aaron Bentley (abentley)]

This looks good, except that there are now two definitions assertTextMatchesExpressionIgnoreWhitespace; one in TestCaseWithFactory and one in BrowserTestCase. Actually, it's not factory-specific, so it should probably go in TestCase.

review: Approve

=== modified file 'cronscripts/merge-proposal-jobs.py'
--- cronscripts/merge-proposal-jobs.py	2010-05-19 18:07:56 +0000
+++ cronscripts/merge-proposal-jobs.py	2010-10-07 14:14:12 +0000
@@ -32,7 +32,7 @@
     def __init__(self):
         super(RunMergeProposalJobs, self).__init__(
             runner_class=TwistedJobRunner,
-            script_name='merge-proposal-jobs')
+            name='merge-proposal-jobs')
 
 
 if __name__ == '__main__':

=== added file 'cronscripts/process-job-source-groups.py'
--- cronscripts/process-job-source-groups.py	1970-01-01 00:00:00 +0000
+++ cronscripts/process-job-source-groups.py	2010-10-07 14:14:12 +0000
@@ -0,0 +1,114 @@
+#!/usr/bin/python -S
+#
+# Copyright 2009, 2010 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Handle jobs for multiple job source classes."""
+
+__metaclass__ = type
+
+from optparse import IndentedHelpFormatter
+import os
+import subprocess
+import sys
+import textwrap
+
+import _pythonpath
+
+from canonical.config import config
+from lp.services.propertycache import cachedproperty
+from lp.services.scripts.base import LaunchpadCronScript
+
+
+class LongEpilogHelpFormatter(IndentedHelpFormatter):
+    """Preserve newlines in epilog."""
+
+    def format_epilog(self, epilog):
+        if epilog:
+            return '\n%s\n' % epilog
+        else:
+            return ""
+
+
+class ProcessJobSourceGroups(LaunchpadCronScript):
+    """Handle each job source in a separate process with ProcessJobSource."""
+
+    def add_my_options(self):
+        self.parser.usage = "%prog [ -e JOB_SOURCE ] GROUP [GROUP]..."
+        self.parser.epilog = (
+            textwrap.fill(
+            "At least one group must be specified. Excluding job sources "
+            "is useful when you want to run all the other job sources in "
+            "a group.")
+            + "\n\n" + self.group_help)
+
+        self.parser.formatter = LongEpilogHelpFormatter()
+        self.parser.add_option(
+            '-e', '--exclude', dest='excluded_job_sources',
+            metavar="JOB_SOURCE", default=[], action='append',
+            help="Exclude specific job sources.")
+        self.parser.add_option(
+            '--wait', dest='do_wait', default=False, action='store_true',
+            help="Wait for the child processes to finish. This is useful "
+                 "for testing, but it shouldn't be used in a cronjob, since "
+                 "it would prevent the cronjob from processing new jobs "
+                 "if just one of the child processes is still processing, "
+                 "and each process only handles a single job source class.")
+
+    def main(self):
+        selected_groups = self.args
+        if len(selected_groups) == 0:
+            self.parser.print_help()
+            sys.exit(1)
+
+        selected_job_sources = set()
+        # Include job sources from selected groups.
+        for group in selected_groups:
+            selected_job_sources.update(self.grouped_sources[group])
+        # Then, exclude job sources.
+        for source in self.options.excluded_job_sources:
+            if source not in selected_job_sources:
+                self.logger.info('%r is not in job source groups %s'
+                                  % (source, self.options.groups))
+            else:
+                selected_job_sources.remove(source)
+        # Process job sources.
+        command = os.path.join(
+            os.path.dirname(sys.argv[0]), 'process-job-source.py')
+        child_args = [command]
+        if self.options.verbose:
+            child_args.append('-v')
+        children = []
+        for job_source in selected_job_sources:
+            child = subprocess.Popen(child_args + [job_source])
+            children.append(child)
+        if self.options.do_wait:
+            for child in children:
+                child.wait()
+
+    @cachedproperty
+    def all_job_sources(self):
+        job_sources = config['process-job-source-groups'].job_sources
+        return [job_source.strip() for job_source in job_sources.split(',')]
+
+    @cachedproperty
+    def grouped_sources(self):
+        groups = {}
+        for source in self.all_job_sources:
+            if source not in config:
+                continue
+            section = config[source]
+            group = groups.setdefault(section.crontab_group, [])
+            group.append(source)
+        return groups
+
+    @cachedproperty
+    def group_help(self):
+        return '\n\n'.join(
+            'Group: %s\n    %s' % (group, '\n    '.join(sources))
+            for group, sources in sorted(self.grouped_sources.items()))
+
+
+if __name__ == '__main__':
+    script = ProcessJobSourceGroups()
+    script.lock_and_run()

=== added file 'cronscripts/process-job-source.py'
--- cronscripts/process-job-source.py	1970-01-01 00:00:00 +0000
+++ cronscripts/process-job-source.py	2010-10-07 14:14:12 +0000
@@ -0,0 +1,65 @@
+#!/usr/bin/python -S
+#
+# Copyright 2009, 2010 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Handle jobs for a specified job source class."""
+
+__metaclass__ = type
+
+import sys
+
+import _pythonpath
+from twisted.python import log
+
+from canonical.config import config
+from lp.services.job import runner
+from lp.services.job.runner import JobCronScript
+
+
+class ProcessJobSource(JobCronScript):
+    """Run jobs for a specified job source class."""
+    usage = (
+        "Usage: %prog [options] JOB_SOURCE\n\n"
+        "For more help, run:\n"
+        "    cronscripts/process-job-source-groups.py --help")
+
+    def __init__(self):
+        super(ProcessJobSource, self).__init__()
+        # The fromlist argument is necessary so that __import__()
+        # returns the bottom submodule instead of the top one.
+        module = __import__(self.config_section.module,
+                            fromlist=[self.job_source_name])
+        self.source_interface = getattr(module, self.job_source_name)
+
+    @property
+    def config_name(self):
+        return self.job_source_name
+
+    @property
+    def name(self):
+        return 'process-job-source-%s' % self.job_source_name
+
+    @property
+    def runner_class(self):
+        runner_class_name = getattr(
+            self.config_section, 'runner_class', 'JobRunner')
+        # Override attributes that are normally set in __init__().
+        return getattr(runner, runner_class_name)
+
+    def handle_options(self):
+        if len(self.args) != 1:
+            self.parser.print_help()
+            sys.exit(1)
+        self.job_source_name = self.args[0]
+        super(ProcessJobSource, self).handle_options()
+
+    def main(self):
+        if self.options.verbose:
+            log.startLogging(sys.stdout)
+        super(ProcessJobSource, self).main()
+
+
+if __name__ == '__main__':
+    script = ProcessJobSource()
+    script.lock_and_run()

=== modified file 'database/schema/security.cfg'
--- database/schema/security.cfg	2010-10-07 14:14:09 +0000
+++ database/schema/security.cfg	2010-10-07 14:14:12 +0000
@@ -715,6 +715,8 @@
 public.teamparticipation                        = SELECT, DELETE
 public.person                                   = SELECT
 public.emailaddress                             = SELECT
+public.job                                      = SELECT, INSERT
+public.persontransferjob                        = SELECT, INSERT
 
 [karma]
 # Update the KarmaCache table
@@ -1877,6 +1879,20 @@
 public.product                          = SELECT, UPDATE
 public.bugtracker                       = SELECT
 
+[process-job-source-groups]
+# Does not need access to tables.
+type=user
+groups=script
+
+[person-transfer-job]
+type=user
+groups=script
+public.emailaddress                     = SELECT
+public.job                              = SELECT, UPDATE
+public.person                           = SELECT
+public.persontransferjob                = SELECT
+public.teammembership                   = SELECT
+
 [weblogstats]
 # For the script that parses our Apache/Squid logfiles and updates statistics
 type=user

=== modified file 'lib/canonical/config/schema-lazr.conf'
--- lib/canonical/config/schema-lazr.conf	2010-09-24 22:30:48 +0000
+++ lib/canonical/config/schema-lazr.conf	2010-10-07 14:14:12 +0000
@@ -2046,3 +2046,17 @@
 
 # datatype: boolean
 send_email: true
+
+[process-job-source-groups]
+# This section is used by cronscripts/process-job-source-groups.py.
+dbuser: process-job-source-groups
+# Each job source class also needs its own config section to specify the
+# dbuser, the crontab_group, and the module that the job source class
+# can be loaded from.
+job_sources: IMembershipNotificationJobSource
+
+[IMembershipNotificationJobSource]
+# This section is used by cronscripts/process-job-source.py.
+module: lp.registry.interfaces.persontransferjob
+dbuser: person-transfer-job
+crontab_group: MAIN

=== modified file 'lib/canonical/launchpad/scripts/tests/__init__.py'
--- lib/canonical/launchpad/scripts/tests/__init__.py	2009-06-25 05:39:50 +0000
+++ lib/canonical/launchpad/scripts/tests/__init__.py	2010-10-07 14:14:12 +0000
@@ -31,5 +31,5 @@
         args, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
     stdout, stderr = process.communicate()
     if process.returncode != expect_returncode:
-        raise AssertionError('Failed:\n%s' % stderr)
+        raise AssertionError('Failed:\n%s\n%s' % (stdout, stderr))
     return (process.returncode, stdout, stderr)

=== modified file 'lib/lp/archiveuploader/tests/test_ppauploadprocessor.py'
--- lib/lp/archiveuploader/tests/test_ppauploadprocessor.py	2010-10-03 15:30:06 +0000
+++ lib/lp/archiveuploader/tests/test_ppauploadprocessor.py	2010-10-07 14:14:12 +0000
@@ -49,6 +49,7 @@
     )
 from lp.soyuz.model.publishing import BinaryPackagePublishingHistory
 from lp.soyuz.tests.test_publishing import SoyuzTestPublisher
+from lp.testing.mail_helpers import run_mail_jobs
 
 
 class TestPPAUploadProcessorBase(TestUploadProcessorBase):
@@ -668,6 +669,7 @@
             ILaunchpadCelebrities).launchpad_beta_testers
         self.name16.leave(beta_testers)
         # Pop the message notifying the membership modification.
+        run_mail_jobs()
         unused = stub.test_emails.pop()
 
         upload_dir = self.queueUpload("bar_1.0-1", "~name16/ubuntu")

=== modified file 'lib/lp/code/interfaces/branchmergeproposal.py'
--- lib/lp/code/interfaces/branchmergeproposal.py	2010-10-03 15:30:06 +0000
+++ lib/lp/code/interfaces/branchmergeproposal.py	2010-10-07 14:14:12 +0000
@@ -85,6 +85,7 @@
     IJob,
     IJobSource,
     IRunnableJob,
+    ITwistedJobSource,
     )
 
 
@@ -551,7 +552,7 @@
         """Destroy this object."""
 
 
-class IBranchMergeProposalJobSource(IJobSource):
+class IBranchMergeProposalJobSource(ITwistedJobSource):
     """A job source that will get all supported merge proposal jobs."""
 
 

=== modified file 'lib/lp/registry/interfaces/persontransferjob.py'
--- lib/lp/registry/interfaces/persontransferjob.py	2010-10-07 14:14:09 +0000
+++ lib/lp/registry/interfaces/persontransferjob.py	2010-10-07 14:14:12 +0000
@@ -61,6 +61,16 @@
 class IMembershipNotificationJob(IPersonTransferJob):
     """A Job to notify new members of a team of that change."""
 
+    member = PublicPersonChoice(
+        title=_('Alias for minor_person attribute'),
+        vocabulary='ValidPersonOrTeam',
+        required=True)
+
+    team = PublicPersonChoice(
+        title=_('Alias for major_person attribute'),
+        vocabulary='ValidPersonOrTeam',
+        required=True)
+
 
 class IMembershipNotificationJobSource(IJobSource):
     """An interface for acquiring IMembershipNotificationJobs."""

=== modified file 'lib/lp/registry/model/persontransferjob.py'
--- lib/lp/registry/model/persontransferjob.py	2010-10-07 14:14:09 +0000
+++ lib/lp/registry/model/persontransferjob.py	2010-10-07 14:14:12 +0000
@@ -104,16 +104,6 @@
         # but the DB representation is unicode.
         self._json_data = json_data.decode('utf-8')
 
-    @classmethod
-    def get(cls, key):
-        """Return the instance of this class whose key is supplied."""
-        store = IMasterStore(PersonTransferJob)
-        instance = store.get(cls, key)
-        if instance is None:
-            raise SQLObjectNotFound(
-                'No occurrence of %s has key %s' % (cls.__name__, key))
-        return instance
-
 
 class PersonTransferJobDerived(BaseRunnableJob):
     """Intermediate class for deriving from PersonTransferJob.

=== modified file 'lib/lp/registry/stories/person/xx-approve-members.txt'
--- lib/lp/registry/stories/person/xx-approve-members.txt	2009-10-26 14:37:31 +0000
+++ lib/lp/registry/stories/person/xx-approve-members.txt	2010-10-07 14:14:12 +0000
@@ -23,6 +23,9 @@
     >>> browser.getControl(name='action_7').value = ['approve']
     >>> browser.getControl(name='comment').value = 'Thanks for your interest'
     >>> browser.getControl('Save changes').click()
+    >>> from lp.testing.mail_helpers import run_mail_jobs
+    >>> login(ANONYMOUS)
+    >>> run_mail_jobs()
     >>> len(stub.test_emails)
     12
     >>> for from_addr, to_addrs, raw_msg in sorted(stub.test_emails):

=== added file 'lib/lp/registry/tests/test_membership_notification_job_creation.py'
--- lib/lp/registry/tests/test_membership_notification_job_creation.py	1970-01-01 00:00:00 +0000
+++ lib/lp/registry/tests/test_membership_notification_job_creation.py	2010-10-07 14:14:12 +0000
@@ -0,0 +1,65 @@
+# Copyright 2010 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Test that MembershipNotificationJobs are created appropriately."""
+
+__metaclass__ = type
+
+from zope.component import getUtility
+
+from canonical.testing import LaunchpadFunctionalLayer
+from lp.registry.interfaces.person import IPersonSet
+from lp.registry.interfaces.persontransferjob import (
+    IMembershipNotificationJobSource,
+    )
+from lp.registry.interfaces.teammembership import (
+    ITeamMembershipSet,
+    TeamMembershipStatus,
+    )
+from lp.registry.model.persontransferjob import MembershipNotificationJob
+from lp.services.job.interfaces.job import JobStatus
+from lp.testing import (
+    login_person,
+    TestCaseWithFactory,
+    )
+from lp.testing.sampledata import ADMIN_EMAIL
+
+
+class CreateMembershipNotificationJobTest(TestCaseWithFactory):
+    """Test that MembershipNotificationJobs are created appropriately."""
+    layer = LaunchpadFunctionalLayer
+
+    def setUp(self):
+        super(CreateMembershipNotificationJobTest, self).setUp()
+        self.person = self.factory.makePerson(name='murdock')
+        self.team = self.factory.makeTeam(name='a-team')
+        self.job_source = getUtility(IMembershipNotificationJobSource)
+
+    def test_setstatus_admin(self):
+        login_person(self.team.teamowner)
+        self.team.addMember(self.person, self.team.teamowner)
+        membership_set = getUtility(ITeamMembershipSet)
+        tm = membership_set.getByPersonAndTeam(self.person, self.team)
+        tm.setStatus(TeamMembershipStatus.ADMIN, self.team.teamowner)
+        jobs = list(self.job_source.iterReady())
+        job_info = [
+            (job.__class__, job.member, job.team, job.status)
+            for job in jobs]
+        self.assertEqual(
+            [(MembershipNotificationJob,
+              self.person,
+              self.team,
+              JobStatus.WAITING),
+            ],
+            job_info)
+
+    def test_setstatus_silent(self):
+        person_set = getUtility(IPersonSet)
+        admin = person_set.getByEmail(ADMIN_EMAIL)
+        login_person(admin)
+        self.team.addMember(self.person, self.team.teamowner)
+        membership_set = getUtility(ITeamMembershipSet)
+        tm = membership_set.getByPersonAndTeam(self.person, self.team)
+        tm.setStatus(
+            TeamMembershipStatus.ADMIN, admin, silent=True)
+        self.assertEqual([], list(self.job_source.iterReady()))

=== added file 'lib/lp/registry/tests/test_process_job_sources_cronjob.py'
--- lib/lp/registry/tests/test_process_job_sources_cronjob.py	1970-01-01 00:00:00 +0000
+++ lib/lp/registry/tests/test_process_job_sources_cronjob.py	2010-10-07 14:14:12 +0000
@@ -0,0 +1,125 @@
+# Copyright 2010 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Test cron script for processing jobs from any job source class."""
+
+__metaclass__ = type
+
+import os
+import subprocess
+
+import transaction
+from zope.component import getUtility
+
+from canonical.config import config
+from canonical.launchpad.scripts.tests import run_script
+from canonical.testing import LaunchpadScriptLayer
+from lp.registry.interfaces.teammembership import (
+    ITeamMembershipSet,
+    TeamMembershipStatus,
+    )
+from lp.testing import (
+    login_person,
+    TestCaseWithFactory,
+    )
+
+
+class ProcessJobSourceTest(TestCaseWithFactory):
+    """Test the process-job-source.py script."""
+    layer = LaunchpadScriptLayer
+    script = 'cronscripts/process-job-source.py'
+
+    def tearDown(self):
+        super(ProcessJobSourceTest, self).tearDown()
+        self.layer.force_dirty_database()
+
+    def test_missing_argument(self):
+        # The script should display usage info when called without any
+        # arguments.
+        returncode, output, error = run_script(
+            self.script, [], expect_returncode=1)
+        self.assertIn('Usage:', output)
+        self.assertIn('process-job-source.py [options] JOB_SOURCE', output)
+
+    def test_empty_queue(self):
+        # The script should just create a lockfile and exit if no jobs
+        # are in the queue.
+        returncode, output, error = run_script(
+            self.script, ['IMembershipNotificationJobSource'])
+        expected = (
+            'INFO    Creating lockfile: .*launchpad-process-job-'
+            'source-IMembershipNotificationJobSource.lock.*')
+        self.assertTextMatchesExpressionIgnoreWhitespace(expected, error)
+
+    def test_processed(self):
+        # The script should output the number of jobs it processed.
+        person = self.factory.makePerson(name='murdock')
+        team = self.factory.makeTeam(name='a-team')
+        login_person(team.teamowner)
+        team.addMember(person, team.teamowner)
+        membership_set = getUtility(ITeamMembershipSet)
+        tm = membership_set.getByPersonAndTeam(person, team)
+        tm.setStatus(TeamMembershipStatus.ADMIN, team.teamowner)
+        transaction.commit()
+        returncode, output, error = run_script(
+            self.script, ['-v', 'IMembershipNotificationJobSource'])
+        self.assertIn(
+            ('DEBUG   Running <MEMBERSHIP_NOTIFICATION branch job (1) '
+             'for murdock as part of a-team. status=Waiting>'),
+            error)
+        self.assertIn('DEBUG   MembershipNotificationJob sent email', error)
+        self.assertIn('Ran 1 MembershipNotificationJob jobs.', error)
+
+
+class ProcessJobSourceGroupsTest(TestCaseWithFactory):
+    """Test the process-job-source-groups.py script."""
+    layer = LaunchpadScriptLayer
+    script = 'cronscripts/process-job-source-groups.py'
+
+    def tearDown(self):
+        super(ProcessJobSourceGroupsTest, self).tearDown()
+        self.layer.force_dirty_database()
+
+    def test_missing_argument(self):
+        # The script should display usage info when called without any
+        # arguments.
+        returncode, output, error = run_script(
+            self.script, [], expect_returncode=1)
+        self.assertIn(
+            ('Usage: process-job-source-groups.py '
+             '[ -e JOB_SOURCE ] GROUP [GROUP]...'),
+            output)
+        self.assertIn('-e JOB_SOURCE, --exclude=JOB_SOURCE', output)
+        self.assertIn('At least one group must be specified.', output)
+        self.assertIn('Group: MAIN\n    IMembershipNotificationJobSource',
+                      output)
+
+    def test_empty_queue(self):
+        # The script should just create a lockfile, launch a child for
+        # each job source class, and then exit if no jobs are in the queue.
+        returncode, output, error = run_script(self.script, ['MAIN'])
+        expected = (
+            '.*Creating lockfile:.*launchpad-processjobsourcegroups.lock'
+            '.*Creating lockfile:.*launchpad-process-job-'
+            'source-IMembershipNotificationJobSource.lock.*')
+        self.assertTextMatchesExpressionIgnoreWhitespace(expected, error)
+
+    def test_processed(self):
+        # The script should output the number of jobs that have been
+        # processed by its child processes.
+        person = self.factory.makePerson(name='murdock')
+        team = self.factory.makeTeam(name='a-team')
+        login_person(team.teamowner)
+        team.addMember(person, team.teamowner)
+        membership_set = getUtility(ITeamMembershipSet)
+        tm = membership_set.getByPersonAndTeam(person, team)
+        tm.setStatus(TeamMembershipStatus.ADMIN, team.teamowner)
+        transaction.commit()
+        returncode, output, error = run_script(
+            self.script, ['-v', '--wait', 'MAIN'])
+        self.assertTextMatchesExpressionIgnoreWhitespace(
+            ('DEBUG Running <MEMBERSHIP_NOTIFICATION branch job (.*) '
+             'for murdock as part of a-team. status=Waiting>'),
+            error)
+        self.assertIn('DEBUG   MembershipNotificationJob sent email', error)
+        self.assertIn('Ran 1 MembershipNotificationJob jobs.', error)

=== modified file 'lib/lp/services/job/interfaces/job.py'
--- lib/lp/services/job/interfaces/job.py	2010-08-20 20:31:18 +0000
+++ lib/lp/services/job/interfaces/job.py	2010-10-07 14:14:12 +0000
@@ -11,6 +11,7 @@
     'IJob',
     'IJobSource',
     'IRunnableJob',
+    'ITwistedJobSource',
     'JobStatus',
     'LeaseHeld',
     ]
@@ -166,3 +167,10 @@
 
     def contextManager():
         """Get a context for running this kind of job in."""
+
+
+class ITwistedJobSource(IJobSource):
+    """Interface for a job source that is usable by the TwistedJobRunner."""
+
+    def get(id):
+        """Get a job by its id."""

=== modified file 'lib/lp/services/job/runner.py'
--- lib/lp/services/job/runner.py	2010-09-20 09:48:58 +0000
+++ lib/lp/services/job/runner.py	2010-10-07 14:14:12 +0000
@@ -77,7 +77,6 @@
 
     # We redefine __eq__ and __ne__ here to prevent the security proxy
     # from mucking up our comparisons in tests and elsewhere.
-
     def __eq__(self, job):
         return (
             self.__class__ is removeSecurityProxy(job.__class__)
@@ -337,13 +336,12 @@
     """Run Jobs via twisted."""
 
     def __init__(self, job_source, dbuser, logger=None, error_utility=None):
-        env = {'PYTHONPATH': os.environ['PYTHONPATH'],
-               'PATH': os.environ['PATH']}
-        lp_config = os.environ.get('LPCONFIG')
-        if lp_config is not None:
-            env['LPCONFIG'] = lp_config
+        env = {'PATH': os.environ['PATH']}
+        for name in ('PYTHONPATH', 'LPCONFIG'):
+            if name in os.environ:
+                env[name] = os.environ[name]
         starter = main.ProcessStarter(
-            packages=('twisted', 'ampoule'), env=env)
+            packages=('_pythonpath', 'twisted', 'ampoule'), env=env)
         super(TwistedJobRunner, self).__init__(logger, error_utility)
         self.job_source = job_source
         import_name = '%s.%s' % (
@@ -367,7 +365,8 @@
         transaction.commit()
         job_id = job.id
         deadline = timegm(job.lease_expires.timetuple())
-        self.logger.debug('Running %r, lease expires %s', job, job.lease_expires)
+        self.logger.debug(
+            'Running %r, lease expires %s', job, job.lease_expires)
         deferred = self.pool.doWork(
             RunJobCommand, job_id = job_id, _deadline=deadline)
         def update(response):
@@ -442,14 +441,21 @@
 class JobCronScript(LaunchpadCronScript):
     """Base class for scripts that run jobs."""
 
-    def __init__(self, runner_class=JobRunner, test_args=None,
-                 script_name=None):
-        self.dbuser = getattr(config, self.config_name).dbuser
-        if script_name is None:
-            script_name = self.config_name
+    config_name = None
+
+    def __init__(self, runner_class=JobRunner, test_args=None, name=None):
         super(JobCronScript, self).__init__(
-            script_name, self.dbuser, test_args)
-        self.runner_class = runner_class
+            name=name, dbuser=None, test_args=test_args)
+        self._runner_class = runner_class
+
+    @property
+    def dbuser(self):
+        return self.config_section.dbuser
+
+    @property
+    def runner_class(self):
+        """Enable subclasses to override with command-line arguments."""
+        return self._runner_class
 
     def job_counts(self, jobs):
         """Return a list of tuples containing the job name and counts."""
@@ -458,8 +464,18 @@
             counts[job.__class__.__name__] += 1
         return sorted(counts.items())
 
+    @property
+    def config_section(self):
+        return getattr(config, self.config_name)
+
     def main(self):
-        errorlog.globalErrorUtility.configure(self.config_name)
+        section = self.config_section
+        if (getattr(section, 'error_dir', None) is not None
+            and getattr(section, 'oops_prefix', None) is not None
+            and getattr(section, 'copy_to_zlog', None) is not None):
+            # If the three variables are not set, we will let the error
+            # utility default to using the [error_reports] config.
+            errorlog.globalErrorUtility.configure(self.config_name)
         job_source = getUtility(self.source_interface)
         runner = self.runner_class.runFromSource(
             job_source, self.dbuser, self.logger)

=== modified file 'lib/lp/services/scripts/base.py'
--- lib/lp/services/scripts/base.py	2010-10-03 15:30:06 +0000
+++ lib/lp/services/scripts/base.py	2010-10-07 14:14:12 +0000
@@ -37,9 +37,6 @@
     )
 from canonical.lp import initZopeless
 
-from lp.services.log.mappingfilter import MappingFilter
-from lp.services.log.nullhandler import NullHandler
-
 
 LOCK_PATH = "/var/lock/"
 UTC = pytz.UTC
@@ -152,22 +149,21 @@
         useful in test scripts.
         """
         if name is None:
-            self.name = self.__class__.__name__.lower()
-        else:
-            self.name = name
-
-        self.dbuser = dbuser
-
-        if self.description is None:
-            description = self.__doc__
-        else:
-            description = self.description
+            self._name = self.__class__.__name__.lower()
+        else:
+            self._name = name
+
+        self._dbuser = dbuser
 
         # The construction of the option parser is a bit roundabout, but
         # at least it's isolated here. First we build the parser, then
         # we add options that our logger object uses, then call our
         # option-parsing hook, and finally pull out and store the
         # supplied options and args.
+        if self.description is None:
+            description = self.__doc__
+        else:
+            description = self.description
         self.parser = OptionParser(usage=self.usage,
                                    description=description)
         scripts.logger_options(self.parser, default=self.loglevel)
@@ -177,9 +173,23 @@
                     "profiling stats in FILE."))
         self.add_my_options()
         self.options, self.args = self.parser.parse_args(args=test_args)
-        self.logger = scripts.logger(self.options, name)
-
-        self.lockfilepath = os.path.join(LOCK_PATH, self.lockfilename)
+
+        # Enable subclasses to easily override these __init__()
+        # arguments using command-line arguments.
+        self.handle_options()
+
+    def handle_options(self):
+        self.logger = scripts.logger(self.options, self.name)
+
+    @property
+    def name(self):
+        """Enable subclasses to override with command-line arguments."""
+        return self._name
+
+    @property
+    def dbuser(self):
+        """Enable subclasses to override with command-line arguments."""
+        return self._dbuser
 
     #
     # Hooks that we expect users to redefine.
@@ -226,6 +236,10 @@
         """
         return "launchpad-%s.lock" % self.name
 
+    @property
+    def lockfilepath(self):
+        return os.path.join(LOCK_PATH, self.lockfilename)
+
     def setup_lock(self):
         """Create lockfile.
 
@@ -275,8 +289,10 @@
 
     @log_unhandled_exception_and_exit
     def run(self, use_web_security=False, implicit_begin=True,
-            isolation=ISOLATION_LEVEL_DEFAULT):
+            isolation=None):
         """Actually run the script, executing zcml and initZopeless."""
+        if isolation is None:
+            isolation = ISOLATION_LEVEL_DEFAULT
         self._init_zca(use_web_security=use_web_security)
         self._init_db(implicit_begin=implicit_begin, isolation=isolation)
 
@@ -339,17 +355,15 @@
     """Logs successful script runs in the database."""
 
     def __init__(self, name=None, dbuser=None, test_args=None):
-        """Initialize, and sys.exit() if the cronscript is disabled.
-
-        Rather than hand editing crontab files, cronscripts can be
-        enabled and disabled using a config file.
-
-        The control file location is specified by
-        config.canonical.cron_control_url.
-        """
         super(LaunchpadCronScript, self).__init__(name, dbuser, test_args)
 
-        name = self.name
+        # self.name is used instead of the name argument, since it may have
+        # have been overridden by command-line parameters or by
+        # overriding the name property.
+        enabled = cronscript_enabled(
+            config.canonical.cron_control_url, self.name, self.logger)
+        if not enabled:
+            sys.exit(0)
 
         # Configure the IErrorReportingUtility we use with defaults.
         # Scripts can override this if they want.
@@ -359,12 +373,10 @@
         globalErrorUtility.copy_to_zlog = False
 
         # WARN and higher log messages should generate OOPS reports.
-        logging.getLogger().addHandler(OopsHandler(name))
-
-        enabled = cronscript_enabled(
-            config.canonical.cron_control_url, name, self.logger)
-        if not enabled:
-            sys.exit(0)
+        # self.name is used instead of the name argument, since it may have
+        # have been overridden by command-line parameters or by
+        # overriding the name property.
+        logging.getLogger().addHandler(OopsHandler(self.name))
 
     @log_unhandled_exception_and_exit
     def record_activity(self, date_started, date_completed):

=== modified file 'lib/lp/testing/__init__.py'
--- lib/lp/testing/__init__.py	2010-09-29 20:29:43 +0000
+++ lib/lp/testing/__init__.py	2010-10-07 14:14:12 +0000
@@ -441,6 +441,16 @@
                 "Expected %s to be %s, but it was %s."
                 % (attribute_name, date, getattr(sql_object, attribute_name)))
 
+    def assertTextMatchesExpressionIgnoreWhitespace(self,
+                                                    regular_expression_txt,
+                                                    text):
+        def normalise_whitespace(text):
+            return ' '.join(text.split())
+        pattern = re.compile(
+            normalise_whitespace(regular_expression_txt), re.S)
+        self.assertIsNot(
+            None, pattern.search(normalise_whitespace(text)), text)
+
     def assertIsInstance(self, instance, assert_class):
         """Assert that an instance is an instance of assert_class.
 
@@ -686,16 +696,6 @@
         super(BrowserTestCase, self).setUp()
         self.user = self.factory.makePerson(password='test')
 
-    def assertTextMatchesExpressionIgnoreWhitespace(self,
-                                                    regular_expression_txt,
-                                                    text):
-        def normalise_whitespace(text):
-            return ' '.join(text.split())
-        pattern = re.compile(
-            normalise_whitespace(regular_expression_txt), re.S)
-        self.assertIsNot(
-            None, pattern.search(normalise_whitespace(text)), text)
-
     def getViewBrowser(self, context, view_name=None, no_login=False):
         login(ANONYMOUS)
         url = canonical_url(context, view_name=view_name)