Celery: how can I route a failed task to a dead-letter queue

Question

I'm a newcomer to celery and I try to integrate this task queue into my project but I still don't figure out how celery handles the failed tasks and I'd like to keep all those in a amqp dead-letter queue.

According to the doc here it seems that raising Reject in a Task having acks_late enabled produces the same effect as acking the message and then we have a few words about dead-letter queues.

So I added a custom default queue to my celery config

celery_app.conf.update(CELERY_ACCEPT_CONTENT=['application/json'],
                       CELERY_TASK_SERIALIZER='json',
                       CELERY_QUEUES=[CELERY_QUEUE,
                                      CELERY_DLX_QUEUE],
                       CELERY_DEFAULT_QUEUE=CELERY_QUEUE_NAME,
                       CELERY_DEFAULT_EXCHANGE=CELERY_EXCHANGE
                       )

and my kombu objects are looking like

CELERY_DLX_EXCHANGE = Exchange(CELERY_DLX_EXCHANGE_NAME, type='direct')
CELERY_DLX_QUEUE = Queue(CELERY_DLX_QUEUE_NAME, exchange=DLX_EXCHANGE,
                             routing_key='celery-dlq')

DEAD_LETTER_CELERY_OPTIONS = {'x-dead-letter-exchange': CELERY_DLX_EXCHANGE_NAME,
                          'x-dead-letter-routing-key': 'celery-dlq'}

CELERY_EXCHANGE = Exchange(CELERY_EXCHANGE_NAME,
                               arguments=DEAD_LETTER_CELERY_OPTIONS,
                               type='direct')

CELERY_QUEUE = Queue(CELERY_QUEUE_NAME,
                         exchange=CELERY_EXCHANGE,
                         routing_key='celery-q')

And the task I'm executing is:

class HookTask(Task):
    acks_late = True

def run(self, ctx, data):
    logger.info('{0} starting {1.name}[{1.request.id}]'.format(self.__class__.__name__.upper(), self))
    self.hook_process(ctx, data)


def on_failure(self, exc, task_id, args, kwargs, einfo):
    logger.error('task_id %s failed, message: %s', task_id, exc.message)

def hook_process(self, t_ctx, body):
    # Build context
    ctx = TaskContext(self.request, t_ctx)
    logger.info('Task_id: %s, handling request %s', ctx.task_id, ctx.req_id)
    raise Reject('no_reason', requeue=False)

I made a little test with it but with no results when raising a Reject exception.

Now I'm wondering if it's a good idea to force the failed task route to the dead-letter queue by overriding the Task.on_failure. I think this would work but I also think that this solution is not so clean because according to what I red celery should do this all alone.

Thanks for your help.

Answer 1

I think you should not add arguments=DEAD_LETTER_CELERY_OPTIONS in CELERY_EXCHANGE. You should add it to CELERY_QUEUE with queue_arguments=DEAD_LETTER_CELERY_OPTIONS.

The following example is what I did and it works fine:

from celery import Celery
from kombu import Exchange, Queue
from celery.exceptions import Reject

app = Celery(
    'tasks',
    broker='amqp://guest@localhost:5672//',
    backend='redis://localhost:6379/0')

dead_letter_queue_option = {
    'x-dead-letter-exchange': 'dlx',
    'x-dead-letter-routing-key': 'dead_letter'
}

default_exchange = Exchange('default', type='direct')
dlx_exchange = Exchange('dlx', type='direct')

default_queue = Queue(
    'default',
    default_exchange,
    routing_key='default',
    queue_arguments=dead_letter_queue_option)
dead_letter_queue = Queue(
    'dead_letter', dlx_exchange, routing_key='dead_letter')

app.conf.task_queues = (default_queue, dead_letter_queue)

app.conf.task_default_queue = 'default'
app.conf.task_default_exchange = 'default'
app.conf.task_default_routing_key = 'default'


@app.task
def add(x, y):
    return x + y


@app.task(acks_late=True)
def div(x, y):
    try:
        z = x / y
        return z
    except ZeroDivisionError as exc:
        raise Reject(exc, requeue=False)

After the creation of queue, you should see that on the 'Features' column, it shows DLX (dead-letter-exchange) and DLK (dead-letter-routing-key) labels.

