Celery Monitoring and Alerting — Complete Guide
In this tutorial, you will learn about Celery Monitoring and Alerting. We cover key concepts, practical examples, and best practices to help you master this topic.
Monitor Celery production systems with metrics, alerts, and dashboards for queue depth, worker health, task latency, and error rates to ensure reliability.
What You Learn
You will learn how to collect Celery metrics, set up Prometheus monitoring, create Grafana dashboards, configure alerts for queue depth and task failures, and monitor worker health.
Why It Matters
Celery in production without monitoring is blind. A queue backing up silently causes hours of delays. A dead worker goes unnoticed until users complain. Monitoring gives you visibility into every aspect of the Celery system.
Real-World Use
DodaTech monitors Celery with Prometheus and Grafana. Alerts fire when any queue exceeds 1000 tasks, any task takes longer than 5 minutes, or any worker goes offline for more than 60 seconds.
Key Metrics to Monitor
| Metric | What It Tells You | Alert Threshold |
|---|---|---|
| Queue depth | Task backlog | > 1000 tasks |
| Active tasks | Current load | > 80% of concurrency |
| Task latency | Processing speed | > 5 minutes |
| Task failure rate | Error rate | > 1% |
| Worker count | Available capacity | < minimum workers |
| Broker memory | Broker health | > 80% of limit |
| Result backend size | Storage usage | > 1GB |
Prometheus Metrics with Celery
# Install: pip install celery-prometheus-exporter
from celery import Celery
from celery_prometheus import register_prometheus_exporter
app = Celery('monitored', broker='redis://localhost:6379/0')
register_prometheus_exporter(app, port=8080)
@app.task
def monitored_task(data):
return f"Processed: {data}"
# Start worker with Prometheus endpoint
celery -A monitored worker --loglevel=info
# Metrics available at http://localhost:8080/metrics
curl http://localhost:8080/metrics | head -20
Expected output:
# HELP celery_task_states Number of tasks in each state
# TYPE celery_task_states gauge
celery_task_states{state="active"} 2.0
celery_task_states{state="pending"} 15.0
celery_task_states{state="failed"} 1.0
# HELP celery_task_runtime_seconds Task runtime in seconds
# TYPE celery_task_runtime_seconds histogram
celery_task_runtime_seconds_bucket{le="0.1"} 10.0
Custom Metrics in Tasks
from celery import Celery
import time
import json
app = Celery('custom_metrics', broker='redis://localhost:6379/0')
# Simple metrics using Redis
import redis
metrics_client = redis.Redis.from_url('redis://localhost:6379/1')
def record_metric(name, value, tags=None):
key = f"celery_metric:{name}"
data = {'value': value, 'timestamp': time.time(), 'tags': tags or {}}
metrics_client.lpush(key, json.dumps(data))
metrics_client.ltrim(key, 0, 9999) # Keep last 10000
@app.task(bind=True)
def monitored_task(self, data):
start = time.time()
record_metric('task.started', 1, {'task': self.name})
try:
result = data.upper()
duration = time.time() - start
record_metric('task.completed', 1, {'task': self.name})
record_metric('task.duration', duration, {'task': self.name})
return result
except Exception as e:
record_metric('task.failed', 1, {'task': self.name, 'error': str(e)})
raise
Queue Depth Monitoring
import redis
import time
class QueueMonitor:
def __init__(self, broker_url='redis://localhost:6379/0'):
self.client = redis.Redis.from_url(broker_url)
def get_queue_depth(self, queue_name='celery'):
return self.client.llen(queue_name)
def get_all_queues(self, queue_names):
depths = {}
for name in queue_names:
depths[name] = self.get_queue_depth(name)
return depths
def check_alerts(self, thresholds):
alerts = []
for name, depth in self.get_all_queues(thresholds.keys()).items():
if depth > thresholds[name]:
alerts.append({
'queue': name,
'depth': depth,
'threshold': thresholds[name],
'severity': 'critical' if depth > thresholds[name] * 2 else 'warning'
})
return alerts
monitor = QueueMonitor()
thresholds = {'celery': 100, 'high': 50, 'batch': 1000}
while True:
depths = monitor.get_all_queues(thresholds.keys())
alerts = monitor.check_alerts(thresholds)
print(f"\nQueue Status at {time.strftime('%H:%M:%S')}:")
for name, depth in depths.items():
status = 'OK' if depth <= thresholds[name] else 'ALERT'
print(f" {name:15s} {depth:6d} {thresholds[name]:6d} [{status}]")
if alerts:
print("\nAlerts:")
for a in alerts:
print(f" [{a['severity'].upper()}] {a['queue']}: {a['depth']} > {a['threshold']}")
time.sleep(10)
Worker Health Check
from celery import Celery
import subprocess
import json
app = Celery('health', broker='redis://localhost:6379/0')
def check_worker_health():
"""Check all workers via Celery inspect."""
from celery.task.control import inspect
i = inspect()
# Ping all workers
ping_result = i.ping()
if not ping_result:
return {'status': 'critical', 'message': 'No workers responding'}
# Get active tasks
active = i.active() or {}
reserved = i.reserved() or {}
stats = i.stats() or {}
workers = []
for worker_name in ping_result:
worker_stats = stats.get(worker_name, {})
workers.append({
'name': worker_name,
'alive': True,
'active_tasks': len(active.get(worker_name, [])),
'reserved_tasks': len(reserved.get(worker_name, [])),
'processed_total': worker_stats.get('total', {}).get('tasks.process', 0),
'pool_size': worker_stats.get('pool', {}).get('max-concurrency', 0),
})
return {
'status': 'healthy',
'workers': workers,
'total_workers': len(workers),
}
print(json.dumps(check_worker_health(), indent=2))
Expected output:
{
"status": "healthy",
"workers": [
{
"name": "celery@worker1",
"alive": true,
"active_tasks": 2,
"reserved_tasks": 5,
"processed_total": 1500,
"pool_size": 8
}
],
"total_workers": 1
}
Celery Event Logging
from celery import Celery
import logging
app = Celery('logging_demo', broker='redis://localhost:6379/0')
# Set up Celery logging
app.conf.update(
worker_log_format='[%(asctime)s %(levelname)s %(processName)s] %(message)s',
worker_task_log_format='[%(asctime)s %(levelname)s %(processName)s] %(task_name)s: %(message)s',
worker_log_color=False,
)
# Log all task events
from celery.signals import (
task_prerun, task_postrun, task_failure, task_success,
worker_ready, worker_shutdown
)
@task_prerun.connect
def task_prerun_handler(task_id=None, task=None, args=None, kwargs=None, **kw):
logging.info(f"Task started: {task.name}[{task_id[:8]}]")
@task_postrun.connect
def task_postrun_handler(task_id=None, task=None, state=None, **kw):
logging.info(f"Task finished: {task.name}[{task_id[:8]}] state={state}")
@task_failure.connect
def task_failure_handler(task_id=None, exception=None, traceback=None, **kw):
logging.error(f"Task failed: {task_id[:8]}: {exception}")
@worker_ready.connect
def worker_ready_handler(**kw):
logging.info("Worker ready")
@app.task
def logged_task(data):
return f"Logged: {data}"
Common Mistakes
1. Only Monitoring Queue Depth
Queue depth alone is insufficient. A queue with 0 tasks could mean no work or a dead producer. Monitor task rates, worker count, and latency together.
2. Setting Alert Thresholds Too Low
A queue depth of 100 may be normal during a batch job. Set thresholds based on historical data. Start high and tune down.
3. Not Alerting on Zero Workers
Zero active workers is the most critical alert. No workers means no tasks are processed. If all workers are down, every second costs.
4. Ignoring Broker Memory
The broker (Redis/RabbitMQ) stores all queued tasks. If broker memory fills up, tasks are rejected. Monitor broker memory usage and set maxmemory policies.
5. Not Monitoring Task Duration
A task that normally takes 1 second taking 60 seconds indicates a problem. Monitor p95 and p99 task latency to catch performance degradation early.
Practice Questions
1. What are the most important Celery metrics to monitor?
Queue depth, active tasks, worker count, task failure rate, task latency (p95/p99), and broker memory usage.
2. How do you expose Celery metrics to Prometheus?
Use the celery-prometheus-exporter package. Register the exporter with your Celery app and it exposes metrics on an HTTP endpoint.
3. What is a reasonable queue depth alert threshold?
Start at 1000 tasks. Tune based on your processing rate. A queue of 1000 that clears in 30 seconds is fine. A queue of 1000 that never shrinks is a problem.
4. How do you detect a dead worker?
Ping workers with inspect().ping(). If a worker does not respond in 5 seconds, mark it as dead. Alert if less than the minimum number of workers respond.
Challenge
Design a complete Celery monitoring system: collect metrics (queue depth, task latency, failure rate, worker count), store them in a time-series database, create a Grafana dashboard with 5 panels, and configure alerts for: queue depth > 1000, task latency > 5 min p99, failure rate > 5%, and worker count < 3.
FAQ
Mini Project: Celery Monitoring Dashboard
# celery_monitor.py
from celery import Celery
import redis
import time
import json
import os
app = Celery('monitoring_demo', broker='redis://localhost:6379/0')
class CeleryMonitor:
def __init__(self):
self.broker = redis.Redis.from_url('redis://localhost:6379/0')
self.tasks = {}
def get_metrics(self):
metrics = {
'timestamp': time.time(),
'queues': self._get_queue_depths(),
'workers': self._get_worker_status(),
'task_stats': self._get_task_stats(),
}
return metrics
def _get_queue_depths(self):
depths = {}
for key in self.broker.scan_iter('celery*'):
try:
depths[key.decode()] = self.broker.llen(key)
except:
pass
return depths
def _get_worker_status(self):
try:
i = app.control.inspect()
ping = i.ping() or {}
active = i.active() or {}
stats = i.stats() or {}
workers = {}
for name in ping:
s = stats.get(name, {})
workers[name] = {
'alive': True,
'active': len(active.get(name, [])),
'pool_size': s.get('pool', {}).get('max-concurrency', 0),
'total_processed': s.get('total', {}),
}
return workers
except Exception as e:
return {'error': str(e)}
def _get_task_stats(self):
try:
i = app.control.inspect()
stats = i.stats() or {}
totals = {}
for worker, s in stats.items():
for task, count in s.get('total', {}).items():
totals[task] = totals.get(task, 0) + count
return totals
except Exception as e:
return {'error': str(e)}
def print_report(self):
metrics = self.get_metrics()
os.system('clear')
print(f"Celery Monitor - {time.ctime()}")
print("=" * 60)
print(f"\nQueues:")
for name, depth in sorted(metrics['queues'].items()):
print(f" {name:20s}: {depth} tasks")
print(f"\nWorkers:")
for name, info in metrics['workers'].items():
if name == 'error':
print(f" Error: {info}")
else:
print(f" {name:30s}: {info['active']} active / {info['pool_size']} pool")
print(f"\nTask Totals:")
for task, count in sorted(metrics['task_stats'].items())[:10]:
print(f" {task:40s}: {count}")
if __name__ == '__main__':
monitor = CeleryMonitor()
try:
while True:
monitor.print_report()
time.sleep(5)
except KeyboardInterrupt:
print("\nMonitoring stopped")
What's Next
Now that you understand monitoring and alerting, explore error handling patterns for building robust task pipelines, then learn about Celery with Django for web application integration.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro