diff --git a/api/pkg/di/container.go b/api/pkg/di/container.go index 8d9e778d4..e7ca20d28 100644 --- a/api/pkg/di/container.go +++ b/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) diff --git a/api/pkg/handlers/message_thread_handler_test.go b/api/pkg/handlers/message_thread_handler_test.go index e545adc31..887bc8b6f 100644 --- a/api/pkg/handlers/message_thread_handler_test.go +++ b/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) {} diff --git a/api/pkg/listeners/read_receipts_test_helpers_test.go b/api/pkg/listeners/read_receipts_test_helpers_test.go index 3227685c5..7e5074f18 100644 --- a/api/pkg/listeners/read_receipts_test_helpers_test.go +++ b/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 diff --git a/api/pkg/repositories/gorm_heartbeat_monitor_repository.go b/api/pkg/repositories/gorm_heartbeat_monitor_repository.go index 50a2cefb4..85d1b2098 100644 --- a/api/pkg/repositories/gorm_heartbeat_monitor_repository.go +++ b/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{ diff --git a/api/pkg/repositories/gorm_heartbeat_monitor_repository_test.go b/api/pkg/repositories/gorm_heartbeat_monitor_repository_test.go new file mode 100644 index 000000000..51e792e85 --- /dev/null +++ b/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) + }) + } +} diff --git a/api/pkg/repositories/gorm_message_thread_repository_test.go b/api/pkg/repositories/gorm_message_thread_repository_test.go index 6f2fde4a3..99dc61b1f 100644 --- a/api/pkg/repositories/gorm_message_thread_repository_test.go +++ b/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{} diff --git a/api/pkg/services/contact_service_test.go b/api/pkg/services/contact_service_test.go index 2c5498039..0d4ef892b 100644 --- a/api/pkg/services/contact_service_test.go +++ b/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 } diff --git a/api/pkg/services/http_notification_sender_test.go b/api/pkg/services/http_notification_sender_test.go index 018765cf0..ce6d8e5a5 100644 --- a/api/pkg/services/http_notification_sender_test.go +++ b/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 } diff --git a/api/pkg/services/message_service_test.go b/api/pkg/services/message_service_test.go index be17a3a14..f84a83767 100644 --- a/api/pkg/services/message_service_test.go +++ b/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{}) {} diff --git a/api/pkg/services/message_thread_service_contacts_test.go b/api/pkg/services/message_thread_service_contacts_test.go index a941a0f7b..66a0b00ea 100644 --- a/api/pkg/services/message_thread_service_contacts_test.go +++ b/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 { diff --git a/api/pkg/services/phone_notification_service_test.go b/api/pkg/services/phone_notification_service_test.go index 9f0c42a46..2a6b5358b 100644 --- a/api/pkg/services/phone_notification_service_test.go +++ b/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()) } diff --git a/api/pkg/telemetry/gorm_logger.go b/api/pkg/telemetry/gorm_logger.go index 7f9d9517c..1b8623892 100644 --- a/api/pkg/telemetry/gorm_logger.go +++ b/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) diff --git a/api/pkg/telemetry/logger.go b/api/pkg/telemetry/logger.go index 4285a7481..aacd84b51 100644 --- a/api/pkg/telemetry/logger.go +++ b/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) diff --git a/api/pkg/telemetry/otel_tracer.go b/api/pkg/telemetry/otel_tracer.go index 34981f843..54bef2a82 100644 --- a/api/pkg/telemetry/otel_tracer.go +++ b/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) { diff --git a/api/pkg/telemetry/sampling_span_processor.go b/api/pkg/telemetry/sampling_span_processor.go new file mode 100644 index 000000000..1c683143c --- /dev/null +++ b/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 +} diff --git a/api/pkg/telemetry/sampling_span_processor_test.go b/api/pkg/telemetry/sampling_span_processor_test.go new file mode 100644 index 000000000..2ac06a85a --- /dev/null +++ b/api/pkg/telemetry/sampling_span_processor_test.go @@ -0,0 +1,323 @@ +package telemetry + +import ( + "context" + "encoding/binary" + "errors" + "fmt" + "testing" + "time" + + "github.com/hirosassa/zerodriver" + "github.com/rs/zerolog" + "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/codes" + "go.opentelemetry.io/otel/metric" + sdkmetric "go.opentelemetry.io/otel/sdk/metric" + "go.opentelemetry.io/otel/sdk/metric/metricdata" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + "go.opentelemetry.io/otel/sdk/trace/tracetest" + "go.opentelemetry.io/otel/trace" +) + +func TestTraceSamplingSupportsUUIDTraceIDs(t *testing.T) { + sampled := 0 + for index := range 1024 { + var traceID trace.TraceID + binary.BigEndian.PutUint64(traceID[8:], uint64(index)) + traceID[6] = 0x40 + traceID[8] |= 0x80 + if isSampledTrace(traceID) { + sampled++ + } + } + require.InDelta(t, 512, sampled, 64) +} + +func TestSamplingSpanProcessor(t *testing.T) { + t.Run("sampled trace", func(t *testing.T) { + test := newSamplingTestSpan(t, true) + test.finish(t, "test", "sibling") + }) + t.Run("unsampled trace", func(t *testing.T) { + test := newSamplingTestSpan(t, false) + test.finish(t) + }) + t.Run("info log does not promote span", func(t *testing.T) { + test := newSamplingTestSpan(t, false) + test.logger.Info("info") + test.finish(t) + }) +} + +func TestSamplingSpanProcessorErrors(t *testing.T) { + t.Run("error status", func(t *testing.T) { + test := newSamplingTestSpan(t, false) + test.span.SetStatus(codes.Error, "failed") + test.finish(t, "test") + }) + t.Run("exception without error status", func(t *testing.T) { + test := newSamplingTestSpan(t, false) + test.span.RecordError(errors.New("failed")) + test.finish(t, "test") + }) +} + +func TestSamplingSpanProcessorLogs(t *testing.T) { + logs := map[string]func(Logger){ + "warning": func(logger Logger) { + logger.Warn(errors.New("warning")) + }, + "decorated warning": func(logger Logger) { + logger.WithString("key", "value").WithService("test").Warn(errors.New("warning")) + }, + "error": func(logger Logger) { + logger.Error(errors.New("failed")) + }, + "nil warning": func(logger Logger) { + logger.Warn(nil) + }, + } + for name, log := range logs { + t.Run(name, func(t *testing.T) { + test := newSamplingTestSpan(t, false) + log(test.logger) + test.finish(t, "test") + exported := test.exporter.GetSpans()[0] + require.Contains(t, exported.Attributes, attribute.Bool("httpsms.must_sample", true)) + require.Empty(t, exported.Events) + }) + } +} + +func TestSamplingSpanProcessorMustSample(t *testing.T) { + t.Run("true retains span", func(t *testing.T) { + test := newSamplingTestSpan(t, false) + test.span.SetAttributes(attribute.Bool("httpsms.must_sample", true)) + test.finish(t, "test") + }) + t.Run("false does not retain span", func(t *testing.T) { + test := newSamplingTestSpan(t, false) + test.span.SetAttributes(attribute.Bool("httpsms.must_sample", false)) + test.finish(t) + }) +} + +func TestSamplingSpanProcessorGormLogs(t *testing.T) { + t.Run("warning", func(t *testing.T) { + test := newSamplingTestSpan(t, false) + NewGormLogger(test.tracer, test.logger).Warn(test.ctx, "warning %s", "message") + test.finish(t, "test") + }) + t.Run("query error", func(t *testing.T) { + test := newSamplingTestSpan(t, false) + NewGormLogger(test.tracer, test.logger).Trace(test.ctx, time.Now(), func() (string, int64) { + return "SELECT 1", 0 + }, errors.New("failed")) + test.finish(t, "test") + }) + t.Run("successful query does not promote span", func(t *testing.T) { + test := newSamplingTestSpan(t, false) + NewGormLogger(test.tracer, test.logger).Trace(test.ctx, time.Now(), func() (string, int64) { + return "SELECT 1", 1 + }, nil) + test.finish(t) + }) +} + +func TestSamplingSpanProcessorSlowDatabaseSpans(t *testing.T) { + tests := []struct { + name string + systemKey attribute.Key + duration time.Duration + want []string + }{ + {name: "slow database", systemKey: "db.system.name", duration: 201 * time.Millisecond, want: []string{"test"}}, + {name: "legacy database attribute", systemKey: "db.system", duration: time.Second, want: []string{"test"}}, + {name: "at threshold", systemKey: "db.system.name", duration: 200 * time.Millisecond}, + {name: "fast database", systemKey: "db.system.name", duration: time.Millisecond}, + {name: "slow non-database", systemKey: "service.name", duration: time.Second}, + } + for _, spec := range tests { + t.Run(spec.name, func(t *testing.T) { + start := time.Now() + test := newSamplingTestSpan(t, false, trace.WithTimestamp(start)) + test.span.SetAttributes(spec.systemKey.String("postgresql")) + test.span.End(trace.WithTimestamp(start.Add(spec.duration))) + test.finish(t, spec.want...) + }) + } +} + +func TestSamplingSpanProcessorTruncatedTelemetry(t *testing.T) { + scenarios := map[string]func(*samplingTestSpan){ + "evicted exception": func(test *samplingTestSpan) { + test.span.RecordError(errors.New("failed")) + test.fillEvents() + }, + "warning after event limit": func(test *samplingTestSpan) { + test.logger.Warn(errors.New("warning")) + test.fillEvents() + }, + "warning after attribute exhaustion and event eviction": func(test *samplingTestSpan) { + test.fillAttributes() + test.logger.Warn(errors.New("warning")) + test.fillEvents() + }, + "dropped retention flag": func(test *samplingTestSpan) { + test.fillAttributes() + test.span.SetAttributes(attribute.Bool("httpsms.must_sample", true)) + }, + } + for name, record := range scenarios { + t.Run(name, func(t *testing.T) { + test := newSamplingTestSpan(t, false) + record(test) + test.finish(t, "test") + }) + } +} + +func TestSamplingSpanProcessorTruncatedLinks(t *testing.T) { + link := trace.Link{SpanContext: trace.NewSpanContext(trace.SpanContextConfig{ + TraceID: trace.TraceID{15: 2}, + SpanID: trace.SpanID{7: 2}, + })} + t.Run("dropped links", func(t *testing.T) { + test := newSamplingTestSpan(t, false) + for range 200 { + test.span.AddLink(link) + } + test.finish(t, "test") + }) + t.Run("dropped link attributes", func(t *testing.T) { + var attributes []attribute.KeyValue + for index := range 200 { + attributes = append(attributes, attribute.Int(fmt.Sprintf("attribute.%d", index), index)) + } + link.Attributes = attributes + test := newSamplingTestSpan(t, false, trace.WithLinks(link)) + test.finish(t, "test") + }) +} + +func TestMetricExemplarsOnlyReferenceSampledTraces(t *testing.T) { + reader := sdkmetric.NewManualReader() + provider := sdkmetric.NewMeterProvider( + sdkmetric.WithReader(reader), + sdkmetric.WithExemplarFilter(SampledTraceExemplarFilter), + ) + t.Cleanup(func() { require.NoError(t, provider.Shutdown(context.Background())) }) + histogram, err := provider.Meter("test").Float64Histogram("duration") + require.NoError(t, err) + + for _, sampled := range []bool{false, true} { + traceID := trace.TraceID{6: 0x40, 8: 0x80, 15: 1} + if sampled { + traceID[15] = 2 + } + ctx := trace.ContextWithSpanContext(context.Background(), trace.NewSpanContext(trace.SpanContextConfig{ + TraceID: traceID, + SpanID: trace.SpanID{7: 1}, + TraceFlags: trace.FlagsSampled, + })) + histogram.Record(ctx, 10, metric.WithAttributes(attribute.Bool("sampled", sampled))) + } + + var result metricdata.ResourceMetrics + require.NoError(t, reader.Collect(context.Background(), &result)) + data, ok := result.ScopeMetrics[0].Metrics[0].Data.(metricdata.Histogram[float64]) + require.True(t, ok) + require.Len(t, data.DataPoints, 2) + for _, point := range data.DataPoints { + require.Equal(t, uint64(1), point.Count) + require.Equal(t, float64(10), point.Sum) + sampled, _ := point.Attributes.Value("sampled") + if sampled.AsBool() { + require.Len(t, point.Exemplars, 1) + } else { + require.Empty(t, point.Exemplars) + } + } +} + +func TestSamplingSpanProcessorTruncatedEventAttributes(t *testing.T) { + test := newSamplingTestSpan(t, false) + var attributes []attribute.KeyValue + for index := range 200 { + attributes = append(attributes, attribute.Int(fmt.Sprintf("attribute.%d", index), index)) + } + test.span.AddEvent("event", trace.WithAttributes(attributes...)) + test.finish(t, "test") +} + +type samplingTestSpan struct { + ctx context.Context + span trace.Span + logger Logger + tracer Tracer + provider *sdktrace.TracerProvider + exporter *tracetest.InMemoryExporter +} + +func newSamplingTestSpan(t *testing.T, sampled bool, options ...trace.SpanStartOption) *samplingTestSpan { + t.Helper() + exporter := tracetest.NewInMemoryExporter() + provider := sdktrace.NewTracerProvider( + sdktrace.WithSampler(sdktrace.AlwaysSample()), + sdktrace.WithSpanProcessor(NewSamplingSpanProcessor(exporter)), + ) + t.Cleanup(func() { + require.NoError(t, provider.Shutdown(context.Background())) + }) + + // Fixed UUID-shaped IDs exercise both outcomes of the stable hash decision. + traceID := trace.TraceID{6: 0x40, 8: 0x80, 15: 1} + if sampled { + traceID[15] = 2 + } + ctx := trace.ContextWithRemoteSpanContext(context.Background(), trace.NewSpanContext(trace.SpanContextConfig{ + TraceID: traceID, + SpanID: trace.SpanID{7: 1}, + })) + ctx, span := provider.Tracer("test").Start(ctx, "test", options...) + driver := zerolog.Nop() + logger := NewZerologLogger("", nil, &zerodriver.Logger{Logger: &driver}, nil) + tracer := NewOtelLogger("", logger) + return &samplingTestSpan{ + ctx: ctx, + span: span, + logger: tracer.CtxLogger(logger, span), + tracer: tracer, + provider: provider, + exporter: exporter, + } +} + +func (test *samplingTestSpan) finish(t *testing.T, want ...string) { + t.Helper() + test.span.End() + // Normal siblings share the same 50% decision, but are not promoted by errors. + _, sibling := test.provider.Tracer("test").Start(test.ctx, "sibling") + sibling.End() + require.NoError(t, test.provider.ForceFlush(context.Background())) + + var names []string + for _, span := range test.exporter.GetSpans() { + names = append(names, span.Name) + } + require.Equal(t, want, names) +} + +func (test *samplingTestSpan) fillEvents() { + for range 200 { + test.span.AddEvent("info") + } +} + +func (test *samplingTestSpan) fillAttributes() { + for index := range 200 { + test.span.SetAttributes(attribute.Int(fmt.Sprintf("attribute.%d", index), index)) + } +} diff --git a/api/pkg/telemetry/zerolog_logger.go b/api/pkg/telemetry/zerolog_logger.go index c5230bc01..a0e6c4ca0 100644 --- a/api/pkg/telemetry/zerolog_logger.go +++ b/api/pkg/telemetry/zerolog_logger.go @@ -10,20 +10,19 @@ import ( ) type zerologLogger struct { - zerolog *zerodriver.Logger - spanContext *trace.SpanContext - fields map[string]string - projectID string - level zerolog.Level + zerolog *zerodriver.Logger + span trace.Span + fields map[string]string + projectID string } // NewZerologLogger creates a new instance of the zerolog logger -func NewZerologLogger(projectID string, fields map[string]string, driver *zerodriver.Logger, span *trace.SpanContext) Logger { +func NewZerologLogger(projectID string, fields map[string]string, driver *zerodriver.Logger, span trace.Span) Logger { logger := &zerologLogger{ - zerolog: driver, - fields: fields, - projectID: projectID, - spanContext: span, + zerolog: driver, + fields: fields, + projectID: projectID, + span: span, } zerolog.SetGlobalLevel(zerolog.TraceLevel) @@ -32,12 +31,7 @@ func NewZerologLogger(projectID string, fields map[string]string, driver *zerodr // WithService creates a new structured zerolog logger instance with a service name func (logger *zerologLogger) WithService(service string) Logger { - return NewZerologLogger( - logger.projectID, - logger.addField(string(semconv.ServiceNamespaceKey), service), - logger.zerolog, - logger.spanContext, - ) + return logger.WithString(string(semconv.ServiceNamespaceKey), service) } func (logger *zerologLogger) Printf(s string, i ...interface{}) { @@ -50,7 +44,7 @@ func (logger *zerologLogger) WithString(key string, value string) Logger { logger.projectID, logger.addField(key, value), logger.zerolog, - logger.spanContext, + logger.span, ) } @@ -66,6 +60,7 @@ func (logger *zerologLogger) Trace(value string) { // Warn logs a new message with warning level. func (logger *zerologLogger) Warn(err error) { + logger.markSpanForSampling() logger.decorateEvent(logger.zerolog.Warn()).Err(err).Send() } @@ -76,29 +71,32 @@ func (logger *zerologLogger) Debug(value string) { // Fatal logs a new message with fatal level. func (logger *zerologLogger) Fatal(err error) { + logger.markSpanForSampling() logger.decorateEvent(logger.zerolog.Fatal()).Err(err).Send() } // Error logs an error func (logger *zerologLogger) Error(err error) { + logger.markSpanForSampling() logger.decorateEvent(logger.zerolog.Error()).Err(err).Send() } -// WithSpan adds a spanContext to a logger -func (logger *zerologLogger) WithSpan(spanContext trace.SpanContext) Logger { +// WithSpan adds a span to a logger. +func (logger *zerologLogger) WithSpan(span trace.Span) Logger { return NewZerologLogger( logger.projectID, logger.fields, logger.zerolog, - &spanContext, + span, ) } func (logger *zerologLogger) decorateEvent(event *zerodriver.Event) *zerolog.Event { - if logger.spanContext != nil { - event.Str("trace_id", logger.spanContext.TraceID().String()) - event.Str("span_id", logger.spanContext.SpanID().String()) - event.Bool("trace_sampled", logger.spanContext.IsSampled()) + if logger.span != nil { + spanContext := logger.span.SpanContext() + event.Str("trace_id", spanContext.TraceID().String()) + event.Str("span_id", spanContext.SpanID().String()) + event.Bool("trace_sampled", spanContext.IsSampled()) } for key, value := range logger.fields { event.Str(key, value) @@ -106,6 +104,12 @@ func (logger *zerologLogger) decorateEvent(event *zerodriver.Event) *zerolog.Eve return event.Event } +func (logger *zerologLogger) markSpanForSampling() { + if logger.span != nil { + logger.span.SetAttributes(mustSampleAttribute.Bool(true)) + } +} + func (logger *zerologLogger) addField(key string, value string) map[string]string { fields := map[string]string{} for oldKey, oldValue := range logger.fields { diff --git a/api/pkg/telemetry/zerolog_logger_test.go b/api/pkg/telemetry/zerolog_logger_test.go new file mode 100644 index 000000000..91a9324ed --- /dev/null +++ b/api/pkg/telemetry/zerolog_logger_test.go @@ -0,0 +1,96 @@ +package telemetry + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "runtime" + "testing" + + "github.com/hirosassa/zerodriver" + "github.com/rs/zerolog" + "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel/trace" +) + +func TestLoggerWithSpanRetainsWarnings(t *testing.T) { + test := newSamplingTestSpan(t, false) + var output bytes.Buffer + driver := zerolog.New(&output) + logger := NewZerologLogger("", nil, &zerodriver.Logger{Logger: &driver}, nil).WithSpan(test.span) + logger.WithString("field", "value").WithService("test").Warn(errors.New("warning")) + test.finish(t, "test") + + var record struct { + TraceID string `json:"trace_id"` + SpanID string `json:"span_id"` + Sampled bool `json:"trace_sampled"` + Field string `json:"field"` + Namespace string `json:"service.namespace"` + Error string `json:"error"` + } + require.NoError(t, json.Unmarshal(output.Bytes(), &record)) + require.Equal(t, test.span.SpanContext().TraceID().String(), record.TraceID) + require.Equal(t, test.span.SpanContext().SpanID().String(), record.SpanID) + require.True(t, record.Sampled) + require.Equal(t, "value", record.Field) + require.Equal(t, "test", record.Namespace) + require.Equal(t, "warning", record.Error) +} + +func TestZerologLoggerPreservesCaller(t *testing.T) { + for _, withSpan := range []bool{false, true} { + for _, severity := range []string{"warn", "error", "info"} { + t.Run(fmt.Sprintf("with_span=%t/%s", withSpan, severity), func(t *testing.T) { + logger, output := newCallerTestLogger(withSpan) + wantCaller := emitLogAtCaller(logger, severity) + assertCallerLog(t, output.Bytes(), wantCaller, severity) + }) + } + } +} + +func newCallerTestLogger(withSpan bool) (Logger, *bytes.Buffer) { + output := &bytes.Buffer{} + driver := zerolog.New(output).With().CallerWithSkipFrameCount(3).Logger() + logger := NewZerologLogger("", nil, &zerodriver.Logger{Logger: &driver}, nil) + if withSpan { + logger = logger.WithSpan(trace.SpanFromContext(context.Background())) + } + return logger.WithString("field", "value").WithService("test"), output +} + +func assertCallerLog(t *testing.T, output []byte, wantCaller string, severity string) { + t.Helper() + var record struct { + Caller string `json:"caller"` + Error string `json:"error"` + Message string `json:"message"` + } + require.NoError(t, json.Unmarshal(output, &record)) + require.Equal(t, wantCaller, record.Caller) + if severity == "info" { + require.Equal(t, "original message", record.Message) + return + } + require.Equal(t, "original error", record.Error) +} + +func emitLogAtCaller(logger Logger, severity string) string { + switch severity { + case "warn": + _, file, line, _ := runtime.Caller(0) + logger.Warn(errors.New("original error")) + return fmt.Sprintf("%s:%d", file, line+1) + case "error": + _, file, line, _ := runtime.Caller(0) + logger.Error(errors.New("original error")) + return fmt.Sprintf("%s:%d", file, line+1) + default: + _, file, line, _ := runtime.Caller(0) + logger.Info("original message") + return fmt.Sprintf("%s:%d", file, line+1) + } +} diff --git a/api/pkg/validators/contact_handler_validator_test.go b/api/pkg/validators/contact_handler_validator_test.go index bb145dbf1..b199df8b6 100644 --- a/api/pkg/validators/contact_handler_validator_test.go +++ b/api/pkg/validators/contact_handler_validator_test.go @@ -424,7 +424,7 @@ func (logger *contactValidatorNoopLogger) WithService(_ string) telemetry.Logger func (logger *contactValidatorNoopLogger) WithString(_, _ string) telemetry.Logger { return logger } -func (logger *contactValidatorNoopLogger) WithSpan(_ trace.SpanContext) telemetry.Logger { +func (logger *contactValidatorNoopLogger) WithSpan(_ trace.Span) telemetry.Logger { return logger } func (logger *contactValidatorNoopLogger) Trace(_ string) {}