Merge lp:~abentley/launchpad/job-db-user into lp:launchpad

[Aaron Bentley]

[Aaron Bentley (abentley)]

= Summary =
Ensure Twisted job runner uses the correct database user. Currently,
update_preview_diff fails because it runs as lp_main, which is not permitted to
run on asuka.

== Proposed fix ==
Pass the database user as a parameter to the child process.

== Pre-implementation notes ==
None

== Implementation details ==
Unfortunately, Ampoule uses the JobRunnerProcess class as a ContextManager, not
the instance of JobRunnerProcess that it creates. I'll propose fixing that
in Ampoule itself, but this approach works in the meantime, and database users
are effectively global.

== Tests ==
bin/test test_runner -t timeout

== Demo and Q/A == Create a merge proposal on staging, get a LOSA to add
--twisted to the flags, push changes to the source branch, and see whether the
diff is updated.

= Launchpad lint =

Checking for conflicts. and issues in doctests and templates.
Running jslint, xmllint, pyflakes, and pylint.
Using normal rules.

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

== Pylint notices ==

lib/lp/services/job/runner.py
    31: [F0401] Unable to import 'lazr.delegates' (No module named delegates)

[Edwin Grubbs (edwin-grubbs)]

>=== modified file 'lib/lp/services/job/runner.py'
>--- lib/lp/services/job/runner.py 2010-02-16 19:36:10 +0000
>+++ lib/lp/services/job/runner.py 2010-02-24 20:07:05 +0000
>@@ -203,7 +204,7 @@
> return cls(job_class.iterReady(), logger)
>
> @classmethod
>- def runFromSource(cls, job_source, logger):
>+ def runFromSource(cls, job_source, dbuser, logger):
> """Run all ready jobs provided by the specified source."""
> with removeSecurityProxy(job_source.contextManager()):
> logger.info("Running synchronously.")
>@@ -364,10 +368,10 @@
> self.terminated()
>
> @classmethod
>- def runFromSource(cls, job_source, logger, error_utility=None):
>+ def runFromSource(cls, job_source, dbuser, logger, error_utility=None):
> """Run all ready jobs provided by the specified source."""
> logger.info("Running through Twisted.")
>- runner = cls(job_source, logger, error_utility)
>+ runner = cls(job_source, dbuser, logger, error_utility)
> reactor.callWhenRunning(runner.runAll)
> handler = getsignal(SIGCHLD)
> try:

Hi Aaron,

This looks good. Please make the runFromSource() parameter lists match
for the two classes and add info in the docstring about whether the
dbuser parameter is used or ignored. The diff

merge-conditional

-Edwin

review: Approve

=== modified file 'lib/lp/services/job/runner.py'
--- lib/lp/services/job/runner.py	2010-02-16 19:36:10 +0000
+++ lib/lp/services/job/runner.py	2010-02-24 21:47:18 +0000
@@ -31,6 +31,7 @@
 from lazr.delegates import delegates
 
 from canonical.config import config
