feat(output): HTTP Gzip compression (#31)

* feat(output): HTTP Gzip compression

* feat(output): CompressionStrategy

pkg/output/http/compression.go

@@ -0,0 +1,8 @@
+package http
+
+type CompressionStrategy string
+
+var (
+	CompressionStrategyNone CompressionStrategy = "none"
+	CompressionStrategyGzip CompressionStrategy = "gzip"
+)

pkg/output/http/config.go

@@ -6,12 +6,13 @@ import (
 )
 
 type Config struct {
-	Address            string            `yaml:"address"`
-	Headers            map[string]string `yaml:"headers"`
-	MaxQueueSize       int               `yaml:"max_queue_size" default:"51200"`
-	BatchTimeout       time.Duration     `yaml:"batch_timeout" default:"5s"`
-	ExportTimeout      time.Duration     `yaml:"export_timeout" default:"30s"`
-	MaxExportBatchSize int               `yaml:"max_export_batch_size" default:"512"`
+	Address            string              `yaml:"address"`
+	Headers            map[string]string   `yaml:"headers"`
+	MaxQueueSize       int                 `yaml:"max_queue_size" default:"51200"`
+	BatchTimeout       time.Duration       `yaml:"batch_timeout" default:"5s"`
+	ExportTimeout      time.Duration       `yaml:"export_timeout" default:"30s"`
+	MaxExportBatchSize int                 `yaml:"max_export_batch_size" default:"512"`
+	Compression        CompressionStrategy `yaml:"compression" default:"none"`
 }
 
 func (c *Config) Validate() error {

pkg/output/http/exporter.go

@@ -2,6 +2,7 @@ package http
 
 import (
 	"bytes"
+	"compress/gzip"
 	"context"
 	"fmt"
 	"io"
@@ -60,8 +61,18 @@ func (e *EventExporter) sendUpstream(ctx context.Context, events []*xatu.Decorat
 		body += string(eventAsJSON) + "\n"
 	}
 
+	buf := bytes.NewBufferString(body)
+	if e.config.Compression == CompressionStrategyGzip {
+		compressed, err := e.gzip(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, bytes.NewBufferString(body))
+	req, err := http.NewRequestWithContext(ctx, httpMethod, e.config.Address, buf)
 	if err != nil {
 		return err
 	}
@@ -72,6 +83,10 @@ func (e *EventExporter) sendUpstream(ctx context.Context, events []*xatu.Decorat
 
 	req.Header.Set("Content-Type", "application/x-ndjson")
 
+	if e.config.Compression == CompressionStrategyGzip {
+		req.Header.Set("Content-Encoding", "gzip")
+	}
+
 	rsp, err = e.client.Do(req)
 	if err != nil {
 		return err
@@ -90,3 +105,19 @@ func (e *EventExporter) sendUpstream(ctx context.Context, events []*xatu.Decorat
 
 	return nil
 }
+
+func (e *EventExporter) 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
+}