make tracer better

pkg/tracing/interfaces.go

@@ -0,0 +1,26 @@
+package tracing
+
+import (
+	"context"
+
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/trace"
+)
+
+// TracerInterface defines the methods for tracing
+type TracerInterface interface {
+	StartSpan(ctx context.Context, operation string, attrs ...attribute.KeyValue) (context.Context, trace.Span)
+	StartClientSpan(ctx context.Context, operation string, attrs ...attribute.KeyValue) (context.Context, trace.Span)
+	RecordError(span trace.Span, err error)
+	AddAttributes(span trace.Span, attrs ...attribute.KeyValue)
+	AddGoogleCloudAttributes(span trace.Span, projectID, region, zone string)
+	AddServiceAttributes(span trace.Span, serviceName, serviceVersion, environment string)
+	AddRequestAttributes(span trace.Span, method, path, userAgent string, statusCode int)
+	AddDatabaseAttributes(span trace.Span, operation, table string, duration float64)
+	AddKafkaAttributes(span trace.Span, topic, operation string, partition int32, offset int64)
+}
+
+// ConfigInterface defines the methods for configuration
+type ConfigInterface interface {
+	Validate() error
+}

pkg/tracing/kafka.go

@@ -0,0 +1,41 @@
+package tracing
+
+import (
+	"context"
+
+	"github.com/IBM/sarama"
+	"go.opentelemetry.io/otel/propagation"
+)
+
+// InjectTraceContext injects OpenTelemetry trace context into Kafka message headers
+// for propagation to downstream consumers.
+func InjectTraceContext(ctx context.Context, headers []sarama.RecordHeader) []sarama.RecordHeader {
+	carrier := propagation.MapCarrier{}
+	propagator := propagation.TraceContext{}
+	propagator.Inject(ctx, carrier)
+
+	// Create new headers slice to avoid mutation
+	newHeaders := make([]sarama.RecordHeader, len(headers), len(headers)+len(carrier))
+	copy(newHeaders, headers)
+
+	for k, v := range carrier {
+		newHeaders = append(newHeaders, sarama.RecordHeader{
+			Key:   []byte(k),
+			Value: []byte(v),
+		})
+	}
+
+	return newHeaders
+}
+
+// ExtractTraceContext extracts OpenTelemetry trace context from Kafka message headers
+// for use in downstream consumers.
+func ExtractTraceContext(ctx context.Context, headers []sarama.RecordHeader) context.Context {
+	carrier := propagation.MapCarrier{}
+	for _, h := range headers {
+		carrier[string(h.Key)] = string(h.Value)
+	}
+
+	propagator := propagation.TraceContext{}
+	return propagator.Extract(ctx, carrier)
+}

pkg/tracing/tracer.go

