add zstd compression support to the sumo exporter

Signed-off-by: Sean Porter <[email protected]>

pkg/exporter/sumologicexporter/README.md

@@ -17,7 +17,7 @@ exporters:
     # if sumologicextension is not being used, the endpoint is required
     endpoint: <HTTP_Source_URL>
     # Compression encoding format, empty string means no compression, default = gzip
-    compress_encoding: {gzip, deflate, ""}
+    compress_encoding: {gzip, deflate, ztsd, ""}
     # max HTTP request body size in bytes before compression (if applied),
     # NOTE: this limit does not apply to data sent in otlp format,
     #   to limit size of otlp requests, please use the batch processor:

pkg/exporter/sumologicexporter/compress.go

@@ -21,6 +21,7 @@ import (
 
 	"github.com/klauspost/compress/flate"
 	"github.com/klauspost/compress/gzip"
+	"github.com/klauspost/compress/zstd"
 )
 
 type compressor struct {
@@ -49,6 +50,11 @@ func newCompressor(format CompressEncodingType) (compressor, error) {
 		if err != nil {
 			return compressor{}, err
 		}
+	case ZSTDCompression:
+		writer, err = zstd.NewWriter(io.Discard)
+		if err != nil {
+			return compressor{}, err
+		}
 	case NoCompression:
 		writer = nil
 	default:

pkg/exporter/sumologicexporter/compress_test.go

@@ -22,6 +22,7 @@ import (
 	"strings"
 	"testing"
 
+	"github.com/klauspost/compress/zstd"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -155,6 +156,31 @@ func TestCompressCloseError(t *testing.T) {
 	assert.EqualError(t, err, "close error")
 }
 
+func TestCompressZSTD(t *testing.T) {
+	const message = "This is an example log"
+
+	c, err := newCompressor(ZSTDCompression)
+	require.NoError(t, err)
+
+	body := strings.NewReader(message)
+
+	data, err := c.compress(body)
+	require.NoError(t, err)
+
+	assert.Equal(t, message, decodeZstd(t, data))
+}
+
+func decodeZstd(t *testing.T, data io.Reader) string {
+	r, err := zstd.NewReader(data)
+	require.NoError(t, err)
+
+	var buf []byte
+	buf, err = io.ReadAll(r)
+	require.NoError(t, err)
+
+	return string(buf)
+}
+
 func BenchmarkCompression(b *testing.B) {
 	const (
 		messageBlock       = "This is an example log"
@@ -188,6 +214,19 @@ func BenchmarkCompression(b *testing.B) {
 				return "", err
 			}
 
+			return string(buf), nil
+		case string(ZSTDCompression):
+			r, err := zstd.NewReader(data)
+			if err != nil {
+				return "", err
+			}
+
+			var buf []byte
+			buf, err = io.ReadAll(r)
+			if err != nil {
+				return "", err
+			}
+
 			return string(buf), nil
 
 		default:
@@ -204,6 +243,9 @@ func BenchmarkCompression(b *testing.B) {
 		{
 			encoding: string(GZIPCompression),
 		},
+		{
+			encoding: string(ZSTDCompression),
+		},
 	}
 
 	for _, tc := range testcases {

pkg/exporter/sumologicexporter/config.go

@@ -32,7 +32,7 @@ type Config struct {
 	exporterhelper.QueueSettings  `mapstructure:"sending_queue"`
 	exporterhelper.RetrySettings  `mapstructure:"retry_on_failure"`
 
-	// Compression encoding format, either empty string, gzip or deflate (default gzip)
+	// Compression encoding format, either empty string, gzip, deflate or zstd (default gzip)
 	// Empty string means no compression
 	CompressEncoding CompressEncodingType `mapstructure:"compress_encoding"`
 	// Max HTTP request body size in bytes before compression (if applied).
@@ -176,6 +176,7 @@ func (cet CompressEncodingType) Validate() error {
 	case GZIPCompression:
 	case NoCompression:
 	case DeflateCompression:
+	case ZSTDCompression:
 
 	default:
 		return fmt.Errorf("invalid compression encoding type: %v", cet)
@@ -205,6 +206,8 @@ const (
 	GZIPCompression CompressEncodingType = "gzip"
 	// DeflateCompression represents compress_encoding: deflate
 	DeflateCompression CompressEncodingType = "deflate"
+	// ZSTDCompression represents compress_encoding: zstd
+	ZSTDCompression CompressEncodingType = "zstd"
 	// NoCompression represents disabled compression
 	NoCompression CompressEncodingType = ""
 	// MetricsPipeline represents metrics pipeline