Celery Events: Monitoring Task Lifecycle with the Celery Event System
In this tutorial, you will learn about Celery Events: Monitoring Task Lifecycle with the Celery Event System. We cover key concepts, practical examples, and best practices to help you master this topic.
The Celery event system emits lifecycle events for tasks and workers — task-sent, task-started, task-succeeded, task-failed, worker-online, worker-heartbeat — enabling real-time monitoring, custom dashboards, and automated responses to task state changes.
sequenceDiagram
participant Client
participant Broker
participant Worker
participant Events
Client->>Broker: Send Task
Broker->>Events: task-sent
Worker->>Broker: Receive Task
Worker->>Events: task-received
Worker->>Events: task-started
Worker->>Worker: Execute
Worker->>Events: task-succeeded / task-failed
Worker->>Events: worker-heartbeat (every 5s)
What You'll Learn
- Celery event types: task lifecycle, worker lifecycle, heartbeat
- Capturing and processing events with Python
- Building custom event monitoring dashboards
- Event-based alerting and auto-scaling
Why It Matters
Without events, monitoring Celery means polling the result backend for task states — which is slow, resource-intensive, and misses worker-level events. The event system provides real-time push-based observability with sub-second latency.
Real-World Use
DodaTech's Celery event monitor listens for task-failed events and automatically creates incident tickets. Worker-offline events trigger auto-scaling to launch replacement workers. Task latency events (time between task-sent and task-started) feed a Grafana dashboard that tracks queue health.
Capturing Celery Events
Listen for and Process Celery events:
from celery import Celery
from celery.events import EventReceiver
import json
app = Celery('events_demo', broker='redis://localhost:6379/0')
@app.task
def sample_task(duration=1):
import time
time.sleep(duration)
return f"Completed in {duration}s"
class EventMonitor:
def __init__(self, app):
self.app = app
self.events = []
def process_event(self, event):
"""Process a single Celery event."""
event_type = event.get("type")
timestamp = event.get("timestamp")
uuid = event.get("uuid", "unknown")
if event_type == "task-sent":
print(f"[SENT] Task {uuid} sent to queue")
elif event_type == "task-received":
print(f"[RECEIVED] Task {uuid} received by worker")
elif event_type == "task-started":
print(f"[STARTED] Task {uuid} started execution")
elif event_type == "task-succeeded":
result = event.get("result", "")
print(f"[SUCCESS] Task {uuid} completed: {str(result)[:50]}")
elif event_type == "task-failed":
exception = event.get("exception", "")
print(f"[FAILED] Task {uuid} failed: {exception}")
elif event_type == "worker-heartbeat":
hostname = event.get("hostname", "unknown")
active = event.get("active", 0)
processed = event.get("processed", 0)
print(f"[HEART] Worker {hostname}: {active} active, {processed} total")
self.events.append(event)
def capture(self, duration=10):
"""Capture events for a specified duration."""
import time
connection = self.app.connection()
recv = EventReceiver(connection, handlers={
"*": self.process_event,
})
recv.capture(limit=None, timeout=duration, wakeup_after=1)
def summary(self):
"""Generate event summary."""
counts = {}
for event in self.events:
etype = event.get("type", "unknown")
counts[etype] = counts.get(etype, 0) + 1
return {"total": len(self.events), "by_type": counts}
# Start event capture in background
import threading
import time
monitor = EventMonitor(app)
def capture_events():
try:
monitor.capture(duration=5)
except Exception as e:
print(f"Capture ended: {e}")
t = threading.Thread(target=capture_events, daemon=True)
t.start()
time.sleep(0.5)
result = sample_task.delay(0.2)
print(f"Task sent: {result.id}")
time.sleep(3)
summary = monitor.summary()
print(f"\nEvent Summary: {json.dumps(summary, indent=2)}")
Expected output:
[SENT] Task 550e8400-... sent to queue
[RECEIVED] Task 550e8400-... received by worker
[STARTED] Task 550e8400-... started execution
[SUCCESS] Task 550e8400-... completed: Completed in 0.2s
[HEART] Worker celery@hostname: 1 active, 150 total
Event Summary: {
"total": 5,
"by_type": {
"task-sent": 1,
"task-received": 1,
"task-started": 1,
"task-succeeded": 1,
"worker-heartbeat": 1
}
}
Custom Event Monitoring
Build a custom monitor with statistics:
from celery import Celery
from collections import defaultdict
import time
import json
app = Celery('events_demo', broker='redis://localhost:6379/0')
class TaskStatisticsMonitor:
def __init__(self):
self.task_times = defaultdict(list)
self.task_counts = defaultdict(int)
self.task_errors = defaultdict(int)
self.current_tasks = {}
self.workers = {}
def on_task_sent(self, event):
uuid = event["uuid"]
self.current_tasks[uuid] = {"sent_at": event["timestamp"]}
def on_task_received(self, event):
uuid = event["uuid"]
if uuid in self.current_tasks:
self.current_tasks[uuid]["received_at"] = event["timestamp"]
def on_task_started(self, event):
uuid = event["uuid"]
if uuid in self.current_tasks:
self.current_tasks[uuid]["started_at"] = event["timestamp"]
def on_task_succeeded(self, event):
uuid = event["uuid"]
name = event.get("name", "unknown")
self.task_counts[name] += 1
if uuid in self.current_tasks:
info = self.current_tasks[uuid]
if "started_at" in info and "sent_at" in info:
queue_time = info["started_at"] - info["sent_at"]
run_time = event["timestamp"] - info["started_at"]
self.task_times[name].append({
"queue_time": queue_time,
"run_time": run_time,
})
del self.current_tasks[uuid]
def on_task_failed(self, event):
uuid = event["uuid"]
name = event.get("name", "unknown")
self.task_counts[name] += 1
self.task_errors[name] += 1
self.current_tasks.pop(uuid, None)
def on_worker_heartbeat(self, event):
hostname = event["hostname"]
self.workers[hostname] = {
"last_seen": time.time(),
"active": event.get("active", 0),
"processed": event.get("processed", 0),
}
def get_report(self):
return {
"task_counts": dict(self.task_counts),
"task_errors": dict(self.task_errors),
"workers": self.workers,
"active_tasks": len(self.current_tasks),
}
monitor = TaskStatisticsMonitor()
import threading
def capture(monitor, app):
from celery.events import EventReceiver
connection = app.connection()
recv = EventReceiver(connection, handlers={
"task-sent": monitor.on_task_sent,
"task-received": monitor.on_task_received,
"task-started": monitor.on_task_started,
"task-succeeded": monitor.on_task_succeeded,
"task-failed": monitor.on_task_failed,
"worker-heartbeat": monitor.on_worker_heartbeat,
})
recv.capture(limit=None, timeout=5, wakeup_after=1)
@app.task
def fast_task():
return "fast"
@app.task
def slow_task():
time.sleep(0.3)
return "slow"
@app.task
def failing_task():
raise ValueError("Intentional failure")
t = threading.Thread(target=capture, args=(monitor, app), daemon=True)
t.start()
time.sleep(0.5)
fast_task.delay()
slow_task.delay()
failing_task.delay()
time.sleep(4)
report = monitor.get_report()
print(f"Tasks: {report['task_counts']}")
print(f"Errors: {report['task_errors']}")
print(f"Active workers: {len(report['workers'])}")
Expected output:
Tasks: {'events_demo.fast_task': 1, 'events_demo.slow_task': 1, 'events_demo.failing_task': 1}
Errors: {'events_demo.failing_task': 1}
Active workers: 1
Common Mistakes
- Not consuming events fast enough — the event queue has a limited buffer. If your event consumer is slow, events are dropped. Use a dedicated event processing worker with its own concurrency.
- Using events for persistent monitoring — events are fire-and-forget. If your monitor is down, events are lost. For persistent monitoring, also poll the result backend periodically.
- Subscribing to all events in production — high-traffic Celery clusters emit thousands of events per second. Only subscribe to events you need. Filter by type in the handler registration.
- Ignoring worker-heartbeat events — heartbeats detect worker failures. If a heartbeat stops, the worker may have crashed. Monitor heartbeat frequency and alert on missing heartbeats.
- Not deduplicating events — the same event may be delivered multiple times if the event receiver reconnects. Use event UUIDs for idempotent processing.
Practice Questions
- What lifecycle events does Celery emit for tasks?
- How does the event system differ from polling the result backend?
- What information is included in a task-started event?
- How do worker-heartbeat events help detect worker failures?
- What precautions are needed when consuming events at scale?
Challenge
Build a real-time Celery event dashboard that: (1) captures all task lifecycle events, (2) displays active task count per queue, (3) shows task latency (queue time + execution time) per task type, (4) alerts when error rate exceeds 5% in a 5-minute window, (5) tracks worker heartbeats and alerts when a worker misses 3 consecutive beats, (6) displays throughput (tasks/sec) per task type, and (7) exports all metrics to Prometheus.
FAQ
Mini Project
Build a Celery event-based auto-scaler that: (1) captures task-received and task-completed events, (2) calculates queue depth (total received - completed - failed), (3) when queue depth exceeds 100 for 30 seconds, launches a new Celery worker via Docker/Kubernetes API, (4) when queue depth is below 10 for 5 minutes, stops an idle worker, and (5) reports scaling actions to a monitoring channel.
What's Next
Continue with Celery Signals to learn about Celery's signal system for extending task behavior. Then explore Celery Logging for configuring task and worker logging.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro