Celery Worker Inspection: Monitoring Tasks, Workers, and Queues at Runtime

In this tutorial, you will learn about Celery Worker Inspection: Monitoring Tasks, Workers, and Queues at Runtime. We cover key concepts, practical examples, and best practices to help you master this topic.

Celery worker inspection provides runtime visibility into worker state through the inspect API and CLI, enabling queries for active tasks, registered task types, worker statistics, scheduled ETA tasks, reserved prefetched tasks, and queue depths.

flowchart LR
    Admin[Monitor/Admin] -->|inspect()| I[Inspect API]
    I -->|active()| W1[Worker: Active Tasks]
    I -->|registered()| W2[Worker: Registered Tasks]
    I -->|stats()| W3[Worker: Statistics]
    I -->|scheduled()| W4[Worker: Scheduled Tasks]
    I -->|reserved()| W5[Worker: Reserved Tasks]
    I -->|active_queues()| W6[Worker: Queues]
    W1 -->|Reply| Admin

What You'll Learn

• Inspect API methods and usage
• Active, reserved, scheduled task queries
• Worker statistics and health
• Queue inspection and broker state
• Building monitoring dashboards

Why It Matters

Without inspection, you have no visibility into what workers are doing. You cannot tell if a task is stuck, if workers are overloaded, or if tasks are being prefetched. Inspection provides the data needed for debugging and capacity planning.

Real-World Use

DodaTech's monitoring system calls inspect.active() every 15 seconds to detect stuck tasks. If a task runs for more than 5 minutes, an alert fires. Inspect.stats() feeds a capacity planning dashboard showing per-worker throughput trends.

Basic Inspection

from celery import Celery
import json

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

@app.task
def task_a(x):
    return x * 2

@app.task
def task_b(x):
    return x ** 2

@app.task
def task_c(x):
    return x + 1

def inspect_workers():
    i = app.control.inspect()
    report = {
        'active': i.active(),
        'registered': i.registered(),
        'scheduled': i.scheduled(),
        'reserved': i.reserved(),
        'stats': {k: v for k, v in (i.stats() or {}).items()},
    }
    print(json.dumps(report, indent=2, default=str))
    return report

task_a.delay(10)
task_b.delay(5)
task_c.delay(100)
inspect_workers()

Expected output:

{
  "active": {"celery@host": [{"id": "id1", "name": "task_a"}]},
  "registered": {"celery@host": ["inspect.task_a", "inspect.task_b", "inspect.task_c"]},
  "scheduled": {"celery@host": []},
  "reserved": {"celery@host": [{"id": "id2", "name": "task_b"}]},
  "stats": {"celery@host": {"total": 150, "active": 1, "processed": 100}}
}

Worker Statistics

from celery import Celery
import json

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

@app.task
def benchmark_task(n):
    total = sum(i * i for i in range(n))
    return total

def get_worker_health():
    i = app.control.inspect()
    stats = i.stats()
    if not stats:
        return {"status": "unreachable"}

    health = {}
    for worker, s in stats.items():
        health[worker] = {
            'total_tasks_processed': s.get('total', 0),
            'uptime_seconds': s.get('uptime', 0),
            'pool_size': s.get('pool', {}).get('max-concurrency', 0),
            'pool_active': s.get('pool', {}).get('writes', {}).get('inqueues', 0),
            'prefetch_count': s.get('prefetch_count', 0),
            'broker': {
                'hostname': s.get('broker', {}).get('hostname', 'unknown'),
                'port': s.get('broker', {}).get('port', 0),
            },
            'clock': s.get('clock', 'unknown'),
            'pid': s.get('pid', 0),
        }
    return health

for i in range(10):
    benchmark_task.delay(10000)

health = get_worker_health()
print(json.dumps(health, indent=2))

Expected output:

{
  "celery@host": {
    "total_tasks_processed": 150,
    "uptime_seconds": 86400,
    "pool_size": 8,
    "pool_active": 3,
    "prefetch_count": 10,
    "broker": {"hostname": "localhost", "port": 6379},
    "pid": 12345
  }
}

Queue Inspection

from celery import Celery
import json

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

app.conf.task_queues = {
    'default': {'exchange': 'default', 'routing_key': 'default'},
    'high_priority': {'exchange': 'high_priority', 'routing_key': 'high_priority'},
    'bulk': {'exchange': 'bulk', 'routing_key': 'bulk'},
}

def inspect_queues():
    i = app.control.inspect()
    active_queues = i.active_queues()
    if not active_queues:
        return {"queues": []}

    queue_info = {}
    for worker, queues in active_queues.items():
        for q in queues:
            name = q.get('name')
            if name not in queue_info:
                queue_info[name] = {
                    'workers': [],
                    'binding': q.get('binding'),
                    'exchange': q.get('exchange', {}).get('name'),
                }
            queue_info[name]['workers'].append(worker)

    return queue_info

queue_status = inspect_queues()
print(json.dumps(queue_status, indent=2))

Expected output:

{
  "default": {
    "workers": ["celery@host"],
    "exchange": "default"
  },
  "high_priority": {
    "workers": ["celery@host"],
    "exchange": "high_priority"
  },
  "bulk": {
    "workers": ["celery@host"],
    "exchange": "bulk"
  }
}

Common Mistakes

• Calling inspect on unreachable workers -- inspect returns None for unreachable workers. Always check for None before accessing reply data. Use try/except to handle connection failures.
• Ignoring reserved tasks -- active shows currently executing tasks. reserved shows tasks prefetched but not started. reserved tasks are invisible unless you query for them. Always check both.
• Statistics counter overflow -- stats.total is an integer that can overflow on long-running workers processing millions of tasks. Use the clock value for monotonic ordering instead.
• Inspect with large reply data -- on clusters with 100+ workers, inspect replies can be megabytes. Use destination to target specific workers or implement pagination for large clusters.
• Not inspecting prefetch count -- high prefetch count with task_acks_late means many tasks are reserved. If the worker crashes, all reserved tasks are lost or delayed. Monitor prefetch count as a risk metric.

Practice Questions

1. What is the difference between active, reserved, and scheduled tasks?
2. How do you get per-worker statistics?
3. What information does active_queues() provide?
4. How can you detect a stuck task using inspect?
5. Why should you check reserved tasks in addition to active tasks?

Challenge

Build a real-time worker monitoring dashboard that: (1) polls inspect.active(), inspect.reserved(), and inspect.stats() every 10 seconds, (2) maintains a 30-minute rolling history of active/reserved counts, (3) detects anomalies (task running > 5 minutes, prefetch count > 50), (4) shows per-worker throughput (tasks/sec) over time, and (5) generates a weekly report showing peak usage times and worker utilization rates.

FAQ

Mini Project

Build a Celery worker health monitor: (1) expose a Flask/HTTP endpoint that returns JSON from inspect().stats() and inspect().active() for all workers, (2) cache the inspect results for 5 seconds to reduce broker load, (3) add an endpoint for per-worker detail showing last 100 completed tasks with timestamps, (4) implement a dead-worker detector that alerts if a worker misses 3 consecutive inspect pings, and (5) serve a real-time dashboard using server-sent events.

What's Next

Continue with Custom Task Classes to learn how to create reusable task base classes. Then explore Unit Testing for reliable task testing strategies.