feat(pubsub): add open telemetry trace support

[hongalex]

This PR adds the foundation for tracing of a publisher and subscriber for Pub/Sub. This PR attempts to mimic existing messaging system tracing and follows the semantic conventions defined here. This is a draft and is subject to backwards incompatible changes.

Here's a sample that demonstrates using this otel enabled library w/ Google Cloud Tracing as the exporter. The output looks something like the following:

[quartzmo]

Is this already set in getSpanAttributes on line 532, above?

[quartzmo]

I think I understand: After injecting the span into the message attributes, the size of the message needs to be updated.

[quartzmo]

Does the "publish RPC" span also need to be updated with the message ID?

[Gentoli]

why did this context switch to background?

[Gentoli]

this does not work when the gcp identity don't have roles/pubsub.viewer, it should fallback to the subscription name in the err != nil case, otherwise the span name will be " receive"

[Gentoli]

this span can block for a long time if there is no message to be processed in the topic.
It can potentially cause the consumer span to be started before the message is published:

Process P:     | topic send |
--
Process C: |---sub receive---|---sub process---|

where both sub receive and sub process are children of topic send.

This is what it looks like in honeycomb (with no parent trace in the message)

PS: It also causes the span based metrics to be messed up.

[Gentoli]

msg.OrderingKey is empty on unordered, maybe there should be a default when setting it in attributes.

google-cloud-go/internal/pubsub/message.go

Lines 58 to 60 in dd8973b

	// OrderingKey identifies related messages for which publish order should
	// be respected. If empty string is used, message will be sent unordered.
	OrderingKey string

[mbn18]

Hey, is there any expectation to when this will be supported? thanks

[hongalex]

Hey, is there any expectation to when this will be supported? thanks

Thanks for your patience. This was briefly paused to account for a different design decision for tracing. Initially, we were following the specifications on this otel messaging page on batch receiving which we think no longer makes sense for Pub/Sub users. We believe that it would be more useful for users to trace messages individually to better view the lifespan of a single message, which deviates slightly from otel semantic convention's approach of tracking multiple messages in a single trace. If you're interested in learning more, I'd be happy to share.

With that said, we're currently prioritizing other features at the moment, including exactly-once delivery and BigQuery subscriptions, which should be out in the next month or so. Afterwards, resuming work on tracing will be the highest priority item.

[mbn18]

@hongalex , thanks for letting me know.

For the mean while I took your propagator design and implemented it in my code. Will add it here so other can use it till a final solution exist

package whatever

import (
	"cloud.google.com/go/pubsub"
	"context"
	"go.opentelemetry.io/otel"
)

const pubsubAttribPrefix = "whatever"

// PubsubMessageCarrier injects and extracts traces from a pubsub.Message.
type PubsubMessageCarrier struct {
	msg *pubsub.Message
}

// NewPubsubMessageCarrier creates a new PubsubMessageCarrier.
func NewPubsubMessageCarrier(msg *pubsub.Message) PubsubMessageCarrier {
	return PubsubMessageCarrier{msg: msg}
}

// Get retrieves a single value for a given key.
func (c PubsubMessageCarrier) Get(key string) string {
	return c.msg.Attributes[pubsubAttribPrefix+"_"+key]
}

// Set sets an attribute.
func (c PubsubMessageCarrier) Set(key, val string) {
	c.msg.Attributes[pubsubAttribPrefix+"_"+key] = val
}

// Keys returns a slice of all keys in the carrier.
func (c PubsubMessageCarrier) Keys() []string {
	i := 0
	out := make([]string, len(c.msg.Attributes))
	for k := range c.msg.Attributes {
		out[i] = k
		i++
	}
	return out
}

func PubSubMessageInjectContext(ctx context.Context, msg *pubsub.Message) {
	otel.GetTextMapPropagator().Inject(ctx, NewPubsubMessageCarrier(msg))
}

func PubSubMessageExtractContext(ctx context.Context, msg *pubsub.Message) context.Context {
	return otel.GetTextMapPropagator().Extract(ctx, NewPubsubMessageCarrier(msg))
}

*** Note, this does not provide attributes that need to be added to the designated span. like semconv.MessagingDestinationKindTopic

[Mistic92]

Hi, what is the current status?

[panperla]

Are there any alternative thread? Or it just died and nobody is looking into it?

[meredithslota]

We are still tracking this here: #4665