[exporterhelper] Preserve client metadata in the persistent queue (#13220)

Preserve client metadata key values in persistent queue.

It allows request client metadata to propagate through the persistent
queue used by the exporters. The same way as it's done for the in-memory
queue. Currently, it is behind the exporter.PersistRequestContext
feature gate, which can be enabled by adding
`--feature-gates=exporter.PersistRequestContext` to the collector
command line. An exporter buffer stored by a previous version of the
collector (or by a collector with the feature gate disabled) can be read
by a newer collector with the feature enabled. However, the reverse is
not supported: a buffer stored by a newer collector with the feature
enabled cannot be read by an older collector (or by a collector with the
feature gate disabled).

Resolves
#10110,
#11780
and
open-telemetry/opentelemetry-collector-contrib#38666

Follow-up work:

To fully propagate the client context, we also add support for
serializing remaining `client.Client` fields:
```
	Addr net.Addr
	Auth AuthData
```

Then, we probably want to make this configurable or rely on the
transform processor to be able to manipulate the context

Author: dmitryax

.chloggen/persist-request-context.yaml

@@ -5,15 +5,17 @@ change_type: enhancement
 component: exporterhelper
 
 # A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
-note: Add an option to preserve request span context in the persistent queue
+note: Preserve request span context and client metadata in the persistent queue.
 
 # One or more tracking issues or pull requests related to the change
-issues: [11740]
+issues: [11740, 13220]
 
 # (Optional) One or more lines of additional information to render under the primary note.
 # These lines will be padded with 2 spaces and then inserted directly into the document.
 # Use pipe (|) for multiline entries.
 subtext: |
+  It allows internal collector spans and client metadata to propagate through the persistent queue used by 
+  the exporters. The same way as it's done for the in-memory queue.
   Currently, it is behind the exporter.PersistRequestContext feature gate, which can be enabled by adding 
   `--feature-gates=exporter.PersistRequestContext` to the collector command line. An exporter buffer stored by
   a previous version of the collector (or by a collector with the feature gate disabled) can be read by a newer

exporter/debugexporter/go.mod

@@ -41,6 +41,7 @@ require (
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
+	go.opentelemetry.io/collector/client v1.34.0 // indirect
 	go.opentelemetry.io/collector/config/configretry v1.34.0 // indirect
 	go.opentelemetry.io/collector/consumer/consumererror v0.128.0 // indirect
 	go.opentelemetry.io/collector/consumer/consumererror/xconsumererror v0.128.0 // indirect

exporter/exporterhelper/xexporterhelper/go.mod

@@ -37,6 +37,7 @@ require (
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
+	go.opentelemetry.io/collector/client v1.34.0 // indirect
 	go.opentelemetry.io/collector/config/configretry v1.34.0 // indirect
 	go.opentelemetry.io/collector/extension v1.34.0 // indirect
 	go.opentelemetry.io/collector/extension/xextension v0.128.0 // indirect

exporter/exportertest/go.mod

@@ -33,6 +33,7 @@ require (
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
+	go.opentelemetry.io/collector/client v1.34.0 // indirect
 	go.opentelemetry.io/collector/consumer/xconsumer v0.128.0 // indirect
 	go.opentelemetry.io/collector/extension v1.34.0 // indirect
 	go.opentelemetry.io/collector/extension/xextension v0.128.0 // indirect

pdata/xpdata/go.mod

@@ -5,6 +5,7 @@ go 1.23.0
 require (
 	github.com/gogo/protobuf v1.3.2
 	github.com/stretchr/testify v1.10.0
+	go.opentelemetry.io/collector/client v0.0.0-00010101000000-000000000000
 	go.opentelemetry.io/collector/pdata v1.34.0
 	go.opentelemetry.io/collector/pdata/pprofile v0.128.0
 	go.opentelemetry.io/collector/pdata/testdata v0.128.0
@@ -33,3 +34,7 @@ replace go.opentelemetry.io/collector/pdata => ..
 replace go.opentelemetry.io/collector/pdata/pprofile => ../pprofile
 
 replace go.opentelemetry.io/collector/pdata/testdata => ../testdata
+
+replace go.opentelemetry.io/collector/client => ../../client
+
+replace go.opentelemetry.io/collector/consumer => ../../consumer

pdata/xpdata/request/context.go

@@ -8,42 +8,94 @@ import (
 
 	"go.opentelemetry.io/otel/trace"
 
+	"go.opentelemetry.io/collector/client"
+	pdataint "go.opentelemetry.io/collector/pdata/internal"
+	protocommon "go.opentelemetry.io/collector/pdata/internal/data/protogen/common/v1"
+	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/xpdata/request/internal"
 )
 
+var readOnlyState = pdataint.StateReadOnly
+
 // encodeContext encodes the context into a map of strings.
 func encodeContext(ctx context.Context) internal.RequestContext {
-	rc := internal.RequestContext{}
+	return internal.RequestContext{
+		SpanContext:    encodeSpanContext(ctx),
+		ClientMetadata: encodeClientMetadata(ctx),
+	}
+}
+
+func encodeSpanContext(ctx context.Context) *internal.SpanContext {
 	spanCtx := trace.SpanContextFromContext(ctx)
-	if spanCtx.IsValid() {
-		traceID := spanCtx.TraceID()
-		spanID := spanCtx.SpanID()
-		rc.SpanContext = &internal.SpanContext{
-			TraceId:    traceID[:],
-			SpanId:     spanID[:],
-			TraceFlags: uint32(spanCtx.TraceFlags()),
-			TraceState: spanCtx.TraceState().String(),
-			Remote:     spanCtx.IsRemote(),
+	if !spanCtx.IsValid() {
+		return nil
+	}
+	traceID := spanCtx.TraceID()
+	spanID := spanCtx.SpanID()
+	return &internal.SpanContext{
+		TraceId:    traceID[:],
+		SpanId:     spanID[:],
+		TraceFlags: uint32(spanCtx.TraceFlags()),
+		TraceState: spanCtx.TraceState().String(),
+		Remote:     spanCtx.IsRemote(),
+	}
+}
+
+func encodeClientMetadata(ctx context.Context) []protocommon.KeyValue {
+	clientMetadata := client.FromContext(ctx).Metadata
+	metadataMap, metadataFound := pcommon.Map{}, false
+	for k := range clientMetadata.Keys() {
+		if !metadataFound {
+			metadataMap, metadataFound = pcommon.NewMap(), true
+		}
+		vals := metadataMap.PutEmptySlice(k)
+		for i := 0; i < len(clientMetadata.Get(k)); i++ {
+			vals.AppendEmpty().SetStr(clientMetadata.Get(k)[i])
 		}
 	}
-	return rc
+	if metadataFound {
+		return *pdataint.GetOrigMap(pdataint.Map(metadataMap))
+	}
+	return nil
 }
 
 // decodeContext decodes the context from the bytes map.
-func decodeContext(rc *internal.RequestContext) context.Context {
-	if rc == nil || rc.SpanContext == nil {
-		return context.Background()
+func decodeContext(ctx context.Context, rc *internal.RequestContext) context.Context {
+	if rc == nil {
+		return ctx
+	}
+	ctx = decodeSpanContext(ctx, rc.SpanContext)
+	return decodeClientMetadata(ctx, rc.ClientMetadata)
+}
+
+func decodeSpanContext(ctx context.Context, sc *internal.SpanContext) context.Context {
+	if sc == nil {
+		return ctx
 	}
 	traceID := trace.TraceID{}
-	copy(traceID[:], rc.SpanContext.TraceId)
+	copy(traceID[:], sc.TraceId)
 	spanID := trace.SpanID{}
-	copy(spanID[:], rc.SpanContext.SpanId)
-	traceState, _ := trace.ParseTraceState(rc.SpanContext.TraceState)
+	copy(spanID[:], sc.SpanId)
+	traceState, _ := trace.ParseTraceState(sc.TraceState)
 	return trace.ContextWithSpanContext(context.Background(), trace.NewSpanContext(trace.SpanContextConfig{
 		TraceID:    traceID,
 		SpanID:     spanID,
-		TraceFlags: trace.TraceFlags(rc.SpanContext.TraceFlags),
+		TraceFlags: trace.TraceFlags(sc.TraceFlags),
 		TraceState: traceState,
-		Remote:     rc.SpanContext.Remote,
+		Remote:     sc.Remote,
 	}))
 }
+
+func decodeClientMetadata(ctx context.Context, clientMetadata []protocommon.KeyValue) context.Context {
+	if len(clientMetadata) == 0 {
+		return ctx
+	}
+	metadataMap := make(map[string][]string, len(clientMetadata))
+	for k, vals := range pcommon.Map(pdataint.NewMap(&clientMetadata, &readOnlyState)).All() {
+		metadataMap[k] = make([]string, vals.Slice().Len())
+		for i, v := range vals.Slice().All() {
+			metadataMap[k][i] = v.Str()
+		}
+	}
+	return client.NewContext(ctx, client.Info{Metadata: client.NewMetadata(metadataMap)})
+}

pdata/xpdata/request/context_test.go

@@ -7,27 +7,38 @@ import (
 	"context"
 	"testing"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/otel/trace"
 
+	"go.opentelemetry.io/collector/client"
 	"go.opentelemetry.io/collector/pdata/xpdata/request/internal"
 )
 
-func BenchmarkEncodeDecodeContext(b *testing.B) {
-	spanCtx := fakeSpanContext(b)
+func TestEncodeDecodeContext(t *testing.T) {
+	spanCtx := fakeSpanContext(t)
+	clientMetadata := client.NewMetadata(map[string][]string{
+		"key1": {"value1"},
+		"key2": {"value2", "value3"},
+	})
+
+	// Encode a context with a span and client metadata
 	ctx := trace.ContextWithSpanContext(context.Background(), spanCtx)
+	ctx = client.NewContext(ctx, client.Info{
+		Metadata: clientMetadata,
+	})
+	reqCtx := encodeContext(ctx)
+	buf, err := reqCtx.Marshal()
+	require.NoError(t, err)
+
+	// Decode the context
+	gotReqCtx := internal.RequestContext{}
+	err = gotReqCtx.Unmarshal(buf)
+	require.NoError(t, err)
+	gotCtx := decodeContext(context.Background(), &gotReqCtx)
+	assert.Equal(t, spanCtx, trace.SpanContextFromContext(gotCtx))
+	assert.Equal(t, clientMetadata, client.FromContext(gotCtx).Metadata)
 
-	b.ResetTimer()
-	b.ReportAllocs()
-	for i := 0; i < b.N; i++ {
-		reqCtx := encodeContext(ctx)
-		buf, err := reqCtx.Marshal()
-		require.NoError(b, err)
-
-		gotReqCtx := internal.RequestContext{}
-		err = gotReqCtx.Unmarshal(buf)
-		require.NoError(b, err)
-		gotCtx := decodeContext(&gotReqCtx)
-		require.Equal(b, spanCtx, trace.SpanContextFromContext(gotCtx))
-	}
+	// Decode nil request context
+	assert.Equal(t, context.Background(), decodeContext(context.Background(), nil))
 }