Add support in aws-s3 input for s3 notification from SNS to SQS

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

@@ -53,8 +53,12 @@ func nonRetryableErrorWrap(err error) error {
 // s3EventsV2 is the notification message that Amazon S3 sends to notify of S3 changes.
 // This was derived from the version 2.2 schema.
 // 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.
 type s3EventsV2 struct {
- Records []s3EventV2 `json:"Records"`
+ TopicArn string `json:"TopicArn"`
+ Message string `json:"Message"`
+ Records []s3EventV2 `json:"Records"`
 }
 
 // s3EventV2 is a S3 change notification event.
@@ -189,6 +193,18 @@ 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 the notification is from S3 -> SNS -> SQS
+ if events.TopicArn != "" {
+ dec := json.NewDecoder(strings.NewReader(events.Message))
+ if err := dec.Decode(&events); 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)
+ }
+ }
+ return p.getS3Info(events)
+}
+
+func (p *sqsS3EventProcessor) getS3Info(events s3EventsV2) ([]s3EventV2, error) {
  var out []s3EventV2
  for _, record := range events.Records {
  if !p.isObjectCreatedEvents(record) {
@@ -211,7 +227,6 @@ func (p *sqsS3EventProcessor) getS3Notifications(body string) ([]s3EventV2, erro
 
  out = append(out, record)
  }
-
  return out, nil
 }
 

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

@@ -184,6 +184,16 @@ func TestSqsProcessor_getS3Notifications(t *testing.T) {
  require.NoError(t, err)
  assert.Len(t, events, 0)
  })
+
+ t.Run("sns-sqs notification", func(t *testing.T) {
+ msg := newSNSSQSMessage()
+ events, err := p.getS3Notifications(*msg.Body)
+ require.NoError(t, err)
+ assert.Len(t, events, 1)
+ assert.Equal(t, "test-object-key", events[0].S3.Object.Key)
+ assert.Equal(t, "arn:aws:s3:::vpc-flow-logs-ks", events[0].S3.Bucket.ARN)
+ assert.Equal(t, "vpc-flow-logs-ks", events[0].S3.Bucket.Name)
+ })
 }
 
 func TestNonRecoverableError(t *testing.T) {

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

@@ -126,6 +126,28 @@ func newSQSMessage(events ...s3EventV2) sqs.Message {
  }
 }
 
+func newSNSSQSMessage() sqs.Message {
+ body, err := json.Marshal(s3EventsV2{
+ TopicArn: "arn:aws:sns:us-east-1:1234:sns-topic",
+ Message: "{\"Records\":[{\"eventSource\":\"aws:s3\",\"awsRegion\":\"us-east-1\",\"eventName\":\"ObjectCreated:Put\",\"s3\":{\"configurationId\":\"sns-notification-vpc-flow-logs\",\"bucket\":{\"name\":\"vpc-flow-logs-ks\",\"arn\":\"arn:aws:s3:::vpc-flow-logs-ks\"},\"object\":{\"key\":\"test-object-key\"}}}]}",
+ })
+ if err != nil {
+ panic(err)
+ }
+
+ hash := sha256.Sum256(body)
+ id, _ := uuid.FromBytes(hash[:16])
+ messageID := id.String()
+ receipt := "receipt-" + messageID
+ bodyStr := string(body)
+
+ return sqs.Message{
+ Body: &bodyStr,
+ MessageId: &messageID,
+ ReceiptHandle: &receipt,
+ }
+}
+
 func newS3Event(key string) s3EventV2 {
  record := s3EventV2{
  AWSRegion: "us-east-1",