instasarama

This example demonstrates how to instrument an async Kafka producer using instasarama. Error handling is omitted for brevity.

package main

import (
	"github.com/IBM/sarama"
	instana "github.com/instana/go-sensor"
	"github.com/instana/go-sensor/instrumentation/instasarama"
	"github.com/opentracing/opentracing-go/ext"
)

func main() {
	sensor := instana.NewSensor("my-service")
	brokers := []string{"localhost:9092"}

	config := sarama.NewConfig()
	// enable the use record headers added in kafka v0.11.0 and used to propagate
	// trace context
	config.Version = sarama.V0_11_0_0

	// create a new instrumented instance of sarama.SyncProducer
	producer, _ := instasarama.NewAsyncProducer(brokers, config, sensor)

	// start a new entry span
	sp := sensor.Tracer().StartSpan("my-producing-method")
	ext.SpanKind.Set(sp, "entry")

	msg := &sarama.ProducerMessage{
		// ...
	}

	// inject the span before passing the message to producer
	producer.Input() <- instasarama.ProducerMessageWithSpan(msg, sp)
}

Output:

This example demonstrates how to instrument a Kafka consumer using instasarama and extract the trace context to ensure continuation. Error handling is omitted for brevity.

// (c) Copyright IBM Corp. 2023

//go:build go1.17
// +build go1.17

package main

import (
	"fmt"

	"github.com/IBM/sarama"
	instana "github.com/instana/go-sensor"
	"github.com/instana/go-sensor/instrumentation/instasarama"
	"github.com/opentracing/opentracing-go"
)

// This example demonstrates how to instrument a Kafka consumer using instasarama
// and extract the trace context to ensure continuation. Error handling is omitted for brevity.
func main() {
	sensor := instana.NewSensor("my-service")
	brokers := []string{"localhost:9092"}

	conf := sarama.NewConfig()
	conf.Version = sarama.V0_11_0_0

	// create a new instrumented instance of sarama.Consumer
	consumer, _ := instasarama.NewConsumer(brokers, conf, sensor)

	c, _ := consumer.ConsumePartition("test-topic-1", 0, sarama.OffsetNewest)
	defer c.Close()

	for msg := range c.Messages() {
		fmt.Println("Got messagge", msg)
		processMessage(msg, sensor)
	}
}

func processMessage(msg *sarama.ConsumerMessage, sensor instana.TracerLogger) {
	// extract trace context and start a new span
	parentCtx, _ := instasarama.SpanContextFromConsumerMessage(msg, sensor)

	sp := sensor.Tracer().StartSpan("process-message", opentracing.ChildOf(parentCtx))
	defer sp.Finish()

	// process message
}

Output:

This example demonstrates how to instrument a Kafka consumer group using instasarama and extract the trace context to ensure continuation. Error handling is omitted for brevity.

// (c) Copyright IBM Corp. 2023

//go:build go1.17
// +build go1.17

package main

import (
	"context"

	"github.com/IBM/sarama"
	instana "github.com/instana/go-sensor"
	"github.com/instana/go-sensor/instrumentation/instasarama"
	"github.com/opentracing/opentracing-go"
)

// This example demonstrates how to instrument a Kafka consumer group using instasarama
// and extract the trace context to ensure continuation. Error handling is omitted for brevity.
func main() {
	sensor := instana.NewSensor("my-service")
	brokers := []string{"localhost:9092"}
	topics := []string{"records", "more-records"}

	conf := sarama.NewConfig()
	conf.Version = sarama.V0_11_0_0

	client, _ := sarama.NewConsumerGroup(brokers, "my-service-consumers", conf)
	defer client.Close()

	ctx := context.Background()
	consumer := instasarama.WrapConsumerGroupHandler(&Consumer{sensor: sensor}, sensor)

	// start consuming
	for {
		_ = client.Consume(ctx, topics, consumer)

		// ...
	}
}

type Consumer struct {
	sensor instana.TracerLogger
}

func (*Consumer) Setup(sarama.ConsumerGroupSession) error {
	// setup consumer
	return nil
}

func (*Consumer) Cleanup(sarama.ConsumerGroupSession) error {
	// cleanup consumer
	return nil
}

func (c *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
	for msg := range claim.Messages() {
		c.processMessage(msg)
		session.MarkMessage(msg, "")
	}

	return nil
}

func (c *Consumer) processMessage(msg *sarama.ConsumerMessage) {
	// extract trace context and start a new span
	parentCtx, _ := instasarama.SpanContextFromConsumerMessage(msg, c.sensor)

	sp := c.sensor.Tracer().StartSpan("process-message", opentracing.ChildOf(parentCtx))
	defer sp.Finish()

	// process message
}

Output:

This example demonstrates how to instrument a sync Kafka producer using instasarama. Error handling is omitted for brevity.

package main

import (
	"github.com/IBM/sarama"
	instana "github.com/instana/go-sensor"
	"github.com/instana/go-sensor/instrumentation/instasarama"
	"github.com/opentracing/opentracing-go/ext"
)

func main() {
	sensor := instana.NewSensor("my-service")
	brokers := []string{"localhost:9092"}

	config := sarama.NewConfig()
	// sarama requires Producer.Return.Successes to be set for sync producers
	config.Producer.Return.Successes = true
	// enable the use record headers added in kafka v0.11.0 and used to propagate
	// trace context
	config.Version = sarama.V0_11_0_0

	// create a new instrumented instance of sarama.SyncProducer
	producer, _ := instasarama.NewSyncProducer(brokers, config, sensor)

	// start a new entry span
	sp := sensor.Tracer().StartSpan("my-producing-method")
	ext.SpanKind.Set(sp, "entry")

	msg := &sarama.ProducerMessage{
		Topic:  "test-topic-1",
		Offset: sarama.OffsetNewest,
		Value:  sarama.StringEncoder("I am a message"),
		// ...
	}

	// inject the span before passing the message to producer
	msg = instasarama.ProducerMessageWithSpan(msg, sp)

	// pass it to the producer
	producer.SendMessage(msg)
}

Output: