Merge lp:~mwhudson/launchpad/reduce-concurrent-job-count into lp:launchpad

[Michael Hudson-Doyle]

Allow a code import machine to set its own limit for how many jobs to run, and lower the default to 3.

[Michael Hudson-Doyle (mwhudson)]

As discussed in the linked bug, we should probably do something more sophisticated here to give a machine a reasonable number of jobs to run, but this is much easier for sure.

[Michael Nelson (michael.nelson)]

Hi Michael,

doesn't this need to be a change on the lp-production-configs? Currently there I see:

production/launchpad-lazr.conf
55:max_jobs_per_machine: 2

review: Needs Information (code)

[Michael Hudson-Doyle (mwhudson)]

Hm, I don't think that config setting takes effect at the moment. I'll investigate further!

[Michael Hudson-Doyle (mwhudson)]

Yeah, it doesn't apply, so I'll remove it: https://code.edge.launchpad.net/~mwhudson/lp-production-configs/remove-production-max_workers_per_machine/+merge/19671

[Tim Penhey (thumper)]

review: Approve

[Michael Hudson-Doyle (mwhudson)]

OK, so as discussed in IRC, I've made the worker limit settable on the code-import-dispatcher command line, and lowered the default to 3. Please re-review.

[Tim Penhey (thumper)]

lib/lp/codehosting/codeimport/dispatcher.py

You have:

    if worker_limit is None:
        self.worker_limit = \
            config.codeimportdispatcher.max_jobs_per_machine

Which doesn't set self.worker_limit if worker_limit is passed in.

    if worker_limit is None:
        worker_limit = config.codeimportdispatcher.max_jobs_per_machine
    self.worker_limit = worker_limit

perhaps.

How hard would it be to add an int Option? That way the OptionParser object does the integer validation and conversion for you.

  --max-jobs

review: Needs Fixing

[Michael Hudson-Doyle (mwhudson)]

Tim,

can you rereview now please?

Cheers,
mwh

[Tim Penhey (thumper)]

Looks good now

review: Approve

=== modified file 'cronscripts/code-import-dispatcher.py'
--- cronscripts/code-import-dispatcher.py	2009-10-13 14:38:07 +0000
+++ cronscripts/code-import-dispatcher.py	2010-02-22 19:42:20 +0000
@@ -18,6 +18,12 @@
 
 class CodeImportDispatcherScript(LaunchpadScript):
 