NOTE: You should delete the previous queues, if you have already created them in RabbitMQ. This is because celery won't delete the existing queue and re-create a new one.

Answer 2

I am having a similar case and I faced the same problems. I also wanted a solution that was based on configuration and not hard coded values. The proposed solution of Hengfeng Li was very helpfull and helped me understand the mechanism and the concepts. But there was a problem with the declaration of dead-letter queues. Specifically if you injected the DLQ in the task_default_queues, the Celery was consuming the queue and it was always empty. So a manual way of declaring DL(X/Q) was needed.

I used Celery's Bootsteps as they provide a good control on the stage that the code was run. My initial experiment was to create them exactly after the app creation but this created stalled connection after the forking of processes and it created an ugly exception. With a bootstep that runs exactly after the Pool step you can be guaranteed that it runs in the begining of each worker after it is forked and the connection pool is ready.

Finally I created a decorator that converts uncaught exceptions to task rejections by reraising with celery's Reject. Special care is taken for cases where a task is already decided on how to be handled, such as retries.

Here is a full working example. Try to run the task div.delay(1, 0) and see how it works.

from celery import Celery
from celery.exceptions import Reject, TaskPredicate
from functools import wraps
from kombu import Exchange, Queue

from celery import bootsteps


class Config(object):

    APP_NAME = 'test'

    task_default_queue = '%s_celery' % APP_NAME
    task_default_exchange = "%s_celery" % APP_NAME
    task_default_exchange_type = 'direct'
    task_default_routing_key = task_default_queue
    task_create_missing_queues = False
    task_acks_late = True

    # Configuration for DLQ support
    dead_letter_exchange = '%s_dlx' % APP_NAME
    dead_letter_exchange_type = 'direct'
    dead_letter_queue = '%s_dlq' % APP_NAME
    dead_letter_routing_key = dead_letter_queue


class DeclareDLXnDLQ(bootsteps.StartStopStep):
    """
    Celery Bootstep to declare the DL exchange and queues before the worker starts
        processing tasks
    """
    requires = {'celery.worker.components:Pool'}

    def start(self, worker):
        app = worker.app

        # Declare DLX and DLQ
        dlx = Exchange(
            app.conf.dead_letter_exchange,
            type=app.conf.dead_letter_exchange_type)

        dead_letter_queue = Queue(
            app.conf.dead_letter_queue,
            dlx,
            routing_key=app.conf.dead_letter_routing_key)

        with worker.app.pool.acquire() as conn:
            dead_letter_queue.bind(conn).declare()


app = Celery('tasks', broker='pyamqp://guest@localhost//')
app.config_from_object(Config)


# Declare default queues
# We bypass the default mechanism tha creates queues in order to declare special queue arguments for DLX support
default_exchange = Exchange(
    app.conf.task_default_exchange,
    type=app.conf.task_default_exchange_type)
default_queue = Queue(
        app.conf.task_default_queue,
        default_exchange,
        routing_key=app.conf.task_default_routing_key,
        queue_arguments={
            'x-dead-letter-exchange': app.conf.dead_letter_exchange,
            'x-dead-letter-routing-key': app.conf.dead_letter_routing_key
        })

# Inject the default queue in celery application
app.conf.task_queues = (default_queue,)

# Inject extra bootstep that declares DLX and DLQ
app.steps['worker'].add(DeclareDLXnDLQ)


def onfailure_reject(requeue=False):
    """
    When a task has failed it will raise a Reject exception so
    that the message will be requeued or marked for insertation in Dead Letter Exchange
    """

    def _decorator(f):
        @wraps(f)
        def _wrapper(*args, **kwargs):

            try:
                return f(*args, **kwargs)
            except TaskPredicate:
                raise   # Do not handle TaskPredicate like Retry or Reject
            except Exception as e:
                print("Rejecting")
                raise Reject(str(e), requeue=requeue)
        return _wrapper

    return _decorator


@app.task()
@onfailure_reject()
def div(x, y):
    return x / y

Edit: I updated the code to use the new configuration schema of celery (lower-case) as I found some compatibility issues in Celery 4.1.0.