Jetstream: key header changes when received by the consumer

[Alturino]

Observed behavior

The header key changes when received by the consumer the key observed in publisher is
Traceparent but when received by the consumer the header key changes to traceparent

Expected behavior

The header key must not changes when received by the consumer when sent from publisher the key is Traceparent it must preserve the key Traceparent when it is consumed by the consumer

Server and client version

nats:2.11-alpine

Host environment

No response

Steps to reproduce

func publish(ctx context.Context, js jetstream.JetStream) {
	ctx, span := otel.Tracer("publisher").Start(ctx, "publish")
	defer span.End()

	header := make(nats.Header)
	carrier := propagation.HeaderCarrier(header)
	inOtel.GetTextMapPropagator().Inject(ctx, carrier)

	log.Println("publisher headers:", header)

	msgID := randomstring.HumanFriendlyEnglishString(16)
	fut, err := js.PublishMsgAsync(
		&nats.Msg{
			Header:  header,
			Data:    []byte("hello from sender"),
			Subject: "events.download." + msgID,
		},
		jetstream.WithMsgID(msgID),
	)
	if err != nil {
		err = fmt.Errorf("failed to PublishMsgAsync with err: %w", err)
		log.Fatalln(err.Error())
	}
	log.Printf("published message to subject: %s", fut.Msg().Subject)
}
func consume(cs jetstream.Consumer) {
	cc, err := cs.Consume(func(msg jetstream.Msg) {
		carrier := propagation.HeaderCarrier(msg.Headers())
		ctx := inOtel.GetTextMapPropagator().Extract(context.Background(), carrier)

		ctx, span := otel.Tracer("consumer").Start(ctx, "received message")
		defer span.End()

		handleMessage(ctx, msg)

		defer func() {
			log.Println("acknowledging message")
			if err := msg.Ack(); err != nil {
				log.Printf("failed to acknowledge message with err: %s", err.Error())
				return
			}
			log.Println("message acknowledged")
		}()
	})
	if err != nil {
		err = fmt.Errorf("failed to consume with err: %w", err)
		log.Fatalln(err.Error())
	}
	defer func() {
		log.Println("stopping consumer connection")
		cc.Stop()
	}()
}
func handleMessage(ctx context.Context, msg jetstream.Msg) {
	ctx, span := otel.Tracer("consumer").Start(ctx, "handleMessage")
	defer span.End()

	log.Printf("message received from subject: %s, message data: %s", msg.Subject(), msg.Data())
}

[piotrpio]

Hello @Alturino, thanks for reporting the issue. NATS Server always uses lowercase traceparent header specifically in order to comply with W3C Trace Context spec (https://www.w3.org/TR/trace-context/#header-name).

See the code in nats-server: https://github.com/nats-io/nats-server/blob/7cdd5cfdcae083a7666713e51142c124d37b66ff/server/msgtrace.go#L37

[Alturino]

Hi @piotrpio I've seen this issues (nats-io/nats-server#5057) nats contributor is trying to keep the traceparent to comply with W3C Trace Context spec (https://www.w3.org/TR/trace-context/#header-name).

while the opentelemetry HeaderCarrier is using http.Header which changes header key to the Canonical-Key

func (hc HeaderCarrier) Get(key string) string {
	return http.Header(hc).Get(key)
}

func (hc HeaderCarrier) Set(key, value string) {
	http.Header(hc).Set(key, value)
}

func (hc HeaderCarrier) Values(key string) []string {
	return http.Header(hc).Values(key)
}

and these issues #1868 #1573 they intent to implement the keys method because of TextMapCarrier interface

type TextMapCarrier interface {
	// DO NOT CHANGE: any modification will not be backwards compatible and
	// must never be done outside of a new major release.

	// Get returns the value associated with the passed key.
	Get(key string) string
	// DO NOT CHANGE: any modification will not be backwards compatible and
	// must never be done outside of a new major release.

	// Set stores the key-value pair.
	Set(key, value string)
	// DO NOT CHANGE: any modification will not be backwards compatible and
	// must never be done outside of a new major release.

	// Keys lists the keys stored in this carrier.
	Keys() []string
	// DO NOT CHANGE: any modification will not be backwards compatible and
	// must never be done outside of a new major release.
}

My current solution is just using https://pkg.go.dev/go.opentelemetry.io/contrib/propagators/jaeger

What do you think is it nats.go problem or the nats-server problem or is it opentelemetry problem ?

[Alturino]

because of this open-telemetry/opentelemetry-go#6808 they cannot deprecate the Keys() method so I think it's best to nats.Header to implement Keys method

[paivagustavo]

I've just hit this myself as well. We've upgrated to Nats 2.12 and our traces were broken. Seems like this feature was added to the nats-server 2.11, which forces the header to be lower-cased breaking otel go's header propagation.

I agree with @Alturino that nats should probably implement otel go's header implementation so we're not forced to use propagation.HeaderCarrier or implement our own propagation map carrier, for now i'm forced to implementing it myself:

type natsHeader nats.Header

var _ propagation.TextMapCarrier = (*natsHeader)(nil)

func (h natsHeader) Get(key string) string {
	if len(h) == 0 {
		return ""
	}
	values := h[key]
	if len(values) == 0 {
		return ""
	}
	return values[0]
}

func (h natsHeader) Set(key, value string) {
	h[key] = []string{value}
}

func (h natsHeader) Keys() []string {
	return slices.Collect(maps.Keys(h))
}

And then I can extract the span from the nat's message with:

ctx = otel.GetTextMapPropagator().Extract(ctx, natsHeader(msg.Headers()))