Tracing Distributed Transactions Across Async Queue Boundaries with OpenTelemetry Context Propagation — code snippets

snippet-1.go

package tracing

import (
	"github.com/IBM/sarama"
	"go.opentelemetry.io/otel/propagation"
)

// SaramaProducerCarrier injects trace context into Sarama producer message headers.
type SaramaProducerCarrier struct {
	Headers *[]sarama.RecordHeader
}

// Get returns the value associated with the passed key.
func (c SaramaProducerCarrier) Get(key string) string {
	if c.Headers == nil {
		return ""
	}
	for _, h := range *c.Headers {
		if string(h.Key) == key {
			return string(h.Value)
		}
	}
	return ""
}

// Set sets the key-value pair in the message headers.
func (c SaramaProducerCarrier) Set(key, val string) {
	for i, h := range *c.Headers {
		if string(h.Key) == key {
			(*c.Headers)[i].Value = []byte(val)
			return
		}
	}
	*c.Headers = append(*c.Headers, sarama.RecordHeader{
		Key:   []byte(key),
		Value: []byte(val),
	})
}

// Keys lists the keys in the carrier.
func (c SaramaProducerCarrier) Keys() []string {
	if c.Headers == nil {
		return nil
	}
	keys := make([]string, len(*c.Headers))
	for i, h := range *c.Headers {
		keys[i] = string(h.Key)
	}
	return keys
}

// Ensure interface compatibility at compile time.
var _ propagation.TextMapCarrier = (*SaramaProducerCarrier)(nil)

snippet-2.go

package tracing

import (
	"context"
	"fmt"

	"github.com/IBM/sarama"
	"go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/trace"
)

const TracerName = "order-service-producer"

type OrderPublisher struct {
	producer sarama.SyncProducer
	tracer   trace.Tracer
}

func NewOrderPublisher(p sarama.SyncProducer) *OrderPublisher {
	return &OrderPublisher{
		producer: p,
		tracer:   otel.Tracer(TracerName),
	}
}

func (op *OrderPublisher) PublishOrderCreated(ctx context.Context, orderID string, payload []byte) error {
	// 1. Start a producer span using the appropriate span kind
	ctx, span := op.tracer.Start(ctx, "orders.publish_created",
		trace.WithSpanKind(trace.SpanKindProducer),
		trace.WithAttributes(
			attribute.String("messaging.system", "kafka"),
			attribute.String("messaging.destination", "orders"),
			attribute.String("messaging.destination_kind", "topic"),
			attribute.String("messaging.kafka.message_key", orderID),
		),
	)
	defer span.End()

	// 2. Prepare the Kafka message headers
	headers := []sarama.RecordHeader{
		{Key: []byte("content-type"), Value: []byte("application/json")},
	}

	// 3. Inject the span context into the headers using the custom carrier
	carrier := SaramaProducerCarrier{Headers: &headers}
	otel.GetTextMapPropagator().Inject(ctx, carrier)

	msg := &sarama.ProducerMessage{
		Topic:   "orders",
		Key:     sarama.StringEncoder(orderID),
		Value:   sarama.ByteEncoder(payload),
		Headers: headers,
	}

	// 4. Send the message
	partition, offset, err := op.producer.SendMessage(msg)
	if err != nil {
		span.RecordError(err)
		return fmt.Errorf("failed to send kafka message: %w", err)
	}

	span.SetAttributes(
		attribute.Int32("messaging.kafka.partition", partition),
		attribute.Int64("messaging.kafka.offset", offset),
	)

	return nil
}

snippet-3.go

package tracing

import (
	"github.com/IBM/sarama"
	"go.opentelemetry.io/otel/propagation"
)

// SaramaConsumerCarrier extracts trace context from Sarama consumer message headers.
type SaramaConsumerCarrier struct {
	Headers []*sarama.RecordHeader
}

// Get returns the value associated with the passed key.
func (c SaramaConsumerCarrier) Get(key string) string {
	for _, h := range c.Headers {
		if h != nil && string(h.Key) == key {
			return string(h.Value)
		}
	}
	return ""
}

// Set sets the key-value pair in the message headers.
func (c SaramaConsumerCarrier) Set(key, val string) {
	for i, h := range c.Headers {
		if h != nil && string(h.Key) == key {
			c.Headers[i].Value = []byte(val)
			return
		}
	}
	c.Headers = append(c.Headers, &sarama.RecordHeader{
		Key:   []byte(key),
		Value: []byte(val),
	})
}

// Keys lists the keys in the carrier.
func (c SaramaConsumerCarrier) Keys() []string {
	keys := make([]string, len(c.Headers))
	for i, h := range c.Headers {
		if h != nil {
			keys[i] = string(h.Key)
		}
	}
	return keys
}

var _ propagation.TextMapCarrier = (*SaramaConsumerCarrier)(nil)

snippet-4.go

package tracing

import (
	"context"
	"encoding/json"
	"log"

	"github.com/IBM/sarama"
	"go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/trace"
)

const ConsumerTracerName = "payment-service-consumer"

type OrderConsumer struct {
	tracer trace.Tracer
}

