Collecting Metrics from Celery Workers

A Celery deployment has two places where things go wrong, and they need different metrics. Inside the workers, tasks succeed, fail, retry and take longer than expected. Outside them, in the broker, tasks wait — and when workers are stuck or too few, the waiting is the whole story, invisible to the workers themselves. This article covers both: recording task outcomes with Celery's signals, aggregating across the prefork pool, and measuring the backlog from the broker. It is part of metrics in multi-process Python servers in the Python metrics and instrumentation section.

Prerequisites

pip install "celery>=5.3.0,<6.0.0" "prometheus-client>=0.20.0,<1.0.0" "redis>=5.0.0,<6.0.0"

A Celery application with the prefork pool, and a Prometheus server that can scrape both the workers and a small broker exporter.

Implementation steps

Step 1 — Define task metrics. Counters for outcomes, a histogram for runtime, and the task name as the principal label. Task names are a bounded set defined in code, so they are safe as label values; task arguments are not, and never belong in labels.

# myservice/celery_metrics.py
from prometheus_client import Counter, Histogram, Gauge

TASKS = Counter("celery_tasks_total", "Tasks finished", ["task", "outcome"])
RUNTIME = Histogram("celery_task_runtime_seconds", "Task runtime", ["task"],
                    buckets=(0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60, 120, 300))
RETRIES = Counter("celery_task_retries_total", "Task retries", ["task"])
ACTIVE = Gauge("celery_tasks_active", "Tasks running", ["task"], multiprocess_mode="livesum")

The runtime buckets reach much further than an HTTP latency histogram's, because task runtimes span seconds to minutes. Choosing them follows the same reasoning as choosing histogram buckets for latency SLOs: put boundaries where decisions are made.

Step 2 — Record with signals. The signals fire in the pool process running the task, before and after it, and on failure and retry.

import time
from celery import signals
from myservice.celery_metrics import TASKS, RUNTIME, RETRIES, ACTIVE

_started: dict[str, float] = {}

@signals.task_prerun.connect
def _prerun(task_id, task, **_):
    _started[task_id] = time.perf_counter()
    ACTIVE.labels(task=task.name).inc()

@signals.task_postrun.connect
def _postrun(task_id, task, state, **_):
    ACTIVE.labels(task=task.name).dec()
    t0 = _started.pop(task_id, None)
    if t0 is not None:
        RUNTIME.labels(task=task.name).observe(time.perf_counter() - t0)
    TASKS.labels(task=task.name, outcome=(state or "UNKNOWN").lower()).inc()

@signals.task_retry.connect
def _retry(request, **_):
    RETRIES.labels(task=request.task).inc()

task_postrun fires for successes and failures alike, with the final state, so one handler counts every outcome. Retries are counted separately because a retried task's postrun reports RETRY, and a retry rate is a distinct signal — often the first sign a downstream dependency is struggling.

Step 3 — Aggregate across the pool. Set PROMETHEUS_MULTIPROC_DIR in the worker's environment and clean it at startup, exactly as for Gunicorn in Prometheus multiprocess mode with Gunicorn. Celery's worker_process_shutdown signal is the equivalent of Gunicorn's child_exit for gauge cleanup.

import os
from prometheus_client import multiprocess

@signals.worker_process_shutdown.connect
def _pool_exit(pid=None, **_):
    multiprocess.mark_process_dead(pid or os.getpid())

Step 4 — Serve from the main process. The main worker process runs no tasks, and it is the one process that lives for the worker's whole life, so it is where the HTTP server belongs. The worker_init signal fires there once.

from prometheus_client import CollectorRegistry, start_http_server

@signals.worker_init.connect
def _serve_metrics(**_):
    registry = CollectorRegistry()
    multiprocess.MultiProcessCollector(registry)
    start_http_server(9808, registry=registry)

Measuring the backlog from the broker

Queue depth is the metric most needed during a Celery incident, and workers cannot provide it. A worker knows about the tasks it has prefetched, not the thousands still in the broker. When every worker is busy with a slow task, the workers report nothing new at all.

A small exporter, running separately from the workers, reads the broker directly. For Redis, each queue is a list and its length is the depth; for RabbitMQ, the management API reports messages ready and unacknowledged per queue.

# broker_exporter.py — runs as its own small deployment
import json, time, redis
from prometheus_client import Gauge, start_http_server

DEPTH = Gauge("celery_queue_depth", "Tasks waiting", ["queue"])
OLDEST = Gauge("celery_queue_oldest_task_age_seconds", "Age of the oldest waiting task", ["queue"])
QUEUES = ["default", "reports"]

def main():
    r = redis.Redis.from_url("redis://redis:6379/0")
    start_http_server(9809)
    while True:
        for q in QUEUES:
            DEPTH.labels(queue=q).set(r.llen(q))
            tail = r.lindex(q, -1)                 # Celery pops from the tail
            age = 0.0
            if tail:
                sent = json.loads(tail)["headers"].get("sent_at")
                age = max(0.0, time.time() - sent) if sent else 0.0
            OLDEST.labels(queue=q).set(age)
        time.sleep(15)

The age of the oldest task needs an enqueue timestamp, which Celery does not add by default. A before_task_publish signal in the producers can add one to the headers:

@signals.before_task_publish.connect
def _stamp(headers=None, **_):
    headers["sent_at"] = time.time()

The exporter is deliberately independent of the workers: its own deployment, its own scrape target, its own failure modes. If it ran inside a worker it would share the worker's fate, and the one incident it exists for — every worker stuck — would silence it too. It is also cheap, a couple of broker commands per queue every fifteen seconds, so running two replicas for resilience costs nothing measurable.

Age is a better alerting signal than depth. A depth of two thousand is fine for a queue that drains ten thousand a minute and alarming for one that drains ten; an oldest-task age of twenty minutes means the same thing everywhere — work is twenty minutes late.

Reading the worker and broker metrics together

Each set of metrics is useful alone, and the combination diagnoses most Celery incidents in a few minutes. The useful comparison is between the rate tasks arrive, the rate they finish, and how long each one takes.

Too few workers shows as rising age with normal runtimes and a completion rate that is flat at its ceiling: every worker is busy and every task is behaving, there just is not enough capacity. Autoscaling on oldest-task age, rather than on CPU, responds to exactly this case.

Slow tasks show as rising age with the runtime histogram shifting into its upper buckets. The cause is almost always a dependency — a database or an HTTP API the task calls — and a trace of one slow task names it, which is where tracing slow SQL queries in Python picks up.

Stuck workers are the case the broker exporter exists for. Completions fall to zero, the runtime histogram records nothing because nothing finishes, and the active gauge sits at the concurrency limit. The cause is a call with no timeout; Celery's task_time_limit bounds the damage by killing the pool process, and the resulting failures appear in the outcome counter.

Failure loops can leave age flat while the retry counter climbs, because tasks are being taken and put back. A retry rate alert catches them before the retries exhaust and the failures become permanent.

Configuration options

Verification

Enqueue a known number of tasks and compare counters:

from myservice.tasks import noop
for _ in range(500):
    noop.delay()

curl -s localhost:9808/metrics | grep 'celery_tasks_total{outcome="success",task="myservice.tasks.noop"}'

Expected Output: five hundred, whichever pool processes ran them.

celery_tasks_total{outcome="success",task="myservice.tasks.noop"} 500.0

Then stop the workers, enqueue a hundred more, and watch the broker exporter: depth reads one hundred and the oldest-task age rises by fifteen seconds each scrape — the incident view, with no worker running.

Common mistakes

Metrics server started in every pool process. Error signature: Address already in use in the pool, or metrics from one random child. Root cause: the server started at import. Remediation: worker_init, which runs once in the main process.

No multiprocess mode. Error signature: task counters stuck at zero. Root cause: the main process serves its own registry, and it runs no tasks. Remediation: the multiprocess directory and collector.

Task arguments as labels. Error signature: a series count that grows with every order or user. Root cause: unbounded values in labels. Remediation: the task name only; arguments belong on spans or log records.

Queue depth from inspect(). Error signature: depths that look fine during an outage. Root cause: inspect asks workers, which see only prefetched tasks and do not answer when stuck. Remediation: read the broker.

Alerting on depth alone. Error signature: pages for harmless bursts, silence for a slow trickle that nobody drains. Root cause: depth without drain rate. Remediation: alert on the oldest task's age.

Prefetch hiding the backlog. Error signature: broker depth near zero while tasks are late. Root cause: workers prefetching many tasks each, moving the backlog from the broker into worker memory, where the exporter cannot see it. Remediation: worker_prefetch_multiplier = 1 for long tasks, and acks_late so unfinished tasks return to the queue.

Timestamps from the wrong clock. Error signature: negative or wildly large ages. Root cause: producer and exporter clocks disagree. Remediation: synchronised clocks, and clamping age at zero as the exporter does.

No time limit on tasks. Error signature: active gauge pinned at the concurrency limit for hours. Root cause: a blocking call without a timeout. Remediation: task_time_limit and task_soft_time_limit, with the soft limit logged and counted.

Frequently Asked Questions