Every OpenTelemetry tutorial ends at "add the middleware to FastAPI." That gets
you traces for HTTP requests. It tells you nothing about what happens inside
your Celery workers.
The hard part is propagating trace context from your FastAPI handler into the
Celery task, so a single request trace shows the full chain: HTTP → queue →
worker → database.
Here's the exact setup I use in production.
Why context propagation breaks by default
When FastAPI handles a request, the OTEL SDK creates a span and stores the
trace context in Python's contextvars. When you call .delay() to queue a
Celery task, that context is not automatically passed — Celery serialises
the task arguments to JSON and ships them to Redis. The trace context lives
in memory, not in the message. The worker starts with no idea what trace
it belongs to.
The fix: manually inject the trace context into the task headers, then
extract it in the worker before creating child spans.
Step 1: Inject context when queuing
from opentelemetry import propagate
from opentelemetry.trace import get_current_span
def queue_task_with_context(payload: dict):
carrier = {}
propagate.inject(carrier) # pulls current trace context into dict
process_order.apply_async(
kwargs={"payload": payload},
headers={"otel_context": carrier},
)
Step 2: Extract context in the worker
from celery.signals import task_prerun
from opentelemetry import propagate, trace
tracer = trace.get_tracer(__name__)
@task_prerun.connect
def on_task_prerun(task_id, task, args, kwargs, **kw):
headers = task.request.get("otel_context", {})
ctx = propagate.extract(headers) # restore parent trace context
span = tracer.start_span(
name=f"celery.{task.name}",
context=ctx,
kind=trace.SpanKind.CONSUMER,
)
task.request.otel_span = span
task.request.otel_token = trace.use_span(span, end_on_exit=False).__enter__()
from celery.signals import task_postrun, task_failure
@task_postrun.connect
def on_task_postrun(task, **kw):
span = getattr(task.request, "otel_span", None)
if span:
span.set_status(trace.StatusCode.OK)
span.end()
@task_failure.connect
def on_task_failure(task, exception, **kw):
span = getattr(task.request, "otel_span", None)
if span:
span.record_exception(exception)
span.set_status(trace.StatusCode.ERROR)
span.end()
# SDK setup in your FastAPI app
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace.export import BatchSpanProcessor
exporter = OTLPSpanExporter(endpoint="http://otel-collector:4317", insecure=True)
provider.add_span_processor(BatchSpanProcessor(exporter))
What you get
One trace ID that follows a request from the HTTP layer through the queue
into the Celery worker and down into your database query. When something
breaks at 2 AM, you open Grafana, search by trace ID, and see exactly
where it failed and how long each step took.
Without this, you have logs from three different services with no way to
connect them. With this, you have one timeline.
The full runnable repo with docker-compose, FastAPI app, Celery worker,
and Grafana Tempo is at github.com/itsjayanth/observability-starter-kit.
