[AWS] Skip s3 test events in s3-sqs filebeat input (#47635)

* Skip s3 test events

* add changelog

* update changelog

* keep only event field in s3TestEvent

* add Event into s3EventsV2

* add comment and change variable name for processing s3-sns-sqs

Author: kaiyan-sheng

changelog/fragments/1763061151-bug-skip-s3-test-events-in-filebeat-s3-input.yaml

@@ -0,0 +1,45 @@
+# REQUIRED
+# Kind can be one of:
+# - breaking-change: a change to previously-documented behavior
+# - deprecation: functionality that is being removed in a later release
+# - bug-fix: fixes a problem in a previous version
+# - enhancement: extends functionality but does not break or fix existing behavior
+# - feature: new functionality
+# - known-issue: problems that we are aware of in a given version
+# - security: impacts on the security of a product or a user’s deployment.
+# - upgrade: important information for someone upgrading from a prior version
+# - other: does not fit into any of the other categories
+kind: bug-fix
+
+# REQUIRED for all kinds
+# Change summary; a 80ish characters long description of the change.
+summary: Skip s3 test events in filebeat s3 input
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# Long description; in case the summary is not enough to describe the change
+# this field accommodate a description without length limits.
+# description:
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# impact:
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# action:
+
+# REQUIRED for all kinds
+# Affected component; usually one of "elastic-agent", "fleet-server", "filebeat", "metricbeat", "auditbeat", "all", etc.
+component: filebeat
+
+# AUTOMATED
+# OPTIONAL to manually add other PR URLs
+# PR URL: A link the PR that added the changeset.
+# If not present is automatically filled by the tooling finding the PR where this changelog fragment has been added.
+# NOTE: the tooling supports backports, so it's able to fill the original PR number instead of the backport PR number.
+# Please provide it if you are adding a fragment for a different PR.
+# pr: https://github.com/owner/repo/1234
+
+# AUTOMATED
+# OPTIONAL to manually add other issue URLs
+# Issue URL; optional; the GitHub issue related to this changeset (either closes or is part of).
+# If not present is automatically filled by the tooling with the issue linked to the PR number.
+# issue: https://github.com/owner/repo/1234

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

@@ -60,7 +60,9 @@ func nonRetryableErrorWrap(err error) error {
 // https://docs.aws.amazon.com/AmazonS3/latest/userguide/notification-content-structure.html
 // If the notification message is sent from SNS to SQS, then Records will be
 // replaced by TopicArn and Message fields.
+// The Event field is present in test event notifications (s3:TestEvent) but not in regular events.
 type s3EventsV2 struct {
+	Event    string      `json:"Event"` // Present in test events (s3:TestEvent), empty in regular events
 	TopicArn string      `json:"TopicArn"`
 	Message  string      `json:"Message"`
 	Records  []s3EventV2 `json:"Records"`
@@ -323,13 +325,26 @@ func (p *sqsS3EventProcessor) getS3Notifications(body string) ([]s3EventV2, erro
 		return nil, fmt.Errorf("failed to decode SQS message body as an S3 notification: %w", err)
 	}
 
+	// Check if this is a test event and skip it
+	if events.Event == "s3:TestEvent" {
+		p.log.Debugw("Skipping S3 test event notification", "sqs_message_body", body)
+		return nil, nil
+	}
+
 	// Check if the notification is from S3 -> SNS -> SQS
 	if events.TopicArn != "" {
+		// Check if the inner message is a test event before unmarshaling
+		var innerEvents s3EventsV2
 		dec := json.NewDecoder(strings.NewReader(events.Message))
-		if err := dec.Decode(&events); err != nil {
+		if err := dec.Decode(&innerEvents); err != nil {
 			p.log.Debugw("Invalid SQS message body.", "sqs_message_body", body)
 			return nil, fmt.Errorf("failed to decode SQS message body as an S3 notification: %w", err)
 		}
+		if innerEvents.Event == "s3:TestEvent" {
+			p.log.Debugw("Skipping S3 test event notification (via SNS)", "sqs_message_body", body)
+			return nil, nil
+		}
+		events = innerEvents
 	}
 
 	// Check if the notification is from S3 -> EventBridge -> SQS

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

@@ -282,6 +282,30 @@ func TestSqsProcessor_getS3Notifications(t *testing.T) {
 		require.NoError(t, err)
 		assert.Equal(t, 0, len(events))
 	})
+
+	t.Run("s3:TestEvent messages are skipped", func(t *testing.T) {
+		testEventMsg := `{
+			"Service":"Amazon S3",
+			"Event":"s3:TestEvent",
+			"Time":"2014-10-13T15:57:02.089Z",
+			"Bucket":"amzn-s3-demo-bucket",
+			"RequestId":"5582815E1AEA5ADF",
+			"HostId":"8cLeGAmw098X5cv4Zkwcmo8vvZa3eH3eKxsPzbB9wrR+YstdA6Knx4Ip8EXAMPLE"
+		}`
+		events, err := p.getS3Notifications(testEventMsg)
+		require.NoError(t, err)
+		assert.Equal(t, 0, len(events), "Test events should be skipped and return no events")
+	})
+
+	t.Run("s3:TestEvent messages via SNS are skipped", func(t *testing.T) {
+		testEventViaSNS := `{
+			"TopicArn":"arn:aws:sns:us-east-1:123456789012:test-topic",
+			"Message":"{\"Service\":\"Amazon S3\",\"Event\":\"s3:TestEvent\",\"Time\":\"2014-10-13T15:57:02.089Z\",\"Bucket\":\"amzn-s3-demo-bucket\",\"RequestId\":\"5582815E1AEA5ADF\",\"HostId\":\"8cLeGAmw098X5cv4Zkwcmo8vvZa3eH3eKxsPzbB9wrR+YstdA6Knx4Ip8EXAMPLE\"}"
+		}`
+		events, err := p.getS3Notifications(testEventViaSNS)
+		require.NoError(t, err)
+		assert.Equal(t, 0, len(events), "Test events via SNS should be skipped and return no events")
+	})
 }
 
 func TestNonRecoverableError(t *testing.T) {