standardize attribute names and span names

pubsub/iterator.go

@@ -30,8 +30,8 @@ import (
 	"cloud.google.com/go/pubsub/internal/distribution"
 	gax "github.com/googleapis/gax-go/v2"
 	"github.com/googleapis/gax-go/v2/apierror"
-	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/propagation"
 	semconv "go.opentelemetry.io/otel/semconv/v1.21.0"
 	"go.opentelemetry.io/otel/trace"
 	"google.golang.org/grpc"
@@ -72,6 +72,7 @@ type messageIterator struct {
 	po         *pullOptions
 	ps         *pullStream
 	subc       *vkit.SubscriberClient
+	subID      string
 	subName    string
 	kaTick     <-chan time.Time // keep-alive (deadline extensions)
 	ackTicker  *time.Ticker     // message acks
@@ -132,12 +133,15 @@ func newMessageIterator(subc *vkit.SubscriberClient, subName string, po *pullOpt
 	cctx, cancel := context.WithCancel(context.Background())
 	cctx = withSubscriptionKey(cctx, subName)
 
+	subID := strings.Split(subName, "/")[3]
+
 	it := &messageIterator{
 		ctx:                cctx,
 		cancel:             cancel,
 		ps:                 ps,
 		po:                 po,
 		subc:               subc,
+		subID:              subID,
 		subName:            subName,
 		kaTick:             time.After(keepAlivePeriod),
 		ackTicker:          ackTicker,
@@ -305,15 +309,13 @@ func (it *messageIterator) receive(maxToPull int32) ([]*Message, error) {
 
 		opts := getSubSpanAttributes(it.subName, m, semconv.MessagingOperationReceive)
 		if m.Attributes != nil {
-			ctx = otel.GetTextMapPropagator().Extract(ctx, newMessageCarrier(m))
-
-			log.Printf("context extracted: %+v\n", ctx)
+			ctx = propagation.TraceContext{}.Extract(ctx, newMessageCarrier(m))
 		}
-		_, span := tracer().Start(ctx, fmt.Sprintf("%s %s", it.subName, subscribeReceiveSpanName), opts...)
+		_, span := tracer().Start(ctx, fmt.Sprintf("%s %s", it.subID, subscribeSpanName), opts...)
 		span.SetAttributes(
 			attribute.Bool(eosAttribute, it.enableExactlyOnceDelivery),
 			attribute.String(ackIDAttribute, ackID),
-			attribute.Int(numBatchedMessagesAttribute, len(pendingMessages)),
+			semconv.MessagingBatchMessageCount(len(pendingMessages)),
 		)
 		it.activeSpan.Store(ackID, span)
 	}
@@ -539,7 +541,7 @@ func (it *messageIterator) sendAck(m map[string]*AckResult) {
 					links = append(links, trace.Link{SpanContext: parentSpan.SpanContext()})
 				}
 			}
-			_, ackSpan := tracer().Start(context.Background(), ackSpanName, trace.WithLinks(links...))
+			_, ackSpan := tracer().Start(context.Background(), fmt.Sprintf("%s %s", it.subID, ackSpanName), trace.WithLinks(links...))
 			defer ackSpan.End()
 			ackSpan.SetAttributes(semconv.MessagingBatchMessageCount(numBatch))
 
@@ -594,14 +596,15 @@ func (it *messageIterator) sendModAck(m map[string]*AckResult, deadline time.Dur
 	for len(ackIDs) > 0 {
 		toSend, ackIDs = splitRequestIDs(ackIDs, ackIDBatchSize)
 
-		var spanName, eventName string
+		spanName := it.subID + " "
+		var eventName string
 		if isNack {
 			recordStat(it.ctx, NackCount, int64(len(toSend)))
-			spanName = nackSpanName
+			spanName += spanName + nackSpanName
 			eventName = "nack"
 		} else {
 			recordStat(it.ctx, ModAckCount, int64(len(toSend)))
-			spanName = modAckSpanName
+			spanName = spanName + modAckSpanName
 			eventName = "modack"
 		}
 
@@ -627,8 +630,8 @@ func (it *messageIterator) sendModAck(m map[string]*AckResult, deadline time.Dur
 				defer mSpan.End()
 				if !isNack {
 					mSpan.SetAttributes(
-						attribute.Int(modackDeadlineSecondsAttribute, int(deadlineSec)),
-						attribute.Bool(initialModackAttribute, isReceipt))
+						attribute.Int(ackDeadlineSecAttribute, int(deadlineSec)),
+						attribute.Bool(receiptModackAttribute, isReceipt))
 				}
 				mSpan.SetAttributes(semconv.MessagingBatchMessageCount(numBatch))
 			}

pubsub/subscription.go

@@ -29,7 +29,7 @@ import (
 	pb "cloud.google.com/go/pubsub/apiv1/pubsubpb"
 	"cloud.google.com/go/pubsub/internal/scheduler"
 	gax "github.com/googleapis/gax-go/v2"
-	semconv "go.opentelemetry.io/otel/semconv/v1.4.0"
+	semconv "go.opentelemetry.io/otel/semconv/v1.21.0"
 	"go.opentelemetry.io/otel/trace"
 	"golang.org/x/sync/errgroup"
 	"google.golang.org/grpc/codes"
@@ -1426,7 +1426,7 @@ func (s *Subscription) Receive(ctx context.Context, f func(context.Context, *Mes
 						m := msg.(*Message)
 						schedulerSpan.End()
 						defer wg.Done()
-						_, cSpan := tracer().Start(ctx3, fmt.Sprintf("%s %s", s.String(), subscribeProcessSpanName))
+						_, cSpan := tracer().Start(ctx3, fmt.Sprintf("%s %s", s.ID(), subscribeProcessSpanName))
 						defer cSpan.End()
 						old2 := ackh.doneFunc
 						ackh.doneFunc = func(ackID string, ack bool, r *ipubsub.AckResult, receiveTime time.Time) {

pubsub/topic.go

@@ -33,10 +33,9 @@ import (
 	gax "github.com/googleapis/gax-go/v2"
 	"go.opencensus.io/stats"
 	"go.opencensus.io/tag"
-	"go.opentelemetry.io/otel"
-	"go.opentelemetry.io/otel/attribute"
 	otelcodes "go.opentelemetry.io/otel/codes"
-	semconv "go.opentelemetry.io/otel/semconv/v1.4.0"
+	"go.opentelemetry.io/otel/propagation"
+	semconv "go.opentelemetry.io/otel/semconv/v1.21.0"
 	"go.opentelemetry.io/otel/trace"
 	"google.golang.org/api/support/bundler"
 	"google.golang.org/grpc"
@@ -572,7 +571,7 @@ var errTopicOrderingNotEnabled = errors.New("Topic.EnableMessageOrdering=false,
 // need to be stopped by calling t.Stop(). Once stopped, future calls to Publish
 // will immediately return a PublishResult with an error.
 func (t *Topic) Publish(ctx context.Context, msg *Message) *PublishResult {
-	ctx, publishSpan := startPublishSpan(ctx, msg, t.String())
+	ctx, publishSpan := startPublishSpan(ctx, msg, t.ID())
 	ctx, err := tag.New(ctx, tag.Insert(keyStatus, "OK"), tag.Upsert(keyTopic, t.name))
 	if err != nil {
 		log.Printf("pubsub: cannot create context with tag in Publish: %v", err)
@@ -765,19 +764,28 @@ func (t *Topic) publishMessageBundle(ctx context.Context, bms []*bundledMessage)
 		// key, it doesn't matter which we read from.
 		orderingKey = bms[0].msg.OrderingKey
 	}
+
+	links := make([]trace.Link, 0, numMsgs)
+	for _, bm := range bms {
+		links = append(links, trace.Link{SpanContext: bm.span.SpanContext()})
+	}
+
+	topicID := strings.Split(t.name, "/")[3]
+	_, pSpan := tracer().Start(ctx, fmt.Sprintf("%s %s", topicID, publishRPCSpanName), trace.WithLinks(links...))
+	pSpan.SetAttributes(semconv.MessagingBatchMessageCount(numMsgs))
+	defer pSpan.End()
+
 	for i, bm := range bms {
 		pbMsgs[i] = &pb.PubsubMessage{
 			Data:        bm.msg.Data,
 			Attributes:  bm.msg.Attributes,
 			OrderingKey: bm.msg.OrderingKey,
 		}
 		if bm.msg.Attributes != nil {
-			ctx = otel.GetTextMapPropagator().Extract(ctx, newMessageCarrier(bm.msg))
+			ctx = propagation.TraceContext{}.Extract(ctx, newMessageCarrier(bm.msg))
 		}
-		_, pSpan := tracer().Start(ctx, publishRPCSpanName)
-		pSpan.SetAttributes(attribute.Int(numBatchedMessagesAttribute, numMsgs))
+
 		defer bm.span.End()
-		defer pSpan.End()
 		bm.msg = nil // release bm.msg for GC
 	}
 

pubsub/trace.go

@@ -28,7 +28,7 @@ import (
 	"go.opentelemetry.io/otel/attribute"
 	otelcodes "go.opentelemetry.io/otel/codes"
 	"go.opentelemetry.io/otel/propagation"
-	semconv "go.opentelemetry.io/otel/semconv/v1.4.0"
+	semconv "go.opentelemetry.io/otel/semconv/v1.21.0"
 	"go.opentelemetry.io/otel/trace"
 	pb "google.golang.org/genproto/googleapis/pubsub/v1"
 	"google.golang.org/protobuf/proto"
@@ -310,36 +310,34 @@ func (c messageCarrier) Keys() []string {
 
 const (
 	// publish span names
-	publisherSpanName          = "send"
+	publisherSpanName          = "create"
 	publishFlowControlSpanName = "publisher flow control"
-	publishBatcherSpanName     = "publish scheduler"
+	publishBatcherSpanName     = "publisher batching"
 	publishRPCSpanName         = "publish"
 
 	// subscribe span names
-	subscribeReceiveSpanName      = "receive"
+	subscribeSpanName             = "subscribe"
 	subscriberFlowControlSpanName = "subscriber flow control"
 	subscribeProcessSpanName      = "process"
 	subscribeSchedulerSpanName    = "subscribe scheduler"
-	receiptModAckSpanName         = "send initial ModifyAckDeadline"
-	modAckSpanName                = "send ModifyAckDeadline"
-	ackSpanName                   = "send Acknowledge"
-	nackSpanName                  = "send Negative Acknowledge"
+	modAckSpanName                = "modify ack deadline"
+	ackSpanName                   = "ack"
+	nackSpanName                  = "nack"
 
 	// custom pubsub specific attributes
-	numBatchedMessagesAttribute    = "messaging.gcp.pubsub.num_messages_in_batch"
-	subscriptionAttribute          = "messaging.gcp.pubsub.subscription"
-	orderingAttribute              = "messaging.gcp.pubsub.ordering_key"
-	deliveryAttemptAttribute       = "messaging.gcp.pubsub.delivery_attempt"
-	eosAttribute                   = "messaging.gcp.pppubsub.exactly_once_delivery"
-	ackIDAttribute                 = "messaging.gcp.pubsub.ack_id"
-	resultAttribute                = "messaging.gcp.pubsub.result"
-	modackDeadlineSecondsAttribute = "messaging.gcp.pubsub.modack_deadline_seconds"
-	initialModackAttribute         = "messaging.gcp.pubsub.is_initial_modack"
+	pubsubPrefix             = "messaging.gcp_pubsub."
+	orderingAttribute        = pubsubPrefix + "message.ordering_key"
+	deliveryAttemptAttribute = pubsubPrefix + "message.delivery_attempt"
+	eosAttribute             = pubsubPrefix + "exactly_once_delivery"
+	ackIDAttribute           = pubsubPrefix + "message.ack_id"
+	resultAttribute          = pubsubPrefix + "result"
+	ackDeadlineSecAttribute  = pubsubPrefix + "ack_deadline_seconds"
+	receiptModackAttribute   = pubsubPrefix + "is_receipt_modack"
 )
 
-func startPublishSpan(ctx context.Context, m *Message, topicName string) (context.Context, trace.Span) {
-	opts := getPublishSpanAttributes(topicName, m)
-	return tracer().Start(ctx, fmt.Sprintf("%s %s", topicName, publisherSpanName), opts...)
+func startPublishSpan(ctx context.Context, m *Message, topicID string) (context.Context, trace.Span) {
+	opts := getPublishSpanAttributes(topicID, m)
+	return tracer().Start(ctx, fmt.Sprintf("%s %s", topicID, publisherSpanName), opts...)
 }
 
 func startPublishFlowControlSpan(ctx context.Context) (context.Context, trace.Span) {
@@ -362,10 +360,10 @@ func getPublishSpanAttributes(topic string, msg *Message, opts ...attribute.KeyV
 	ss := []trace.SpanStartOption{
 		trace.WithAttributes(
 			semconv.MessagingSystemKey.String("pubsub"),
-			semconv.MessagingDestinationKey.String(topic),
-			semconv.MessagingDestinationKindTopic,
-			semconv.MessagingMessageIDKey.String(msg.ID),
-			semconv.MessagingMessagePayloadSizeBytesKey.Int(msgSize),
+			semconv.MessagingDestinationName(topic),
+			semconv.MessagingDestinationTemplate("projects/{projectID}/subscriptions/{subscriptionID}"),
+			semconv.MessagingMessageID(msg.ID),
+			semconv.MessagingMessagePayloadSizeBytes(msgSize),
 			attribute.String(orderingAttribute, msg.OrderingKey),
 		),
 		trace.WithAttributes(opts...),
@@ -381,7 +379,7 @@ func injectPropagation(ctx context.Context, msg *Message) {
 		if msg.Attributes == nil {
 			msg.Attributes = make(map[string]string)
 		}
-		otel.GetTextMapPropagator().Inject(ctx, newMessageCarrier(msg))
+		propagation.TraceContext{}.Inject(ctx, newMessageCarrier(msg))
 	}
 }
 
@@ -394,10 +392,9 @@ func getSubSpanAttributes(sub string, msg *Message, opts ...attribute.KeyValue)
 	ss := []trace.SpanStartOption{
 		trace.WithAttributes(
 			semconv.MessagingSystemKey.String("pubsub"),
-			semconv.MessagingDestinationKindTopic,
-			semconv.MessagingMessageIDKey.String(msg.ID),
-			semconv.MessagingMessagePayloadSizeBytesKey.Int(msgSize),
-			attribute.String(subscriptionAttribute, sub),
+			semconv.MessagingDestinationName(sub),
+			semconv.MessagingMessageID(msg.ID),
+			semconv.MessagingMessagePayloadSizeBytes(msgSize),
 			attribute.String(orderingAttribute, msg.OrderingKey),
 		),
 		trace.WithAttributes(opts...),

pubsub/trace_test.go

@@ -32,7 +32,7 @@ import (
 	"go.opentelemetry.io/otel/sdk/instrumentation"
 	sdktrace "go.opentelemetry.io/otel/sdk/trace"
 	"go.opentelemetry.io/otel/sdk/trace/tracetest"
-	semconv "go.opentelemetry.io/otel/semconv/v1.4.0"
+	semconv "go.opentelemetry.io/otel/semconv/v1.21.0"
 	"go.opentelemetry.io/otel/trace"
 	"google.golang.org/protobuf/proto"
 )
@@ -58,7 +58,7 @@ func TestTrace_MessageCarrier(t *testing.T) {
 	}
 
 	newCtx := context.Background()
-	otel.GetTextMapPropagator().Extract(newCtx, newMessageCarrier(msg))
+	propagation.TraceContext{}.Extract(newCtx, newMessageCarrier(msg))
 	if _, ok := msg.Attributes[googclientPrefix+"traceparent"]; !ok {
 		t.Fatalf("expected traceparent in message attributes, found empty string")
 	}
@@ -95,8 +95,7 @@ func TestTrace_PublishSpan(t *testing.T) {
 			Name:     fmt.Sprintf("%s %s", topicName, publisherSpanName),
 			SpanKind: trace.SpanKindProducer,
 			Attributes: []attribute.KeyValue{
-				semconv.MessagingDestinationKindTopic,
-				semconv.MessagingDestinationKey.String(topicName),
+				semconv.MessagingDestinationName(topicName),
 				// Hardcoded since the fake server always returns m0 first.
 				semconv.MessagingMessageIDKey.String("m0"),
 				semconv.MessagingMessagePayloadSizeBytesKey.Int(msgSize),
@@ -126,7 +125,7 @@ func TestTrace_PublishSpan(t *testing.T) {
 		tracetest.SpanStub{
 			Name: publishRPCSpanName,
 			Attributes: []attribute.KeyValue{
-				attribute.Int(numBatchedMessagesAttribute, 1),
+				semconv.MessagingBatchMessageCount(1),
 			},
 			InstrumentationLibrary: instrumentation.Scope{
 				Name:    "cloud.google.com/go/pubsub",
@@ -349,10 +348,10 @@ func TestTrace_SubscribeSpans(t *testing.T) {
 			},
 		},
 		tracetest.SpanStub{
-			Name:     fmt.Sprintf("%s %s", subName, subscribeReceiveSpanName),
+			Name:     fmt.Sprintf("%s %s", subName, subscribeSpanName),
 			SpanKind: trace.SpanKindConsumer,
 			Attributes: []attribute.KeyValue{
-				semconv.MessagingDestinationKindTopic,
+				semconv.MessagingDestinationName(topicName),
 				// Hardcoded since the fake server always returns m0 first.
 				semconv.MessagingMessageIDKey.String("m0"),
 				semconv.MessagingMessagePayloadSizeBytesKey.Int(msgSize),
@@ -363,8 +362,7 @@ func TestTrace_SubscribeSpans(t *testing.T) {
 				attribute.String(resultAttribute, "ack"),
 				attribute.String(orderingAttribute, m.OrderingKey),
 				semconv.MessagingSystemKey.String("pubsub"),
-				attribute.Int(numBatchedMessagesAttribute, 1),
-				attribute.String(subscriptionAttribute, subName),
+				semconv.MessagingBatchMessageCount(1),
 			},
 			InstrumentationLibrary: instrumentation.Scope{
 				Name:    "cloud.google.com/go/pubsub",
@@ -399,9 +397,9 @@ func TestTrace_SubscribeSpans(t *testing.T) {
 				Version: internal.Version,
 			},
 			Attributes: []attribute.KeyValue{
-				attribute.Bool(initialModackAttribute, true),
-				attribute.Int(modackDeadlineSecondsAttribute, 10),
-				attribute.Int(numBatchedMessagesAttribute, 1),
+				attribute.Bool(receiptModackAttribute, true),
+				attribute.Int(ackDeadlineSecAttribute, 10),
+				semconv.MessagingBatchMessageCount(1),
 			},
 		},
 	}
@@ -457,8 +455,7 @@ func getPublishSpanStubsWithError(topicName string, m *Message, msgSize int, err
 			Name:     fmt.Sprintf("%s %s", topicName, publisherSpanName),
 			SpanKind: trace.SpanKindProducer,
 			Attributes: []attribute.KeyValue{
-				semconv.MessagingDestinationKindTopic,
-				semconv.MessagingDestinationKey.String(topicName),
+				semconv.MessagingDestinationName(topicName),
 				semconv.MessagingMessageIDKey.String(""),
 				semconv.MessagingMessagePayloadSizeBytesKey.Int(msgSize),
 				attribute.String(orderingAttribute, m.OrderingKey),