Celery beat and workers healthchecks.

Helthcheck for celery worker:

This command only works when remote control is enabled.

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

When a celery worker uses a solo pool, healthcheck waits for the task to finish. In this case, you must increase the timeout waiting for a response.

Helthcheck for celery beat:

I could not find a good solution in the documentation, so I came up with my own.
It works for default celery beat scheduler.

from datetime import datetime, timedelta
import pytz
import shelve

now = datetime.now(tz=pytz.utc)
file_data = shelve.open('celerybeat-schedule') # Name of the file used by PersistentScheduler to store the last run times of periodic tasks. 

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