From 53343161692300aa4a2d0adf349d80d4e059a3ce Mon Sep 17 00:00:00 2001 From: sundaresanr Date: Mon, 2 Nov 2026 13:57:49 -0800 Subject: [PATCH] feat(aws_s3 source): parse EventBridge-format S3 events from SQS When a bucket routes object events through EventBridge to SQS, each message arrives in EventBridge's envelope instead of the native S3 `Records` format. The parser rejected these messages and, by default, deleted them without reading the object. Parse the EventBridge shape and convert it into the existing S3 event record, so everything after parsing is shared with native notifications. --- src/sources/aws_s3/sqs.rs | 178 ++++++++++++++++++++++++++++++++++++-- 1 file changed, 172 insertions(+), 6 deletions(-) diff --git a/src/sources/aws_s3/sqs.rs b/src/sources/aws_s3/sqs.rs index cf05b13e3309b..6ba95c779db52 100644 --- a/src/sources/aws_s3/sqs.rs +++ b/src/sources/aws_s3/sqs.rs @@ -64,6 +64,10 @@ use crate::{ static SUPPORTED_S3_EVENT_VERSION: LazyLock = LazyLock::new(|| semver::VersionReq::parse("~2").unwrap()); +// EventBridge versions `detail` independently of native notifications; currently 1.x. +static SUPPORTED_EVENTBRIDGE_S3_EVENT_VERSION: LazyLock = + LazyLock::new(|| semver::VersionReq::parse("~1").unwrap()); + /// Configuration for deferring events based on their age. #[serde_as] #[configurable_component] @@ -652,29 +656,43 @@ impl IngestorProcess { Ok(()) } SqsEvent::Event(s3_event) => self.handle_s3_event(s3_event).await, + SqsEvent::EventBridge(eb_event) => self.handle_eventbridge_event(eb_event).await, } } async fn handle_s3_event(&mut self, s3_event: S3Event) -> Result<(), ProcessingError> { for record in s3_event.records { + let event_version: semver::Version = record.event_version.clone().into(); + if !SUPPORTED_S3_EVENT_VERSION.matches(&event_version) { + return Err(ProcessingError::UnsupportedS3EventVersion { + version: event_version, + }); + } self.handle_s3_event_record(record, self.log_namespace) .await? } Ok(()) } - async fn handle_s3_event_record( + async fn handle_eventbridge_event( &mut self, - s3_event: S3EventRecord, - log_namespace: LogNamespace, + eb_event: EventBridgeS3Event, ) -> Result<(), ProcessingError> { - let event_version: semver::Version = s3_event.event_version.clone().into(); - if !SUPPORTED_S3_EVENT_VERSION.matches(&event_version) { + let event_version: semver::Version = eb_event.detail.event_version.clone().into(); + if !SUPPORTED_EVENTBRIDGE_S3_EVENT_VERSION.matches(&event_version) { return Err(ProcessingError::UnsupportedS3EventVersion { - version: event_version.clone(), + version: event_version, }); } + self.handle_s3_event_record(eb_event.into(), self.log_namespace) + .await + } + async fn handle_s3_event_record( + &mut self, + s3_event: S3EventRecord, + log_namespace: LogNamespace, + ) -> Result<(), ProcessingError> { if s3_event.event_name.kind != "ObjectCreated" { emit!(SqsS3EventRecordInvalidEventIgnored { bucket: &s3_event.s3.bucket.name, @@ -1026,9 +1044,52 @@ pub struct SnsNotification { #[serde(untagged)] enum SqsEvent { Event(S3Event), + EventBridge(EventBridgeS3Event), TestEvent(S3TestEvent), } +// https://docs.aws.amazon.com/AmazonS3/latest/userguide/ev-events.html +#[derive(Clone, Debug, Deserialize)] +#[serde(rename_all = "kebab-case")] +pub struct EventBridgeS3Event { + pub detail_type: String, + pub source: String, + pub region: String, + pub time: DateTime, + pub detail: EventBridgeS3Detail, +} + +#[derive(Clone, Debug, Deserialize)] +#[serde(rename_all = "kebab-case")] +pub struct EventBridgeS3Detail { + pub event_version: S3EventVersion, + pub bucket: S3Bucket, + pub object: S3Object, + #[serde(default)] + pub reason: String, +} + +impl From for S3EventRecord { + fn from(event: EventBridgeS3Event) -> Self { + // "Object Created" -> "ObjectCreated", matching the native event-name kinds. + let kind = event.detail_type.replace(' ', ""); + S3EventRecord { + event_version: event.detail.event_version, + event_source: event.source, + aws_region: event.region, + event_name: S3EventName { + kind, + name: event.detail.reason, + }, + event_time: event.time, + s3: S3Message { + bucket: event.detail.bucket, + object: event.detail.object, + }, + } + } +} + #[derive(Clone, Debug, Deserialize)] #[serde(rename_all = "PascalCase")] pub struct S3TestEvent { @@ -1346,3 +1407,108 @@ fn test_unrecognized_notification_fails_to_parse() { assert!(serde_json::from_str::(body).is_err()); } + +#[cfg(test)] +fn eventbridge_body(detail_type: &str, key: &str, reason: &str) -> String { + format!( + r#"{{ + "version": "0", + "id": "17793124-05d4-b198-2fde-7ededc63b103", + "detail-type": "{detail_type}", + "source": "aws.s3", + "account": "111122223333", + "time": "2021-11-12T00:00:00Z", + "region": "ca-central-1", + "resources": ["arn:aws:s3:::amzn-s3-demo-bucket1"], + "detail": {{ + "version": "0", + "event-version": "1.2", + "bucket": {{ + "name": "amzn-s3-demo-bucket1", + "aws-generated-tags": {{"aws:cloudformation:logical-id": "S3DemoBucket"}} + }}, + "object": {{ + "key": "{key}", + "size": 5, + "etag": "b1946ac92492d2347c6235b4d2611184", + "version-id": "IYV3p45BT0ac8hjHg1houSdS1a.Mro8e", + "sequencer": "617f08299329d189" + }}, + "request-id": "N4N7GDK58NMKJ12R", + "requester": "123456789012", + "source-ip-address": "192.0.2.1", + "reason": "{reason}" + }} + }}"# + ) +} + +#[test] +fn test_eventbridge_object_created() { + let body = eventbridge_body("Object Created", "logs/example-key.json.gz", "PutObject"); + let SqsEvent::EventBridge(event) = serde_json::from_str::(&body).unwrap() else { + panic!("expected an EventBridge event"); + }; + assert_eq!(event.detail.event_version.major, 1); + + let record: S3EventRecord = event.into(); + assert_eq!(record.event_name.kind, "ObjectCreated"); + assert_eq!(record.event_name.name, "PutObject"); + assert_eq!(record.event_source, "aws.s3"); + assert_eq!(record.aws_region, "ca-central-1"); + assert_eq!( + record.event_time, + Utc.with_ymd_and_hms(2021, 11, 12, 0, 0, 0).unwrap() + ); + assert_eq!(record.s3.bucket.name, "amzn-s3-demo-bucket1"); + assert_eq!(record.s3.object.key, "logs/example-key.json.gz"); +} + +#[test] +fn test_eventbridge_object_deleted() { + let body = eventbridge_body("Object Deleted", "example-key", "DeleteObject"); + let SqsEvent::EventBridge(event) = serde_json::from_str::(&body).unwrap() else { + panic!("expected an EventBridge event"); + }; + let record: S3EventRecord = event.into(); + assert_eq!(record.event_name.kind, "ObjectDeleted"); +} + +#[test] +fn test_eventbridge_restore_without_reason() { + let body = eventbridge_body("Object Restore Completed", "example-key", "") + .replace(r#""reason": """#, r#""source-storage-class": "GLACIER""#); + let SqsEvent::EventBridge(event) = serde_json::from_str::(&body).unwrap() else { + panic!("expected an EventBridge event"); + }; + let record: S3EventRecord = event.into(); + assert_eq!(record.event_name.kind, "ObjectRestoreCompleted"); + assert_eq!(record.event_name.name, ""); +} + +#[test] +fn test_eventbridge_key_is_url_decoded() { + let body = eventbridge_body("Object Created", "a+b%2Bc%25d", "PutObject"); + let SqsEvent::EventBridge(event) = serde_json::from_str::(&body).unwrap() else { + panic!("expected an EventBridge event"); + }; + assert_eq!(event.detail.object.key, "a b+c%d"); +} + +#[test] +fn test_eventbridge_via_sns() { + let inner = eventbridge_body("Object Created", "example-key", "PutObject"); + let sns_body = serde_json::json!({ + "Type": "Notification", + "MessageId": "82833b5c-8d5d-56d0-b0e1-7511f8253eb8", + "TopicArn": "arn:aws:sns:us-east-1:111122223333:Topic", + "Message": inner, + "Timestamp": "2021-11-12T00:00:01.000Z", + }) + .to_string(); + let sns: SnsNotification = serde_json::from_str(&sns_body).unwrap(); + assert!(matches!( + serde_json::from_str::(&sns.message).unwrap(), + SqsEvent::EventBridge(_) + )); +}