Job Dashboard with Bull Board and Flower
In this tutorial, you will learn about Job Dashboard with Bull Board and Flower. We cover key concepts, practical examples, and best practices to help you master this topic.
Set up job dashboards using Bull Board for Node.js and Flower for Celery to monitor queues, workers, job details, and retry failed jobs from a web UI.
What You Learn
You will learn how to install and configure Bull Board and Flower, customize dashboards, monitor queues and workers, and perform operations like retrying and removing jobs.
Why It Matters
A job dashboard provides real-time visibility into queue health, worker status, and job details. It enables operators to retry failed jobs, inspect stalled jobs, and understand system behavior at a glance.
Real-World Use
DodaTech uses Bull Board for Node.js queues and Flower for Celery queues. Operators monitor queue depth, worker count, and job failures. Failed jobs are retried with one click from the dashboard.
Dashboard Architecture
flowchart TD
Q[Redis / RabbitMQ] --> BB[Bull Board]
Q --> F[Flower]
BB -->|Web UI| U1[Node.js Admin]
F -->|Web UI| U2[Python Admin]
BB -->|API| M[Metrics]
F -->|API| M
M --> G[Grafana]
Bull Board Setup
# Bull Board is a Node.js tool, but we simulate its API
import json
import time
class BullBoardAPI:
def __init__(self):
self.queues = {}
def register_queue(self, name, jobs=None):
self.queues[name] = {
'name': name,
'jobs': jobs or [],
'meta': {
'waiting': 0,
'active': 0,
'completed': 0,
'failed': 0,
'delayed': 0,
}
}
def add_job(self, queue_name, job):
if queue_name in self.queues:
self.queues[queue_name]['jobs'].append(job)
self._update_meta(queue_name)
def _update_meta(self, queue_name):
jobs = self.queues[queue_name]['jobs']
meta = self.queues[queue_name]['meta']
meta['waiting'] = len([j for j in jobs if j['status'] == 'waiting'])
meta['active'] = len([j for j in jobs if j['status'] == 'active'])
meta['completed'] = len([j for j in jobs if j['status'] == 'completed'])
meta['failed'] = len([j for j in jobs if j['status'] == 'failed'])
def get_queue_status(self, queue_name):
return self.queues.get(queue_name, {}).get('meta', {})
def retry_job(self, queue_name, job_id):
for job in self.queues[queue_name]['jobs']:
if job['id'] == job_id and job['status'] == 'failed':
job['status'] = 'waiting'
job['retry_count'] = job.get('retry_count', 0) + 1
self._update_meta(queue_name)
return True
return False
def remove_job(self, queue_name, job_id):
self.queues[queue_name]['jobs'] = [
j for j in self.queues[queue_name]['jobs']
if j['id'] != job_id
]
self._update_meta(queue_name)
def dashboard_data(self):
return {
name: {
'meta': q['meta'],
'recent_jobs': q['jobs'][-10:],
}
for name, q in self.queues.items()
}
dashboard = BullBoardAPI()
dashboard.register_queue('email')
dashboard.register_queue('scans')
dashboard.register_queue('cleanup')
dashboard.add_job('email', {'id': 'e-1', 'name': 'welcome_email',
'status': 'completed', 'duration': 0.5})
dashboard.add_job('email', {'id': 'e-2', 'name': 'receipt_email',
'status': 'failed', 'error': 'SMTP timeout',
'duration': 30.0})
dashboard.add_job('scans', {'id': 's-1', 'name': 'malware_scan',
'status': 'active', 'progress': 45})
print("Email queue:", dashboard.get_queue_status('email'))
print("Retry e-2:", dashboard.retry_job('email', 'e-2'))
print("After retry:", dashboard.get_queue_status('email'))
Expected output:
Email queue: {'waiting': 0, 'active': 0, 'completed': 1, 'failed': 1, 'delayed': 0}
Retry e-2: True
After retry: {'waiting': 1, 'active': 0, 'completed': 1, 'failed': 0, 'delayed': 0}
Flower Dashboard for Celery
import json
import time
class FlowerSimulator:
def __init__(self):
self.workers = {}
self.tasks = {}
self.broker = {'queue_depth': 0}
def register_worker(self, name, concurrency=4):
self.workers[name] = {
'name': name,
'concurrency': concurrency,
'active': 0,
'processed': 0,
'failed': 0,
'status': 'online',
'last_heartbeat': time.time(),
}
def add_task(self, task_id, name, queue, state='PENDING'):
self.tasks[task_id] = {
'task_id': task_id,
'name': name,
'queue': queue,
'state': state,
'received': time.time(),
'worker': None,
}
def task_received(self, task_id, worker_name):
if task_id in self.tasks:
self.tasks[task_id]['state'] = 'RECEIVED'
self.tasks[task_id]['worker'] = worker_name
self.broker['queue_depth'] = max(0, self.broker['queue_depth'] - 1)
if worker_name in self.workers:
self.workers[worker_name]['active'] += 1
def task_succeeded(self, task_id):
if task_id in self.tasks:
self.tasks[task_id]['state'] = 'SUCCESS'
worker = self.tasks[task_id]['worker']
if worker and worker in self.workers:
self.workers[worker]['active'] -= 1
self.workers[worker]['processed'] += 1
def task_failed(self, task_id):
if task_id in self.tasks:
self.tasks[task_id]['state'] = 'FAILURE'
worker = self.tasks[task_id]['worker']
if worker and worker in self.workers:
self.workers[worker]['active'] -= 1
self.workers[worker]['failed'] += 1
def get_worker_summary(self):
return [{
'name': w['name'],
'status': w['status'],
'active': w['active'],
'processed': w['processed'],
'failed': w['failed'],
} for w in self.workers.values()]
def get_broker_status(self):
return {
'queue_depth': self.broker['queue_depth'],
'worker_count': len(self.workers),
}
flower = FlowerSimulator()
flower.register_worker('worker-1', concurrency=4)
flower.register_worker('worker-2', concurrency=4)
flower.add_task('t-1', 'send_email', 'email', 'PENDING')
flower.add_task('t-2', 'scan_file', 'scans', 'PENDING')
flower.broker['queue_depth'] = 2
flower.task_received('t-1', 'worker-1')
flower.task_succeeded('t-1')
flower.task_received('t-2', 'worker-2')
flower.task_failed('t-2')
print("Workers:", json.dumps(flower.get_worker_summary(), indent=2))
print("Broker:", flower.get_broker_status())
Expected output:
Workers: [
{"name": "worker-1", "status": "online", "active": 0, "processed": 1, "failed": 0},
{"name": "worker-2", "status": "online", "active": 0, "processed": 0, "failed": 1}
]
Broker: {'queue_depth': 0, 'worker_count': 2}
Custom Dashboard Builder
import json
import time
class CustomDashboard:
def __init__(self):
self.panels = {}
def add_panel(self, name, panel_type, query):
self.panels[name] = {
'type': panel_type,
'query': query,
'data': None,
}
def refresh(self, data_source):
for name, panel in self.panels.items():
if panel['query'] == 'queue_depth':
panel['data'] = {q: data_source.get_depth(q)
for q in data_source.queues}
elif panel['query'] == 'worker_count':
panel['data'] = {
'total': len(data_source.workers),
'online': sum(1 for w in data_source.workers if w['status'] == 'online'),
}
elif panel['query'] == 'failure_rate':
total = data_source.total_jobs()
failed = data_source.failed_jobs()
panel['data'] = {'rate': (failed / total * 100) if total > 0 else 0}
elif panel['query'] == 'throughput':
panel['data'] = data_source.throughput_last_hour()
def render_html(self):
html = '<div class="dashboard">\n'
for name, panel in self.panels.items():
html += f' <div class="panel" id="{name}">\n'
html += f' <h3>{name}</h3>\n'
html += f' <pre>{json.dumps(panel["data"], indent=2)}</pre>\n'
html += ' </div>\n'
html += '</div>'
return html
class MockDataSource:
def __init__(self):
self.queues = ['email', 'scans', 'cleanup']
self.workers = [
{'name': 'w1', 'status': 'online'},
{'name': 'w2', 'status': 'online'},
{'name': 'w3', 'status': 'offline'},
]
self._total = 1500
self._failed = 23
def get_depth(self, queue):
depths = {'email': 45, 'scans': 120, 'cleanup': 0}
return depths.get(queue, 0)
def total_jobs(self):
return self._total
def failed_jobs(self):
return self._failed
def throughput_last_hour(self):
return {'email': 320, 'scans': 150, 'cleanup': 12}
dash = CustomDashboard()
dash.add_panel('Queue Depth', 'gauge', 'queue_depth')
dash.add_panel('Workers', 'stat', 'worker_count')
dash.add_panel('Failure Rate', 'gauge', 'failure_rate')
dash.add_panel('Throughput', 'chart', 'throughput')
ds = MockDataSource()
dash.refresh(ds)
print(dash.render_html())
Expected output:
<div class="dashboard">
<div class="panel" id="Queue Depth">
<h3>Queue Depth</h3>
<pre>{"email": 45, "scans": 120, "cleanup": 0}</pre>
</div>
...
</div>
Common Mistakes
1. Dashboard Without Authentication
Job dashboards expose sensitive system information. Always protect with authentication (basic auth, OAuth, or VPN).
2. Not Using Real-Time Updates
Static pages require manual refresh. Use Websocket or Server-Sent Events for live dashboard updates.
3. Overloading with Too Many Metrics
A dashboard with 50 charts is useless. Focus on 5-10 key metrics: queue depth, throughput, error rate, latency, worker count.
4. No Alert Integration
Dashboards are passive. Integrate with alerting so operators are notified before they check the dashboard.
5. Ignoring Historical Data
Current state without trends is misleading. Add time-series charts showing metrics over the last hour, day, and week.
Practice Questions
1. What information does a job dashboard show?
Queue depth per queue, active/idle workers, job counts by status (waiting, active, completed, failed), and recent job details with duration.
2. How does Bull Board differ from Flower?
Bull Board is for Node.js Bull queues. Flower is for Python Celery queues. Both provide web UI for monitoring and managing jobs.
3. What operations can you perform from a dashboard?
View job details, retry failed jobs, remove stalled jobs, pause/resume queues, and inspect worker status.
4. Why add authentication to dashboards?
Job dashboards expose internal queue structure, worker topology, and job data. Unauthenticated access is a security risk.
Challenge
Build a dashboard that shows: real-time queue depth for 3 queues, worker count and status, job throughput (last hour), failure rate with trend, and ability to retry failed jobs with one click.
FAQ
Mini Project: Dashboard UI
import json
import time
class JobDashboard:
def __init__(self):
self.queues = {}
def update_queue(self, name, waiting=0, active=0, completed=0, failed=0):
self.queues[name] = {
'waiting': waiting, 'active': active,
'completed': completed, 'failed': failed,
'total': waiting + active + completed + failed,
}
def overall_health(self):
total_failed = sum(q['failed'] for q in self.queues.values())
total = sum(q['total'] for q in self.queues.values())
error_rate = (total_failed / total * 100) if total > 0 else 0
if error_rate > 5:
return 'critical'
elif error_rate > 1:
return 'warning'
return 'healthy'
def to_json(self):
return {
'queues': self.queues,
'health': self.overall_health(),
'updated_at': time.time(),
}
dash = JobDashboard()
dash.update_queue('email', waiting=5, active=2, completed=100, failed=1)
dash.update_queue('scans', waiting=20, active=4, completed=500, failed=15)
dash.update_queue('cleanup', waiting=0, active=0, completed=200, failed=0)
print(json.dumps(dash.to_json(), indent=2))
Expected output:
{
"queues": {
"email": {"waiting": 5, "active": 2, "completed": 100, "failed": 1, "total": 108},
"scans": {"waiting": 20, "active": 4, "completed": 500, "failed": 15, "total": 539},
"cleanup": {"waiting": 0, "active": 0, "completed": 200, "failed": 0, "total": 200}
},
"health": "critical",
"updated_at": ...
}
What's Next
Now that you understand dashboards, explore job monitoring alerting for production alerting, then learn about structured logging for jobs.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro