pubsub

README

Instana instrumentation for Google Cloud Pub/Sub

This module contains instrumentation code for Google Cloud Pub/Sub producers and consumers that use cloud.google.com/go/pubsub library starting from v1.3.1 and above.

Installation

To add the module to your go.mod file run the following command in your project directory:

$ go get github.com/instana/go-sensor/instrumentation/cloud.google.com/go/pubsub

Usage

This module is a drop-in replacement for cloud.google.com/go/pubsub in common publisher/subscriber use cases. It provides type aliases for pubsub.Message, pubsub.PublishResult and pubsub.Subscription, however if your code performs any configuration calls to the Pub/Sub API, such as updating the topic configuration or creating subscriptions, you might need add the named import for cloud.google.com/go/pubsub as well.

The instrumentation is implemented as a thin wrapper around service object methods and does not change their behavior. Thus, any limitations/usage patterns/recommendations for the original method also apply to the wrapped one.

In most cases it is enough to change the import path from cloud.google.com/go/pubsub to github.com/instana/go-sensor/instrumentation/cloud.google.com/go/pubsub and add an instance of instana.Sensor to the list of pubsub.NewClient() arguments to start tracing your communication over Google Cloud Pub/Sub with Instana.

Instrumenting push delivery handlers

In case your service uses push method to receive Google Cloud Pub/Sub messages, this instrumentation package provides an HTTP middleware pubsub.TracingHandlerFunc() that extracts Instana trace context and ensures trace continuation:

