chore(internal.gzip): cleanup CompressWithGzip (#12587)

internal/internal.go

@@ -194,30 +194,39 @@ func (r *ReadWaitCloser) Close() error {
 	return err
 }
 
-// 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.ReadCloser, error) {
+// CompressWithGzip takes an io.Reader as input and pipes it through a
+// gzip.Writer returning an io.Reader containing the gzipped data.
+// Errors occurring during compression are returned to the instance reading
+// from the returned reader via through the corresponding read call
+// (e.g. io.Copy or io.ReadAll).
+func CompressWithGzip(data io.Reader) io.ReadCloser {
 	pipeReader, pipeWriter := io.Pipe()
 	gzipWriter := gzip.NewWriter(pipeWriter)
 
-	rc := &ReadWaitCloser{
-		pipeReader: pipeReader,
-	}
-
-	rc.wg.Add(1)
-	var err error
+	// Start copying from the uncompressed reader to the output reader
+	// in the background until the input reader is closed (or errors out).
 	go func() {
-		_, err = io.Copy(gzipWriter, data)
-		gzipWriter.Close()
-		// subsequent reads from the read half of the pipe will
-		// return no bytes and the error err, or EOF if err is nil.
-		pipeWriter.CloseWithError(err)
-		rc.wg.Done()
+		// This copy will block until "data" reached EOF or an error occurs
+		_, err := io.Copy(gzipWriter, data)
+
+		// Close the compression writer and make sure we do not overwrite
+		// the copy error if any.
+		gzipErr := gzipWriter.Close()
+		if err == nil {
+			err = gzipErr
+		}
+
+		// Subsequent reads from the output reader (connected to "pipeWriter"
+		// via pipe) will return the copy (or closing) error if any to the
+		// instance reading from the reader returned by the CompressWithGzip
+		// function. If "err" is nil, the below function will correctly report
+		// io.EOF.
+		_ = pipeWriter.CloseWithError(err)
 	}()
 
-	return pipeReader, err
+	// Return a reader which then can be read by the caller to collect the
+	// compressed stream.
+	return pipeReader
 }
 
 // ParseTimestamp parses a Time according to the standard Telegraf options.

internal/internal_test.go

@@ -174,9 +174,7 @@ func TestCompressWithGzip(t *testing.T) {
 	testData := "the quick brown fox jumps over the lazy dog"
 	inputBuffer := bytes.NewBuffer([]byte(testData))
 
-	outputBuffer, err := CompressWithGzip(inputBuffer)
-	require.NoError(t, err)
-
+	outputBuffer := CompressWithGzip(inputBuffer)
 	gzipReader, err := gzip.NewReader(outputBuffer)
 	require.NoError(t, err)
 	defer gzipReader.Close()
@@ -188,37 +186,71 @@ func TestCompressWithGzip(t *testing.T) {
 }
 
 type mockReader struct {
-	readN uint64 // record the number of calls to Read
+	err    error
+	ncalls uint64 // record the number of calls to Read
+	msg    []byte
 }
 
 func (r *mockReader) Read(p []byte) (n int, err error) {
-	r.readN++
-	return rand.Read(p)
+	r.ncalls++
+
+	if len(r.msg) > 0 {
+		n, err = copy(p, r.msg), io.EOF
+	} else {
+		n, err = rand.Read(p)
+	}
+	if r.err == nil {
+		return n, err
+	}
+	return n, r.err
 }
 
 func TestCompressWithGzipEarlyClose(t *testing.T) {
 	mr := &mockReader{}
 
-	rc, err := CompressWithGzip(mr)
-	require.NoError(t, err)
-
+	rc := CompressWithGzip(mr)
 	n, err := io.CopyN(io.Discard, rc, 10000)
 	require.NoError(t, err)
 	require.Equal(t, int64(10000), n)
 
-	r1 := mr.readN
-	err = rc.Close()
-	require.NoError(t, err)
+	r1 := mr.ncalls
+	require.NoError(t, rc.Close())
 
 	n, err = io.CopyN(io.Discard, rc, 10000)
-	require.Error(t, io.EOF, err)
+	require.ErrorIs(t, err, io.ErrClosedPipe)
 	require.Equal(t, int64(0), n)
 
-	r2 := mr.readN
+	r2 := mr.ncalls
 	// no more read to the source after closing
 	require.Equal(t, r1, r2)
 }
 
+func TestCompressWithGzipErrorPropagationCopy(t *testing.T) {
+	errs := []error{io.ErrClosedPipe, io.ErrNoProgress, io.ErrUnexpectedEOF}
+	for _, expected := range errs {
+		r := &mockReader{msg: []byte("this is a test"), err: expected}
+
+		rc := CompressWithGzip(r)
+		n, err := io.Copy(io.Discard, rc)
+		require.Greater(t, n, int64(0))
+		require.ErrorIs(t, err, expected)
+		require.NoError(t, rc.Close())
+	}
+}
+
+func TestCompressWithGzipErrorPropagationReadAll(t *testing.T) {
+	errs := []error{io.ErrClosedPipe, io.ErrNoProgress, io.ErrUnexpectedEOF}
+	for _, expected := range errs {
+		r := &mockReader{msg: []byte("this is a test"), err: expected}
+
+		rc := CompressWithGzip(r)
+		buf, err := io.ReadAll(rc)
+		require.NotEmpty(t, buf)
+		require.ErrorIs(t, err, expected)
+		require.NoError(t, rc.Close())
+	}
+}
+
 func TestAlignDuration(t *testing.T) {
 	tests := []struct {
 		name     string

plugins/inputs/http/http.go

@@ -2,7 +2,6 @@
 package http
 
 import (
-	"bytes"
 	"context"
 	_ "embed"
 	"fmt"
@@ -214,15 +213,7 @@ func makeRequestBodyReader(contentEncoding, body string) (io.Reader, error) {
 
 	var reader io.Reader = strings.NewReader(body)
 	if contentEncoding == "gzip" {
-		rc, err := internal.CompressWithGzip(reader)
-		if err != nil {
-			return nil, err
-		}
-		data, err := io.ReadAll(rc)
-		if err != nil {
-			return nil, err
-		}
-		return bytes.NewReader(data), nil
+		return internal.CompressWithGzip(reader), nil
 	}
 
 	return reader, nil

plugins/outputs/http/http.go

@@ -136,10 +136,7 @@ func (h *HTTP) writeMetric(reqBody []byte) error {
 
 	var err error
 	if h.ContentEncoding == "gzip" {
-		rc, err := internal.CompressWithGzip(reqBodyBuffer)
-		if err != nil {
-			return err
-		}
+		rc := internal.CompressWithGzip(reqBodyBuffer)
 		defer rc.Close()
 		reqBodyBuffer = rc
 	}

plugins/outputs/influxdb/http.go

@@ -485,12 +485,7 @@ func (c *httpClient) requestBodyReader(metrics []telegraf.Metric) (io.ReadCloser
 	reader := influx.NewReader(metrics, c.config.Serializer)
 
 	if c.config.ContentEncoding == "gzip" {
-		rc, err := internal.CompressWithGzip(reader)
-		if err != nil {
-			return nil, err
-		}
-
-		return rc, nil
+		return internal.CompressWithGzip(reader), nil
 	}
 
 	return io.NopCloser(reader), nil

plugins/outputs/influxdb_v2/http.go

@@ -391,12 +391,7 @@ func (c *httpClient) requestBodyReader(metrics []telegraf.Metric) (io.ReadCloser
 	reader := influx.NewReader(metrics, c.serializer)
 
 	if c.ContentEncoding == "gzip" {
-		rc, err := internal.CompressWithGzip(reader)
-		if err != nil {
-			return nil, err
-		}
-
-		return rc, nil
+		return internal.CompressWithGzip(reader), nil
 	}
 
 	return io.NopCloser(reader), nil

plugins/outputs/loki/loki.go

@@ -143,10 +143,7 @@ func (l *Loki) writeMetrics(s Streams) error {
 	var reqBodyBuffer io.Reader = bytes.NewBuffer(bs)
 
 	if l.GZipRequest {
-		rc, err := internal.CompressWithGzip(reqBodyBuffer)
-		if err != nil {
-			return err
-		}
+		rc := internal.CompressWithGzip(reqBodyBuffer)
 		defer rc.Close()
 		reqBodyBuffer = rc
 	}

plugins/outputs/sensu/sensu.go

@@ -213,10 +213,7 @@ func (s *Sensu) writeMetrics(reqBody []byte) error {
 	method := http.MethodPost
 
 	if s.ContentEncoding == "gzip" {
-		rc, err := internal.CompressWithGzip(reqBodyBuffer)
-		if err != nil {
-			return err
-		}
+		rc := internal.CompressWithGzip(reqBodyBuffer)
 		defer rc.Close()
 		reqBodyBuffer = rc
 	}