+    def add_my_options(self):
+        self.parser.add_option(
+            "--max-jobs", dest="max_jobs", type=int,
+            default=config.codeimportdispatcher.max_jobs_per_machine,
+            help="The maximum number of jobs to run on this machine.")
+
     def run(self, use_web_security=False, implicit_begin=True,
             isolation=None):
         """See `LaunchpadScript.run`.
@@ -30,7 +36,8 @@
     def main(self):
         globalErrorUtility.configure('codeimportdispatcher')
 
-        CodeImportDispatcher(self.logger).findAndDispatchJob(
+        dispatcher = CodeImportDispatcher(self.logger, self.options.max_jobs)
+        dispatcher.findAndDispatchJob(
             ServerProxy(config.codeimportdispatcher.codeimportscheduler_url))
 
 

=== modified file 'lib/canonical/config/schema-lazr.conf'
--- lib/canonical/config/schema-lazr.conf	2010-02-20 04:13:47 +0000
+++ lib/canonical/config/schema-lazr.conf	2010-02-22 19:42:20 +0000
@@ -449,7 +449,7 @@
 codeimportscheduler_url: http://xmlrpc-private.launchpad.dev:8087/codeimportscheduler
 
 # The maximum number of jobs to run on a machine at one time.
-max_jobs_per_machine: 10
+max_jobs_per_machine: 3
 
 # See [error_reports].
 copy_to_zlog: False

=== modified file 'lib/lp/code/doc/xmlrpc-codeimport-scheduler.txt'
--- lib/lp/code/doc/xmlrpc-codeimport-scheduler.txt	2009-10-22 11:55:51 +0000
+++ lib/lp/code/doc/xmlrpc-codeimport-scheduler.txt	2010-02-22 19:42:20 +0000
@@ -32,12 +32,9 @@
 getJobForMachine(), that returns the id of the job that the code
 import slave should next run.
 
-    >>> codeimportscheduler_api.getJobForMachine('bazaar-importer')
+    >>> codeimportscheduler_api.getJobForMachine('bazaar-importer', 2)
     1
 
-    >>> from canonical.database.sqlbase import flush_database_updates
-    >>> flush_database_updates()
-
 The method just calls the 'getJobForMachine' method from the
 ICodeImportJobSet interface, and tests all the details of what it does
 can be found in the tests for ICodeImportJobSet.
@@ -49,7 +46,7 @@
     >>> codeimportscheduler = xmlrpclib.ServerProxy(
     ...     'http://xmlrpc-private.launchpad.dev:8087/codeimportscheduler',
     ...     transport=XMLRPCTestTransport())
-    >>> codeimportscheduler.getJobForMachine('bazaar-importer')
+    >>> codeimportscheduler.getJobForMachine('bazaar-importer', 2)
     0
 
 This includes the behaviour of auto-creating machine rows for
@@ -60,7 +57,7 @@
     >>> print getUtility(ICodeImportMachineSet).getByHostname(
     ...     'doesnt-exist-yet')
     None
-    >>> codeimportscheduler.getJobForMachine('doesnt-exist-yet')
+    >>> codeimportscheduler.getJobForMachine('doesnt-exist-yet', 1)
     0
     >>> new_machine = getUtility(ICodeImportMachineSet).getByHostname(
     ...     'doesnt-exist-yet')

=== modified file 'lib/lp/code/interfaces/codeimportjob.py'
--- lib/lp/code/interfaces/codeimportjob.py	2009-06-25 04:06:00 +0000
+++ lib/lp/code/interfaces/codeimportjob.py	2010-02-22 19:42:20 +0000
@@ -122,7 +122,7 @@
     # we implement endpoint specific authentication for the private xml-rpc
     # server.
 
-    def getJobForMachine(hostname):
+    def getJobForMachine(hostname, worker_limit):
         """Select a job for the given machine to run and mark it as started.
 
         If there is not already a CodeImportMachine with the given hostname,

=== modified file 'lib/lp/code/interfaces/codeimportmachine.py'
--- lib/lp/code/interfaces/codeimportmachine.py	2009-06-25 04:06:00 +0000
+++ lib/lp/code/interfaces/codeimportmachine.py	2010-02-22 19:42:20 +0000
@@ -47,7 +47,7 @@
         description=_("When the controller deamon last recorded it was"
                       " running."))
 
-    def shouldLookForJob():
+    def shouldLookForJob(worker_limit):
         """Should we look for a job to run on this machine?
 
         There are three reasons we might not look for a job:

=== modified file 'lib/lp/code/interfaces/codeimportscheduler.py'
--- lib/lp/code/interfaces/codeimportscheduler.py	2009-06-25 04:06:00 +0000
+++ lib/lp/code/interfaces/codeimportscheduler.py	2010-02-22 19:42:20 +0000
@@ -28,7 +28,7 @@
     when they need more work to do.
     """
 
-    def getJobForMachine(hostname):
+    def getJobForMachine(hostname, worker_limit):
         """Get a job to run on the slave 'hostname'.
 
         This method selects the most appropriate job for the machine,

=== modified file 'lib/lp/code/model/codeimportjob.py'
--- lib/lp/code/model/codeimportjob.py	2010-01-27 02:48:13 +0000
+++ lib/lp/code/model/codeimportjob.py	2010-02-22 19:42:20 +0000
@@ -102,7 +102,7 @@
         except SQLObjectNotFound:
             return None
 
-    def getJobForMachine(self, hostname):
+    def getJobForMachine(self, hostname, worker_limit):
         """See `ICodeImportJobSet`."""
         job_workflow = getUtility(ICodeImportJobWorkflow)
         for job in self.getReclaimableJobs():
@@ -111,7 +111,7 @@
         if machine is None:
             machine = getUtility(ICodeImportMachineSet).new(
                 hostname, CodeImportMachineState.ONLINE)
-        elif not machine.shouldLookForJob():
+        elif not machine.shouldLookForJob(worker_limit):
             return None
         job = CodeImportJob.selectOne(
             """id IN (SELECT id FROM CodeImportJob

=== modified file 'lib/lp/code/model/codeimportmachine.py'
--- lib/lp/code/model/codeimportmachine.py	2009-06-25 04:06:00 +0000
+++ lib/lp/code/model/codeimportmachine.py	2010-02-22 19:42:20 +0000
@@ -51,7 +51,7 @@
         'CodeImportEvent', joinColumn='machine',
         orderBy=['-date_created', '-id'])
 
