Skip to content

Commit

Permalink
fix(pubsub): use official semconv variable whenever possible (#10904)
Browse files Browse the repository at this point in the history
Signed-off-by: Koichi Shiraishi <[email protected]>
Co-authored-by: Alex Hong <[email protected]>
  • Loading branch information
zchee and hongalex authored Nov 12, 2024
1 parent e85151d commit 1ce4b6d
Show file tree
Hide file tree
Showing 3 changed files with 13 additions and 16 deletions.
2 changes: 1 addition & 1 deletion pubsub/iterator.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"),
),
Expand Down
19 changes: 8 additions & 11 deletions pubsub/trace.go
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand All @@ -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),
Expand All @@ -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
Expand Down
8 changes: 4 additions & 4 deletions pubsub/trace_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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)),
Expand Down Expand Up @@ -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),
},
Expand Down

0 comments on commit 1ce4b6d

Please sign in to comment.