Improving memory performance when it comes to snappy

.chloggen/msg_fix-snappy-lazy-read.yaml

@@ -0,0 +1,25 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. otlpreceiver)
+component: confighttp
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Snappy compression to lazy read for memory efficiency 
+
+# One or more tracking issues or pull requests related to the change
+issues: [11177]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

config/confighttp/compress_readcloser.go

@@ -0,0 +1,23 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package confighttp // import "go.opentelemetry.io/collector/config/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()
+}

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")
+			}
+		})
+	}
+}

config/confighttp/compression.go

@@ -58,17 +58,13 @@ var availableDecoders = map[string]func(body io.ReadCloser) (io.ReadCloser, erro
 		}
 		return zr, nil
 	},
+	//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) {
-		sr := snappy.NewReader(body)
-		sb := new(bytes.Buffer)
-		_, err := io.Copy(sb, sr)
-		if err != nil {
-			return nil, err
-		}
-		if err = body.Close(); err != nil {
-			return nil, err
-		}
-		return io.NopCloser(sb), nil
+		// Lazy Reading content to improve memory efficiency
+		return &compressReadCloser{
+			Reader: snappy.NewReader(body),
+			orig:   body,
+		}, nil
 	},
 }
 

config/confighttp/compression_test.go

@@ -232,7 +232,7 @@ func TestHTTPContentDecompressionHandler(t *testing.T) {
 			encoding: "snappy",
 			reqBody:  bytes.NewBuffer(testBody),
 			respCode: http.StatusBadRequest,
-			respBody: "snappy: corrupt input\n",
+			respBody: "snappy: corrupt input",
 		},
 		{
 			name:     "UnsupportedCompression",