contrib/IBM/sarama: adds trace for IBM/sarama (#2142)

Co-authored-by: Zarir Hamza <[email protected]>

contrib/IBM/sarama.v1/example_test.go

@@ -0,0 +1,84 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2016 Datadog, Inc.
+
+package sarama_test
+
+import (
+ "log"
+
+ saramatrace "gopkg.in/DataDog/dd-trace-go.v1/contrib/IBM/sarama.v1"
+ "gopkg.in/DataDog/dd-trace-go.v1/ddtrace/tracer"
+
+ "github.com/IBM/sarama"
+)
+
+func Example_asyncProducer() {
+ cfg := sarama.NewConfig()
+ cfg.Version = sarama.V0_11_0_0 // minimum version that supports headers which are required for tracing
+
+ producer, err := sarama.NewAsyncProducer([]string{"localhost:9092"}, cfg)
+ if err != nil {
+ panic(err)
+ }
+ defer producer.Close()
+
+ producer = saramatrace.WrapAsyncProducer(cfg, producer)
+
+ msg := &sarama.ProducerMessage{
+ Topic: "some-topic",
+ Value: sarama.StringEncoder("Hello World"),
+ }
+ producer.Input() <- msg
+}
+
+func Example_syncProducer() {
+ cfg := sarama.NewConfig()
+ cfg.Producer.Return.Successes = true
+
+ producer, err := sarama.NewSyncProducer([]string{"localhost:9092"}, cfg)
+ if err != nil {
+ panic(err)
+ }
+ defer producer.Close()
+
+ producer = saramatrace.WrapSyncProducer(cfg, producer)
+
+ msg := &sarama.ProducerMessage{
+ Topic: "some-topic",
+ Value: sarama.StringEncoder("Hello World"),
+ }
+ _, _, err = producer.SendMessage(msg)
+ if err != nil {
+ panic(err)
+ }
+}
+
+func Example_consumer() {
+ consumer, err := sarama.NewConsumer([]string{"localhost:9092"}, nil)
+ if err != nil {
+ panic(err)
+ }
+ defer consumer.Close()
+
+ consumer = saramatrace.WrapConsumer(consumer)
+
+ partitionConsumer, err := consumer.ConsumePartition("some-topic", 0, sarama.OffsetNewest)
+ if err != nil {
+ panic(err)
+ }
+ defer partitionConsumer.Close()
+
+ consumed := 0
+ for msg := range partitionConsumer.Messages() {
+ // if you want to use the kafka message as a parent span:
+ if spanctx, err := tracer.Extract(saramatrace.NewConsumerMessageCarrier(msg)); err == nil {
+ // you can create a span using ChildOf(spanctx)
+ _ = spanctx
+ }
+
+ log.Printf("Consumed message offset %d\n", msg.Offset)
+ consumed++
+ }
+}

contrib/IBM/sarama.v1/headers.go

@@ -0,0 +1,96 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2016 Datadog, Inc.
+
+package sarama
+
+import (
+ "gopkg.in/DataDog/dd-trace-go.v1/ddtrace/tracer"
+
+ "github.com/IBM/sarama"
+)
+
+// A ProducerMessageCarrier injects and extracts traces from a sarama.ProducerMessage.
+type ProducerMessageCarrier struct {
+ msg *sarama.ProducerMessage
+}
+
+var _ interface {
+ tracer.TextMapReader
+ tracer.TextMapWriter
+} = (*ProducerMessageCarrier)(nil)
+
+// ForeachKey iterates over every header.
+func (c ProducerMessageCarrier) ForeachKey(handler func(key, val string) error) error {
+ for _, h := range c.msg.Headers {
+ err := handler(string(h.Key), string(h.Value))
+ if err != nil {
+ return err
+ }
+ }
+ return nil
+}
+
+// Set sets a header.
+func (c ProducerMessageCarrier) Set(key, val string) {
+ // ensure uniqueness of keys
+ for i := 0; i < len(c.msg.Headers); i++ {
+ if string(c.msg.Headers[i].Key) == key {
+ c.msg.Headers = append(c.msg.Headers[:i], c.msg.Headers[i+1:]...)
+ i--
+ }
+ }
+ c.msg.Headers = append(c.msg.Headers, sarama.RecordHeader{
+ Key: []byte(key),
+ Value: []byte(val),
+ })
+}
+
+// NewProducerMessageCarrier creates a new ProducerMessageCarrier.
+func NewProducerMessageCarrier(msg *sarama.ProducerMessage) ProducerMessageCarrier {
+ return ProducerMessageCarrier{msg}
+}
+
+// A ConsumerMessageCarrier injects and extracts traces from a sarama.ConsumerMessage.
+type ConsumerMessageCarrier struct {
+ msg *sarama.ConsumerMessage
+}
+
+var _ interface {
+ tracer.TextMapReader
+ tracer.TextMapWriter
+} = (*ConsumerMessageCarrier)(nil)
+
+// NewConsumerMessageCarrier creates a new ConsumerMessageCarrier.
+func NewConsumerMessageCarrier(msg *sarama.ConsumerMessage) ConsumerMessageCarrier {
+ return ConsumerMessageCarrier{msg}
+}
+
+// ForeachKey iterates over every header.
+func (c ConsumerMessageCarrier) ForeachKey(handler func(key, val string) error) error {
+ for _, h := range c.msg.Headers {
+ if h != nil {
+ err := handler(string(h.Key), string(h.Value))
+ if err != nil {
+ return err
+ }
+ }
+ }
+ return nil
+}
+
+// Set sets a header.
+func (c ConsumerMessageCarrier) Set(key, val string) {
+ // ensure uniqueness of keys
+ for i := 0; i < len(c.msg.Headers); i++ {
+ if c.msg.Headers[i] != nil && string(c.msg.Headers[i].Key) == key {
+ c.msg.Headers = append(c.msg.Headers[:i], c.msg.Headers[i+1:]...)
+ i--
+ }
+ }
+ c.msg.Headers = append(c.msg.Headers, &sarama.RecordHeader{
+ Key: []byte(key),
+ Value: []byte(val),
+ })
+}

contrib/IBM/sarama.v1/option.go

@@ -0,0 +1,75 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2016 Datadog, Inc.
+
+package sarama
+
+import (
+ "math"
+
+ "gopkg.in/DataDog/dd-trace-go.v1/internal"
+ "gopkg.in/DataDog/dd-trace-go.v1/internal/namingschema"
+)
+
+const defaultServiceName = "kafka"
+
+type config struct {
+ consumerServiceName string
+ producerServiceName string
+ consumerSpanName string
+ producerSpanName string
+ analyticsRate float64
+}
+
+func defaults(cfg *config) {
+ cfg.consumerServiceName = namingschema.NewDefaultServiceName(defaultServiceName).GetName()
+ cfg.producerServiceName = namingschema.NewDefaultServiceName(
+ defaultServiceName,
+ namingschema.WithOverrideV0(defaultServiceName),
+ ).GetName()
+
+ cfg.consumerSpanName = namingschema.NewKafkaInboundOp().GetName()
+ cfg.producerSpanName = namingschema.NewKafkaOutboundOp().GetName()
+
+ // cfg.analyticsRate = globalconfig.AnalyticsRate()
+ if internal.BoolEnv("DD_TRACE_SARAMA_ANALYTICS_ENABLED", false) {
+ cfg.analyticsRate = 1.0
+ } else {
+ cfg.analyticsRate = math.NaN()
+ }
+}
+
+// An Option is used to customize the config for the sarama tracer.
+type Option func(cfg *config)
+
+// WithServiceName sets the given service name for the intercepted client.
+func WithServiceName(name string) Option {
+ return func(cfg *config) {
+ cfg.consumerServiceName = name
+ cfg.producerServiceName = name
+ }
+}
+
+// WithAnalytics enables Trace Analytics for all started spans.
+func WithAnalytics(on bool) Option {
+ return func(cfg *config) {
+ if on {
+ cfg.analyticsRate = 1.0
+ } else {
+ cfg.analyticsRate = math.NaN()
+ }
+ }
+}
+
+// WithAnalyticsRate sets the sampling rate for Trace Analytics events
+// correlated to started spans.
+func WithAnalyticsRate(rate float64) Option {
+ return func(cfg *config) {
+ if rate >= 0.0 && rate <= 1.0 {
+ cfg.analyticsRate = rate
+ } else {
+ cfg.analyticsRate = math.NaN()
+ }
+ }
+}