diff --git a/Cargo.lock b/Cargo.lock
index 391d78773fb7c..f12099842b7e9 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -982,6 +982,31 @@ dependencies = [
  "url",
 ]
 
+[[package]]
+name = "aws-sdk-sns"
+version = "0.24.0"
+source = "git+https://github.com/vectordotdev/aws-sdk-rust?rev=3d6aefb7fcfced5fc2a7e761a87e4ddbda1ee670#3d6aefb7fcfced5fc2a7e761a87e4ddbda1ee670"
+dependencies = [
+ "aws-credential-types",
+ "aws-endpoint",
+ "aws-http",
+ "aws-sig-auth",
+ "aws-smithy-async",
+ "aws-smithy-client",
+ "aws-smithy-http",
+ "aws-smithy-http-tower",
+ "aws-smithy-json",
+ "aws-smithy-query",
+ "aws-smithy-types",
+ "aws-smithy-xml",
+ "aws-types",
+ "bytes 1.4.0",
+ "http",
+ "regex",
+ "tokio-stream",
+ "tower",
+]
+
 [[package]]
 name = "aws-sdk-sqs"
 version = "0.24.0"
@@ -9492,6 +9517,7 @@ dependencies = [
  "aws-sdk-firehose",
  "aws-sdk-kinesis",
  "aws-sdk-s3",
+ "aws-sdk-sns",
  "aws-sdk-sqs",
  "aws-sigv4",
  "aws-smithy-async",
diff --git a/Cargo.toml b/Cargo.toml
index 531cc7ba0afe4..a11fd043d53fc 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -169,6 +169,7 @@ metrics-tracing-context = { version = "0.14.0", default-features = false }
 # depending on a fork to circumvent https://github.com/awslabs/aws-sdk-rust/issues/749
 aws-sdk-s3 = { git = "https://github.com/vectordotdev/aws-sdk-rust", rev = "3d6aefb7fcfced5fc2a7e761a87e4ddbda1ee670", default-features = false, features = ["native-tls"], optional = true }
 aws-sdk-sqs = { git = "https://github.com/vectordotdev/aws-sdk-rust", rev = "3d6aefb7fcfced5fc2a7e761a87e4ddbda1ee670", default-features = false, features = ["native-tls"], optional = true }
+aws-sdk-sns = { git = "https://github.com/vectordotdev/aws-sdk-rust", rev = "3d6aefb7fcfced5fc2a7e761a87e4ddbda1ee670", default-features = false, features = ["native-tls"], optional = true }
 aws-sdk-cloudwatch = { git = "https://github.com/vectordotdev/aws-sdk-rust", rev = "3d6aefb7fcfced5fc2a7e761a87e4ddbda1ee670", default-features = false, features = ["native-tls"], optional = true }
 aws-sdk-cloudwatchlogs = { git = "https://github.com/vectordotdev/aws-sdk-rust", rev = "3d6aefb7fcfced5fc2a7e761a87e4ddbda1ee670", default-features = false, features = ["native-tls"], optional = true }
 aws-sdk-elasticsearch = { git = "https://github.com/vectordotdev/aws-sdk-rust", rev = "3d6aefb7fcfced5fc2a7e761a87e4ddbda1ee670", default-features = false, features = ["native-tls"], optional = true }
@@ -612,6 +613,7 @@ sinks-logs = [
   "sinks-aws_kinesis_streams",
   "sinks-aws_s3",
   "sinks-aws_sqs",
+  "sinks-aws_sns",
   "sinks-axiom",
   "sinks-azure_blob",
   "sinks-azure_monitor_logs",
@@ -671,6 +673,7 @@ sinks-aws_kinesis_firehose = ["aws-core", "dep:aws-sdk-firehose"]
 sinks-aws_kinesis_streams = ["aws-core", "dep:aws-sdk-kinesis"]
 sinks-aws_s3 = ["dep:base64", "dep:md-5", "aws-core", "dep:aws-sdk-s3"]
 sinks-aws_sqs = ["aws-core", "dep:aws-sdk-sqs"]
+sinks-aws_sns = ["aws-core", "dep:aws-sdk-sns"]
 sinks-axiom = ["sinks-elasticsearch"]
 sinks-azure_blob = ["dep:azure_core", "dep:azure_identity", "dep:azure_storage", "dep:azure_storage_blobs"]
 sinks-azure_monitor_logs = []
@@ -779,6 +782,7 @@ aws-integration-tests = [
   "aws-kinesis-streams-integration-tests",
   "aws-s3-integration-tests",
   "aws-sqs-integration-tests",
+  "aws-sns-integration-tests",
 ]
 
 azure-integration-tests = [
@@ -792,7 +796,8 @@ aws-ecs-metrics-integration-tests = ["sources-aws_ecs_metrics"]
 aws-kinesis-firehose-integration-tests = ["sinks-aws_kinesis_firehose", "dep:aws-sdk-elasticsearch", "sinks-elasticsearch"]
 aws-kinesis-streams-integration-tests = ["sinks-aws_kinesis_streams"]
 aws-s3-integration-tests = ["sinks-aws_s3", "sources-aws_s3"]
-aws-sqs-integration-tests = ["sinks-aws_sqs", "sources-aws_sqs"]
+aws-sqs-integration-tests = ["sinks-aws_sqs"]
+aws-sns-integration-tests = ["sinks-aws_sns"]
 axiom-integration-tests = ["sinks-axiom"]
 azure-blob-integration-tests = ["sinks-azure_blob"]
 chronicle-integration-tests = ["sinks-gcp"]
diff --git a/LICENSE-3rdparty.csv b/LICENSE-3rdparty.csv
index b6d5a876eae7e..8ad918b9800e7 100644
--- a/LICENSE-3rdparty.csv
+++ b/LICENSE-3rdparty.csv
@@ -53,6 +53,7 @@ aws-sdk-cloudwatchlogs,https://github.com/awslabs/aws-sdk-rust,Apache-2.0,"AWS R
 aws-sdk-firehose,https://github.com/awslabs/aws-sdk-rust,Apache-2.0,"AWS Rust SDK Team <aws-sdk-rust@amazon.com>, Russell Cohen <rcoh@amazon.com>"
 aws-sdk-kinesis,https://github.com/awslabs/aws-sdk-rust,Apache-2.0,"AWS Rust SDK Team <aws-sdk-rust@amazon.com>, Russell Cohen <rcoh@amazon.com>"
 aws-sdk-s3,https://github.com/awslabs/aws-sdk-rust,Apache-2.0,"AWS Rust SDK Team <aws-sdk-rust@amazon.com>, Russell Cohen <rcoh@amazon.com>"
+aws-sdk-sns,https://github.com/awslabs/aws-sdk-rust,Apache-2.0,"AWS Rust SDK Team <aws-sdk-rust@amazon.com>, Russell Cohen <rcoh@amazon.com>"
 aws-sdk-sqs,https://github.com/awslabs/aws-sdk-rust,Apache-2.0,"AWS Rust SDK Team <aws-sdk-rust@amazon.com>, Russell Cohen <rcoh@amazon.com>"
 aws-sdk-sso,https://github.com/awslabs/aws-sdk-rust,Apache-2.0,"AWS Rust SDK Team <aws-sdk-rust@amazon.com>, Russell Cohen <rcoh@amazon.com>"
 aws-sdk-sts,https://github.com/awslabs/aws-sdk-rust,Apache-2.0,"AWS Rust SDK Team <aws-sdk-rust@amazon.com>, Russell Cohen <rcoh@amazon.com>"
diff --git a/scripts/integration/aws/compose.yaml b/scripts/integration/aws/compose.yaml
index 07ed1fde4311d..e635c31f12a2f 100644
--- a/scripts/integration/aws/compose.yaml
+++ b/scripts/integration/aws/compose.yaml
@@ -6,7 +6,7 @@ services:
   mock-localstack:
     image: docker.io/localstack/localstack-full:0.11.6
     environment:
-    - SERVICES=kinesis,s3,cloudwatch,elasticsearch,es,firehose,sqs
+    - SERVICES=kinesis,s3,cloudwatch,elasticsearch,es,firehose,sqs,sns
   mock-watchlogs:
     image: docker.io/luciofranco/mockwatchlogs:latest
   mock-ecs:
diff --git a/scripts/integration/aws/test.yaml b/scripts/integration/aws/test.yaml
index 6e554fca22c1a..466eb9cfd172e 100644
--- a/scripts/integration/aws/test.yaml
+++ b/scripts/integration/aws/test.yaml
@@ -13,6 +13,7 @@ env:
   KINESIS_ADDRESS: http://mock-localstack:4566
   S3_ADDRESS: http://mock-localstack:4566
   SQS_ADDRESS: http://mock-localstack:4566
+  SNS_ADDRESS: http://mock-localstack:4566
   WATCHLOGS_ADDRESS: http://mock-watchlogs:6000
 
 matrix:
diff --git a/src/sinks/aws_s_s/client.rs b/src/sinks/aws_s_s/client.rs
new file mode 100644
index 0000000000000..88c86f5f9631b
--- /dev/null
+++ b/src/sinks/aws_s_s/client.rs
@@ -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>>;
+}
diff --git a/src/sinks/aws_s_s/config.rs b/src/sinks/aws_s_s/config.rs
new file mode 100644
index 0000000000000..aaf0a112d2f6b
--- /dev/null
+++ b/src/sinks/aws_s_s/config.rs
@@ -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()?)
+}
diff --git a/src/sinks/aws_s_s/mod.rs b/src/sinks/aws_s_s/mod.rs
new file mode 100644
index 0000000000000..98f14b44f9bde
--- /dev/null
+++ b/src/sinks/aws_s_s/mod.rs
@@ -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;
diff --git a/src/sinks/aws_sqs/request_builder.rs b/src/sinks/aws_s_s/request_builder.rs
similarity index 72%
rename from src/sinks/aws_sqs/request_builder.rs
rename to src/sinks/aws_s_s/request_builder.rs
index 34f54b5c1c0a2..fb977244e932c 100644
--- a/src/sinks/aws_sqs/request_builder.rs
+++ b/src/sinks/aws_s_s/request_builder.rs
@@ -2,7 +2,7 @@ use bytes::Bytes;
 use vector_common::request_metadata::{MetaDescriptive, RequestMetadata};
 use vector_core::ByteSizeOf;
 
