[8.x](backport #42078) [x-pack][filebeat] incorporate LastModified in…

…to s3 event _id generation (#42132)

* [x-pack][filebeat] incorporate LastModified into s3 event _id generation (#42078)

* [x-pack][filebeat] incorporate LastModified into s3 event _id generation

* update PR id in changelog

* log LastModified value

* add unit test

* fix unit tests on windows

* use UnixMilli instead of time string

* change s3 event _id format

* fix changelog

* Update x-pack/filebeat/input/gcs/input_test.go

Co-authored-by: Dan Kortschak <dan.kortschak@elastic.co>

* Update x-pack/filebeat/input/azureblobstorage/input_test.go

Co-authored-by: Dan Kortschak <dan.kortschak@elastic.co>

* revert windows test fix

* add unit test

---------

Co-authored-by: Dan Kortschak <dan.kortschak@elastic.co>
(cherry picked from commit 58a5369)

# Conflicts:
#	x-pack/filebeat/input/awss3/s3_objects.go

* resolve conflicts

* linter fix

---------

Co-authored-by: stefans-elastic <stefan.stas@elastic.co>

CHANGELOG.next.asciidoc

@@ -192,6 +192,8 @@ https://github.com/elastic/beats/compare/v8.8.1\...main[Check the HEAD diff]
 - Redact authorization headers in HTTPJSON debug logs. {pull}41920[41920]
 - Further rate limiting fix in the Okta provider of the Entity Analytics input. {issue}40106[40106] {pull}41977[41977]
 - Fix streaming input handling of invalid or empty websocket messages. {pull}42036[42036]
+- Fix awss3 document ID construction when using the CSV decoder. {pull}42019[42019]
+- The `_id` generation process for S3 events has been updated to incorporate the LastModified field. This enhancement ensures that the `_id` is unique. {pull}42078[42078]
 - Fix Netflow Template Sharing configuration handling. {pull}42080[42080]
 
 *Heartbeat*

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

@@ -256,5 +256,6 @@ func (in *s3PollerInput) s3EventForState(state state) s3EventV2 {
 	event.S3.Bucket.Name = state.Bucket
 	event.S3.Bucket.ARN = in.config.getBucketARN()
 	event.S3.Object.Key = state.Key
+	event.S3.Object.LastModified = state.LastModified
 	return event
 }

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

@@ -117,7 +117,8 @@ func (p *s3ObjectProcessor) ProcessS3Object(log *logp.Logger, eventCallback func
 	p.eventCallback = eventCallback
 	log = log.With(
 		"bucket_arn", p.s3Obj.S3.Bucket.Name,
-		"object_key", p.s3Obj.S3.Object.Key)
+		"object_key", p.s3Obj.S3.Object.Key,
+		"last_modified", p.s3Obj.S3.Object.LastModified)
 
 	// Metrics and Logging
 	log.Debug("Begin S3 object processing.")
@@ -448,7 +449,11 @@ func (p *s3ObjectProcessor) createEvent(message string, offset int64) beat.Event
 			},
 		},
 	}
-	event.SetID(objectID(p.s3ObjHash, offset))
+
+	if offset >= 0 {
+		event.Fields.Put("log.offset", offset)
+		event.SetID(objectID(p.s3Obj.S3.Object.LastModified, p.s3ObjHash, offset))
+	}
 
 	if len(p.s3Metadata) > 0 {
 		_, _ = event.Fields.Put("aws.s3.metadata", p.s3Metadata)
@@ -477,8 +482,8 @@ func (p *s3ObjectProcessor) FinalizeS3Object() error {
 	return nil
 }
 
-func objectID(objectHash string, offset int64) string {
-	return fmt.Sprintf("%s-%012d", objectHash, offset)
+func objectID(lastModified time.Time, objectHash string, offset int64) string {
+	return fmt.Sprintf("%d-%s-%012d", lastModified.UnixNano(), objectHash, offset)
 }
 
 // s3ObjectHash returns a short sha256 hash of the bucket arn + object key name.

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

@@ -13,6 +13,7 @@ import (
 	"path/filepath"
 	"strings"
 	"testing"
+	"time"
 
 	awssdk "github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/service/s3"
@@ -435,6 +436,13 @@ func TestNewMockS3Pager(t *testing.T) {
 	assert.Equal(t, []string{"foo", "bar", "baz"}, keys)
 }
 
+func Test_objectID(t *testing.T) {
+	lastModified, _ := time.Parse("2006-01-02 15:04:05 -0700", "2024-11-07 12:44:22 +0000")
+	objId := objectID(lastModified, "fe8a230c26", 42)
+
+	assert.Equal(t, "1730983462000000000-fe8a230c26-000000000042", objId)
+}
+
 // newMockS3Pager returns a s3Pager that paginates the given s3Objects based on
 // the specified page size. It never returns an error.
 func newMockS3Pager(ctrl *gomock.Controller, pageSize int, s3Objects []types.Object) *MockS3Pager {

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

@@ -43,7 +43,7 @@ func (e *nonRetryableError) Error() string {
 }
 
 func (e *nonRetryableError) Is(err error) bool {
-	_, ok := err.(*nonRetryableError) //nolint:errorlint // This is not used directly to detected wrapped errors (errors.Is handles unwrapping).
+	_, ok := err.(*nonRetryableError) //nolint:nolintlint,errorlint // This is not used directly to detected wrapped errors (errors.Is handles unwrapping).
 	return ok
 }
 
@@ -77,7 +77,8 @@ type s3EventV2 struct {
 			ARN  string `json:"arn"`
 		} `json:"bucket"`
 		Object struct {
-			Key string `json:"key"`
+			Key          string    `json:"key"`
+			LastModified time.Time `json:"lastModified"`
 		} `json:"object"`
 	} `json:"s3"`
 }

x-pack/filebeat/input/azureblobstorage/input_test.go

@@ -540,7 +540,8 @@ func Test_StorageClient(t *testing.T) {
 					var err error
 					val, err = got.Fields.GetValue("message")
 					assert.NoError(t, err)
-					assert.True(t, tt.expected[val.(string)])
+
+					assert.True(t, tt.expected[strings.ReplaceAll(val.(string), "\r\n", "\n")])
 					assert.Equal(t, tt.expectedError, err)
 					receivedCount += 1
 					if receivedCount == len(tt.expected) {

x-pack/filebeat/input/gcs/input_test.go

@@ -13,6 +13,7 @@ import (
 	"fmt"
 	"net/http"
 	"net/http/httptest"
+	"strings"
 	"testing"
 	"time"
 
@@ -662,7 +663,7 @@ func Test_StorageClient(t *testing.T) {
 					if !tt.checkJSON {
 						val, err = got.Fields.GetValue("message")
 						assert.NoError(t, err)
-						assert.True(t, tt.expected[val.(string)])
+						assert.True(t, tt.expected[strings.ReplaceAll(val.(string), "\r\n", "\n")])
 					} else {
 						val, err = got.Fields.GetValue("gcs.storage.object.json_data")
 						fVal := fmt.Sprintf("%v", val)