[exporterhelper] Fix exporter.PersistRequestContext feature gate

dmitryax · dmitryax · commit 8af6b02f7a78 · 2025-07-09T08:24:15.000-07:00
It was ignored before because the PersistRequestContextOnRead and PersistRequestContextOnWrite variables were initialized before the feature gate was set from the command line arguments.
diff --git a/.chloggen/fix_exporter-PersistRequestContext-fg.yaml b/.chloggen/fix_exporter-PersistRequestContext-fg.yaml
@@ -0,0 +1,20 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: bug_fix
+
+# The name of the component, or a single word describing the area of concern, (e.g. otlpreceiver)
+component: exporterhelper
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Fix exporter.PersistRequestContext feature gate
+
+# One or more tracking issues or pull requests related to the change
+issues: [13342]
+
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]
diff --git a/exporter/exporterhelper/internal/queue/fg.go b/exporter/exporterhelper/internal/queue/fg.go
@@ -13,9 +13,9 @@ var PersistRequestContextFeatureGate = featuregate.GlobalRegistry().MustRegister
 	featuregate.WithRegisterDescription("controls whether context should be stored alongside requests in the persistent queue"),
 )
 
-// assign the feature gate to separate variables to make it possible to override the behavior in tests
+// assign the feature gate to separate functions to make it possible to override the behavior in tests
 // on write and read paths separately.
 var (
-	PersistRequestContextOnRead  = PersistRequestContextFeatureGate.IsEnabled()
-	PersistRequestContextOnWrite = PersistRequestContextFeatureGate.IsEnabled()
+	PersistRequestContextOnRead  = PersistRequestContextFeatureGate.IsEnabled
+	PersistRequestContextOnWrite = PersistRequestContextFeatureGate.IsEnabled
 )
