Extract CompressWithGzip into the internal package

internal/internal.go

@@ -3,8 +3,10 @@ package internal
 import (
 	"bufio"
 	"bytes"
+	"compress/gzip"
 	"crypto/rand"
 	"errors"
+	"io"
 	"log"
 	"math/big"
 	"os"
@@ -208,3 +210,21 @@ func ExitStatus(err error) (int, bool) {
 	}
 	return 0, false
 }
+
+// CompressWithGzip takes an io.Reader as input and pipes
+// it through a gzip.Writer returning an io.Reader containing
+// the gzipped data.
+// An error is returned if passing data to the gzip.Writer fails
+func CompressWithGzip(data io.Reader) (io.Reader, error) {
+	pipeReader, pipeWriter := io.Pipe()
+	gzipWriter := gzip.NewWriter(pipeWriter)
+
+	var err error
+	go func() {
+		_, err = io.Copy(gzipWriter, data)
+		gzipWriter.Close()
+		pipeWriter.Close()
+	}()
+
+	return pipeReader, err
+}

internal/internal_test.go

@@ -1,6 +1,9 @@
 package internal
 
 import (
+	"bytes"
+	"compress/gzip"
+	"io/ioutil"
 	"os/exec"
 	"testing"
 	"time"
@@ -162,3 +165,20 @@ func TestDuration(t *testing.T) {
 	d.UnmarshalTOML([]byte(`1.5`))
 	assert.Equal(t, time.Second, d.Duration)
 }
+
+func TestCompressWithGzip(t *testing.T) {
+	testData := "the quick brown fox jumps over the lazy dog"
+	inputBuffer := bytes.NewBuffer([]byte(testData))
+
+	outputBuffer, err := CompressWithGzip(inputBuffer)
+	assert.NoError(t, err)
+
+	gzipReader, err := gzip.NewReader(outputBuffer)
+	assert.NoError(t, err)
+	defer gzipReader.Close()
+
+	output, err := ioutil.ReadAll(gzipReader)
+	assert.NoError(t, err)
+
+	assert.Equal(t, testData, string(output))
+}

plugins/outputs/http/http.go

@@ -2,7 +2,6 @@ package http
 
 import (
 	"bytes"
-	"compress/gzip"
 	"context"
 	"fmt"
 	"io"
@@ -173,7 +172,7 @@ func (h *HTTP) write(reqBody []byte) error {
 	var err error
 
 	if h.ContentEncoding == "gzip" && h.Method == http.MethodPost {
-		reqBodyBuffer, err = compressWithGzip(reqBodyBuffer)
+		reqBodyBuffer, err = internal.CompressWithGzip(reqBodyBuffer)
 		if err != nil {
 			return err
 		}
@@ -210,20 +209,6 @@ func (h *HTTP) write(reqBody []byte) error {
 	return nil
 }
 
-func compressWithGzip(data io.Reader) (io.Reader, error) {
-	pr, pw := io.Pipe()
-	gw := gzip.NewWriter(pw)
-	var err error
-
-	go func() {
-		_, err = io.Copy(gw, data)
-		gw.Close()
-		pw.Close()
-	}()
-
-	return pr, err
-}
-
 func init() {
 	outputs.Add("http", func() telegraf.Output {
 		return &HTTP{

plugins/outputs/influxdb/http.go

@@ -1,7 +1,6 @@
 package influxdb
 
 import (
-	"compress/gzip"
 	"context"
 	"crypto/tls"
 	"encoding/json"
@@ -16,6 +15,7 @@ import (
 	"time"
 
 	"github.com/influxdata/telegraf"
+	"github.com/influxdata/telegraf/internal"
 	"github.com/influxdata/telegraf/plugins/serializers/influx"
 )
 
@@ -360,7 +360,7 @@ func (c *httpClient) makeQueryRequest(query string) (*http.Request, error) {
 func (c *httpClient) makeWriteRequest(body io.Reader) (*http.Request, error) {
 	var err error
 	if c.ContentEncoding == "gzip" {
-		body, err = compressWithGzip(body)
+		body, err = internal.CompressWithGzip(body)
 		if err != nil {
 			return nil, err
 		}
@@ -381,20 +381,6 @@ func (c *httpClient) makeWriteRequest(body io.Reader) (*http.Request, error) {
 	return req, nil
 }
 
-func compressWithGzip(data io.Reader) (io.Reader, error) {
-	pr, pw := io.Pipe()
-	gw := gzip.NewWriter(pw)
-	var err error
-
-	go func() {
-		_, err = io.Copy(gw, data)
-		gw.Close()
-		pw.Close()
-	}()
-
-	return pr, err
-}
-
 func (c *httpClient) addHeaders(req *http.Request) {
 	if c.Username != "" || c.Password != "" {
 		req.SetBasicAuth(c.Username, c.Password)

plugins/outputs/influxdb_v2/http.go

@@ -1,7 +1,6 @@
 package influxdb_v2
 
 import (
-	"compress/gzip"
 	"context"
 	"crypto/tls"
 	"encoding/json"
@@ -17,6 +16,7 @@ import (
 	"time"
 
 	"github.com/influxdata/telegraf"
+	"github.com/influxdata/telegraf/internal"
 	"github.com/influxdata/telegraf/plugins/serializers/influx"
 )
 
@@ -231,7 +231,7 @@ func (c *httpClient) Write(ctx context.Context, metrics []telegraf.Metric) error
 func (c *httpClient) makeWriteRequest(body io.Reader) (*http.Request, error) {
 	var err error
 	if c.ContentEncoding == "gzip" {
-		body, err = compressWithGzip(body)
+		body, err = internal.CompressWithGzip(body)
 		if err != nil {
 			return nil, err
 		}
@@ -258,20 +258,6 @@ func (c *httpClient) addHeaders(req *http.Request) {
 	}
 }
 
-func compressWithGzip(data io.Reader) (io.Reader, error) {
-	pipeReader, pipeWriter := io.Pipe()
-	gzipWriter := gzip.NewWriter(pipeWriter)
-	var err error
-
-	go func() {
-		_, err = io.Copy(gzipWriter, data)
-		gzipWriter.Close()
-		pipeWriter.Close()
-	}()
-
-	return pipeReader, err
-}
-
 func makeWriteURL(loc url.URL, org, bucket string) (string, error) {
 	params := url.Values{}
 	params.Set("bucket", bucket)