-    def shouldLookForJob(self):
+    def shouldLookForJob(self, worker_limit):
         """See `ICodeImportMachine`."""
         job_count = self.current_jobs.count()
 
@@ -64,8 +64,7 @@
                     CodeImportMachineOfflineReason.QUIESCED)
             return False
         elif self.state == CodeImportMachineState.ONLINE:
-            max_jobs = config.codeimportdispatcher.max_jobs_per_machine
-            return job_count < max_jobs
+            return job_count < worker_limit
         else:
             raise AssertionError(
                 "Unknown machine state %r??" % self.state)

=== modified file 'lib/lp/code/model/tests/test_codeimport.py'
--- lib/lp/code/model/tests/test_codeimport.py	2010-02-01 03:55:59 +0000
+++ lib/lp/code/model/tests/test_codeimport.py	2010-02-22 19:42:20 +0000
@@ -195,7 +195,7 @@
 
     def makeApprovedImportWithRunningJob(self):
         code_import = self.makeApprovedImportWithPendingJob()
-        job = CodeImportJobSet().getJobForMachine('machine')
+        job = CodeImportJobSet().getJobForMachine('machine', 10)
         self.assertEqual(code_import.import_job, job)
         return code_import
 

=== modified file 'lib/lp/code/model/tests/test_codeimportjob.py'
--- lib/lp/code/model/tests/test_codeimportjob.py	2010-02-01 03:55:59 +0000
+++ lib/lp/code/model/tests/test_codeimportjob.py	2010-02-22 19:42:20 +0000
@@ -114,7 +114,7 @@
     def assertJobIsSelected(self, desired_job):
         """Assert that the expected job is chosen by getJobForMachine."""
         observed_job = getUtility(ICodeImportJobSet).getJobForMachine(
-            self.machine.hostname)
+            self.machine.hostname, 10)
         self.assert_(observed_job is not None, "No job was selected.")
         self.assertEqual(desired_job, observed_job,
                          "Expected job not selected.")
@@ -122,7 +122,7 @@
     def assertNoJobSelected(self):
         """Assert that no job is selected."""
         observed_job = getUtility(ICodeImportJobSet).getJobForMachine(
-            'machine')
+            'machine', 10)
         self.assert_(observed_job is None, "Job unexpectedly selected.")
 
     def test_nothingSelectedIfNothingCreated(self):
@@ -274,7 +274,7 @@
         machine = self.factory.makeCodeImportMachine(set_online=True)
         login(ANONYMOUS)
         getUtility(ICodeImportJobSet).getJobForMachine(
-            machine.hostname)
+            machine.hostname, 10)
         login_for_code_imports()
         # Now there are no reclaimable jobs.
         self.assertReclaimableJobs([])

=== modified file 'lib/lp/code/model/tests/test_codeimportmachine.py'
--- lib/lp/code/model/tests/test_codeimportmachine.py	2009-06-25 04:06:00 +0000
+++ lib/lp/code/model/tests/test_codeimportmachine.py	2010-02-22 19:42:20 +0000
@@ -9,7 +9,6 @@
 
 from zope.component import getUtility
 
-from canonical.config import config
 from canonical.database.constants import UTC_NOW
 from lp.code.enums import (
     CodeImportMachineOfflineReason, CodeImportMachineState)
@@ -39,13 +38,13 @@
     def test_machineIsOffline(self):
         # When the machine is offline, we shouldn't look for any jobs.
         self.machine.setOffline(CodeImportMachineOfflineReason.STOPPED)
