Skip to content
Pydantic Docs

Celery

See every background task your Celery workers run: the task name, how long it took, and whether it failed, as a span (one unit of work with a name, a start, and a duration) in Logfire. Spans link together into a trace (the full journey of one request), so a task shows up right next to the request that enqueued it. Tasks scheduled with Celery beat are covered too.

What you’ll capture

  • Each task run as a span, with its duration and status
  • The task name and the arguments it ran with
  • Failed tasks, with the error
  • Tasks fired on a schedule by Celery beat

Before you start

You’ll need a Logfire project. Open Add data in your project (top navigation) and follow the setup for your language: it signs your machine in with logfire auth (a browser sign-in, no token to copy) and, for production or other languages, creates a write token (the credential your app uses to send data). New to Logfire? Start with Getting Started.

Installation

Install logfire with the celery extra:

Terminal
pip install 'logfire[celery]'

Usage

Call logfire.configure(), then logfire.instrument_celery() to record every task. Call both inside Celery’s worker_process_init signal so they run once in each worker process, after the prefork pool forks it.

The example below uses Redis as the broker. You can start one with Docker:

Terminal
docker run --rm -d -p 127.0.0.1:6379:6379 redis

Then run the worker with celery -A tasks worker --loglevel=info:

tasks.py
from celery import Celery
from celery.signals import worker_process_init

import logfire


@worker_process_init.connect(weak=False)  # (1)
def init_worker(*args, **kwargs):
  logfire.configure(service_name='worker')  # (2)
  logfire.instrument_celery()


app = Celery('tasks', broker='redis://localhost:6379/0')  # (3)


@app.task
def add(x: int, y: int):
  return x + y


add.delay(42, 50)  # (4)

Celery emits signals at points in the application lifecycle. worker_process_init fires once inside each worker process, after Celery forks it from the parent. Initializing in worker_init (which runs in the parent, before the fork) would build the exporter's background thread in the wrong process, so spans wouldn't export.

Set a service_name so you can tell this service's spans apart from the rest in Logfire.

Install redis with pip install redis.

Trigger the task. In your own app you'll more likely use app.send_task("tasks.add", args=[42, 50]), which hands the task to the broker and returns immediately.

Verify it worked

Trigger a task, then open the Live view. Within a few seconds you’ll see a span named after the task, with its duration and status: click it to see the arguments it ran with.

Troubleshooting

Not seeing your tasks? Check that logfire.configure() ran before instrument_celery(), that your write token is set (run logfire projects use <your-project> locally, or set the LOGFIRE_TOKEN environment variable in production; see Getting Started), and that you called instrument_celery() exactly once per worker process.

Advanced

Distributed tracing

To follow a task from the code that enqueued it all the way to the worker that ran it, call logfire.instrument_celery() in both places:

  1. The worker processes that execute tasks.
  2. The application that enqueues tasks (for example, your Django or FastAPI web server).

This propagates the trace context (the small piece of tracking information passed alongside each task) from the enqueuing app to the worker, so both ends appear in one trace. See the distributed tracing guide for details.

Celery beat

If you schedule periodic tasks with Celery beat, instrument the beat process too, in its own beat_init signal. Building on the example above:

tasks.py
from celery import Celery
from celery.signals import beat_init, worker_process_init

import logfire


@worker_process_init.connect(weak=False)
def init_worker(*args, **kwargs):
  logfire.configure(service_name='worker')
  logfire.instrument_celery()


@beat_init.connect(weak=False)  # (1)
def init_beat(*args, **kwargs):
  logfire.configure(service_name='beat')  # (2)
  logfire.instrument_celery()


app = Celery('tasks', broker='redis://localhost:6379/0')
app.conf.beat_schedule = {  # (3)
  'add-every-30-seconds': {
      'task': 'tasks.add',
      'schedule': 30.0,
      'args': (16, 16),
  },
}


@app.task
def add(x: int, y: int):
  return x + y

The beat_init signal fires once, as the beat process starts.

Use a different service_name so beat's spans are easy to tell apart.

Add a task to the beat schedule. See the beat schedule docs for the full format.

This schedules add to run every 30 seconds with the arguments 16 and 16. Run the beat process with:

Terminal
celery -A tasks beat --loglevel=info

Passing options to the OpenTelemetry instrumentor

The keyword arguments of logfire.instrument_celery() are passed straight to the CeleryInstrumentor().instrument() method. See the OpenTelemetry Celery instrumentation docs for the full option list.

Reference