diff --git a/exporter/exporterhelper/logs.go b/exporter/exporterhelper/logs.go
@@ -59,7 +59,7 @@ type logsEncoding struct{}
 var _ QueueBatchEncoding[Request] = logsEncoding{}
 
 func (logsEncoding) Unmarshal(bytes []byte) (context.Context, Request, error) {
-	if queue.PersistRequestContextOnRead {
+	if queue.PersistRequestContextOnRead() {
 		ctx, logs, err := pdatareq.UnmarshalLogs(bytes)
 		if errors.Is(err, pdatareq.ErrInvalidFormat) {
 			// fall back to unmarshaling without context
@@ -78,7 +78,7 @@ func (logsEncoding) Unmarshal(bytes []byte) (context.Context, Request, error) {
 
 func (logsEncoding) Marshal(ctx context.Context, req Request) ([]byte, error) {
 	logs := req.(*logsRequest).ld
-	if queue.PersistRequestContextOnWrite {
+	if queue.PersistRequestContextOnWrite() {
 		return pdatareq.MarshalLogs(ctx, logs)
 	}
 	return logsMarshaler.MarshalLogs(logs)
diff --git a/exporter/exporterhelper/logs_test.go b/exporter/exporterhelper/logs_test.go
@@ -170,7 +170,8 @@ func TestLogsRequest_Default_ExportError(t *testing.T) {
 }
 
 func TestLogs_WithPersistentQueue(t *testing.T) {
-	fgOrigState := queue.PersistRequestContextFeatureGate.IsEnabled()
+	fgOrigReadState := queue.PersistRequestContextOnRead
+	fgOrigWriteState := queue.PersistRequestContextOnWrite
 	qCfg := NewDefaultQueueConfig()
 	storageID := component.MustNewIDWithName("file_storage", "storage")
 	qCfg.StorageID = configoptional.Some(storageID)
@@ -212,11 +213,11 @@ func TestLogs_WithPersistentQueue(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			queue.PersistRequestContextOnRead = tt.fgEnabledOnRead
-			queue.PersistRequestContextOnWrite = tt.fgEnabledOnWrite
+			queue.PersistRequestContextOnRead = func() bool { return tt.fgEnabledOnRead }
+			queue.PersistRequestContextOnWrite = func() bool { return tt.fgEnabledOnWrite }
 			t.Cleanup(func() {
-				queue.PersistRequestContextOnRead = fgOrigState
-				queue.PersistRequestContextOnWrite = fgOrigState
+				queue.PersistRequestContextOnRead = fgOrigReadState
+				queue.PersistRequestContextOnWrite = fgOrigWriteState
 			})
 
 			ls := consumertest.LogsSink{}
diff --git a/exporter/exporterhelper/metrics.go b/exporter/exporterhelper/metrics.go
@@ -59,7 +59,7 @@ type metricsEncoding struct{}
 var _ QueueBatchEncoding[Request] = metricsEncoding{}
 
 func (metricsEncoding) Unmarshal(bytes []byte) (context.Context, Request, error) {
-	if queue.PersistRequestContextOnRead {
+	if queue.PersistRequestContextOnRead() {
 		ctx, metrics, err := pdatareq.UnmarshalMetrics(bytes)
 		if errors.Is(err, pdatareq.ErrInvalidFormat) {
 			// fall back to unmarshaling without context
@@ -77,7 +77,7 @@ func (metricsEncoding) Unmarshal(bytes []byte) (context.Context, Request, error)
 
 func (metricsEncoding) Marshal(ctx context.Context, req Request) ([]byte, error) {
 	metrics := req.(*metricsRequest).md
-	if queue.PersistRequestContextOnWrite {
+	if queue.PersistRequestContextOnWrite() {
 		return pdatareq.MarshalMetrics(ctx, metrics)
 	}
 	return metricsMarshaler.MarshalMetrics(metrics)
diff --git a/exporter/exporterhelper/metrics_test.go b/exporter/exporterhelper/metrics_test.go
@@ -170,7 +170,8 @@ func TestMetricsRequest_Default_ExportError(t *testing.T) {
 }
 
 func TestMetrics_WithPersistentQueue(t *testing.T) {
-	fgOrigState := queue.PersistRequestContextFeatureGate.IsEnabled()
+	fgOrigReadState := queue.PersistRequestContextOnRead
+	fgOrigWriteState := queue.PersistRequestContextOnWrite
 	qCfg := NewDefaultQueueConfig()
 	storageID := component.MustNewIDWithName("file_storage", "storage")
 	qCfg.StorageID = configoptional.Some(storageID)
@@ -212,11 +213,11 @@ func TestMetrics_WithPersistentQueue(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			queue.PersistRequestContextOnRead = tt.fgEnabledOnRead
-			queue.PersistRequestContextOnWrite = tt.fgEnabledOnWrite
+			queue.PersistRequestContextOnRead = func() bool { return tt.fgEnabledOnRead }
+			queue.PersistRequestContextOnWrite = func() bool { return tt.fgEnabledOnWrite }
 			t.Cleanup(func() {
-				queue.PersistRequestContextOnRead = fgOrigState
-				queue.PersistRequestContextOnWrite = fgOrigState
+				queue.PersistRequestContextOnRead = fgOrigReadState
+				queue.PersistRequestContextOnWrite = fgOrigWriteState
 			})
 
 			ms := consumertest.MetricsSink{}
diff --git a/exporter/exporterhelper/traces.go b/exporter/exporterhelper/traces.go
@@ -59,7 +59,7 @@ type tracesEncoding struct{}
 var _ QueueBatchEncoding[Request] = tracesEncoding{}
 
 func (tracesEncoding) Unmarshal(bytes []byte) (context.Context, Request, error) {
-	if queue.PersistRequestContextOnRead {
+	if queue.PersistRequestContextOnRead() {
 		ctx, traces, err := pdatareq.UnmarshalTraces(bytes)
 		if errors.Is(err, pdatareq.ErrInvalidFormat) {
 			// fall back to unmarshaling without context
@@ -77,7 +77,7 @@ func (tracesEncoding) Unmarshal(bytes []byte) (context.Context, Request, error)
 
 func (tracesEncoding) Marshal(ctx context.Context, req Request) ([]byte, error) {
 	traces := req.(*tracesRequest).td
-	if queue.PersistRequestContextOnWrite {
+	if queue.PersistRequestContextOnWrite() {
 		return pdatareq.MarshalTraces(ctx, traces)
 	}
 	return tracesMarshaler.MarshalTraces(traces)
diff --git a/exporter/exporterhelper/traces_test.go b/exporter/exporterhelper/traces_test.go
@@ -168,7 +168,8 @@ func TestTracesRequest_Default_ExportError(t *testing.T) {
 }
 
 func TestTraces_WithPersistentQueue(t *testing.T) {
-	fgOrigState := queue.PersistRequestContextFeatureGate.IsEnabled()
+	fgOrigReadState := queue.PersistRequestContextOnRead
+	fgOrigWriteState := queue.PersistRequestContextOnWrite
 	qCfg := NewDefaultQueueConfig()
 	storageID := component.MustNewIDWithName("file_storage", "storage")
 	qCfg.StorageID = configoptional.Some(storageID)
@@ -210,11 +211,11 @@ func TestTraces_WithPersistentQueue(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			queue.PersistRequestContextOnRead = tt.fgEnabledOnRead
-			queue.PersistRequestContextOnWrite = tt.fgEnabledOnWrite
+			queue.PersistRequestContextOnRead = func() bool { return tt.fgEnabledOnRead }
+			queue.PersistRequestContextOnWrite = func() bool { return tt.fgEnabledOnWrite }
 			t.Cleanup(func() {
-				queue.PersistRequestContextOnRead = fgOrigState
-				queue.PersistRequestContextOnWrite = fgOrigState
+				queue.PersistRequestContextOnRead = fgOrigReadState
+				queue.PersistRequestContextOnWrite = fgOrigWriteState
 			})
 
 			ts := consumertest.TracesSink{}
diff --git a/exporter/exporterhelper/xexporterhelper/profiles.go b/exporter/exporterhelper/xexporterhelper/profiles.go
@@ -62,7 +62,7 @@ type profilesEncoding struct{}
 var _ exporterhelper.QueueBatchEncoding[request.Request] = profilesEncoding{}
 
 func (profilesEncoding) Unmarshal(bytes []byte) (context.Context, request.Request, error) {
-	if queue.PersistRequestContextOnRead {
+	if queue.PersistRequestContextOnRead() {
 		ctx, profiles, err := pdatareq.UnmarshalProfiles(bytes)
 		if errors.Is(err, pdatareq.ErrInvalidFormat) {
 			// fall back to unmarshaling without context
@@ -80,7 +80,7 @@ func (profilesEncoding) Unmarshal(bytes []byte) (context.Context, request.Reques
 
 func (profilesEncoding) Marshal(ctx context.Context, req request.Request) ([]byte, error) {
 	profiles := req.(*profilesRequest).pd
-	if queue.PersistRequestContextOnWrite {
+	if queue.PersistRequestContextOnWrite() {
 		return pdatareq.MarshalProfiles(ctx, profiles)
 	}
 	return profilesMarshaler.MarshalProfiles(profiles)
diff --git a/exporter/exporterhelper/xexporterhelper/profiles_test.go b/exporter/exporterhelper/xexporterhelper/profiles_test.go
@@ -168,7 +168,8 @@ func TestProfilesRequestExporter_Default_ExportError(t *testing.T) {
 }
 
 func TestProfiles_WithPersistentQueue(t *testing.T) {
-	fgOrigState := queue.PersistRequestContextFeatureGate.IsEnabled()
+	fgOrigReadState := queue.PersistRequestContextOnRead
+	fgOrigWriteState := queue.PersistRequestContextOnWrite
 	qCfg := exporterhelper.NewDefaultQueueConfig()
 	storageID := component.MustNewIDWithName("file_storage", "storage")
 	qCfg.StorageID = configoptional.Some(storageID)
@@ -210,11 +211,11 @@ func TestProfiles_WithPersistentQueue(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			queue.PersistRequestContextOnRead = tt.fgEnabledOnRead
-			queue.PersistRequestContextOnWrite = tt.fgEnabledOnWrite
+			queue.PersistRequestContextOnRead = func() bool { return tt.fgEnabledOnRead }
+			queue.PersistRequestContextOnWrite = func() bool { return tt.fgEnabledOnWrite }
 			t.Cleanup(func() {
-				queue.PersistRequestContextOnRead = fgOrigState
-				queue.PersistRequestContextOnWrite = fgOrigState
+				queue.PersistRequestContextOnRead = fgOrigReadState
+				queue.PersistRequestContextOnWrite = fgOrigWriteState
 			})
 
 			ts := consumertest.ProfilesSink{}

Original file line number	Diff line number	Diff line change
`@@ -13,9 +13,9 @@ var PersistRequestContextFeatureGate = featuregate.GlobalRegistry().MustRegister`
`13`	`13`	`featuregate.WithRegisterDescription("controls whether context should be stored alongside requests in the persistent queue"),`
`14`	`14`	`)`
`15`	`15`
`16`		`-// assign the feature gate to separate variables to make it possible to override the behavior in tests`
	`16`	`+// assign the feature gate to separate functions to make it possible to override the behavior in tests`
`17`	`17`	`// on write and read paths separately.`
`18`	`18`	`var (`
`19`		`- PersistRequestContextOnRead = PersistRequestContextFeatureGate.IsEnabled()`
`20`		`- PersistRequestContextOnWrite = PersistRequestContextFeatureGate.IsEnabled()`
	`19`	`+ PersistRequestContextOnRead = PersistRequestContextFeatureGate.IsEnabled`
	`20`	`+ PersistRequestContextOnWrite = PersistRequestContextFeatureGate.IsEnabled`
`21`	`21`	`)`