How to monitor events from workers in a Celery-Django application?

Question

According to the celery tutorial regarding real-time monitoring of celery workers, one can also programmatically capture the events produced by the workers and take action accordingly.

My question is how can I integrate a monitor as the one in this example, in a Celery-Django application?

EDIT: The code example in the tutorial looks like:

from celery import Celery

def my_monitor(app):
    state = app.events.State()

    def announce_failed_tasks(event):
        state.event(event)
        task_id = event['uuid']

        print('TASK FAILED: %s[%s] %s' % (
            event['name'], task_id, state[task_id].info(), ))
    with app.connection() as connection:
        recv = app.events.Receiver(connection, handlers={
                'task-failed': announce_failed_tasks,
                'worker-heartbeat': announce_dead_workers,
        })
        recv.capture(limit=None, timeout=None, wakeup=True)

if __name__ == '__main__':
    celery = Celery(broker='amqp://guest@localhost//')
    my_monitor(celery)

So I want to capture task_failed event sent by the worker, and to get its task_id like the tutorial shows, to get the result for this task from the result-backend that was configured for my application and process it further. My problem is that it is not obvious to me how to get the application, as in a django-celery project it is not transparent to me the instantiation of Celery library.

I am also open to any other idea as to how to process the results when a worker has finished executing a task.

I think you'll have to be a bit more specific, what events need capturing? Do you have any example code? — danodonovan, Mar 01 '13 at 14:06

Clara · Accepted Answer · 2013-03-19T11:54:57.477

Ok, I found a way of doing this, though I am not sure that this is the solution, but it works for me. The monitor function basically connects directly to the broker and listens to different types of events. My code looks like this:

from celery.events import EventReceiver
from kombu import Connection as BrokerConnection

def my_monitor:
    connection = BrokerConnection('amqp://guest:guest@localhost:5672//')

    def on_event(event):
        print "EVENT HAPPENED: ", event

    def on_task_failed(event):
        exception = event['exception']
        print "TASK FAILED!", event, " EXCEPTION: ", exception

    while True:
        try:
            with connection as conn:
                recv = EventReceiver(conn,
                                 handlers={'task-failed' : on_task_failed,
                                           'task-succeeded' : on_event,
                                           'task-sent' : on_event,
                                           'task-received' : on_event,
                                           'task-revoked' : on_event,
                                           'task-started' : on_event,
                                           # OR: '*' : on_event
                                           })
            recv.capture(limit=None, timeout=None)
    except (KeyboardInterrupt, SystemExit):
        print "EXCEPTION KEYBOARD INTERRUPT"
        sys.exit()

This is all. And I run this in a different process than the normal application, meaning that I create a child process of my celery application which only runs this function. HTH

Hi, thanks, your question is basically what I am trying to do now. Where are you putting this code in your Django project? Could you explain about creating the child process of your celery app? At the moment my celery app gets configued in `myproj/myproj/celery.py`(as per http://docs.celeryproject.org/en/latest/django/first-steps-with-django.html#using-celery-with-django) — fpghost, Jan 28 '16 at 07:54
Hi! I haven't worked on this in a long time, so things might have changed in Celery itself within the last n releases. Basically I was starting a Python daemon process like: daemon_process = Process(target=results_processing.my_monitor) daemon_process.daemon = True daemon_process.start() in one of the modules that is called when the app starts — Clara, Jan 28 '16 at 16:15
I was using Django and initiated this monitor via that. In Django < 1.9 I used to be able to start monitoring in the `proj/proj/celery.py` file, just by `my_monitor(app)` after defining the celery app. Now in Django 1.9 that results in a `AppRegistryNotReady` exc (I think importing models in `__init__.py` of apps is now not allowed --- I should note that my monitor relies on some models). I ended up starting the monitor in the `AppConfig.ready()` method of the django app whose models my monitor relied on (this ensures that the app has finished registration). HTH — fpghost, Jun 07 '16 at 15:08

score 8 · Answer 2 · answered Sep 08 '15 at 04:24

Beware of a couple of gotchas

You need to set CELERY_SEND_EVENTS flag as true in your celery config.
You can also set the event monitor in a new thread from your worker.

Here is my implementation:

class MonitorThread(object):
    def __init__(self, celery_app, interval=1):
        self.celery_app = celery_app
        self.interval = interval

        self.state = self.celery_app.events.State()

        self.thread = threading.Thread(target=self.run, args=())
        self.thread.daemon = True
        self.thread.start()

    def catchall(self, event):
        if event['type'] != 'worker-heartbeat':
            self.state.event(event)

        # logic here

    def run(self):
        while True:
            try:
                with self.celery_app.connection() as connection:
                    recv = self.celery_app.events.Receiver(connection, handlers={
                        '*': self.catchall
                    })
                    recv.capture(limit=None, timeout=None, wakeup=True)

            except (KeyboardInterrupt, SystemExit):
                raise

            except Exception:
                # unable to capture
                pass

            time.sleep(self.interval)

if __name__ == '__main__':
    app = get_celery_app() # returns app
    MonitorThread(app)
    app.start()

How to monitor events from workers in a Celery-Django application?

2 Answers2

Linked