Determine if celery is available and if it works

Question

I am using Celery with Django for an online game.

I wrote middleware to check if celery is available and working based on this answer: Determine if celery is available / work

My code looks something like this:

from celery.task.control import inspect

class CeleryCheckMiddleware(object):

    def process_request(self, request):
        insp = inspect().stats()
        if not insp:
           return render(...)
        else:
           return None

But I forgot the disclaimer in the comment at the bottom of this answer: "I found that the above adds two reply.celery.pidbox queues to rabbitmq every time it starts. This leads to a gradual increase in rabbitmq memory usage.

I now (only in a day!) Notice random 500 errors, starting from the line insp = inspect().stats()and ending OSError: [Errno 4] Interrupted system call.

Is there a safe way to check the availability and functioning of celery?

Answer 1

. , async . , , , ...

@app.job
def celery_alive():
    return "OK"

def process_request(self, request):
    res = celery_alive.apply_async()
    try:
        return "OK" == res.get(timeout=settings.ACCEPTABLE_TRANSACTION_TIME)
    except TimeoutError as e:
        return False