func NewOrderConsumer() *OrderConsumer {
	return &OrderConsumer{
		tracer: otel.Tracer(ConsumerTracerName),
	}
}

func (oc *OrderConsumer) HandleMessage(msg *sarama.ConsumerMessage) {
	// 1. Extract context from incoming message headers
	carrier := SaramaConsumerCarrier{Headers: msg.Headers}
	ctx := otel.GetTextMapPropagator().Extract(context.Background(), carrier)

	// 2. Start a consumer span, linked to the extracted parent context
	ctx, span := oc.tracer.Start(ctx, "orders.consume_created",
		trace.WithSpanKind(trace.SpanKindConsumer),
		trace.WithAttributes(
			attribute.String("messaging.system", "kafka"),
			attribute.String("messaging.destination", msg.Topic),
			attribute.String("messaging.destination_kind", "topic"),
			attribute.Int32("messaging.kafka.partition", msg.Partition),
			attribute.Int64("messaging.kafka.offset", msg.Offset),
			attribute.String("messaging.kafka.message_key", string(msg.Key)),
		),
	)
	defer span.End()

	// 3. Process message payload
	var order struct {
		ID    string  `json:"order_id"`
		Total float64 `json:"total"`
	}
	if err := json.Unmarshal(msg.Value, &order); err != nil {
		span.RecordError(err)
		return
	}

	span.SetAttributes(attribute.String("order.id", order.ID))

	// Execute actual business logic (e.g., process payment)
	if err := oc.processPayment(ctx, order.ID, order.Total); err != nil {
		span.RecordError(err)
		log.Printf("Failed to process payment for order %s: %v", order.ID, err)
	}
}

func (oc *OrderConsumer) processPayment(ctx context.Context, orderID string, total float64) error {
	// Inner database or HTTP call span will inherit the context and properly nest
	_, span := oc.tracer.Start(ctx, "payment.execute")
	defer span.End()
	
	// Payment gateway business logic goes here
	return nil
}

snippet-5.go

package tracing

import (
	amqp "github.com/rabbitmq/amqp091-go"
	"go.opentelemetry.io/otel/propagation"
)

// AmqpHeadersCarrier propagates trace context inside RabbitMQ AMQP message headers.
type AmqpHeadersCarrier struct {
	Table amqp.Table
}

// Get returns the value associated with the passed key.
func (c AmqpHeadersCarrier) Get(key string) string {
	if c.Table == nil {
		return ""
	}
	val, ok := c.Table[key]
	if !ok {
		return ""
	}
	strVal, ok := val.(string)
	if !ok {
		return ""
	}
	return strVal
}

// Set sets the key-value pair in the RabbitMQ table.
func (c AmqpHeadersCarrier) Set(key, val string) {
	if c.Table == nil {
		return
	}
	c.Table[key] = val
}

// Keys returns all keys in the carrier.
func (c AmqpHeadersCarrier) Keys() []string {
	keys := make([]string, 0, len(c.Table))
	for k := range c.Table {
		keys = append(keys, k)
	}
	return keys
}

var _ propagation.TextMapCarrier = (*AmqpHeadersCarrier)(nil)

snippet-6.go

package tracing

import (
	"context"
	"log"

	"go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
	"go.opentelemetry.io/otel/propagation"
	"go.opentelemetry.io/otel/sdk/resource"
	sdktrace "go.opentelemetry.io/otel/sdk/trace"
	semconv "go.opentelemetry.io/otel/semconv/v1.24.0"
)

// InitTracerProvider initializes an OTLP-backed Tracer Provider with custom sampling rules.
func InitTracerProvider(ctx context.Context, serviceName string, otlpEndpoint string, sampleRatio float64) (*sdktrace.TracerProvider, error) {
	exporter, err := otlptracegrpc.New(ctx,
		otlptracegrpc.WithInsecure(),
		otlptracegrpc.WithEndpoint(otlpEndpoint),
	)
	if err != nil {
		return nil, err
	}

	res, err := resource.New(ctx,
		resource.WithAttributes(
			semconv.ServiceNameKey.String(serviceName),
			semconv.DeploymentEnvironmentKey.String("production"),
		),
	)
	if err != nil {
		return nil, err
	}

	// ParentBased sampler respects the parent's sampling decision.
	// If no parent is present, it falls back to the specified TraceIDRatioBased sampler.
	sampler := sdktrace.ParentBased(
		sdktrace.TraceIDRatioBased(sampleRatio),
	)

	tp := sdktrace.NewTracerProvider(
		sdktrace.WithSampler(sampler),
		sdktrace.WithBatcher(exporter),
		sdktrace.WithResource(res),
	)

	otel.SetTracerProvider(tp)
	
	// Register the global propagator for tracing and baggage
	otel.SetTextMapPropagator(propagation.NewCompositeTextMapPropagator(
		propagation.TraceContext{},
		propagation.Baggage{},
	))

	log.Printf("Successfully initialized OTel Tracer Provider for %s with sample ratio %.2f", serviceName, sampleRatio)
	return tp, nil
}