fix(pubsub): use official semconv variable whenever possible (#10904)

zchee · hongalex · web-flow · commit 1ce4b6dc3165 · 2024-11-12T16:58:16.000-05:00
Signed-off-by: Koichi Shiraishi &lt;zchee.io@gmail.com&gt;
Co-authored-by: Alex Hong &lt;9397363+hongalex@users.noreply.github.com&gt;
diff --git a/pubsub/iterator.go b/pubsub/iterator.go
@@ -340,7 +340,7 @@ func (it *messageIterator) receive(maxToPull int32) ([]*Message, error) {
 				opts,
 				trace.WithAttributes(
 					attribute.Bool(eosAttribute, it.enableExactlyOnceDelivery),
-					attribute.String(ackIDAttribute, ackID),
+					semconv.MessagingGCPPubsubMessageAckID(ackID),
 					semconv.MessagingBatchMessageCount(len(msgs)),
 					semconv.CodeFunction("receive"),
 				),
diff --git a/pubsub/trace.go b/pubsub/trace.go
@@ -360,14 +360,11 @@ const (
 	resultExpired = "expired"
 
 	// custom pubsub specific attributes
-	gcpProjectIDAttribute    = "gcp.project_id"
-	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"
-	receiptModackAttribute   = pubsubPrefix + "is_receipt_modack"
+	gcpProjectIDAttribute  = "gcp.project_id"
+	pubsubPrefix           = "messaging.gcp_pubsub."
+	eosAttribute           = pubsubPrefix + "exactly_once_delivery"
+	resultAttribute        = pubsubPrefix + "result"
+	receiptModackAttribute = pubsubPrefix + "is_receipt_modack"
 )
 
 func startSpan(ctx context.Context, spanType, resourceID string, opts ...trace.SpanStartOption) (context.Context, trace.Span) {
@@ -383,7 +380,7 @@ func getPublishSpanAttributes(project, dst string, msg *Message, attrs ...attrib
 		trace.WithAttributes(
 			semconv.MessagingMessageID(msg.ID),
 			semconv.MessagingMessageBodySize(len(msg.Data)),
-			attribute.String(orderingAttribute, msg.OrderingKey),
+			semconv.MessagingGCPPubsubMessageOrderingKey(msg.OrderingKey),
 		),
 		trace.WithAttributes(attrs...),
 		trace.WithSpanKind(trace.SpanKindProducer),
@@ -397,13 +394,13 @@ func getSubscriberOpts(project, dst string, msg *Message, attrs ...attribute.Key
 		trace.WithAttributes(
 			semconv.MessagingMessageID(msg.ID),
 			semconv.MessagingMessageBodySize(len(msg.Data)),
-			attribute.String(orderingAttribute, msg.OrderingKey),
+			semconv.MessagingGCPPubsubMessageOrderingKey(msg.OrderingKey),
 		),
 		trace.WithAttributes(attrs...),
 		trace.WithSpanKind(trace.SpanKindConsumer),
 	}
 	if msg.DeliveryAttempt != nil {
-		opts = append(opts, trace.WithAttributes(attribute.Int(deliveryAttemptAttribute, *msg.DeliveryAttempt)))
+		opts = append(opts, trace.WithAttributes(semconv.MessagingGCPPubsubMessageDeliveryAttempt(*msg.DeliveryAttempt)))
 	}
 	opts = append(opts, getCommonOptions(project, dst)...)
 	return opts
diff --git a/pubsub/trace_test.go b/pubsub/trace_test.go
@@ -93,7 +93,7 @@ func TestTrace_PublishSpan(t *testing.T) {
 			Attributes: []attribute.KeyValue{
 				semconv.CodeFunction("Publish"),
 				semconv.MessagingDestinationName(topicID),
-				attribute.String(orderingAttribute, m.OrderingKey),
+				semconv.MessagingGCPPubsubMessageOrderingKey(m.OrderingKey),
 				// Hardcoded since the fake server always returns m0 first.
 				semconv.MessagingMessageIDKey.String("m0"),
 				semconv.MessagingSystemGCPPubsub,
@@ -371,8 +371,8 @@ func TestTrace_SubscribeSpans(t *testing.T) {
 				// Hardcoded since the fake server always returns m0 first.
 				semconv.MessagingMessageIDKey.String("m0"),
 				// The fake server uses message ID as ackID, this is not the case with live service.
-				attribute.String(ackIDAttribute, "m0"),
-				attribute.String(orderingAttribute, m.OrderingKey),
+				semconv.MessagingGCPPubsubMessageAckID("m0"),
+				semconv.MessagingGCPPubsubMessageOrderingKey(m.OrderingKey),
 				attribute.String(resultAttribute, resultAcked),
 				semconv.MessagingSystemGCPPubsub,
 				semconv.MessagingMessageBodySize(len(m.Data)),
@@ -604,7 +604,7 @@ func getPublishSpanStubsWithError(topicID string, m *Message, err error) tracete
 				semconv.MessagingDestinationName(topicID),
 				semconv.MessagingMessageIDKey.String(""),
 				semconv.MessagingMessageBodySize(len(m.Data)),
-				attribute.String(orderingAttribute, m.OrderingKey),
+				semconv.MessagingGCPPubsubMessageOrderingKey(m.OrderingKey),
 				semconv.MessagingSystemGCPPubsub,
 				attribute.String(gcpProjectIDAttribute, projName),
 			},