+from canonical.lp import initZopeless
 from canonical.launchpad import scripts
 from canonical.launchpad.webapp import errorlog
 from canonical.twistedsupport.task import (
@@ -203,8 +204,11 @@
         return cls(job_class.iterReady(), logger)
 
     @classmethod
-    def runFromSource(cls, job_source, logger):
-        """Run all ready jobs provided by the specified source."""
+    def runFromSource(cls, job_source, dbuser, logger):
+        """Run all ready jobs provided by the specified source.
+
+        The dbuser parameter is ignored.
+        """
         with removeSecurityProxy(job_source.contextManager()):
             logger.info("Running synchronously.")
             runner = cls.fromReady(job_source, logger)
@@ -240,19 +244,22 @@
 class JobRunnerProcess(child.AMPChild):
     """Base class for processes that run jobs."""
 
-    def __init__(self, job_source_name):
+    def __init__(self, job_source_name, dbuser):
         child.AMPChild.__init__(self)
         module, name = job_source_name.rsplit('.', 1)
         source_module = __import__(module, fromlist=[name])
         self.job_source = getattr(source_module, name)
         self.context_manager = self.job_source.contextManager()
+        # icky, but it's really a global value anyhow.
+        self.__class__.dbuser = dbuser
 
-    @staticmethod
-    def __enter__():
+    @classmethod
+    def __enter__(cls):
         def handler(signum, frame):
             raise TimeoutError
         scripts.execute_zcml_for_scripts(use_web_security=False)
         signal(SIGHUP, handler)
+        initZopeless(dbuser=cls.dbuser)
 
     @staticmethod
     def __exit__(exc_type, exc_val, exc_tb):
@@ -284,7 +291,7 @@
 class TwistedJobRunner(BaseJobRunner):
     """Run Jobs via twisted."""
 
-    def __init__(self, job_source, logger=None, error_utility=None):
+    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')
@@ -297,8 +304,8 @@
         import_name = '%s.%s' % (
             removeSecurityProxy(job_source).__module__, job_source.__name__)
         self.pool = pool.ProcessPool(
-            JobRunnerProcess, ampChildArgs=[import_name], starter=starter,
-            min=0, timeout_signal=SIGHUP)
+            JobRunnerProcess, ampChildArgs=[import_name, str(dbuser)],
+            starter=starter, min=0, timeout_signal=SIGHUP)
 
     def runJobInSubprocess(self, job):
         """Run the job_class with the specified id in the process pool.
@@ -364,10 +371,13 @@
         self.terminated()
 
     @classmethod
-    def runFromSource(cls, job_source, logger, error_utility=None):
-        """Run all ready jobs provided by the specified source."""
+    def runFromSource(cls, job_source, dbuser, logger):
+        """Run all ready jobs provided by the specified source.
+
+        The dbuser parameter is not ignored.
+        """
         logger.info("Running through Twisted.")
-        runner = cls(job_source, logger, error_utility)
+        runner = cls(job_source, dbuser, logger)
         reactor.callWhenRunning(runner.runAll)
         handler = getsignal(SIGCHLD)
         try:
@@ -381,14 +391,15 @@
     """Base class for scripts that run jobs."""
 
     def __init__(self, runner_class=JobRunner):
-        dbuser = getattr(config, self.config_name).dbuser
-        super(JobCronScript, self).__init__(self.config_name, dbuser)
+        self.dbuser = getattr(config, self.config_name).dbuser
+        super(JobCronScript, self).__init__(self.config_name, self.dbuser)
         self.runner_class = runner_class
 
     def main(self):
         errorlog.globalErrorUtility.configure(self.config_name)
         job_source = getUtility(self.source_interface)
-        runner = self.runner_class.runFromSource(job_source, self.logger)
+        runner = self.runner_class.runFromSource(
+            job_source, self.dbuser, self.logger)
         self.logger.info(
             'Ran %d %s jobs.',
             len(runner.completed_jobs), self.source_interface.__name__)

=== modified file 'lib/lp/services/job/tests/test_runner.py'
--- lib/lp/services/job/tests/test_runner.py	2010-02-12 20:43:54 +0000
+++ lib/lp/services/job/tests/test_runner.py	2010-02-24 21:47:18 +0000
@@ -26,6 +26,8 @@
 from lp.services.job.model.job import Job
 from lp.testing import TestCaseWithFactory
 from canonical.launchpad.webapp import errorlog
+from canonical.launchpad.webapp.interfaces import (
+    IStoreSelector, MAIN_STORE, DEFAULT_FLAVOR)
 
 
 class NullJob(BaseRunnableJob):
@@ -283,6 +285,10 @@
     def run(self):
         if self.id == 2:
             sleep(30)
+        else:
+            store = getUtility(IStoreSelector).get(MAIN_STORE, DEFAULT_FLAVOR)
+            assert (
+                'user=branchscanner' in store._connection._raw_connection.dsn)
 
 
 class ListLogger:
@@ -306,7 +312,9 @@
         followed by a job that is sure to time out.
         """
         logger = ListLogger()
-        runner = TwistedJobRunner.runFromSource(StuckJob, logger)
+        runner = TwistedJobRunner.runFromSource(
+            StuckJob, 'branchscanner', logger)
+
         self.assertEqual(1, len(runner.completed_jobs))
         self.assertEqual(1, len(runner.incomplete_jobs))
         oops = errorlog.globalErrorUtility.getLastOopsReport()
@@ -327,7 +335,7 @@
         class DummyRunner:
 
             @classmethod
-            def runFromSource(cls, source, logger):
+            def runFromSource(cls, source, dbuser, logger):
                 expected_config = errorlog.ErrorReportingUtility()
                 expected_config.configure('update_preview_diffs')
                 self.assertEqual(