enhancement(aws_s3 source): optionally configure source to defer or delete older notifications (#22691)

* make configurable to defer or delete old messages in queue

* formatting

* update docs

* name field more inline with other fields (poll_secs)

* add changelog

* avoid unwrap

* rename to match config of deferred instead of retry

* nest deferred configurations under deferred.  require them to be together

* correct changelog

* add ticks to emphasize the variable names

* remove panic.  update send_message to require a target queue_url

* internal_log_rate_limit & fmt

Author: akutta

changelog.d/s3-source-defer-or-delete.enhancement.md

@@ -0,0 +1,3 @@
+Add deferred processing options (`deferred.max_age_secs`, `deferred.queue_url`) to automatically route older event notifications to a separate queue, enabling prioritized processing of recent files
+
+authors: akutta

src/internal_events/aws_sqs.rs

@@ -9,6 +9,7 @@ use vector_lib::internal_event::{error_stage, error_type};
 mod s3 {
     use aws_sdk_sqs::types::{
         BatchResultErrorEntry, DeleteMessageBatchRequestEntry, DeleteMessageBatchResultEntry,
+        SendMessageBatchRequestEntry, SendMessageBatchResultEntry,
     };
 
     use super::*;
@@ -114,6 +115,80 @@ mod s3 {
             .increment(1);
         }
     }
+
+    #[derive(Debug)]
+    pub struct SqsMessageSentSucceeded {
+        pub message_ids: Vec<SendMessageBatchResultEntry>,
+    }
+
+    impl InternalEvent for SqsMessageSentSucceeded {
+        fn emit(self) {
+            trace!(message = "Deferred SQS message(s).",
+            message_ids = %self.message_ids.iter()
+                .map(|x| x.id.as_str())
+                .collect::<Vec<_>>()
+                .join(", "));
+            counter!("sqs_message_defer_succeeded_total").increment(self.message_ids.len() as u64);
+        }
+    }
+
+    #[derive(Debug)]
+    pub struct SqsMessageSentPartialError {
+        pub entries: Vec<BatchResultErrorEntry>,
+    }
+
+    impl InternalEvent for SqsMessageSentPartialError {
+        fn emit(self) {
+            error!(
+                message = "Sending of deferred SQS message(s) failed.",
+                message_ids = %self.entries.iter()
+                    .map(|x| format!("{}/{}", x.id, x.code))
+                    .collect::<Vec<_>>()
+                    .join(", "),
+                error_code = "failed_deferring_some_sqs_messages",
+                error_type = error_type::ACKNOWLEDGMENT_FAILED,
+                stage = error_stage::PROCESSING,
+                internal_log_rate_limit = true,
+            );
+            counter!(
+                "component_errors_total",
+                "error_code" => "failed_deferring_some_sqs_messages",
+                "error_type" => error_type::ACKNOWLEDGMENT_FAILED,
+                "stage" => error_stage::PROCESSING,
+            )
+            .increment(1);
+        }
+    }
+
+    #[derive(Debug)]
+    pub struct SqsMessageSendBatchError<E> {
+        pub entries: Vec<SendMessageBatchRequestEntry>,
+        pub error: E,
+    }
+
+    impl<E: std::fmt::Display> InternalEvent for SqsMessageSendBatchError<E> {
+        fn emit(self) {
+            error!(
+                message = "Sending of deferred SQS message(s) failed.",
+                message_ids = %self.entries.iter()
+                    .map(|x| x.id.as_str())
+                    .collect::<Vec<_>>()
+                    .join(", "),
+                error = %self.error,
+                error_code = "failed_deferring_all_sqs_messages",
+                error_type = error_type::ACKNOWLEDGMENT_FAILED,
+                stage = error_stage::PROCESSING,
+                internal_log_rate_limit = true,
+            );
+            counter!(
+                "component_errors_total",
+                "error_code" => "failed_deferring_all_sqs_messages",
+                "error_type" => error_type::ACKNOWLEDGMENT_FAILED,
+                "stage" => error_stage::PROCESSING,
+            )
+            .increment(1);
+        }
+    }
 }
 
 #[derive(Debug)]