Récupérer la liste des tâches dans une file d'attente dans Celery

Comment puis-je récupérer une liste de tâches dans une file d'attente qui doivent encore être traitées?

MODIFIER: voir les autres réponses pour obtenir une liste des tâches dans la file d'attente.

Vous devriez regarder ici: Guide du céleri - Inspection des ouvriers

Fondamentalement ceci:

from celery.app.control import Inspect

# Inspect all nodes.
i = Inspect()

# Show the items that have an ETA or are scheduled for later processing
i.scheduled()

# Show tasks that are currently active.
i.active()

# Show tasks that have been claimed by workers
i.reserved()

Selon ce que vous voulez

si vous utilisez rabbitMQ, utilisez ceci dans le terminal:

sudo rabbitmqctl list_queues

il imprimera la liste des files d'attente avec le nombre de tâches en attente. par exemple:

Listing queues ...
0b27d8c59fba4974893ec22d478a7093    0
0e0a2da9828a48bc86fe993b210d984f    0
10@torob2.celery.pidbox 0
11926b79e30a4f0a9d95df61b6f402f7    0
15c036ad25884b82839495fb29bd6395    1
celerey_mail_worker@torob2.celery.pidbox    0
celery  166
celeryev.795ec5bb-a919-46a8-80c6-5d91d2fcf2aa   0
celeryev.faa4da32-a225-4f6c-be3b-d8814856d1b6   0

le nombre dans la colonne de droite correspond au nombre de tâches dans la file d'attente. ci-dessus, la file d'attente de céleri a 166 tâches en attente.

Si vous n'utilisez pas de tâches prioritaires, c'est en fait assez simple si vous utilisez Redis. Pour obtenir le nombre de tâches:

redis-cli -h HOST -p PORT -n DATABASE_NUMBER llen QUEUE_NAME

Mais, les tâches prioritaires utilisent une clé différente dans redis , donc l'image complète est légèrement plus compliquée. L'image complète est que vous devez interroger redis pour chaque priorité de tâche. En python (et du projet Flower), cela ressemble à:

PRIORITY_SEP = '\x06\x16'
DEFAULT_PRIORITY_STEPS = [0, 3, 6, 9]


def make_queue_name_for_pri(queue, pri):
    """Make a queue name for redis

    Celery uses PRIORITY_SEP to separate different priorities of tasks into
    different queues in Redis. Each queue-priority combination becomes a key in
    redis with names like:

     - batch1\x06\x163 <-- P3 queue named batch1

    There's more information about this in Github, but it doesn't look like it 
    will change any time soon:

      - https://github.com/celery/kombu/issues/422

    In that ticket the code below, from the Flower project, is referenced:

      - https://github.com/mher/flower/blob/master/flower/utils/broker.py#L135

    :param queue: The name of the queue to make a name for.
    :param pri: The priority to make a name with.
    :return: A name for the queue-priority pair.
    """
    if pri not in DEFAULT_PRIORITY_STEPS:
        raise ValueError('Priority not in priority steps')
    return '{0}{1}{2}'.format(*((queue, PRIORITY_SEP, pri) if pri else
                                (queue, '', '')))


def get_queue_length(queue_name='celery'):
    """Get the number of tasks in a celery queue.

    :param queue_name: The name of the queue you want to inspect.
    :return: the number of items in the queue.
    """
    priority_names = [make_queue_name_for_pri(queue_name, pri) for pri in
                      DEFAULT_PRIORITY_STEPS]
    r = redis.StrictRedis(
        host=settings.REDIS_HOST,
        port=settings.REDIS_PORT,
        db=settings.REDIS_DATABASES['CELERY'],
    )
    return sum([r.llen(x) for x in priority_names])

Si vous souhaitez obtenir une tâche réelle, vous pouvez utiliser quelque chose comme:

redis-cli -h HOST -p PORT -n DATABASE_NUMBER lrange QUEUE_NAME 0 -1

