Merge lp:~adeuring/launchpad/celery-config into lp:launchpad

[Abel Deuring]

configuration for the Celery based job runner.

This branch adds configuration details we need for the Celery based
job runner.

Celery reads its configuration from a Python file (we use
lp.services.job.celeryconfig). This module reads the "real config data"
from our regular lp.services.config.config module.

lp.services.job.celeryconfig already existed; I added several parameters
that were missing, and I added some sanity checks.

Note that the config module is used in two different places: The app
servers load it in order to send messages to a RabbitMQ server; several
celeryd instances consume these messages and run the jobs described by
them.

Normally, a celeryconfig modules looks like so:

PARAMETER_1 = value_1
PARAMETER_2 = value_2

i.e., the variables are set when the module is loaded. This makes it
difficult to test variants of the configuration -- re-loading a Python
module can be quite painful, so I moved this into a function configure().

The check for circular chain of linked fallback queues is quite naive and
inefficient, but I think it is good enough for our purposes: It is
unlikely that we will have, let's say, 500 different queues where some
complex fallback chains need to be tested.

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

= Launchpad lint =

Checking for conflicts and issues in changed files.

Linting changed files:
  lib/lp/services/config/schema-lazr.conf
  lib/lp/services/job/celeryconfig.py
  lib/lp/services/job/tests/test_celery_configuration.py

./lib/lp/services/config/schema-lazr.conf
     501: Line exceeds 80 characters.
    1112: Line exceeds 80 characters.
    1119: Line exceeds 80 characters.
    1711: Line exceeds 80 characters.

These are lines I did not change.

[Aaron Bentley (abentley)]

I think we want the timeout to be 5 minutes. That's what it's currently set to, and it will ensure that ~99% of tasks run in the fast lane.

It's generally a bad idea to assume that sys.argv is defined. It won't be when Python is loaded as an extension. I've been bitten by that in Launchpad work before. If you must use sys.argv, you need to handle the case where it's not defined.

Since we expect to use binding keys that match the queue names, please don't require the binding key to be specified.

I think it would be nicer if configure accepted "argv" and returned a dict. Then you could do "globals().update(configure(getattr(sys, 'argv', [''])))" to actually set it. That would make it easier to test, and also shorter. You could even pull the function into a different module so that testing didn't have to actually set the globals.

check_job_specific_celeryd_configutartion is misspelled.

review: Needs Fixing

[Abel Deuring (adeuring)]

fixed

review: Resubmit

[Aaron Bentley (abentley)]

review: Approve

=== modified file 'lib/lp/services/config/schema-lazr.conf'
--- lib/lp/services/config/schema-lazr.conf	2012-04-06 17:28:25 +0000
+++ lib/lp/services/config/schema-lazr.conf	2012-04-18 17:27:21 +0000
@@ -1955,3 +1955,38 @@
 module: lp.answers.interfaces.questionjob
 dbuser: answertracker
 crontab_group: MAIN
+
+[job_runner_queues]
+# The names of all queues.
+queues: job job_slow branch_write_job branch_write_job_slow
+
+# The main job queue.
+[job]
+# The maximum job run time in seconds. When a job runs longer, Celery
+# terminates it with a SoftTimeLimitExceeded exception.
+timeout: 300
+# If a job times out, it will be queued again in the fallback queue.
+fallback_queue: job_slow
+concurrency: 3
+
+# The queue for jobs that time out in the queue "job".
+[job_slow]
+# The maximum job run time in seconds. When a job runs longer, Celery
+# terminates it with a SoftTimeLimitExceeded exception.
+timeout: 86400
+fallback_queue:
+concurrency: 1
+
+[branch_write_job]
+# The maximum job run time in seconds. When a job runs longer, Celery
+# terminates it with a SoftTimeLimitExceeded exception.
+timeout: 300
+fallback_queue: branch_write_job_slow
+concurrency: 3
+
+[branch_write_job_slow]
+# The maximum job run time in seconds. When a job runs longer, Celery
+# terminates it with a SoftTimeLimitExceeded exception.
+timeout: 86400
+fallback_queue:
+concurrency: 1

=== modified file 'lib/lp/services/job/celeryconfig.py'
--- lib/lp/services/job/celeryconfig.py	2012-04-10 20:24:43 +0000
+++ lib/lp/services/job/celeryconfig.py	2012-04-18 17:27:21 +0000
@@ -1,16 +1,88 @@
+# Copyright 2012 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+import argparse
+import sys
 from lp.services.config import config
