kafka exporter: seperate message conversion for otlp and jaeger (#3166)

* kafka exporter: seperate message conversion for otlp and jaeger

* fix test

* fix format

* Revert "fix format"

This reverts commit f02e8fb.

* Revert "fix test"

This reverts commit 23230cc.

* Revert "kafka exporter: seperate message conversion for otlp and jaeger"

This reverts commit 4184718.

* convert traces/metrics/logs to sarama.ProducerMessage directly

* fix linting error (impi) and changed CHANGELOG.md

Co-authored-by: Bogdan Drutu <[email protected]>

Author: Frefreak

CHANGELOG.md

@@ -6,6 +6,7 @@
 
 ## 🛑 Breaking changes 🛑
 
+- Change `Marshal` signatures in kafkaexporter's Marshalers to directly convert pdata to `sarama.ProducerMessage` (#3162)
 - Remove `tracetranslator.DetermineValueType`, only used internally by Zipkin (#3114)
 - Remove OpenCensus conventions, should not be used (#3113)
 - Remove Zipkin specific translation constants, move to internal (#3112)

exporter/kafkaexporter/factory_test.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"testing"
 
+	"github.com/Shopify/sarama"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/zap"
@@ -128,7 +129,7 @@ type customMarshaler struct {
 
 var _ TracesMarshaler = (*customMarshaler)(nil)
 
-func (c customMarshaler) Marshal(_ pdata.Traces) ([]Message, error) {
+func (c customMarshaler) Marshal(_ pdata.Traces, topic string) ([]*sarama.ProducerMessage, error) {
 	panic("implement me")
 }
 

exporter/kafkaexporter/jaeger_marshaler.go

@@ -17,6 +17,7 @@ package kafkaexporter
 import (
 	"bytes"
 
+	"github.com/Shopify/sarama"
 	"github.com/gogo/protobuf/jsonpb"
 	jaegerproto "github.com/jaegertracing/jaeger/model"
 
@@ -31,12 +32,13 @@ type jaegerMarshaler struct {
 
 var _ TracesMarshaler = (*jaegerMarshaler)(nil)
 
-func (j jaegerMarshaler) Marshal(traces pdata.Traces) ([]Message, error) {
+func (j jaegerMarshaler) Marshal(traces pdata.Traces, topic string) ([]*sarama.ProducerMessage, error) {
 	batches, err := jaegertranslator.InternalTracesToJaegerProto(traces)
 	if err != nil {
 		return nil, err
 	}
-	var messages []Message
+	var messages []*sarama.ProducerMessage
+
 	var errs []error
 	for _, batch := range batches {
 		for _, span := range batch.Spans {
@@ -48,7 +50,11 @@ func (j jaegerMarshaler) Marshal(traces pdata.Traces) ([]Message, error) {
 				continue
 			}
 			key := []byte(span.TraceID.String())
-			messages = append(messages, Message{Value: bts, Key: key})
+			messages = append(messages, &sarama.ProducerMessage{
+				Topic: topic,
+				Value: sarama.ByteEncoder(bts),
+				Key:   sarama.ByteEncoder(key),
+			})
 		}
 	}
 	return messages, consumererror.Combine(errs)

exporter/kafkaexporter/jaeger_marshaler_test.go

@@ -18,6 +18,7 @@ import (
 	"bytes"
 	"testing"
 
+	"github.com/Shopify/sarama"
 	"github.com/gogo/protobuf/jsonpb"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -50,14 +51,14 @@ func TestJaegerMarshaler(t *testing.T) {
 	tests := []struct {
 		unmarshaler TracesMarshaler
 		encoding    string
-		messages    []Message
+		messages    []*sarama.ProducerMessage
 	}{
 		{
 			unmarshaler: jaegerMarshaler{
 				marshaler: jaegerProtoSpanMarshaler{},
 			},
 			encoding: "jaeger_proto",
-			messages: []Message{{Value: jaegerProtoBytes, Key: messageKey}},
+			messages: []*sarama.ProducerMessage{{Topic: "topic", Value: sarama.ByteEncoder(jaegerProtoBytes), Key: sarama.ByteEncoder(messageKey)}},
 		},
 		{
 			unmarshaler: jaegerMarshaler{
@@ -66,12 +67,12 @@ func TestJaegerMarshaler(t *testing.T) {
 				},
 			},
 			encoding: "jaeger_json",
-			messages: []Message{{Value: jsonByteBuffer.Bytes(), Key: messageKey}},
+			messages: []*sarama.ProducerMessage{{Topic: "topic", Value: sarama.ByteEncoder(jsonByteBuffer.Bytes()), Key: sarama.ByteEncoder(messageKey)}},
 		},
 	}
 	for _, test := range tests {
 		t.Run(test.encoding, func(t *testing.T) {
-			messages, err := test.unmarshaler.Marshal(td)
+			messages, err := test.unmarshaler.Marshal(td, "topic")
 			require.NoError(t, err)
 			assert.Equal(t, test.messages, messages)
 			assert.Equal(t, test.encoding, test.unmarshaler.Encoding())
@@ -86,7 +87,7 @@ func TestJaegerMarshaler_error_covert_traceID(t *testing.T) {
 	td := pdata.NewTraces()
 	td.ResourceSpans().AppendEmpty().InstrumentationLibrarySpans().AppendEmpty().Spans().AppendEmpty()
 	// fails in zero traceID
-	messages, err := marshaler.Marshal(td)
+	messages, err := marshaler.Marshal(td, "topic")
 	require.Error(t, err)
 	assert.Nil(t, messages)
 }

exporter/kafkaexporter/kafka_exporter.go

@@ -37,11 +37,11 @@ type kafkaTracesProducer struct {
 }
 
 func (e *kafkaTracesProducer) traceDataPusher(_ context.Context, td pdata.Traces) error {
-	messages, err := e.marshaler.Marshal(td)
+	messages, err := e.marshaler.Marshal(td, e.topic)
 	if err != nil {
 		return consumererror.Permanent(err)
 	}
-	err = e.producer.SendMessages(producerMessages(messages, e.topic))
+	err = e.producer.SendMessages(messages)
 	if err != nil {
 		return err
 	}
@@ -61,11 +61,11 @@ type kafkaMetricsProducer struct {
 }
 
 func (e *kafkaMetricsProducer) metricsDataPusher(_ context.Context, md pdata.Metrics) error {
-	messages, err := e.marshaler.Marshal(md)
+	messages, err := e.marshaler.Marshal(md, e.topic)
 	if err != nil {
 		return consumererror.Permanent(err)
 	}
-	err = e.producer.SendMessages(producerMessages(messages, e.topic))
+	err = e.producer.SendMessages(messages)
 	if err != nil {
 		return err
 	}
@@ -85,11 +85,11 @@ type kafkaLogsProducer struct {
 }
 
 func (e *kafkaLogsProducer) logsDataPusher(_ context.Context, ld pdata.Logs) error {
-	messages, err := e.marshaler.Marshal(ld)
+	messages, err := e.marshaler.Marshal(ld, e.topic)
 	if err != nil {
 		return consumererror.Permanent(err)
 	}
-	err = e.producer.SendMessages(producerMessages(messages, e.topic))
+	err = e.producer.SendMessages(messages)
 	if err != nil {
 		return err
 	}
@@ -184,15 +184,3 @@ func newLogsExporter(config Config, params component.ExporterCreateParams, marsh
 	}, nil
 
 }
-
-func producerMessages(messages []Message, topic string) []*sarama.ProducerMessage {
-	producerMessages := make([]*sarama.ProducerMessage, len(messages))
-	for i := range messages {
-		producerMessages[i] = &sarama.ProducerMessage{
-			Topic: topic,
-			Value: sarama.ByteEncoder(messages[i].Value),
-			Key:   sarama.ByteEncoder(messages[i].Key),
-		}
-	}
-	return producerMessages
-}

exporter/kafkaexporter/kafka_exporter_test.go

@@ -270,7 +270,7 @@ type logsErrorMarshaler struct {
 	err error
 }
 
-func (e metricsErrorMarshaler) Marshal(_ pdata.Metrics) ([]Message, error) {
+func (e metricsErrorMarshaler) Marshal(_ pdata.Metrics, _ string) ([]*sarama.ProducerMessage, error) {
 	return nil, e.err
 }
 
@@ -280,15 +280,15 @@ func (e metricsErrorMarshaler) Encoding() string {
 
 var _ TracesMarshaler = (*tracesErrorMarshaler)(nil)
 
-func (e tracesErrorMarshaler) Marshal(_ pdata.Traces) ([]Message, error) {
+func (e tracesErrorMarshaler) Marshal(_ pdata.Traces, _ string) ([]*sarama.ProducerMessage, error) {
 	return nil, e.err
 }
 
 func (e tracesErrorMarshaler) Encoding() string {
 	panic("implement me")
 }
 
-func (e logsErrorMarshaler) Marshal(_ pdata.Logs) ([]Message, error) {
+func (e logsErrorMarshaler) Marshal(_ pdata.Logs, _ string) ([]*sarama.ProducerMessage, error) {
 	return nil, e.err
 }
 

exporter/kafkaexporter/marshaler.go

@@ -15,42 +15,38 @@
 package kafkaexporter
 
 import (
+	"github.com/Shopify/sarama"
+
 	"go.opentelemetry.io/collector/consumer/pdata"
 )
 
 // TracesMarshaler marshals traces into Message array.
 type TracesMarshaler interface {
-	// Marshal serializes spans into Messages
-	Marshal(traces pdata.Traces) ([]Message, error)
+	// Marshal serializes spans into sarama's ProducerMessages
+	Marshal(traces pdata.Traces, topic string) ([]*sarama.ProducerMessage, error)
 
 	// Encoding returns encoding name
 	Encoding() string
 }
 
 // MetricsMarshaler marshals metrics into Message array
 type MetricsMarshaler interface {
-	// Marshal serializes metrics into Messages
-	Marshal(metrics pdata.Metrics) ([]Message, error)
+	// Marshal serializes metrics into sarama's ProducerMessages
+	Marshal(metrics pdata.Metrics, topic string) ([]*sarama.ProducerMessage, error)
 
 	// Encoding returns encoding name
 	Encoding() string
 }
 
 // LogsMarshaler marshals logs into Message array
 type LogsMarshaler interface {
-	// Marshal serializes logs into Messages
-	Marshal(logs pdata.Logs) ([]Message, error)
+	// Marshal serializes logs into sarama's ProducerMessages
+	Marshal(logs pdata.Logs, topic string) ([]*sarama.ProducerMessage, error)
 
 	// Encoding returns encoding name
 	Encoding() string
 }
 
-// Message encapsulates Kafka's message payload.
-type Message struct {
-	Value []byte
-	Key   []byte
-}
-
 // tracesMarshalers returns map of supported encodings with TracesMarshaler.
 func tracesMarshalers() map[string]TracesMarshaler {
 	otlppb := &otlpTracesPbMarshaler{}

exporter/kafkaexporter/otlp_marshaler.go

@@ -15,6 +15,8 @@
 package kafkaexporter
 
 import (
+	"github.com/Shopify/sarama"
+
 	"go.opentelemetry.io/collector/consumer/pdata"
 )
 
@@ -28,12 +30,17 @@ func (m *otlpTracesPbMarshaler) Encoding() string {
 	return defaultEncoding
 }
 
-func (m *otlpTracesPbMarshaler) Marshal(td pdata.Traces) ([]Message, error) {
+func (m *otlpTracesPbMarshaler) Marshal(td pdata.Traces, topic string) ([]*sarama.ProducerMessage, error) {
 	bts, err := td.ToOtlpProtoBytes()
 	if err != nil {
 		return nil, err
 	}
-	return []Message{{Value: bts}}, nil
+	return []*sarama.ProducerMessage{
+		{
+			Topic: topic,
+			Value: sarama.ByteEncoder(bts),
+		},
+	}, nil
 }
 
 type otlpMetricsPbMarshaler struct {
@@ -43,12 +50,17 @@ func (m *otlpMetricsPbMarshaler) Encoding() string {
 	return defaultEncoding
 }
 
-func (m *otlpMetricsPbMarshaler) Marshal(md pdata.Metrics) ([]Message, error) {
+func (m *otlpMetricsPbMarshaler) Marshal(md pdata.Metrics, topic string) ([]*sarama.ProducerMessage, error) {
 	bts, err := md.ToOtlpProtoBytes()
 	if err != nil {
 		return nil, err
 	}
-	return []Message{{Value: bts}}, nil
+	return []*sarama.ProducerMessage{
+		{
+			Topic: topic,
+			Value: sarama.ByteEncoder(bts),
+		},
+	}, nil
 }
 
 type otlpLogsPbMarshaler struct {
@@ -58,10 +70,15 @@ func (m *otlpLogsPbMarshaler) Encoding() string {
 	return defaultEncoding
 }
 
-func (m *otlpLogsPbMarshaler) Marshal(ld pdata.Logs) ([]Message, error) {
+func (m *otlpLogsPbMarshaler) Marshal(ld pdata.Logs, topic string) ([]*sarama.ProducerMessage, error) {
 	bts, err := ld.ToOtlpProtoBytes()
 	if err != nil {
 		return nil, err
 	}
-	return []Message{{Value: bts}}, nil
+	return []*sarama.ProducerMessage{
+		{
+			Topic: topic,
+			Value: sarama.ByteEncoder(bts),
+		},
+	}, nil
 }

exporter/kafkaexporter/otlp_marshaler_test.go

@@ -28,10 +28,12 @@ func TestOTLPTracesPbMarshaler(t *testing.T) {
 	td := testdata.GenerateTracesTwoSpansSameResource()
 	m := otlpTracesPbMarshaler{}
 	assert.Equal(t, "otlp_proto", m.Encoding())
-	messages, err := m.Marshal(td)
+	messages, err := m.Marshal(td, "topic")
 	require.NoError(t, err)
 	require.Len(t, messages, 1)
-	extracted, err := pdata.TracesFromOtlpProtoBytes(messages[0].Value)
+	messageBytes, err := messages[0].Value.Encode()
+	require.NoError(t, err)
+	extracted, err := pdata.TracesFromOtlpProtoBytes(messageBytes)
 	require.NoError(t, err)
 	assert.EqualValues(t, td, extracted)
 }
@@ -40,10 +42,12 @@ func TestOTLPMetricsPbMarshaler(t *testing.T) {
 	md := testdata.GenerateMetricsTwoMetrics()
 	m := otlpMetricsPbMarshaler{}
 	assert.Equal(t, "otlp_proto", m.Encoding())
-	messages, err := m.Marshal(md)
+	messages, err := m.Marshal(md, "topic")
 	require.NoError(t, err)
 	require.Len(t, messages, 1)
-	extracted, err := pdata.MetricsFromOtlpProtoBytes(messages[0].Value)
+	messageBytes, err := messages[0].Value.Encode()
+	require.NoError(t, err)
+	extracted, err := pdata.MetricsFromOtlpProtoBytes(messageBytes)
 	require.NoError(t, err)
 	assert.EqualValues(t, md, extracted)
 }
@@ -52,10 +56,12 @@ func TestOTLPLogsPbMarshaler(t *testing.T) {
 	ld := testdata.GenerateLogsOneLogRecord()
 	m := otlpLogsPbMarshaler{}
 	assert.Equal(t, "otlp_proto", m.Encoding())
-	messages, err := m.Marshal(ld)
+	messages, err := m.Marshal(ld, "topic")
 	require.NoError(t, err)
 	require.Len(t, messages, 1)
-	extracted, err := pdata.LogsFromOtlpProtoBytes(messages[0].Value)
+	messageBytes, err := messages[0].Value.Encode()
+	require.NoError(t, err)
+	extracted, err := pdata.LogsFromOtlpProtoBytes(messageBytes)
 	require.NoError(t, err)
 	assert.EqualValues(t, ld, extracted)
 }