Add support for compression options in Kafka exporter

exporter/kafkaexporter/README.md

@@ -61,6 +61,13 @@ The following settings can be optionally configured:
   User should calculate this as `num_seconds * requests_per_second` where:
     - `num_seconds` is the number of seconds to buffer in case of a backend outage
     - `requests_per_second` is the average number of requests per seconds.
+- `compression` (default = none): The compression codec to compress the messages sent to Kafka. Four compression methods are supported: `gzip`, `snappy`, `lz4` and `zstd`.
+- `compression_level` (default = -1000): The corresponding compression level which measures the compression degree.
+  Lower compression level would lead to fast compression as well as relatively big file size. The valid ranges for each compression method are:
+  - `gzip`: `1 - 9`
+  - `snappy`: Snappy does not support compression level.
+  - `lz4`: `1 - 17`
+  - `zstd`: `1 - 22`
 
 Example configuration:
 

exporter/kafkaexporter/compression.go

@@ -0,0 +1,41 @@
+// Copyright  The OpenTelemetry Authors
+//
+// 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.
+
+package kafkaexporter
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/Shopify/sarama"
+)
+
+func configureCompression(compression string, saramaConfig *sarama.Config) error {
+	switch strings.ToLower(compression) {
+	case "none":
+		saramaConfig.Producer.Compression = sarama.CompressionNone
+	case "gzip":
+		saramaConfig.Producer.Compression = sarama.CompressionGZIP
+	case "snappy":
+		saramaConfig.Producer.Compression = sarama.CompressionSnappy
+	case "lz4":
+		saramaConfig.Producer.Compression = sarama.CompressionLZ4
+	case "zstd":
+		saramaConfig.Producer.Compression = sarama.CompressionZSTD
+	default:
+		return fmt.Errorf("invalid compression %q: can be one of \"none\" , \"gzip\", \"snappy\", \"lz4\" or \"zstd\"", compression)
+	}
+
+	return nil
+}

exporter/kafkaexporter/compression_test.go

@@ -0,0 +1,65 @@
+// Copyright  The OpenTelemetry Authors
+//
+// 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.
+
+package kafkaexporter
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/Shopify/sarama"
+)
+
+func TestConfigureCompression(t *testing.T) {
+	saramaSample := &sarama.Config{}
+
+	tests := []struct {
+		compression  string
+		saramaConfig *sarama.Config
+		wantErr      string
+	}{
+		{
+			compression:  "none",
+			saramaConfig: saramaSample,
+		},
+		{
+			compression:  "gzip",
+			saramaConfig: saramaSample,
+		},
+		{
+			compression:  "snappy",
+			saramaConfig: saramaSample,
+		},
+		{
+			compression:  "lz4",
+			saramaConfig: saramaSample,
+		},
+		{
+			compression:  "zstd",
+			saramaConfig: saramaSample,
+		},
+		{
+			compression:  "somerandomcompression",
+			saramaConfig: saramaSample,
+			wantErr: "invalid compression \"somerandomcompression\": can be one of \"none\" , \"gzip\", \"snappy\", \"lz4\" or \"zstd\"",
+		},
+	}
+	for _, tt := range tests {
+		t.Run("", func(t *testing.T) {
+			if err := configureCompression(tt.compression, tt.saramaConfig); (err != nil) && strings.Compare(err.Error(), tt.wantErr) != 0 {
+				t.Errorf("ConfigureCompression() error = %v, wantErr %v", err, tt.wantErr)
+			}
+		})
+	}
+}

exporter/kafkaexporter/config.go

@@ -44,6 +44,12 @@ type Config struct {
 
 	// Authentication defines used authentication mechanism.
 	Authentication Authentication `mapstructure:"auth"`
+
+	// CompressionCodec is the chosen compression method.
+	CompressionCodec string `mapstructure:"compression"`
+
+	// CompressionLevel is the chosen compression level, if applicable.
+	CompressionLevel int `mapstructure:"compression_level"`
 }
 
 // Metadata defines configuration for retrieving metadata from the broker.

exporter/kafkaexporter/config_test.go

@@ -71,5 +71,7 @@ func TestLoadConfig(t *testing.T) {
 				Backoff: defaultMetadataRetryBackoff,
 			},
 		},
+		CompressionCodec: "gzip",
+		CompressionLevel: 4,
 	}, c)
 }

exporter/kafkaexporter/factory.go

@@ -18,6 +18,8 @@ import (
 	"context"
 	"time"
 
+	"github.com/Shopify/sarama"
+
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config"
 	"go.opentelemetry.io/collector/consumer"
@@ -37,6 +39,10 @@ const (
 	defaultMetadataRetryBackoff = time.Millisecond * 250
 	// default from sarama.NewConfig()
 	defaultMetadataFull = true
+	// By default, don't use compression.
+	defaultCompression = "none"
+	// default from sarama.NewConfig()
+	defaultCompressionLevel = sarama.CompressionLevelDefault
 )
 
 // FactoryOption applies changes to kafkaExporterFactory.
@@ -87,6 +93,8 @@ func createDefaultConfig() config.Exporter {
 				Backoff: defaultMetadataRetryBackoff,
 			},
 		},
+		CompressionCodec: defaultCompression,
+		CompressionLevel: defaultCompressionLevel,
 	}
 }
 

exporter/kafkaexporter/kafka_exporter.go

@@ -112,6 +112,10 @@ func newSaramaProducer(config Config) (sarama.SyncProducer, error) {
 	c.Metadata.Full = config.Metadata.Full
 	c.Metadata.Retry.Max = config.Metadata.Retry.Max
 	c.Metadata.Retry.Backoff = config.Metadata.Retry.Backoff
+	if err := configureCompression(config.CompressionCodec, c); err != nil {
+		return nil, err
+	}
+	c.Producer.CompressionLevel = config.CompressionLevel
 	if config.ProtocolVersion != "" {
 		version, err := sarama.ParseKafkaVersion(config.ProtocolVersion)
 		if err != nil {

exporter/kafkaexporter/kafka_exporter_test.go

@@ -101,6 +101,8 @@ func TestNewExporter_err_auth_type(t *testing.T) {
 		Metadata: Metadata{
 			Full: false,
 		},
+		CompressionCodec: defaultCompression,
+		CompressionLevel: defaultCompressionLevel,
 	}
 	texp, err := newTracesExporter(c, component.ExporterCreateParams{Logger: zap.NewNop()}, tracesMarshalers())
 	assert.Error(t, err)

exporter/kafkaexporter/testdata/config.yaml

@@ -22,6 +22,8 @@ exporters:
       initial_interval: 10s
       max_interval: 60s
       max_elapsed_time: 10m
+    compression: gzip
+    compression_level: 4
 
 processors:
   nop: