Add instrumentation for confluent-kafka-go

.github/dependabot.yml

@@ -22,6 +22,14 @@ updates:
     directory: "/instrumentation/database/sql/splunksql/test"
     schedule:
       interval: "daily"
+  - package-ecosystem: "gomod"
+    directory: "/instrumentation/github.com/confluentinc/confluent-kafka-go/kafka/splunkkafka"
+    schedule:
+      interval: "daily"
+  - package-ecosystem: "gomod"
+    directory: "/instrumentation/github.com/confluentinc/confluent-kafka-go/kafka/splunkkafka/test"
+    schedule:
+      interval: "daily"
   - package-ecosystem: "gomod"
     directory: "/instrumentation/github.com/go-sql-driver/mysql/splunkmysql"
     schedule:

CHANGELOG.md

@@ -28,6 +28,10 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 - Add the
   `github.com/signalfx/splunk-otel-go/instrumentation/github.com/jinzhu/gorm/splunkgorm`
   instrumentation for the `github.com/jinzhu/gorm` package. (#98)
+- Add the
+  `github.com/signalfx/splunk-otel-go/instrumentation/github.com/confluentinc/confluent-kafka-go/kafka/splunkkafka`
+  instrumentation for the `github.com/confluentinc/confluent-kafka-go/kafka`
+  package. (#100)
 
 ### Changed
 

Makefile

@@ -15,7 +15,7 @@
 SHELL := /bin/bash
 
 GO               = go
-TIMEOUT          = 15
+TIMEOUT          = 60
 PKGS             = ./...
 BUILD_DIR        = ./build
 TEST_RESULTS     = $(CURDIR)/test-results

README.md

@@ -102,6 +102,7 @@ Additional recommended Splunk specific instrumentations:
 
 - [`splunkgorm`](./instrumentation/github.com/jinzhu/gorm/splunkgorm)
 - [`splunkhttp`](./instrumentation/net/http/splunkhttp)
+- [`splunkkafka`](./instrumentation/github.com/confluentinc/confluent-kafka-go/kafka/splunkkafka)
 - [`splunkmysql`](./instrumentation/github.com/go-sql-driver/mysql/splunkmysql)
 - [`splunkpgx`](./instrumentation/github.com/jackc/pgx/splunkpgx)
 - [`splunkpq`](./instrumentation/github.com/lib/pq/splunkpq)

instrumentation/github.com/confluentinc/confluent-kafka-go/kafka/splunkkafka/README.md

@@ -0,0 +1,29 @@
+# Splunk instrumentation for `github.com/confluentinc/confluent-kafka-go/kafka/splunkkafka`
+
+This instrumentation is for the
+[github.com/confluentinc/confluent-kafka-go/kafka](https://github.com/confluentinc/confluent-kafka-go)
+package.
+
+## Compatibility
+
+The Producer will end spans when a delivery report is returned. Setting
+`"go.delivery.reports"` to `false` will disable the delivery reports and can
+result in an build up of un-ended spans. If delivery reports are disabled, an
+un-instrumented Producer should be used instead.
+
+This instrumentation was built to support
+[v1.7.0](https://github.com/confluentinc/confluent-kafka-go/releases/tag/v1.7.0)
+of github.com/confluentinc/confluent-kafka-go/kafka/splunkkafka. Similar to the
+instrumented package, librdkafka 1.6.0+ is required. This means you will need
+to use an environment that supports the [pre-built
+binaries](https://github.com/confluentinc/confluent-kafka-go#librdkafka), or
+[install](https://github.com/confluentinc/confluent-kafka-go#installing-librdkafka)
+the library manually. Important to note, similar to the instrumented package,
+**cgo is required** and **this instrumentation does not support the Windows
+operating system**.
+
+## Getting started
+
+The `NewConsumer` and `NewProducer` functions are provided as drop-in
+replacements of the equivalent from the `kafka` package. See [these
+examples](./example_test.go) for how to use these functions.

instrumentation/github.com/confluentinc/confluent-kafka-go/kafka/splunkkafka/carrier.go

@@ -0,0 +1,71 @@
+// Copyright Splunk Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//go:build cgo && (linux || darwin)
+// +build cgo
+// +build linux darwin
+
+package splunkkafka
+
+import (
+	"github.com/confluentinc/confluent-kafka-go/kafka"
+	"go.opentelemetry.io/otel/propagation"
+)
+
+// textMapCarrier wraps a kafka.Message so it can be used used by a
+// TextMapPropagator to propagate tracing context.
+type textMapCarrier struct {
+	msg *kafka.Message
+}
+
+var _ propagation.TextMapCarrier = (*textMapCarrier)(nil)
+
+// NewMessageCarrier returns a TextMapCarrier that will encode and decode
+// tracing information to and from the passed message.
+func NewMessageCarrier(message *kafka.Message) propagation.TextMapCarrier {
+	return &textMapCarrier{message}
+}
+
+// Get returns the value associated with the passed key.
+func (c *textMapCarrier) Get(key string) string {
+	for _, h := range c.msg.Headers {
+		if h.Key == key {
+			return string(h.Value)
+		}
+	}
+	return ""
+}
+
+// Set stores the key-value pair.
+func (c *textMapCarrier) Set(key, value string) {
+	// Ensure the uniqueness of the key.
+	for i := len(c.msg.Headers) - 1; i >= 0; i-- {
+		if c.msg.Headers[i].Key == key {
+			c.msg.Headers = append(c.msg.Headers[:i], c.msg.Headers[i+1:]...)
+		}
+	}
+	c.msg.Headers = append(c.msg.Headers, kafka.Header{
+		Key:   key,
+		Value: []byte(value),
+	})
+}
+
+// Keys lists the keys stored in this carrier.
+func (c *textMapCarrier) Keys() []string {
+	out := make([]string, len(c.msg.Headers))
+	for i, h := range c.msg.Headers {
+		out[i] = h.Key
+	}
+	return out
+}

instrumentation/github.com/confluentinc/confluent-kafka-go/kafka/splunkkafka/carrier_test.go

@@ -0,0 +1,91 @@
+// Copyright Splunk Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//go:build cgo && (linux || darwin)
+// +build cgo
+// +build linux darwin
+
+package splunkkafka_test
+
+import (
+	"testing"
+
+	"github.com/confluentinc/confluent-kafka-go/kafka"
+	"github.com/signalfx/splunk-otel-go/instrumentation/github.com/confluentinc/confluent-kafka-go/kafka/splunkkafka"
+	"github.com/stretchr/testify/assert"
+)
+
+const (
+	k, v = "key", "value"
+)
+
+func TestCarrierGet(t *testing.T) {
+	msg := &kafka.Message{
+		Headers: []kafka.Header{
+			{Key: k, Value: []byte(v)},
+		},
+	}
+	carrier := splunkkafka.NewMessageCarrier(msg)
+	assert.Equal(t, v, carrier.Get(k))
+}
+
+func TestCarrierGetEmpty(t *testing.T) {
+	msg := &kafka.Message{}
+	carrier := splunkkafka.NewMessageCarrier(msg)
+	assert.Equal(t, "", carrier.Get("key"))
+}
+
+func TestCarrierSet(t *testing.T) {
+	msg := &kafka.Message{}
+	carrier := splunkkafka.NewMessageCarrier(msg)
+	carrier.Set(k, v)
+	var got string
+	for _, h := range msg.Headers {
+		if h.Key == k {
+			got = string(h.Value)
+		}
+	}
+	assert.Equal(t, v, got)
+}
+
+func TestCarrierSetOverwrites(t *testing.T) {
+	msg := &kafka.Message{
+		Headers: []kafka.Header{
+			{Key: k, Value: []byte("not value")},
+			{Key: k, Value: []byte("also not value")},
+		},
+	}
+	carrier := splunkkafka.NewMessageCarrier(msg)
+	carrier.Set(k, v)
+	var got string
+	for _, h := range msg.Headers {
+		if h.Key == k {
+			got = string(h.Value)
+		}
+	}
+	assert.Equal(t, v, got)
+}
+
+func TestCarrierKeys(t *testing.T) {
+	keys := []string{"one", "two", "three"}
+	msg := &kafka.Message{
+		Headers: []kafka.Header{
+			{Key: keys[0], Value: []byte("")},
+			{Key: keys[1], Value: []byte("")},
+			{Key: keys[2], Value: []byte("")},
+		},
+	}
+	carrier := splunkkafka.NewMessageCarrier(msg)
+	assert.Equal(t, keys, carrier.Keys())
+}

instrumentation/github.com/confluentinc/confluent-kafka-go/kafka/splunkkafka/config.go

@@ -0,0 +1,100 @@
+// Copyright Splunk Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//go:build cgo && (linux || darwin)
+// +build cgo
+// +build linux darwin
+
+package splunkkafka
+
+import (
+	splunkotel "github.com/signalfx/splunk-otel-go"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/propagation"
+	semconv "go.opentelemetry.io/otel/semconv/v1.4.0"
+	"go.opentelemetry.io/otel/trace"
+)
+
+// instrumentationName is the instrumentation library identifier for a Tracer.
+const instrumentationName = "github.com/signalfx/splunk-otel-go/instrumentation/github.com/confluentinc/confluent-kafka-go/kafka/splunkkafka"
+
+// config contains tracing configuration options.
+type config struct {
+	Tracer     trace.Tracer
+	Propagator propagation.TextMapPropagator
+	Attributes []attribute.KeyValue
+}
+
+func newConfig(options ...Option) config {
+	c := config{
+		Attributes: []attribute.KeyValue{
+			semconv.MessagingSystemKey.String("kafka"),
+		},
+	}
+	for _, o := range options {
+		if o != nil {
+			o.apply(&c)
+		}
+	}
+	if c.Tracer == nil {
+		c.Tracer = otel.Tracer(
+			instrumentationName,
+			trace.WithInstrumentationVersion(splunkotel.Version()),
+		)
+	}
+	if c.Propagator == nil {
+		c.Propagator = otel.GetTextMapPropagator()
+	}
+	return c
+}
+
+// Option applies options to a tracing configuration.
+type Option interface {
+	apply(*config)
+}
+
+type optionFunc func(*config)
+
+func (o optionFunc) apply(c *config) {
+	o(c)
+}
+
+// WithTracerProvider returns an Option that sets the TracerProvider used with
+// this instrumentation library.
+func WithTracerProvider(tp trace.TracerProvider) Option {
+	return optionFunc(func(c *config) {
+		c.Tracer = tp.Tracer(
+			instrumentationName,
+			trace.WithInstrumentationVersion(splunkotel.Version()),
+		)
+	})
+}
+
+// WithPropagator specifies the TextMapPropagator to use when extracting and
+// injecting cross-cutting concerns. If none is specified, the global
+// TextMapPropagator will be used.
+func WithPropagator(propagator propagation.TextMapPropagator) Option {
+	return optionFunc(func(cfg *config) {
+		cfg.Propagator = propagator
+	})
+}
+
+// WithAttributes returns an Option that appends attr to the attributes set
+// for every span created with this instrumentation library.
+func WithAttributes(attr []attribute.KeyValue) Option {
+	return optionFunc(func(c *config) {
+		c.Attributes = append(c.Attributes, attr...)
+	})
+}