Visitar URL original
fix(pubsub/v2): fix span linking and add missing attributes (#14326) · googleapis/google-cloud-go@ec20708 · GitHub
Skip to content

Commit ec20708

Browse files
hongalexhongalex
authored
fix(pubsub/v2): fix span linking and add missing attributes (#14326)
This PR addresses the following 1. the `subscribe` span kind should be `INTERNAL` 2. the `process` span kind should be `CONSUMER` 3. Add `gcp.resource.name` to spans
1 parent 5b553fe commit ec20708

5 files changed

Lines changed: 76 additions & 39 deletions

File tree

‎pubsub/v2/iterator.go‎

Lines changed: 7 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -365,7 +365,7 @@ func (it *messageIterator) receive() ([]*Message, error) {
365365
if m.Attributes != nil {
366366
ctx = propagation.TraceContext{}.Extract(ctx, newMessageCarrier(m))
367367
}
368-
opts := getSubscriberOpts(it.projectID, it.subID, m)
368+
opts := getSubscribeSpanAttributes(it.projectID, it.subName, m)
369369
opts = append(
370370
opts,
371371
trace.WithAttributes(
@@ -673,7 +673,7 @@ func (it *messageIterator) sendAck(m map[string]*AckResult) {
673673

674674
// Create the single ack span for this request, and for each
675675
// message, add Subscribe<->Ack links.
676-
opts := getCommonOptions(it.projectID, it.subID)
676+
opts := getCommonOptions(it.projectID, it.subName)
677677
opts = append(
678678
opts,
679679
trace.WithLinks(links...),
@@ -682,7 +682,8 @@ func (it *messageIterator) sendAck(m map[string]*AckResult) {
682682
semconv.CodeFunction("sendAck"),
683683
),
684684
)
685-
_, ackSpan := startSpan(context.Background(), ackSpanName, it.subID, opts...)
685+
var ackSpan trace.Span
686+
ctx, ackSpan = startSpan(ctx, ackSpanName, it.subID, opts...)
686687
defer ackSpan.End()
687688
if ackSpan.SpanContext().IsSampled() {
688689
for _, s := range subscribeSpans {
@@ -759,7 +760,7 @@ func (it *messageIterator) sendModAck(ctx context.Context, m map[string]*AckResu
759760

760761
// Create the single modack/nack span for this request, and for each
761762
// message, add Subscribe<->Modack links.
762-
opts := getCommonOptions(it.projectID, it.subID)
763+
opts := getCommonOptions(it.projectID, it.subName)
763764
opts = append(
764765
opts,
765766
trace.WithLinks(links...),
@@ -777,7 +778,8 @@ func (it *messageIterator) sendModAck(ctx context.Context, m map[string]*AckResu
777778
),
778779
)
779780
}
780-
_, mSpan := startSpan(context.Background(), spanName, it.subID, opts...)
781+
var mSpan trace.Span
782+
ctx, mSpan = startSpan(ctx, spanName, it.subID, opts...)
781783
defer mSpan.End()
782784
if mSpan.SpanContext().IsSampled() {
783785
for _, s := range subscribeSpans {

‎pubsub/v2/publisher.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -205,7 +205,7 @@ var errPublisherOrderingNotEnabled = errors.New("Publisher.EnableMessageOrdering
205205
func (t *Publisher) Publish(ctx context.Context, msg *Message) *PublishResult {
206206
var createSpan trace.Span
207207
if t.enableTracing {
208-
opts := getPublishSpanAttributes(t.c.projectID, t.ID(), msg)
208+
opts := getPublishSpanAttributes(t.c.projectID, t.name, msg)
209209
opts = append(opts, trace.WithAttributes(semconv.CodeFunction("Publish")))
210210
ctx, createSpan = startSpan(ctx, createSpanName, t.ID(), opts...)
211211
}
@@ -423,7 +423,7 @@ func (t *Publisher) publishMessageBundle(ctx context.Context, bms []*bundledMess
423423

424424
projectID, topicID := parseResourceName(t.name)
425425
var pSpan trace.Span
426-
opts := getCommonOptions(projectID, topicID)
426+
opts := getCommonOptions(projectID, t.name)
427427
// Add link to publish RPC span of createSpan(s).
428428
opts = append(opts, trace.WithLinks(links...))
429429
opts = append(

‎pubsub/v2/subscriber.go‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -450,7 +450,8 @@ func (s *Subscriber) Receive(ctx context.Context, f func(context.Context, *Messa
450450
if iter.enableTracing {
451451
schedulerSpan.End()
452452
// Start the process span, and augment the done function to end this span and record events.
453-
otelCtx, ps = startSpan(otelCtx, processSpanName, s.ID())
453+
opts := getProcessSpanAttributes(s.c.projectID, s.name, m)
454+
otelCtx, ps = startSpan(otelCtx, processSpanName, s.ID(), opts...)
454455
old := ackh.doneFunc
455456
ackh.doneFunc = func(ackID string, ack bool, r *ipubsub.AckResult, receiveTime time.Time) {
456457
var eventString string

‎pubsub/v2/trace.go‎

Lines changed: 29 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ import (
1919
"fmt"
2020
"log"
2121
"maps"
22+
"strings"
2223
"sync"
2324

2425
pb "cloud.google.com/go/pubsub/v2/apiv1/pubsubpb"
@@ -365,8 +366,8 @@ const (
365366
resultNacked = "nacked"
366367
resultExpired = "expired"
367368

368-
// custom pubsub specific attributes
369-
gcpProjectIDAttribute = "gcp.project_id"
369+
gcpProjectID = "gcp.project_id"
370+
gcpResourceName = "gcp.resource.name"
370371
pubsubPrefix = "messaging.gcp_pubsub."
371372
eosAttribute = pubsubPrefix + "exactly_once_delivery"
372373
resultAttribute = pubsubPrefix + "result"
@@ -395,7 +396,7 @@ func getPublishSpanAttributes(project, dst string, msg *Message, attrs ...attrib
395396
return opts
396397
}
397398

398-
func getSubscriberOpts(project, dst string, msg *Message, attrs ...attribute.KeyValue) []trace.SpanStartOption {
399+
func getProcessSpanAttributes(project, dst string, msg *Message, attrs ...attribute.KeyValue) []trace.SpanStartOption {
399400
opts := []trace.SpanStartOption{
400401
trace.WithAttributes(
401402
semconv.MessagingMessageID(msg.ID),
@@ -412,16 +413,38 @@ func getSubscriberOpts(project, dst string, msg *Message, attrs ...attribute.Key
412413
return opts
413414
}
414415

415-
func getCommonOptions(projectID, destination string) []trace.SpanStartOption {
416+
func getCommonOptions(projectID, resourceName string) []trace.SpanStartOption {
417+
slash := strings.LastIndex(resourceName, "/")
418+
id := resourceName
419+
if slash != -1 {
420+
id = resourceName[slash+1:]
421+
}
416422
opts := []trace.SpanStartOption{
417423
trace.WithAttributes(
418-
attribute.String(gcpProjectIDAttribute, projectID),
424+
attribute.String(gcpProjectID, projectID),
425+
attribute.String(gcpResourceName, resourceName),
419426
semconv.MessagingSystemGCPPubsub,
420-
semconv.MessagingDestinationName(destination),
427+
semconv.MessagingDestinationName(id),
421428
),
422429
}
423430
return opts
431+
}
424432

433+
func getSubscribeSpanAttributes(project, resourceName string, msg *Message, attrs ...attribute.KeyValue) []trace.SpanStartOption {
434+
opts := []trace.SpanStartOption{
435+
trace.WithAttributes(
436+
semconv.MessagingMessageID(msg.ID),
437+
semconv.MessagingMessageBodySize(len(msg.Data)),
438+
semconv.MessagingGCPPubsubMessageOrderingKey(msg.OrderingKey),
439+
),
440+
trace.WithAttributes(attrs...),
441+
trace.WithSpanKind(trace.SpanKindInternal),
442+
}
443+
if msg.DeliveryAttempt != nil {
444+
opts = append(opts, trace.WithAttributes(semconv.MessagingGCPPubsubMessageDeliveryAttempt(*msg.DeliveryAttempt)))
445+
}
446+
opts = append(opts, getCommonOptions(project, resourceName)...)
447+
return opts
425448
}
426449

427450
// spanRecordError records the error, sets the status to error, and ends the span.

‎pubsub/v2/trace_test.go‎

Lines changed: 36 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -86,6 +86,7 @@ func TestTrace_PublishSpan(t *testing.T) {
8686
}
8787

8888
topicID := "t"
89+
topicName := fmt.Sprintf("projects/%s/topics/%s", testutil.ProjID(), topicID)
8990

9091
expectedSpans := tracetest.SpanStubs{
9192
tracetest.SpanStub{
@@ -99,7 +100,8 @@ func TestTrace_PublishSpan(t *testing.T) {
99100
semconv.MessagingMessageIDKey.String("m0"),
100101
semconv.MessagingSystemGCPPubsub,
101102
semconv.MessagingMessageBodySize(len(m.Data)),
102-
attribute.String(gcpProjectIDAttribute, projName),
103+
attribute.String(gcpProjectID, projName),
104+
attribute.String(gcpResourceName, topicName),
103105
},
104106
Events: []sdktrace.Event{
105107
{
@@ -148,7 +150,8 @@ func TestTrace_PublishSpan(t *testing.T) {
148150
semconv.MessagingDestinationName(topicID),
149151
semconv.CodeFunction("publishMessageBundle"),
150152
semconv.MessagingBatchMessageCount(1),
151-
attribute.String(gcpProjectIDAttribute, projName),
153+
attribute.String(gcpProjectID, projName),
154+
attribute.String(gcpResourceName, topicName),
152155
},
153156
InstrumentationLibrary: instrumentation.Scope{
154157
Name: "cloud.google.com/go/pubsub/v2",
@@ -165,7 +168,6 @@ func TestTrace_PublishSpan(t *testing.T) {
165168
},
166169
},
167170
}
168-
topicName := fmt.Sprintf("projects/%s/topics/%s", testutil.ProjID(), topicID)
169171
publisher := mustCreateTopic(t, c, topicName)
170172
defer publisher.Stop()
171173
if m.OrderingKey != "" {
@@ -338,8 +340,15 @@ func TestTrace_SubscribeSpans(t *testing.T) {
338340
expectedSpans := tracetest.SpanStubs{
339341
tracetest.SpanStub{
340342
Name: fmt.Sprintf("%s %s", subID, processSpanName),
341-
SpanKind: trace.SpanKindInternal,
343+
SpanKind: trace.SpanKindConsumer,
342344
Attributes: []attribute.KeyValue{
345+
semconv.MessagingMessageIDKey.String("m0"),
346+
semconv.MessagingMessageBodySize(len(m.Data)),
347+
semconv.MessagingGCPPubsubMessageOrderingKey(m.OrderingKey),
348+
attribute.String(gcpProjectID, projName),
349+
attribute.String(gcpResourceName, subName),
350+
semconv.MessagingSystemGCPPubsub,
351+
semconv.MessagingDestinationName(subID),
343352
semconv.MessagingOperationTypeDeliver,
344353
},
345354
Events: []sdktrace.Event{
@@ -354,21 +363,20 @@ func TestTrace_SubscribeSpans(t *testing.T) {
354363
},
355364
tracetest.SpanStub{
356365
Name: fmt.Sprintf("%s %s", subID, subscribeSpanName),
357-
SpanKind: trace.SpanKindConsumer,
366+
SpanKind: trace.SpanKindInternal,
358367
Attributes: []attribute.KeyValue{
359-
semconv.CodeFunction("receive"),
360-
semconv.MessagingBatchMessageCount(1),
368+
semconv.MessagingMessageIDKey.String("m0"),
369+
semconv.MessagingMessageBodySize(len(m.Data)),
370+
semconv.MessagingGCPPubsubMessageOrderingKey(m.OrderingKey),
371+
attribute.String(gcpProjectID, projName),
372+
attribute.String(gcpResourceName, subName),
373+
semconv.MessagingSystemGCPPubsub,
361374
semconv.MessagingDestinationName(subID),
362375
attribute.Bool(eosAttribute, enableEOS),
363-
// Hardcoded since the fake server always returns m0 first.
364-
semconv.MessagingMessageIDKey.String("m0"),
365-
// The fake server uses message ID as ackID, this is not the case with live service.
366376
semconv.MessagingGCPPubsubMessageAckID("m0"),
367-
semconv.MessagingGCPPubsubMessageOrderingKey(m.OrderingKey),
377+
semconv.MessagingBatchMessageCount(1),
378+
semconv.CodeFunction("receive"),
368379
attribute.String(resultAttribute, resultAcked),
369-
semconv.MessagingSystemGCPPubsub,
370-
semconv.MessagingMessageBodySize(len(m.Data)),
371-
attribute.String(gcpProjectIDAttribute, projName),
372380
},
373381
Events: []sdktrace.Event{
374382
{
@@ -422,11 +430,12 @@ func TestTrace_SubscribeSpans(t *testing.T) {
422430
},
423431
},
424432
Attributes: []attribute.KeyValue{
425-
semconv.CodeFunction("sendAck"),
426-
semconv.MessagingBatchMessageCount(1),
433+
attribute.String(gcpProjectID, projName),
434+
attribute.String(gcpResourceName, subName),
427435
semconv.MessagingSystemGCPPubsub,
428436
semconv.MessagingDestinationName(subID),
429-
attribute.String(gcpProjectIDAttribute, projName),
437+
semconv.MessagingBatchMessageCount(1),
438+
semconv.CodeFunction("sendAck"),
430439
},
431440
InstrumentationLibrary: instrumentation.Scope{
432441
Name: "cloud.google.com/go/pubsub/v2",
@@ -449,13 +458,14 @@ func TestTrace_SubscribeSpans(t *testing.T) {
449458
},
450459
},
451460
Attributes: []attribute.KeyValue{
452-
semconv.CodeFunction("sendModAck"),
453-
attribute.Bool(receiptModackAttribute, true),
454-
semconv.MessagingGCPPubsubMessageAckDeadline(10),
455-
semconv.MessagingBatchMessageCount(1),
461+
attribute.String(gcpProjectID, projName),
462+
attribute.String(gcpResourceName, subName),
456463
semconv.MessagingSystemGCPPubsub,
457464
semconv.MessagingDestinationName(subID),
458-
attribute.String(gcpProjectIDAttribute, projName),
465+
semconv.MessagingBatchMessageCount(1),
466+
semconv.CodeFunction("sendModAck"),
467+
semconv.MessagingGCPPubsubMessageAckDeadline(10),
468+
attribute.Bool(receiptModackAttribute, true),
459469
},
460470
},
461471
}
@@ -586,13 +596,14 @@ func getPublishSpanStubsWithError(topicID string, m *Message, err error) tracete
586596
Name: fmt.Sprintf("%s %s", topicID, createSpanName),
587597
SpanKind: trace.SpanKindProducer,
588598
Attributes: []attribute.KeyValue{
589-
semconv.CodeFunction("Publish"),
590-
semconv.MessagingDestinationName(topicID),
591599
semconv.MessagingMessageIDKey.String(""),
592600
semconv.MessagingMessageBodySize(len(m.Data)),
593601
semconv.MessagingGCPPubsubMessageOrderingKey(m.OrderingKey),
602+
attribute.String(gcpProjectID, projName),
603+
attribute.String(gcpResourceName, fmt.Sprintf("projects/%s/topics/%s", projName, topicID)),
594604
semconv.MessagingSystemGCPPubsub,
595-
attribute.String(gcpProjectIDAttribute, projName),
605+
semconv.MessagingDestinationName(topicID),
606+
semconv.CodeFunction("Publish"),
596607
},
597608
InstrumentationLibrary: instrumentation.Scope{
598609
Name: "cloud.google.com/go/pubsub/v2",

0 commit comments

Comments
 (0)