De là, vous devrez désérialiser la liste renvoyée. Dans mon cas, j'ai pu accomplir cela avec quelque chose comme:

r = redis.StrictRedis(
    host=settings.REDIS_HOST,
    port=settings.REDIS_PORT,
    db=settings.REDIS_DATABASES['CELERY'],
)
l = r.lrange('celery', 0, -1)
pickle.loads(base64.decodestring(json.loads(l[0])['body']))

Sachez simplement que la désérialisation peut prendre un moment et que vous devrez ajuster les commandes ci-dessus pour travailler avec différentes priorités.

Pour récupérer des tâches depuis le backend, utilisez ceci

from amqplib import client_0_8 as amqp
conn = amqp.Connection(host="localhost:5672 ", userid="guest",
                       password="guest", virtual_host="/", insist=False)
chan = conn.channel()
name, jobs, consumers = chan.queue_declare(queue="queue_name", passive=True)

Si vous utilisez Celery + Django, le moyen le plus simple d'inspecter les tâches à l'aide de commandes directement depuis votre terminal dans votre environnement virtuel ou en utilisant un chemin complet vers céleri:

Doc : http://docs.celleryproject.org/en/latest/userguide/workers.html?highlight=revoke#inspecting-workers

$ celery inspect reserved
$ celery inspect active
$ celery inspect registered
$ celery inspect scheduled

De plus, si vous utilisez Celery + RabbitMQ, vous pouvez inspecter la liste des files d'attente à l'aide de la commande suivante:

Plus d'infos : https://linux.die.net/man/1/rabbitmqctl

$ sudo rabbitmqctl list_queues

Une solution de copier-coller pour Redis avec sérialisation json:

def get_celery_queue_items(queue_name):
    import base64
    import json  

    # Get a configured instance of a celery app:
    from yourproject.celery import app as celery_app

    with celery_app.pool.acquire(block=True) as conn:
        tasks = conn.default_channel.client.lrange(queue_name, 0, -1)
        decoded_tasks = []

    for task in tasks:
        j = json.loads(task)
        body = json.loads(base64.b64decode(j['body']))
        decoded_tasks.append(body)

    return decoded_tasks

Cela fonctionne avec Django. N'oubliez pas de changer yourproject.celery.

