Celery job re-triggering

[poautran]

In GitLab by @woutdenolf on May 10, 2025, 15:49 GMT+2:

Flower stores stringified args and kwargs which get truncated so that's not a solution.

Store every task that arrives at a worker:

from celery.signals import task_received

@task_received.connect
def store_received_task(request=None, **kwargs):
    task_data = {
        "task_name": request.name,
        "args": request.args,
        "kwargs": request.kwargs,
    }
    with open("executed_tasks.log", "a") as f:
        f.write(json.dumps(task_data) + "\n")

or

@task_prerun.connect
def store_task_execution(task_id=None, task=None, args=None, kwargs=None, **_):
    task_data = {
        "task_id": task_id,
        "task_name": task.name,
        "args": args,
        "kwargs": kwargs,
    }
    with open("executed_tasks.log", "a") as f:
        f.write(json.dumps(task_data) + "\n")

Or on the client side

from celery.signals import before_task_publish
import json

@before_task_publish.connect
def log_task_args(sender=None, body=None, exchange=None, routing_key=None, headers=None, **_):
    task_data = {
        "task_name": sender,
        "args": body[0],
        "kwargs": body[1],
        "routing_key": routing_key,
        "exchange": exchange,
    }
    with open("submitted_tasks.log", "a") as f:
        f.write(json.dumps(task_data) + "\n")

See Celery signals docs: https://docs.celeryq.dev/en/stable/userguide/signals.html

Migrated from GitLab: https://gitlab.esrf.fr/workflow/ewoks/ewoksjob/-/issues/62

[poautran]

In GitLab by @woutdenolf on May 10, 2025, 16:26 GMT+2:

Or alternatively we create a celery-agnotic solution.