-host, port = config.rabbitmq.host.split(':')
-BROKER_HOST = host
-BROKER_PORT = port
-BROKER_USER = config.rabbitmq.userid
-BROKER_PASSWORD = config.rabbitmq.password
-BROKER_VHOST = config.rabbitmq.virtual_host
-CELERY_IMPORTS = ("lp.services.job.celeryjob", )
-CELERY_RESULT_BACKEND = "amqp"
-CELERY_QUEUES = {
-    "branch_write_job": {"binding_key": "branch_write_job"},
-    "job": {"binding_key": "job"},
-}
-CELERY_DEFAULT_EXCHANGE = "job"
-CELERY_DEFAULT_QUEUE = "job"
-CELERY_CREATE_MISSING_QUEUES = False
+
+
+class ConfigurationError(Exception):
+    """Errors raised due to misconfiguration."""
+
+
+def check_circular_fallbacks(queue):
+    """Check for curcular fallback queues.
+
+    A circular chain of fallback queues could keep a job forever queued
+    if it times out in all queues.
+    """
+    linked_queues = []
+    while config[queue].fallback_queue != '':
+        linked_queues.append(queue)
+        queue = config[queue].fallback_queue
+        if queue in linked_queues:
+            raise ConfigurationError(
+                'Circular chain of fallback queues: '
+                '%s already in %s' % (queue, linked_queues))
+
+
+def configure(argv):
+    """Set the Celery parameters.
+
+    Doing this in a function is convenient for testing.
+    """
+    result = {}
+    celery_queues = {}
+    queue_names = config.job_runner_queues.queues
+    queue_names = queue_names.split(' ')
+    for queue_name in queue_names:
+        celery_queues[queue_name] = {
+            'binding_key': queue_name,
+            }
+        check_circular_fallbacks(queue_name)
+
+    parser = argparse.ArgumentParser()
+    parser.add_argument('-Q',  '--queues')
+    args = parser.parse_known_args(argv)
+    queues = args[0].queues
+    # A queue must be specified as a command line parameter for each
+    # celeryd instance, but this is not required for a Launchpad app server.
+    if 'celeryd' in argv[0]:
+        if queues is None or queues == '':
+            raise ConfigurationError('A queue must be specified.')
+        queues = queues.split(',')
+        # Allow only one queue per celeryd instance. More than one queue
+        # would require a check for consistent timeout values, and especially
+        # a better way to specify a fallback queue.
+        if len(queues) > 1:
+            raise ConfigurationError(
+                'A celeryd instance may serve only one queue.')
+        queue = queues[0]
+        if queue not in celery_queues:
+            raise ConfigurationError(
+                'Queue %s is not configured in schema-lazr.conf' % queue)
+        result['CELERYD_TASK_SOFT_TIME_LIMIT'] = config[queue].timeout
+        if config[queue].fallback_queue != '':
+            result['FALLBACK'] = config[queue].fallback_queue
+        result['CELERYD_CONCURRENCY'] = config[queue].concurrency
+
+    host, port = config.rabbitmq.host.split(':')
+
+    result['BROKER_HOST'] = host
+    result['BROKER_PORT'] = port
+    result['BROKER_USER'] = config.rabbitmq.userid
+    result['BROKER_PASSWORD'] = config.rabbitmq.password
+    result['BROKER_VHOST'] = config.rabbitmq.virtual_host
+    result['CELERY_CREATE_MISSING_QUEUES'] = False
+    result['CELERY_DEFAULT_EXCHANGE'] = 'job'
+    result['CELERY_DEFAULT_QUEUE'] = 'job'
+    result['CELERY_IMPORTS'] = ("lp.services.job.celeryjob", )
+    result['CELERY_QUEUES'] = celery_queues
+    result['CELERY_RESULT_BACKEND'] = 'amqp'
+    return result
+
+try:
+    globals().update(configure(getattr(sys, 'argv', [''])))
+except ConfigurationError, error:
+    print >>sys.stderr, error
+    sys.exit(1)

