OpenTelemetry Integration Guide

Celery has official auto-instrumentation — use it instead of writing manual spans:

pip install opentelemetry-instrumentation-celery

from celery.signals import worker_process_init
from opentelemetry.instrumentation.celery import CeleryInstrumentor

@worker_process_init.connect(weak=False)
def init_celery_tracing(*args, **kwargs):
    CeleryInstrumentor().instrument()

For RQ, APScheduler, raw threads, or anything else, use the tracer API directly:

from opentelemetry import trace
from opentelemetry.trace.status import Status, StatusCode

tracer = trace.get_tracer("my-service-worker")

def process_email(payload: dict) -> None:
    with tracer.start_as_current_span(
        "email.send",
        attributes={
            "messaging.system": "rq",
            "messaging.operation": "process",
            "messaging.destination.name": "emails",
            "code.function": "process_email",
        },
    ) as span:
        try:
            send_email(payload)
        except Exception as exc:
            span.record_exception(exc)
            span.set_status(Status(StatusCode.ERROR))
            raise

import (
    "context"
    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/attribute"
    "go.opentelemetry.io/otel/codes"
    "go.opentelemetry.io/otel/trace"
    "github.com/hibiken/asynq"
)

var tracer = otel.Tracer("my-service-worker")

func ProcessEmail(ctx context.Context, t *asynq.Task) error {
    ctx, span := tracer.Start(ctx, "email.send",
        trace.WithSpanKind(trace.SpanKindConsumer),
        trace.WithAttributes(
            attribute.String("messaging.system", "asynq"),
            attribute.String("messaging.operation", "process"),
            attribute.String("messaging.destination.name", t.Type()),
            attribute.String("code.function", "ProcessEmail"),
        ))
    defer span.End()

    if err := sendEmail(ctx, t.Payload()); err != nil {
        span.RecordError(err)
        span.SetStatus(codes.Error, err.Error())
        return err
    }
    return nil
}

use opentelemetry::{global, trace::{Tracer, Status, SpanKind}, KeyValue};

pub async fn process_email(payload: EmailJob) -> Result<(), Error> {
    let tracer = global::tracer("my-service-worker");
    let mut span = tracer
        .span_builder("email.send")
        .with_kind(SpanKind::Consumer)
        .with_attributes(vec![
            KeyValue::new("messaging.system", "redis"),
            KeyValue::new("messaging.operation", "process"),
            KeyValue::new("messaging.destination.name", "emails"),
            KeyValue::new("code.function", "process_email"),
        ])
        .start(&tracer);

    match send_email(payload).await {
        Ok(()) => Ok(()),
        Err(e) => {
            span.record_error(&e);
            span.set_status(Status::error(e.to_string()));
            Err(e)
        }
    }
}

Sidekiq has official auto-instrumentation — use it instead of writing manual spans:

require 'opentelemetry/instrumentation/sidekiq'

OpenTelemetry::SDK.configure do |c|
  c.use 'OpenTelemetry::Instrumentation::Sidekiq'
end

For ActiveJob, raw threads, or anything else, use the tracer API directly:

tracer = OpenTelemetry.tracer_provider.tracer('my-service-worker')

def process_email(payload)
  tracer.in_span('email.send', kind: :consumer, attributes: {
    'messaging.system' => 'sidekiq',
    'messaging.operation' => 'process',
    'messaging.destination.name' => 'emails',
    'code.function' => 'process_email',
  }) do |span|
    begin
      send_email(payload)
    rescue => e
      span.record_exception(e)
      span.status = OpenTelemetry::Trace::Status.error(e.message)
      raise
    end
  end
end