feat: add LogProcessor to baggagecopy

CHANGELOG.md

@@ -15,6 +15,9 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 - Add the `WithSource` option to the `go.opentelemetry.io/contrib/bridges/otelslog` log bridge to set the `code.*` attributes in the log record that includes the source location where the record was emitted. (#6253)
 - Add `ContextWithStartTime` and `StartTimeFromContext` to `go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp`, which allows setting the start time using go context. (#6137)
 - Set the `code.*` attributes in `go.opentelemetry.io/contrib/bridges/otelzap` if the `zap.Logger` was created with the `AddCaller` or `AddStacktrace` option. (#6268)
+- Add a `LogProcessor` to `go.opentelemetry.io/contrib/processors/baggagecopy` to copy baggage members to log records. (#6277)
+  - Use `baggagecopy.NewLogProcessor` when configuring a Log Provider.
+    - `NewLogProcessor` accepts a `Filter` function type that selects which baggage members are added to the log record.
 
 ### Changed 
 

processors/baggagecopy/doc.go

@@ -1,27 +1,37 @@
 // Copyright The OpenTelemetry Authors
 // SPDX-License-Identifier: Apache-2.0
 
-// Package baggagecopy is an OpenTelemetry [Span Processor] that reads key/values
-// stored in [Baggage] in the starting span's parent context and adds them as
-// attributes to the span.
+// Package baggagecopy is an OpenTelemetry [Span Processor] and [Log Record Processor]
+// that reads key/values stored in [Baggage] in context provided to copy onto the span or log.
+//
+// The SpanProcessor retrieves [Baggage] from the starting span's parent context
+// and adds them as attributes to the span.
 //
 // Keys and values added to Baggage will appear on all subsequent child spans for
 // a trace within this service and will be propagated to external services via
 // propagation headers.
 // If the external services also have a Baggage span processor, the keys and
 // values will appear in those child spans as well.
 //
+// The LogProcessor retrieves [Baggage] from the the context provided when
+// emitting the log and adds them as attributes to the log.
+// Baggage may be propagated to external services via propagation headers.
+// and be used to add context to logs if the service also has a Baggage log processor.
+//
 // Do not put sensitive information in Baggage.
 //
 // # Usage
 //
 // Add the span processor when configuring the tracer provider.
 //
+// Add the log processor when configuring the logger provider.
+//
 // The convenience function [AllowAllBaggageKeys] is provided to
-// allow all baggage keys to be copied to the span. Alternatively, you can
+// allow all baggage keys to be copied. Alternatively, you can
 // provide a custom baggage key predicate to select which baggage keys you want
 // to copy.
 //
 // [Span Processor]: https://opentelemetry.io/docs/specs/otel/trace/sdk/#span-processor
+// [Log Record Processor]: https://opentelemetry.io/docs/specs/otel/logs/sdk/#logrecordprocessor
 // [Baggage]: https://opentelemetry.io/docs/specs/otel/api/baggage
 package baggagecopy // import "go.opentelemetry.io/contrib/processors/baggagecopy"

processors/baggagecopy/example_test.go

@@ -9,16 +9,17 @@ import (
 
 	"go.opentelemetry.io/contrib/processors/baggagecopy"
 	"go.opentelemetry.io/otel/baggage"
+	"go.opentelemetry.io/otel/sdk/log"
 	"go.opentelemetry.io/otel/sdk/trace"
 )
 
-func ExampleNew_allKeys() {
+func ExampleNewSpanProcessor_allKeys() {
 	trace.NewTracerProvider(
 		trace.WithSpanProcessor(baggagecopy.NewSpanProcessor(baggagecopy.AllowAllMembers)),
 	)
 }
 
-func ExampleNew_keysWithPrefix() {
+func ExampleNewSpanProcessor_keysWithPrefix() {
 	trace.NewTracerProvider(
 		trace.WithSpanProcessor(
 			baggagecopy.NewSpanProcessor(
@@ -30,7 +31,7 @@ func ExampleNew_keysWithPrefix() {
 	)
 }
 
-func ExampleNew_keysMatchingRegex() {
+func ExampleNewSpanProcessor_keysMatchingRegex() {
 	expr := regexp.MustCompile(`^key.+`)
 	trace.NewTracerProvider(
 		trace.WithSpanProcessor(
@@ -42,3 +43,34 @@ func ExampleNew_keysMatchingRegex() {
 		),
 	)
 }
+
+func ExampleNewLogProcessor_allKeys() {
+	log.NewLoggerProvider(
+		log.WithProcessor(baggagecopy.NewLogProcessor(baggagecopy.AllowAllMembers)),
+	)
+}
+
+func ExampleNewLogProcessor_keysWithPrefix() {
+	log.NewLoggerProvider(
+		log.WithProcessor(
+			baggagecopy.NewLogProcessor(
+				func(m baggage.Member) bool {
+					return strings.HasPrefix(m.Key(), "my-key")
+				},
+			),
+		),
+	)
+}
+
+func ExampleNewLogProcessor_keysMatchingRegex() {
+	expr := regexp.MustCompile(`^key.+`)
+	log.NewLoggerProvider(
+		log.WithProcessor(
+			baggagecopy.NewLogProcessor(
+				func(m baggage.Member) bool {
+					return expr.MatchString(m.Key())
+				},
+			),
+		),
+	)
+}

processors/baggagecopy/go.mod

@@ -5,7 +5,9 @@ go 1.22
 require (
 	github.com/stretchr/testify v1.9.0
 	go.opentelemetry.io/otel v1.31.0
+	go.opentelemetry.io/otel/log v0.7.0
 	go.opentelemetry.io/otel/sdk v1.31.0
+	go.opentelemetry.io/otel/sdk/log v0.7.0
 )
 
 require (

processors/baggagecopy/go.sum

@@ -15,10 +15,14 @@ github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsT
 github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
 go.opentelemetry.io/otel v1.31.0 h1:NsJcKPIW0D0H3NgzPDHmo0WW6SptzPdqg/L1zsIm2hY=
 go.opentelemetry.io/otel v1.31.0/go.mod h1:O0C14Yl9FgkjqcCZAsE053C13OaddMYr/hz6clDkEJE=
+go.opentelemetry.io/otel/log v0.7.0 h1:d1abJc0b1QQZADKvfe9JqqrfmPYQCz2tUSO+0XZmuV4=
+go.opentelemetry.io/otel/log v0.7.0/go.mod h1:2jf2z7uVfnzDNknKTO9G+ahcOAyWcp1fJmk/wJjULRo=
 go.opentelemetry.io/otel/metric v1.31.0 h1:FSErL0ATQAmYHUIzSezZibnyVlft1ybhy4ozRPcF2fE=
 go.opentelemetry.io/otel/metric v1.31.0/go.mod h1:C3dEloVbLuYoX41KpmAhOqNriGbA+qqH6PQ5E5mUfnY=
 go.opentelemetry.io/otel/sdk v1.31.0 h1:xLY3abVHYZ5HSfOg3l2E5LUj2Cwva5Y7yGxnSW9H5Gk=
 go.opentelemetry.io/otel/sdk v1.31.0/go.mod h1:TfRbMdhvxIIr/B2N2LQW2S5v9m3gOQ/08KsbbO5BPT0=
+go.opentelemetry.io/otel/sdk/log v0.7.0 h1:dXkeI2S0MLc5g0/AwxTZv6EUEjctiH8aG14Am56NTmQ=
+go.opentelemetry.io/otel/sdk/log v0.7.0/go.mod h1:oIRXpW+WD6M8BuGj5rtS0aRu/86cbDV/dAfNaZBIjYM=
 go.opentelemetry.io/otel/trace v1.31.0 h1:ffjsj1aRouKewfr85U2aGagJ46+MvodynlQ1HYdmJys=
 go.opentelemetry.io/otel/trace v1.31.0/go.mod h1:TXZkRk7SM2ZQLtR6eoAWQFIHPvzQ06FJAsO1tJg480A=
 golang.org/x/sys v0.26.0 h1:KHjCJyddX0LoSTb3J+vWpupP9p0oznkqVk/IfjymZbo=

processors/baggagecopy/log_processor.go

@@ -0,0 +1,57 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package baggagecopy // import "go.opentelemetry.io/contrib/processors/baggagecopy"
+
+import (
+	"context"
+
+	"go.opentelemetry.io/otel/baggage"
+	api "go.opentelemetry.io/otel/log"
+	"go.opentelemetry.io/otel/sdk/log"
+)
+
+// LogProcessor is a [log.Processor] implementation that adds baggage
+// members onto a log as attributes.
+type LogProcessor struct {
+	filter Filter
+}
+
+var _ log.Processor = (*LogProcessor)(nil)
+
+// NewLogProcessor returns a new [LogProcessor].
+//
+// The Baggage log processor adds attributes to a log record that are found
+// in Baggage in the parent context at the moment the log is emitted.
+// The passed filter determines which baggage members are added to the span.
+//
+// If filter is nil, all baggage members will be added.
+func NewLogProcessor(filter Filter) *LogProcessor {
+	return &LogProcessor{
+		filter: filter,
+	}
+}
+
+// OnEmit adds Baggage member to a log record as attributes that are pulled from
+// the Baggage found in ctx. Baggage members are filtered by the filter passed
+// to NewLogProcessor.
+func (processor LogProcessor) OnEmit(ctx context.Context, record *log.Record) error {
+	filter := processor.filter
+	if filter == nil {
+		filter = AllowAllMembers
+	}
+
+	for _, member := range baggage.FromContext(ctx).Members() {
+		if filter(member) {
+			record.AddAttributes(api.String(member.Key(), member.Value()))
+		}
+	}
+
+	return nil
+}
+
+// Shutdown is called when the [log.Processor] is shutting down and is a no-op for this processor.
+func (processor LogProcessor) Shutdown(context.Context) error { return nil }
+
+// ForceFlush is called to ensure all logs are flushed to the output and is a no-op for this processor.
+func (processor LogProcessor) ForceFlush(context.Context) error { return nil }

processors/baggagecopy/log_processor_test.go

@@ -0,0 +1,134 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package baggagecopy
+
+import (
+	"context"
+	"regexp"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"go.opentelemetry.io/otel/baggage"
+	api "go.opentelemetry.io/otel/log"
+	"go.opentelemetry.io/otel/sdk/log"
+)
+
+var _ log.Processor = &processor{}
+
+type processor struct {
+	records []*log.Record
+}
+
+func (p *processor) OnEmit(ctx context.Context, r *log.Record) error {
+	p.records = append(p.records, r)
+	return nil
+}
+
+func (p *processor) Shutdown(ctx context.Context) error { return nil }
+
+func (p *processor) ForceFlush(ctx context.Context) error { return nil }
+
+func NewTestProcessor() *processor {
+	return &processor{}
+}
+
+func TestLogProcessorOnEmit(t *testing.T) {
+	tests := []struct {
+		name    string
+		baggage baggage.Baggage
+		filter  Filter
+		want    []api.KeyValue
+	}{
+		{
+			name: "all baggage attributes",
+			baggage: func() baggage.Baggage {
+				b, _ := baggage.New()
+				b = addEntryToBaggage(t, b, "baggage.test", "baggage value")
+				return b
+			}(),
+			filter: AllowAllMembers,
+			want:   []api.KeyValue{api.String("baggage.test", "baggage value")},
+		},
+		{
+			name: "baggage attributes with prefix",
+			baggage: func() baggage.Baggage {
+				b, _ := baggage.New()
+				b = addEntryToBaggage(t, b, "baggage.test", "baggage value")
+				return b
+			}(),
+			filter: func(m baggage.Member) bool {
+				return strings.HasPrefix(m.Key(), "baggage.")
+			},
+			want: []api.KeyValue{api.String("baggage.test", "baggage value")},
+		},
+		{
+			name: "baggage attributes with regex",
+			baggage: func() baggage.Baggage {
+				b, _ := baggage.New()
+				b = addEntryToBaggage(t, b, "baggage.test", "baggage value")
+				return b
+			}(),
+			filter: func(m baggage.Member) bool {
+				return regexp.MustCompile(`^baggage\..*`).MatchString(m.Key())
+			},
+			want: []api.KeyValue{api.String("baggage.test", "baggage value")},
+		},
+		{
+			name: "only adds baggage entries that match predicate",
+			baggage: func() baggage.Baggage {
+				b, _ := baggage.New()
+				b = addEntryToBaggage(t, b, "baggage.test", "baggage value")
+				b = addEntryToBaggage(t, b, "foo", "bar")
+				return b
+			}(),
+			filter: func(m baggage.Member) bool {
+				return m.Key() == "baggage.test"
+			},
+			want: []api.KeyValue{api.String("baggage.test", "baggage value")},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ctx := baggage.ContextWithBaggage(context.Background(), tt.baggage)
+
+			wrapped := &processor{}
+			lp := log.NewLoggerProvider(
+				log.WithProcessor(NewLogProcessor(tt.filter)),
+				log.WithProcessor(wrapped),
+			)
+
+			lp.Logger("test").Emit(ctx, api.Record{})
+
+			require.Len(t, wrapped.records, 1)
+			require.Equal(t, len(tt.want), wrapped.records[0].AttributesLen())
+
+			var got []api.KeyValue
+			wrapped.records[0].WalkAttributes(func(kv api.KeyValue) bool {
+				got = append(got, kv)
+				return true
+			})
+
+			require.Equal(t, tt.want, got)
+		})
+	}
+}
+
+func TestZeroLogProcessorNoPanic(t *testing.T) {
+	lp := new(LogProcessor)
+
+	m, err := baggage.NewMember("key", "val")
+	require.NoError(t, err)
+	b, err := baggage.New(m)
+	require.NoError(t, err)
+
+	ctx := baggage.ContextWithBaggage(context.Background(), b)
+	assert.NotPanics(t, func() {
+		_ = lp.OnEmit(ctx, &log.Record{})
+		_ = lp.Shutdown(ctx)
+		_ = lp.ForceFlush(ctx)
+	})
+}