=== added file 'lib/lp/services/job/tests/test_celery_configuration.py'
--- lib/lp/services/job/tests/test_celery_configuration.py	1970-01-01 00:00:00 +0000
+++ lib/lp/services/job/tests/test_celery_configuration.py	2012-04-18 17:27:21 +0000
@@ -0,0 +1,148 @@
+# Copyright 2012 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+from contextlib import contextmanager
+
+from lp.services.config import config
+from lp.testing import TestCase
+from lp.testing.layers import RabbitMQLayer
+
+
+@contextmanager
+def changed_config(changes):
+    config.push('test_changes', changes)
+    yield
+    config.pop('test_changes')
+
+
+class TestCeleryConfiguration(TestCase):
+    layer = RabbitMQLayer
+
+    def check_default_common_parameters(self, config):
+        # Tests for default config values that are set for app servers
+        # and for celeryd instances.
+
+        # Four queues are defined; the binding key for each queue is
+        # just the queue name.
+        queue_names = [
+            'branch_write_job', 'branch_write_job_slow', 'job', 'job_slow']
+        queues = config['CELERY_QUEUES']
+        self.assertEqual(queue_names, sorted(queues))
+        for name in queue_names:
+            self.assertEqual(name, queues[name]['binding_key'])
+
+        self.assertEqual('localhost', config['BROKER_HOST'])
+        # BROKER_PORT changes between test runs, so just check that it
+        # is defined.
+        self.assertTrue('BROKER_PORT' in config)
+        self.assertEqual('guest', config['BROKER_USER'])
+        self.assertEqual('guest', config['BROKER_PASSWORD'])
+        self.assertEqual('/', config['BROKER_VHOST'])
+        self.assertFalse(config['CELERY_CREATE_MISSING_QUEUES'])
+        self.assertEqual('job', config['CELERY_DEFAULT_EXCHANGE'])
+        self.assertEqual('job', config['CELERY_DEFAULT_QUEUE'])
+        self.assertEqual(
+            ('lp.services.job.celeryjob', ), config['CELERY_IMPORTS'])
+        self.assertEqual('amqp', config['CELERY_RESULT_BACKEND'])
+
+    def test_app_server_configuration(self):
+        from lp.services.job.celeryconfig import configure
+        config = configure([''])
+        self.check_default_common_parameters(config)
+
+    def check_job_specific_celeryd_configuration(self, expected, config):
+        self.check_default_common_parameters(config)
+        self.assertEqual(
+            expected['concurrency'], config['CELERYD_CONCURRENCY'])
+        self.assertEqual(
+            expected['timeout'], config['CELERYD_TASK_SOFT_TIME_LIMIT'])
+        self.assertEqual(
+            expected['fallback'], config.get('FALLBACK', None))
+
+    def test_default_celeryd_configuration_fast_lanes(self):
+        from lp.services.job.celeryconfig import configure
+        expected = {
+            'concurrency': 3,
+            'fallback': 'job_slow',
+            'timeout': 300,
+            }
+        config = configure(['celeryd', '-Q', 'job'])
+        self.check_default_common_parameters(config)
+        self.check_job_specific_celeryd_configuration(expected, config)
+        config = configure(['celeryd', '-Q', 'branch_write_job'])
+        self.check_default_common_parameters(config)
+        expected['fallback'] = 'branch_write_job_slow'
+        self.check_job_specific_celeryd_configuration(expected, config)
+
+    def test_default_celeryd_configuration_slow_lanes(self):
+        from lp.services.job.celeryconfig import configure
+        expected = {
+            'concurrency': 1,
+            'fallback': None,
+            'timeout': 86400,
+            }
+        config = configure(['celeryd', '-Q', 'job_slow'])
+        self.check_default_common_parameters(config)
+        self.check_job_specific_celeryd_configuration(expected, config)
+        config = configure(['celeryd', '-Q', 'branch_write_job_slow'])
+        self.check_default_common_parameters(config)
+        self.check_job_specific_celeryd_configuration(expected, config)
+
+    def test_circular_fallback_lanes(self):
+        # Circular fallback lanes are detected.
+        # Import late because the RabbitMQ parameters are set during layer
+        # setup.
+        from lp.services.job.celeryconfig import (
+            ConfigurationError,
+            configure,
+            )
+        with changed_config(
+            """
+            [job_slow]
+            fallback_queue: job
+        """):
+            error = (
+                "Circular chain of fallback queues: job already in "
+                "['job', 'job_slow']"
+                )
+            self.assertRaisesWithContent(
+                ConfigurationError, error, configure, [''])
+
+    def test_missing_queue_parameter_for_celeryd(self):
+        # An exception is raised when celeryd is started without
+        # the parameter -Q.
+        # Import late because the RabbitMQ parameters are set during layer
+        # setup.
+        from lp.services.job.celeryconfig import (
+            ConfigurationError,
+            configure,
+            )
+        error = 'A queue must be specified.'
+        self.assertRaisesWithContent(
+            ConfigurationError, error, configure, ['celeryd'])
+
+    def test_two_queues_for_celeryd(self):
+        # An exception is raised when celeryd is started for two queues.
+        # Import late because the RabbitMQ parameters are set during layer
+        # setup.
+        from lp.services.job.celeryconfig import (
+            ConfigurationError,
+            configure,
+            )
+        error = 'A celeryd instance may serve only one queue.'
+        self.assertRaisesWithContent(
+            ConfigurationError, error, configure,
+            ['celeryd', '--queue=job,branch_write_job'])
+
+    def test_unconfigured_queue_for_celeryd(self):
+        # An exception is raised when celeryd is started for a queue that
+        # is not configured.
+        # Import late because the RabbitMQ parameters are set during layer
+        # setup.
+        from lp.services.job.celeryconfig import (
+            ConfigurationError,
+            configure,
+            )
+        error = 'Queue foo is not configured in schema-lazr.conf'
+        self.assertRaisesWithContent(
+            ConfigurationError, error, configure, ['celeryd', '--queue=foo'])