feat(api): retain error spans and sample remaining traces at 50%

api/pkg/di/container.go

@@ -1925,7 +1925,7 @@ func (container *Container) initializeGoogleTraceProvider(version string, namesp
 	}
 
 	tp := trace.NewTracerProvider(
-		trace.WithBatcher(traceExporter),
+		trace.WithSpanProcessor(telemetry.NewSamplingSpanProcessor(traceExporter)),
 		trace.WithSampler(trace.AlwaysSample()),
 		trace.WithResource(container.OtelResources(version, namespace)),
 	)
@@ -1938,6 +1938,7 @@ func (container *Container) initializeGoogleTraceProvider(version string, namesp
 
 	meterProvider := metric.NewMeterProvider(
 		metric.WithReader(metric.NewPeriodicReader(metricExporter)),
+		metric.WithExemplarFilter(telemetry.SampledTraceExemplarFilter),
 		metric.WithResource(container.OtelResources(version, namespace)),
 	)
 	otel.SetMeterProvider(meterProvider)
@@ -1970,7 +1971,7 @@ func (container *Container) initializeAxiomTraceProvider(version string, namespa
 	}
 
 	tp := trace.NewTracerProvider(
-		trace.WithBatcher(traceExporter),
+		trace.WithSpanProcessor(telemetry.NewSamplingSpanProcessor(traceExporter)),
 		trace.WithSampler(trace.AlwaysSample()),
 		trace.WithResource(container.OtelResources(version, namespace)),
 	)
@@ -1997,6 +1998,7 @@ func (container *Container) initializeAxiomTraceProvider(version string, namespa
 
 	meterProvider := metric.NewMeterProvider(
 		metric.WithReader(metric.NewPeriodicReader(metricExporter)),
+		metric.WithExemplarFilter(telemetry.SampledTraceExemplarFilter),
 		metric.WithResource(container.OtelResources(version, namespace)),
 	)
 	otel.SetMeterProvider(meterProvider)

api/pkg/handlers/message_thread_handler_test.go

@@ -165,7 +165,7 @@ func (logger *messageThreadHandlerNoopLogger) WithService(_ string) telemetry.Lo
 
 func (logger *messageThreadHandlerNoopLogger) WithString(_, _ string) telemetry.Logger { return logger }
 
-func (logger *messageThreadHandlerNoopLogger) WithSpan(_ trace.SpanContext) telemetry.Logger {
+func (logger *messageThreadHandlerNoopLogger) WithSpan(_ trace.Span) telemetry.Logger {
 	return logger
 }
 func (logger *messageThreadHandlerNoopLogger) Trace(_ string)                    {}

api/pkg/listeners/read_receipts_test_helpers_test.go

@@ -12,16 +12,16 @@ import (
 
 type noopListenerLogger struct{}
 
-func (logger *noopListenerLogger) Error(error)                                 {}
-func (logger *noopListenerLogger) WithService(string) telemetry.Logger         { return logger }
-func (logger *noopListenerLogger) WithString(string, string) telemetry.Logger  { return logger }
-func (logger *noopListenerLogger) WithSpan(trace.SpanContext) telemetry.Logger { return logger }
-func (logger *noopListenerLogger) Trace(string)                                {}
-func (logger *noopListenerLogger) Info(string)                                 {}
-func (logger *noopListenerLogger) Warn(error)                                  {}
-func (logger *noopListenerLogger) Debug(string)                                {}
-func (logger *noopListenerLogger) Fatal(error)                                 {}
-func (logger *noopListenerLogger) Printf(string, ...interface{})               {}
+func (logger *noopListenerLogger) Error(error)                                {}
+func (logger *noopListenerLogger) WithService(string) telemetry.Logger        { return logger }
+func (logger *noopListenerLogger) WithString(string, string) telemetry.Logger { return logger }
+func (logger *noopListenerLogger) WithSpan(trace.Span) telemetry.Logger       { return logger }
+func (logger *noopListenerLogger) Trace(string)                               {}
+func (logger *noopListenerLogger) Info(string)                                {}
+func (logger *noopListenerLogger) Warn(error)                                 {}
+func (logger *noopListenerLogger) Debug(string)                               {}
+func (logger *noopListenerLogger) Fatal(error)                                {}
+func (logger *noopListenerLogger) Printf(string, ...interface{})              {}
 
 type listenerMessageThreadRepository struct {
 	activity repositories.MessageThreadActivityUpdate

api/pkg/repositories/gorm_heartbeat_monitor_repository.go

@@ -53,7 +53,7 @@ func (repository *gormHeartbeatMonitorRepository) UpdatePhoneOnline(ctx context.
 	ctx, cancel := context.WithTimeout(ctx, dbOperationDuration)
 	defer cancel()
 
-	err := repository.db.
+	err := repository.db.WithContext(ctx).
 		Model(&entities.HeartbeatMonitor{}).
 		Where("id = ?", monitorID).
 		Where("user_id = ?", userID).
@@ -75,7 +75,7 @@ func (repository *gormHeartbeatMonitorRepository) UpdateQueueID(ctx context.Cont
 	ctx, cancel := context.WithTimeout(ctx, dbOperationDuration)
 	defer cancel()
 
-	err := repository.db.
+	err := repository.db.WithContext(ctx).
 		Model(&entities.HeartbeatMonitor{}).
 		Where("id = ?", monitorID).
 		Updates(map[string]any{

api/pkg/repositories/gorm_heartbeat_monitor_repository_test.go

@@ -0,0 +1,47 @@
+package repositories
+
+import (
+	"context"
+	"testing"
+
+	"github.com/NdoleStudio/httpsms/pkg/telemetry"
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/require"
+	"gorm.io/driver/postgres"
+	"gorm.io/gorm"
+)
+
+func TestHeartbeatMonitorUpdatesCarryContext(t *testing.T) {
+	updates := map[string]func(HeartbeatMonitorRepository, context.Context) error{
+		"phone online": func(repository HeartbeatMonitorRepository, ctx context.Context) error {
+			return repository.UpdatePhoneOnline(ctx, "user-id", uuid.New(), true)
+		},
+		"queue ID": func(repository HeartbeatMonitorRepository, ctx context.Context) error {
+			return repository.UpdateQueueID(ctx, uuid.New(), "queue-id")
+		},
+	}
+	for name, update := range updates {
+		t.Run(name, func(t *testing.T) {
+			db, err := gorm.Open(postgres.New(postgres.Config{Conn: &messageThreadTestConnPool{}}), &gorm.Config{
+				DisableAutomaticPing:   true,
+				SkipDefaultTransaction: true,
+				DryRun:                 true,
+			})
+			require.NoError(t, err)
+			type contextKey struct{}
+			ctx := context.WithValue(context.Background(), contextKey{}, "request")
+			called := false
+			require.NoError(t, db.Callback().Update().Before("gorm:update").Register("check_context", func(db *gorm.DB) {
+				called = true
+				require.Equal(t, "request", db.Statement.Context.Value(contextKey{}))
+				_, hasDeadline := db.Statement.Context.Deadline()
+				require.True(t, hasDeadline)
+			}))
+			logger := &messageThreadTestLogger{}
+			repository := NewGormHeartbeatMonitorRepository(logger, telemetry.NewOtelLogger("test", logger), db)
+
+			require.NoError(t, update(repository, ctx))
+			require.True(t, called)
+		})
+	}
+}

api/pkg/repositories/gorm_message_thread_repository_test.go

@@ -104,13 +104,13 @@ func (logger *messageThreadTestLogger) WithService(string) telemetry.Logger { re
 
 func (logger *messageThreadTestLogger) WithString(string, string) telemetry.Logger { return logger }
 
-func (logger *messageThreadTestLogger) WithSpan(trace.SpanContext) telemetry.Logger { return logger }
-func (logger *messageThreadTestLogger) Trace(string)                                {}
-func (logger *messageThreadTestLogger) Info(string)                                 {}
-func (logger *messageThreadTestLogger) Warn(error)                                  {}
-func (logger *messageThreadTestLogger) Debug(string)                                {}
-func (logger *messageThreadTestLogger) Fatal(error)                                 {}
-func (logger *messageThreadTestLogger) Printf(string, ...interface{})               {}
+func (logger *messageThreadTestLogger) WithSpan(trace.Span) telemetry.Logger { return logger }
+func (logger *messageThreadTestLogger) Trace(string)                         {}
+func (logger *messageThreadTestLogger) Info(string)                          {}
+func (logger *messageThreadTestLogger) Warn(error)                           {}
+func (logger *messageThreadTestLogger) Debug(string)                         {}
+func (logger *messageThreadTestLogger) Fatal(error)                          {}
+func (logger *messageThreadTestLogger) Printf(string, ...interface{})        {}
 
 func TestMessageThreadStoreIncrementsUnreadCountOnConflict(t *testing.T) {
 	pool := &messageThreadTestConnPool{}

api/pkg/services/contact_service_test.go

@@ -255,7 +255,7 @@ func (l *recordingLogger) Debug(v string) {
 
 func (l *recordingLogger) WithService(_ string) telemetry.Logger   { return l }
 func (l *recordingLogger) WithString(_, _ string) telemetry.Logger { return l }
-func (l *recordingLogger) WithSpan(_ trace.SpanContext) telemetry.Logger {
+func (l *recordingLogger) WithSpan(_ trace.Span) telemetry.Logger {
 	return l
 }
 

api/pkg/services/http_notification_sender_test.go

@@ -418,6 +418,8 @@ type httpNotificationRecordingLogger struct {
 	errors []string
 }
 
+var _ telemetry.Logger = (*httpNotificationRecordingLogger)(nil)
+
 func (logger *httpNotificationRecordingLogger) Error(err error) {
 	logger.errors = append(logger.errors, err.Error())
 }
@@ -428,7 +430,7 @@ func (logger *httpNotificationRecordingLogger) WithString(string, string) teleme
 	return logger
 }
 
-func (logger *httpNotificationRecordingLogger) WithSpan(trace.SpanContext) telemetry.Logger {
+func (logger *httpNotificationRecordingLogger) WithSpan(trace.Span) telemetry.Logger {
 	return logger
 }
 

api/pkg/services/message_service_test.go

@@ -142,13 +142,13 @@ type noopLogger struct{}
 
 var _ telemetry.Logger = (*noopLogger)(nil)
 
-func (l *noopLogger) Error(_ error)                                 {}
-func (l *noopLogger) WithService(_ string) telemetry.Logger         { return l }
-func (l *noopLogger) WithString(_, _ string) telemetry.Logger       { return l }
-func (l *noopLogger) WithSpan(_ trace.SpanContext) telemetry.Logger { return l }
-func (l *noopLogger) Trace(_ string)                                {}
-func (l *noopLogger) Info(_ string)                                 {}
-func (l *noopLogger) Warn(_ error)                                  {}
-func (l *noopLogger) Debug(_ string)                                {}
-func (l *noopLogger) Fatal(_ error)                                 {}
-func (l *noopLogger) Printf(_ string, _ ...interface{})             {}
+func (l *noopLogger) Error(_ error)                           {}
+func (l *noopLogger) WithService(_ string) telemetry.Logger   { return l }
+func (l *noopLogger) WithString(_, _ string) telemetry.Logger { return l }
+func (l *noopLogger) WithSpan(_ trace.Span) telemetry.Logger  { return l }
+func (l *noopLogger) Trace(_ string)                          {}
+func (l *noopLogger) Info(_ string)                           {}
+func (l *noopLogger) Warn(_ error)                            {}
+func (l *noopLogger) Debug(_ string)                          {}
+func (l *noopLogger) Fatal(_ error)                           {}
+func (l *noopLogger) Printf(_ string, _ ...interface{})       {}

api/pkg/services/message_thread_service_contacts_test.go

@@ -61,13 +61,13 @@ func (logger *messageThreadContactLogger) WithService(string) telemetry.Logger {
 
 func (logger *messageThreadContactLogger) WithString(string, string) telemetry.Logger { return logger }
 
-func (logger *messageThreadContactLogger) WithSpan(trace.SpanContext) telemetry.Logger { return logger }
-func (logger *messageThreadContactLogger) Trace(string)                                {}
-func (logger *messageThreadContactLogger) Info(string)                                 {}
-func (logger *messageThreadContactLogger) Warn(error)                                  {}
-func (logger *messageThreadContactLogger) Debug(string)                                {}
-func (logger *messageThreadContactLogger) Fatal(error)                                 {}
-func (logger *messageThreadContactLogger) Printf(string, ...interface{})               {}
+func (logger *messageThreadContactLogger) WithSpan(trace.Span) telemetry.Logger { return logger }
+func (logger *messageThreadContactLogger) Trace(string)                         {}
+func (logger *messageThreadContactLogger) Info(string)                          {}
+func (logger *messageThreadContactLogger) Warn(error)                           {}
+func (logger *messageThreadContactLogger) Debug(string)                         {}
+func (logger *messageThreadContactLogger) Fatal(error)                          {}
+func (logger *messageThreadContactLogger) Printf(string, ...interface{})        {}
 
 func newMessageThreadContactServiceForTest(repository repositories.MessageThreadRepository, provider contactMapProvider, logger telemetry.Logger) *MessageThreadService {
 	if logger == nil {

api/pkg/services/phone_notification_service_test.go

@@ -78,9 +78,9 @@ func (logger *phoneNotificationLogger) WithString(string, string) telemetry.Logg
 	return logger
 }
 
-func (logger *phoneNotificationLogger) WithSpan(trace.SpanContext) telemetry.Logger { return logger }
-func (logger *phoneNotificationLogger) Trace(string)                                {}
-func (logger *phoneNotificationLogger) Info(string)                                 {}
+func (logger *phoneNotificationLogger) WithSpan(trace.Span) telemetry.Logger { return logger }
+func (logger *phoneNotificationLogger) Trace(string)                         {}
+func (logger *phoneNotificationLogger) Info(string)                          {}
 func (logger *phoneNotificationLogger) Warn(err error) {
 	logger.warnings = append(logger.warnings, err.Error())
 }

api/pkg/telemetry/gorm_logger.go

@@ -28,20 +28,20 @@ func (gorm *gormLogger) LogMode(_ logger.LogLevel) logger.Interface {
 }
 
 func (gorm *gormLogger) Info(ctx context.Context, s string, i ...any) {
-	gorm.logger.WithSpan(gorm.tracer.Span(ctx).SpanContext()).Info(fmt.Sprintf(s, i...))
+	gorm.logger.WithSpan(gorm.tracer.Span(ctx)).Info(fmt.Sprintf(s, i...))
 }
 
 func (gorm *gormLogger) Warn(ctx context.Context, s string, i ...any) {
-	gorm.logger.WithSpan(gorm.tracer.Span(ctx).SpanContext()).Warn(fmt.Errorf(s, i...))
+	gorm.logger.WithSpan(gorm.tracer.Span(ctx)).Warn(fmt.Errorf(s, i...))
 }
 
 func (gorm *gormLogger) Error(ctx context.Context, s string, i ...any) {
-	gorm.logger.WithSpan(gorm.tracer.Span(ctx).SpanContext()).Error(fmt.Errorf(s, i...))
+	gorm.logger.WithSpan(gorm.tracer.Span(ctx)).Error(fmt.Errorf(s, i...))
 }
 
 func (gorm *gormLogger) Trace(ctx context.Context, begin time.Time, fc func() (sql string, rowsAffected int64), err error) {
 	elapsed := time.Since(begin)
-	l := gorm.logger.WithSpan(gorm.tracer.Span(ctx).SpanContext()).WithString("latency", elapsed.String())
+	l := gorm.logger.WithSpan(gorm.tracer.Span(ctx)).WithString("latency", elapsed.String())
 	sql, rows := fc()
 	msg := fmt.Sprintf("[ROWS:%d][%s]", rows, sql)
 

api/pkg/telemetry/logger.go

@@ -15,8 +15,8 @@ type Logger interface {
 	// WithString creates a new structured logger instance with a string
 	WithString(key string, value string) Logger
 
-	// WithSpan creates a new structured logger instance for a spanContext
-	WithSpan(span trace.SpanContext) Logger
+	// WithSpan creates a new structured logger instance for a span.
+	WithSpan(span trace.Span) Logger
 
 	// Trace logs a new message with trace level.
 	Trace(value string)

api/pkg/telemetry/otel_tracer.go

@@ -34,7 +34,7 @@ func (tracer *otelTracer) StartFromFiberCtx(c fiber.Ctx, name ...string) (contex
 }
 
 func (tracer *otelTracer) CtxLogger(logger Logger, span trace.Span) Logger {
-	return logger.WithSpan(span.SpanContext())
+	return logger.WithSpan(span)
 }
 
 func (tracer *otelTracer) StartWithLogger(c context.Context, logger Logger, name ...string) (context.Context, trace.Span, Logger) {

api/pkg/telemetry/sampling_span_processor.go

@@ -0,0 +1,104 @@
+package telemetry
+
+import (
+	"context"
+	"crypto/sha256"
+	"time"
+
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/codes"
+	sdktrace "go.opentelemetry.io/otel/sdk/trace"
+	"go.opentelemetry.io/otel/trace"
+)
+
+const (
+	mustSampleAttribute       attribute.Key = "httpsms.must_sample"
+	slowDatabaseSpanThreshold               = 200 * time.Millisecond
+)
+
+type samplingSpanProcessor struct {
+	sdktrace.SpanProcessor
+}
+
+// NewSamplingSpanProcessor exports warning/error spans, database spans over 200ms,
+// and 50% of other spans. Truncated spans are retained conservatively.
+// The provider must record all spans so the decision can be made after they end.
+func NewSamplingSpanProcessor(exporter sdktrace.SpanExporter) sdktrace.SpanProcessor {
+	return &samplingSpanProcessor{
+		SpanProcessor: sdktrace.NewBatchSpanProcessor(exporter),
+	}
+}
+
+func (processor *samplingSpanProcessor) OnEnd(span sdktrace.ReadOnlySpan) {
+	if isSampledTrace(span.SpanContext().TraceID()) || shouldRetainSpan(span) {
+		processor.SpanProcessor.OnEnd(span)
+	}
+}
+
+// SampledTraceExemplarFilter links metrics only to the stable trace-ID sample.
+func SampledTraceExemplarFilter(ctx context.Context) bool {
+	spanContext := trace.SpanContextFromContext(ctx)
+	return spanContext.IsValid() && spanContext.IsSampled() && isSampledTrace(spanContext.TraceID())
+}
+
+func isSampledTrace(traceID trace.TraceID) bool {
+	// UUID version/variant bits must not bias the trace-ID decision.
+	hash := sha256.Sum256(traceID[:])
+	return hash[0] < 128
+}
+
+func shouldRetainSpan(span sdktrace.ReadOnlySpan) bool {
+	// Drop counters survive buffer limits, unlike severity attributes and events.
+	return span.DroppedEvents() > 0 ||
+		span.DroppedAttributes() > 0 ||
+		span.Status().Code == codes.Error ||
+		hasRetentionFlag(span.Attributes()) ||
+		hasExceptionalEvents(span.Events()) ||
+		hasTruncatedLinks(span) ||
+		isSlowDatabaseSpan(span)
+}
+
+func hasTruncatedLinks(span sdktrace.ReadOnlySpan) bool {
+	if span.DroppedLinks() > 0 {
+		return true
+	}
+	for _, link := range span.Links() {
+		if link.DroppedAttributeCount > 0 {
+			return true
+		}
+	}
+	return false
+}
+
+func isSlowDatabaseSpan(span sdktrace.ReadOnlySpan) bool {
+	if span.EndTime().Sub(span.StartTime()) <= slowDatabaseSpanThreshold {
+		return false
+	}
+	for _, value := range span.Attributes() {
+		switch value.Key {
+		case "db.system.name", "db.system":
+			if value.Value.AsString() != "" {
+				return true
+			}
+		}
+	}
+	return false
+}
+
+func hasExceptionalEvents(events []sdktrace.Event) bool {
+	for _, event := range events {
+		if event.Name == "exception" || event.DroppedAttributeCount > 0 {
+			return true
+		}
+	}
+	return false
+}
+
+func hasRetentionFlag(attributes []attribute.KeyValue) bool {
+	for _, value := range attributes {
+		if value.Key == mustSampleAttribute && value.Value.AsBool() {
+			return true
+		}
+	}
+	return false
+}