Proposal: Non X-ray Propagator for SQS using Message Attributes. #7620
Description
Summary
Add SQS trace propagator to enable distributed tracing across AWS SQS message queues in OpenTelemetry Go.
Problem
Trace context is lost when messages flow through SQS queues, breaking distributed tracing in microservices.
Proposed Draft
package sqs
import (
"context"
"github.com/aws/aws-sdk-go-v2/aws"
"github.com/aws/aws-sdk-go-v2/service/sqs/types"
"go.opentelemetry.io/otel/propagation"
)
type PropagatorConfig struct {
enabled bool
propagator propagation.TextMapPropagator
}
type PropagatorOption interface {
apply(*PropagatorConfig)
}
type propagatorOptionFunc func(*PropagatorConfig)
func (f propagatorOptionFunc) apply(config *PropagatorConfig) {
f(config)
}
func WithEnabled(enabled bool) PropagatorOption {
return propagatorOptionFunc(func(config *PropagatorConfig) {
config.enabled = enabled
})
}
func WithPropagator(propagator propagation.TextMapPropagator) PropagatorOption {
return propagatorOptionFunc(func(config *PropagatorConfig) {
config.propagator = propagator
})
}
type Propagator struct {
config PropagatorConfig
}
func NewSQSPropagator(opts ...PropagatorOption) *Propagator {
config := PropagatorConfig{
enabled: false,
propagator: propagation.NewCompositeTextMapPropagator(
propagation.TraceContext{},
propagation.Baggage{},
),
}
for _, opt := range opts {
opt.apply(&config)
}
return &Propagator{config: config}
}
var (
_ propagation.TextMapPropagator = &Propagator{}
_ propagation.TextMapCarrier = &MessageAttributeCarrier{}
)
var _ propagation.TextMapPropagator = &Propagator{}
func (p Propagator) Inject(ctx context.Context, carrier propagation.TextMapCarrier) {
if !p.config.enabled {
return
}
carrierAttributes, ok := carrier.(MessageAttributeCarrier)
if !ok || carrierAttributes.attributes == nil {
return
}
p.config.propagator.Inject(ctx, carrierAttributes)
}
func (p Propagator) Extract(ctx context.Context, carrier propagation.TextMapCarrier) context.Context {
if !p.config.enabled {
return ctx
}
carrierAttributes, ok := carrier.(MessageAttributeCarrier)
if !ok || carrierAttributes.attributes == nil {
return ctx
}
return p.config.propagator.Extract(ctx, carrierAttributes)
}
func (p Propagator) Fields() []string {
if !p.config.enabled && p.config.propagator == nil {
return []string{}
}
return p.config.propagator.Fields()
}
type MessageAttributeCarrier struct {
attributes map[string]types.MessageAttributeValue
}
func (m MessageAttributeCarrier) Get(key string) string {
if attr, exists := m.attributes[key]; exists && attr.StringValue != nil {
return *attr.StringValue
}
return ""
}
func (m MessageAttributeCarrier) Set(key string, value string) {
if m.attributes == nil {
m.attributes = make(map[string]types.MessageAttributeValue)
}
m.attributes[key] = types.MessageAttributeValue{
StringValue: aws.String(value),
DataType: aws.String("String"),
}
}
func (m MessageAttributeCarrier) Keys() []string {
keys := make([]string, 0, len(m.attributes))
for key := range m.attributes {
keys = append(keys, key)
}
return keys
}