@@ -2,7 +2,6 @@ package tracing
 
 import (
 	"context"
-	"log/slog"
 	"time"
 
 	"go.opentelemetry.io/otel"
@@ -11,34 +10,58 @@ import (
 	"go.opentelemetry.io/otel/trace"
 )
 
+const (
+	AttrGCPProjectID = "gcp.project_id"
+	AttrGCPRegion    = "gcp.region"
+	AttrGCPZone      = "gcp.zone"
+
+	AttrServiceName        = "service.name"
+	AttrServiceVersion     = "service.version"
+	AttrServiceEnvironment = "service.environment"
+
+	AttrHTTPMethod     = "http.method"
+	AttrHTTPRoute      = "http.route"
+	AttrHTTPUserAgent  = "http.user_agent"
+	AttrHTTPStatusCode = "http.status_code"
+
+	AttrDBOperation  = "db.operation"
+	AttrDBTable      = "db.table"
+	AttrDBDurationMs = "db.duration_ms"
+
+	AttrMessagingSystem         = "messaging.system"
+	AttrMessagingDestination    = "messaging.destination"
+	AttrMessagingOperation      = "messaging.operation"
+	AttrMessagingKafkaPartition = "messaging.kafka.partition"
+	AttrMessagingKafkaOffset    = "messaging.kafka.offset"
+)
+
 // Tracer provides Google Cloud compliant tracing
 type Tracer struct {
 	tracer trace.Tracer
-	logger *slog.Logger
 }
 
 // NewTracer creates a new tracer instance
-func NewTracer(tracer trace.Tracer, logger *slog.Logger) *Tracer {
+func NewTracer(tracer trace.Tracer) *Tracer {
 	return &Tracer{
 		tracer: tracer,
-		logger: logger,
 	}
 }
 
-// StartSpan creates a new span with Google Cloud attributes
-func (t *Tracer) StartSpan(ctx context.Context, operation string, attrs ...attribute.KeyValue) (context.Context, trace.Span) {
-	ctx, span := t.tracer.Start(ctx, operation,
-		trace.WithAttributes(attrs...),
-		trace.WithSpanKind(trace.SpanKindServer),
-	)
-	return ctx, span
+// StartServerSpan creates a new server span with Google Cloud attributes
+func (t *Tracer) StartServerSpan(ctx context.Context, operation string, attrs ...attribute.KeyValue) (context.Context, trace.Span) {
+	return t.startSpan(ctx, operation, trace.SpanKindServer, attrs...)
 }
 
 // StartClientSpan creates a new client span
 func (t *Tracer) StartClientSpan(ctx context.Context, operation string, attrs ...attribute.KeyValue) (context.Context, trace.Span) {
+	return t.startSpan(ctx, operation, trace.SpanKindClient, attrs...)
+}
+
+// startSpan is a helper to start a span with given kind and attributes
+func (t *Tracer) startSpan(ctx context.Context, operation string, kind trace.SpanKind, attrs ...attribute.KeyValue) (context.Context, trace.Span) {
 	ctx, span := t.tracer.Start(ctx, operation,
 		trace.WithAttributes(attrs...),
-		trace.WithSpanKind(trace.SpanKindClient),
+		trace.WithSpanKind(kind),
 	)
 	return ctx, span
 }
@@ -59,48 +82,48 @@ func (t *Tracer) AddAttributes(span trace.Span, attrs ...attribute.KeyValue) {
 // AddGoogleCloudAttributes adds Google Cloud specific attributes
 func (t *Tracer) AddGoogleCloudAttributes(span trace.Span, projectID, region, zone string) {
 	span.SetAttributes(
-		attribute.String("gcp.project_id", projectID),
-		attribute.String("gcp.region", region),
-		attribute.String("gcp.zone", zone),
+		attribute.String(AttrGCPProjectID, projectID),
+		attribute.String(AttrGCPRegion, region),
+		attribute.String(AttrGCPZone, zone),
 	)
 }
 
 // AddServiceAttributes adds service-specific attributes
 func (t *Tracer) AddServiceAttributes(span trace.Span, serviceName, serviceVersion, environment string) {
 	span.SetAttributes(
-		attribute.String("service.name", serviceName),
-		attribute.String("service.version", serviceVersion),
-		attribute.String("service.environment", environment),
+		attribute.String(AttrServiceName, serviceName),
+		attribute.String(AttrServiceVersion, serviceVersion),
+		attribute.String(AttrServiceEnvironment, environment),
 	)
 }
 
 // AddRequestAttributes adds HTTP request attributes
 func (t *Tracer) AddRequestAttributes(span trace.Span, method, path, userAgent string, statusCode int) {
 	span.SetAttributes(
-		attribute.String("http.method", method),
-		attribute.String("http.route", path),
-		attribute.String("http.user_agent", userAgent),
-		attribute.Int("http.status_code", statusCode),
+		attribute.String(AttrHTTPMethod, method),
+		attribute.String(AttrHTTPRoute, path),
+		attribute.String(AttrHTTPUserAgent, userAgent),
+		attribute.Int(AttrHTTPStatusCode, statusCode),
 	)
 }
 
 // AddDatabaseAttributes adds database operation attributes
 func (t *Tracer) AddDatabaseAttributes(span trace.Span, operation, table string, duration time.Duration) {
 	span.SetAttributes(
-		attribute.String("db.operation", operation),
-		attribute.String("db.table", table),
-		attribute.Float64("db.duration_ms", float64(duration.Milliseconds())),
+		attribute.String(AttrDBOperation, operation),
+		attribute.String(AttrDBTable, table),
+		attribute.Float64(AttrDBDurationMs, float64(duration.Milliseconds())),
 	)
 }
 
 // AddKafkaAttributes adds Kafka operation attributes
 func (t *Tracer) AddKafkaAttributes(span trace.Span, topic, operation string, partition int32, offset int64) {
 	span.SetAttributes(
-		attribute.String("messaging.system", "kafka"),
-		attribute.String("messaging.destination", topic),
-		attribute.String("messaging.operation", operation),
-		attribute.Int64("messaging.kafka.partition", int64(partition)),
-		attribute.Int64("messaging.kafka.offset", offset),
+		attribute.String(AttrMessagingSystem, "kafka"),
+		attribute.String(AttrMessagingDestination, topic),
+		attribute.String(AttrMessagingOperation, operation),
+		attribute.Int64(AttrMessagingKafkaPartition, int64(partition)),
+		attribute.Int64(AttrMessagingKafkaOffset, offset),
 	)
 }
 

services/url/cmd/url/main.go

@@ -73,8 +73,9 @@ func main() {
 	defer dbPool.Close()
 
 	// Initialize layers
-	ps := storage.NewPostgresStorage(dbPool)
-	urlSvc := service.NewURLService(ps, l)
+	tracer := tracing.NewTracer(tracing.GetTracer(serviceName))
+	ps := storage.NewPostgresStorage(dbPool, tracer)
+	urlSvc := service.NewURLService(ps, l, tracer)
 	healthSvc := service.NewHealthService(ps, l)
 
 	// Kafka producers setup
@@ -102,15 +103,15 @@ func main() {
 	var wg sync.WaitGroup
 
 	l.Info("Before NewProducer")
-	notificationProducer := kafka.NewProducer(kafkaAsyncProducer, kafkaNotifTopic, l, &wg)
+	notificationProducer := kafka.NewProducer(kafkaAsyncProducer, kafkaNotifTopic, l, &wg, tracer)
 	l.Info("After NewProducer")
 
 	l.Info("Calling notificationProducer.Start()")
 
 	notificationProducer.Start(ctx)
 
 	httpClient := &http.Client{Timeout: 5 * time.Second}
-	chkr := checker.NewURLChecker(urlSvc, l, httpClient, 1*time.Minute, notificationProducer)
+	chkr := checker.NewURLChecker(urlSvc, l, httpClient, 1*time.Minute, notificationProducer, tracer)
 	go chkr.Start(ctx)
 
 	urlHandler := handler.NewURLHandler(urlSvc, l)

services/url/internal/checker/checker.go

@@ -7,6 +7,9 @@ import (
 	"sync"
 	"time"
 
+	"go.opentelemetry.io/otel/attribute"
+
+	"github.com/samims/hcaas/pkg/tracing"
 	"github.com/samims/hcaas/services/url/internal/kafka"
 	"github.com/samims/hcaas/services/url/internal/metrics"
 	"github.com/samims/hcaas/services/url/internal/model"
@@ -24,6 +27,7 @@ type URLChecker struct {
 	httpClient           *http.Client
 	interval             time.Duration
 	notificationProducer kafka.NotificationProducer
+	tracer               *tracing.Tracer
 }
 
 func NewURLChecker(
@@ -32,17 +36,22 @@ func NewURLChecker(
 	client *http.Client,
 	interval time.Duration,
 	producer kafka.NotificationProducer,
+	tracer *tracing.Tracer,
 ) *URLChecker {
 	if producer == nil {
 		// This panic indicates a serious configuration error that should be caught
 		panic("NewURLChecker: notificationProducer cannot be nil")
 	}
+	if tracer == nil {
+		panic("NewURLChecker: tracer cannot be nil")
+	}
 	return &URLChecker{
 		svc:                  svc,
 		logger:               logger,
 		httpClient:           client,
 		interval:             interval,
 		notificationProducer: producer,
+		tracer:               tracer,
 	}
 }
 
@@ -64,9 +73,13 @@ func (uc *URLChecker) Start(ctx context.Context) {
 }
 
 func (uc *URLChecker) CheckAllURLs(ctx context.Context) {
+	ctx, span := uc.tracer.StartServerSpan(ctx, "CheckAllURLs")
+	defer span.End()
+
 	urls, err := uc.svc.GetAll(ctx)
 	if err != nil {
 		uc.logger.Error("Failed to fetch URLs", slog.Any("error", err))
+		span.RecordError(err)
 		return
 	}
 
@@ -80,18 +93,28 @@ func (uc *URLChecker) CheckAllURLs(ctx context.Context) {
 			sem <- struct{}{}
 			defer func() { <-sem }()
 
+			ctx, span := uc.tracer.StartClientSpan(ctx, "CheckURL")
+			defer span.End()
+
 			uc.logger.Info("Checking URL", slog.String("id", url.ID), slog.String("address", url.Address))
 
 			status := uc.ping(ctx, url.Address)
 			uc.logger.Info("After ping", slog.String("url_id", url.ID), slog.Any("address", url.Address), slog.String("status", status))
 
+			span.SetAttributes(
+				attribute.String("url.id", url.ID),
+				attribute.String("url.address", url.Address),
+				attribute.String("url.status", status),
+			)
+
 			err := uc.svc.UpdateStatus(ctx, url.ID, status)
 			if err != nil {
 				uc.logger.Error("Failed to update URL status",
 					slog.String("urlID", url.ID),
 					slog.String("status", status),
 					slog.Any("error", err),
 				)
+				span.RecordError(err)
 			} else {
 				uc.logger.Info("URL status updated",
 					slog.String("urlID", url.ID),
@@ -112,6 +135,7 @@ func (uc *URLChecker) CheckAllURLs(ctx context.Context) {
 						uc.logger.Error("Failed to publish notification",
 							slog.String("url_id", url.ID),
 							slog.Any("error", err))
+						span.RecordError(err)
 					}
 				}
 			}