[extension/awslogsencodingextension] Fix concurrent usage of gzip reader (open-telemetry#40838)

<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->
#### Description

The gzip reader was returned to the pool before we finished using it,
which could cause multiple go routines using the same reader.

<!-- Issue number (e.g. open-telemetry#1234) or full URL to issue, if applicable. -->
#### Link to tracking issue
N/A.

<!--Describe what testing was performed and which tests were added.-->
#### Testing

There was one unit test added to check this isn't happening anymore.

<!--Describe the documentation added.-->
#### Documentation

N/A

<!--Please delete paragraphs that you did not use before submitting.-->

Author: constanca-m

.chloggen/gzip-concurrent.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: bug_fix
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: awslogsencodingextension
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Fix bug in which concurrent go routines can end up using the same gzip reader
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [40838]
+
+# (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:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# 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: []

extension/encoding/awslogsencodingextension/extension.go

@@ -23,6 +23,12 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/extension/encoding/awslogsencodingextension/internal/unmarshaler/waf"
 )
 
+const (
+	gzipEncoding    = "gzip"
+	bytesEncoding   = "bytes"
+	parquetEncoding = "parquet"
+)
+
 var _ encoding.LogsUnmarshalerExtension = (*encodingExtension)(nil)
 
 type encodingExtension struct {
@@ -80,57 +86,62 @@ func (*encodingExtension) Shutdown(_ context.Context) error {
 }
 
 func (e *encodingExtension) getGzipReader(buf []byte) (io.Reader, error) {
-	var errGzipReader error
+	var err error
 	gzipReader, ok := e.gzipPool.Get().(*gzip.Reader)
 	if !ok {
-		gzipReader, errGzipReader = gzip.NewReader(bytes.NewReader(buf))
+		gzipReader, err = gzip.NewReader(bytes.NewReader(buf))
 	} else {
-		errGzipReader = gzipReader.Reset(bytes.NewReader(buf))
+		err = gzipReader.Reset(bytes.NewBuffer(buf))
 	}
-	if errGzipReader != nil {
+	if err != nil {
 		if gzipReader != nil {
 			e.gzipPool.Put(gzipReader)
 		}
-		return nil, fmt.Errorf("failed to decompress content: %w", errGzipReader)
+		return nil, fmt.Errorf("failed to decompress content: %w", err)
 	}
-	defer func() {
-		_ = gzipReader.Close()
-		e.gzipPool.Put(gzipReader)
-	}()
 	return gzipReader, nil
 }
 
-func (e *encodingExtension) getReaderFromFormat(buf []byte) (io.Reader, error) {
+func (e *encodingExtension) getReaderFromFormat(buf []byte) (string, io.Reader, error) {
 	switch e.format {
 	case formatWAFLog, formatCloudWatchLogsSubscriptionFilter:
-		return e.getGzipReader(buf)
+		reader, err := e.getGzipReader(buf)
+		return gzipEncoding, reader, err
 	case formatS3AccessLog:
-		return bytes.NewReader(buf), nil
+		return bytesEncoding, bytes.NewReader(buf), nil
 	case formatVPCFlowLog:
 		switch e.vpcFormat {
 		case fileFormatParquet:
-			return nil, fmt.Errorf("%q still needs to be implemented", e.vpcFormat)
+			return parquetEncoding, nil, fmt.Errorf("%q still needs to be implemented", e.vpcFormat)
 		case fileFormatPlainText:
-			return e.getGzipReader(buf)
+			reader, err := e.getGzipReader(buf)
+			return gzipEncoding, reader, err
 		default:
 			// should not be possible
-			return nil, fmt.Errorf(
+			return "", nil, fmt.Errorf(
 				"unsupported file fileFormat %q for VPC flow log, expected one of %q",
 				e.vpcFormat,
 				supportedVPCFlowLogFileFormat,
 			)
 		}
 	default:
 		// should not be possible
-		return nil, fmt.Errorf("unimplemented: format %q has no reader", e.format)
+		return "", nil, fmt.Errorf("unimplemented: format %q has no reader", e.format)
 	}
 }
 
 func (e *encodingExtension) UnmarshalLogs(buf []byte) (plog.Logs, error) {
-	reader, err := e.getReaderFromFormat(buf)
+	encodingReader, reader, err := e.getReaderFromFormat(buf)
 	if err != nil {
 		return plog.Logs{}, fmt.Errorf("failed to get reader for %q logs: %w", e.format, err)
 	}
+	defer func() {
+		if encodingReader == gzipEncoding {
+			r := reader.(*gzip.Reader)
+			_ = r.Close()
+			e.gzipPool.Put(r)
+		}
+	}()
 
 	logs, err := e.unmarshaler.UnmarshalAWSLogs(reader)
 	if err != nil {

extension/encoding/awslogsencodingextension/extension_test.go

@@ -5,12 +5,17 @@ package awslogsencodingextension
 
 import (
 	"bytes"
+	"os"
+	"sync"
 	"testing"
 
 	"github.com/klauspost/compress/gzip"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/extension/extensiontest"
+
+	subscriptionfilter "github.com/open-telemetry/opentelemetry-collector-contrib/extension/encoding/awslogsencodingextension/internal/unmarshaler/subscription-filter"
 )
 
 func TestNew_CloudWatchLogsSubscriptionFilter(t *testing.T) {
@@ -91,7 +96,7 @@ func TestGetReaderFromFormat(t *testing.T) {
 	for name, test := range tests {
 		t.Run(name, func(t *testing.T) {
 			e := &encodingExtension{format: test.format}
-			reader, err := e.getReaderFromFormat(test.buf)
+			_, reader, err := e.getReaderFromFormat(test.buf)
 			if test.expectedErr != "" {
 				require.ErrorContains(t, err, test.expectedErr)
 				return
@@ -101,3 +106,49 @@ func TestGetReaderFromFormat(t *testing.T) {
 		})
 	}
 }
+
+// readAndCompressLogFile reads the data inside it, compresses it
+// and returns a GZIP reader for it.
+func readAndCompressLogFile(t *testing.T, file string) []byte {
+	data, err := os.ReadFile(file)
+	require.NoError(t, err)
+	var compressedData bytes.Buffer
+	gzipWriter := gzip.NewWriter(&compressedData)
+	_, err = gzipWriter.Write(data)
+	require.NoError(t, err)
+	err = gzipWriter.Close()
+	require.NoError(t, err)
+	return compressedData.Bytes()
+}
+
+func TestConcurrentGzipReaderUsage(t *testing.T) {
+	// Create an encoding extension for cloudwatch format to test the
+	// gzip reader and check that it works as expected for non concurrent
+	// and concurrent usage
+	ext := &encodingExtension{
+		unmarshaler: subscriptionfilter.NewSubscriptionFilterUnmarshaler(component.BuildInfo{}),
+		format:      formatCloudWatchLogsSubscriptionFilter,
+		gzipPool:    sync.Pool{},
+	}
+
+	cloudwatchData := readAndCompressLogFile(t, "testdata/cloudwatch_log.json")
+	testUnmarshall := func() {
+		_, err := ext.UnmarshalLogs(cloudwatchData)
+		require.NoError(t, err)
+	}
+
+	// non concurrent
+	testUnmarshall()
+
+	// concurrent usage
+	concurrent := 20
+	wg := sync.WaitGroup{}
+	for i := 0; i < concurrent; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			testUnmarshall()
+		}()
+	}
+	wg.Wait()
+}

extension/encoding/awslogsencodingextension/testdata/cloudwatch_log.json

@@ -0,0 +1,16 @@
+{
+    "owner": "123456789012",
+    "logGroup": "CloudTrail",
+    "logStream": "123456789012_CloudTrail_us-east-1",
+    "subscriptionFilters": [
+        "Destination"
+    ],
+    "messageType": "DATA_MESSAGE",
+    "logEvents": [
+        {
+            "id": "31953106606966983378809025079804211143289615424298221568",
+            "timestamp": 1432826855000,
+            "message": "{\"eventVersion\":\"1.03\",\"userIdentity\":{\"type\":\"Root\"}"
+        }
+    ]
+}