-        self.assertFalse(self.machine.shouldLookForJob())
+        self.assertFalse(self.machine.shouldLookForJob(10))
 
     def test_machineIsQuiescingNoJobsRunning(self):
         # When the machine is quiescing and no jobs are running on this
         # machine, we should set the machine to OFFLINE and not look for jobs.
         self.machine.setQuiescing(self.factory.makePerson())
-        self.assertFalse(self.machine.shouldLookForJob())
+        self.assertFalse(self.machine.shouldLookForJob(10))
         self.assertEqual(self.machine.state, CodeImportMachineState.OFFLINE)
 
     def test_machineIsQuiescingWithJobsRunning(self):
@@ -53,20 +52,19 @@
         # machine, we shouldn't look for any more jobs.
         self.createJobRunningOnMachine(self.machine)
         self.machine.setQuiescing(self.factory.makePerson())
-        self.assertFalse(self.machine.shouldLookForJob())
+        self.assertFalse(self.machine.shouldLookForJob(10))
         self.assertEqual(self.machine.state, CodeImportMachineState.QUIESCING)
 
     def test_enoughJobsRunningOnMachine(self):
         # When there are already enough jobs running on this machine, we
         # shouldn't look for any more jobs.
-        for i in range(config.codeimportdispatcher.max_jobs_per_machine):
-            self.createJobRunningOnMachine(self.machine)
-        self.assertFalse(self.machine.shouldLookForJob())
+        self.createJobRunningOnMachine(self.machine)
+        self.assertFalse(self.machine.shouldLookForJob(worker_limit=1))
 
     def test_shouldLook(self):
         # If the machine is online and there are not already
         # max_jobs_per_machine jobs running, then we should look for a job.
-        self.assertTrue(self.machine.shouldLookForJob())
+        self.assertTrue(self.machine.shouldLookForJob(worker_limit=1))
 
     def test_noHeartbeatWhenCreated(self):
         # Machines are created with a NULL heartbeat.
@@ -75,18 +73,18 @@
     def test_noHeartbeatUpdateWhenOffline(self):
         # When the machine is offline, the heartbeat is not updated.
         self.machine.setOffline(CodeImportMachineOfflineReason.STOPPED)
-        self.machine.shouldLookForJob()
+        self.machine.shouldLookForJob(10)
         self.assertTrue(self.machine.heartbeat is None)
 
     def test_heartbeatUpdateWhenQuiescing(self):
         # When the machine is quiescing, the heartbeat is updated.
         self.machine.setQuiescing(self.factory.makePerson())
-        self.machine.shouldLookForJob()
+        self.machine.shouldLookForJob(10)
         self.assertSqlAttributeEqualsDate(self.machine, 'heartbeat', UTC_NOW)
 
     def test_heartbeatUpdateWhenOnline(self):
         # When the machine is online, the heartbeat is updated.
-        self.machine.shouldLookForJob()
+        self.machine.shouldLookForJob(10)
         self.assertSqlAttributeEqualsDate(self.machine, 'heartbeat', UTC_NOW)
 
 

=== modified file 'lib/lp/code/xmlrpc/codeimportscheduler.py'
--- lib/lp/code/xmlrpc/codeimportscheduler.py	2009-06-25 04:06:00 +0000
+++ lib/lp/code/xmlrpc/codeimportscheduler.py	2010-02-22 19:42:20 +0000
@@ -22,9 +22,10 @@
 
     implements(ICodeImportScheduler)
 
-    def getJobForMachine(self, hostname):
+    def getJobForMachine(self, hostname, worker_limit):
         """See `ICodeImportScheduler`."""
-        job = getUtility(ICodeImportJobSet).getJobForMachine(hostname)
+        job = getUtility(ICodeImportJobSet).getJobForMachine(
+            hostname, worker_limit)
         if job is not None:
             return job.id
         else:

