Handling s3:TestEvent in aws_s3 source

[sean-snyk]

When using the S3 source, AWS sends a s3:TestEvent as the first message into the SQS queue (ref official docs, blog post).

The message looks like:

{
    "Messages": [
        {
            "MessageId": "c8f6eb9e-4c90-4154-9502-5a4d9dc5c78e",
            "ReceiptHandle": "XXX...XXX==",
            "MD5OfBody": "b33da647a485a3745e1111277ec590d2",
            "Body": "{\"Service\":\"Amazon S3\",\"Event\":\"s3:TestEvent\",\"Time\":\"2022-08-11T14:22:40.084Z\",\"Bucket\":\"vector-log-pipeline--xxx\",\"RequestId\":\"1YXFCPKNH9RYDP5P\",\"HostId\":\"Jq470yic8yEPATd6Hw3BmamYkq7l/vb9QKrIZWp46LnrGQmvDsGnitgmylQs/jfW7QbE2RkML1w=\"}"
        }
    ]
}

And vector doesn't like it, with our logs filling up with:

{
  "timestamp": "2022-08-15T09:18:16.410889Z",
  "level": "ERROR",
  "message": "Failed to process SQS message.",
  "message_id": "c8f6eb9e-4c90-4154-9502-5a4d9dc5c78e",
  "error": "Could not parse SQS message with id c8f6eb9e-4c90-4154-9502-5a4d9dc5c78e as S3 notification: missing field `Records` at line 1 column 247",
  "error_code": "failed_processing_sqs_message",
  "error_type": "parser_failed",
  "stage": "processing",
  "target": "vector::internal_events::aws_sqs::s3",
  "span": {
    "component_id": "s3",
    "component_kind": "source",
    "component_name": "s3",
    "component_type": "aws_s3",
    "name": "source"
  },
  "spans": [
    {
      "component_id": "s3",
      "component_kind": "source",
      "component_name": "s3",
      "component_type": "aws_s3",
      "name": "source"
    }
  ]
}

What options do we have for making vector handle it quietly/properly?

[bencord0]

Would something like this work?

i.e. Define a S3TestEvent struct and test if the event successfully deserializes, and skip processing early for this edge case.

diff --git a/src/sources/aws_s3/sqs.rs b/src/sources/aws_s3/sqs.rs
index 1b0bd2493..a143db7bd 100644
--- a/src/sources/aws_s3/sqs.rs
+++ b/src/sources/aws_s3/sqs.rs
@@ -374,7 +374,13 @@ impl IngestorProcess {
     }
 
     async fn handle_sqs_message(&mut self, message: Message) -> Result<(), ProcessingError> {
-        let s3_event: S3Event = serde_json::from_str(message.body.unwrap_or_default().as_ref())
+        let message_body = message.body.unwrap_or_default().as_ref();
+        if let Result<S3TestEvent>::Ok(s3_test_event) = serde_json::from_str(message_body) {
+            debug!(message = "Found S3 Test Event", ?message.message_id);
+            return Ok(())
+        };
+
+        let s3_event: S3Event = serde_json::from_str(message_body)
             .context(InvalidSqsMessageSnafu {
                 message_id: message.message_id.unwrap_or_else(|| "<empty>".to_owned()),
             })?;
@@ -594,6 +600,15 @@ impl IngestorProcess {
     }
 }
 
+// https://docs.aws.amazon.com/AmazonS3/latest/userguide/how-to-enable-disable-notification-intro.html
+#[derive(Clone, Debug, Deserialize)]
+#[serde(rename_all = "PascalCase")]
+pub struct S3TestEvent {
+    pub service: String,
+    pub event_name: S3EventName,
+    pub bucket: String,
+}
+
 // https://docs.aws.amazon.com/AmazonS3/latest/dev/notification-content-structure.html
 #[derive(Clone, Debug, Deserialize, Serialize)]
 #[serde(rename_all = "PascalCase")]

[jszwedko]

Hi @bencord0 and @sean-snyk !

Good catch. I think the changes proposed here make sense: consume and discard the test event with a debug message. Would you want to submit that as a PR?

My only suggestion is that we could model it by using an enum for serde to deserialize into like:

enum Message {
  TestEvent(S3TestEvent)
  Event(S3Event)
}

and then the caller could look like:

let s3_event: Message = serde_json::from_str(message_body)
             .context(InvalidSqsMessageSnafu {
                 message_id: message.message_id.unwrap_or_else(|| "<empty>".to_owned()),
             })?;

match s3_event {
  TestEvent(event) => { ... debug log ...},
  Event(event) => { ... handle event ... },
}

[bencord0]

Sorry for the delay, I needed to get some guidance wrt. the CLA. We're all good now, so I've filed this PR.
#14572