Celery healthchecks for django

celery_healthcheck.sh

celery inspect ping -d <worker_name> --timeout=<timeout_time>

celerybeat_healthcheck.py

from datetime import datetime, timedelta

import celery
import pytz
from celery.beat import Service
from django.core.management.base import BaseCommand


class Command(BaseCommand):
    help = 'Celery beat healthcheck'

    def handle(self, *args, **options):
        schedule = Service(celery.current_app).get_scheduler().get_schedule()
        now = datetime.now(tz=pytz.utc)

        for task_name, task in schedule.items():
            try:
                assert now < task.last_run_at + task.schedule.run_every
            except AttributeError:
                assert timedelta() < task.schedule.remaining_estimate(task.last_run_at)