-use super::config::SqsSinkConfig;
+use crate::codecs::EncodingConfig;
 use crate::{
     codecs::{Encoder, Transformer},
     event::{Event, EventFinalizers, Finalizable},
@@ -15,37 +15,39 @@ use crate::{
 };
 
 #[derive(Clone)]
-pub struct SqsMetadata {
-    pub finalizers: EventFinalizers,
-    pub message_group_id: Option<String>,
-    pub message_deduplication_id: Option<String>,
+pub(super) struct SSMetadata {
+    pub(super) finalizers: EventFinalizers,
+    pub(super) message_group_id: Option<String>,
+    pub(super) message_deduplication_id: Option<String>,
 }
 
 #[derive(Clone)]
-pub(crate) struct SqsRequestBuilder {
+pub(super) struct SSRequestBuilder {
     encoder: (Transformer, Encoder<()>),
     message_group_id: Option<Template>,
     message_deduplication_id: Option<Template>,
-    queue_url: String,
 }
 
-impl SqsRequestBuilder {
-    pub fn new(config: SqsSinkConfig) -> crate::Result<Self> {
-        let transformer = config.encoding.transformer();
-        let serializer = config.encoding.build()?;
+impl SSRequestBuilder {
+    pub(super) fn new(
+        message_group_id: Option<Template>,
+        message_deduplication_id: Option<Template>,
+        encoding_config: EncodingConfig,
+    ) -> crate::Result<Self> {
+        let transformer = encoding_config.transformer();
+        let serializer = encoding_config.build()?;
         let encoder = Encoder::<()>::new(serializer);
 
         Ok(Self {
             encoder: (transformer, encoder),
-            message_group_id: config.message_group_id()?,
-            message_deduplication_id: config.message_deduplication_id()?,
-            queue_url: config.queue_url,
+            message_group_id,
+            message_deduplication_id,
         })
     }
 }
 
-impl RequestBuilder<Event> for SqsRequestBuilder {
-    type Metadata = SqsMetadata;
+impl RequestBuilder<Event> for SSRequestBuilder {
+    type Metadata = SSMetadata;
     type Events = Event;
     type Encoder = (Transformer, Encoder<()>);
     type Payload = Bytes;
@@ -95,17 +97,17 @@ impl RequestBuilder<Event> for SqsRequestBuilder {
 
         let builder = RequestMetadataBuilder::from_event(&event);
 
-        let sqs_metadata = SqsMetadata {
+        let metadata = SSMetadata {
             finalizers: event.take_finalizers(),
             message_group_id,
             message_deduplication_id,
         };
-        (sqs_metadata, builder, event)
+        (metadata, builder, event)
     }
 
     fn build_request(
         &self,
-        sqs_metadata: Self::Metadata,
+        client_metadata: Self::Metadata,
         metadata: RequestMetadata,
         payload: EncodeResult<Self::Payload>,
     ) -> Self::Request {
@@ -114,23 +116,21 @@ impl RequestBuilder<Event> for SqsRequestBuilder {
 
         SendMessageEntry {
             message_body,
-            message_group_id: sqs_metadata.message_group_id,
-            message_deduplication_id: sqs_metadata.message_deduplication_id,
-            queue_url: self.queue_url.clone(),
-            finalizers: sqs_metadata.finalizers,
+            message_group_id: client_metadata.message_group_id,
+            message_deduplication_id: client_metadata.message_deduplication_id,
+            finalizers: client_metadata.finalizers,
             metadata,
         }
     }
 }
 
 #[derive(Debug, Clone)]
-pub(crate) struct SendMessageEntry {
-    pub message_body: String,
-    pub message_group_id: Option<String>,
-    pub message_deduplication_id: Option<String>,
-    pub queue_url: String,
-    finalizers: EventFinalizers,
-    pub metadata: RequestMetadata,
+pub(super) struct SendMessageEntry {
+    pub(super) message_body: String,
+    pub(super) message_group_id: Option<String>,
+    pub(super) message_deduplication_id: Option<String>,
+    pub(super) finalizers: EventFinalizers,
+    pub(super) metadata: RequestMetadata,
 }
 
 impl ByteSizeOf for SendMessageEntry {
diff --git a/src/sinks/aws_s_s/retry.rs b/src/sinks/aws_s_s/retry.rs
new file mode 100644
index 0000000000000..249c1ad24c35d
--- /dev/null
+++ b/src/sinks/aws_s_s/retry.rs
@@ -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,
+        }
+    }
+}
diff --git a/src/sinks/aws_s_s/service.rs b/src/sinks/aws_s_s/service.rs
new file mode 100644
index 0000000000000..8d5fd53c17dbd
--- /dev/null
+++ b/src/sinks/aws_s_s/service.rs
@@ -0,0 +1,87 @@
+use std::marker::PhantomData;
+use std::task::{Context, Poll};
+
+use aws_sdk_sqs::types::SdkError;
+use futures::future::BoxFuture;
+use tower::Service;
+use vector_common::request_metadata::GroupedCountByteSize;
+use vector_core::{event::EventStatus, stream::DriverResponse, ByteSizeOf};
+
+use super::{client::Client, request_builder::SendMessageEntry};
+
+pub(super) struct SSService<C, E>
+where
+    C: Client<E> + Clone + Send + Sync + 'static,
+    E: std::fmt::Debug + std::fmt::Display + std::error::Error + Sync + Send + 'static,
+{
+    client: C,
+    _phantom: PhantomData<fn() -> E>,
+}
+
+impl<C, E> SSService<C, E>
+where
+    C: Client<E> + Clone + Send + Sync + 'static,
+    E: std::fmt::Debug + std::fmt::Display + std::error::Error + Sync + Send + 'static,
+{
+    pub(super) const fn new(client: C) -> Self {
+        Self {
+            client,
+            _phantom: PhantomData,
+        }
+    }
+}
+
+impl<C, E> Clone for SSService<C, E>
+where
+    C: Client<E> + Clone + Send + Sync + 'static,
+    E: std::fmt::Debug + std::fmt::Display + std::error::Error + Sync + Send + 'static,
+{
+    fn clone(&self) -> SSService<C, E> {
+        SSService {
+            client: self.client.clone(),
+            _phantom: PhantomData,
+        }
+    }
+}
+
+impl<C, E> Service<SendMessageEntry> for SSService<C, E>
+where
+    C: Client<E> + Clone + Send + Sync + 'static,
+    E: std::fmt::Debug + std::fmt::Display + std::error::Error + Sync + Send + 'static,
+{
+    type Response = SendMessageResponse;
+    type Error = SdkError<E>;
+    type Future = BoxFuture<'static, Result<Self::Response, Self::Error>>;
+
+    // Emission of an internal event in case of errors is handled upstream by the caller.
+    fn poll_ready(&mut self, _cx: &mut Context) -> Poll<Result<(), Self::Error>> {
+        Poll::Ready(Ok(()))
+    }
+
+    // Emission of internal events for errors and dropped events is handled upstream by the caller.
+    fn call(&mut self, entry: SendMessageEntry) -> Self::Future {
+        let byte_size = entry.size_of();
+        let client = self.client.clone();
+
+        Box::pin(async move { client.send_message(entry, byte_size).await })
+    }
+}
+
+pub(super) struct SendMessageResponse {
+    pub(crate) byte_size: usize,
+    pub(crate) json_byte_size: GroupedCountByteSize,
+}
+
+impl DriverResponse for SendMessageResponse {
+    fn event_status(&self) -> EventStatus {
+        EventStatus::Delivered
+    }
+
+    fn events_sent(&self) -> &GroupedCountByteSize {
+        &self.json_byte_size
+    }
+
+    fn bytes_sent(&self) -> Option<usize> {
+        Some(self.byte_size)
+    }
+}
diff --git a/src/sinks/aws_sqs/sink.rs b/src/sinks/aws_s_s/sink.rs
similarity index 59%
rename from src/sinks/aws_sqs/sink.rs
rename to src/sinks/aws_s_s/sink.rs
index 6d87b34a47c6d..b8bfa16da936c 100644
--- a/src/sinks/aws_sqs/sink.rs
+++ b/src/sinks/aws_s_s/sink.rs
@@ -1,12 +1,12 @@
 use std::num::NonZeroUsize;
 
-use aws_sdk_sqs::Client as SqsClient;
 use futures::stream::BoxStream;
 use futures_util::StreamExt;
 use vector_core::sink::StreamSink;
 
-use super::{config::SqsSinkConfig, request_builder::SqsRequestBuilder, service::SqsService};
+use super::{client::Client, request_builder::SSRequestBuilder, service::SSService};
 use crate::internal_events::SinkRequestBuildError;
+use crate::sinks::aws_s_s::retry::SSRetryLogic;
 use crate::{
     event::Event,
     sinks::util::{
@@ -24,18 +24,29 @@ impl SinkBatchSettings for SqsSinkDefaultBatchSettings {
 }
 
 #[derive(Clone)]
-pub(crate) struct SqsSink {
-    request_builder: SqsRequestBuilder,
-    service: SqsService,
+pub(super) struct SSSink<C, E>
+where
+    C: Client<E> + Clone + Send + Sync + 'static,
+    E: std::fmt::Debug + std::fmt::Display + std::error::Error + Sync + Send + 'static,
+{
+    request_builder: SSRequestBuilder,
+    service: SSService<C, E>,
     request: TowerRequestConfig,
 }
 
-impl SqsSink {
-    pub fn new(config: SqsSinkConfig, client: SqsClient) -> crate::Result<Self> {
-        let request = config.request;
-        Ok(SqsSink {
-            request_builder: SqsRequestBuilder::new(config)?,
-            service: SqsService::new(client),
+impl<C, E> SSSink<C, E>
+where
+    C: Client<E> + Clone + Send + Sync + 'static,
+    E: std::fmt::Debug + std::fmt::Display + std::error::Error + Sync + Send + 'static,
+{
+    pub(super) fn new(
+        request_builder: SSRequestBuilder,
+        request: TowerRequestConfig,
+        publisher: C,
+    ) -> crate::Result<Self> {
+        Ok(SSSink {
+            request_builder,
+            service: SSService::new(publisher),
             request,
         })
     }
@@ -45,8 +56,9 @@ impl SqsSink {
             .request
             .unwrap_with(&TowerRequestConfig::default().timeout_secs(30));
         let request_builder_concurrency_limit = NonZeroUsize::new(50);
+        let retry_logic: SSRetryLogic<E> = super::retry::SSRetryLogic::new();
         let service = tower::ServiceBuilder::new()
-            .settings(request, super::retry::SqsRetryLogic)
+            .settings(request, retry_logic)
             .service(self.service);
 
         input
@@ -64,7 +76,11 @@ impl SqsSink {
 }
 
 #[async_trait::async_trait]
-impl StreamSink<Event> for SqsSink {
+impl<C, E> StreamSink<Event> for SSSink<C, E>
+where
+    C: Client<E> + Clone + Send + Sync + 'static,
+    E: std::fmt::Debug + std::fmt::Display + std::error::Error + Sync + Send + 'static,
+{
     async fn run(mut self: Box<Self>, input: BoxStream<'_, Event>) -> Result<(), ()> {
         self.run_inner(input).await
     }
diff --git a/src/sinks/aws_s_s/sns/client.rs b/src/sinks/aws_s_s/sns/client.rs
new file mode 100644
index 0000000000000..12a6685b222f7
--- /dev/null
+++ b/src/sinks/aws_s_s/sns/client.rs
@@ -0,0 +1,41 @@
+use super::{Client, SendMessageEntry, SendMessageResponse};
+use aws_sdk_sns::{error::PublishError, types::SdkError};
+use futures::TryFutureExt;
+use tracing::Instrument;
+
+#[derive(Clone, Debug)]
+pub(super) struct SnsMessagePublisher {
+    client: aws_sdk_sns::Client,
+    topic_arn: String,
+}
+
+impl SnsMessagePublisher {
+    pub(super) const fn new(client: aws_sdk_sns::Client, topic_arn: String) -> Self {
+        Self { client, topic_arn }
+    }
+}
+#[async_trait::async_trait]
+impl Client<PublishError> for SnsMessagePublisher {
+    async fn send_message(
+        &self,
+        entry: SendMessageEntry,
+        byte_size: usize,
+    ) -> Result<SendMessageResponse, SdkError<PublishError>> {
+        self.client
+            .publish()
+            .message(entry.message_body)
+            .set_message_group_id(entry.message_group_id)
+            .set_message_deduplication_id(entry.message_deduplication_id)
+            .topic_arn(self.topic_arn.clone())
+            .send()
+            .map_ok(|_| SendMessageResponse {
+                byte_size,
+                json_byte_size: entry
+                    .metadata
+                    .events_estimated_json_encoded_byte_size()
+                    .clone(),
+            })
+            .instrument(info_span!("request").or_current())
+            .await
+    }
+}
diff --git a/src/sinks/aws_s_s/sns/config.rs b/src/sinks/aws_s_s/sns/config.rs
new file mode 100644
index 0000000000000..a9ea5f283f579
--- /dev/null
+++ b/src/sinks/aws_s_s/sns/config.rs
@@ -0,0 +1,130 @@
+use aws_sdk_sns::Client as SnsClient;
+
+use crate::aws::RegionOrEndpoint;
+
+use crate::config::{
+    AcknowledgementsConfig, DataType, GenerateConfig, Input, ProxyConfig, SinkConfig, SinkContext,
+};
+use vector_config::configurable_component;
+
+use super::{
+    client::SnsMessagePublisher, message_deduplication_id, message_group_id, BaseSSSinkConfig,
+    SSRequestBuilder, SSSink,
+};
+use crate::aws::create_client;
+use crate::aws::ClientBuilder;
+
+/// Configuration for the `aws_sns` sink.
+#[configurable_component(sink(
+    "aws_sns",
+    "Publish observability events to AWS Simple Notification Service topics."
+))]
+#[derive(Clone, Debug)]
+#[serde(deny_unknown_fields)]
+pub(super) struct SnsSinkConfig {
+    /// The ARN of the Amazon SNS topic to which messages are sent.
+    #[configurable(validation(format = "uri"))]
+    #[configurable(metadata(docs::examples = "arn:aws:sns:us-east-2:123456789012:MyTopic"))]
+    pub(super) topic_arn: String,
+
+    #[serde(flatten)]
+    pub(super) region: RegionOrEndpoint,
+
+    #[serde(flatten)]
+    pub(super) base_config: BaseSSSinkConfig,
+}
+
+impl GenerateConfig for SnsSinkConfig {
+    fn generate_config() -> toml::Value {
+        toml::from_str(
+            r#"topic_arn = "arn:aws:sns:us-east-2:123456789012:MyTopic"
+            region = "us-east-2"
+            encoding.codec = "json""#,
+        )
+        .unwrap()
+    }
+}
+
+impl SnsSinkConfig {
+    pub(super) async fn create_client(&self, proxy: &ProxyConfig) -> crate::Result<SnsClient> {
+        create_client::<SnsClientBuilder>(
+            &self.base_config.auth,
+            self.region.region(),
+            self.region.endpoint(),
+            proxy,
+            &self.base_config.tls,
+            true,
+        )
+        .await
+    }
+}
+
+#[async_trait::async_trait]
+#[typetag::serde(name = "aws_sns")]
+impl SinkConfig for SnsSinkConfig {
+    async fn build(
+        &self,
+        cx: SinkContext,
+    ) -> crate::Result<(crate::sinks::VectorSink, crate::sinks::Healthcheck)> {
+        let client = self.create_client(&cx.proxy).await?;
+
+        let publisher = SnsMessagePublisher::new(client.clone(), self.topic_arn.clone());
+
+        let healthcheck = Box::pin(healthcheck(client.clone(), self.topic_arn.clone()));
+
+        let message_group_id = message_group_id(
+            self.base_config.message_group_id.clone(),
+            self.topic_arn.ends_with(".fifo"),
+        );
+        let message_deduplication_id =
+            message_deduplication_id(self.base_config.message_deduplication_id.clone());
+
+        let sink = SSSink::new(
+            SSRequestBuilder::new(
+                message_group_id?,
+                message_deduplication_id?,
+                self.base_config.encoding.clone(),
+            )?,
+            self.base_config.request,
+            publisher,
+        )?;
+        Ok((
+            crate::sinks::VectorSink::from_event_streamsink(sink),
+            healthcheck,
+        ))
+    }
+
+    fn input(&self) -> Input {
+        Input::new(self.base_config.encoding.config().input_type() & DataType::Log)
+    }
+
+    fn acknowledgements(&self) -> &AcknowledgementsConfig {
+        &self.base_config.acknowledgements
+    }
+}
+
+pub(super) struct SnsClientBuilder;
+
+impl ClientBuilder for SnsClientBuilder {
+    type Config = aws_sdk_sns::config::Config;
+    type Client = aws_sdk_sns::client::Client;
+    type DefaultMiddleware = aws_sdk_sns::middleware::DefaultMiddleware;
+
+    fn default_middleware() -> Self::DefaultMiddleware {
+        aws_sdk_sns::middleware::DefaultMiddleware::new()
+    }
+
+    fn build(client: aws_smithy_client::Client, config: &aws_types::SdkConfig) -> Self::Client {
+        aws_sdk_sns::client::Client::with_config(client, config.into())
+    }
+}
+
+pub(super) async fn healthcheck(client: SnsClient, topic_arn: String) -> crate::Result<()> {
+    client
+        .get_topic_attributes()
+        .topic_arn(topic_arn.clone())
+        .send()
+        .await
+        .map(|_| ())
+        .map_err(Into::into)
+}
diff --git a/src/sinks/aws_s_s/sns/integration_tests.rs b/src/sinks/aws_s_s/sns/integration_tests.rs
new file mode 100644
index 0000000000000..cf79d4d97c5f8
--- /dev/null
+++ b/src/sinks/aws_s_s/sns/integration_tests.rs
@@ -0,0 +1,195 @@
+use std::collections::HashMap;
+
+use aws_sdk_sns::{Client as SnsClient, Region};
+use aws_sdk_sqs::model::QueueAttributeName;
+use aws_sdk_sqs::Client as SqsClient;
+use codecs::TextSerializerConfig;
+use tokio::time::{sleep, Duration};
+
+use super::{
+    config::SnsClientBuilder,
+    config::{healthcheck, SnsSinkConfig},
+    BaseSSSinkConfig,
+};
+use crate::common::sqs::SqsClientBuilder;
+use crate::config::{SinkConfig, SinkContext};
+use crate::{
+    aws::{create_client, AwsAuthentication, RegionOrEndpoint},
+    config::ProxyConfig,
+    test_util::{
+        components::{run_and_assert_sink_compliance, AWS_SINK_TAGS},
+        random_lines_with_stream, random_string,
+    },
+};
+
+fn sns_address() -> String {
+    std::env::var("SNS_ADDRESS").unwrap_or_else(|_| "http://localhost:4566".into())
+}
+
+async fn create_sns_test_client() -> SnsClient {
+    let auth = AwsAuthentication::test_auth();
+
+    let endpoint = sns_address();
+    let proxy = ProxyConfig::default();
+    create_client::<SnsClientBuilder>(
+        &auth,
+        Some(Region::new("localstack")),
+        Some(endpoint),
+        &proxy,
+        &None,
+        true,
+    )
+    .await
+    .unwrap()
+}
+
+fn sqs_address() -> String {
+    std::env::var("SQS_ADDRESS").unwrap_or_else(|_| "http://localhost:4566".into())
+}
+
+async fn create_sqs_test_client() -> SqsClient {
+    let auth = AwsAuthentication::test_auth();
+
+    let endpoint = sqs_address();
+    let proxy = ProxyConfig::default();
+    create_client::<SqsClientBuilder>(
+        &auth,
+        Some(Region::new("localstack")),
+        Some(endpoint),
+        &proxy,
+        &None,
+        true,
+    )
+    .await
+    .unwrap()
+}
+
+#[tokio::test]
+async fn sns_send_message_batch() {
+    let topic_name = gen_topic_name();
+    let topic_arn = ensure_topic(topic_name.clone()).await;
+
+    let sqs_client = create_sqs_test_client().await;
+    let queue_url = ensure_queue(&sqs_client, gen_queue_name()).await;
+    let queue_arn = get_queue_arn(&sqs_client, queue_url.clone()).await;
+
+    let sns_client = create_sns_test_client().await;
+
+    let base_config = BaseSSSinkConfig {
+        encoding: TextSerializerConfig::default().into(),
+        message_group_id: None,
+        message_deduplication_id: None,
+        request: Default::default(),
+        tls: Default::default(),
+        assume_role: None,
+        auth: Default::default(),
+        acknowledgements: Default::default(),
+    };
+
+    let config = SnsSinkConfig {
+        region: RegionOrEndpoint::with_both("local", sns_address().as_str()),
+        topic_arn: topic_arn.clone(),
+        base_config,
+    };
+
+    healthcheck(sns_client.clone(), config.topic_arn.clone())
+        .await
+        .unwrap();
+
+    subscribe_queue_to_topic(&sns_client, &topic_arn, &queue_arn).await;
+
+    let cx = SinkContext::default();
+    let sink = config.build(cx).await.unwrap().0;
+
+    let (mut input_lines, events) = random_lines_with_stream(100, 10, None);
+    run_and_assert_sink_compliance(sink, events, &AWS_SINK_TAGS).await;
+
+    sleep(Duration::from_secs(1)).await;
+
+    let response = sqs_client
+        .receive_message()
+        .max_number_of_messages(input_lines.len() as i32)
+        .queue_url(queue_url)
+        .send()
+        .await
+        .unwrap();
+
+    let mut output_lines = response
+        .clone()
+        .messages
+        .unwrap()
+        .into_iter()
+        .map(|e| e.body.unwrap())
+        .collect::<Vec<_>>();
+
+    input_lines.sort();
+    output_lines.sort();
+
+    assert_eq!(output_lines, input_lines);
+    assert_eq!(input_lines.len(), response.messages.unwrap().len());
+}
+
+async fn ensure_topic(topic_arn: String) -> String {
+    let client = create_sns_test_client().await;
+
+    let attributes: Option<HashMap<String, String>> = None;
+
+    client
+        .create_topic()
+        .set_attributes(attributes)
+        .name(topic_arn)
+        .send()
+        .await
+        .unwrap()
+        .topic_arn
+        .unwrap()
+}
+
+fn gen_topic_name() -> String {
+    format!("test-{}", random_string(10).to_lowercase())
+}
+
+async fn ensure_queue(client: &SqsClient, queue_name: String) -> String {
+    client
+        .create_queue()
+        .queue_name(queue_name)
+        .send()
+        .await
+        .unwrap()
+        .queue_url
+        .unwrap()
+}
+
+async fn get_queue_arn(client: &SqsClient, queue_url: String) -> String {
+    let arn_attribute = QueueAttributeName::QueueArn;
+    client
+        .get_queue_attributes()
+        .queue_url(queue_url)
+        .send()
+        .await
+        .unwrap()
+        .attributes
+        .unwrap()
+        .get(&arn_attribute)
+        .unwrap()
+        .clone()
+}
+
+fn gen_queue_name() -> String {
+    format!("test-{}", random_string(10).to_lowercase())
+}
+
+async fn subscribe_queue_to_topic(sns_client: &SnsClient, topic_arn: &str, queue_arn: &str) {
+    let mut attributes = HashMap::new();
+    attributes.insert("RawMessageDelivery".to_string(), "true".to_string());
+
+    sns_client
+        .subscribe()
+        .protocol("sqs")
+        .endpoint(queue_arn)
+        .set_attributes(Some(attributes))
+        .topic_arn(topic_arn)
+        .send()
+        .await
+        .unwrap();
+}
diff --git a/src/sinks/aws_s_s/sns/mod.rs b/src/sinks/aws_s_s/sns/mod.rs
new file mode 100644
index 0000000000000..3d83d21d71216
--- /dev/null
+++ b/src/sinks/aws_s_s/sns/mod.rs
@@ -0,0 +1,12 @@
+mod client;
+mod config;
+#[cfg(all(test, feature = "aws-sns-integration-tests"))]
+mod integration_tests;
+
+use super::{
+    client::Client,
+    config::{message_deduplication_id, message_group_id, BaseSSSinkConfig},
+    request_builder::{SSRequestBuilder, SendMessageEntry},
+    service::SendMessageResponse,
+    sink::SSSink,
+};
diff --git a/src/sinks/aws_s_s/sqs/client.rs b/src/sinks/aws_s_s/sqs/client.rs
new file mode 100644
index 0000000000000..334b902c2e47c
--- /dev/null
+++ b/src/sinks/aws_s_s/sqs/client.rs
@@ -0,0 +1,41 @@
+use super::{Client, SendMessageEntry, SendMessageResponse};
+use aws_sdk_sqs::{error::SendMessageError, types::SdkError};
+use futures::TryFutureExt;
+use tracing::Instrument;
+
+#[derive(Clone, Debug)]
+pub(super) struct SqsMessagePublisher {
+    client: aws_sdk_sqs::Client,
+    queue_url: String,
+}
+
+impl SqsMessagePublisher {
+    pub(super) const fn new(client: aws_sdk_sqs::Client, queue_url: String) -> Self {
+        Self { client, queue_url }
+    }
+}
+#[async_trait::async_trait]
+impl Client<SendMessageError> for SqsMessagePublisher {
+    async fn send_message(
+        &self,
+        entry: SendMessageEntry,
+        byte_size: usize,
+    ) -> Result<SendMessageResponse, SdkError<SendMessageError>> {
+        self.client
+            .send_message()
+            .message_body(entry.message_body)
+            .set_message_group_id(entry.message_group_id)
+            .set_message_deduplication_id(entry.message_deduplication_id)
+            .queue_url(self.queue_url.clone())
+            .send()
+            .map_ok(|_| SendMessageResponse {
+                byte_size,
+                json_byte_size: entry
+                    .metadata
+                    .events_estimated_json_encoded_byte_size()
+                    .clone(),
+            })
+            .instrument(info_span!("request").or_current())
+            .await
+    }
+}
diff --git a/src/sinks/aws_s_s/sqs/config.rs b/src/sinks/aws_s_s/sqs/config.rs
new file mode 100644
index 0000000000000..803e6e253e15d
--- /dev/null
+++ b/src/sinks/aws_s_s/sqs/config.rs
@@ -0,0 +1,114 @@
+use aws_sdk_sqs::Client as SqsClient;
+
+use crate::aws::RegionOrEndpoint;
+
+use crate::config::{
+    AcknowledgementsConfig, DataType, GenerateConfig, Input, ProxyConfig, SinkConfig, SinkContext,
+};
+use vector_config::configurable_component;
+
+use super::{
+    client::SqsMessagePublisher, message_deduplication_id, message_group_id, BaseSSSinkConfig,
+    SSRequestBuilder, SSSink,
+};
+use crate::{aws::create_client, common::sqs::SqsClientBuilder};
+
+/// Configuration for the `aws_sqs` sink.
+#[configurable_component(sink(
+    "aws_sqs",
+    "Publish observability events to AWS Simple Queue Service topics."
+))]
+#[derive(Clone, Debug)]
+#[serde(deny_unknown_fields)]
+pub(super) struct SqsSinkConfig {
+    /// The URL of the Amazon SQS queue to which messages are sent.
+    #[configurable(validation(format = "uri"))]
+    #[configurable(metadata(
+        docs::examples = "https://sqs.us-east-2.amazonaws.com/123456789012/MyQueue"
+    ))]
+    pub(super) queue_url: String,
+
+    #[serde(flatten)]
+    pub(super) region: RegionOrEndpoint,
+
+    #[serde(flatten)]
+    pub(super) base_config: BaseSSSinkConfig,
+}
+
+impl GenerateConfig for SqsSinkConfig {
+    fn generate_config() -> toml::Value {
+        toml::from_str(
+            r#"queue_url = "https://sqs.us-east-2.amazonaws.com/123456789012/MyQueue"
+            region = "us-east-2"
+            encoding.codec = "json""#,
+        )
+        .unwrap()
+    }
+}
+
+impl SqsSinkConfig {
+    pub(super) async fn create_client(&self, proxy: &ProxyConfig) -> crate::Result<SqsClient> {
+        create_client::<SqsClientBuilder>(
+            &self.base_config.auth,
+            self.region.region(),
+            self.region.endpoint(),
+            proxy,
+            &self.base_config.tls,
+            true,
+        )
+        .await
+    }
+}
+
+#[async_trait::async_trait]
+#[typetag::serde(name = "aws_sqs")]
+impl SinkConfig for SqsSinkConfig {
+    async fn build(
+        &self,
+        cx: SinkContext,
+    ) -> crate::Result<(crate::sinks::VectorSink, crate::sinks::Healthcheck)> {
+        let client = self.create_client(&cx.proxy).await?;
+
+        let publisher = SqsMessagePublisher::new(client.clone(), self.queue_url.clone());
+
+        let healthcheck = Box::pin(healthcheck(client.clone(), self.queue_url.clone()));
+        let message_group_id = message_group_id(
+            self.base_config.message_group_id.clone(),
+            self.queue_url.ends_with(".fifo"),
+        );
+        let message_deduplication_id =
+            message_deduplication_id(self.base_config.message_deduplication_id.clone());
+
+        let sink = SSSink::new(
+            SSRequestBuilder::new(
+                message_group_id?,
+                message_deduplication_id?,
+                self.base_config.encoding.clone(),
+            )?,
+            self.base_config.request,
+            publisher,
+        )?;
+        Ok((
+            crate::sinks::VectorSink::from_event_streamsink(sink),
+            healthcheck,
+        ))
+    }
+
+    fn input(&self) -> Input {
+        Input::new(self.base_config.encoding.config().input_type() & DataType::Log)
+    }
+
+    fn acknowledgements(&self) -> &AcknowledgementsConfig {
+        &self.base_config.acknowledgements
+    }
+}
+
+pub(super) async fn healthcheck(client: SqsClient, queue_url: String) -> crate::Result<()> {
+    client
+        .get_queue_attributes()
+        .queue_url(queue_url)
+        .send()
+        .await
+        .map(|_| ())
+        .map_err(Into::into)
+}
diff --git a/src/sinks/aws_sqs/integration_tests.rs b/src/sinks/aws_s_s/sqs/integration_tests.rs
similarity index 88%
rename from src/sinks/aws_sqs/integration_tests.rs
rename to src/sinks/aws_s_s/sqs/integration_tests.rs
index e8d37f1266c79..a3c7171857391 100644
--- a/src/sinks/aws_sqs/integration_tests.rs
+++ b/src/sinks/aws_s_s/sqs/integration_tests.rs
@@ -1,17 +1,16 @@
-#![cfg(all(test, feature = "aws-sqs-integration-tests"))]
-
 use std::collections::HashMap;
 
 use aws_sdk_sqs::{model::QueueAttributeName, Client as SqsClient, Region};
 use codecs::TextSerializerConfig;
 use tokio::time::{sleep, Duration};
 
-use super::{config::SqsSinkConfig, sink::SqsSink};
+use crate::config::{SinkConfig, SinkContext};
+use crate::sinks::aws_s_s::sqs::config::{healthcheck, SqsSinkConfig};
+use crate::sinks::aws_s_s::sqs::BaseSSSinkConfig;
 use crate::{
     aws::{create_client, AwsAuthentication, RegionOrEndpoint},
     common::sqs::SqsClientBuilder,
     config::ProxyConfig,
-    sinks::VectorSink,
     test_util::{
         components::{run_and_assert_sink_compliance, AWS_SINK_TAGS},
         random_lines_with_stream, random_string,
@@ -47,9 +46,7 @@ async fn sqs_send_message_batch() {
 
     let client = create_test_client().await;
 
-    let config = SqsSinkConfig {
-        queue_url: queue_url.clone(),
-        region: RegionOrEndpoint::with_both("local", sqs_address().as_str()),
+    let base_config = BaseSSSinkConfig {
         encoding: TextSerializerConfig::default().into(),
         message_group_id: None,
         message_deduplication_id: None,
@@ -60,10 +57,19 @@ async fn sqs_send_message_batch() {
         acknowledgements: Default::default(),
     };
 
-    config.clone().healthcheck(client.clone()).await.unwrap();
+    let config = SqsSinkConfig {
+        region: RegionOrEndpoint::with_both("local", sqs_address().as_str()),
+        queue_url: queue_url.clone(),
+        base_config,
+    };
+
+    healthcheck(client.clone(), config.queue_url.clone())
+        .await
+        .unwrap();
+
+    let cx = SinkContext::default();
 
-    let sink = SqsSink::new(config, client.clone()).unwrap();
-    let sink = VectorSink::from_event_streamsink(sink);
+    let sink = config.build(cx).await.unwrap().0;
 
     let (mut input_lines, events) = random_lines_with_stream(100, 10, None);
     run_and_assert_sink_compliance(sink, events, &AWS_SINK_TAGS).await;
diff --git a/src/sinks/aws_s_s/sqs/mod.rs b/src/sinks/aws_s_s/sqs/mod.rs
new file mode 100644
index 0000000000000..943fc8bacb8ab
--- /dev/null
+++ b/src/sinks/aws_s_s/sqs/mod.rs
@@ -0,0 +1,13 @@
+mod client;
+mod config;
+
+#[cfg(all(test, feature = "aws-sqs-integration-tests"))]
+mod integration_tests;
+
+use super::{
+    client::Client,
+    config::{message_deduplication_id, message_group_id, BaseSSSinkConfig},
+    request_builder::{SSRequestBuilder, SendMessageEntry},
+    service::SendMessageResponse,
+    sink::SSSink,
+};
diff --git a/src/sinks/aws_sqs/config.rs b/src/sinks/aws_sqs/config.rs
deleted file mode 100644
index 984f40aa7bf40..0000000000000
--- a/src/sinks/aws_sqs/config.rs
+++ /dev/null
@@ -1,183 +0,0 @@
-use std::convert::TryFrom;
-
-use aws_sdk_sqs::Client as SqsClient;
-use futures::FutureExt;
-use serde::{Deserialize, Serialize};
-use snafu::{ResultExt, Snafu};
-use vector_config::configurable_component;
-
-use crate::{
-    aws::{create_client, AwsAuthentication, RegionOrEndpoint},
-    codecs::EncodingConfig,
-    common::sqs::SqsClientBuilder,
-    config::{
-        AcknowledgementsConfig, DataType, GenerateConfig, Input, ProxyConfig, SinkConfig,
-        SinkContext,
-    },
-    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 },
-}
-
-/// Configuration for the `aws_sqs` sink.
-#[configurable_component(sink(
-    "aws_sqs",
-    "Publish observability events to AWS Simple Queue Service topics."
-))]
-#[derive(Clone, Debug)]
-#[serde(deny_unknown_fields)]
-pub struct SqsSinkConfig {
-    /// The URL of the Amazon SQS queue to which messages are sent.
-    #[configurable(validation(format = "uri"))]
-    #[configurable(metadata(
-        docs::examples = "https://sqs.us-east-2.amazonaws.com/123456789012/MyQueue"
-    ))]
-    pub queue_url: String,
-
-    #[serde(flatten)]
-    pub region: RegionOrEndpoint,
-
-    #[configurable(derived)]
-    pub 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 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 message_deduplication_id: Option<String>,
-
-    #[configurable(derived)]
-    #[serde(default)]
-    pub request: TowerRequestConfig,
-
-    #[configurable(derived)]
-    pub 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 assume_role: Option<String>,
-
-    #[configurable(derived)]
-    #[serde(default)]
-    pub 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,
-}
-
-#[derive(Deserialize, Serialize, Debug, Eq, PartialEq, Clone, Derivative)]
-#[serde(rename_all = "snake_case")]
-pub enum Encoding {
-    Text,
-    Json,
-}
-
-impl GenerateConfig for SqsSinkConfig {
-    fn generate_config() -> toml::Value {
-        toml::from_str(
-            r#"queue_url = "https://sqs.us-east-2.amazonaws.com/123456789012/MyQueue"
-            region = "us-east-2"
-            encoding.codec = "json""#,
-        )
-        .unwrap()
-    }
-}
-
-#[async_trait::async_trait]
-#[typetag::serde(name = "aws_sqs")]
-impl SinkConfig for SqsSinkConfig {
-    async fn build(
-        &self,
-        cx: SinkContext,
-    ) -> crate::Result<(crate::sinks::VectorSink, crate::sinks::Healthcheck)> {
-        let client = self.create_client(&cx.proxy).await?;
-        let healthcheck = self.clone().healthcheck(client.clone()).boxed();
-        let sink = super::sink::SqsSink::new(self.clone(), client)?;
-        Ok((
-            crate::sinks::VectorSink::from_event_streamsink(sink),
-            healthcheck,
-        ))
-    }
-
-    fn input(&self) -> Input {
-        Input::new(self.encoding.config().input_type() & DataType::Log)
-    }
-
-    fn acknowledgements(&self) -> &AcknowledgementsConfig {
-        &self.acknowledgements
-    }
-}
-
-impl SqsSinkConfig {
-    pub async fn healthcheck(self, client: SqsClient) -> crate::Result<()> {
-        client
-            .get_queue_attributes()
-            .queue_url(self.queue_url.clone())
-            .send()
-            .await
-            .map(|_| ())
-            .map_err(Into::into)
-    }
-
-    pub async fn create_client(&self, proxy: &ProxyConfig) -> crate::Result<SqsClient> {
-        create_client::<SqsClientBuilder>(
-            &self.auth,
-            self.region.region(),
-            self.region.endpoint(),
-            proxy,
-            &self.tls,
-            true,
-        )
-        .await
-    }
-
-    pub fn message_group_id(&self) -> crate::Result<Option<Template>> {
-        let fifo = self.queue_url.ends_with(".fifo");
-        match (self.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 fn message_deduplication_id(&self) -> crate::Result<Option<Template>> {
-        Ok(self
-            .message_deduplication_id
-            .clone()
-            .map(Template::try_from)
-            .transpose()?)
-    }
-}
diff --git a/src/sinks/aws_sqs/mod.rs b/src/sinks/aws_sqs/mod.rs
deleted file mode 100644
index 8f2919860a239..0000000000000
--- a/src/sinks/aws_sqs/mod.rs
+++ /dev/null
@@ -1,9 +0,0 @@
-mod config;
-mod request_builder;
-mod retry;
-mod service;
-mod sink;
-
-mod integration_tests;
-
-pub use self::config::SqsSinkConfig;
diff --git a/src/sinks/aws_sqs/retry.rs b/src/sinks/aws_sqs/retry.rs
deleted file mode 100644
index 1a34ab7915353..0000000000000
--- a/src/sinks/aws_sqs/retry.rs
+++ /dev/null
@@ -1,16 +0,0 @@
-use aws_sdk_sqs::{error::SendMessageError, types::SdkError};
-
-use super::service::SendMessageResponse;
-use crate::{aws::is_retriable_error, sinks::util::retries::RetryLogic};
-
-#[derive(Debug, Clone)]
-pub(super) struct SqsRetryLogic;
-
-impl RetryLogic for SqsRetryLogic {
-    type Error = SdkError<SendMessageError>;
-    type Response = SendMessageResponse;
-
-    fn is_retriable_error(&self, error: &Self::Error) -> bool {
-        is_retriable_error(error)
-    }
-}
diff --git a/src/sinks/aws_sqs/service.rs b/src/sinks/aws_sqs/service.rs
deleted file mode 100644
index e08f68c2753ba..0000000000000
--- a/src/sinks/aws_sqs/service.rs
+++ /dev/null
@@ -1,76 +0,0 @@
-use std::task::{Context, Poll};
-
-use aws_sdk_sqs::{error::SendMessageError, types::SdkError, Client as SqsClient};
-use futures::{future::BoxFuture, TryFutureExt};
-use tower::Service;
-use tracing::Instrument;
-use vector_common::request_metadata::GroupedCountByteSize;
-use vector_core::{event::EventStatus, stream::DriverResponse, ByteSizeOf};
-
-use super::request_builder::SendMessageEntry;
-
-#[derive(Clone)]
-pub(crate) struct SqsService {
-    client: SqsClient,
-}
-
-impl SqsService {
-    pub const fn new(client: SqsClient) -> Self {
-        Self { client }
-    }
-}
-
-impl Service<SendMessageEntry> for SqsService {
-    type Response = SendMessageResponse;
-    type Error = SdkError<SendMessageError>;
-    type Future = BoxFuture<'static, Result<Self::Response, Self::Error>>;
-
-    // Emission of an internal event in case of errors is handled upstream by the caller.
-    fn poll_ready(&mut self, _cx: &mut Context) -> Poll<Result<(), Self::Error>> {
-        Poll::Ready(Ok(()))
-    }
-
-    // Emission of internal events for errors and dropped events is handled upstream by the caller.
-    fn call(&mut self, entry: SendMessageEntry) -> Self::Future {
-        let byte_size = entry.size_of();
-        let client = self.client.clone();
-
-        Box::pin(async move {
-            client
-                .send_message()
-                .message_body(entry.message_body)
-                .set_message_group_id(entry.message_group_id)
-                .set_message_deduplication_id(entry.message_deduplication_id)
-                .queue_url(entry.queue_url)
-                .send()
-                .map_ok(|_| SendMessageResponse {
-                    byte_size,
-                    json_byte_size: entry
-                        .metadata
-                        .events_estimated_json_encoded_byte_size()
-                        .clone(),
-                })
-                .instrument(info_span!("request").or_current())
-                .await
-        })
-    }
-}
-
-pub(crate) struct SendMessageResponse {
-    byte_size: usize,
-    json_byte_size: GroupedCountByteSize,
-}
-
-impl DriverResponse for SendMessageResponse {
-    fn event_status(&self) -> EventStatus {
-        EventStatus::Delivered
-    }
-
-    fn events_sent(&self) -> &GroupedCountByteSize {
-        &self.json_byte_size
-    }
-
-    fn bytes_sent(&self) -> Option<usize> {
-        Some(self.byte_size)
-    }
-}
diff --git a/src/sinks/mod.rs b/src/sinks/mod.rs
index d349167f24f06..1a5b03e06c63e 100644
--- a/src/sinks/mod.rs
+++ b/src/sinks/mod.rs
@@ -20,8 +20,8 @@ pub mod aws_cloudwatch_metrics;
 pub mod aws_kinesis;
 #[cfg(feature = "sinks-aws_s3")]
 pub mod aws_s3;
-#[cfg(feature = "sinks-aws_sqs")]
-pub mod aws_sqs;
+#[cfg(any(feature = "sinks-aws_sqs", feature = "sinks-sinks-aws_sns"))]
+pub mod aws_s_s;
 #[cfg(feature = "sinks-axiom")]
 pub mod axiom;
 #[cfg(feature = "sinks-azure_blob")]
diff --git a/website/cue/reference/components/sinks/base/aws_sns.cue b/website/cue/reference/components/sinks/base/aws_sns.cue
new file mode 100644
index 0000000000000..8d7bc06f2405f
--- /dev/null
+++ b/website/cue/reference/components/sinks/base/aws_sns.cue
@@ -0,0 +1,550 @@
+package metadata
+
+base: components: sinks: aws_sns: configuration: {
+	acknowledgements: {
+		description: """
+			Controls how acknowledgements are handled for this sink.
+
+			See [End-to-end Acknowledgements][e2e_acks] for more information on how event acknowledgement is handled.
+
+			[e2e_acks]: https://vector.dev/docs/about/under-the-hood/architecture/end-to-end-acknowledgements/
+			"""
+		required: false
+		type: object: options: enabled: {
+			description: """
+				Whether or not end-to-end acknowledgements are enabled.
+
+				When enabled for a sink, any source connected to that sink, where the source supports
+				end-to-end acknowledgements as well, waits for events to be acknowledged by the sink
+				before acknowledging them at the source.
+
+				Enabling or disabling acknowledgements at the sink level takes precedence over any global
+				[`acknowledgements`][global_acks] configuration.
+
+				[global_acks]: https://vector.dev/docs/reference/configuration/global-options/#acknowledgements
+				"""
+			required: false
+			type: bool: {}
+		}
+	}
+	auth: {
+		description: "Configuration of the authentication strategy for interacting with AWS services."
+		required:    false
+		type: object: options: {
+			access_key_id: {
+				description: "The AWS access key ID."
+				required:    true
+				type: string: examples: ["AKIAIOSFODNN7EXAMPLE"]
+			}
+			assume_role: {
+				description: """
+					The ARN of an [IAM role][iam_role] to assume.
+
+					[iam_role]: https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles.html
+					"""
+				required: true
+				type: string: examples: ["arn:aws:iam::123456789098:role/my_role"]
+			}
+			credentials_file: {
+				description: "Path to the credentials file."
+				required:    true
+				type: string: examples: ["/my/aws/credentials"]
+			}
+			external_id: {
+				description: """
+					The optional unique external ID in conjunction with role to assume.
+
+					[external_id]: https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles_create_for-user_externalid.html
+					"""
+				required: false
+				type: string: examples: ["randomEXAMPLEidString"]
+			}
+			imds: {
+				description: "Configuration for authenticating with AWS through IMDS."
+				required:    false
+				type: object: options: {
+					connect_timeout_seconds: {
+						description: "Connect timeout for IMDS."
+						required:    false
+						type: uint: {
+							default: 1
+							unit:    "seconds"
+						}
+					}
+					max_attempts: {
+						description: "Number of IMDS retries for fetching tokens and metadata."
+						required:    false
+						type: uint: default: 4
+					}
+					read_timeout_seconds: {
+						description: "Read timeout for IMDS."
+						required:    false
+						type: uint: {
+							default: 1
+							unit:    "seconds"
+						}
+					}
+				}
+			}
+			load_timeout_secs: {
+				description: """
+					Timeout for successfully loading any credentials, in seconds.
+
+					Relevant when the default credentials chain or `assume_role` is used.
+					"""
+				required: false
+				type: uint: {
+					examples: [30]
+					unit: "seconds"
+				}
+			}
+			profile: {
+				description: """
+					The credentials profile to use.
+
+					Used to select AWS credentials from a provided credentials file.
+					"""
+				required: false
+				type: string: {
+					default: "default"
+					examples: ["develop"]
+				}
+			}
+			region: {
+				description: """
+					The [AWS region][aws_region] to send STS requests to.
+
+					If not set, this defaults to the configured region
+					for the service itself.
+
+					[aws_region]: https://docs.aws.amazon.com/general/latest/gr/rande.html#regional-endpoints
+					"""
+				required: false
+				type: string: examples: ["us-west-2"]
+			}
+			secret_access_key: {
+				description: "The AWS secret access key."
+				required:    true
+				type: string: examples: ["wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"]
+			}
+		}
+	}
+	encoding: {
+		description: "Configures how events are encoded into raw bytes."
+		required:    true
+		type: object: options: {
+			avro: {
+				description:   "Apache Avro-specific encoder options."
+				relevant_when: "codec = \"avro\""
+				required:      true
+				type: object: options: schema: {
+					description: "The Avro schema."
+					required:    true
+					type: string: examples: ["{ \"type\": \"record\", \"name\": \"log\", \"fields\": [{ \"name\": \"message\", \"type\": \"string\" }] }"]
+				}
+			}
+			codec: {
+				description: "The codec to use for encoding events."
+				required:    true
+				type: string: enum: {
+					avro: """
+						Encodes an event as an [Apache Avro][apache_avro] message.
+
+						[apache_avro]: https://avro.apache.org/
+						"""
+					csv: """
+						Encodes an event as a CSV message.
+
+						This codec must be configured with fields to encode.
+						"""
+					gelf: """
+						Encodes an event as a [GELF][gelf] message.
+
+						[gelf]: https://docs.graylog.org/docs/gelf
+						"""
+					json: """
+						Encodes an event as [JSON][json].
+
+						[json]: https://www.json.org/
+						"""
+					logfmt: """
+						Encodes an event as a [logfmt][logfmt] message.
+
+						[logfmt]: https://brandur.org/logfmt
+						"""
+					native: """
+						Encodes an event in the [native Protocol Buffers format][vector_native_protobuf].
+
+						This codec is **[experimental][experimental]**.
+
+						[vector_native_protobuf]: https://github.com/vectordotdev/vector/blob/master/lib/vector-core/proto/event.proto
+						[experimental]: https://vector.dev/highlights/2022-03-31-native-event-codecs
+						"""
+					native_json: """
+						Encodes an event in the [native JSON format][vector_native_json].
+
+						This codec is **[experimental][experimental]**.
+
+						[vector_native_json]: https://github.com/vectordotdev/vector/blob/master/lib/codecs/tests/data/native_encoding/schema.cue
+						[experimental]: https://vector.dev/highlights/2022-03-31-native-event-codecs
+						"""
+					raw_message: """
+						No encoding.
+
+						This encoding uses the `message` field of a log event.
+
+						Be careful if you are modifying your log events (for example, by using a `remap`
+						transform) and removing the message field while doing additional parsing on it, as this
+						could lead to the encoding emitting empty strings for the given event.
+						"""
+					text: """
+						Plain text encoding.
+
+						This encoding uses the `message` field of a log event. For metrics, it uses an
+						encoding that resembles the Prometheus export format.
+
+						Be careful if you are modifying your log events (for example, by using a `remap`
+						transform) and removing the message field while doing additional parsing on it, as this
+						could lead to the encoding emitting empty strings for the given event.
+						"""
+				}
+			}
+			csv: {
+				description:   "The CSV Serializer Options."
+				relevant_when: "codec = \"csv\""
+				required:      true
+				type: object: options: fields: {
+					description: """
+						Configures the fields that will be encoded, as well as the order in which they
+						appear in the output.
+
+						If a field is not present in the event, the output will be an empty string.
+
+						Values of type `Array`, `Object`, and `Regex` are not supported and the
+						output will be an empty string.
+						"""
+					required: true
+					type: array: items: type: string: {}
+				}
+			}
+			except_fields: {
+				description: "List of fields that are excluded from the encoded event."
+				required:    false
+				type: array: items: type: string: {}
+			}
+			metric_tag_values: {
+				description: """
+					Controls how metric tag values are encoded.
+
+					When set to `single`, only the last non-bare value of tags are displayed with the
+					metric.  When set to `full`, all metric tags are exposed as separate assignments.
+					"""
+				relevant_when: "codec = \"json\" or codec = \"text\""
+				required:      false
+				type: string: {
+					default: "single"
+					enum: {
+						full: "All tags are exposed as arrays of either string or null values."
+						single: """
+															Tag values are exposed as single strings, the same as they were before this config
+															option. Tags with multiple values show the last assigned value, and null values
+															are ignored.
+															"""
+					}
+				}
+			}
+			only_fields: {
+				description: "List of fields that are included in the encoded event."
+				required:    false
+				type: array: items: type: string: {}
+			}
+			timestamp_format: {
+				description: "Format used for timestamp fields."
+				required:    false
+				type: string: enum: {
+					rfc3339: "Represent the timestamp as a RFC 3339 timestamp."
+					unix:    "Represent the timestamp as a Unix timestamp."
+				}
+			}
+		}
+	}
+	endpoint: {
+		description: "Custom endpoint for use with AWS-compatible services."
+		required:    false
+		type: string: examples: ["http://127.0.0.0:5000/path/to/service"]
+	}
+	message_deduplication_id: {
+		description: """
+			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
+			"""
+		required: false
+		type: string: examples: ["{{ transaction_id }}"]
+	}
+	message_group_id: {
+		description: """
+			The tag that specifies that a message belongs to a specific message group.
+
+			Can be applied only to FIFO queues.
+			"""
+		required: false
+		type: string: examples: ["vector", "vector-%Y-%m-%d"]
+	}
+	region: {
+		description: """
+			The [AWS region][aws_region] of the target service.
+
+			[aws_region]: https://docs.aws.amazon.com/general/latest/gr/rande.html#regional-endpoints
+			"""
+		required: false
+		type: string: examples: ["us-east-1"]
+	}
+	request: {
+		description: """
+			Middleware settings for outbound requests.
+
+			Various settings can be configured, such as concurrency and rate limits, timeouts, etc.
+			"""
+		required: false
+		type: object: options: {
+			adaptive_concurrency: {
+				description: """
+					Configuration of adaptive concurrency parameters.
+
+					These parameters typically do not require changes from the default, and incorrect values can lead to meta-stable or
+					unstable performance and sink behavior. Proceed with caution.
+					"""
+				required: false
+				type: object: options: {
+					decrease_ratio: {
+						description: """
+																The fraction of the current value to set the new concurrency limit when decreasing the limit.
+
+																Valid values are greater than `0` and less than `1`. Smaller values cause the algorithm to scale back rapidly
+																when latency increases.
+
+																Note that the new limit is rounded down after applying this ratio.
+																"""
+						required: false
+						type: float: default: 0.9
+					}
+					ewma_alpha: {
+						description: """
+																The weighting of new measurements compared to older measurements.
+
+																Valid values are greater than `0` and less than `1`.
+
+																ARC uses an exponentially weighted moving average (EWMA) of past RTT measurements as a reference to compare with
+																the current RTT. Smaller values cause this reference to adjust more slowly, which may be useful if a service has
+																unusually high response variability.
+																"""
+						required: false
+						type: float: default: 0.4
+					}
+					initial_concurrency: {
+						description: """
+																The initial concurrency limit to use. If not specified, the initial limit will be 1 (no concurrency).
+
+																It is recommended to set this value to your service's average limit if you're seeing that it takes a
+																long time to ramp up adaptive concurrency after a restart. You can find this value by looking at the
+																`adaptive_concurrency_limit` metric.
+																"""
+						required: false
+						type: uint: default: 1
+					}
+					rtt_deviation_scale: {
+						description: """
+																Scale of RTT deviations which are not considered anomalous.
+
+																Valid values are greater than or equal to `0`, and we expect reasonable values to range from `1.0` to `3.0`.
+
+																When calculating the past RTT average, we also compute a secondary “deviation” value that indicates how variable
+																those values are. We use that deviation when comparing the past RTT average to the current measurements, so we
+																can ignore increases in RTT that are within an expected range. This factor is used to scale up the deviation to
+																an appropriate range.  Larger values cause the algorithm to ignore larger increases in the RTT.
+																"""
+						required: false
+						type: float: default: 2.5
+					}
+				}
+			}
+			concurrency: {
+				description: "Configuration for outbound request concurrency."
+				required:    false
+				type: {
+					string: {
+						default: "none"
+						enum: {
+							adaptive: """
+															Concurrency will be managed by Vector's [Adaptive Request Concurrency][arc] feature.
+
+															[arc]: https://vector.dev/docs/about/under-the-hood/networking/arc/
+															"""
+							none: """
+															A fixed concurrency of 1.
+
+															Only one request can be outstanding at any given time.
+															"""
+						}
+					}
+					uint: {}
+				}
+			}
+			rate_limit_duration_secs: {
+				description: "The time window used for the `rate_limit_num` option."
+				required:    false
+				type: uint: {
+					default: 1
+					unit:    "seconds"
+				}
+			}
+			rate_limit_num: {
+				description: "The maximum number of requests allowed within the `rate_limit_duration_secs` time window."
+				required:    false
+				type: uint: {
+					default: 9223372036854775807
+					unit:    "requests"
+				}
+			}
+			retry_attempts: {
+				description: """
+					The maximum number of retries to make for failed requests.
+
+					The default, for all intents and purposes, represents an infinite number of retries.
+					"""
+				required: false
+				type: uint: {
+					default: 9223372036854775807
+					unit:    "retries"
+				}
+			}
+			retry_initial_backoff_secs: {
+				description: """
+					The amount of time to wait before attempting the first retry for a failed request.
+
+					After the first retry has failed, the fibonacci sequence is used to select future backoffs.
+					"""
+				required: false
+				type: uint: {
+					default: 1
+					unit:    "seconds"
+				}
+			}
+			retry_max_duration_secs: {
+				description: "The maximum amount of time to wait between retries."
+				required:    false
+				type: uint: {
+					default: 3600
+					unit:    "seconds"
+				}
+			}
+			timeout_secs: {
+				description: """
+					The time a request can take before being aborted.
+
+					Datadog highly recommends that you do not lower this value below the service's internal timeout, as this could
+					create orphaned requests, pile on retries, and result in duplicate data downstream.
+					"""
+				required: false
+				type: uint: {
+					default: 60
+					unit:    "seconds"
+				}
+			}
+		}
+	}
+	tls: {
+		description: "TLS configuration."
+		required:    false
+		type: object: options: {
+			alpn_protocols: {
+				description: """
+					Sets the list of supported ALPN protocols.
+
+					Declare the supported ALPN protocols, which are used during negotiation with peer. They are prioritized in the order
+					that they are defined.
+					"""
+				required: false
+				type: array: items: type: string: examples: ["h2"]
+			}
+			ca_file: {
+				description: """
+					Absolute path to an additional CA certificate file.
+
+					The certificate must be in the DER or PEM (X.509) format. Additionally, the certificate can be provided as an inline string in PEM format.
+					"""
+				required: false
+				type: string: examples: ["/path/to/certificate_authority.crt"]
+			}
+			crt_file: {
+				description: """
+					Absolute path to a certificate file used to identify this server.
+
+					The certificate must be in DER, PEM (X.509), or PKCS#12 format. Additionally, the certificate can be provided as
+					an inline string in PEM format.
+
+					If this is set, and is not a PKCS#12 archive, `key_file` must also be set.
+					"""
+				required: false
+				type: string: examples: ["/path/to/host_certificate.crt"]
+			}
+			key_file: {
+				description: """
+					Absolute path to a private key file used to identify this server.
+
+					The key must be in DER or PEM (PKCS#8) format. Additionally, the key can be provided as an inline string in PEM format.
+					"""
+				required: false
+				type: string: examples: ["/path/to/host_certificate.key"]
+			}
+			key_pass: {
+				description: """
+					Passphrase used to unlock the encrypted key file.
+
+					This has no effect unless `key_file` is set.
+					"""
+				required: false
+				type: string: examples: ["${KEY_PASS_ENV_VAR}", "PassWord1"]
+			}
+			verify_certificate: {
+				description: """
+					Enables certificate verification.
+
+					If enabled, certificates must not be expired and must be issued by a trusted
+					issuer. This verification operates in a hierarchical manner, checking that the leaf certificate (the
+					certificate presented by the client/server) is not only valid, but that the issuer of that certificate is also valid, and
+					so on until the verification process reaches a root certificate.
+
+					Relevant for both incoming and outgoing connections.
+
+					Do NOT set this to `false` unless you understand the risks of not verifying the validity of certificates.
+					"""
+				required: false
+				type: bool: {}
+			}
+			verify_hostname: {
+				description: """
+					Enables hostname verification.
+
+					If enabled, the hostname used to connect to the remote host must be present in the TLS certificate presented by
+					the remote host, either as the Common Name or as an entry in the Subject Alternative Name extension.
+
+					Only relevant for outgoing connections.
+
+					Do NOT set this to `false` unless you understand the risks of not verifying the remote hostname.
+					"""
+				required: false
+				type: bool: {}
+			}
+		}
+	}
+	topic_arn: {
+		description: "The ARN of the Amazon SNS topic to which messages are sent."
+		required:    true
+		type: string: examples: ["arn:aws:sns:us-east-2:123456789012:MyTopic"]
+	}
+}