=== modified file 'lib/lp/codehosting/codeimport/dispatcher.py'
--- lib/lp/codehosting/codeimport/dispatcher.py	2009-06-25 04:06:00 +0000
+++ lib/lp/codehosting/codeimport/dispatcher.py	2010-02-22 19:42:20 +0000
@@ -32,12 +32,13 @@
     worker_script = os.path.join(
         config.root, 'scripts', 'code-import-worker-db.py')
 
-    def __init__(self, logger):
+    def __init__(self, logger, worker_limit):
         """Initialize an instance.
 
         :param logger: A `Logger` object.
         """
         self.logger = logger
+        self.worker_limit = worker_limit
 
     def getHostname(self):
         """Return the hostname of this machine.
@@ -66,7 +67,8 @@
     def findAndDispatchJob(self, scheduler_client):
         """Check for and dispatch a job if necessary."""
 
-        job_id = scheduler_client.getJobForMachine(self.getHostname())
+        job_id = scheduler_client.getJobForMachine(
+            self.getHostname(), self.worker_limit)
 
         if job_id == 0:
             self.logger.info("No jobs pending.")

=== modified file 'lib/lp/codehosting/codeimport/tests/test_dispatcher.py'
--- lib/lp/codehosting/codeimport/tests/test_dispatcher.py	2009-12-14 18:11:07 +0000
+++ lib/lp/codehosting/codeimport/tests/test_dispatcher.py	2010-02-22 19:42:20 +0000
@@ -11,64 +11,60 @@
 import shutil
 import socket
 import tempfile
-from textwrap import dedent
 from unittest import TestLoader
 
-from twisted.trial.unittest import TestCase
-
-from canonical.config import config
-from lp.codehosting.codeimport.dispatcher import CodeImportDispatcher
 from canonical.launchpad import scripts
 from canonical.launchpad.scripts.logger import QuietFakeLogger
-from canonical.testing.layers import TwistedLaunchpadZopelessLayer
+from canonical.testing.layers import BaseLayer
+
+from lp.codehosting.codeimport.dispatcher import CodeImportDispatcher
+from lp.testing import TestCase
 
 
 class StubSchedulerClient:
-    """A stub scheduler client that returns a pre-arranged answer."""
+    """A scheduler client that returns a pre-arranged answer."""
 
     def __init__(self, id_to_return):
         self.id_to_return = id_to_return
 
-    def getJobForMachine(self, machine):
+    def getJobForMachine(self, machine, limit):
         return self.id_to_return
 
 
+class MockSchedulerClient:
+    """A scheduler client that records calls to `getJobForMachine`."""
+
+    def __init__(self):
+        self.calls = []
+
+    def getJobForMachine(self, machine, limit):
+        self.calls.append((machine, limit))
+        return 0
+
+
 class TestCodeImportDispatcherUnit(TestCase):
     """Unit tests for `CodeImportDispatcher`."""
 
-    layer = TwistedLaunchpadZopelessLayer
+    layer = BaseLayer
 
     def setUp(self):
-        self.config_count = 0
-        self.pushConfig(forced_hostname='none')
-        self.dispatcher = CodeImportDispatcher(QuietFakeLogger())
-
-    def pushConfig(self, **args):
-        """Push some key-value pairs into the codeimportdispatcher config.
-
-        The config values will be restored during test tearDown.
-        """
-        self.config_count += 1
-        name = 'test%d' % self.config_count
-        body = '\n'.join(["%s: %s"%(k, v) for k, v in args.iteritems()])
-        config.push(name, dedent("""
-            [codeimportdispatcher]
-            %s
-            """ % body))
-        self.addCleanup(config.pop, name)
+        TestCase.setUp(self)
+        self.pushConfig('codeimportdispatcher', forced_hostname='none')
+
+    def makeDispatcher(self, worker_limit=10):
+        """Make a `CodeImportDispatcher`."""
+        return CodeImportDispatcher(QuietFakeLogger(), worker_limit)
 
     def test_getHostname(self):
         # By default, getHostname return the same as socket.gethostname()
-        self.assertEqual(
-            self.dispatcher.getHostname(),
-            socket.gethostname())
+        dispatcher = self.makeDispatcher()
+        self.assertEqual(socket.gethostname(), dispatcher.getHostname())
 
     def test_getHostnameOverride(self):
         # getHostname can be overriden by the config for testing, however.
-        self.pushConfig(forced_hostname='test-value')
-        self.assertEqual(
-            self.dispatcher.getHostname(),
-            'test-value')
+        dispatcher = self.makeDispatcher()
+        self.pushConfig('codeimportdispatcher', forced_hostname='test-value')
+        self.assertEqual('test-value', dispatcher.getHostname())
 
     def writePythonScript(self, script_path, script_body):
         """Write out an executable Python script.
