fix(aws_s3 source): Allow region to be optional (#18258)

When constructing the client we detect the region so we can use that for consistency rather than
requiring it to be specified (so that, for example, `AWS_REGION` is used if set).

Signed-off-by: Jesse Szwedko <jesse.szwedko@datadoghq.com>

src/aws/mod.rs

@@ -149,6 +149,19 @@ pub async fn create_client<T: ClientBuilder>(
     tls_options: &Option<TlsConfig>,
     is_sink: bool,
 ) -> crate::Result<T::Client> {
+    create_client_and_region::<T>(auth, region, endpoint, proxy, tls_options, is_sink)
+        .await
+        .map(|(client, _)| client)
+}
+
+pub async fn create_client_and_region<T: ClientBuilder>(
+    auth: &AwsAuthentication,
+    region: Option<Region>,
+    endpoint: Option<String>,
+    proxy: &ProxyConfig,
+    tls_options: &Option<TlsConfig>,
+    is_sink: bool,
+) -> crate::Result<(T::Client, Region)> {
     let retry_config = RetryConfig::disabled();
 
     // The default credentials chains will look for a region if not given but we'd like to
@@ -169,9 +182,10 @@ pub async fn create_client<T: ClientBuilder>(
     let config = config_builder.build();
 
     let client =
-        create_smithy_client::<T>(region, proxy, tls_options, is_sink, retry_config).await?;
+        create_smithy_client::<T>(region.clone(), proxy, tls_options, is_sink, retry_config)
+            .await?;
 
-    Ok(T::build(client, &config))
+    Ok((T::build(client, &config), region))
 }
 
 pub async fn sign_request(

src/sources/aws_s3/mod.rs

@@ -16,7 +16,7 @@ use super::util::MultilineConfig;
 use crate::codecs::DecodingConfig;
 use crate::config::DataType;
 use crate::{
-    aws::{auth::AwsAuthentication, create_client, RegionOrEndpoint},
+    aws::{auth::AwsAuthentication, create_client, create_client_and_region, RegionOrEndpoint},
     common::{s3::S3ClientBuilder, sqs::SqsClientBuilder},
     config::{
         ProxyConfig, SourceAcknowledgementsConfig, SourceConfig, SourceContext, SourceOutput,
@@ -223,16 +223,12 @@ impl AwsS3Config {
         proxy: &ProxyConfig,
         log_namespace: LogNamespace,
     ) -> crate::Result<sqs::Ingestor> {
-        let region = self
-            .region
-            .region()
-            .ok_or(CreateSqsIngestorError::RegionMissing)?;
-
+        let region = self.region.region();
         let endpoint = self.region.endpoint();
 
         let s3_client = create_client::<S3ClientBuilder>(
             &self.auth,
-            Some(region.clone()),
+            region.clone(),
             endpoint.clone(),
             proxy,
             &self.tls_options,
@@ -246,9 +242,9 @@ impl AwsS3Config {
 
         match self.sqs {
             Some(ref sqs) => {
-                let sqs_client = create_client::<SqsClientBuilder>(
+                let (sqs_client, region) = create_client_and_region::<SqsClientBuilder>(
                     &self.auth,
-                    Some(region.clone()),
+                    region.clone(),
                     endpoint,
                     proxy,
                     &sqs.tls_options,
@@ -284,8 +280,6 @@ enum CreateSqsIngestorError {
     Credentials { source: crate::Error },
     #[snafu(display("Configuration for `sqs` required when strategy=sqs"))]
     ConfigMissing,
-    #[snafu(display("Region is required"))]
-    RegionMissing,
     #[snafu(display("Endpoint is invalid"))]
     InvalidEndpoint,
 }