Celery Alerting: Automated Notifications for Worker Health, Task Failures, and Queue Backlogs

In this tutorial, you will learn about Celery Alerting: Automated Notifications for Worker Health, Task Failures, and Queue Backlogs. We cover key concepts, practical examples, and best practices to help you master this topic.

Celery alerting automates notifications for worker failures, task errors, queue backlogs, and latency SLO breaches, integrating with PagerDuty for critical alerts, Slack for warnings, and email for daily health digests.

flowchart LR
    CW[Celery Worker] -->|Events| Monitor[Alert Monitor]
    Monitor -->|Task Failed| Logger[Error Logger]
    Monitor -->|Queue Depth > N| Alert{Alert Type}
    Alert -->|Critical| PD[PagerDuty]
    Alert -->|Warning| Slack[Slack Channel]
    Alert -->|Info| Email[Daily Digest]
    Monitor -->|Latency > SLO| PD
    Monitor -->|Worker Down| PD

What You'll Learn

• Task failure alerting
• Queue backlog detection
• Latency SLO monitoring
• Worker down detection
• Multi-channel notification routing

Why It Matters

Without alerting, task failures and queue backlogs go unnoticed until users report problems. Proactive alerting catches issues minutes after they start, reducing MTTR from hours to minutes and preventing cascading failures.

Real-World Use

DodaTech's alerting system monitors 50+ Celery workers. When a worker's memory exceeds 1.5GB, a warning fires in Slack. If queue depth exceeds 5000 for 5 minutes, a critical alert pages the on-call engineer. The team resolves most issues before users notice.

Task Failure Alerts

from celery import Celery
from celery.signals import task_failure
import smtplib
import json

app = Celery('alerting', broker='redis://localhost:6379/0')

def send_slack_alert(message):
    print(f"[SLACK] {message}")

def send_pagerduty_alert(summary, details):
    print(f"[PAGERDUTY] {summary}")
    print(f"[PAGERDUTY] Details: {json.dumps(details, indent=2)}")

@task_failure.connect
def handle_task_failure(sender=None, task_id=None, exception=None,
                       traceback=None, **kwargs):
    task_name = sender.name if sender else 'unknown'
    alert = {
        'task_name': task_name,
        'task_id': task_id,
        'error': str(exception),
    }

    send_slack_alert(f"Task Failed: {task_name} - {exception}")
    send_pagerduty_alert(f"Task Failure: {task_name}", alert)

@app.task(bind=True, max_retries=1)
def critical_task(self, item_id):
    import random
    if random.random() < 0.3:
        raise ValueError(f"Processing failed for item {item_id}")
    return f"Item {item_id} processed"

for i in range(10):
    critical_task.delay(i)
print("Submitted 10 tasks with 30% failure rate")

Expected output:

Submitted 10 tasks with 30% failure rate
[SLACK] Task Failed: alerting.critical_task - Processing failed for item 3
[PAGERDUTY] Task Failure: alerting.critical_task
[PAGERDUTY] Details: {"task_name": "alerting.critical_task", "task_id": "id", "error": "Processing failed for item 3"}
...

Queue Backlog Detection

from celery import Celery
import redis
import time

app = Celery('alerting', broker='redis://localhost:6379/0')
cache = redis.Redis(host='localhost', port=6379, db=0)

QUEUE_DEPTH_WARNING = 100
QUEUE_DEPTH_CRITICAL = 500
CHECK_INTERVAL = 30

@app.task
def process_document(doc_id):
    time.sleep(0.2)
    return f"Processed {doc_id}"

def check_queue_depth():
    queue_name = 'celery'
    try:
        depth = cache.llen(queue_name)
        print(f"Queue depth: {depth}")

        if depth >= QUEUE_DEPTH_CRITICAL:
            send_slack_alert(f"[CRITICAL] Queue depth {depth} exceeds {QUEUE_DEPTH_CRITICAL}")
            send_pagerduty_alert(
                f"Queue {queue_name} depth critical: {depth}",
                {'queue': queue_name, 'depth': depth, 'threshold': QUEUE_DEPTH_CRITICAL}
            )
        elif depth >= QUEUE_DEPTH_WARNING:
            send_slack_alert(f"[WARNING] Queue depth {depth} exceeds {QUEUE_DEPTH_WARNING}")

        return depth
    except Exception as e:
        print(f"Failed to check queue: {e}")
        return None