@@ -94,6 +90,7 @@
 
         # We create a script that writes its command line arguments to
         # some a temporary file and examine that.
+        dispatcher = self.makeDispatcher()
         tmpdir = tempfile.mkdtemp()
         self.addCleanup(shutil.rmtree, tmpdir)
         script_path = os.path.join(tmpdir, 'script.py')
@@ -102,26 +99,39 @@
             script_path,
             ['import sys',
              'open(%r, "w").write(str(sys.argv[1:]))' % output_path])
-        self.dispatcher.worker_script = script_path
-        proc = self.dispatcher.dispatchJob(10)
+        dispatcher.worker_script = script_path
+        proc = dispatcher.dispatchJob(10)
         proc.wait()
         arglist = self.filterOutLoggingOptions(eval(open(output_path).read()))
-        self.assertEqual(arglist, ['10'])
+        self.assertEqual(['10'], arglist)
 
     def test_findAndDispatchJob_jobWaiting(self):
-        # If there is a job to dispatch, then we call dispatchJob with its id.
+        # If there is a job to dispatch, then we call dispatchJob with its id
+        # and the worker_limit supplied to the dispatcher.
         calls = []
-        self.dispatcher.dispatchJob = lambda job_id: calls.append(job_id)
-        self.dispatcher.findAndDispatchJob(StubSchedulerClient(10))
-        self.assertEqual(calls, [10])
+        dispatcher = self.makeDispatcher()
+        dispatcher.dispatchJob = lambda job_id: calls.append(job_id)
+        dispatcher.findAndDispatchJob(StubSchedulerClient(10))
+        self.assertEqual([10], calls)
 
     def test_findAndDispatchJob_noJobWaiting(self):
         # If there is no job to dispatch, then we just exit quietly.
         calls = []
-        self.dispatcher.dispatchJob = lambda job_id: calls.append(job_id)
-        self.dispatcher.findAndDispatchJob(StubSchedulerClient(0))
-        self.assertEqual(calls, [])
+        dispatcher = self.makeDispatcher()
+        dispatcher.dispatchJob = lambda job_id: calls.append(job_id)
+        dispatcher.findAndDispatchJob(StubSchedulerClient(0))
+        self.assertEqual([], calls)
 
+    def test_findAndDispatchJob_calls_getJobForMachine_with_limit(self):
+        # findAndDispatchJob calls getJobForMachine on the scheduler client
+        # with the hostname and supplied worker limit.
+        worker_limit = self.factory.getUniqueInteger()
+        dispatcher = self.makeDispatcher(worker_limit)
+        scheduler_client = MockSchedulerClient()
+        dispatcher.findAndDispatchJob(scheduler_client)
+        self.assertEqual(
+            [(dispatcher.getHostname(), worker_limit)],
+            scheduler_client.calls)
 
 def test_suite():
     return TestLoader().loadTestsFromName(__name__)

=== modified file 'lib/lp/codehosting/codeimport/tests/test_workermonitor.py'
--- lib/lp/codehosting/codeimport/tests/test_workermonitor.py	2010-02-03 19:29:27 +0000
+++ lib/lp/codehosting/codeimport/tests/test_workermonitor.py	2010-02-22 19:42:20 +0000
@@ -566,7 +566,7 @@
             code_import.updateFromData(
                 {'review_status': CodeImportReviewStatus.REVIEWED},
                 self.factory.makePerson())
-        job = getUtility(ICodeImportJobSet).getJobForMachine('machine')
+        job = getUtility(ICodeImportJobSet).getJobForMachine('machine', 10)
         self.assertEqual(code_import, job.code_import)
         return job