Celery needs to be instrumented on client side for distributed tracing

[alexmojaki]

Demo:

import logfire
from celery import Celery
from celery.signals import worker_init
from fastapi import FastAPI

@worker_init.connect()
def init_worker(*args, **kwargs):
    logfire.configure(service_name="worker")
    logfire.instrument_celery()

app = Celery("tasks", broker="redis://localhost:6379/0")

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

fapp = FastAPI()

logfire.configure()
logfire.instrument_fastapi(fapp)
########### Important:
logfire.instrument_celery()


@fapp.get("/hello")
async def hello(name: str):
    add.delay(42, 50)
    return {"message": f"hello {name}"}


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(fapp)

The top-level logfire.instrument_celery() is required for the context propagation, based on open-telemetry/opentelemetry-python-contrib#1002 (comment)

This should be documented at least, maybe fixed in some other way, e.g. in OTel.

[dhruv-ahuja]

Should this be a warning in the existing documentation, or maybe a separate section where we extend the existing celery example to showcase distributed tracing example?

Code snippet below, in which case user can run the same tasks.py file.

import logfire
from celery import Celery
from celery.signals import worker_init


# celery worker configuration
@worker_init.connect()
def init_worker(*args, **kwargs):
    logfire.configure(service_name='worker')
    logfire.instrument_celery()
    print('logfire instrumented')


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


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


if __name__ == '__main__':
    print('running as a client app')
    logfire.configure(service_name='client-app')
    logfire.instrument_celery()
    add.delay(42, 50)

[alexmojaki]

A separate section sounds good. It should set distributed_tracing=True in the worker to prevent the warning.

[dhruv-ahuja]

created PR for the same