http.Handle("/", pubsub.TracingHandlerFunc(sensor, func (w http.ResponseWriter, req *http.Request) {
	// deserialize and process the message from req.Body
})

The pubsub.TracingHandlerFunc() is a complete replacement for a more generic HTTP instrumentation done with instana.TracingHandlerFunc(), and will use it for any non-Pub/Sub request, so there is no need to use both middleware wrappers on one handler.

Documentation

Overview

Package pubsub provides Instana tracing instrumentation for Google Cloud Pub/Sub producers and consumers that use cloud.google.com/go/pubsub.

Index

• func TracingHandlerFunc(sensor instana.TracerLogger, pathTemplate string, handler http.HandlerFunc) http.HandlerFunc
• type Client
  • func NewClient(ctx context.Context, projectID string, sensor instana.TracerLogger, ...) (*Client, error)
  • func (c *Client) CreateSubscription(ctx context.Context, id string, cfg pubsub.SubscriptionConfig) (*Subscription, error)
  • func (c *Client) CreateTopic(ctx context.Context, id string) (*Topic, error)
  • func (c *Client) CreateTopicWithConfig(ctx context.Context, topicID string, tc *pubsub.TopicConfig) (*Topic, error)
  • func (c *Client) Subscription(id string) *Subscription
  • func (c *Client) SubscriptionInProject(id, projectID string) *Subscription
  • func (c *Client) Subscriptions(ctx context.Context) *SubscriptionIterator
  • func (c *Client) Topic(id string) *Topic
  • func (c *Client) TopicInProject(id, projectID string) *Topic
  • func (c *Client) Topics(ctx context.Context) *TopicIterator
• type Message
• type PublishResult
• type Snapshot
• type Subscription
  • func (sub *Subscription) Receive(ctx context.Context, f func(context.Context, *pubsub.Message)) error
• type SubscriptionIterator
  • func (it *SubscriptionIterator) Next() (*Subscription, error)
• type Topic
  • func (top *Topic) Publish(ctx context.Context, msg *pubsub.Message) *pubsub.PublishResult
  • func (top *Topic) Subscriptions(ctx context.Context) *SubscriptionIterator
• type TopicIterator
  • func (it *TopicIterator) Next() (*Topic, error)

Examples

• TracingHandlerFunc

Functions

func TracingHandlerFunc

func TracingHandlerFunc(sensor instana.TracerLogger, pathTemplate string, handler http.HandlerFunc) http.HandlerFunc

TracingHandlerFunc wraps an HTTP handler that handles Google Cloud Pub/Sub push deliveries sent via POST requests. If a request uses any other method, the wrapper uses instana.TracingHandlerFunc() to trace it as a regular HTTP request.

Please note, that this wrapper consumes the request body in order to to extract the trace context from the message, thus the (net/http.Request).Body is a copy of received data.

Example

This example show how to instrument and HTTP handler that receives Google Cloud Pub/Sub messages via the push delivery method.

package main

import (
	"encoding/json"
	"fmt"
	"net/http"

	instana "github.com/instana/go-sensor"
	"github.com/instana/go-sensor/instrumentation/cloud.google.com/go/pubsub"
)

func main() {
	// Initialize sensor
	sensor := instana.NewSensor("pubsub-consumer")

	// Wrap your Pub/Sub message handler with pubsub.TracingHandlerFunc
	http.Handle("/", pubsub.TracingHandlerFunc(sensor, "/", func(w http.ResponseWriter, req *http.Request) {
		var delivery struct {
			Message struct {
				Data []byte `json:"data"`
			} `son:"message"`
		}

		if err := json.NewDecoder(req.Body).Decode(&delivery); err != nil {
			// handle the error
		}

		fmt.Printf("got %q", delivery.Message.Data)
	}))
}

Types

type Client

type Client struct {
	*pubsub.Client
	// contains filtered or unexported fields
}

Client is an instrumented wrapper for cloud.google.com/go/pubsub.Client that traces message reads and writes to and from Google Cloud Pub/Sub topics. It also and ensures Instana trace propagation across the Pub/Sub producers and consumers.

See https://pkg.go.dev/cloud.google.com/go/pubsub?tab=doc#Client for further details on wrapped type.

func NewClient

func NewClient(ctx context.Context, projectID string, sensor instana.TracerLogger, opts ...option.ClientOption) (*Client, error)

NewClient returns a new wrapped cloud.google.com/go/pubsub.Client that uses provided instana.Sensor to trace the publish/receive operations.

See https://pkg.go.dev/cloud.google.com/go/pubsub?tab=doc#NewClient for further details on wrapped method.

func (*Client) CreateSubscription

func (c *Client) CreateSubscription(ctx context.Context, id string, cfg pubsub.SubscriptionConfig) (*Subscription, error)

CreateSubscription calls CreateSubscription() of underlying Client and wraps the returned subscription.

See https://pkg.go.dev/cloud.google.com/go/pubsub?tab=doc#Client.CreateSubscription for further details on wrapped method.

func (*Client) CreateTopic

func (c *Client) CreateTopic(ctx context.Context, id string) (*Topic, error)

CreateTopic calls CreateTopic() of underlying Client and wraps the returned topic.

See https://pkg.go.dev/cloud.google.com/go/pubsub?tab=doc#Client.CreateTopic for further details on wrapped method.

func (*Client) CreateTopicWithConfig

func (c *Client) CreateTopicWithConfig(ctx context.Context, topicID string, tc *pubsub.TopicConfig) (*Topic, error)

CreateTopicWithConfig calls CreateTopicWithConfig() of underlying Client and wraps returned topic.

See https://pkg.go.dev/cloud.google.com/go/pubsub?tab=doc#Client.CreateTopicWithConfig for further details on wrapped method.

func (*Client) Subscription

func (c *Client) Subscription(id string) *Subscription

Subscription calls Subscription() of underlying Client and wraps the returned subscription.

See https://pkg.go.dev/cloud.google.com/go/pubsub?tab=doc#Client.Subscription for further details on wrapped method.

func (*Client) SubscriptionInProject

func (c *Client) SubscriptionInProject(id, projectID string) *Subscription

SubscriptionInProject calls SubscriptionInProject() of underlying Client and wraps the returned subscription.

See https://pkg.go.dev/cloud.google.com/go/pubsub?tab=doc#Client.SubscriptionInProject for further details on wrapped method.

func (*Client) Subscriptions

func (c *Client) Subscriptions(ctx context.Context) *SubscriptionIterator

Subscriptions calls Subscriptions() of underlying Client and wraps the returned subscription iterator.

See https://pkg.go.dev/cloud.google.com/go/pubsub?tab=doc#Client.Subscriptions for further details on wrapped method.

func (*Client) Topic

func (c *Client) Topic(id string) *Topic

Topic calls Topic() of underlying Client and wraps the returned topic.

See https://pkg.go.dev/cloud.google.com/go/pubsub?tab=doc#Client.Topic for further details on wrapped method.

func (*Client) TopicInProject

func (c *Client) TopicInProject(id, projectID string) *Topic

TopicInProject calls TopicInProject() of underlying Client and wraps the returned topic.

See https://pkg.go.dev/cloud.google.com/go/pubsub?tab=doc#Client.TopicInProject for further details on wrapped method.

func (*Client) Topics

func (c *Client) Topics(ctx context.Context) *TopicIterator

Topics calls Topics() of underlying Client and wraps the returned topic iterator.

See https://pkg.go.dev/cloud.google.com/go/pubsub?tab=doc#Client.Topics for further details on wrapped method.

type Message

type Message = pubsub.Message

Message is a type alias for cloud.google.com/go/pubsub.Message

type PublishResult

type PublishResult = pubsub.PublishResult

PublishResult is a type alias for cloud.google.com/go/pubsub.PublishResult

type Snapshot

type Snapshot = pubsub.Snapshot

Snapshot is a type alias for cloud.google.com/go/pubsub.Snapshot

type Subscription

type Subscription struct {
	*pubsub.Subscription
	// contains filtered or unexported fields
}

Subscription is an instrumented wrapper for cloud.google.com/go/pubsub.Subscription that traces Receive() calls and ensures Instana trace propagation across the Pub/Sub producers and consumers.

See https://pkg.go.dev/cloud.google.com/go/pubsub?tab=doc#Subscription for further details on wrapped type.

func (*Subscription) Receive

func (sub *Subscription) Receive(ctx context.Context, f func(context.Context, *pubsub.Message)) error

Receive wraps the Receive() call of the underlying cloud.google.com/go/pubsub.Subscription starting a new entry span using the provided instana.Sensor and ensuring the trace continuation.

See https://pkg.go.dev/cloud.google.com/go/pubsub?tab=doc#Subscription.Receive for further details on wrapped method.

type SubscriptionIterator

type SubscriptionIterator struct {
	*pubsub.SubscriptionIterator
	// contains filtered or unexported fields
}

SubscriptionIterator is a wrapper for cloud.google.com/go/pubsub.SubscriptionIterator that retrieves and instruments subscriptions in a project.

See https://pkg.go.dev/cloud.google.com/go/pubsub?tab=doc#SubscriptionIterator for further details on wrapped type.

func (*SubscriptionIterator) Next

func (it *SubscriptionIterator) Next() (*Subscription, error)

Next fetches the next subscription in project via the wrapped SubscriptionIterator and returns its wrapped version.

See https://pkg.go.dev/cloud.google.com/go/pubsub?tab=doc#SubscriptionIterator.Next for further details on wrapped method.

type Topic

type Topic struct {
	*pubsub.Topic
	// contains filtered or unexported fields
}

Topic is an instrumented wrapper for cloud.google.com/go/pubsub.Topic that traces Publish() calls and ensures Instana trace propagation across the Pub/Sub producers and consumers.

See https://pkg.go.dev/cloud.google.com/go/pubsub?tab=doc#Topic for further details on wrapped type.

func (*Topic) Publish

func (top *Topic) Publish(ctx context.Context, msg *pubsub.Message) *pubsub.PublishResult

Publish adds the trace context found in ctx to the message and publishes it to the wrapped topic. The exit span created for this operation will be finished only after the message was submitted to the server.

See https://pkg.go.dev/cloud.google.com/go/pubsub?tab=doc#Topic.Publish for further details on wrapped method.

func (*Topic) Subscriptions

func (top *Topic) Subscriptions(ctx context.Context) *SubscriptionIterator

Subscriptions calls Subscriptions() of underlying Topic and wraps the returned subscription iterator.

See https://pkg.go.dev/cloud.google.com/go/pubsub?tab=doc#Topic.Subscriptions for further details on wrapped method.

type TopicIterator

type TopicIterator struct {
	*pubsub.TopicIterator
	// contains filtered or unexported fields
}

TopicIterator is a wrapper for cloud.google.com/go/pubsub.TopicIterator that retrieves and instruments topics in a project.

See https://pkg.go.dev/cloud.google.com/go/pubsub?tab=doc#TopicIterator for further details on wrapped type.

func (*TopicIterator) Next

func (it *TopicIterator) Next() (*Topic, error)

Next fetches the next topic in project via the wrapped TopicIterator and returns its wrapped version.

See https://pkg.go.dev/cloud.google.com/go/pubsub?tab=doc#TopicIterator.Next for further details on wrapped method.