Le module d'inspection du céleri semble ne connaître les tâches que du point de vue des travailleurs. Si vous souhaitez afficher les messages qui sont dans la file d'attente (qui n'ont pas encore été extraits par les travailleurs), je suggère d'utiliser pyrabbit , qui peut s'interfacer avec l'api http rabbitmq pour récupérer toutes sortes d'informations de la file d'attente.

Un exemple peut être trouvé ici: Récupérer la longueur de la file d'attente avec Celery (RabbitMQ, Django)

Je pense que le seul moyen d'obtenir les tâches en attente est de conserver une liste des tâches que vous avez commencées et de laisser la tâche se retirer de la liste lorsqu'elle a commencé.

Avec rabbitmqctl et list_queues, vous pouvez avoir un aperçu du nombre de tâches en attente, mais pas des tâches elles-mêmes: http://www.rabbitmq.com/man/rabbitmqctl.1.man.html

Si ce que vous voulez inclut la tâche en cours de traitement, mais que vous n'êtes pas encore terminé, vous pouvez conserver une liste de vos tâches et vérifier leurs états:

from tasks import add
result = add.delay(4, 4)

result.ready() # True if finished

Vous pouvez également laisser Celery stocker les résultats avec CELERY_RESULT_BACKEND et vérifier lesquelles de vos tâches ne s'y trouvent pas.

Cela a fonctionné pour moi dans mon application:

def get_celery_queue_active_jobs(queue_name):
    connection = <CELERY_APP_INSTANCE>.connection()

    try:
        channel = connection.channel()
        name, jobs, consumers = channel.queue_declare(queue=queue_name, passive=True)
        active_jobs = []

        def dump_message(message):
            active_jobs.append(message.properties['application_headers']['task'])

        channel.basic_consume(queue=queue_name, callback=dump_message)

        for job in range(jobs):
            connection.drain_events()

        return active_jobs
    finally:
        connection.close()

active_jobs sera une liste de chaînes correspondant aux tâches de la file d'attente.

N'oubliez pas d'échanger CELERY_APP_INSTANCE avec le vôtre.

Merci à @ashish de m'avoir pointé dans la bonne direction avec sa réponse ici: https://stackoverflow.com/a/19465670/9843399

Autant que je sache, Celery ne donne pas d'API pour examiner les tâches en attente dans la file d'attente. Ceci est spécifique au courtier. Si vous utilisez Redis comme courtier pour un exemple, alors l'examen des tâches en attente dans la celeryfile d'attente (par défaut) est aussi simple que:

1. se connecter à la base de données du courtier
2. lister les éléments de la celeryliste (commande LRANGE pour un exemple)

Gardez à l'esprit qu'il s'agit de tâches EN ATTENTE d'être sélectionnées par les travailleurs disponibles. Votre cluster peut avoir des tâches en cours d'exécution - celles-ci ne figureront pas dans cette liste car elles ont déjà été sélectionnées.

J'en suis venu à la conclusion que le meilleur moyen d'obtenir le nombre de travaux sur une file d'attente est d'utiliser rabbitmqctlcomme cela a été suggéré à plusieurs reprises ici. Pour permettre à tout utilisateur choisi d'exécuter la commande avec, sudoj'ai suivi les instructions ici (j'ai ignoré la modification de la partie du profil car cela ne me dérange pas de taper sudo avant la commande.)

J'ai également attrapé les jamesc grepet l' cutextrait de code et je l'ai enveloppé dans des appels de sous-processus.

from subprocess import Popen, PIPE
p1 = Popen(["sudo", "rabbitmqctl", "list_queues", "-p", "[name of your virtula host"], stdout=PIPE)
p2 = Popen(["grep", "-e", "^celery\s"], stdin=p1.stdout, stdout=PIPE)
p3 = Popen(["cut", "-f2"], stdin=p2.stdout, stdout=PIPE)
p1.stdout.close()
p2.stdout.close()
print("number of jobs on queue: %i" % int(p3.communicate()[0]))

from celery.task.control import inspect
def key_in_list(k, l):
    return bool([True for i in l if k in i.values()])

def check_task(task_id):
    task_value_dict = inspect().active().values()
    for task_list in task_value_dict:
        if self.key_in_list(task_id, task_list):
             return True
    return False

Si vous contrôlez le code des tâches, vous pouvez contourner le problème en laissant une tâche déclencher une nouvelle tentative triviale la première fois qu'elle s'exécute, puis en vérifiant inspect().reserved(). La nouvelle tentative enregistre la tâche avec le moteur de résultat, et céleri peut le voir. La tâche doit accepter selfou contextcomme premier paramètre afin que nous puissions accéder au nombre de tentatives.

@task(bind=True)
def mytask(self):
    if self.request.retries == 0:
        raise self.retry(exc=MyTrivialError(), countdown=1)
    ...

Cette solution est indépendante du courtier, c'est-à-dire. vous n'avez pas à vous soucier de savoir si vous utilisez RabbitMQ ou Redis pour stocker les tâches.

EDIT: après les tests, j'ai trouvé que ce n'était qu'une solution partielle. La taille de réservé est limitée au paramètre de prélecture pour le worker.

Avec subprocess.run:

import subprocess
import re
active_process_txt = subprocess.run(['celery', '-A', 'my_proj', 'inspect', 'active'],
                                        stdout=subprocess.PIPE).stdout.decode('utf-8')
return len(re.findall(r'worker_pid', active_process_txt))

Attention à changer my_projavecyour_proj