Monitoring Celery with Flower — Complete Guide
In this tutorial, you will learn about Monitoring Celery with Flower. We cover key concepts, practical examples, and best practices to help you master this topic.
Monitor Celery tasks and workers in real time using Flower, a web-based tool that shows task progress, worker status, queue depths, and task history.
What You Learn
You will learn how to install and run Flower, navigate the dashboard, monitor task execution and worker health, manage workers from the UI, and configure authentication and persistence.
Why It Matters
Celery workers operate as separate processes, often on different machines. Without Flower, debugging a slow task or a stuck worker requires checking logs on every machine. Flower provides a unified real-time view of the entire Celery cluster.
Real-World Use
DodaTech operations team keeps Flower open at all times. When a queue backs up, they see it immediately. When a task fails, they inspect the error in Flower without touching server logs. Flower is the first tool used during any Celery incident.
Installing Flower
# Install Flower
pip install flower
# Verify installation
flower --version
Expected output:
2.0.1
Starting Flower
# Basic usage (default port 5555)
celery -A tasks flower
# With custom port
celery -A tasks flower --port=5555
# With basic auth
celery -A tasks flower \
--basic-auth=user:password123
# With URL prefix
celery -A tasks flower --url-prefix=flower
# Enable persistence (saves task history)
celery -A tasks flower --persistent=True \
--db=/var/lib/flower/flower.db
Open http://localhost:5555 in your browser.
Flower Dashboard Overview
flowchart TB
FD[Flower Dashboard] --> M[Main View]
FD --> T[Tasks]
FD --> W[Workers]
FD --> Q[Queues]
FD --> B[Broker]
M --> Stats[Active, Reserved, Scheduled]
T --> TH[Task History]
W --> WS[Worker Status]
Q --> QD[Queue Depth]
B --> BS[Broker Status]
style FD fill:#f90,color:#fff
Programmatic Access with Flower API
import requests
import json
flower_url = "http://localhost:5555"
# Get worker status
workers = requests.get(f"{flower_url}/api/workers").json()
print("Workers:")
for worker_name, info in workers.items():
print(f" {worker_name}:")
print(f" Status: {info.get('status', 'unknown')}")
print(f" Active: {info.get('active', 0)}")
print(f" Processed: {info.get('task_count', 0)}")
print(f" Concurrency: {info.get('concurrency', 0)}")
# Get task list
tasks = requests.get(f"{flower_url}/api/tasks").json()
print(f"\nRecent tasks: {len(tasks)}")
for task_id, info in list(tasks.items())[:5]:
print(f" {task_id[:8]}: {info.get('name')} - {info.get('state')}")
Expected output:
Workers:
celery@worker1:
Status: online
Active: 2
Processed: 150
Concurrency: 8
Recent tasks: 25
550e8400: tasks.process_file - SUCCESS
6ba7b810: tasks.process_file - FAILURE
6ba7b811: tasks.send_email - SUCCESS
Monitoring Tasks with Flower API
import requests
from datetime import datetime
flower = "http://localhost:5555/api"
def get_task_details(task_id):
response = requests.get(f"{flower}/tasks/{task_id}")
return response.json()
def get_failed_tasks(limit=10):
tasks = requests.get(f"{flower}/tasks").json()
failed = []
for task_id, info in tasks.items():
if info.get('state') == 'FAILURE' and len(failed) < limit:
failed.append({
'id': task_id[:8],
'name': info.get('name'),
'args': info.get('args'),
'exception': info.get('exception'),
'traceback': info.get('traceback'),
'timestamp': info.get('timestamp'),
})
return failed
def get_recent_tasks(limit=10):
tasks = requests.get(f"{flower}/tasks").json()
recent = []
for task_id, info in sorted(
tasks.items(),
key=lambda x: x[1].get('timestamp', 0),
reverse=True
)[:limit]:
recent.append({
'id': task_id[:8],
'name': info.get('name'),
'state': info.get('state'),
'runtime': info.get('runtime'),
})
return recent
print("Recent tasks:")
for t in get_recent_tasks():
print(f" {t['id']} {t['name']:30s} {t['state']:10s} {t.get('runtime', 'N/A'):>6s}s")
print("\nFailed tasks:")
for t in get_failed_tasks():
print(f" {t['id']} {t['name']}: {t['exception']}")
Expected output:
Recent tasks:
550e8400 tasks.process_file SUCCESS 1.23s
6ba7b810 tasks.process_file FAILURE N/A
Failed tasks:
6ba7b810 tasks.process_file: ValueError('Invalid data')
Worker Control via Flower
import requests
flower = "http://localhost:5555/api"
# Shut down a worker
response = requests.post(
f"{flower}/worker/shutdown/celery@worker1"
)
print(f"Shutdown worker: {response.status_code}")
# Enable/disable worker
response = requests.post(
f"{flower}/worker/pool/grow/celery@worker1",
json={'n': 2}
)
print(f"Grow pool by 2: {response.status_code}")
response = requests.post(
f"{flower}/worker/pool/shrink/celery@worker1",
json={'n': 1}
)
print(f"Shrink pool by 1: {response.status_code}")
# Revoke a task
task_id = "550e8400-e29b-41d4-a716-446655440000"
response = requests.post(
f"{flower}/task/revoke/{task_id}",
json={'terminate': True}
)
print(f"Revoke task: {response.status_code}")
Flower Configuration File
# flowerconfig.py
import os
# Basic settings
port = 5555
address = '0.0.0.0'
url_prefix = ''
max_tasks = 10000
# Authentication
basic_auth = [
os.environ.get('FLOWER_USER', 'admin'),
os.environ.get('FLOWER_PASSWORD', 'flower_pass'),
]
# Persistence
persistent = True
db = '/var/lib/flower/flower.db'
state_save_interval = 1000
# CORS
cors = ['https://monitoring.dodatech.com']
# Auto-refresh
auto_refresh = True
# Start Flower with config
celery -A tasks flower --conf=flowerconfig.py
Celery Events Requirement
Flower requires Celery events to be enabled:
# Start worker with events
celery -A tasks worker --events --loglevel=info
# Or enable in config
# app.conf.task_events = True
# app.conf.task_send_events = True
Common Mistakes
1. Not Enabling Task Events
Flower needs task events to display task information. Start workers with --events or set task_send_events=True in Celery config.
2. Exposing Flower Without Auth
Flower provides full control over workers and tasks. Always enable basic authentication with a strong password in production.
3. Not Using Persistent Storage
Without --persistent=True, Flower loses all task history when restarted. Enable persistence to keep historical data for debugging.
4. Running Flower on the Same Port as Celery
Flower defaults to port 5555. Celery uses 5672 (RabbitMQ) or 6379 (Redis). Ensure ports do not conflict.
5. Monitoring Too Many Tasks
Flower stores task history in memory by default. For high-throughput systems, limit with --max-tasks=5000 and enable persistent storage.
Practice Questions
1. What port does Flower run on by default?
- Can be changed with
--portflag.
2. How do you enable authentication in Flower?
Use --basic-auth=user:password flag or the basic_auth configuration option.
3. What Celery configuration is required for Flower to work?
Workers must have --events enabled or task_send_events=True in configuration. Without events, Flower shows workers but no task details.
4. How do you persist Flower's task history?
Use --persistent=True --db=/path/to/flower.db. The database file stores task history across restarts.
Challenge
Deploy Flower in production behind a reverse proxy (Nginx) with HTTPS, basic authentication, and persistent storage. Create a monitoring dashboard that checks Flower API every 30 seconds and alerts if any worker is offline or any queue exceeds 1000 tasks.
FAQ
Mini Project: Flower Monitoring Setup
# monitoring.py
import requests
import time
import os
class FlowerMonitor:
def __init__(self, url='http://localhost:5555'):
self.url = url
self.api = f"{url}/api"
def check_workers(self):
workers = requests.get(f"{self.api}/workers").json()
status = {}
for name, info in workers.items():
status[name] = {
'online': info.get('status') == 'online',
'active': info.get('active', 0),
'processed': info.get('task_count', 0),
}
return status
def check_queues(self):
response = requests.get(f"{self.api}/queues/length")
return response.json()
def check_recent_failures(self, minutes=5):
tasks = requests.get(f"{self.api}/tasks").json()
failures = []
now = time.time()
for task_id, info in tasks.items():
if info.get('state') == 'FAILURE':
ts = info.get('timestamp', 0)
if now - ts < minutes * 60:
failures.append({
'id': task_id[:8],
'name': info.get('name'),
'error': info.get('exception'),
})
return failures
def health_report(self):
workers = self.check_workers()
queues = self.check_queues()
failures = self.check_recent_failures()
print(f"Flower Monitor Report - {time.ctime()}")
print("=" * 60)
print(f"\nWorkers ({len(workers)}):")
for name, info in workers.items():
icon = 'OK' if info['online'] else 'OFFLINE'
print(f" [{icon}] {name}: {info['active']} active, "
f"{info['processed']} processed")
print(f"\nQueues:")
for queue, length in queues.items():
status = 'OK' if length < 100 else 'WARN' if length < 500 else 'CRIT'
print(f" [{status}] {queue}: {length} tasks")
if failures:
print(f"\nRecent Failures ({len(failures)}):")
for f in failures[:5]:
print(f" {f['id']} {f['name']}: {f['error']}")
if __name__ == '__main__':
monitor = FlowerMonitor()
while True:
os.system('clear')
monitor.health_report()
time.sleep(10)
What's Next
Now that you understand Flower monitoring, explore monitoring and alerting for production Celery systems, then learn about error handling patterns for building robust task pipelines.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro