OpenTelemetry

Hatchet supports exporting traces from your tasks to an OpenTelemetry Collector to improve visibility into your Hatchet tasks.

Usage

Setup

Hatchet’s SDK provides an instrumentor that auto-instruments Hatchet code if you opt in. Setup is straightforward:

First, install the otel extra with (e.g.) pip install hatchet-sdk[otel]. Then, import the instrumentor:

from path.to.your.trace.provider import trace_provider
from hatchet_sdk.opentelemetry.instrumentor import HatchetInstrumentor
 
HatchetInstrumentor(tracer_provider=trace_provider).instrument()

You bring your own trace provider and plug it into the HatchetInstrumentor, call instrument, and that’s it!

Spans

By default, Hatchet creates spans at the following points in the lifecycle of a task run:

1. When a trigger is run on the client side, e.g. run() or push() is called.
2. When a worker handles a task event, such as starting to run the task or cancelling the task

In addition, you’ll get a handful of attributes set (prefixed by hatchet.) on the task run events, such as the task name and the worker ID, as well as success/failure states, and so on.

Some other important notes:

1. The instrumentor will automatically propagate the trace context between task runs, so if you spawn a task from another task, the child will correctly show up as a child of its parent in the trace waterfall.
2. You can exclude specific attributes from being attached to spans by providing the otel configuration option on the ClientConfig and passing a list of excluded_attributes, which come from this list.

Integrations

Hatchet’s instrumentor is easy to integrate with a number of third-party tracing tools.

Langfuse

For example, you might be interested in using Langfuse for tracing an LLM-intensive application.

First, configure the Langfuse client as described by their documentation:

LANGFUSE_AUTH = base64.b64encode(
    f"{os.getenv('LANGFUSE_PUBLIC_KEY')}:{os.getenv('LANGFUSE_SECRET_KEY')}".encode()
).decode()

os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"] = (
    os.getenv("LANGFUSE_HOST", "https://us.cloud.langfuse.com") + "/api/public/otel"
)
os.environ["OTEL_EXPORTER_OTLP_HEADERS"] = f"Authorization=Basic {LANGFUSE_AUTH}"

## Note: Langfuse sets the global tracer provider, so you don't need to worry about it
lf = Langfuse(
    public_key=os.getenv("LANGFUSE_PUBLIC_KEY"),
    secret_key=os.getenv("LANGFUSE_SECRET_KEY"),
    host=os.getenv("LANGFUSE_HOST", "https://app.langfuse.com"),
)

Langfuse will set the global tracer provider, so you don’t have to do it manually.

Next, create an OpenAI client using Langfuse’s OpenAI wrapper langfuse.openai as a drop-in replacement for the default OpenAI client:

openai = AsyncOpenAI(
    api_key=os.getenv("OPENAI_API_KEY"),
)

And that’s it! Now you’re ready to instrument your Hatchet workers with Langfuse. For example, create a task like this:

HatchetInstrumentor(
    ## Langfuse sets the global tracer provider
    tracer_provider=get_tracer_provider(),
).instrument()

hatchet = Hatchet()


@hatchet.task()
async def langfuse_task(input: EmptyModel, ctx: Context) -> dict[str, str | None]:
    ## Usage, cost, etc. of this call will be send to Langfuse
    generation = await openai.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "You are a helpful assistant."},
            {"role": "user", "content": "Where does Anna Karenina take place?"},
        ],
    )

    location = generation.choices[0].message.content

    return {
        "location": location,
    }

And finally, run the task to view the Langfuse traces (cost, usage, etc.) interspersed with Hatchet’s traces, in addition to any other traces you may have:

tracer = get_client()


async def main() -> None:
    # Traces will send to Langfuse
    # Use `_otel_tracer` to access the OpenTelemetry tracer if you need
    # to e.g. log statuses or attributes manually.
    with tracer._otel_tracer.start_as_current_span(name="trigger") as span:
        result = await langfuse_task.aio_run()
        location = result.get("location")

        if not location:
            span.set_status(StatusCode.ERROR)
            return

        span.set_attribute("location", location)

When you run this task, you’ll see a trace like this in Langfuse!