From 3d47ba71e8b21060b9fce9418b1532a1d063a0a7 Mon Sep 17 00:00:00 2001 From: Jakub Pavlik Date: Sat, 19 Sep 2026 13:49:43 +0200 Subject: [PATCH] fix(aws_s3 source): ignore CloudTrail log delivery notifications A trail's own SNS topic is often the same topic the bucket publishes S3 events to, so the queue receives a second message per log file shaped {"s3Bucket", "s3ObjectKey": [...]}. It matched no SqsEvent variant, so it was logged as a processing error and never deleted, and came back every visibility timeout until retention expired. Parse that shape as its own variant and treat it like s3:TestEvent: emit a warning and sqs_s3_event_record_ignored_total{ignore_type="cloudtrail_notification"}, then delete the message. The variant is last in the untagged enum, so every body that parsed before resolves to the same variant as before. The files are not fetched from it because the S3 event for the same file is already ingested. --- src/internal_events/aws_sqs.rs | 15 +++++ src/sources/aws_s3/sqs.rs | 104 ++++++++++++++++++++++++++++++++- 2 files changed, 118 insertions(+), 1 deletion(-) diff --git a/src/internal_events/aws_sqs.rs b/src/internal_events/aws_sqs.rs index b0c14493ea57a..d2e1e491133fa 100644 --- a/src/internal_events/aws_sqs.rs +++ b/src/internal_events/aws_sqs.rs @@ -372,3 +372,18 @@ impl InternalEvent for SqsS3EventRecordInvalidEventIgnored<'_> { .increment(1); } } + +#[derive(Debug, NamedInternalEvent)] +pub struct SqsCloudTrailNotificationIgnored<'a> { + pub bucket: &'a str, + pub object_count: usize, +} + +impl InternalEvent for SqsCloudTrailNotificationIgnored<'_> { + fn emit(self) { + warn!(message = "Ignored CloudTrail log delivery notification in SQS message; only S3 event notifications are ingested.", + bucket = %self.bucket, object_count = %self.object_count); + counter!("sqs_s3_event_record_ignored_total", "ignore_type" => "cloudtrail_notification") + .increment(1); + } +} diff --git a/src/sources/aws_s3/sqs.rs b/src/sources/aws_s3/sqs.rs index c8a922f7b69e0..c96ee99f00207 100644 --- a/src/sources/aws_s3/sqs.rs +++ b/src/sources/aws_s3/sqs.rs @@ -50,7 +50,7 @@ use crate::{ event::{BatchNotifier, BatchStatus, EstimatedJsonEncodedSizeOf, Event, LogEvent}, internal_events::{ EventsReceived, S3ObjectProcessingFailed, S3ObjectProcessingSucceeded, - SqsMessageDeleteBatchError, SqsMessageDeletePartialError, SqsMessageDeleteSucceeded, + SqsCloudTrailNotificationIgnored, SqsMessageDeleteBatchError, SqsMessageDeletePartialError, SqsMessageDeleteSucceeded, SqsMessageProcessingError, SqsMessageProcessingSucceeded, SqsMessageReceiveError, SqsMessageReceiveSucceeded, SqsMessageSendBatchError, SqsMessageSentPartialError, SqsMessageSentSucceeded, SqsS3EventRecordInvalidEventIgnored, StreamClosedError, @@ -628,6 +628,14 @@ impl IngestorProcess { Ok(()) } SqsEvent::Event(s3_event) => self.handle_s3_event(s3_event).await, + // Returning Ok deletes the message; left unparsed it is redelivered until retention expires. + SqsEvent::CloudTrailNotification(notification) => { + emit!(SqsCloudTrailNotificationIgnored { + bucket: ¬ification.s3_bucket, + object_count: notification.s3_object_key.len(), + }); + Ok(()) + } } } @@ -1003,6 +1011,17 @@ pub struct SnsNotification { enum SqsEvent { Event(S3Event), TestEvent(S3TestEvent), + // Keep last: untagged variants are tried in order, so existing matches are unchanged. + CloudTrailNotification(CloudTrailNotification), +} + +// Sent by a trail's own SNS topic, which customers often share with the bucket's S3 event topic. +// https://docs.aws.amazon.com/awscloudtrail/latest/userguide/configure-cloudtrail-to-send-notifications.html +#[derive(Clone, Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CloudTrailNotification { + pub s3_bucket: String, + pub s3_object_key: Vec, } #[derive(Clone, Debug, Deserialize)] @@ -1257,6 +1276,89 @@ fn test_s3_sns_testevent() { assert_eq!(value.event.name, "TestEvent".to_string()); } +#[test] +fn test_cloudtrail_notification() { + let value: SqsEvent = serde_json::from_str( + r#"{ + "s3Bucket":"bucketname", + "s3ObjectKey":["AWSLogs/123456789012/CloudTrail/us-east-1/2026/09/19/123456789012_CloudTrail_us-east-1_20260919T0410Z_abcdefgh.json.gz"] + }"#, + ) + .unwrap(); + + match value { + SqsEvent::CloudTrailNotification(notification) => { + assert_eq!(notification.s3_bucket, "bucketname".to_string()); + assert_eq!(notification.s3_object_key.len(), 1); + } + other => panic!("expected CloudTrailNotification, got {other:?}"), + } +} + +#[test] +fn test_sns_cloudtrail_notification() { + let sns_value: SnsNotification = serde_json::from_str( + r#"{ + "Type" : "Notification", + "MessageId" : "63a3f6b6-d533-4a47-aef9-fcf5cf758c76", + "TopicArn" : "arn:aws:sns:us-west-2:123456789012:MyTopic", + "Message" : "{\"s3Bucket\":\"bucketname\",\"s3ObjectKey\":[\"AWSLogs/123456789012/CloudTrail/us-east-1/2026/09/19/a.json.gz\",\"AWSLogs/123456789012/CloudTrail/us-east-1/2026/09/19/b.json.gz\"]}", + "Timestamp" : "2012-03-29T05:12:16.901Z", + "SignatureVersion" : "1", + "Signature" : "EXAMPLEnTrFPa3...", + "SigningCertURL" : "https://sns.us-west-2.amazonaws.com/SimpleNotificationService-f3ecfb7224c7233fe7bb5f59f96de52f.pem", + "UnsubscribeURL" : "https://sns.us-west-2.amazonaws.com/?Action=Unsubscribe&SubscriptionArn=arn:aws:sns:us-west-2:123456789012:MyTopic:c7fe3a54-ab0e-4ec2-88e0-db410a0f2bee" + }"#, + ).unwrap(); + + let value: SqsEvent = serde_json::from_str(sns_value.message.as_ref()).unwrap(); + + match value { + SqsEvent::CloudTrailNotification(notification) => { + assert_eq!(notification.s3_bucket, "bucketname".to_string()); + assert_eq!(notification.s3_object_key.len(), 2); + } + other => panic!("expected CloudTrailNotification, got {other:?}"), + } +} + +// The CloudTrail variant must never shadow the shapes that were already handled. +#[test] +fn test_sqs_event_variant_order() { + let s3_event: SqsEvent = serde_json::from_str( + r#"{ + "Records":[{ + "eventVersion":"2.1", + "eventSource":"aws:s3", + "awsRegion":"us-east-1", + "eventTime":"2026-09-19T04:10:00.000Z", + "eventName":"ObjectCreated:Put", + "s3":{ + "bucket":{"name":"bucketname"}, + "object":{"key":"AWSLogs/123456789012/CloudTrail/us-east-1/a.json.gz"} + } + }] + }"#, + ) + .unwrap(); + assert!(matches!(s3_event, SqsEvent::Event(_))); + + let test_event: SqsEvent = serde_json::from_str( + r#"{ + "Service":"Amazon S3", + "Event":"s3:TestEvent", + "Time":"2014-10-13T15:57:02.089Z", + "Bucket":"bucketname" + }"#, + ) + .unwrap(); + assert!(matches!(test_event, SqsEvent::TestEvent(_))); + + // Anything else must keep failing so it stays visible as a processing error. + assert!(serde_json::from_str::(r#"{"s3Bucket":"bucketname"}"#).is_err()); + assert!(serde_json::from_str::(r#"{"detail-type":"Object Created"}"#).is_err()); +} + #[test] fn parse_sqs_config() { let config: Config = toml::from_str(