Vérification de la prochaine exécution pour les tâches périodiques programmées dans Celery (avec Django)

Question

* Utilisation de célery 3.1.25 car django-célery-beat 1.0.1 rencontre un problème de planification des tâches périodiques.

Récemment, j'ai rencontré un problème avec celerybeat, les tâches périodiques avec un intervalle d'un jour ou plus semblant être «oubliées» par le planificateur. Si je change l'intervalle à every 5 seconds la tâche s'exécute normalement (toutes les 5 secondes) et l'attribut last_run_at est mis à jour. Cela signifie que celerybeat répond à l'ordonnanceur dans une certaine mesure, mais si je réinitialise le last_run_at, c'est-à-dire PeriodicTask.objects.update(last_run_at=None), aucune des tâches avec un intervalle de every day ne fonctionne plus. Celherbeat s'est écrasé à un moment donné et cela a peut-être corrompu quelque chose, j'ai donc créé une nouvelle base de données et virtualenv pour voir si le problème persiste. Je voudrais savoir s'il existe un moyen de récupérer le temps d'exécution suivant afin que je n'ai pas à attendre un jour pour savoir si ma tâche périodique a été exécutée ou non.

J'ai également essayé d'utiliser inspect <active/scheduled/reserved> mais tous ont renvoyé empty. Est-ce normal pour les tâches périodiques utilisant le planificateur de base de données de djcelery?

est ici la fonction qui planifie les tâches:

def schedule_data_collection(request, project): 
    if (request.method == 'POST'): 
     interval = request.POST.get('interval') 

     target_project = Project.objects.get(url_path=project)  
     interval_schedule = dict(every=json.loads(interval), period='days') 

     schedule, created = IntervalSchedule.objects.get_or_create(
      every=interval_schedule['every'], 
      period=interval_schedule['period'], 
     ) 

     task_name = '{} data collection'.format(target_project.name) 

     try: 
      task = PeriodicTask.objects.get(name=task_name) 
     except PeriodicTask.DoesNotExist: 
      task = PeriodicTask.objects.create(
       interval=schedule, 
       name=task_name, 
       task='myapp.tasks.collect_tool_data', 
       args=json.dumps([target_project.url_path]) 
      ) 
     else: 
      if task.interval != schedule: 
       task.interval = schedule 

      if task.enabled is False: 
       task.enabled = True 

      task.save() 

     return HttpResponse(task.interval) 
    else: 
     return HttpResponseForbidden() 

Answer 1

Vous pouvez voir votre planificateur en allant dans la coquille et en regardant app.conf.CELERYBEAT_SCEDULE.

celery -A myApp shell 
print(app.conf.CELERYBEAT_SCHEDULE) 

Ceci devrait vous montrer toutes vos tâches périodiques.