Documentation
Datadog Tracer
Integrations
contrib/cloud.google.com/go/pubsub.v2

contrib/cloud.google.com/go/pubsub.v2

Those integration are enabled by having the following import in the project’s orchestrion.tool.go file:

import (
	_ "github.com/DataDog/orchestrion"

	_ "github.com/DataDog/dd-trace-go/contrib/cloud.google.com/go/pubsub.v2/v2" // integration
	//...
)

Package pubsub provides an easy way to publish and receive Google Cloud Pub/Sub messages, hiding the details of the underlying server RPCs

Initialize pubsubtrace

Join Point
Definition of pubsub.Subscriber
Advice
Introduce new declarations:
// Using the following synthetic imports:
import (
	instrumentation "github.com/DataDog/dd-trace-go/v2/instrumentation"
	pubsubtrace "github.com/DataDog/dd-trace-go/v2/contrib/cloud.google.com/go/pubsubtrace"
)
var (
	__dd_instr *instrumentation.Instrumentation
	__dd_pstrace *pubsubtrace.Tracer
)

func init() {
	component := instrumentation.PackageGCPPubsubV2
	__dd_instr = instrumentation.Load(component)
	__dd_pstrace = pubsubtrace.NewTracer(__dd_instr, component)
}

Subscription.Receive

Join Point
Function body
Advice
Prepend statements produced by the following template:
// Using the following synthetic imports:
import (
	pubsubtrace "github.com/DataDog/dd-trace-go/v2/contrib/cloud.google.com/go/pubsubtrace"
)
{{- $subscription := .Function.Receiver -}}
{{- $handler := .Function.Argument 1 -}}
__dd_traceFn := __dd_pstrace.TraceReceiveFunc({{ $subscription }})
__dd_wrapHandler := func(h func(ctx context.Context, msg *Message)) func(ctx context.Context, msg *Message) {
	return func(ctx context.Context, msg *Message) {
		__dd_traceMsg := &pubsubtrace.Message{
			ID:              msg.ID,
			Data:            msg.Data,
			OrderingKey:     msg.OrderingKey,
			Attributes:      msg.Attributes,
			DeliveryAttempt: msg.DeliveryAttempt,
			PublishTime:     msg.PublishTime,
		}
		ctx, closeSpan := __dd_traceFn(ctx, __dd_traceMsg)
		defer closeSpan()
		h(ctx, msg)
	}
}
{{ $handler }} = __dd_wrapHandler({{ $handler }})

Topic.Publish

Join Point
Function body
Advice
Prepend statements produced by the following template:
// Using the following synthetic imports:
import (
	pubsubtrace "github.com/DataDog/dd-trace-go/v2/contrib/cloud.google.com/go/pubsubtrace"
)
{{- $topic := .Function.Receiver -}}
{{- $ctx := .Function.Argument 0 -}}
{{- $msg := .Function.Argument 1 -}}
{{- $publishResult := .Function.Result 0 -}}
__dd_traceMsg := &pubsubtrace.Message{
	ID:              {{ $msg }}.ID,
	Data:            {{ $msg }}.Data,
	OrderingKey:     {{ $msg }}.OrderingKey,
	Attributes:      {{ $msg }}.Attributes,
	DeliveryAttempt: {{ $msg }}.DeliveryAttempt,
	PublishTime:     {{ $msg }}.PublishTime,
}
__dd_ctx, __dd_closeSpan := __dd_pstrace.TracePublish({{ $ctx }}, {{ $topic }}, __dd_traceMsg)
{{ $ctx }} = __dd_ctx
{{ $msg }}.Attributes = __dd_traceMsg.Attributes

defer func() {
	{{ $publishResult }}.DDCloseSpan = __dd_closeSpan
}()

Add fields to PublishResult

Join Point
Definition of pubsub.PublishResult
Advice
Introduce new declarations:
type DDCloseSpanFunc = func(serverID string, err error)
Advice
Add new field DDCloseSpan of type DDCloseSpanFunc.

Update PublishResult.Get

Join Point
Function body
Advice
Prepend statements produced by the following template:
{{- $publishResult := .Function.Receiver -}}
{{- $serverID := .Function.Result 0 -}}
{{- $err := .Function.Result 1 -}}
defer func() {
	if {{ $publishResult }}.DDCloseSpan != nil {
		{{ $publishResult }}.DDCloseSpan({{ $serverID }}, {{ $err }})
	}
}()
contrib/cloud.google.com/go/pubsub.v1contrib/confluentinc/confluent-kafka-go/kafka