[AWS] [S3] Remove url.QueryUnescape() from aws-s3 input in polling mode

CHANGELOG.next.asciidoc

@@ -97,6 +97,7 @@ fields added to events containing the Beats version. {pull}37553[37553]
 - [threatintel] MISP pagination fixes {pull}37898[37898]
 - Fix file handle leak when handling errors in filestream {pull}37973[37973]
 - Prevent HTTPJSON holding response bodies between executions. {issue}35219[35219] {pull}38116[38116]
+- Fix "failed processing S3 event for object key" error on aws-s3 input when key contains the "+" character {issue}38012[38012] {pull}38125[38125]
 
 *Heartbeat*
 

x-pack/filebeat/input/awss3/s3.go

@@ -8,7 +8,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"net/url"
 	"sync"
 	"time"
 
@@ -208,14 +207,7 @@ func (p *s3Poller) GetS3Objects(ctx context.Context, s3ObjectPayloadChan chan<-
 		// Metrics
 		p.metrics.s3ObjectsListedTotal.Add(uint64(totListedObjects))
 		for _, object := range page.Contents {
-			// Unescape s3 key name. For example, convert "%3D" back to "=".
-			filename, err := url.QueryUnescape(*object.Key)
-			if err != nil {
-				p.log.Errorw("Error when unescaping object key, skipping.", "error", err, "s3_object", *object.Key)
-				continue
-			}
-
-			state := newState(bucketName, filename, *object.ETag, p.listPrefix, *object.LastModified)
+			state := newState(bucketName, *object.Key, *object.ETag, p.listPrefix, *object.LastModified)
 			if p.states.MustSkip(state, p.store) {
 				p.log.Debugw("skipping state.", "state", state)
 				continue
@@ -240,7 +232,7 @@ func (p *s3Poller) GetS3Objects(ctx context.Context, s3ObjectPayloadChan chan<-
 				s3ObjectHandler: s3Processor,
 				s3ObjectInfo: s3ObjectInfo{
 					name:         bucketName,
-					key:          filename,
+					key:          *object.Key,
 					etag:         *object.ETag,
 					lastModified: *object.LastModified,
 					listingID:    listingID.String(),

x-pack/filebeat/input/awss3/s3_test.go

@@ -93,6 +93,11 @@ func TestS3Poller(t *testing.T) {
 							Key:          aws.String("key5"),
 							LastModified: aws.Time(time.Now()),
 						},
+						{
+							ETag:         aws.String("etag6"),
+							Key:          aws.String("2024-02-08T08:35:00+00:02.json.gz"),
+							LastModified: aws.Time(time.Now()),
+						},
 					},
 				}, nil
 			})
@@ -124,6 +129,10 @@ func TestS3Poller(t *testing.T) {
 			GetObject(gomock.Any(), gomock.Eq(bucket), gomock.Eq("key5")).
 			Return(nil, errFakeConnectivityFailure)
 
+		mockAPI.EXPECT().
+			GetObject(gomock.Any(), gomock.Eq(bucket), gomock.Eq("2024-02-08T08:35:00+00:02.json.gz")).
+			Return(nil, errFakeConnectivityFailure)
+
 		s3ObjProc := newS3ObjectProcessorFactory(logp.NewLogger(inputName), nil, mockAPI, nil, backupConfig{}, numberOfWorkers)
 		receiver := newS3Poller(logp.NewLogger(inputName), nil, mockAPI, mockPublisher, s3ObjProc, newStates(inputCtx), store, bucket, "key", "region", "provider", numberOfWorkers, pollInterval)
 		require.Error(t, context.DeadlineExceeded, receiver.Poll(ctx))