Comment vérifier l'état d'une tâche dans Celery ?

Question

Comment vérifier si une tâche est en cours d'exécution dans celery (plus précisément, j'utilise celery-django) ?

J'ai lu la documentation, et j'ai cherché sur Google, mais je ne vois pas d'appel comme celui-ci :

my_example_task.state() == RUNNING

Mon cas d'utilisation est que j'ai un service externe (java) pour le transcodage. Lorsque j'envoie un document à transcoder, je veux vérifier si la tâche qui exécute ce service est en cours d'exécution, et si non, la (re)lancer.

J'utilise les versions stables actuelles - 2.4, je crois.

Answer 1

Retourne le task_id (qui est donné par .delay()) et interroge l'instance celery sur l'état de la tâche :

x = method.delay(1,2)
print x.task_id

Lors de la demande, obtenez un nouveau AsyncResult en utilisant ce task_id :

from celery.result import AsyncResult
res = AsyncResult("your-task-id")
res.ready()

Answer 2

Créer un AsyncResult de l'identifiant de la tâche est de la manière recommandée dans le FAQ pour obtenir le statut de la tâche lorsque la seule chose que vous avez est l'id de la tâche.

Cependant, à partir de Celery 3.x, il y a des mises en garde importantes qui risquent de faire mal si l'on n'y prête pas attention. Cela dépend vraiment du scénario d'utilisation spécifique.

Par défaut, Celery n'enregistre pas d'état "en cours d'exécution".

Pour que Celery enregistre qu'une tâche est en cours d'exécution, vous devez configurer task_track_started a True . Voici une tâche simple qui permet de tester cela :

@app.task(bind=True)
def test(self):
    print self.AsyncResult(self.request.id).state

Quand task_track_started es False qui est la valeur par défaut, le spectacle d'état est PENDING même si la tâche a commencé. Si vous définissez task_track_started a True alors l'état sera STARTED .

L'État PENDING signifie "Je ne sais pas".

Un site AsyncResult avec l'État PENDING ne signifie rien de plus que le fait que Celery ne connaît pas l'état de la tâche. Cela peut être dû à un certain nombre de raisons.

D'abord, AsyncResult peut être construit avec des identifiants de tâches invalides. De telles "tâches" seront considérées comme en attente par Celery :

>>> task.AsyncResult("invalid").status
'PENDING'

Ok, donc personne ne va nourrir évidemment ids invalides à AsyncResult . D'accord, mais il a aussi pour effet que AsyncResult considérera également une tâche qui a été exécutée avec succès mais que Celery a oubliée comme étant PENDING . Encore une fois, dans certains cas d'utilisation cela peut être un problème. Une partie du problème dépend de la façon dont Celery est configuré pour conserver les résultats des tâches, car cela dépend de la disponibilité des "pierres tombales" dans le backend des résultats. ("Tombstones" est le terme utilisé dans la documentation de Celery pour désigner les morceaux de données qui enregistrent la façon dont la tâche s'est terminée). Utilisation de AsyncResult ne fonctionnera pas du tout si task_ignore_result es True . Un problème plus contrariant est que Celery expire les pierres tombales par défaut. Le site result_expires est réglé par défaut sur 24 heures. Ainsi, si vous lancez une tâche, et enregistrez l'id dans le stockage à long terme, et plus 24 heures plus tard, vous créez un AsyncResult avec elle, le statut sera PENDING .

Toutes les "vraies tâches" commencent dans le PENDING l'état. Donc, obtenir PENDING sur une tâche pourrait signifier que la tâche a été demandée mais n'a jamais progressé plus loin (pour une raison quelconque). Ou encore que la tâche a été exécutée mais que Celery a oublié son état.

Ouch ! AsyncResult ne fonctionnera pas pour moi. Que puis-je faire d'autre ?

Je préfère garder la trace de objectifs que de garder la trace de la les tâches elles-mêmes . Je conserve certaines informations sur les tâches, mais elles sont vraiment secondaires par rapport au suivi des objectifs. Les objectifs sont stockés dans un espace indépendant de Celery. Lorsqu'une requête doit effectuer un calcul qui dépend de l'atteinte d'un objectif, Celery vérifie si l'objectif a déjà été atteint. Si oui, il utilise l'objectif mis en cache, sinon il lance la tâche qui aura un effet sur l'objectif et envoie au client qui a fait la requête HTTP une réponse indiquant qu'il doit attendre un résultat.

---

Les noms de variables et les hyperliens ci-dessus sont pour Celery 4.x. En 3.x, les variables et les hyperliens correspondants sont les suivants : CELERY_TRACK_STARTED , CELERY_IGNORE_RESULT , CELERY_TASK_RESULT_EXPIRES .

Answer 3

Chaque Task a un .request qui le contient AsyncRequest objet. Ainsi, la ligne suivante donne l'état d'une Task task :

task.AsyncResult(task.request.id).state

Answer 4

Vous pouvez également créer des états personnalisés et mettre à jour leur valeur lors de l'exécution de la tâche. Cet exemple est tiré de la documentation :

@app.task(bind=True)
def upload_files(self, filenames):
    for i, file in enumerate(filenames):
        if not self.request.called_directly:
            self.update_state(state='PROGRESS',
                meta={'current': i, 'total': len(filenames)})

http://celery.readthedocs.org/en/latest/userguide/tasks.html#custom-states

Answer 5

Vieille question mais j'ai récemment rencontré ce problème.

Si vous essayez d'obtenir l'identifiant de la tâche, vous pouvez le faire comme suit :

import celery
from celery_app import add
from celery import uuid

task_id = uuid()
result = add.apply_async((2, 2), task_id=task_id)

Maintenant, vous savez exactement ce qu'est le task_id et vous pouvez l'utiliser pour obtenir le AsyncResult :

# grab the AsyncResult 
result = celery.result.AsyncResult(task_id)

# print the task id
print result.task_id
09dad9cf-c9fa-4aee-933f-ff54dae39bdf

# print the AsyncResult's status
print result.status
SUCCESS

# print the result returned 
print result.result
4