structlog in Celery Workers

Celery configures logging on its own, forks worker processes, and reuses each one for task after task — three behaviours that each break a naive structlog setup in a different way. This page covers the signals that put you back in control, the per-task context binding, and the queue boundary that trace context has to cross. It builds on the structlog processors and pipelines guide, part of the modern Python logging libraries deep dive section.

Prerequisites

pip install "celery>=5.4.0,<6.0.0" \
            "structlog>=24.1.0,<26.0.0" \
            "opentelemetry-instrumentation-celery>=0.48b0,<1.0.0"

export CELERY_BROKER_URL=redis://localhost:6379/0
export LOG_RENDERER=json

Implementation

Step 1 — Stop Celery configuring logging. Connecting setup_logging — even with a handler that does nothing but call your own setup — makes Celery skip its entire logging configuration.

# worker.py
from celery import Celery
from celery.signals import setup_logging, worker_process_init
from observability.log import configure          # the structlog configure() from the parent guide

app = Celery("tasks")

@setup_logging.connect
def _on_setup_logging(**_kwargs):
    configure()                                  # Celery now leaves logging alone entirely

@worker_process_init.connect
def _on_child_start(**_kwargs):
    configure()                                  # explicit in each forked child

Without the first handler, Celery installs its own root handler and colour formatter during worker startup, and every record the worker emits afterwards has a different shape from the ones your application emits — which usually shows up as "the worker logs are not JSON" long after the deploy.

Step 2 — Bind task context on entry. task_prerun fires in the worker process with the task and its ID in scope.

import structlog
from celery.signals import task_prerun, task_postrun

@task_prerun.connect
def _bind_task_context(task_id=None, task=None, **_kwargs):
    structlog.contextvars.clear_contextvars()    # start from nothing, every time
    structlog.contextvars.bind_contextvars(
        task=task.name,
        task_id=task_id,
        retry=task.request.retries,
    )

@task_postrun.connect
def _clear_task_context(**_kwargs):
    structlog.contextvars.clear_contextvars()

Clearing in both handlers is deliberate. task_postrun covers the normal path; the clear at the top of task_prerun covers the case where a worker was killed mid-task or a signal handler failed, leaving context behind that would otherwise be inherited by the next task.

Step 3 — Carry the trace across the queue. The worker shares no memory with the producer, so trace context has to travel in the message. The OpenTelemetry Celery instrumentation does this automatically; by hand it is two signals.

from celery.signals import before_task_publish
from opentelemetry import trace
from opentelemetry.propagate import inject, extract

@before_task_publish.connect
def _inject_trace(headers=None, **_kwargs):
    if headers is not None:
        inject(headers)                          # writes traceparent into the message headers

@task_prerun.connect
def _bind_trace(task=None, **_kwargs):
    ctx = extract(task.request.get("headers") or {})
    span = trace.get_current_span(ctx)
    sc = span.get_span_context()
    if sc.is_valid:
        structlog.contextvars.bind_contextvars(
            trace_id=format(sc.trace_id, "032x"),
            span_id=format(sc.span_id, "016x"),
        )

The full treatment of that boundary — including what happens to retries and to the apply_async path — is in propagating trace context across Celery tasks.

Step 4 — Use the task logger, or don't. Celery's get_task_logger returns a logger under the celery.task namespace with task metadata attached by Celery's own formatter. With structlog binding that metadata into contextvars, it adds nothing — a plain structlog.get_logger(__name__) is simpler and gives the same fields.

import structlog

log = structlog.get_logger(__name__)

@app.task(bind=True, max_retries=3)
def process_order(self, order_id: int) -> None:
    log.info("processing order", order_id=order_id)
    try:
        charge(order_id)
    except PaymentTimeout as exc:
        log.warning("payment timed out, retrying", order_id=order_id)
        raise self.retry(exc=exc, countdown=2 ** self.request.retries)

Step 5 — Keep the worker's own records in the pipeline. Celery's internal loggers — celery.worker, celery.app.trace, celery.redirected — go through the standard library, so they need the same ProcessorFormatter treatment as any other dependency.

CELERY_LOGGERS = {
    "celery":            {"level": "INFO", "propagate": True},
    "celery.app.trace":  {"level": "INFO", "propagate": True},   # task succeeded/failed lines
    "celery.redirected": {"level": "WARNING", "propagate": True},  # stdout/stderr capture
}

celery.app.trace is the one worth keeping at INFO: it emits the "Task succeeded in 0.42s" and "Task raised unexpected" records, which are the worker's equivalent of an access log.

Configuration options

Verification

celery -A worker worker --loglevel=INFO --concurrency=2

Queue a task from a traced web request and check that both sides share a trace ID and that consecutive tasks do not share a task ID.

Expected Output:

{"event": "queueing order", "logger": "api.orders", "level": "info", "trace_id": "4bf92f3577b34da6a3ce929d0e0e4736", "order_id": 8812}
{"event": "processing order", "logger": "tasks.orders", "level": "info", "task": "tasks.process_order", "task_id": "8f1c...b207", "retry": 0, "trace_id": "4bf92f3577b34da6a3ce929d0e0e4736", "order_id": 8812}
{"event": "payment timed out, retrying", "logger": "tasks.orders", "level": "warning", "task_id": "8f1c...b207", "retry": 0, "trace_id": "4bf92f3577b34da6a3ce929d0e0e4736"}
{"event": "processing order", "logger": "tasks.orders", "level": "info", "task_id": "8f1c...b207", "retry": 1, "trace_id": "4bf92f3577b34da6a3ce929d0e0e4736"}

Two properties to check specifically: the trace_id on the first line — emitted by the web process — matches every worker line, and retry increments while task_id does not. Then queue two different tasks in sequence and confirm the second one's records carry its own task_id, which is the test for the clearing step.

Common mistakes

Worker logs are not JSON

Error signature: application records are JSON; Celery's own startup and task records are coloured text. Root cause: Celery configured logging during worker startup and replaced the root handler. Remediation: connect the setup_logging signal, which makes Celery skip its configuration entirely.

Records carry the previous task's ID

Error signature: a task's records are attributed to a task that finished a second earlier. Root cause: contextvars bound in task_prerun and never cleared, in a prefork worker that reuses the process. Remediation: clear in task_postrun and again at the top of task_prerun.

The trace stops at the queue

Error signature: the producing request has a trace; the task has a different one, or none. Root cause: trace context was never written into the message headers, so the worker had nothing to extract. Remediation: inject at before_task_publish and extract at task_prerun, or install the OpenTelemetry Celery instrumentation, which does both.

What a worker's log should contain

A web service's log is organised around requests; a worker's is organised around tasks, and the useful field set differs accordingly. Six fields cover most of what anyone asks of a worker log.

The task name and the task ID, bound in task_prerun, which together identify the unit of work. The retry count, because it is the only field that distinguishes attempts of the same task. The queue, when the worker consumes more than one, since queue depth problems are per-queue. The trace ID, carried from whatever queued the task. And the outcome with a duration, emitted once when the task finishes, which is the record that answers "how long do these take" without needing every intermediate step.

from celery.signals import task_postrun
import time

@task_prerun.connect
def _start(task_id=None, task=None, **_kw):
    task.request.__dict__["_started"] = time.perf_counter()

@task_postrun.connect
def _finish(task_id=None, task=None, state=None, **_kw):
    started = task.request.__dict__.get("_started")
    structlog.get_logger("tasks").info(
        "task finished",
        state=state,
        duration_s=round(time.perf_counter() - started, 3) if started else None,
    )
    structlog.contextvars.clear_contextvars()

That single record per task is worth more than a dozen progress records, because it carries the outcome and the duration as fields you can aggregate rather than as text you can read.

The failure modes worth logging deliberately

Three worker-specific failures produce no useful record unless you arrange one, and all three are common enough to be worth the signal handlers.

A task that raises. Celery logs this through celery.app.trace at ERROR with the traceback, which is adequate — provided that logger propagates to your handlers, which is the configuration point covered above. Connect task_failure if you want the failure in your own event vocabulary rather than Celery's.

A task that is revoked or times out. These do not raise inside the task; the worker terminates it. task_revoked carries the reason, and a soft time limit raises SoftTimeLimitExceeded inside the task, which is catchable and is the better of the two mechanisms precisely because it is.

A worker that stops consuming. Nothing raises. The signal is absence — no task records from that worker — which means the detection belongs in metrics rather than logs: a heartbeat gauge per worker, alerting on staleness. That is the same argument made for any component that can fail by disappearing.

Prefork, threads and the context

The default prefork pool gives each task a whole process, so contextvars behave straightforwardly as long as they are cleared between tasks. Two other pool types change that.

With the threads pool, several tasks run concurrently in one process on different threads. Contextvars are per-thread-of-execution, so bindings do not leak between concurrent tasks — but a module-level logger.bind() result shared across them does, which is an argument for using bind_contextvars rather than a shared bound logger.

With gevent or eventlet, the picture is more complicated: greenlets have their own context semantics, and monkey-patching happens before your configuration runs. If you use those pools, verify the isolation with a test that runs two tasks concurrently and asserts each one's records carry its own task_id — the same test is worth having under any pool, and under those two it is worth running before trusting anything else on this page.

Related

• structlog processors and pipelines — the parent guide: the configuration these signals apply.
• Writing custom structlog processors — processors for task-specific enrichment.
• Testing structlog output with pytest — asserting on the fields these signals bind.
• Propagating trace context across Celery tasks — the queue boundary in full.
• Thread-safe logging in multiprocessing — the same fork semantics for the standard library.

Frequently Asked Questions