From 6c1881cd582d171930c055f97880c7cfbfd23ef2 Mon Sep 17 00:00:00 2001 From: Acho Arnold Ewin Date: Thu, 29 Oct 2026 19:34:07 +0200 Subject: [PATCH 1/8] feat(api): retain error spans and sample remaining traces at 50% Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 60fb2b83-6e9d-4656-9f60-6183c61b716d --- api/pkg/di/container.go | 4 +- api/pkg/telemetry/gorm_logger.go | 8 +- api/pkg/telemetry/otel_tracer.go | 2 +- api/pkg/telemetry/sampling_span_processor.go | 53 ++++++ .../telemetry/sampling_span_processor_test.go | 164 ++++++++++++++++++ api/pkg/telemetry/span_logger.go | 43 +++++ 6 files changed, 267 insertions(+), 7 deletions(-) create mode 100644 api/pkg/telemetry/sampling_span_processor.go create mode 100644 api/pkg/telemetry/sampling_span_processor_test.go create mode 100644 api/pkg/telemetry/span_logger.go diff --git a/api/pkg/di/container.go b/api/pkg/di/container.go index 8d9e778d4..51c1693a4 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)), ) @@ -1970,7 +1970,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)), ) diff --git a/api/pkg/telemetry/gorm_logger.go b/api/pkg/telemetry/gorm_logger.go index 7f9d9517c..7ee37a012 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.tracer.CtxLogger(gorm.logger, 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.tracer.CtxLogger(gorm.logger, 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.tracer.CtxLogger(gorm.logger, 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.tracer.CtxLogger(gorm.logger, 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/otel_tracer.go b/api/pkg/telemetry/otel_tracer.go index 34981f843..b98a2a8f8 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 &spanLogger{Logger: logger.WithSpan(span.SpanContext()), span: 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..2c4d8a248 --- /dev/null +++ b/api/pkg/telemetry/sampling_span_processor.go @@ -0,0 +1,53 @@ +package telemetry + +import ( + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/codes" + sdktrace "go.opentelemetry.io/otel/sdk/trace" +) + +type samplingSpanProcessor struct { + sdktrace.SpanProcessor + sampler sdktrace.Sampler +} + +// NewSamplingSpanProcessor exports warning/error spans and 50% of other spans. +// 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), + sampler: sdktrace.TraceIDRatioBased(0.5), + } +} + +func (processor *samplingSpanProcessor) OnEnd(span sdktrace.ReadOnlySpan) { + if processor.sampler.ShouldSample(sdktrace.SamplingParameters{ + TraceID: span.SpanContext().TraceID(), + }).Decision == sdktrace.RecordAndSample || isWarningOrErrorSpan(span) { + processor.SpanProcessor.OnEnd(span) + } +} + +func isWarningOrErrorSpan(span sdktrace.ReadOnlySpan) bool { + if span.Status().Code == codes.Error || hasWarningOrErrorSeverity(span.Attributes()) { + return true + } + for _, event := range span.Events() { + if event.Name == "exception" || hasWarningOrErrorSeverity(event.Attributes) { + return true + } + } + return false +} + +func hasWarningOrErrorSeverity(attributes []attribute.KeyValue) bool { + for _, value := range attributes { + if value.Key == "log.severity" { + switch value.Value.AsString() { + case "warn", "warning", "error", "fatal": + 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..9f7aa88cd --- /dev/null +++ b/api/pkg/telemetry/sampling_span_processor_test.go @@ -0,0 +1,164 @@ +package telemetry + +import ( + "bytes" + "context" + "errors" + "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" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + "go.opentelemetry.io/otel/sdk/trace/tracetest" + "go.opentelemetry.io/otel/trace" +) + +func TestSamplingSpanProcessor(t *testing.T) { + tests := []struct { + name string + sampled bool + record func(context.Context, trace.Span, Logger, Tracer) + exported bool + }{ + {name: "sampled trace", sampled: true, exported: true}, + {name: "unsampled trace"}, + { + name: "error status", + record: func(_ context.Context, span trace.Span, _ Logger, _ Tracer) { + span.SetStatus(codes.Error, "failed") + }, + exported: true, + }, + { + name: "exception without error status", + record: func(_ context.Context, span trace.Span, _ Logger, _ Tracer) { + span.RecordError(errors.New("failed")) + }, + exported: true, + }, + { + name: "warning log", + record: func(_ context.Context, _ trace.Span, logger Logger, _ Tracer) { + logger.Warn(errors.New("warning")) + }, + exported: true, + }, + { + name: "decorated warning log", + record: func(_ context.Context, _ trace.Span, logger Logger, _ Tracer) { + logger.WithString("key", "value").WithService("test").Warn(errors.New("warning")) + }, + exported: true, + }, + { + name: "error log", + record: func(_ context.Context, _ trace.Span, logger Logger, _ Tracer) { + logger.Error(errors.New("failed")) + }, + exported: true, + }, + { + name: "warning after event limit", + record: func(_ context.Context, span trace.Span, logger Logger, _ Tracer) { + logger.Warn(errors.New("warning")) + for range 200 { + span.AddEvent("info") + } + }, + exported: true, + }, + { + name: "nil warning", + record: func(_ context.Context, _ trace.Span, logger Logger, _ Tracer) { + logger.Warn(nil) + }, + exported: true, + }, + { + name: "gorm warning", + record: func(ctx context.Context, _ trace.Span, logger Logger, tracer Tracer) { + NewGormLogger(tracer, logger).Warn(ctx, "warning %s", "message") + }, + exported: true, + }, + { + name: "gorm query error", + record: func(ctx context.Context, _ trace.Span, logger Logger, tracer Tracer) { + NewGormLogger(tracer, logger).Trace(ctx, time.Now(), func() (string, int64) { + return "SELECT 1", 0 + }, errors.New("failed")) + }, + exported: true, + }, + { + name: "info log", + record: func(_ context.Context, _ trace.Span, logger Logger, _ Tracer) { + logger.Info("info") + }, + }, + { + name: "warning event", + record: func(_ context.Context, span trace.Span, _ Logger, _ Tracer) { + span.AddEvent("log", trace.WithAttributes(attribute.String("log.severity", "warn"))) + }, + exported: true, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + exporter := tracetest.NewInMemoryExporter() + provider := sdktrace.NewTracerProvider( + sdktrace.WithSampler(sdktrace.AlwaysSample()), + sdktrace.WithSpanProcessor(NewSamplingSpanProcessor(exporter)), + ) + t.Cleanup(func() { + require.NoError(t, provider.Shutdown(context.Background())) + }) + + // Ratio sampling uses the lower half of the trace ID. + traceID := trace.TraceID{8: 0xff, 15: 1} + if test.sampled { + traceID[8] = 0 + } + ctx := trace.ContextWithRemoteSpanContext(context.Background(), trace.NewSpanContext(trace.SpanContextConfig{ + TraceID: traceID, + SpanID: trace.SpanID{7: 1}, + })) + ctx, span := provider.Tracer("test").Start(ctx, "test") + var output bytes.Buffer + driver := zerolog.New(&output) + logger := NewZerologLogger("", nil, &zerodriver.Logger{Logger: &driver}, nil) + tracer := NewOtelLogger("", logger) + if test.record != nil { + test.record(ctx, span, tracer.CtxLogger(logger, span), tracer) + } + span.End() + + // Normal siblings share the same 50% decision, but are not promoted by errors. + _, sibling := provider.Tracer("test").Start(ctx, "sibling") + sibling.End() + require.NoError(t, provider.ForceFlush(context.Background())) + + want := 0 + if test.exported { + want++ + } + if test.sampled { + want++ + } + spans := exporter.GetSpans() + require.Len(t, spans, want) + if test.exported { + require.Equal(t, "test", spans[0].Name) + } + if test.record != nil && test.name != "error status" && test.name != "exception without error status" && test.name != "warning event" { + require.NotEmpty(t, output.String()) + } + }) + } +} diff --git a/api/pkg/telemetry/span_logger.go b/api/pkg/telemetry/span_logger.go new file mode 100644 index 000000000..8d7c15e68 --- /dev/null +++ b/api/pkg/telemetry/span_logger.go @@ -0,0 +1,43 @@ +package telemetry + +import ( + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/trace" +) + +type spanLogger struct { + Logger + span trace.Span +} + +func (logger *spanLogger) WithService(service string) Logger { + return &spanLogger{Logger: logger.Logger.WithService(service), span: logger.span} +} + +func (logger *spanLogger) WithString(key string, value string) Logger { + return &spanLogger{Logger: logger.Logger.WithString(key, value), span: logger.span} +} + +func (logger *spanLogger) Warn(err error) { + logger.recordLog("warn", err) + logger.Logger.Warn(err) +} + +func (logger *spanLogger) Error(err error) { + logger.recordLog("error", err) + logger.Logger.Error(err) +} + +func (logger *spanLogger) Fatal(err error) { + logger.recordLog("fatal", err) + logger.Logger.Fatal(err) +} + +func (logger *spanLogger) recordLog(severity string, err error) { + attributes := []attribute.KeyValue{attribute.String("log.severity", severity)} + logger.span.SetAttributes(attributes...) + if err != nil { + attributes = append(attributes, attribute.String("log.message", err.Error())) + } + logger.span.AddEvent("log", trace.WithAttributes(attributes...)) +} From b0f0f22ce33db76b805728e4425765945b868926 Mon Sep 17 00:00:00 2001 From: Acho Arnold Ewin Date: Thu, 29 Oct 2026 20:29:25 +0200 Subject: [PATCH 2/8] fix(api): address trace sampling review feedback Retain spans with truncated telemetry and capture dedicated database slow-query warnings. Preserve original callers for span-linked logs. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 60fb2b83-6e9d-4656-9f60-6183c61b716d --- api/pkg/di/container.go | 8 +++ api/pkg/telemetry/gorm_logger.go | 44 +++++++++---- api/pkg/telemetry/gorm_logger_test.go | 65 +++++++++++++++++++ api/pkg/telemetry/sampling_span_processor.go | 9 ++- .../telemetry/sampling_span_processor_test.go | 60 ++++++++++++++++- api/pkg/telemetry/span_logger.go | 13 +++- api/pkg/telemetry/span_logger_test.go | 63 ++++++++++++++++++ api/pkg/telemetry/zerolog_logger.go | 8 +++ 8 files changed, 251 insertions(+), 19 deletions(-) create mode 100644 api/pkg/telemetry/gorm_logger_test.go create mode 100644 api/pkg/telemetry/span_logger_test.go diff --git a/api/pkg/di/container.go b/api/pkg/di/container.go index 51c1693a4..dc07a3448 100644 --- a/api/pkg/di/container.go +++ b/api/pkg/di/container.go @@ -262,6 +262,14 @@ func (container *Container) DedicatedDB() (db *gorm.DB) { config := &gorm.Config{ TranslateError: true, + Logger: telemetry.NewGormLoggerWithConfig( + container.Tracer(), + container.Logger(6), + gormLogger.Config{ + SlowThreshold: 200 * time.Millisecond, + LogLevel: gormLogger.Warn, + }, + ), } if isLocal() { config = &gorm.Config{Logger: container.GormLogger()} diff --git a/api/pkg/telemetry/gorm_logger.go b/api/pkg/telemetry/gorm_logger.go index 7ee37a012..5fbfdd01c 100644 --- a/api/pkg/telemetry/gorm_logger.go +++ b/api/pkg/telemetry/gorm_logger.go @@ -2,53 +2,75 @@ package telemetry import ( "context" + "errors" "fmt" "time" "github.com/NdoleStudio/stacktrace" + gormdb "gorm.io/gorm" "gorm.io/gorm/logger" ) type gormLogger struct { tracer Tracer logger Logger + config logger.Config } // NewGormLogger creates a new instance of gormLogger -func NewGormLogger(tracer Tracer, logger Logger) logger.Interface { +func NewGormLogger(tracer Tracer, log Logger) logger.Interface { + return NewGormLoggerWithConfig(tracer, log, logger.Config{LogLevel: logger.Info}) +} + +// NewGormLoggerWithConfig creates a span-aware logger with GORM query logging settings. +func NewGormLoggerWithConfig(tracer Tracer, log Logger, config logger.Config) logger.Interface { return &gormLogger{ tracer: tracer, - logger: logger, + logger: log, + config: config, } } // LogMode log mode -func (gorm *gormLogger) LogMode(_ logger.LogLevel) logger.Interface { - return gorm +func (gorm *gormLogger) LogMode(level logger.LogLevel) logger.Interface { + clone := *gorm + clone.config.LogLevel = level + return &clone } func (gorm *gormLogger) Info(ctx context.Context, s string, i ...any) { - gorm.tracer.CtxLogger(gorm.logger, gorm.tracer.Span(ctx)).Info(fmt.Sprintf(s, i...)) + if gorm.config.LogLevel >= logger.Info { + gorm.tracer.CtxLogger(gorm.logger, gorm.tracer.Span(ctx)).Info(fmt.Sprintf(s, i...)) + } } func (gorm *gormLogger) Warn(ctx context.Context, s string, i ...any) { - gorm.tracer.CtxLogger(gorm.logger, gorm.tracer.Span(ctx)).Warn(fmt.Errorf(s, i...)) + if gorm.config.LogLevel >= logger.Warn { + gorm.tracer.CtxLogger(gorm.logger, gorm.tracer.Span(ctx)).Warn(fmt.Errorf(s, i...)) + } } func (gorm *gormLogger) Error(ctx context.Context, s string, i ...any) { - gorm.tracer.CtxLogger(gorm.logger, gorm.tracer.Span(ctx)).Error(fmt.Errorf(s, i...)) + if gorm.config.LogLevel >= logger.Error { + gorm.tracer.CtxLogger(gorm.logger, 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) { + if gorm.config.LogLevel <= logger.Silent { + return + } elapsed := time.Since(begin) l := gorm.tracer.CtxLogger(gorm.logger, gorm.tracer.Span(ctx)).WithString("latency", elapsed.String()) sql, rows := fc() msg := fmt.Sprintf("[ROWS:%d][%s]", rows, sql) - if err != nil { + switch { + case err != nil && gorm.config.LogLevel >= logger.Error && (!errors.Is(err, gormdb.ErrRecordNotFound) || !gorm.config.IgnoreRecordNotFoundError): l.Error(stacktrace.Propagatef(err, "[ROWS:%d][%s]", rows, sql)) - return + case gorm.config.SlowThreshold > 0 && elapsed > gorm.config.SlowThreshold && gorm.config.LogLevel >= logger.Warn: + l.Warn(stacktrace.NewErrorf("SLOW SQL >= %s [ROWS:%d][%s]", gorm.config.SlowThreshold, rows, sql)) + case gorm.config.LogLevel >= logger.Info: + l.Debug(msg) } - - l.Debug(msg) } diff --git a/api/pkg/telemetry/gorm_logger_test.go b/api/pkg/telemetry/gorm_logger_test.go new file mode 100644 index 000000000..d2e7cdae9 --- /dev/null +++ b/api/pkg/telemetry/gorm_logger_test.go @@ -0,0 +1,65 @@ +package telemetry + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "testing" + "time" + + "github.com/hirosassa/zerodriver" + "github.com/rs/zerolog" + "github.com/stretchr/testify/require" + "gorm.io/gorm" + gormlogger "gorm.io/gorm/logger" +) + +func TestGormWarningLogger(t *testing.T) { + tests := []struct { + name string + elapsed time.Duration + err error + logLevel gormlogger.LogLevel + want string + }{ + {name: "fast query", elapsed: time.Millisecond, logLevel: gormlogger.Warn}, + {name: "slow query", elapsed: time.Second, logLevel: gormlogger.Warn, want: "warn"}, + {name: "query error", elapsed: time.Millisecond, err: errors.New("query failed"), logLevel: gormlogger.Warn, want: "error"}, + {name: "record not found", elapsed: time.Millisecond, err: gorm.ErrRecordNotFound, logLevel: gormlogger.Warn, want: "error"}, + {name: "silent query error", elapsed: time.Second, err: errors.New("query failed"), logLevel: gormlogger.Silent}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + var output bytes.Buffer + driver := zerolog.New(&output) + logger := NewZerologLogger("", nil, &zerodriver.Logger{Logger: &driver}, nil) + tracer := NewOtelLogger("", logger) + queryLogger := NewGormLoggerWithConfig(tracer, logger, gormlogger.Config{ + SlowThreshold: 200 * time.Millisecond, + LogLevel: gormlogger.Warn, + }).LogMode(test.logLevel) + queryLogger.Trace(context.Background(), time.Now().Add(-test.elapsed), func() (string, int64) { + return "SELECT 1", 1 + }, test.err) + + if test.want == "" { + require.Empty(t, output.String()) + return + } + var record struct { + Level string `json:"level"` + Error string `json:"error"` + } + require.NoError(t, json.Unmarshal(output.Bytes(), &record)) + require.Equal(t, test.want, record.Level) + require.Contains(t, record.Error, "SELECT 1") + if test.err != nil { + require.Contains(t, record.Error, test.err.Error()) + } else { + require.Contains(t, record.Error, "SLOW SQL") + } + }) + } +} diff --git a/api/pkg/telemetry/sampling_span_processor.go b/api/pkg/telemetry/sampling_span_processor.go index 2c4d8a248..ade244fae 100644 --- a/api/pkg/telemetry/sampling_span_processor.go +++ b/api/pkg/telemetry/sampling_span_processor.go @@ -12,6 +12,7 @@ type samplingSpanProcessor struct { } // NewSamplingSpanProcessor exports warning/error spans and 50% of other spans. +// Spans with dropped telemetry are retained because their severity may be lost. // The provider must record all spans so the decision can be made after they end. func NewSamplingSpanProcessor(exporter sdktrace.SpanExporter) sdktrace.SpanProcessor { return &samplingSpanProcessor{ @@ -23,12 +24,16 @@ func NewSamplingSpanProcessor(exporter sdktrace.SpanExporter) sdktrace.SpanProce func (processor *samplingSpanProcessor) OnEnd(span sdktrace.ReadOnlySpan) { if processor.sampler.ShouldSample(sdktrace.SamplingParameters{ TraceID: span.SpanContext().TraceID(), - }).Decision == sdktrace.RecordAndSample || isWarningOrErrorSpan(span) { + }).Decision == sdktrace.RecordAndSample || shouldRetainSpan(span) { processor.SpanProcessor.OnEnd(span) } } -func isWarningOrErrorSpan(span sdktrace.ReadOnlySpan) bool { +func shouldRetainSpan(span sdktrace.ReadOnlySpan) bool { + // Drop counters survive buffer limits, unlike severity attributes and events. + if span.DroppedEvents() > 0 || span.DroppedAttributes() > 0 { + return true + } if span.Status().Code == codes.Error || hasWarningOrErrorSeverity(span.Attributes()) { return true } diff --git a/api/pkg/telemetry/sampling_span_processor_test.go b/api/pkg/telemetry/sampling_span_processor_test.go index 9f7aa88cd..eee837059 100644 --- a/api/pkg/telemetry/sampling_span_processor_test.go +++ b/api/pkg/telemetry/sampling_span_processor_test.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "errors" + "fmt" "testing" "time" @@ -15,6 +16,7 @@ import ( sdktrace "go.opentelemetry.io/otel/sdk/trace" "go.opentelemetry.io/otel/sdk/trace/tracetest" "go.opentelemetry.io/otel/trace" + gormlogger "gorm.io/gorm/logger" ) func TestSamplingSpanProcessor(t *testing.T) { @@ -40,6 +42,16 @@ func TestSamplingSpanProcessor(t *testing.T) { }, exported: true, }, + { + name: "evicted exception", + record: func(_ context.Context, span trace.Span, _ Logger, _ Tracer) { + span.RecordError(errors.New("failed")) + for range 200 { + span.AddEvent("info") + } + }, + exported: true, + }, { name: "warning log", record: func(_ context.Context, _ trace.Span, logger Logger, _ Tracer) { @@ -71,6 +83,19 @@ func TestSamplingSpanProcessor(t *testing.T) { }, exported: true, }, + { + name: "warning after attribute exhaustion and event eviction", + record: func(_ context.Context, span trace.Span, logger Logger, _ Tracer) { + for index := range 200 { + span.SetAttributes(attribute.Int(fmt.Sprintf("attribute.%d", index), index)) + } + logger.Warn(errors.New("warning")) + for range 200 { + span.AddEvent("info") + } + }, + exported: true, + }, { name: "nil warning", record: func(_ context.Context, _ trace.Span, logger Logger, _ Tracer) { @@ -94,6 +119,18 @@ func TestSamplingSpanProcessor(t *testing.T) { }, exported: true, }, + { + name: "gorm slow query warning", + record: func(ctx context.Context, _ trace.Span, logger Logger, tracer Tracer) { + NewGormLoggerWithConfig(tracer, logger, gormlogger.Config{ + SlowThreshold: 200 * time.Millisecond, + LogLevel: gormlogger.Warn, + }).Trace(ctx, time.Now().Add(-time.Second), func() (string, int64) { + return "SELECT 1", 1 + }, nil) + }, + exported: true, + }, { name: "info log", record: func(_ context.Context, _ trace.Span, logger Logger, _ Tracer) { @@ -107,6 +144,26 @@ func TestSamplingSpanProcessor(t *testing.T) { }, exported: true, }, + { + name: "evicted warning event", + record: func(_ context.Context, span trace.Span, _ Logger, _ Tracer) { + span.AddEvent("log", trace.WithAttributes(attribute.String("log.severity", "warn"))) + for range 200 { + span.AddEvent("info") + } + }, + exported: true, + }, + { + name: "dropped warning attribute", + record: func(_ context.Context, span trace.Span, _ Logger, _ Tracer) { + for index := range 200 { + span.SetAttributes(attribute.Int(fmt.Sprintf("attribute.%d", index), index)) + } + span.SetAttributes(attribute.String("log.severity", "warn")) + }, + exported: true, + }, } for _, test := range tests { @@ -156,9 +213,6 @@ func TestSamplingSpanProcessor(t *testing.T) { if test.exported { require.Equal(t, "test", spans[0].Name) } - if test.record != nil && test.name != "error status" && test.name != "exception without error status" && test.name != "warning event" { - require.NotEmpty(t, output.String()) - } }) } } diff --git a/api/pkg/telemetry/span_logger.go b/api/pkg/telemetry/span_logger.go index 8d7c15e68..4750f7ccd 100644 --- a/api/pkg/telemetry/span_logger.go +++ b/api/pkg/telemetry/span_logger.go @@ -20,17 +20,24 @@ func (logger *spanLogger) WithString(key string, value string) Logger { func (logger *spanLogger) Warn(err error) { logger.recordLog("warn", err) - logger.Logger.Warn(err) + logger.callerLogger().Warn(err) } func (logger *spanLogger) Error(err error) { logger.recordLog("error", err) - logger.Logger.Error(err) + logger.callerLogger().Error(err) } func (logger *spanLogger) Fatal(err error) { logger.recordLog("fatal", err) - logger.Logger.Fatal(err) + logger.callerLogger().Fatal(err) +} + +func (logger *spanLogger) callerLogger() Logger { + if callerLogger, ok := logger.Logger.(interface{ withCallerSkipFrame(int) Logger }); ok { + return callerLogger.withCallerSkipFrame(1) + } + return logger.Logger } func (logger *spanLogger) recordLog(severity string, err error) { diff --git a/api/pkg/telemetry/span_logger_test.go b/api/pkg/telemetry/span_logger_test.go new file mode 100644 index 000000000..2945187cd --- /dev/null +++ b/api/pkg/telemetry/span_logger_test.go @@ -0,0 +1,63 @@ +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 TestSpanLoggerPreservesCaller(t *testing.T) { + for _, wrapped := range []bool{false, true} { + for _, severity := range []string{"warn", "error", "info"} { + t.Run(fmt.Sprintf("wrapped=%t/%s", wrapped, severity), func(t *testing.T) { + var output bytes.Buffer + driver := zerolog.New(&output).With().CallerWithSkipFrameCount(3).Logger() + logger := NewZerologLogger("", nil, &zerodriver.Logger{Logger: &driver}, nil) + if wrapped { + logger = NewOtelLogger("", logger).CtxLogger(logger, trace.SpanFromContext(context.Background())) + } + logger = logger.WithString("field", "value").WithService("test") + + wantCaller := emitLogAtCaller(logger, severity) + var record struct { + Caller string `json:"caller"` + Error string `json:"error"` + Message string `json:"message"` + } + require.NoError(t, json.Unmarshal(output.Bytes(), &record)) + require.Equal(t, wantCaller, record.Caller) + if severity == "info" { + require.Equal(t, "original message", record.Message) + } else { + 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/telemetry/zerolog_logger.go b/api/pkg/telemetry/zerolog_logger.go index c5230bc01..0dd9995b3 100644 --- a/api/pkg/telemetry/zerolog_logger.go +++ b/api/pkg/telemetry/zerolog_logger.go @@ -15,6 +15,7 @@ type zerologLogger struct { fields map[string]string projectID string level zerolog.Level + callerSkip int } // NewZerologLogger creates a new instance of the zerolog logger @@ -95,6 +96,7 @@ func (logger *zerologLogger) WithSpan(spanContext trace.SpanContext) Logger { } func (logger *zerologLogger) decorateEvent(event *zerodriver.Event) *zerolog.Event { + event.CallerSkipFrame(logger.callerSkip) if logger.spanContext != nil { event.Str("trace_id", logger.spanContext.TraceID().String()) event.Str("span_id", logger.spanContext.SpanID().String()) @@ -106,6 +108,12 @@ func (logger *zerologLogger) decorateEvent(event *zerodriver.Event) *zerolog.Eve return event.Event } +func (logger *zerologLogger) withCallerSkipFrame(skip int) Logger { + clone := *logger + clone.callerSkip += skip + return &clone +} + func (logger *zerologLogger) addField(key string, value string) map[string]string { fields := map[string]string{} for oldKey, oldValue := range logger.fields { From 46a404cb4f303aed31bcce427ea2753aa78c7c23 Mon Sep 17 00:00:00 2001 From: Acho Arnold Ewin Date: Thu, 29 Oct 2026 20:46:21 +0200 Subject: [PATCH 3/8] refactor(api): mark spans for sampling without copying logs Keep standalone log output intact and simplify telemetry conditions and test fixtures. Propagate heartbeat update contexts and defer SQL formatting until a log is emitted. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 60fb2b83-6e9d-4656-9f60-6183c61b716d --- api/pkg/di/container.go | 4 +- .../gorm_heartbeat_monitor_repository.go | 4 +- .../gorm_heartbeat_monitor_repository_test.go | 47 +++ api/pkg/telemetry/gorm_logger.go | 96 ++++- api/pkg/telemetry/gorm_logger_test.go | 110 ++++- api/pkg/telemetry/sampling_span_processor.go | 43 +- .../telemetry/sampling_span_processor_test.go | 383 +++++++++--------- api/pkg/telemetry/span_logger.go | 16 +- api/pkg/telemetry/span_logger_test.go | 48 ++- 9 files changed, 492 insertions(+), 259 deletions(-) create mode 100644 api/pkg/repositories/gorm_heartbeat_monitor_repository_test.go diff --git a/api/pkg/di/container.go b/api/pkg/di/container.go index dc07a3448..2e3e614b8 100644 --- a/api/pkg/di/container.go +++ b/api/pkg/di/container.go @@ -262,9 +262,9 @@ func (container *Container) DedicatedDB() (db *gorm.DB) { config := &gorm.Config{ TranslateError: true, - Logger: telemetry.NewGormLoggerWithConfig( + Logger: telemetry.NewGormSpanLogger( container.Tracer(), - container.Logger(6), + gormLogger.Default, gormLogger.Config{ SlowThreshold: 200 * time.Millisecond, LogLevel: gormLogger.Warn, 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/telemetry/gorm_logger.go b/api/pkg/telemetry/gorm_logger.go index 5fbfdd01c..28f57b261 100644 --- a/api/pkg/telemetry/gorm_logger.go +++ b/api/pkg/telemetry/gorm_logger.go @@ -14,7 +14,11 @@ import ( type gormLogger struct { tracer Tracer logger Logger - config logger.Config + config gormLogConfig +} + +type gormLogConfig struct { + logger.Config } // NewGormLogger creates a new instance of gormLogger @@ -27,7 +31,7 @@ func NewGormLoggerWithConfig(tracer Tracer, log Logger, config logger.Config) lo return &gormLogger{ tracer: tracer, logger: log, - config: config, + config: gormLogConfig{Config: config}, } } @@ -57,20 +61,96 @@ func (gorm *gormLogger) Error(ctx context.Context, s string, i ...any) { } func (gorm *gormLogger) Trace(ctx context.Context, begin time.Time, fc func() (sql string, rowsAffected int64), err error) { - if gorm.config.LogLevel <= logger.Silent { + elapsed := time.Since(begin) + level := gorm.config.queryLevel(elapsed, err) + if level == logger.Silent { return } - elapsed := time.Since(begin) l := gorm.tracer.CtxLogger(gorm.logger, gorm.tracer.Span(ctx)).WithString("latency", elapsed.String()) sql, rows := fc() msg := fmt.Sprintf("[ROWS:%d][%s]", rows, sql) - switch { - case err != nil && gorm.config.LogLevel >= logger.Error && (!errors.Is(err, gormdb.ErrRecordNotFound) || !gorm.config.IgnoreRecordNotFoundError): + switch level { + case logger.Error: l.Error(stacktrace.Propagatef(err, "[ROWS:%d][%s]", rows, sql)) - case gorm.config.SlowThreshold > 0 && elapsed > gorm.config.SlowThreshold && gorm.config.LogLevel >= logger.Warn: + case logger.Warn: l.Warn(stacktrace.NewErrorf("SLOW SQL >= %s [ROWS:%d][%s]", gorm.config.SlowThreshold, rows, sql)) - case gorm.config.LogLevel >= logger.Info: + default: l.Debug(msg) } } + +func (config gormLogConfig) queryLevel(elapsed time.Duration, err error) logger.LogLevel { + if config.LogLevel <= logger.Silent { + return logger.Silent + } + switch { + case config.shouldLogError(err): + return logger.Error + case config.shouldLogSlowQuery(elapsed): + return logger.Warn + case config.LogLevel >= logger.Info: + return logger.Info + default: + return logger.Silent + } +} + +func (config gormLogConfig) shouldLogError(err error) bool { + if err == nil || config.LogLevel < logger.Error { + return false + } + return !config.IgnoreRecordNotFoundError || !errors.Is(err, gormdb.ErrRecordNotFound) +} + +func (config gormLogConfig) shouldLogSlowQuery(elapsed time.Duration) bool { + return config.SlowThreshold > 0 && + elapsed > config.SlowThreshold && + config.LogLevel >= logger.Warn +} + +type gormSpanLogger struct { + logger.Interface + tracer Tracer + config gormLogConfig +} + +// NewGormSpanLogger marks warning/error spans without changing the existing logger's output. +func NewGormSpanLogger(tracer Tracer, delegate logger.Interface, config logger.Config) logger.Interface { + return &gormSpanLogger{ + Interface: delegate, + tracer: tracer, + config: gormLogConfig{Config: config}, + } +} + +func (gorm *gormSpanLogger) LogMode(level logger.LogLevel) logger.Interface { + clone := *gorm + clone.Interface = gorm.Interface.LogMode(level) + clone.config.LogLevel = level + return &clone +} + +func (gorm *gormSpanLogger) Warn(ctx context.Context, message string, args ...any) { + if gorm.config.LogLevel >= logger.Warn { + gorm.tracer.Span(ctx).SetAttributes(mustSampleAttribute.Bool(true)) + } + gorm.Interface.Warn(ctx, message, args...) +} + +func (gorm *gormSpanLogger) Error(ctx context.Context, message string, args ...any) { + if gorm.config.LogLevel >= logger.Error { + gorm.tracer.Span(ctx).SetAttributes(mustSampleAttribute.Bool(true)) + } + gorm.Interface.Error(ctx, message, args...) +} + +func (gorm *gormSpanLogger) Trace(ctx context.Context, begin time.Time, fc func() (string, int64), err error) { + gorm.Interface.Trace(ctx, begin, func() (string, int64) { + switch gorm.config.queryLevel(time.Since(begin), err) { + case logger.Error, logger.Warn: + gorm.tracer.Span(ctx).SetAttributes(mustSampleAttribute.Bool(true)) + } + return fc() + }, err) +} diff --git a/api/pkg/telemetry/gorm_logger_test.go b/api/pkg/telemetry/gorm_logger_test.go index d2e7cdae9..980998247 100644 --- a/api/pkg/telemetry/gorm_logger_test.go +++ b/api/pkg/telemetry/gorm_logger_test.go @@ -5,6 +5,7 @@ import ( "context" "encoding/json" "errors" + stdlog "log" "testing" "time" @@ -40,26 +41,103 @@ func TestGormWarningLogger(t *testing.T) { SlowThreshold: 200 * time.Millisecond, LogLevel: gormlogger.Warn, }).LogMode(test.logLevel) + formatted := false queryLogger.Trace(context.Background(), time.Now().Add(-test.elapsed), func() (string, int64) { + formatted = true return "SELECT 1", 1 }, test.err) - if test.want == "" { - require.Empty(t, output.String()) - return - } - var record struct { - Level string `json:"level"` - Error string `json:"error"` - } - require.NoError(t, json.Unmarshal(output.Bytes(), &record)) - require.Equal(t, test.want, record.Level) - require.Contains(t, record.Error, "SELECT 1") - if test.err != nil { - require.Contains(t, record.Error, test.err.Error()) - } else { - require.Contains(t, record.Error, "SLOW SQL") - } + require.Equal(t, test.want != "", formatted) + assertGormQueryLog(t, output.Bytes(), test.want, test.err) }) } } + +func assertGormQueryLog(t *testing.T, output []byte, wantLevel string, err error) { + t.Helper() + if wantLevel == "" { + require.Empty(t, output) + return + } + var record struct { + Level string `json:"level"` + Error string `json:"error"` + } + require.NoError(t, json.Unmarshal(output, &record)) + require.Equal(t, wantLevel, record.Level) + require.Contains(t, record.Error, "SELECT 1") + if err != nil { + require.Contains(t, record.Error, err.Error()) + return + } + require.Contains(t, record.Error, "SLOW SQL") +} + +func TestGormSpanLoggerPreservesDelegate(t *testing.T) { + tests := []struct { + name string + elapsed time.Duration + err error + level gormlogger.LogLevel + message string + spans []string + }{ + {name: "fast query", elapsed: time.Millisecond, level: gormlogger.Warn}, + {name: "slow query", elapsed: time.Second, level: gormlogger.Warn, message: "SLOW SQL", spans: []string{"test"}}, + {name: "query error", err: errors.New("original query error"), level: gormlogger.Warn, message: "original query error", spans: []string{"test"}}, + {name: "info query", elapsed: time.Millisecond, level: gormlogger.Info, message: "SELECT 1"}, + {name: "silent error", err: errors.New("query error"), level: gormlogger.Silent}, + } + for _, spec := range tests { + t.Run(spec.name, func(t *testing.T) { + test := newSamplingTestSpan(t, false) + queryLogger, output := newDelegatedGormLogger(test) + formatted := false + queryLogger.LogMode(spec.level).Trace(test.ctx, time.Now().Add(-spec.elapsed), func() (string, int64) { + formatted = true + return "SELECT 1", 1 + }, spec.err) + + test.finish(t, spec.spans...) + require.Equal(t, spec.message != "", formatted) + assertDelegatedGormOutput(t, output.String(), spec.message) + }) + } +} + +func TestGormSpanLoggerDirectWarningsAndErrors(t *testing.T) { + logs := map[string]func(gormlogger.Interface, context.Context){ + "warning": func(logger gormlogger.Interface, ctx context.Context) { + logger.Warn(ctx, "original %s", "message") + }, + "error": func(logger gormlogger.Interface, ctx context.Context) { + logger.Error(ctx, "original %s", "message") + }, + } + for name, log := range logs { + t.Run(name, func(t *testing.T) { + test := newSamplingTestSpan(t, false) + logger, output := newDelegatedGormLogger(test) + log(logger, test.ctx) + test.finish(t, "test") + require.Contains(t, output.String(), "original message") + }) + } +} + +func newDelegatedGormLogger(test *samplingTestSpan) (gormlogger.Interface, *bytes.Buffer) { + output := &bytes.Buffer{} + config := gormlogger.Config{SlowThreshold: 200 * time.Millisecond, LogLevel: gormlogger.Warn} + delegate := gormlogger.New(stdlog.New(output, "", 0), config) + return NewGormSpanLogger(test.tracer, delegate, config), output +} + +func assertDelegatedGormOutput(t *testing.T, output string, message string) { + t.Helper() + if message == "" { + require.Empty(t, output) + return + } + require.Contains(t, output, message) + require.Contains(t, output, "SELECT 1") +} diff --git a/api/pkg/telemetry/sampling_span_processor.go b/api/pkg/telemetry/sampling_span_processor.go index ade244fae..3fda6b3ac 100644 --- a/api/pkg/telemetry/sampling_span_processor.go +++ b/api/pkg/telemetry/sampling_span_processor.go @@ -6,6 +6,8 @@ import ( sdktrace "go.opentelemetry.io/otel/sdk/trace" ) +const mustSampleAttribute attribute.Key = "httpsms.must_sample" + type samplingSpanProcessor struct { sdktrace.SpanProcessor sampler sdktrace.Sampler @@ -31,28 +33,41 @@ func (processor *samplingSpanProcessor) OnEnd(span sdktrace.ReadOnlySpan) { func shouldRetainSpan(span sdktrace.ReadOnlySpan) bool { // Drop counters survive buffer limits, unlike severity attributes and events. - if span.DroppedEvents() > 0 || span.DroppedAttributes() > 0 { - return true - } - if span.Status().Code == codes.Error || hasWarningOrErrorSeverity(span.Attributes()) { - return true - } - for _, event := range span.Events() { - if event.Name == "exception" || hasWarningOrErrorSeverity(event.Attributes) { + return span.DroppedEvents() > 0 || + span.DroppedAttributes() > 0 || + span.Status().Code == codes.Error || + hasRetentionAttributes(span.Attributes()) || + hasExceptionalEvents(span.Events()) +} + +func hasExceptionalEvents(events []sdktrace.Event) bool { + for _, event := range events { + if event.Name == "exception" || event.DroppedAttributeCount > 0 || hasRetentionAttributes(event.Attributes) { return true } } return false } -func hasWarningOrErrorSeverity(attributes []attribute.KeyValue) bool { +func hasRetentionAttributes(attributes []attribute.KeyValue) bool { for _, value := range attributes { - if value.Key == "log.severity" { - switch value.Value.AsString() { - case "warn", "warning", "error", "fatal": - return true - } + if isRetentionAttribute(value) { + return true } } return false } + +func isRetentionAttribute(value attribute.KeyValue) bool { + if value.Key == mustSampleAttribute { + return value.Value.AsBool() + } + if value.Key != "log.severity" { + return false + } + switch value.Value.AsString() { + case "warn", "warning", "error", "fatal": + return true + } + return false +} diff --git a/api/pkg/telemetry/sampling_span_processor_test.go b/api/pkg/telemetry/sampling_span_processor_test.go index eee837059..1a6c76d96 100644 --- a/api/pkg/telemetry/sampling_span_processor_test.go +++ b/api/pkg/telemetry/sampling_span_processor_test.go @@ -1,7 +1,6 @@ package telemetry import ( - "bytes" "context" "errors" "fmt" @@ -20,199 +19,215 @@ import ( ) func TestSamplingSpanProcessor(t *testing.T) { - tests := []struct { - name string - sampled bool - record func(context.Context, trace.Span, Logger, Tracer) - exported bool - }{ - {name: "sampled trace", sampled: true, exported: true}, - {name: "unsampled trace"}, - { - name: "error status", - record: func(_ context.Context, span trace.Span, _ Logger, _ Tracer) { - span.SetStatus(codes.Error, "failed") - }, - exported: true, - }, - { - name: "exception without error status", - record: func(_ context.Context, span trace.Span, _ Logger, _ Tracer) { - span.RecordError(errors.New("failed")) - }, - exported: true, - }, - { - name: "evicted exception", - record: func(_ context.Context, span trace.Span, _ Logger, _ Tracer) { - span.RecordError(errors.New("failed")) - for range 200 { - span.AddEvent("info") - } - }, - exported: true, - }, - { - name: "warning log", - record: func(_ context.Context, _ trace.Span, logger Logger, _ Tracer) { - logger.Warn(errors.New("warning")) - }, - exported: true, - }, - { - name: "decorated warning log", - record: func(_ context.Context, _ trace.Span, logger Logger, _ Tracer) { - logger.WithString("key", "value").WithService("test").Warn(errors.New("warning")) - }, - exported: true, - }, - { - name: "error log", - record: func(_ context.Context, _ trace.Span, logger Logger, _ Tracer) { - logger.Error(errors.New("failed")) - }, - exported: true, - }, - { - name: "warning after event limit", - record: func(_ context.Context, span trace.Span, logger Logger, _ Tracer) { - logger.Warn(errors.New("warning")) - for range 200 { - span.AddEvent("info") - } - }, - exported: true, - }, - { - name: "warning after attribute exhaustion and event eviction", - record: func(_ context.Context, span trace.Span, logger Logger, _ Tracer) { - for index := range 200 { - span.SetAttributes(attribute.Int(fmt.Sprintf("attribute.%d", index), index)) - } - logger.Warn(errors.New("warning")) - for range 200 { - span.AddEvent("info") - } - }, - exported: true, + 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")) }, - { - name: "nil warning", - record: func(_ context.Context, _ trace.Span, logger Logger, _ Tracer) { - logger.Warn(nil) - }, - exported: true, + "decorated warning": func(logger Logger) { + logger.WithString("key", "value").WithService("test").Warn(errors.New("warning")) }, - { - name: "gorm warning", - record: func(ctx context.Context, _ trace.Span, logger Logger, tracer Tracer) { - NewGormLogger(tracer, logger).Warn(ctx, "warning %s", "message") - }, - exported: true, + "error": func(logger Logger) { + logger.Error(errors.New("failed")) }, - { - name: "gorm query error", - record: func(ctx context.Context, _ trace.Span, logger Logger, tracer Tracer) { - NewGormLogger(tracer, logger).Trace(ctx, time.Now(), func() (string, int64) { - return "SELECT 1", 0 - }, errors.New("failed")) - }, - exported: true, + "nil warning": func(logger Logger) { + logger.Warn(nil) }, - { - name: "gorm slow query warning", - record: func(ctx context.Context, _ trace.Span, logger Logger, tracer Tracer) { - NewGormLoggerWithConfig(tracer, logger, gormlogger.Config{ - SlowThreshold: 200 * time.Millisecond, - LogLevel: gormlogger.Warn, - }).Trace(ctx, time.Now().Add(-time.Second), func() (string, int64) { - return "SELECT 1", 1 - }, nil) - }, - exported: true, + } + 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("slow query warning", func(t *testing.T) { + test := newSamplingTestSpan(t, false) + NewGormLoggerWithConfig(test.tracer, test.logger, gormlogger.Config{ + SlowThreshold: 200 * time.Millisecond, + LogLevel: gormlogger.Warn, + }).Trace(test.ctx, time.Now().Add(-time.Second), func() (string, int64) { + return "SELECT 1", 1 + }, nil) + test.finish(t, "test") + }) +} + +func TestSamplingSpanProcessorWarningEvent(t *testing.T) { + test := newSamplingTestSpan(t, false) + test.span.AddEvent("log", trace.WithAttributes(attribute.String("log.severity", "warn"))) + test.finish(t, "test") +} + +func TestSamplingSpanProcessorTruncatedTelemetry(t *testing.T) { + scenarios := map[string]func(*samplingTestSpan){ + "evicted exception": func(test *samplingTestSpan) { + test.span.RecordError(errors.New("failed")) + test.fillEvents() }, - { - name: "info log", - record: func(_ context.Context, _ trace.Span, logger Logger, _ Tracer) { - logger.Info("info") - }, + "evicted warning event": func(test *samplingTestSpan) { + test.span.AddEvent("log", trace.WithAttributes(attribute.String("log.severity", "warn"))) + test.fillEvents() }, - { - name: "warning event", - record: func(_ context.Context, span trace.Span, _ Logger, _ Tracer) { - span.AddEvent("log", trace.WithAttributes(attribute.String("log.severity", "warn"))) - }, - exported: true, + "warning after event limit": func(test *samplingTestSpan) { + test.logger.Warn(errors.New("warning")) + test.fillEvents() }, - { - name: "evicted warning event", - record: func(_ context.Context, span trace.Span, _ Logger, _ Tracer) { - span.AddEvent("log", trace.WithAttributes(attribute.String("log.severity", "warn"))) - for range 200 { - span.AddEvent("info") - } - }, - exported: true, + "warning after attribute exhaustion and event eviction": func(test *samplingTestSpan) { + test.fillAttributes() + test.logger.Warn(errors.New("warning")) + test.fillEvents() }, - { - name: "dropped warning attribute", - record: func(_ context.Context, span trace.Span, _ Logger, _ Tracer) { - for index := range 200 { - span.SetAttributes(attribute.Int(fmt.Sprintf("attribute.%d", index), index)) - } - span.SetAttributes(attribute.String("log.severity", "warn")) - }, - exported: true, + "dropped warning attribute": func(test *samplingTestSpan) { + test.fillAttributes() + test.span.SetAttributes(attribute.String("log.severity", "warn")) }, } - - for _, test := range tests { - t.Run(test.name, func(t *testing.T) { - exporter := tracetest.NewInMemoryExporter() - provider := sdktrace.NewTracerProvider( - sdktrace.WithSampler(sdktrace.AlwaysSample()), - sdktrace.WithSpanProcessor(NewSamplingSpanProcessor(exporter)), - ) - t.Cleanup(func() { - require.NoError(t, provider.Shutdown(context.Background())) - }) - - // Ratio sampling uses the lower half of the trace ID. - traceID := trace.TraceID{8: 0xff, 15: 1} - if test.sampled { - traceID[8] = 0 - } - ctx := trace.ContextWithRemoteSpanContext(context.Background(), trace.NewSpanContext(trace.SpanContextConfig{ - TraceID: traceID, - SpanID: trace.SpanID{7: 1}, - })) - ctx, span := provider.Tracer("test").Start(ctx, "test") - var output bytes.Buffer - driver := zerolog.New(&output) - logger := NewZerologLogger("", nil, &zerodriver.Logger{Logger: &driver}, nil) - tracer := NewOtelLogger("", logger) - if test.record != nil { - test.record(ctx, span, tracer.CtxLogger(logger, span), tracer) - } - span.End() - - // Normal siblings share the same 50% decision, but are not promoted by errors. - _, sibling := provider.Tracer("test").Start(ctx, "sibling") - sibling.End() - require.NoError(t, provider.ForceFlush(context.Background())) - - want := 0 - if test.exported { - want++ - } - if test.sampled { - want++ - } - spans := exporter.GetSpans() - require.Len(t, spans, want) - if test.exported { - require.Equal(t, "test", spans[0].Name) - } + for name, record := range scenarios { + t.Run(name, func(t *testing.T) { + test := newSamplingTestSpan(t, false) + record(test) + test.finish(t, "test") }) } } + +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)) + } + attributes = append(attributes, attribute.String("log.severity", "warn")) + test.span.AddEvent("log", 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) *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())) + }) + + // Ratio sampling uses the lower half of the trace ID. + traceID := trace.TraceID{8: 0xff, 15: 1} + if sampled { + traceID[8] = 0 + } + ctx := trace.ContextWithRemoteSpanContext(context.Background(), trace.NewSpanContext(trace.SpanContextConfig{ + TraceID: traceID, + SpanID: trace.SpanID{7: 1}, + })) + ctx, span := provider.Tracer("test").Start(ctx, "test") + 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/span_logger.go b/api/pkg/telemetry/span_logger.go index 4750f7ccd..ff548a888 100644 --- a/api/pkg/telemetry/span_logger.go +++ b/api/pkg/telemetry/span_logger.go @@ -1,7 +1,6 @@ package telemetry import ( - "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/trace" ) @@ -19,17 +18,17 @@ func (logger *spanLogger) WithString(key string, value string) Logger { } func (logger *spanLogger) Warn(err error) { - logger.recordLog("warn", err) + logger.span.SetAttributes(mustSampleAttribute.Bool(true)) logger.callerLogger().Warn(err) } func (logger *spanLogger) Error(err error) { - logger.recordLog("error", err) + logger.span.SetAttributes(mustSampleAttribute.Bool(true)) logger.callerLogger().Error(err) } func (logger *spanLogger) Fatal(err error) { - logger.recordLog("fatal", err) + logger.span.SetAttributes(mustSampleAttribute.Bool(true)) logger.callerLogger().Fatal(err) } @@ -39,12 +38,3 @@ func (logger *spanLogger) callerLogger() Logger { } return logger.Logger } - -func (logger *spanLogger) recordLog(severity string, err error) { - attributes := []attribute.KeyValue{attribute.String("log.severity", severity)} - logger.span.SetAttributes(attributes...) - if err != nil { - attributes = append(attributes, attribute.String("log.message", err.Error())) - } - logger.span.AddEvent("log", trace.WithAttributes(attributes...)) -} diff --git a/api/pkg/telemetry/span_logger_test.go b/api/pkg/telemetry/span_logger_test.go index 2945187cd..6dd90105d 100644 --- a/api/pkg/telemetry/span_logger_test.go +++ b/api/pkg/telemetry/span_logger_test.go @@ -19,32 +19,40 @@ func TestSpanLoggerPreservesCaller(t *testing.T) { for _, wrapped := range []bool{false, true} { for _, severity := range []string{"warn", "error", "info"} { t.Run(fmt.Sprintf("wrapped=%t/%s", wrapped, severity), func(t *testing.T) { - var output bytes.Buffer - driver := zerolog.New(&output).With().CallerWithSkipFrameCount(3).Logger() - logger := NewZerologLogger("", nil, &zerodriver.Logger{Logger: &driver}, nil) - if wrapped { - logger = NewOtelLogger("", logger).CtxLogger(logger, trace.SpanFromContext(context.Background())) - } - logger = logger.WithString("field", "value").WithService("test") - + logger, output := newCallerTestLogger(wrapped) wantCaller := emitLogAtCaller(logger, severity) - var record struct { - Caller string `json:"caller"` - Error string `json:"error"` - Message string `json:"message"` - } - require.NoError(t, json.Unmarshal(output.Bytes(), &record)) - require.Equal(t, wantCaller, record.Caller) - if severity == "info" { - require.Equal(t, "original message", record.Message) - } else { - require.Equal(t, "original error", record.Error) - } + assertCallerLog(t, output.Bytes(), wantCaller, severity) }) } } } +func newCallerTestLogger(wrapped bool) (Logger, *bytes.Buffer) { + output := &bytes.Buffer{} + driver := zerolog.New(output).With().CallerWithSkipFrameCount(3).Logger() + logger := NewZerologLogger("", nil, &zerodriver.Logger{Logger: &driver}, nil) + if wrapped { + logger = NewOtelLogger("", logger).CtxLogger(logger, 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": From 1d2aaf94fd6d9a0cc48348a0d9c282df7bfc9652 Mon Sep 17 00:00:00 2001 From: Acho Arnold Ewin Date: Thu, 29 Oct 2026 20:59:25 +0200 Subject: [PATCH 4/8] fix(api): preserve GORM callers and honor severity variants Keep original query callers in standard GORM log output and honor opt-in parameter filtering. Retain warning and error events regardless of severity capitalization. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 60fb2b83-6e9d-4656-9f60-6183c61b716d --- api/pkg/di/container.go | 3 +- api/pkg/telemetry/gorm_logger.go | 77 ++++++++++++++----- api/pkg/telemetry/gorm_logger_test.go | 75 +++++++++++++++++- api/pkg/telemetry/sampling_span_processor.go | 4 +- .../telemetry/sampling_span_processor_test.go | 10 +++ 5 files changed, 145 insertions(+), 24 deletions(-) diff --git a/api/pkg/di/container.go b/api/pkg/di/container.go index 2e3e614b8..3e0cd398f 100644 --- a/api/pkg/di/container.go +++ b/api/pkg/di/container.go @@ -264,10 +264,11 @@ func (container *Container) DedicatedDB() (db *gorm.DB) { TranslateError: true, Logger: telemetry.NewGormSpanLogger( container.Tracer(), - gormLogger.Default, + log.New(os.Stdout, "\r\n", log.LstdFlags), gormLogger.Config{ SlowThreshold: 200 * time.Millisecond, LogLevel: gormLogger.Warn, + Colorful: true, }, ), } diff --git a/api/pkg/telemetry/gorm_logger.go b/api/pkg/telemetry/gorm_logger.go index 28f57b261..585b7079b 100644 --- a/api/pkg/telemetry/gorm_logger.go +++ b/api/pkg/telemetry/gorm_logger.go @@ -4,6 +4,8 @@ import ( "context" "errors" "fmt" + "runtime" + "strings" "time" "github.com/NdoleStudio/stacktrace" @@ -12,9 +14,9 @@ import ( ) type gormLogger struct { + gormLogConfig tracer Tracer logger Logger - config gormLogConfig } type gormLogConfig struct { @@ -29,40 +31,40 @@ func NewGormLogger(tracer Tracer, log Logger) logger.Interface { // NewGormLoggerWithConfig creates a span-aware logger with GORM query logging settings. func NewGormLoggerWithConfig(tracer Tracer, log Logger, config logger.Config) logger.Interface { return &gormLogger{ - tracer: tracer, - logger: log, - config: gormLogConfig{Config: config}, + gormLogConfig: gormLogConfig{Config: config}, + tracer: tracer, + logger: log, } } // LogMode log mode func (gorm *gormLogger) LogMode(level logger.LogLevel) logger.Interface { clone := *gorm - clone.config.LogLevel = level + clone.LogLevel = level return &clone } func (gorm *gormLogger) Info(ctx context.Context, s string, i ...any) { - if gorm.config.LogLevel >= logger.Info { + if gorm.LogLevel >= logger.Info { gorm.tracer.CtxLogger(gorm.logger, gorm.tracer.Span(ctx)).Info(fmt.Sprintf(s, i...)) } } func (gorm *gormLogger) Warn(ctx context.Context, s string, i ...any) { - if gorm.config.LogLevel >= logger.Warn { + if gorm.LogLevel >= logger.Warn { gorm.tracer.CtxLogger(gorm.logger, gorm.tracer.Span(ctx)).Warn(fmt.Errorf(s, i...)) } } func (gorm *gormLogger) Error(ctx context.Context, s string, i ...any) { - if gorm.config.LogLevel >= logger.Error { + if gorm.LogLevel >= logger.Error { gorm.tracer.CtxLogger(gorm.logger, 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) - level := gorm.config.queryLevel(elapsed, err) + level := gorm.queryLevel(elapsed, err) if level == logger.Silent { return } @@ -74,7 +76,7 @@ func (gorm *gormLogger) Trace(ctx context.Context, begin time.Time, fc func() (s case logger.Error: l.Error(stacktrace.Propagatef(err, "[ROWS:%d][%s]", rows, sql)) case logger.Warn: - l.Warn(stacktrace.NewErrorf("SLOW SQL >= %s [ROWS:%d][%s]", gorm.config.SlowThreshold, rows, sql)) + l.Warn(stacktrace.NewErrorf("SLOW SQL >= %s [ROWS:%d][%s]", gorm.SlowThreshold, rows, sql)) default: l.Debug(msg) } @@ -109,37 +111,44 @@ func (config gormLogConfig) shouldLogSlowQuery(elapsed time.Duration) bool { config.LogLevel >= logger.Warn } +func (config gormLogConfig) ParamsFilter(_ context.Context, sql string, params ...interface{}) (string, []interface{}) { + if config.ParameterizedQueries { + return sql, nil + } + return sql, params +} + type gormSpanLogger struct { logger.Interface + gormLogConfig tracer Tracer - config gormLogConfig } -// NewGormSpanLogger marks warning/error spans without changing the existing logger's output. -func NewGormSpanLogger(tracer Tracer, delegate logger.Interface, config logger.Config) logger.Interface { +// NewGormSpanLogger marks warning/error spans while keeping GORM's standard output format. +func NewGormSpanLogger(tracer Tracer, writer logger.Writer, config logger.Config) logger.Interface { return &gormSpanLogger{ - Interface: delegate, - tracer: tracer, - config: gormLogConfig{Config: config}, + Interface: logger.New(&gormSourceWriter{Writer: writer}, config), + gormLogConfig: gormLogConfig{Config: config}, + tracer: tracer, } } func (gorm *gormSpanLogger) LogMode(level logger.LogLevel) logger.Interface { clone := *gorm clone.Interface = gorm.Interface.LogMode(level) - clone.config.LogLevel = level + clone.LogLevel = level return &clone } func (gorm *gormSpanLogger) Warn(ctx context.Context, message string, args ...any) { - if gorm.config.LogLevel >= logger.Warn { + if gorm.LogLevel >= logger.Warn { gorm.tracer.Span(ctx).SetAttributes(mustSampleAttribute.Bool(true)) } gorm.Interface.Warn(ctx, message, args...) } func (gorm *gormSpanLogger) Error(ctx context.Context, message string, args ...any) { - if gorm.config.LogLevel >= logger.Error { + if gorm.LogLevel >= logger.Error { gorm.tracer.Span(ctx).SetAttributes(mustSampleAttribute.Bool(true)) } gorm.Interface.Error(ctx, message, args...) @@ -147,10 +156,38 @@ func (gorm *gormSpanLogger) Error(ctx context.Context, message string, args ...a func (gorm *gormSpanLogger) Trace(ctx context.Context, begin time.Time, fc func() (string, int64), err error) { gorm.Interface.Trace(ctx, begin, func() (string, int64) { - switch gorm.config.queryLevel(time.Since(begin), err) { + switch gorm.queryLevel(time.Since(begin), err) { case logger.Error, logger.Warn: gorm.tracer.Span(ctx).SetAttributes(mustSampleAttribute.Bool(true)) } return fc() }, err) } + +type gormSourceWriter struct { + logger.Writer +} + +func (writer *gormSourceWriter) Printf(format string, args ...interface{}) { + if len(args) > 0 { + if source, ok := args[0].(string); ok { + args[0] = gormQueryCaller(source) + } + } + writer.Writer.Printf(format, args...) +} + +func gormQueryCaller(fallback string) string { + _, adapterFile, _, _ := runtime.Caller(0) + var callers [32]uintptr + frames := runtime.CallersFrames(callers[:runtime.Callers(2, callers[:])]) + for { + frame, more := frames.Next() + if frame.File != adapterFile && !strings.HasPrefix(frame.Function, "gorm.io/gorm") { + return fmt.Sprintf("%s:%d", frame.File, frame.Line) + } + if !more { + return fallback + } + } +} diff --git a/api/pkg/telemetry/gorm_logger_test.go b/api/pkg/telemetry/gorm_logger_test.go index 980998247..c31a9231d 100644 --- a/api/pkg/telemetry/gorm_logger_test.go +++ b/api/pkg/telemetry/gorm_logger_test.go @@ -5,7 +5,9 @@ import ( "context" "encoding/json" "errors" + "fmt" stdlog "log" + "runtime" "testing" "time" @@ -128,8 +130,7 @@ func TestGormSpanLoggerDirectWarningsAndErrors(t *testing.T) { func newDelegatedGormLogger(test *samplingTestSpan) (gormlogger.Interface, *bytes.Buffer) { output := &bytes.Buffer{} config := gormlogger.Config{SlowThreshold: 200 * time.Millisecond, LogLevel: gormlogger.Warn} - delegate := gormlogger.New(stdlog.New(output, "", 0), config) - return NewGormSpanLogger(test.tracer, delegate, config), output + return NewGormSpanLogger(test.tracer, stdlog.New(output, "", 0), config), output } func assertDelegatedGormOutput(t *testing.T, output string, message string) { @@ -141,3 +142,73 @@ func assertDelegatedGormOutput(t *testing.T, output string, message string) { require.Contains(t, output, message) require.Contains(t, output, "SELECT 1") } + +func TestGormSpanLoggerPreservesCaller(t *testing.T) { + for _, severity := range []string{"warn", "error", "slow query", "query error"} { + t.Run(severity, func(t *testing.T) { + test := newSamplingTestSpan(t, false) + logger, output := newDelegatedGormLogger(test) + caller := emitGormLogAtCaller(logger, test.ctx, severity) + require.Contains(t, output.String(), caller) + test.finish(t, "test") + }) + } +} + +func emitGormLogAtCaller(logger gormlogger.Interface, ctx context.Context, severity string) string { + query := func() (string, int64) { return "SELECT 1", 1 } + switch severity { + case "warn": + _, file, line, _ := runtime.Caller(0) + logger.Warn(ctx, "warning") + return fmt.Sprintf("%s:%d", file, line+1) + case "error": + _, file, line, _ := runtime.Caller(0) + logger.Error(ctx, "error") + return fmt.Sprintf("%s:%d", file, line+1) + case "slow query": + _, file, line, _ := runtime.Caller(0) + logger.Trace(ctx, time.Now().Add(-time.Second), query, nil) + return fmt.Sprintf("%s:%d", file, line+1) + default: + _, file, line, _ := runtime.Caller(0) + logger.Trace(ctx, time.Now(), query, errors.New("query error")) + return fmt.Sprintf("%s:%d", file, line+1) + } +} + +func TestGormLoggerParameterizedQueries(t *testing.T) { + for _, delegated := range []bool{false, true} { + for _, parameterized := range []bool{false, true} { + t.Run(fmt.Sprintf("delegated=%t/parameterized=%t", delegated, parameterized), func(t *testing.T) { + test := newSamplingTestSpan(t, false) + config := gormlogger.Config{ParameterizedQueries: parameterized, LogLevel: gormlogger.Warn} + logger := newParameterFilterLogger(test, config, delegated) + filter, ok := logger.(interface { + ParamsFilter(context.Context, string, ...interface{}) (string, []interface{}) + }) + require.True(t, ok) + sql, params := filter.ParamsFilter(test.ctx, "SELECT ?", "example-value") + require.Equal(t, "SELECT ?", sql) + assertGormQueryParams(t, params, parameterized) + test.finish(t) + }) + } + } +} + +func newParameterFilterLogger(test *samplingTestSpan, config gormlogger.Config, delegated bool) gormlogger.Interface { + if delegated { + return NewGormSpanLogger(test.tracer, stdlog.New(&bytes.Buffer{}, "", 0), config) + } + return NewGormLoggerWithConfig(test.tracer, test.logger, config) +} + +func assertGormQueryParams(t *testing.T, params []interface{}, parameterized bool) { + t.Helper() + if parameterized { + require.Empty(t, params) + return + } + require.Equal(t, []interface{}{"example-value"}, params) +} diff --git a/api/pkg/telemetry/sampling_span_processor.go b/api/pkg/telemetry/sampling_span_processor.go index 3fda6b3ac..958c13d21 100644 --- a/api/pkg/telemetry/sampling_span_processor.go +++ b/api/pkg/telemetry/sampling_span_processor.go @@ -1,6 +1,8 @@ package telemetry import ( + "strings" + "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/codes" sdktrace "go.opentelemetry.io/otel/sdk/trace" @@ -65,7 +67,7 @@ func isRetentionAttribute(value attribute.KeyValue) bool { if value.Key != "log.severity" { return false } - switch value.Value.AsString() { + switch strings.ToLower(value.Value.AsString()) { case "warn", "warning", "error", "fatal": return true } diff --git a/api/pkg/telemetry/sampling_span_processor_test.go b/api/pkg/telemetry/sampling_span_processor_test.go index 1a6c76d96..3201019a4 100644 --- a/api/pkg/telemetry/sampling_span_processor_test.go +++ b/api/pkg/telemetry/sampling_span_processor_test.go @@ -118,6 +118,16 @@ func TestSamplingSpanProcessorWarningEvent(t *testing.T) { test.finish(t, "test") } +func TestSamplingSpanProcessorUppercaseSeverity(t *testing.T) { + for _, severity := range []string{"WARN", "WARNING", "ERROR", "FATAL"} { + t.Run(severity, func(t *testing.T) { + test := newSamplingTestSpan(t, false) + test.span.AddEvent("log", trace.WithAttributes(attribute.String("log.severity", severity))) + test.finish(t, "test") + }) + } +} + func TestSamplingSpanProcessorTruncatedTelemetry(t *testing.T) { scenarios := map[string]func(*samplingTestSpan){ "evicted exception": func(test *samplingTestSpan) { From f5ce3d202cf88ff51ea322478c966466842a894b Mon Sep 17 00:00:00 2001 From: Acho Arnold Ewin Date: Fri, 30 Oct 2026 01:34:47 +0200 Subject: [PATCH 5/8] refactor(api): restore original GORM logging Remove the extra GORM logger, configuration layer, parameter filters, and caller repair. Retain database spans over 200ms in the export processor without changing logging. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 60fb2b83-6e9d-4656-9f60-6183c61b716d --- api/pkg/di/container.go | 9 - api/pkg/telemetry/gorm_logger.go | 161 +------------ api/pkg/telemetry/gorm_logger_test.go | 214 ------------------ api/pkg/telemetry/sampling_span_processor.go | 28 ++- .../telemetry/sampling_span_processor_test.go | 38 +++- 5 files changed, 64 insertions(+), 386 deletions(-) delete mode 100644 api/pkg/telemetry/gorm_logger_test.go diff --git a/api/pkg/di/container.go b/api/pkg/di/container.go index 3e0cd398f..51c1693a4 100644 --- a/api/pkg/di/container.go +++ b/api/pkg/di/container.go @@ -262,15 +262,6 @@ func (container *Container) DedicatedDB() (db *gorm.DB) { config := &gorm.Config{ TranslateError: true, - Logger: telemetry.NewGormSpanLogger( - container.Tracer(), - log.New(os.Stdout, "\r\n", log.LstdFlags), - gormLogger.Config{ - SlowThreshold: 200 * time.Millisecond, - LogLevel: gormLogger.Warn, - Colorful: true, - }, - ), } if isLocal() { config = &gorm.Config{Logger: container.GormLogger()} diff --git a/api/pkg/telemetry/gorm_logger.go b/api/pkg/telemetry/gorm_logger.go index 585b7079b..7ee37a012 100644 --- a/api/pkg/telemetry/gorm_logger.go +++ b/api/pkg/telemetry/gorm_logger.go @@ -2,192 +2,53 @@ package telemetry import ( "context" - "errors" "fmt" - "runtime" - "strings" "time" "github.com/NdoleStudio/stacktrace" - gormdb "gorm.io/gorm" "gorm.io/gorm/logger" ) type gormLogger struct { - gormLogConfig tracer Tracer logger Logger } -type gormLogConfig struct { - logger.Config -} - // NewGormLogger creates a new instance of gormLogger -func NewGormLogger(tracer Tracer, log Logger) logger.Interface { - return NewGormLoggerWithConfig(tracer, log, logger.Config{LogLevel: logger.Info}) -} - -// NewGormLoggerWithConfig creates a span-aware logger with GORM query logging settings. -func NewGormLoggerWithConfig(tracer Tracer, log Logger, config logger.Config) logger.Interface { +func NewGormLogger(tracer Tracer, logger Logger) logger.Interface { return &gormLogger{ - gormLogConfig: gormLogConfig{Config: config}, - tracer: tracer, - logger: log, + tracer: tracer, + logger: logger, } } // LogMode log mode -func (gorm *gormLogger) LogMode(level logger.LogLevel) logger.Interface { - clone := *gorm - clone.LogLevel = level - return &clone +func (gorm *gormLogger) LogMode(_ logger.LogLevel) logger.Interface { + return gorm } func (gorm *gormLogger) Info(ctx context.Context, s string, i ...any) { - if gorm.LogLevel >= logger.Info { - gorm.tracer.CtxLogger(gorm.logger, gorm.tracer.Span(ctx)).Info(fmt.Sprintf(s, i...)) - } + gorm.tracer.CtxLogger(gorm.logger, gorm.tracer.Span(ctx)).Info(fmt.Sprintf(s, i...)) } func (gorm *gormLogger) Warn(ctx context.Context, s string, i ...any) { - if gorm.LogLevel >= logger.Warn { - gorm.tracer.CtxLogger(gorm.logger, gorm.tracer.Span(ctx)).Warn(fmt.Errorf(s, i...)) - } + gorm.tracer.CtxLogger(gorm.logger, gorm.tracer.Span(ctx)).Warn(fmt.Errorf(s, i...)) } func (gorm *gormLogger) Error(ctx context.Context, s string, i ...any) { - if gorm.LogLevel >= logger.Error { - gorm.tracer.CtxLogger(gorm.logger, gorm.tracer.Span(ctx)).Error(fmt.Errorf(s, i...)) - } + gorm.tracer.CtxLogger(gorm.logger, 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) - level := gorm.queryLevel(elapsed, err) - if level == logger.Silent { - return - } l := gorm.tracer.CtxLogger(gorm.logger, gorm.tracer.Span(ctx)).WithString("latency", elapsed.String()) sql, rows := fc() msg := fmt.Sprintf("[ROWS:%d][%s]", rows, sql) - switch level { - case logger.Error: + if err != nil { l.Error(stacktrace.Propagatef(err, "[ROWS:%d][%s]", rows, sql)) - case logger.Warn: - l.Warn(stacktrace.NewErrorf("SLOW SQL >= %s [ROWS:%d][%s]", gorm.SlowThreshold, rows, sql)) - default: - l.Debug(msg) - } -} - -func (config gormLogConfig) queryLevel(elapsed time.Duration, err error) logger.LogLevel { - if config.LogLevel <= logger.Silent { - return logger.Silent - } - switch { - case config.shouldLogError(err): - return logger.Error - case config.shouldLogSlowQuery(elapsed): - return logger.Warn - case config.LogLevel >= logger.Info: - return logger.Info - default: - return logger.Silent - } -} - -func (config gormLogConfig) shouldLogError(err error) bool { - if err == nil || config.LogLevel < logger.Error { - return false - } - return !config.IgnoreRecordNotFoundError || !errors.Is(err, gormdb.ErrRecordNotFound) -} - -func (config gormLogConfig) shouldLogSlowQuery(elapsed time.Duration) bool { - return config.SlowThreshold > 0 && - elapsed > config.SlowThreshold && - config.LogLevel >= logger.Warn -} - -func (config gormLogConfig) ParamsFilter(_ context.Context, sql string, params ...interface{}) (string, []interface{}) { - if config.ParameterizedQueries { - return sql, nil - } - return sql, params -} - -type gormSpanLogger struct { - logger.Interface - gormLogConfig - tracer Tracer -} - -// NewGormSpanLogger marks warning/error spans while keeping GORM's standard output format. -func NewGormSpanLogger(tracer Tracer, writer logger.Writer, config logger.Config) logger.Interface { - return &gormSpanLogger{ - Interface: logger.New(&gormSourceWriter{Writer: writer}, config), - gormLogConfig: gormLogConfig{Config: config}, - tracer: tracer, - } -} - -func (gorm *gormSpanLogger) LogMode(level logger.LogLevel) logger.Interface { - clone := *gorm - clone.Interface = gorm.Interface.LogMode(level) - clone.LogLevel = level - return &clone -} - -func (gorm *gormSpanLogger) Warn(ctx context.Context, message string, args ...any) { - if gorm.LogLevel >= logger.Warn { - gorm.tracer.Span(ctx).SetAttributes(mustSampleAttribute.Bool(true)) - } - gorm.Interface.Warn(ctx, message, args...) -} - -func (gorm *gormSpanLogger) Error(ctx context.Context, message string, args ...any) { - if gorm.LogLevel >= logger.Error { - gorm.tracer.Span(ctx).SetAttributes(mustSampleAttribute.Bool(true)) - } - gorm.Interface.Error(ctx, message, args...) -} - -func (gorm *gormSpanLogger) Trace(ctx context.Context, begin time.Time, fc func() (string, int64), err error) { - gorm.Interface.Trace(ctx, begin, func() (string, int64) { - switch gorm.queryLevel(time.Since(begin), err) { - case logger.Error, logger.Warn: - gorm.tracer.Span(ctx).SetAttributes(mustSampleAttribute.Bool(true)) - } - return fc() - }, err) -} - -type gormSourceWriter struct { - logger.Writer -} - -func (writer *gormSourceWriter) Printf(format string, args ...interface{}) { - if len(args) > 0 { - if source, ok := args[0].(string); ok { - args[0] = gormQueryCaller(source) - } + return } - writer.Writer.Printf(format, args...) -} -func gormQueryCaller(fallback string) string { - _, adapterFile, _, _ := runtime.Caller(0) - var callers [32]uintptr - frames := runtime.CallersFrames(callers[:runtime.Callers(2, callers[:])]) - for { - frame, more := frames.Next() - if frame.File != adapterFile && !strings.HasPrefix(frame.Function, "gorm.io/gorm") { - return fmt.Sprintf("%s:%d", frame.File, frame.Line) - } - if !more { - return fallback - } - } + l.Debug(msg) } diff --git a/api/pkg/telemetry/gorm_logger_test.go b/api/pkg/telemetry/gorm_logger_test.go deleted file mode 100644 index c31a9231d..000000000 --- a/api/pkg/telemetry/gorm_logger_test.go +++ /dev/null @@ -1,214 +0,0 @@ -package telemetry - -import ( - "bytes" - "context" - "encoding/json" - "errors" - "fmt" - stdlog "log" - "runtime" - "testing" - "time" - - "github.com/hirosassa/zerodriver" - "github.com/rs/zerolog" - "github.com/stretchr/testify/require" - "gorm.io/gorm" - gormlogger "gorm.io/gorm/logger" -) - -func TestGormWarningLogger(t *testing.T) { - tests := []struct { - name string - elapsed time.Duration - err error - logLevel gormlogger.LogLevel - want string - }{ - {name: "fast query", elapsed: time.Millisecond, logLevel: gormlogger.Warn}, - {name: "slow query", elapsed: time.Second, logLevel: gormlogger.Warn, want: "warn"}, - {name: "query error", elapsed: time.Millisecond, err: errors.New("query failed"), logLevel: gormlogger.Warn, want: "error"}, - {name: "record not found", elapsed: time.Millisecond, err: gorm.ErrRecordNotFound, logLevel: gormlogger.Warn, want: "error"}, - {name: "silent query error", elapsed: time.Second, err: errors.New("query failed"), logLevel: gormlogger.Silent}, - } - - for _, test := range tests { - t.Run(test.name, func(t *testing.T) { - var output bytes.Buffer - driver := zerolog.New(&output) - logger := NewZerologLogger("", nil, &zerodriver.Logger{Logger: &driver}, nil) - tracer := NewOtelLogger("", logger) - queryLogger := NewGormLoggerWithConfig(tracer, logger, gormlogger.Config{ - SlowThreshold: 200 * time.Millisecond, - LogLevel: gormlogger.Warn, - }).LogMode(test.logLevel) - formatted := false - queryLogger.Trace(context.Background(), time.Now().Add(-test.elapsed), func() (string, int64) { - formatted = true - return "SELECT 1", 1 - }, test.err) - - require.Equal(t, test.want != "", formatted) - assertGormQueryLog(t, output.Bytes(), test.want, test.err) - }) - } -} - -func assertGormQueryLog(t *testing.T, output []byte, wantLevel string, err error) { - t.Helper() - if wantLevel == "" { - require.Empty(t, output) - return - } - var record struct { - Level string `json:"level"` - Error string `json:"error"` - } - require.NoError(t, json.Unmarshal(output, &record)) - require.Equal(t, wantLevel, record.Level) - require.Contains(t, record.Error, "SELECT 1") - if err != nil { - require.Contains(t, record.Error, err.Error()) - return - } - require.Contains(t, record.Error, "SLOW SQL") -} - -func TestGormSpanLoggerPreservesDelegate(t *testing.T) { - tests := []struct { - name string - elapsed time.Duration - err error - level gormlogger.LogLevel - message string - spans []string - }{ - {name: "fast query", elapsed: time.Millisecond, level: gormlogger.Warn}, - {name: "slow query", elapsed: time.Second, level: gormlogger.Warn, message: "SLOW SQL", spans: []string{"test"}}, - {name: "query error", err: errors.New("original query error"), level: gormlogger.Warn, message: "original query error", spans: []string{"test"}}, - {name: "info query", elapsed: time.Millisecond, level: gormlogger.Info, message: "SELECT 1"}, - {name: "silent error", err: errors.New("query error"), level: gormlogger.Silent}, - } - for _, spec := range tests { - t.Run(spec.name, func(t *testing.T) { - test := newSamplingTestSpan(t, false) - queryLogger, output := newDelegatedGormLogger(test) - formatted := false - queryLogger.LogMode(spec.level).Trace(test.ctx, time.Now().Add(-spec.elapsed), func() (string, int64) { - formatted = true - return "SELECT 1", 1 - }, spec.err) - - test.finish(t, spec.spans...) - require.Equal(t, spec.message != "", formatted) - assertDelegatedGormOutput(t, output.String(), spec.message) - }) - } -} - -func TestGormSpanLoggerDirectWarningsAndErrors(t *testing.T) { - logs := map[string]func(gormlogger.Interface, context.Context){ - "warning": func(logger gormlogger.Interface, ctx context.Context) { - logger.Warn(ctx, "original %s", "message") - }, - "error": func(logger gormlogger.Interface, ctx context.Context) { - logger.Error(ctx, "original %s", "message") - }, - } - for name, log := range logs { - t.Run(name, func(t *testing.T) { - test := newSamplingTestSpan(t, false) - logger, output := newDelegatedGormLogger(test) - log(logger, test.ctx) - test.finish(t, "test") - require.Contains(t, output.String(), "original message") - }) - } -} - -func newDelegatedGormLogger(test *samplingTestSpan) (gormlogger.Interface, *bytes.Buffer) { - output := &bytes.Buffer{} - config := gormlogger.Config{SlowThreshold: 200 * time.Millisecond, LogLevel: gormlogger.Warn} - return NewGormSpanLogger(test.tracer, stdlog.New(output, "", 0), config), output -} - -func assertDelegatedGormOutput(t *testing.T, output string, message string) { - t.Helper() - if message == "" { - require.Empty(t, output) - return - } - require.Contains(t, output, message) - require.Contains(t, output, "SELECT 1") -} - -func TestGormSpanLoggerPreservesCaller(t *testing.T) { - for _, severity := range []string{"warn", "error", "slow query", "query error"} { - t.Run(severity, func(t *testing.T) { - test := newSamplingTestSpan(t, false) - logger, output := newDelegatedGormLogger(test) - caller := emitGormLogAtCaller(logger, test.ctx, severity) - require.Contains(t, output.String(), caller) - test.finish(t, "test") - }) - } -} - -func emitGormLogAtCaller(logger gormlogger.Interface, ctx context.Context, severity string) string { - query := func() (string, int64) { return "SELECT 1", 1 } - switch severity { - case "warn": - _, file, line, _ := runtime.Caller(0) - logger.Warn(ctx, "warning") - return fmt.Sprintf("%s:%d", file, line+1) - case "error": - _, file, line, _ := runtime.Caller(0) - logger.Error(ctx, "error") - return fmt.Sprintf("%s:%d", file, line+1) - case "slow query": - _, file, line, _ := runtime.Caller(0) - logger.Trace(ctx, time.Now().Add(-time.Second), query, nil) - return fmt.Sprintf("%s:%d", file, line+1) - default: - _, file, line, _ := runtime.Caller(0) - logger.Trace(ctx, time.Now(), query, errors.New("query error")) - return fmt.Sprintf("%s:%d", file, line+1) - } -} - -func TestGormLoggerParameterizedQueries(t *testing.T) { - for _, delegated := range []bool{false, true} { - for _, parameterized := range []bool{false, true} { - t.Run(fmt.Sprintf("delegated=%t/parameterized=%t", delegated, parameterized), func(t *testing.T) { - test := newSamplingTestSpan(t, false) - config := gormlogger.Config{ParameterizedQueries: parameterized, LogLevel: gormlogger.Warn} - logger := newParameterFilterLogger(test, config, delegated) - filter, ok := logger.(interface { - ParamsFilter(context.Context, string, ...interface{}) (string, []interface{}) - }) - require.True(t, ok) - sql, params := filter.ParamsFilter(test.ctx, "SELECT ?", "example-value") - require.Equal(t, "SELECT ?", sql) - assertGormQueryParams(t, params, parameterized) - test.finish(t) - }) - } - } -} - -func newParameterFilterLogger(test *samplingTestSpan, config gormlogger.Config, delegated bool) gormlogger.Interface { - if delegated { - return NewGormSpanLogger(test.tracer, stdlog.New(&bytes.Buffer{}, "", 0), config) - } - return NewGormLoggerWithConfig(test.tracer, test.logger, config) -} - -func assertGormQueryParams(t *testing.T, params []interface{}, parameterized bool) { - t.Helper() - if parameterized { - require.Empty(t, params) - return - } - require.Equal(t, []interface{}{"example-value"}, params) -} diff --git a/api/pkg/telemetry/sampling_span_processor.go b/api/pkg/telemetry/sampling_span_processor.go index 958c13d21..d5d4bc319 100644 --- a/api/pkg/telemetry/sampling_span_processor.go +++ b/api/pkg/telemetry/sampling_span_processor.go @@ -2,21 +2,25 @@ package telemetry import ( "strings" + "time" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/codes" sdktrace "go.opentelemetry.io/otel/sdk/trace" ) -const mustSampleAttribute attribute.Key = "httpsms.must_sample" +const ( + mustSampleAttribute attribute.Key = "httpsms.must_sample" + slowDatabaseSpanThreshold = 200 * time.Millisecond +) type samplingSpanProcessor struct { sdktrace.SpanProcessor sampler sdktrace.Sampler } -// NewSamplingSpanProcessor exports warning/error spans and 50% of other spans. -// Spans with dropped telemetry are retained because their severity may be lost. +// 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{ @@ -39,7 +43,23 @@ func shouldRetainSpan(span sdktrace.ReadOnlySpan) bool { span.DroppedAttributes() > 0 || span.Status().Code == codes.Error || hasRetentionAttributes(span.Attributes()) || - hasExceptionalEvents(span.Events()) + hasExceptionalEvents(span.Events()) || + isSlowDatabaseSpan(span) +} + +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 { diff --git a/api/pkg/telemetry/sampling_span_processor_test.go b/api/pkg/telemetry/sampling_span_processor_test.go index 3201019a4..4872ae8c4 100644 --- a/api/pkg/telemetry/sampling_span_processor_test.go +++ b/api/pkg/telemetry/sampling_span_processor_test.go @@ -15,7 +15,6 @@ import ( sdktrace "go.opentelemetry.io/otel/sdk/trace" "go.opentelemetry.io/otel/sdk/trace/tracetest" "go.opentelemetry.io/otel/trace" - gormlogger "gorm.io/gorm/logger" ) func TestSamplingSpanProcessor(t *testing.T) { @@ -100,18 +99,39 @@ func TestSamplingSpanProcessorGormLogs(t *testing.T) { }, errors.New("failed")) test.finish(t, "test") }) - t.Run("slow query warning", func(t *testing.T) { + t.Run("successful query does not promote span", func(t *testing.T) { test := newSamplingTestSpan(t, false) - NewGormLoggerWithConfig(test.tracer, test.logger, gormlogger.Config{ - SlowThreshold: 200 * time.Millisecond, - LogLevel: gormlogger.Warn, - }).Trace(test.ctx, time.Now().Add(-time.Second), func() (string, int64) { + NewGormLogger(test.tracer, test.logger).Trace(test.ctx, time.Now(), func() (string, int64) { return "SELECT 1", 1 }, nil) - test.finish(t, "test") + 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 TestSamplingSpanProcessorWarningEvent(t *testing.T) { test := newSamplingTestSpan(t, false) test.span.AddEvent("log", trace.WithAttributes(attribute.String("log.severity", "warn"))) @@ -181,7 +201,7 @@ type samplingTestSpan struct { exporter *tracetest.InMemoryExporter } -func newSamplingTestSpan(t *testing.T, sampled bool) *samplingTestSpan { +func newSamplingTestSpan(t *testing.T, sampled bool, options ...trace.SpanStartOption) *samplingTestSpan { t.Helper() exporter := tracetest.NewInMemoryExporter() provider := sdktrace.NewTracerProvider( @@ -201,7 +221,7 @@ func newSamplingTestSpan(t *testing.T, sampled bool) *samplingTestSpan { TraceID: traceID, SpanID: trace.SpanID{7: 1}, })) - ctx, span := provider.Tracer("test").Start(ctx, "test") + ctx, span := provider.Tracer("test").Start(ctx, "test", options...) driver := zerolog.Nop() logger := NewZerologLogger("", nil, &zerodriver.Logger{Logger: &driver}, nil) tracer := NewOtelLogger("", logger) From 736a098fa9293022d458e332bf11a6fa06f82427 Mon Sep 17 00:00:00 2001 From: Acho Arnold Ewin Date: Fri, 30 Oct 2026 01:53:28 +0200 Subject: [PATCH 6/8] refactor(api): use existing logger for span retention Bind actual spans through Logger.WithSpan instead of adding a logger wrapper. Remove caller offsets and unused severity parsing while preserving log output. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 60fb2b83-6e9d-4656-9f60-6183c61b716d --- .../handlers/message_thread_handler_test.go | 2 +- .../read_receipts_test_helpers_test.go | 20 +++---- .../gorm_message_thread_repository_test.go | 14 ++--- api/pkg/services/contact_service_test.go | 2 +- .../services/http_notification_sender_test.go | 4 +- api/pkg/services/message_service_test.go | 20 +++---- .../message_thread_service_contacts_test.go | 14 ++--- .../phone_notification_service_test.go | 6 +- api/pkg/telemetry/gorm_logger.go | 8 +-- api/pkg/telemetry/logger.go | 4 +- api/pkg/telemetry/otel_tracer.go | 2 +- api/pkg/telemetry/sampling_span_processor.go | 23 ++------ .../telemetry/sampling_span_processor_test.go | 27 +-------- api/pkg/telemetry/span_logger.go | 40 ------------- api/pkg/telemetry/zerolog_logger.go | 56 +++++++++---------- ..._logger_test.go => zerolog_logger_test.go} | 39 ++++++++++--- .../contact_handler_validator_test.go | 2 +- 17 files changed, 115 insertions(+), 168 deletions(-) delete mode 100644 api/pkg/telemetry/span_logger.go rename api/pkg/telemetry/{span_logger_test.go => zerolog_logger_test.go} (54%) 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_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 7ee37a012..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.tracer.CtxLogger(gorm.logger, gorm.tracer.Span(ctx)).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.tracer.CtxLogger(gorm.logger, gorm.tracer.Span(ctx)).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.tracer.CtxLogger(gorm.logger, gorm.tracer.Span(ctx)).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.tracer.CtxLogger(gorm.logger, gorm.tracer.Span(ctx)).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 b98a2a8f8..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 &spanLogger{Logger: logger.WithSpan(span.SpanContext()), span: span} + 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 index d5d4bc319..ecb6b0b24 100644 --- a/api/pkg/telemetry/sampling_span_processor.go +++ b/api/pkg/telemetry/sampling_span_processor.go @@ -1,7 +1,6 @@ package telemetry import ( - "strings" "time" "go.opentelemetry.io/otel/attribute" @@ -42,7 +41,7 @@ func shouldRetainSpan(span sdktrace.ReadOnlySpan) bool { return span.DroppedEvents() > 0 || span.DroppedAttributes() > 0 || span.Status().Code == codes.Error || - hasRetentionAttributes(span.Attributes()) || + hasRetentionFlag(span.Attributes()) || hasExceptionalEvents(span.Events()) || isSlowDatabaseSpan(span) } @@ -64,32 +63,18 @@ func isSlowDatabaseSpan(span sdktrace.ReadOnlySpan) bool { func hasExceptionalEvents(events []sdktrace.Event) bool { for _, event := range events { - if event.Name == "exception" || event.DroppedAttributeCount > 0 || hasRetentionAttributes(event.Attributes) { + if event.Name == "exception" || event.DroppedAttributeCount > 0 { return true } } return false } -func hasRetentionAttributes(attributes []attribute.KeyValue) bool { +func hasRetentionFlag(attributes []attribute.KeyValue) bool { for _, value := range attributes { - if isRetentionAttribute(value) { + if value.Key == mustSampleAttribute && value.Value.AsBool() { return true } } return false } - -func isRetentionAttribute(value attribute.KeyValue) bool { - if value.Key == mustSampleAttribute { - return value.Value.AsBool() - } - if value.Key != "log.severity" { - return false - } - switch strings.ToLower(value.Value.AsString()) { - case "warn", "warning", "error", "fatal": - return true - } - return false -} diff --git a/api/pkg/telemetry/sampling_span_processor_test.go b/api/pkg/telemetry/sampling_span_processor_test.go index 4872ae8c4..464d27829 100644 --- a/api/pkg/telemetry/sampling_span_processor_test.go +++ b/api/pkg/telemetry/sampling_span_processor_test.go @@ -132,32 +132,12 @@ func TestSamplingSpanProcessorSlowDatabaseSpans(t *testing.T) { } } -func TestSamplingSpanProcessorWarningEvent(t *testing.T) { - test := newSamplingTestSpan(t, false) - test.span.AddEvent("log", trace.WithAttributes(attribute.String("log.severity", "warn"))) - test.finish(t, "test") -} - -func TestSamplingSpanProcessorUppercaseSeverity(t *testing.T) { - for _, severity := range []string{"WARN", "WARNING", "ERROR", "FATAL"} { - t.Run(severity, func(t *testing.T) { - test := newSamplingTestSpan(t, false) - test.span.AddEvent("log", trace.WithAttributes(attribute.String("log.severity", severity))) - test.finish(t, "test") - }) - } -} - func TestSamplingSpanProcessorTruncatedTelemetry(t *testing.T) { scenarios := map[string]func(*samplingTestSpan){ "evicted exception": func(test *samplingTestSpan) { test.span.RecordError(errors.New("failed")) test.fillEvents() }, - "evicted warning event": func(test *samplingTestSpan) { - test.span.AddEvent("log", trace.WithAttributes(attribute.String("log.severity", "warn"))) - test.fillEvents() - }, "warning after event limit": func(test *samplingTestSpan) { test.logger.Warn(errors.New("warning")) test.fillEvents() @@ -167,9 +147,9 @@ func TestSamplingSpanProcessorTruncatedTelemetry(t *testing.T) { test.logger.Warn(errors.New("warning")) test.fillEvents() }, - "dropped warning attribute": func(test *samplingTestSpan) { + "dropped retention flag": func(test *samplingTestSpan) { test.fillAttributes() - test.span.SetAttributes(attribute.String("log.severity", "warn")) + test.span.SetAttributes(attribute.Bool("httpsms.must_sample", true)) }, } for name, record := range scenarios { @@ -187,8 +167,7 @@ func TestSamplingSpanProcessorTruncatedEventAttributes(t *testing.T) { for index := range 200 { attributes = append(attributes, attribute.Int(fmt.Sprintf("attribute.%d", index), index)) } - attributes = append(attributes, attribute.String("log.severity", "warn")) - test.span.AddEvent("log", trace.WithAttributes(attributes...)) + test.span.AddEvent("event", trace.WithAttributes(attributes...)) test.finish(t, "test") } diff --git a/api/pkg/telemetry/span_logger.go b/api/pkg/telemetry/span_logger.go deleted file mode 100644 index ff548a888..000000000 --- a/api/pkg/telemetry/span_logger.go +++ /dev/null @@ -1,40 +0,0 @@ -package telemetry - -import ( - "go.opentelemetry.io/otel/trace" -) - -type spanLogger struct { - Logger - span trace.Span -} - -func (logger *spanLogger) WithService(service string) Logger { - return &spanLogger{Logger: logger.Logger.WithService(service), span: logger.span} -} - -func (logger *spanLogger) WithString(key string, value string) Logger { - return &spanLogger{Logger: logger.Logger.WithString(key, value), span: logger.span} -} - -func (logger *spanLogger) Warn(err error) { - logger.span.SetAttributes(mustSampleAttribute.Bool(true)) - logger.callerLogger().Warn(err) -} - -func (logger *spanLogger) Error(err error) { - logger.span.SetAttributes(mustSampleAttribute.Bool(true)) - logger.callerLogger().Error(err) -} - -func (logger *spanLogger) Fatal(err error) { - logger.span.SetAttributes(mustSampleAttribute.Bool(true)) - logger.callerLogger().Fatal(err) -} - -func (logger *spanLogger) callerLogger() Logger { - if callerLogger, ok := logger.Logger.(interface{ withCallerSkipFrame(int) Logger }); ok { - return callerLogger.withCallerSkipFrame(1) - } - return logger.Logger -} diff --git a/api/pkg/telemetry/zerolog_logger.go b/api/pkg/telemetry/zerolog_logger.go index 0dd9995b3..a0e6c4ca0 100644 --- a/api/pkg/telemetry/zerolog_logger.go +++ b/api/pkg/telemetry/zerolog_logger.go @@ -10,21 +10,19 @@ import ( ) type zerologLogger struct { - zerolog *zerodriver.Logger - spanContext *trace.SpanContext - fields map[string]string - projectID string - level zerolog.Level - callerSkip int + 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) @@ -33,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{}) { @@ -51,7 +44,7 @@ func (logger *zerologLogger) WithString(key string, value string) Logger { logger.projectID, logger.addField(key, value), logger.zerolog, - logger.spanContext, + logger.span, ) } @@ -67,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() } @@ -77,30 +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 { - event.CallerSkipFrame(logger.callerSkip) - 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) @@ -108,10 +104,10 @@ func (logger *zerologLogger) decorateEvent(event *zerodriver.Event) *zerolog.Eve return event.Event } -func (logger *zerologLogger) withCallerSkipFrame(skip int) Logger { - clone := *logger - clone.callerSkip += skip - return &clone +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 { diff --git a/api/pkg/telemetry/span_logger_test.go b/api/pkg/telemetry/zerolog_logger_test.go similarity index 54% rename from api/pkg/telemetry/span_logger_test.go rename to api/pkg/telemetry/zerolog_logger_test.go index 6dd90105d..91a9324ed 100644 --- a/api/pkg/telemetry/span_logger_test.go +++ b/api/pkg/telemetry/zerolog_logger_test.go @@ -15,11 +15,36 @@ import ( "go.opentelemetry.io/otel/trace" ) -func TestSpanLoggerPreservesCaller(t *testing.T) { - for _, wrapped := range []bool{false, true} { +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("wrapped=%t/%s", wrapped, severity), func(t *testing.T) { - logger, output := newCallerTestLogger(wrapped) + 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) }) @@ -27,12 +52,12 @@ func TestSpanLoggerPreservesCaller(t *testing.T) { } } -func newCallerTestLogger(wrapped bool) (Logger, *bytes.Buffer) { +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 wrapped { - logger = NewOtelLogger("", logger).CtxLogger(logger, trace.SpanFromContext(context.Background())) + if withSpan { + logger = logger.WithSpan(trace.SpanFromContext(context.Background())) } return logger.WithString("field", "value").WithService("test"), output } 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) {} From 641545132d7cf6a73bd817d2c1b0d1bc64920c0c Mon Sep 17 00:00:00 2001 From: Acho Arnold Ewin Date: Fri, 30 Oct 2026 02:11:06 +0200 Subject: [PATCH 7/8] fix(api): align metric exemplars with exported traces Share the trace-ID decision with exemplar filtering without dropping metric measurements. Retain spans with dropped links or link attributes. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 60fb2b83-6e9d-4656-9f60-6183c61b716d --- api/pkg/di/container.go | 2 + api/pkg/telemetry/sampling_span_processor.go | 33 ++++++++-- .../telemetry/sampling_span_processor_test.go | 66 +++++++++++++++++++ 3 files changed, 96 insertions(+), 5 deletions(-) diff --git a/api/pkg/di/container.go b/api/pkg/di/container.go index 51c1693a4..e7ca20d28 100644 --- a/api/pkg/di/container.go +++ b/api/pkg/di/container.go @@ -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) @@ -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/telemetry/sampling_span_processor.go b/api/pkg/telemetry/sampling_span_processor.go index ecb6b0b24..1bd0c4ca2 100644 --- a/api/pkg/telemetry/sampling_span_processor.go +++ b/api/pkg/telemetry/sampling_span_processor.go @@ -1,11 +1,13 @@ package telemetry import ( + "context" "time" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/codes" sdktrace "go.opentelemetry.io/otel/sdk/trace" + "go.opentelemetry.io/otel/trace" ) const ( @@ -13,9 +15,10 @@ const ( slowDatabaseSpanThreshold = 200 * time.Millisecond ) +var exportSampler = sdktrace.TraceIDRatioBased(0.5) + type samplingSpanProcessor struct { sdktrace.SpanProcessor - sampler sdktrace.Sampler } // NewSamplingSpanProcessor exports warning/error spans, database spans over 200ms, @@ -24,18 +27,25 @@ type samplingSpanProcessor struct { func NewSamplingSpanProcessor(exporter sdktrace.SpanExporter) sdktrace.SpanProcessor { return &samplingSpanProcessor{ SpanProcessor: sdktrace.NewBatchSpanProcessor(exporter), - sampler: sdktrace.TraceIDRatioBased(0.5), } } func (processor *samplingSpanProcessor) OnEnd(span sdktrace.ReadOnlySpan) { - if processor.sampler.ShouldSample(sdktrace.SamplingParameters{ - TraceID: span.SpanContext().TraceID(), - }).Decision == sdktrace.RecordAndSample || shouldRetainSpan(span) { + 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 { + return exportSampler.ShouldSample(sdktrace.SamplingParameters{TraceID: traceID}).Decision == sdktrace.RecordAndSample +} + func shouldRetainSpan(span sdktrace.ReadOnlySpan) bool { // Drop counters survive buffer limits, unlike severity attributes and events. return span.DroppedEvents() > 0 || @@ -43,9 +53,22 @@ func shouldRetainSpan(span sdktrace.ReadOnlySpan) bool { 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 diff --git a/api/pkg/telemetry/sampling_span_processor_test.go b/api/pkg/telemetry/sampling_span_processor_test.go index 464d27829..a18a25ccb 100644 --- a/api/pkg/telemetry/sampling_span_processor_test.go +++ b/api/pkg/telemetry/sampling_span_processor_test.go @@ -12,6 +12,9 @@ import ( "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" @@ -161,6 +164,69 @@ func TestSamplingSpanProcessorTruncatedTelemetry(t *testing.T) { } } +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{8: 0xff, 15: 1} + if sampled { + traceID[8] = 0 + } + 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 From 0467d4651e410dc91baa011ceb94032cff152a7e Mon Sep 17 00:00:00 2001 From: Acho Arnold Ewin Date: Fri, 30 Oct 2026 02:21:56 +0200 Subject: [PATCH 8/8] fix(api): hash full trace IDs before sampling Avoid UUID version and variant bits biasing the shared 50% decision. Keep exported spans and metric exemplars aligned for UUID-shaped IDs. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 60fb2b83-6e9d-4656-9f60-6183c61b716d --- api/pkg/telemetry/sampling_span_processor.go | 7 +++--- .../telemetry/sampling_span_processor_test.go | 25 +++++++++++++++---- 2 files changed, 24 insertions(+), 8 deletions(-) diff --git a/api/pkg/telemetry/sampling_span_processor.go b/api/pkg/telemetry/sampling_span_processor.go index 1bd0c4ca2..1c683143c 100644 --- a/api/pkg/telemetry/sampling_span_processor.go +++ b/api/pkg/telemetry/sampling_span_processor.go @@ -2,6 +2,7 @@ package telemetry import ( "context" + "crypto/sha256" "time" "go.opentelemetry.io/otel/attribute" @@ -15,8 +16,6 @@ const ( slowDatabaseSpanThreshold = 200 * time.Millisecond ) -var exportSampler = sdktrace.TraceIDRatioBased(0.5) - type samplingSpanProcessor struct { sdktrace.SpanProcessor } @@ -43,7 +42,9 @@ func SampledTraceExemplarFilter(ctx context.Context) bool { } func isSampledTrace(traceID trace.TraceID) bool { - return exportSampler.ShouldSample(sdktrace.SamplingParameters{TraceID: traceID}).Decision == sdktrace.RecordAndSample + // 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 { diff --git a/api/pkg/telemetry/sampling_span_processor_test.go b/api/pkg/telemetry/sampling_span_processor_test.go index a18a25ccb..2ac06a85a 100644 --- a/api/pkg/telemetry/sampling_span_processor_test.go +++ b/api/pkg/telemetry/sampling_span_processor_test.go @@ -2,6 +2,7 @@ package telemetry import ( "context" + "encoding/binary" "errors" "fmt" "testing" @@ -20,6 +21,20 @@ import ( "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) @@ -198,9 +213,9 @@ func TestMetricExemplarsOnlyReferenceSampledTraces(t *testing.T) { require.NoError(t, err) for _, sampled := range []bool{false, true} { - traceID := trace.TraceID{8: 0xff, 15: 1} + traceID := trace.TraceID{6: 0x40, 8: 0x80, 15: 1} if sampled { - traceID[8] = 0 + traceID[15] = 2 } ctx := trace.ContextWithSpanContext(context.Background(), trace.NewSpanContext(trace.SpanContextConfig{ TraceID: traceID, @@ -257,10 +272,10 @@ func newSamplingTestSpan(t *testing.T, sampled bool, options ...trace.SpanStartO require.NoError(t, provider.Shutdown(context.Background())) }) - // Ratio sampling uses the lower half of the trace ID. - traceID := trace.TraceID{8: 0xff, 15: 1} + // Fixed UUID-shaped IDs exercise both outcomes of the stable hash decision. + traceID := trace.TraceID{6: 0x40, 8: 0x80, 15: 1} if sampled { - traceID[8] = 0 + traceID[15] = 2 } ctx := trace.ContextWithRemoteSpanContext(context.Background(), trace.NewSpanContext(trace.SpanContextConfig{ TraceID: traceID,