How do we completed stop tasks in celery periodic_task?

Question

Revoking a task on @periodic_task sends Discarding revoked tasks & Due task to workers.

[2018-09-17 12:23:50,864: INFO/MainProcess] Received task: cimexapp.tasks.add[xxxxxxx] [2018-09-17 12:23:50,864: INFO/MainProcess] Discarding revoked task: cimexapp.tasks.add[xxxxxxx] [2018-09-17 12:24:00,865: INFO/Beat] Scheduler: Sending due task cimexapp.tasks.add (cimexapp.tasks.add) [2018-09-17 12:24:00,869: INFO/MainProcess] Received task: cimexapp.tasks.add[xxxxxxx] [2018-09-17 12:24:00,869: INFO/MainProcess] Discarding revoked task: cimexapp.tasks.add[xxxxxxx] [2018-09-17 12:24:10,865: INFO/Beat] Scheduler: Sending due task cimexapp.tasks.add (cimexapp.tasks.add) [2018-09-17 12:24:10,868: INFO/MainProcess] Received task: cimexapp.tasks.add[xxxxxxx] [2018-09-17 12:24:10,869: INFO/MainProcess] Discarding revoked task: cimexapp.tasks.add[xxxxxxx]

tasks.py

@periodic_task(run_every=timedelta(seconds=10),options={"task_id":"xxxxxxx"})
def add():
     call(["ping","-c10","google.com"])


def stop():
    x = revoke("xxxxxxx",terminate=True,signal="KILL")
    print(x)
    print('DONE')

I had created task_id with a name so that it could be easy for me to kill it by calling id.

How do I completely stop it from sending tasks ? I don't want to kill all workers with

pkill -9 -f 'celery worker'
celery -A PROJECTNAME control shutdown

I just want to stop the tasks/workers for add() function.

a_k_v a_k_v · Accepted Answer · 2018-09-20T05:39:14

One possible way is to store the tasks in the database and add remove tasks dynamically. You can use database backed celery beat scheduler for the same. Refer https://django-celery-beat.readthedocs.io/en/latest/. PeriodicTask database store the periodic tasks. You can manipulate the periodic task by using database commands (Django ORM).

This is how I handled the dynamic tasks (Create and stop tasks dynamically).

from django_celery_beat.models import PeriodicTask, IntervalSchedule, CrontabSchedule

chon_schedule = CrontabSchedule.objects.create(minute='40', hour='08', day_of_week='*', day_of_month='*', month_of_year='*') # To create a cron schedule. 
schedule = IntervalSchedule.objects.create(every=10, period=IntervalSchedule.SECONDS) # To create a schedule to run everu 10 min.
PeriodicTask.objects.create(crontab=chon_schedule, name='name_to_identify_task',task='name_of_task') # It creates a entry in the database describing that periodic task (With cron schedule).
task = PeriodicTask.objects.create(interval=schedule, name='run for every 10 min', task='for_each_ten_min', ) # It creates a periodic task with interval schedule

Whenever you update a PeriodicTask a counter in this table is also incremented, which tells the celery beat service to reload the schedule from the database.

So you don't need to restart the or kill the beat. If you want to stop a task when particular criteria met then

periodic_task = PeriodicTask.objects.get(name='run for every 10 min')
periodic_task.enabled = False
periodic_task.save()

When enabled is False then the periodic task becomes idle. You can again make it active by making enable = True.

If you no longer needs the task then you can simply delete the entry.

When creating your Project model object create periodic task too. Just create cron schedule or interval schedule based on your scenario. Then Create PeriodicTask object you can give Project.name to PeriodicTask name(so you can easily relate the project object with PeriodicTask object. Thats it, From that moment onwards the task will handle by celery beat.

If you want to disable or enable Periodic task dynamically just set enabled flag in PeriodicTask as follows

    task = PeriodicTask.objects.get(name='task_name')
    task.enabled = False
    task.save()

How do we completed stop tasks in celery periodic_task?

2 Answers