#sentry

#sentry

Sentry is a great tool for monitoring celery tasks, and alerting when they fail or don't run on time. But it requires a bit of work to setup properly. Below is some sample code for setting up sentry monitoring of periodic tasks, followed by an explanation. import math import sentry_sdk from celery import signals from sentry_sdk import monitor from sentry_sdk.integrations.celery import CeleryIntegration @signals.beat_init.connect # if you use beats @signals.celeryd_init.connect def init_sentry(**kwargs): sentry_sdk.init( dsn=..., integrations=[ CeleryIntegration(monitor_beat_tasks=False) ] ) @signals.worker_shutdown.connect @signals.task_postrun.connect def flush_sentry(**kwargs): sentry_sdk.flush(timeout=5) def add_periodic_task(celery, schedule, task): max_runtime = math.ceil(schedule * 4 / 60) monitor_config = { "recovery_threshold": 1, "failure_issue_threshold": 10, "checkin_margin": max_runtime, "max_runtime": max_runtime, "schedule": { "type": "interval", "value": math.ceil(schedule / 60.0) "unit": "minute" } } name = task.__name__ task = monitor(monitor_slug=name, monitor_config=monitor_config)(task) celery.add_periodic_task(schedule, celery.task(task).s(), name=name)