cadasta-workertoolbox 0.1.17

Tracking Scheduled Tasks

To keep our system aware of all tasks being scheduled, the Cadasta Platform has a process running to consume task messages off of a task-monitor queue and insert those messages into our database. To support this design, all task producers (including worker nodes) must publish their task messages to both the normal destination queues and the task-monitor queue. This is acheived by registering all queues with a Topic Exchange, setting the task-monitor queue to subscribe to all messages sent to the exchange, and setting standard work queues to subscribe to messages with a matching routing_key. Being that the Cadasta Platform is designed to work with Amazon SQS and the SQS backend only keeps exchange/queue declarations in memory, each message producer must have this set up within their configuration. (For more reading on Exchanges, see the RabbitMQ Tutorials 1-5)

Tracking Task Results

Tasks results are inserted by each worker into the Platform DB. For this reason, it is important that each worker have network access to the Platform DB (via AWS Security Groups). Additionally, each worker should have a provided username and password that grants them authorization to write to the Platform DB’s Result Table. For reasons of security, it is advised that these credentials be permitted to only access this single table. The Result Table has a one-to-one relation via the task_id column to the Task Table. This should not be enforced via a constraint, as it is possible for a task’s result to be entered into the DB before the sync-tasks service enters the task into the Task Table.

cadasta.workertoolbox.conf.Config

The Config class was built to simplify configuring Celery settings, helping to ensure that all workers adhere to the architecture requirements of the Cadasta asynchronous system. It essentially offers a diff between Celery’s default configuration and the configuration required by our system. It is the aim of the class to not require much customization on the part of the developer, however some customization may be needed when altering configuration between environments (e.g. if dev settings vary greatly from prod settings).

Any Celery setting may be submitted. It is internal convention that we use the Celery’s newer lowercase settings rather than their older uppercase counterparts. This will ensure that they are displayed when calling repr on the Conf instance.

Once applied, all settings (and internal variables) are available on the Celery app instance’s app.conf object.

{
    'region': 'us-west-2',
    'queue_name_prefix': '{}-'.format(QUEUE_NAME_PREFIX)
}

set([
    Queue('celery', exchange, routing_key='celery'),
    Queue(platform_queue, exchange, routing_key='#'),
] + [
    Queue(q_name, exchange, routing_key=q_name)
    for q_name in queues
])

cadasta.workertoolbox.setup.setup_app

After the Celery application is provided a configuration object, there are other steups that must follow to properly configure the application. For example, the exchanges and queues described in the configuration must be declared. This function calls those required followup procedures. Typically, it is called automatically by the `worker_init <http://docs.celeryproject.org/en/latest/userguide/signals.html#worker-init>`__ signal, however it must be called manually by codebases that are run only as task producers or from within a Python shell.

It takes two arguments:

• app - A Celery() app instance. Required

• throw - Boolean stipulating if errors should be raise on failed setup. Otherwise, errors will simply be logged to the module logger at exception level. Optional, default: True

cadasta.workertoolbox.tests.build_functional_tests

When provided with a Celery app instance, this function generates a suite of functional tests to ensure that the provided application’s configuration and functionality conforms with the architecture of the Cadasta asynchronous system.

An example, where an instanciated and configured Celery() app instance exists in a parallel celery module:

from cadasta.workertoolbox.tests import build_functional_tests

from .celery import app

FunctionalTests = build_functional_tests(app)

To run these tests, use your standard test runner (e.g. pytest) or call manually from the command-line:

python -m unittest path/to/tests.py

Celery Best Practices and Gotchas

• To make the most of task performance, take advantage of Celery’s Canvas functionality.

• Avoid launching synchronous subtasks.

• Pursue granularity. when writing tasks.

• If your task schedules other tasks (eg a chain or chord), it is important to pass the parent task’s callbacks/errbacks to the last task of the subtasks. (TODO: Add helper function)

• At time of writing, chords with single tasks don’t respect callbacks/errbacks (celery/celery#3317, celery/celery#3709, celery/celery#3597).

Common Issues

celery.exceptions.NotRegistered: '...'

[2017-08-09 10:49:23,338: ERROR/MainProcess] Pool callback raised exception: Task of kind 'msg.email_err' never registered, please make sure it's imported.
Traceback (most recent call last):
  File "/Users/alukach/.virtualenvs/export-worker/lib/python3.6/site-packages/kombu/utils/objects.py", line 42, in __get__
    return obj.__dict__[self.__name__]
KeyError: 'type'

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/Users/alukach/.virtualenvs/export-worker/lib/python3.6/site-packages/billiard/pool.py", line 1747, in safe_apply_callback
    fun(*args, **kwargs)
  File "/Users/alukach/.virtualenvs/export-worker/lib/python3.6/site-packages/celery/worker/request.py", line 366, in on_failure
    self.id, exc, request=self, store_result=self.store_errors,
  File "/Users/alukach/.virtualenvs/export-worker/lib/python3.6/site-packages/celery/backends/base.py", line 168, in mark_as_failure
    self._call_task_errbacks(request, exc, traceback)
  File "/Users/alukach/.virtualenvs/export-worker/lib/python3.6/site-packages/celery/backends/base.py", line 174, in _call_task_errbacks
    if arity_greater(errback.type.__header__, 1):
  File "/Users/alukach/.virtualenvs/export-worker/lib/python3.6/site-packages/kombu/utils/objects.py", line 44, in __get__
    value = obj.__dict__[self.__name__] = self.__get(obj)
  File "/Users/alukach/.virtualenvs/export-worker/lib/python3.6/site-packages/celery/canvas.py", line 490, in type
    return self._type or self.app.tasks[self['task']]
  File "/Users/alukach/.virtualenvs/export-worker/lib/python3.6/site-packages/celery/app/registry.py", line 19, in __missing__
    raise self.NotRegistered(key)
celery.exceptions.NotRegistered: 'msg.email_err'

This occurs when a task experiences an exception and has an errback/link_error. For whatever reason, it attempts to run the task right then and there (as opposed to scheduling it in the queues). This is problematic if the errback is for a task not available in the worker’s codebase. For whatever reason, this does not seem to occur in tasks that are run within a chord. See celery/kombu#4022.

Testing

pip install -r requirements-test.txt
./runtests

Deploying

pip install -r requirements-deploy.txt
python setup.py test clean build tag publish