check_queue_depth()
for i in range(50):
    process_document.delay(i)
check_queue_depth()

Expected output:

Queue depth: 0
Queue depth: 75

Latency SLO Monitoring

from celery import Celery
from celery.signals import task_postrun
import time

app = Celery('alerting', broker='redis://localhost:6379/0')

SLO_TARGETS = {
    'alerting.quick_task': {'p99': 0.5, 'p95': 0.2},
    'alerting.slow_task': {'p99': 5.0, 'p95': 3.0},
}

latency_history = {}

@task_postrun.connect
def check_latency_slo(sender=None, task_id=None, **kwargs):
    task_name = sender.name if sender else 'unknown'
    if task_name not in SLO_TARGETS:
        return

    if task_id not in latency_history:
        return

    duration = time.time() - latency_history.pop(task_id, 0)
    slo = SLO_TARGETS[task_name]

    if duration > slo['p99']:
        send_slack_alert(f"[SLO] {task_name} took {duration:.2f}s (p99 SLO: {slo['p99']}s)")

@app.task
def quick_task():
    return "Quick result"

@app.task
def slow_task():
    time.sleep(1.5)
    return "Slow result"

quick_task.delay()
slow_task.delay()
time.sleep(2)

Expected output:

[SLO] alerting.slow_task took 1.52s (p99 SLO: 5.0s)

Common Mistakes

• Alert fatigue from flapping tasks -- a flapping task (succeeds, fails, succeeds) triggers separate alerts each cycle. Implement hysteresis: alert only after N consecutive failures within M minutes.
• No deduplication of alerts -- if 100 tasks fail simultaneously, 100 alerts fire. Aggregate by error type: group all ValueError failures in 5-minute window into one alert with count.
• Ignoring alert routing by severity -- task failures go to Slack (warning) while worker down goes to PagerDuty (critical). Define severity tiers and route accordingly. Not all alerts require waking someone.
• Alerts without runbooks -- every alert should have a linked runbook or playbook. Include the alert in the notification with remediation steps. Without runbooks, alerts cause confusion during incidents.
• Not testing alert paths -- email servers fail, Slack API changes, PagerDuty integrations break. Test alert delivery monthly with scheduled test alerts that verify end-to-end notification delivery.

Practice Questions

1. How do you aggregate multiple task failures into a single alert?
2. What is the difference between warning and critical alert severity for queue depth?
3. How do you monitor p99 latency SLO for Celery tasks?
4. What information should a task failure alert include?
5. How do you prevent alert fatigue from flapping tasks?

Challenge

Build a multi-channel alerting system for Celery: (1) define alert severity levels (info=email, warning=Slack, critical=PagerDuty) with per-severity aggregation windows, (2) implement alert deduplication that groups identical errors from the same task type into 5-minute buckets with occurrence count, (3) create alert rules for: task failure rate >5% in 5min, queue depth >1000 sustained, worker count drops below minimum, latency p99 exceeds 2x baseline, (4) include runbook links in every alert notification, (5) implement a daily digest email with summary of all alerts from the past 24 hours, and (6) add monthly alert trend reporting.

FAQ

Mini Project

Build a complete Celery alerting pipeline: (1) instrument tasks with failure and latency signals that emit structured events, (2) build an alert aggregator that listens to events and deduplicates within 5-minute windows, (3) route alerts by severity: critical (PagerDuty via Events API v2), warning (Slack Webhook), info (email via SendGrid), (4) implement escalation: if critical alert not acknowledged in 15 minutes, escalate to secondary on-call, (5) create a weekly report showing alert breakdown by task type and time of day, (6) add an alert history dashboard in Grafana with filterable timeline.

What's Next

Continue with Performance Optimization to learn worker tuning for high throughput. Then explore Troubleshooting Guide for debugging common production issues.