forked from vectordotdev/vector
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat(new sink): add AWS Simple Notification Service
aws_sns sink (v…
…ectordotdev#18141) * ci: speed up tests by only running sqs tests * chore: prepare split in sqs and sns * chore: wrap up split for foundation * refactor: separate config * refactor: implement separate publisher * refactor: drop no longer needed message builder * refactor: extract error type * style: name variable correctly * feat: add sns 🎉 * refactor: abstract retry logic * chore: remove sqs from shared modules * chore: cleanup * chore: drop temporary changes * chore: update mod to include sns module * refactor: move healthcheck out of config into separate function * refactor: simplify request builder setup * refactor: message id * nit: make functions just visibile in sub module * fix: add pub(super) * fix: sub methods attributes pub(super) * Update Cargo.toml Co-authored-by: neuronull <neuronull@pm.me> * Update src/sinks/aws_s_s/mod.rs Co-authored-by: neuronull <neuronull@pm.me> * Update src/sinks/aws_s_s/retry.rs Co-authored-by: neuronull <neuronull@pm.me> * Update src/sinks/mod.rs Co-authored-by: neuronull <neuronull@pm.me> * Update src/sinks/aws_s_s/sns/config.rs Co-authored-by: neuronull <neuronull@pm.me> * chore: first bunch of comments * chore: second bunch of comments * chore: enable all integration tests * fix: move test * fix: dead_code warning * fix: dead code warning * docs: autogenerated aws_sns docs * fix: move region serde to downstream impl * update licenses --------- Co-authored-by: Kristof Herrmann <kristof.herrmann@rwth-aachen.de> Co-authored-by: ArzelaAscoIi <37148029+ArzelaAscoIi@users.noreply.github.com> Co-authored-by: neuronull <neuronull@pm.me>
- Loading branch information
4 people
committed
Aug 18, 2023
1 parent
833ac19
commit 7b2bddc
Showing
26 changed files
with
1,464 additions
and
341 deletions.
There are no files selected for viewing
Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,14 @@ | ||
| use super::{request_builder::SendMessageEntry, service::SendMessageResponse}; | ||
| use aws_sdk_sqs::types::SdkError; | ||
|
|
||
| #[async_trait::async_trait] | ||
| pub(super) trait Client<R> | ||
| where | ||
| R: std::fmt::Debug + std::fmt::Display + std::error::Error, | ||
| { | ||
| async fn send_message( | ||
| &self, | ||
| entry: SendMessageEntry, | ||
| byte_size: usize, | ||
| ) -> Result<SendMessageResponse, SdkError<R>>; | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,99 @@ | ||
| use std::convert::TryFrom; | ||
|
|
||
| use snafu::{ResultExt, Snafu}; | ||
|
|
||
| use vector_config::configurable_component; | ||
|
|
||
| use crate::{ | ||
| aws::AwsAuthentication, | ||
| codecs::EncodingConfig, | ||
| config::AcknowledgementsConfig, | ||
| sinks::util::TowerRequestConfig, | ||
| template::{Template, TemplateParseError}, | ||
| tls::TlsConfig, | ||
| }; | ||
|
|
||
| #[derive(Debug, Snafu)] | ||
| pub(super) enum BuildError { | ||
| #[snafu(display("`message_group_id` should be defined for FIFO queue."))] | ||
| MessageGroupIdMissing, | ||
| #[snafu(display("`message_group_id` is not allowed with non-FIFO queue."))] | ||
| MessageGroupIdNotAllowed, | ||
| #[snafu(display("invalid topic template: {}", source))] | ||
| TopicTemplate { source: TemplateParseError }, | ||
| #[snafu(display("invalid message_deduplication_id template: {}", source))] | ||
| MessageDeduplicationIdTemplate { source: TemplateParseError }, | ||
| } | ||
|
|
||
| /// Base Configuration `aws_s_s` for sns and sqs sink. | ||
| #[configurable_component] | ||
| #[derive(Clone, Debug)] | ||
| #[serde(deny_unknown_fields)] | ||
| pub(super) struct BaseSSSinkConfig { | ||
| #[configurable(derived)] | ||
| pub(super) encoding: EncodingConfig, | ||
|
|
||
| /// The tag that specifies that a message belongs to a specific message group. | ||
| /// | ||
| /// Can be applied only to FIFO queues. | ||
| #[configurable(metadata(docs::examples = "vector"))] | ||
| #[configurable(metadata(docs::examples = "vector-%Y-%m-%d"))] | ||
| pub(super) message_group_id: Option<String>, | ||
|
|
||
| /// The message deduplication ID value to allow AWS to identify duplicate messages. | ||
| /// | ||
| /// This value is a template which should result in a unique string for each event. See the [AWS | ||
| /// documentation][deduplication_id_docs] for more about how AWS does message deduplication. | ||
| /// | ||
| /// [deduplication_id_docs]: https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/using-messagededuplicationid-property.html | ||
| #[configurable(metadata(docs::examples = "{{ transaction_id }}"))] | ||
| pub(super) message_deduplication_id: Option<String>, | ||
|
|
||
| #[configurable(derived)] | ||
| #[serde(default)] | ||
| pub(super) request: TowerRequestConfig, | ||
|
|
||
| #[configurable(derived)] | ||
| pub(super) tls: Option<TlsConfig>, | ||
|
|
||
| /// The ARN of an [IAM role][iam_role] to assume at startup. | ||
| /// | ||
| /// [iam_role]: https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles.html | ||
| #[configurable(deprecated)] | ||
| #[configurable(metadata(docs::hidden))] | ||
| pub(super) assume_role: Option<String>, | ||
|
|
||
| #[configurable(derived)] | ||
| #[serde(default)] | ||
| pub(super) auth: AwsAuthentication, | ||
|
|
||
| #[configurable(derived)] | ||
| #[serde( | ||
| default, | ||
| deserialize_with = "crate::serde::bool_or_struct", | ||
| skip_serializing_if = "crate::serde::skip_serializing_if_default" | ||
| )] | ||
| pub(super) acknowledgements: AcknowledgementsConfig, | ||
| } | ||
|
|
||
| pub(super) fn message_group_id( | ||
| message_group_id: Option<String>, | ||
| fifo: bool, | ||
| ) -> crate::Result<Option<Template>> { | ||
| match (message_group_id.as_ref(), fifo) { | ||
| (Some(value), true) => Ok(Some( | ||
| Template::try_from(value.clone()).context(TopicTemplateSnafu)?, | ||
| )), | ||
| (Some(_), false) => Err(Box::new(BuildError::MessageGroupIdNotAllowed)), | ||
| (None, true) => Err(Box::new(BuildError::MessageGroupIdMissing)), | ||
| (None, false) => Ok(None), | ||
| } | ||
| } | ||
| pub(super) fn message_deduplication_id( | ||
| message_deduplication_id: Option<String>, | ||
| ) -> crate::Result<Option<Template>> { | ||
| Ok(message_deduplication_id | ||
| .clone() | ||
| .map(Template::try_from) | ||
| .transpose()?) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,12 @@ | ||
| mod client; | ||
| mod config; | ||
| mod request_builder; | ||
| mod retry; | ||
| mod service; | ||
| mod sink; | ||
|
|
||
| #[cfg(feature = "sinks-aws_sqs")] | ||
| mod sqs; | ||
|
|
||
| #[cfg(feature = "sinks-aws_sns")] | ||
| mod sns; |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,44 @@ | ||
| use aws_sdk_sqs::types::SdkError; | ||
| use std::marker::PhantomData; | ||
|
|
||
| use super::service::SendMessageResponse; | ||
| use crate::{aws::is_retriable_error, sinks::util::retries::RetryLogic}; | ||
|
|
||
| #[derive(Debug)] | ||
| pub(super) struct SSRetryLogic<E> { | ||
| _phantom: PhantomData<fn() -> E>, | ||
| } | ||
|
|
||
| impl<E> SSRetryLogic<E> | ||
| where | ||
| E: std::fmt::Debug + std::fmt::Display + std::error::Error + Sync + Send + 'static, | ||
| { | ||
| pub(super) fn new() -> SSRetryLogic<E> { | ||
| Self { | ||
| _phantom: PhantomData, | ||
| } | ||
| } | ||
| } | ||
|
|
||
| impl<E> RetryLogic for SSRetryLogic<E> | ||
| where | ||
| E: std::fmt::Debug + std::fmt::Display + std::error::Error + Sync + Send + 'static, | ||
| { | ||
| type Error = SdkError<E>; | ||
| type Response = SendMessageResponse; | ||
|
|
||
| fn is_retriable_error(&self, error: &Self::Error) -> bool { | ||
| is_retriable_error(error) | ||
| } | ||
| } | ||
|
|
||
| impl<E> Clone for SSRetryLogic<E> | ||
| where | ||
| E: std::fmt::Debug + std::fmt::Display + std::error::Error + Sync + Send + 'static, | ||
| { | ||
| fn clone(&self) -> SSRetryLogic<E> { | ||
| SSRetryLogic { | ||
| _phantom: PhantomData, | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.