opentelemetry-instrumentation-absurd

OpenTelemetry instrumentation for the Absurd durable-task SDK (absurd-sdk).

Absurd (absurd_sdk.AsyncAbsurd) is a Postgres-backed durable task queue: a producer spawn()s a task and a worker work_batch() claims and runs the registered handler, possibly in a different process. Without instrumentation the two halves live in separate traces, so a task's execution cannot be linked back to whatever spawned it.

This instrumentor follows the standard producer/consumer span shape for a task queue:

• AsyncAbsurd.spawn() is wrapped to open a CLIENT span named after the task and inject the current trace context into the task's headers. Absurd persists headers on the task row and hands them back to the handler via ctx.headers.
• AsyncAbsurd.register_task() is wrapped so every registered handler runs inside a running:<task> SERVER span whose parent is extracted from those headers.

The two spans share one trace, so a task's execution is linked back to whatever spawned it — even when the spawner and the worker run in different processes.

trigger (your span)
└── report                 CLIENT   ← spawn()  (process A)
    └── running:report     SERVER   ← handler  (process B)

Installation

pip install opentelemetry-instrumentation-absurd
# or
uv add opentelemetry-instrumentation-absurd

Usage

Install the instrumentor once per process, on both the spawn side and the worker side:

from opentelemetry.instrumentation.absurd import AbsurdInstrumentor

AbsurdInstrumentor().instrument()

That is all — no changes to your spawn() calls or task handlers are required. Only AsyncAbsurd is instrumented.

You can pass an explicit TracerProvider:

AbsurdInstrumentor().instrument(tracer_provider=my_tracer_provider)

Because this package ships an opentelemetry_instrumentor entry point, it is also picked up automatically by opentelemetry-instrument:

opentelemetry-instrument python your_app.py

Producer / consumer example

import asyncio

from absurd_sdk import AsyncAbsurd
from opentelemetry.instrumentation.absurd import AbsurdInstrumentor

AbsurdInstrumentor().instrument()

app = AsyncAbsurd('postgresql://localhost/absurd', queue_name='workflows')


@app.register_task(name='report')
async def report(params, ctx):
    # Runs inside a `running:report` SERVER span parented to the spawner.
    return {'value': params['value'] * 2}


async def main() -> None:
    # Opens a `report` CLIENT span and injects the trace context into the task headers.
    await app.spawn('report', {'value': 7}, queue='workflows')


asyncio.run(main())

Span attributes

Both spans carry these attributes (set when available):

The trace context is propagated through the task headers under the otel.absurd.context key using the globally configured OpenTelemetry propagator (W3C traceparent by default), so any caller headers you pass are preserved.

How it works

spawn() and the registered handler are wrapped with wrapt. On spawn() the current context is injected into a fresh carrier nested under a single header key; the worker extracts that carrier and parents its SERVER span to the spawner's CLIENT span. If a task was spawned before instrumentation (no carrier present), the handler still runs in its own SERVER span — it just starts a new trace.

Compatibility

• Python 3.10+
• absurd-sdk >= 0.4.0
• opentelemetry-api >= 1.0

Development

make install   # uv sync --all-groups
make test      # uv run pytest
make check     # ruff check + ruff format --check

AI disclaimer

This code was written with the help of AI.

License

MIT