diff --git a/internal/events/rabbitmq/consumer/handler.go b/internal/events/rabbitmq/consumer/handler.go index e40b439..4778d54 100644 --- a/internal/events/rabbitmq/consumer/handler.go +++ b/internal/events/rabbitmq/consumer/handler.go @@ -40,7 +40,16 @@ func (r *rabbitmqConsumer) handler(msgs <-chan amqp.Delivery, handler events.Han ), ) - logger := r.logger.With().Str("topic", topic).Ctx(ctx).Logger() + logCtx := r.logger.With().Str("topic", topic).Ctx(ctx) + // Enrich with the OTEL trace_id so every log across the message + // lifecycle (decode errors, ack/nack failures, retries, and the + // application handler itself) carries the same id as the producer + // span. ExtractTrace above already restored the trace context from + // the message headers. + if sc := trace.SpanContextFromContext(ctx); sc.HasTraceID() { + logCtx = logCtx.Str("trace_id", sc.TraceID().String()) + } + logger := logCtx.Logger() ctx = logger.WithContext(ctx) ctx = thunderContext.ContextWithMetadata(ctx, metadataFromAmqpTable(msg.Headers)) // ensures that the correlation ID is propagated or generated diff --git a/pkg/grpc/server.go b/pkg/grpc/server.go index d27fe8b..71ca068 100644 --- a/pkg/grpc/server.go +++ b/pkg/grpc/server.go @@ -26,12 +26,7 @@ type NewServerParams struct { func NewServer(params NewServerParams) *BareServer { grpcServer := &BareServer{} - // We want to add the MetadataPropagator interceptor first and - // logger interceptor last. - params.Interceptors = append( - []grpc.UnaryServerInterceptor{UnaryServerMetadataPropagator}, - append(params.Interceptors, grpcLoggerInterceptor(params.Logger))..., - ) + params.Interceptors = composeInterceptors(params.Logger, params.Interceptors) // default max message size is 4MB maxReceiveMessageSize := 4 * 1024 * 1024 @@ -81,3 +76,19 @@ func grpcLoggerInterceptor(logger *zerolog.Logger) func(context.Context, interfa return h, err } } + +// composeInterceptors builds the server's unary interceptor chain: +// MetadataPropagator first, then the base logger, then the supplied +// interceptors. The base logger MUST come before supplied interceptors: +// they enrich the context logger (e.g. audit fields, trace_id), and if +// the base logger ran after them it would replace the context logger and +// silently drop that enrichment before the handler runs. +func composeInterceptors(logger *zerolog.Logger, supplied []grpc.UnaryServerInterceptor) []grpc.UnaryServerInterceptor { + return append( + []grpc.UnaryServerInterceptor{ + UnaryServerMetadataPropagator, + grpcLoggerInterceptor(logger), + }, + supplied..., + ) +} diff --git a/pkg/grpc/server_test.go b/pkg/grpc/server_test.go new file mode 100644 index 0000000..2fe1f73 --- /dev/null +++ b/pkg/grpc/server_test.go @@ -0,0 +1,103 @@ +package grpc + +import ( + "bytes" + "context" + "encoding/json" + "testing" + + "github.com/rs/zerolog" + "google.golang.org/grpc" +) + +// chain replicates grpc.ChainUnaryInterceptor semantics: interceptors run in +// slice order, each wrapping the next, with the handler innermost. +func chain(interceptors []grpc.UnaryServerInterceptor, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) grpc.UnaryHandler { + chained := handler + for i := len(interceptors) - 1; i >= 0; i-- { + ic := interceptors[i] + next := chained + chained = func(ctx context.Context, req interface{}) (interface{}, error) { + return ic(ctx, req, info, next) + } + } + return chained +} + +// TestComposedChainPreservesSuppliedEnrichment is the regression for the +// interceptor-ordering bug: the base logger interceptor used to be appended +// AFTER supplied interceptors, replacing the context logger and silently +// dropping their enrichment (trace_id, audit fields) before the handler ran. +// +// Unlike a hand-assembled chain, this test takes the interceptor slice from +// composeInterceptors — the same production function NewServer uses — so +// reverting the ordering there makes this test fail. +func TestComposedChainPreservesSuppliedEnrichment(t *testing.T) { + var buf bytes.Buffer + logger := zerolog.New(&buf) + + // Supplied interceptor: enriches the context logger, exactly like + // backend-commons logs.UnaryServerInterceptor does for trace_id. + enriching := func( + ctx context.Context, + req interface{}, + info *grpc.UnaryServerInfo, + handler grpc.UnaryHandler, + ) (interface{}, error) { + l := zerolog.Ctx(ctx).With(). + Str("trace_id", "trace-abc-123"). + Str("grpc_method", info.FullMethod). + Logger() + return handler(l.WithContext(ctx), req) + } + + // Compose through the PRODUCTION function used by NewServer. + interceptors := composeInterceptors(&logger, []grpc.UnaryServerInterceptor{enriching}) + + // Handler logs from its context, like real service handlers do. + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + zerolog.Ctx(ctx).Info().Msg("handled") + return "ok", nil + } + + info := &grpc.UnaryServerInfo{FullMethod: "/test.Service/Do"} + if _, err := chain(interceptors, info, handler)(context.Background(), "req"); err != nil { + t.Fatalf("unexpected error: %v", err) + } + + var entry map[string]any + if err := json.Unmarshal(buf.Bytes(), &entry); err != nil { + t.Fatalf("failed to parse handler log: %v\noutput: %s", err, buf.String()) + } + + if entry["trace_id"] != "trace-abc-123" { + t.Errorf("handler log trace_id = %v, want trace-abc-123 (supplied enrichment was dropped)", entry["trace_id"]) + } + if entry["grpc_method"] != "/test.Service/Do" { + t.Errorf("handler log grpc_method = %v, want /test.Service/Do", entry["grpc_method"]) + } +} + +// TestComposedChainBaseLoggerReachesHandlerWithoutSupplied guards the base +// behavior: with no supplied interceptors, the handler still gets the base +// logger in its context. +func TestComposedChainBaseLoggerReachesHandlerWithoutSupplied(t *testing.T) { + var buf bytes.Buffer + logger := zerolog.New(&buf) + + interceptors := composeInterceptors(&logger, nil) + + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + zerolog.Ctx(ctx).Info().Str("marker", "base").Msg("handled") + return "ok", nil + } + + info := &grpc.UnaryServerInfo{FullMethod: "/test.Service/Do"} + if _, err := chain(interceptors, info, handler)(context.Background(), "req"); err != nil { + t.Fatalf("unexpected error: %v", err) + } + + if !bytes.Contains(buf.Bytes(), []byte(`"marker":"base"`)) { + t.Errorf("handler did not log through the base logger: %s", buf.String()) + } +}