OTEL-2540 create persistent_queue_context and test files

Author: jackgopack4

exporter/exporterhelper/internal/queuebatch/persistent_queue.go

@@ -6,21 +6,18 @@ package queuebatch // import "go.opentelemetry.io/collector/exporter/exporterhel
 import (
 	"context"
 	"encoding/binary"
-	"encoding/hex"
 	"encoding/json"
 	"errors"
 	"fmt"
 	"strconv"
 	"sync"
 
-	"go.opentelemetry.io/otel/trace"
 	"go.uber.org/zap"
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/experr"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
 	"go.opentelemetry.io/collector/extension/xextension/storage"
-	"go.opentelemetry.io/collector/featuregate"
 	"go.opentelemetry.io/collector/pipeline"
 )
 
@@ -37,24 +34,14 @@ const (
 	// queueMetadataKey is the new single key for all queue metadata.
 	// TODO: Enable when https://github.com/open-telemetry/opentelemetry-collector/issues/12890 is done
 	//nolint:unused
-	queueMetadataKey           = "qmv0"
-	errInvalidTraceFlagsLength = "trace flags must only be 1 byte"
+	queueMetadataKey = "qmv0"
 )
 
 var (
 	errValueNotSet        = errors.New("value not set")
 	errInvalidValue       = errors.New("invalid value")
 	errNoStorageClient    = errors.New("no storage client extension found")
 	errWrongExtensionType = errors.New("requested extension is not a storage extension")
-
-	// persistRequestContextFeatureGate controls whether request context should be persisted in the queue.
-	persistRequestContextFeatureGate = featuregate.GlobalRegistry().MustRegister(
-		"exporter.PersistRequestContext",
-		featuregate.StageAlpha,
-		featuregate.WithRegisterFromVersion("v0.127.0"),
-		featuregate.WithRegisterDescription("controls whether context should be stored alongside requests in the persistent queue"),
-		featuregate.WithRegisterReferenceURL("https://github.com/open-telemetry/opentelemetry-collector/pull/12934"),
-	)
 )
 
 var indexDonePool = sync.Pool{
@@ -258,80 +245,6 @@ func (pq *persistentQueue[T]) Offer(ctx context.Context, req T) error {
 	return pq.putInternal(ctx, req)
 }
 
-// necessary due to SpanContext and SpanContextConfig not supporting Unmarshal interface,
-// see https://github.com/open-telemetry/opentelemetry-go/issues/1819.
-type spanContext struct {
-	TraceID    string
-	SpanID     string
-	TraceFlags string
-	TraceState string
-	Remote     bool
-}
-
-func localSpanContextFromTraceSpanContext(sc trace.SpanContext) spanContext {
-	return spanContext{
-		TraceID:    sc.TraceID().String(),
-		SpanID:     sc.SpanID().String(),
-		TraceFlags: sc.TraceFlags().String(),
-		TraceState: sc.TraceState().String(),
-		Remote:     sc.IsRemote(),
-	}
-}
-
-func contextWithLocalSpanContext(ctx context.Context, sc spanContext) context.Context {
-	traceID, err := trace.TraceIDFromHex(sc.TraceID)
-	if err != nil {
-		return ctx
-	}
-	spanID, err := trace.SpanIDFromHex(sc.SpanID)
-	if err != nil {
-		return ctx
-	}
-	traceFlags, err := traceFlagsFromHex(sc.TraceFlags)
-	if err != nil {
-		return ctx
-	}
-	traceState, err := trace.ParseTraceState(sc.TraceState)
-	if err != nil {
-		return ctx
-	}
-
-	return trace.ContextWithSpanContext(ctx, trace.NewSpanContext(trace.SpanContextConfig{
-		TraceID:    traceID,
-		SpanID:     spanID,
-		TraceFlags: *traceFlags,
-		TraceState: traceState,
-		Remote:     sc.Remote,
-	}))
-}
-
-// requestContext wraps trace.SpanContext to allow for unmarshaling as well as
-// future metadata key/value pairs to be added.
-type requestContext struct {
-	SpanContext spanContext
-}
-
-// reverse of code in trace library https://github.com/open-telemetry/opentelemetry-go/blob/v1.35.0/trace/trace.go#L143-L168
-func traceFlagsFromHex(hexStr string) (*trace.TraceFlags, error) {
-	decoded, err := hex.DecodeString(hexStr)
-	if err != nil {
-		return nil, err
-	}
-	if len(decoded) != 1 {
-		return nil, errors.New(errInvalidTraceFlagsLength)
-	}
-	traceFlags := trace.TraceFlags(decoded[0])
-	return &traceFlags, nil
-}
-
-func getAndMarshalSpanContext(ctx context.Context) ([]byte, error) {
-	if !persistRequestContextFeatureGate.IsEnabled() {
-		return nil, nil
-	}
-	rc := localSpanContextFromTraceSpanContext(trace.SpanContextFromContext(ctx))
-	return json.Marshal(requestContext{SpanContext: rc})
-}
-
 // putInternal is the internal version that requires caller to hold the mutex lock.
 func (pq *persistentQueue[T]) putInternal(ctx context.Context, req T) error {
 	reqSize := pq.set.sizer.Sizeof(req)

exporter/exporterhelper/internal/queuebatch/persistent_queue_context.go

@@ -0,0 +1,102 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package queuebatch // import "go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch"
+
+import (
+	"context"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+
+	"go.opentelemetry.io/otel/trace"
+
+	"go.opentelemetry.io/collector/featuregate"
+)
+
+const (
+	errInvalidTraceFlagsLength = "trace flags must only be 1 byte"
+)
+
+// persistRequestContextFeatureGate controls whether request context should be persisted in the queue.
+var persistRequestContextFeatureGate = featuregate.GlobalRegistry().MustRegister(
+	"exporter.PersistRequestContext",
+	featuregate.StageAlpha,
+	featuregate.WithRegisterFromVersion("v0.127.0"),
+	featuregate.WithRegisterDescription("controls whether context should be stored alongside requests in the persistent queue"),
+	featuregate.WithRegisterReferenceURL("https://github.com/open-telemetry/opentelemetry-collector/pull/12934"),
+)
+
+// requestContext wraps trace.SpanContext to allow for unmarshaling as well as
+// future metadata key/value pairs to be added.
+type requestContext struct {
+	SpanContext spanContext
+}
+
+// necessary due to SpanContext and SpanContextConfig not supporting Unmarshal interface,
+// see https://github.com/open-telemetry/opentelemetry-go/issues/1819.
+type spanContext struct {
+	TraceID    string
+	SpanID     string
+	TraceFlags string
+	TraceState string
+	Remote     bool
+}
+
+func localSpanContextFromTraceSpanContext(sc trace.SpanContext) spanContext {
+	return spanContext{
+		TraceID:    sc.TraceID().String(),
+		SpanID:     sc.SpanID().String(),
+		TraceFlags: sc.TraceFlags().String(),
+		TraceState: sc.TraceState().String(),
+		Remote:     sc.IsRemote(),
+	}
+}
+
+func contextWithLocalSpanContext(ctx context.Context, sc spanContext) context.Context {
+	traceID, err := trace.TraceIDFromHex(sc.TraceID)
+	if err != nil {
+		return ctx
+	}
+	spanID, err := trace.SpanIDFromHex(sc.SpanID)
+	if err != nil {
+		return ctx
+	}
+	traceFlags, err := traceFlagsFromHex(sc.TraceFlags)
+	if err != nil {
+		return ctx
+	}
+	traceState, err := trace.ParseTraceState(sc.TraceState)
+	if err != nil {
+		return ctx
+	}
+
+	return trace.ContextWithSpanContext(ctx, trace.NewSpanContext(trace.SpanContextConfig{
+		TraceID:    traceID,
+		SpanID:     spanID,
+		TraceFlags: *traceFlags,
+		TraceState: traceState,
+		Remote:     sc.Remote,
+	}))
+}
+
+// reverse of code in trace library https://github.com/open-telemetry/opentelemetry-go/blob/v1.35.0/trace/trace.go#L143-L168
+func traceFlagsFromHex(hexStr string) (*trace.TraceFlags, error) {
+	decoded, err := hex.DecodeString(hexStr)
+	if err != nil {
+		return nil, err
+	}
+	if len(decoded) != 1 {
+		return nil, errors.New(errInvalidTraceFlagsLength)
+	}
+	traceFlags := trace.TraceFlags(decoded[0])
+	return &traceFlags, nil
+}
+
+func getAndMarshalSpanContext(ctx context.Context) ([]byte, error) {
+	if !persistRequestContextFeatureGate.IsEnabled() {
+		return nil, nil
+	}
+	rc := localSpanContextFromTraceSpanContext(trace.SpanContextFromContext(ctx))
+	return json.Marshal(requestContext{SpanContext: rc})
+}