feat(output-http): Add more compression options (#341)

deploy/local/docker-compose/xatu-server.yaml

@@ -45,5 +45,5 @@ services:
         batchTimeout: 3s
         exportTimeout: 30s
         maxExportBatchSize: 5000
-        compression: none
+        compression: zstd
         keepAlive: true

pkg/output/http/compression.go

@@ -1,8 +1,111 @@
 package http
 
+import (
+	"bytes"
+	"compress/gzip"
+	"compress/zlib"
+	"net/http"
+
+	"github.com/golang/snappy"
+	"github.com/klauspost/compress/zstd"
+)
+
 type CompressionStrategy string
 
 var (
-	CompressionStrategyNone CompressionStrategy = "none"
-	CompressionStrategyGzip CompressionStrategy = "gzip"
+	CompressionStrategyNone   CompressionStrategy = "none"
+	CompressionStrategyGzip   CompressionStrategy = "gzip"
+	CompressionStrategyZstd   CompressionStrategy = "zstd"
+	CompressionStrategyZlib   CompressionStrategy = "zlib"
+	CompressionStrategySnappy CompressionStrategy = "snappy"
 )
+
+type Compressor struct {
+	Strategy CompressionStrategy
+}
+
+func (c *Compressor) Compress(in *bytes.Buffer) (*bytes.Buffer, error) {
+	switch c.Strategy {
+	case CompressionStrategyGzip:
+		return c.gzipCompress(in)
+	case CompressionStrategyZstd:
+		return c.zstdCompress(in)
+	case CompressionStrategyZlib:
+		return c.zlibCompress(in)
+	case CompressionStrategySnappy:
+		return c.snappyCompress(in)
+	default:
+		return in, nil
+	}
+}
+
+func (c *Compressor) gzipCompress(in *bytes.Buffer) (*bytes.Buffer, error) {
+	out := &bytes.Buffer{}
+	g := gzip.NewWriter(out)
+
+	_, err := g.Write(in.Bytes())
+	if err != nil {
+		return out, err
+	}
+
+	if err := g.Close(); err != nil {
+		return out, err
+	}
+
+	return out, nil
+}
+
+func (c *Compressor) zstdCompress(in *bytes.Buffer) (*bytes.Buffer, error) {
+	out := &bytes.Buffer{}
+
+	z, err := zstd.NewWriter(out)
+	if err != nil {
+		return out, err
+	}
+
+	_, err = z.Write(in.Bytes())
+	if err != nil {
+		return out, err
+	}
+
+	if err := z.Close(); err != nil {
+		return out, err
+	}
+
+	return out, nil
+}
+
+func (c *Compressor) zlibCompress(in *bytes.Buffer) (*bytes.Buffer, error) {
+	out := &bytes.Buffer{}
+	z := zlib.NewWriter(out)
+
+	_, err := z.Write(in.Bytes())
+	if err != nil {
+		return out, err
+	}
+
+	if err := z.Close(); err != nil {
+		return out, err
+	}
+
+	return out, nil
+}
+
+func (c *Compressor) snappyCompress(in *bytes.Buffer) (*bytes.Buffer, error) {
+	compressed := snappy.Encode(nil, in.Bytes())
+
+	return bytes.NewBuffer(compressed), nil
+}
+
+func (c *Compressor) AddHeaders(req *http.Request) {
+	switch c.Strategy {
+	case CompressionStrategyGzip:
+		req.Header.Set("Content-Encoding", "gzip")
+	case CompressionStrategyZstd:
+		req.Header.Set("Content-Encoding", "zstd")
+	case CompressionStrategyZlib:
+		req.Header.Set("Content-Encoding", "deflate")
+	case CompressionStrategySnappy:
+		req.Header.Set("Content-Encoding", "snappy")
+	}
+}

pkg/output/http/exporter.go

@@ -2,7 +2,6 @@ package http
 
 import (
 	"bytes"
-	"compress/gzip"
 	"context"
 	"fmt"
 	"io"
@@ -18,10 +17,10 @@ import (
 )
 
 type ItemExporter struct {
-	config *Config
-	log    logrus.FieldLogger
-
-	client *http.Client
+	config     *Config
+	log        logrus.FieldLogger
+	compressor *Compressor
+	client     *http.Client
 }
 
 func NewItemExporter(name string, config *Config, log logrus.FieldLogger) (ItemExporter, error) {
@@ -38,6 +37,7 @@ func NewItemExporter(name string, config *Config, log logrus.FieldLogger) (ItemE
 			Transport: t,
 			Timeout:   config.ExportTimeout,
 		},
+		compressor: &Compressor{Strategy: config.Compression},
 	}, nil
 }
 
@@ -82,15 +82,14 @@ func (e *ItemExporter) sendUpstream(ctx context.Context, items []*xatu.Decorated
 	}
 
 	buf := bytes.NewBufferString(body)
-	if e.config.Compression == CompressionStrategyGzip {
-		compressed, err := e.gzip(buf)
-		if err != nil {
-			return err
-		}
 
-		buf = compressed
+	compressed, err := e.compressor.Compress(buf)
+	if err != nil {
+		return err
 	}
 
+	buf = compressed
+
 	// TODO: check that this also handles processor timeout
 	req, err := http.NewRequestWithContext(ctx, httpMethod, e.config.Address, buf)
 	if err != nil {
@@ -103,9 +102,7 @@ func (e *ItemExporter) sendUpstream(ctx context.Context, items []*xatu.Decorated
 
 	req.Header.Set("Content-Type", "application/x-ndjson")
 
-	if e.config.Compression == CompressionStrategyGzip {
-		req.Header.Set("Content-Encoding", "gzip")
-	}
+	e.compressor.AddHeaders(req)
 
 	rsp, err = e.client.Do(req)
 	if err != nil {
@@ -125,19 +122,3 @@ func (e *ItemExporter) sendUpstream(ctx context.Context, items []*xatu.Decorated
 
 	return nil
 }
-
-func (e *ItemExporter) gzip(in *bytes.Buffer) (*bytes.Buffer, error) {
-	out := &bytes.Buffer{}
-	g := gzip.NewWriter(out)
-
-	_, err := g.Write(in.Bytes())
-	if err != nil {
-		return out, err
-	}
-
-	if err := g.Close(); err != nil {
-		return out, err
-	}
-
-	return out, nil
-}