celery scheduling task periodically but not executing them

Question

My django app have two task out of which one is periodic task.

normal task: AddScore periodic task: CalculateTopScore

class CalculateTopScore(celery.Task):

    default_retry_delay = settings.DEFAULT_RETRY_DELAY
    max_retries = settings.DEFAULT_MAX_RETRIES
    name = 'games.tasks.CalculateTopScore'

    def run(self):
        try:
            # Code to run

        except Exception, err:
            logger.exception("Error in running task calculate_top_score")
            self.retry(exc=err)

        return True

    def on_failure(self, exc, task_id):

        failure = "%s task for calculate_top_score failed permanently." % task_id
        logger.error(failure)

    def on_success(self):

        task_info = 'calculate_top_score task successfully'
        logger.info(task_info)

I want to execute this task periodically every 30 minute.

Here are the settings i'm using:

#celery/settings.py

import djcelery
import kombu

from celery.schedules import crontab
from config.celery import exchanges


djcelery.setup_loader()

CELERYBEAT_SCHEDULER = "djcelery.schedulers.DatabaseScheduler"

CELERYBEAT_SCHEDULE = {
    "calcluate_score": {
        "task": "games.tasks.CalculateTopScore",
        "schedule": crontab(minute='*/30'),
        "args": (),
    },
}

CELERY_QUEUES = (
    kombu.Queue('add_score',
                exchange=exchanges.add_score_exchange,
                routing_key='add.scores'),
    kombu.Queue('calcluate_score',
                 exchange=exchanges.calcluate_score_exchange,
                 routing_key='calculate.scores'),
)

CELERY_ROUTES = ('config.celery.routers.CeleryTaskRouter',)

# Default delay(in seconds) for retrying tasks.
DEFAULT_RETRY_DELAY = 60

# Maximum retry count
DEFAULT_MAX_RETRIES = 6

CELERY_IGNORE_RESULT = True

exchanges.py file

#exchanges.py

from kombu import Exchange

add_score_exchange = Exchange('add_score', type='direct')
calcluate_score_exchange = Exchange('calcluate_score', type='direct')

routes.py file

ROUTES = {
    'players.tasks.AddScore': {
        'exchange': 'add_score',
        'exchange_type': 'direct',
        'routing_key': 'add.score',
    },
   'games.tasks.CalculateTopScore': {
        'exchange': 'calculate_score',
        'exchange_type': 'direct',
        'routing_key': 'calculate.score',
    },
}


class CeleryTaskRouter(object):
    """ This is a basic celery task router.
    """

    def route_for_task(self, task, arg=None, kwargs=None):
        return ROUTES.get(task)

On our production server i run celeryd with following arguments: celeryd worker -B

Now in logs i observe that celerybeat schedules the task every 30 minutes but worker doesn't know at all about the task scheduled and hence its not executed.

Why? any config/settings missing? How to execute a class based task periodically?

Please help

Answer 1

You probably have to specify the queues you want celery to get tasks from. Default is to use the celery queue.

Try:

celeryd worker -B -Q add_score,calculate_score