feat: kafka bus tracer support

pkg/bus/bus.go

@@ -1,6 +1,10 @@
 package bus
 
-import "context"
+import (
+	"context"
+
+	"go.opentelemetry.io/otel/attribute"
+)
 
 type BusMessageType string
 
@@ -34,3 +38,10 @@ type MessageBusWriter[T BusMessage[any]] interface {
 	PublishMany(context.Context, []T) error
 	Close() error
 }
+
+func BusMessageAttributes(msg BusMessage[any]) []attribute.KeyValue {
+	return []attribute.KeyValue{
+		attribute.String("sro.bus.message.type", string(msg.GetType())),
+		attribute.String("sro.bus.message.id", msg.GetId()),
+	}
+}

pkg/bus/bus_kafka.go

@@ -5,6 +5,7 @@ import (
 	"sync"
 
 	"github.com/ShatteredRealms/go-common-service/pkg/config"
+	"go.opentelemetry.io/otel/trace"
 )
 
 var (
@@ -17,6 +18,8 @@ type kafkaBus[T BusMessage[any]] struct {
 	brokers config.ServerAddresses
 	topic   string
 
+	tracer trace.Tracer
+
 	mu sync.Mutex
 	wg sync.WaitGroup
 }

pkg/bus/bus_kafka_reader.go

@@ -12,6 +12,7 @@ import (
 	"github.com/ShatteredRealms/go-common-service/pkg/config"
 	"github.com/ShatteredRealms/go-common-service/pkg/log"
 	"github.com/segmentio/kafka-go"
+	"go.opentelemetry.io/otel"
 )
 
 // MessageBusReader is for reading messages from the message bus synchronously.
@@ -236,6 +237,7 @@ func NewKafkaMessageBusReader[T BusMessage[any]](brokers config.ServerAddresses,
 		kafkaBus: &kafkaBus[T]{
 			brokers: brokers,
 			topic:   string(msg.GetType()),
+			tracer:  otel.Tracer(fmt.Sprintf("sro.bus.kafka.reader.%s", msg.GetType())),
 		},
 		groupId: groupId,
 	}

pkg/bus/bus_kafka_writer.go

@@ -13,6 +13,7 @@ import (
 	"github.com/ShatteredRealms/go-common-service/pkg/config"
 	"github.com/ShatteredRealms/go-common-service/pkg/log"
 	"github.com/segmentio/kafka-go"
+	"go.opentelemetry.io/otel"
 )
 
 // MessageBusWriter is for writing message asynchronously to the message bus.
@@ -23,13 +24,18 @@ type kafkaBusWriter[T BusMessage[any]] struct {
 
 // Publish implements MessageBus.
 func (k *kafkaBusWriter[T]) Publish(ctx context.Context, msg T) error {
+	ctx, span := k.tracer.Start(ctx, "Publish")
+	defer span.End()
+	span.SetAttributes(BusMessageAttributes(msg)...)
 	k.setupWriter()
 
 	val, err := k.encodeMessage(msg)
 	if err != nil {
 		return err
 	}
 
+	ctx, innerSpan := k.tracer.Start(ctx, "Publish.SendMessages")
+	defer innerSpan.End()
 	k.wg.Add(1)
 	defer k.wg.Done()
 	err = k.Writer.WriteMessages(ctx, kafka.Message{
@@ -47,6 +53,14 @@ func (k *kafkaBusWriter[T]) PublishMany(ctx context.Context, msgs []T) error {
 	k.wg.Add(1)
 	defer k.wg.Done()
 
+	if len(msgs) == 0 {
+		return nil
+	}
+
+	ctx, span := k.tracer.Start(ctx, "PublishMany")
+	span.SetAttributes(BusMessageAttributes(msgs[0])[0])
+	defer span.End()
+
 	k.setupWriter()
 
 	messages := make([]kafka.Message, len(msgs))
@@ -55,6 +69,7 @@ func (k *kafkaBusWriter[T]) PublishMany(ctx context.Context, msgs []T) error {
 	errsMu := sync.Mutex{}
 	wg := sync.WaitGroup{}
 
+	ctx, innerSpan := k.tracer.Start(ctx, "PublishMany.CreateMessages")
 	for chunk := range slices.Chunk(msgs, runtime.NumCPU()) {
 		wg.Add(1)
 		go func(chunk []T) {
@@ -79,10 +94,13 @@ func (k *kafkaBusWriter[T]) PublishMany(ctx context.Context, msgs []T) error {
 	}
 
 	wg.Wait()
+	innerSpan.End()
 	if errs != nil {
 		return errs
 	}
 
+	ctx, innerSpan = k.tracer.Start(ctx, "PublishMany.SendMessages")
+	defer innerSpan.End()
 	err := k.Writer.WriteMessages(ctx, messages...)
 	if err != nil {
 		return fmt.Errorf("%w: %w", ErrSendingMessage, err)
@@ -139,6 +157,7 @@ func NewKafkaMessageBusWriter[T BusMessage[any]](brokers config.ServerAddresses,
 		kafkaBus: &kafkaBus[T]{
 			brokers: brokers,
 			topic:   string(msg.GetType()),
+			tracer:  otel.Tracer(fmt.Sprintf("sro.bus.kafka.reader.%s", msg.GetType())),
 		},
 	}
 }

pkg/srospan/attributes.go

@@ -1,6 +1,8 @@
 package srospan
 
-import "go.opentelemetry.io/otel/attribute"
+import (
+	"go.opentelemetry.io/otel/attribute"
+)
 
 func SourceOwnerId(val string) attribute.KeyValue {
 	return attribute.String("sro.source.owner.id", val)