celery errback complex flows

Question

I'm trying to understand properly how Celery works (v5.2.7) and I'm doing a bunch of canvas experiments. Although I'm decently able to combine tasks with chains/ groups and chords, I'm really struggling at creating 'complex' flows (i.e. more than one errback) when errors happen arbitrarily in tasks. In particular, I'm trying to setup chains and groups of errbacks to be triggered in case of error.

I create the following tasks:

from celery import Celery
app = Celery('tasks', broker='redis://localhost:6379/0', backend='redis://localhost')

@app.task(name="tasks.raise_exception")
def raise_exception():
    raise Exception("raised exception!")

@app.task(name="tasks.error_handler")
def error_handler(request, exc, traceback):
    print('Task {0} raised exception: {1!r}
{2!r}'.format(
          request.id, exc, traceback))

@app.task(name="tasks.error_handler_forward_error")
def error_handler_forward_error(request, exc, traceback):
    print('Task {0} raised exception: {1!r}
{2!r}'.format(
          request.id, exc, traceback))
    return request, exc, traceback
    # raise exc

The last task is tested both trying to return the parameters (I suppose this should call its callback) or raise the received exc. Now I create some task flows in a producer process and see if I'm able to call chained errbacks in some way. I try both with using plain signature methods and canvas primitives.

from tasks import raise_exception, error_handler, error_handler_forward_error

# This doesn't work: error_handler is not executed
# here error_handler_forward_error RAISES exc
err1 = error_handler_forward_error.s()
err2 = error_handler.s()
err1.on_error(err2) 

s = raise_exception.s()
s.on_error(err1)
result = s.delay()

# This doesn't work: error_handler is not executed
# here error_handler_forward_error RETURNS request, exc, traceback
err1 = error_handler_forward_error.s()
err2 = error_handler.s()
err1.link(err2)

s = raise_exception.s()
s.on_error(err1)
result = s.delay()

# This doesn't work: no errback is executed
# here error_handler_forward_error RETURNS request, exc, traceback
c = (error_handler_forward_error.s() | error_handler.s())

s = raise_exception.s()
s.on_error(c)

result = s.delay()

In the last example, when trying to use a chain as a callback, nothing works and I get a huge amount of exception stacktrace. This seems at the core of the problem: TypeError: Object of type Context is not JSON serializable

Trying to create a group of errbacks neither works:

from celery import group

# this doesn't work: no errback is executed
g = group([error_handler.s() | error_handler.s()])

s = raise_exception.s()
s.on_error(g)

result = s.delay()

The error seems to be: TypeError: error_handler() missing 2 required positional arguments: 'exc' and 'traceback'

In fact, when leaving error_handler with just one parameter the group is executed:

@app.task(name="tasks.error_handler")
def error_handler(request):
    print("error_handler")
    print(f"request: {request}")

The request ID is printed, but I've lost the exception and stacktrace, which are equally important parameters.

I went through the entire manual, but it seems that there is no useful info to address the case of complex errbacks workflows. At the best of my knowledge, it seems to me that Celery poorly manages flows of error tasks, and the only way is to declare one unique errback to manage everything. Although this can work, it is inconvenient when the errback needs to do many things (e.g. update a db, send an email, etc.). I wish I could create independent error management tasks for such operations, which could possibly fail on their own and need to be independently retried upon db/network errors, etc. There should be a way, but I cannot find it.

celery errback complex flows

0 Answers