How to retrieve all pending and active tasks with celery

Viewed 1223

I am trying to retrieve all currently pending and active tasks from Celery.

active, scheduled and reserved are simple to obtain:

i = celery_app.control.inspect()
tasks=list()
for k in i.active():
    tasks += i.active()[k]
for k in i.reserved():
    tasks += i.reserved()[k]
for k in i.scheduled():
    tasks += i.scheduled()[k]

However, this does not include tasks that have not been scheduled or reserved yet. The following code retrieves these tasks, but obviously only works with Redis, assumes a JSON serialization, and it's not ideal for a variety of reasons.

r = redis.StrictRedis(
        host='localhost',
        port='6379',
    )
tasks = [json.loads(pending_task)['headers'] for pending_task in r.lrange('celery', 0, -1)]

Is there a way to retrieve these pending tasks in a form similar to inspect.active() with easy access to args, kwargs, id (preferably through Celery, so I do not need to specify the Celery backend and serialization, etc)?

EDIT (hedging my bet)

From what I can tell this is not simple, and there is no i.pending() in celery. So, I have a few sub questions which will allow me to write this myself.

  1. How to get a list of active celery queues? (the above assumes default queue only) Perhaps something like queues = [q['exchange']['name'] for host in i.active_queues().values() for q in host])
  2. How to generically get messages in those queues? This would be a generic version of r.lrange('celery', 0, -1) above.
  3. How to generically decode/de-serialize messages? This would be a generic version of json.loads(...) above celery_app.backend.decode

With that info it should be straightforward to pull the necessary information from the message headers.

0 Answers
Related