From ecac0305261b7dbb52301b78bd423df15678d0a6 Mon Sep 17 00:00:00 2001
From: Sean Marciniak <smarciniak@splunk.com>
Date: Fri, 20 Sep 2024 12:50:22 +0930
Subject: [PATCH] Adding compressReadCloser to ensure underlying resources are
 closed

---
 config/confighttp/compress_readcloser.go      | 23 ++++++
 config/confighttp/compress_readcloser_test.go | 77 +++++++++++++++++++
 config/confighttp/compression.go              |  5 +-
 3 files changed, 104 insertions(+), 1 deletion(-)
 create mode 100644 config/confighttp/compress_readcloser.go
 create mode 100644 config/confighttp/compress_readcloser_test.go

diff --git a/config/confighttp/compress_readcloser.go b/config/confighttp/compress_readcloser.go
new file mode 100644
index 00000000000..40c8855bad2
--- /dev/null
+++ b/config/confighttp/compress_readcloser.go
@@ -0,0 +1,23 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package confighttp
+
+import "io"
+
+// compressReadCloser couples the original compressed reader
+// and the compression reader to ensure that the original body
+// is correctly closed to ensure resources are freed.
+type compressReadCloser struct {
+	io.Reader
+	orig io.ReadCloser
+}
+
+var (
+	_ io.Reader = (*compressReadCloser)(nil)
+	_ io.Closer = (*compressReadCloser)(nil)
+)
+
+func (crc *compressReadCloser) Close() error {
+	return crc.orig.Close()
+}
diff --git a/config/confighttp/compress_readcloser_test.go b/config/confighttp/compress_readcloser_test.go
new file mode 100644
index 00000000000..fd97a15e242
--- /dev/null
+++ b/config/confighttp/compress_readcloser_test.go
@@ -0,0 +1,77 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package confighttp
+
+import (
+	"bytes"
+	"errors"
+	"io"
+	"testing"
+	"testing/iotest"
+
+	"github.com/stretchr/testify/require"
+)
+
+type errorReadCloser struct {
+	io.Reader
+	err error
+}
+
+func (erc errorReadCloser) Close() error {
+	return erc.err
+}
+
+func TestCompressReadCloser(t *testing.T) {
+	t.Parallel()
+
+	for _, tc := range []struct {
+		name    string
+		wrapper func(r io.Reader) io.ReadCloser
+		content []byte
+		errVal  string
+	}{
+		{
+			name: "non mutating wrapper",
+			wrapper: func(r io.Reader) io.ReadCloser {
+				return errorReadCloser{
+					Reader: r,
+					err:    nil,
+				}
+			},
+			content: []byte("hello world"),
+			errVal:  "",
+		},
+		{
+			name: "failed reader",
+			wrapper: func(r io.Reader) io.ReadCloser {
+				return errorReadCloser{
+					Reader: r,
+					err:    errors.New("failed to close reader"),
+				}
+			},
+			errVal: "failed to close reader",
+		},
+	} {
+		tc := tc
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			orig := bytes.NewBuffer([]byte("hello world"))
+
+			crc := &compressReadCloser{
+				Reader: orig,
+				orig:   tc.wrapper(orig),
+			}
+
+			require.NoError(t, iotest.TestReader(crc, orig.Bytes()), "Must be able to read original content")
+
+			err := crc.Close()
+			if tc.errVal != "" {
+				require.EqualError(t, err, tc.errVal, "Must match the expected error message")
+			} else {
+				require.NoError(t, err, "Must not error when closing reader")
+			}
+		})
+	}
+}
diff --git a/config/confighttp/compression.go b/config/confighttp/compression.go
index 3f780d16269..6680b61ff4f 100644
--- a/config/confighttp/compression.go
+++ b/config/confighttp/compression.go
@@ -61,7 +61,10 @@ var availableDecoders = map[string]func(body io.ReadCloser) (io.ReadCloser, erro
 	//nolint:unparam // Ignoring the linter request to remove error return since it needs to match the method signature
 	"snappy": func(body io.ReadCloser) (io.ReadCloser, error) {
 		// Lazy Reading content to improve memory efficiency
-		return io.NopCloser(snappy.NewReader(body)), nil
+		return &compressReadCloser{
+			Reader: snappy.NewReader(body),
+			orig:   body,
+		}, nil
 	},
 }