Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat(aws_s3 sink): add option to use virtual addressing #21999

Merged
merged 3 commits into from
Dec 20, 2024
Merged

feat(aws_s3 sink): add option to use virtual addressing #21999

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
d03f11b
feat(aws_s3 sink): add option to use virtual addressing
sam6258 Nov 22, 2024
337b667
fix(aws_s3 sink): move force_path_style to S3ClientBuilder
sam6258 Dec 19, 2024
a0ac4a0
fix(aws_s3 sink): simplify S3ClientBuilder logic
sam6258 Dec 19, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
43 changes: 17 additions & 26 deletions src/aws/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -123,10 +123,7 @@ pub trait ClientBuilder {
type Client;

/// Build the client using the given config settings.
fn build(config: &SdkConfig) -> Self::Client;

/// Build the client using the given config settings and path style addressing.
fn build_and_force_path_style(config: &aws_types::SdkConfig) -> Self::Client;
fn build(&self, config: &SdkConfig) -> Self::Client;
}

fn region_provider(
Expand Down Expand Up @@ -164,38 +161,36 @@ async fn resolve_region(
}

/// Create the SDK client using the provided settings.
pub async fn create_client<T: ClientBuilder>(
pub async fn create_client<T>(
builder: &T,
auth: &AwsAuthentication,
region: Option<Region>,
endpoint: Option<String>,
proxy: &ProxyConfig,
tls_options: &Option<TlsConfig>,
timeout: &Option<AwsTimeout>,
force_path_style: impl Into<bool>,
) -> crate::Result<T::Client> {
create_client_and_region::<T>(
auth,
region,
endpoint,
proxy,
tls_options,
timeout,
force_path_style,
)
.await
.map(|(client, _)| client)
) -> crate::Result<T::Client>
where
T: ClientBuilder,
{
create_client_and_region::<T>(builder, auth, region, endpoint, proxy, tls_options, timeout)
.await
.map(|(client, _)| client)
}

/// Create the SDK client and resolve the region using the provided settings.
pub async fn create_client_and_region<T: ClientBuilder>(
pub async fn create_client_and_region<T>(
builder: &T,
auth: &AwsAuthentication,
region: Option<Region>,
endpoint: Option<String>,
proxy: &ProxyConfig,
tls_options: &Option<TlsConfig>,
timeout: &Option<AwsTimeout>,
force_path_style: impl Into<bool>,
) -> crate::Result<(T::Client, Region)> {
) -> crate::Result<(T::Client, Region)>
where
T: ClientBuilder,
{
let retry_config = RetryConfig::disabled();

// The default credentials chains will look for a region if not given but we'd like to
Expand Down Expand Up @@ -252,11 +247,7 @@ pub async fn create_client_and_region<T: ClientBuilder>(

let config = config_builder.build();

if force_path_style.into() {
Ok((T::build_and_force_path_style(&config), region))
} else {
Ok((T::build(&config), region))
}
Ok((T::build(builder, &config), region))
}

#[derive(Snafu, Debug)]
Expand Down
18 changes: 10 additions & 8 deletions src/common/s3.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,18 +2,20 @@ use aws_sdk_s3::config;

use crate::aws::ClientBuilder;

pub(crate) struct S3ClientBuilder;
pub(crate) struct S3ClientBuilder {
pub force_path_style: Option<bool>,
}

impl ClientBuilder for S3ClientBuilder {
type Client = aws_sdk_s3::client::Client;

fn build(config: &aws_types::SdkConfig) -> Self::Client {
let config = config::Builder::from(config).build();
aws_sdk_s3::client::Client::from_conf(config)
}
fn build(&self, config: &aws_types::SdkConfig) -> Self::Client {
let mut builder = config::Builder::from(config);

if let Some(true) = self.force_path_style {
builder = builder.force_path_style(true);
}
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think you could avoid the if let Some if you just do: builder = builder.set_force_path_style(self.force_path_style). The set_* variants on the AWS SDK take Options.

Copy link
Contributor Author

@sam6258 sam6258 Dec 19, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done! I opted to use force_path_style still so I could unwrap to true by default in case someone forgets to set the value in S3ClientBuilder. That way it maintains vector's previous behavior even in a programming error.


fn build_and_force_path_style(config: &aws_types::SdkConfig) -> Self::Client {
let config = config::Builder::from(config).force_path_style(true).build();
aws_sdk_s3::client::Client::from_conf(config)
aws_sdk_s3::client::Client::from_conf(builder.build())
}
}
6 changes: 1 addition & 5 deletions src/common/sqs.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,11 +5,7 @@ pub(crate) struct SqsClientBuilder;
impl ClientBuilder for SqsClientBuilder {
type Client = aws_sdk_sqs::client::Client;

fn build(config: &aws_types::SdkConfig) -> Self::Client {
fn build(&self, config: &aws_types::SdkConfig) -> Self::Client {
aws_sdk_sqs::client::Client::new(config)
}

fn build_and_force_path_style(config: &aws_types::SdkConfig) -> Self::Client {
SqsClientBuilder::build(config)
}
}
8 changes: 2 additions & 6 deletions src/secrets/aws_secrets_manager.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,14 +13,10 @@ pub(crate) struct SecretsManagerClientBuilder;
impl ClientBuilder for SecretsManagerClientBuilder {
type Client = Client;

fn build(config: &aws_types::SdkConfig) -> Self::Client {
fn build(&self, config: &aws_types::SdkConfig) -> Self::Client {
let config = config::Builder::from(config).build();
Client::from_conf(config)
}

fn build_and_force_path_style(config: &aws_types::SdkConfig) -> Self::Client {
SecretsManagerClientBuilder::build(config)
}
}

/// Configuration for the `aws_secrets_manager` secrets backend.
Expand Down Expand Up @@ -61,13 +57,13 @@ impl SecretBackend for AwsSecretsManagerBackend {
_: &mut signal::SignalRx,
) -> crate::Result<HashMap<String, String>> {
let client = create_client::<SecretsManagerClientBuilder>(
&SecretsManagerClientBuilder {},
&self.auth,
self.region.region(),
self.region.endpoint(),
&ProxyConfig::default(),
&self.tls,
&None,
false,
)
.await?;

Expand Down
8 changes: 2 additions & 6 deletions src/sinks/aws_cloudwatch_logs/config.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,13 +33,9 @@ pub struct CloudwatchLogsClientBuilder;
impl ClientBuilder for CloudwatchLogsClientBuilder {
type Client = aws_sdk_cloudwatchlogs::client::Client;

fn build(config: &aws_types::SdkConfig) -> Self::Client {
fn build(&self, config: &aws_types::SdkConfig) -> Self::Client {
aws_sdk_cloudwatchlogs::client::Client::new(config)
}

fn build_and_force_path_style(config: &aws_types::SdkConfig) -> Self::Client {
CloudwatchLogsClientBuilder::build(config)
}
}

#[configurable_component]
Expand Down Expand Up @@ -173,13 +169,13 @@ pub struct CloudwatchLogsSinkConfig {
impl CloudwatchLogsSinkConfig {
pub async fn create_client(&self, proxy: &ProxyConfig) -> crate::Result<CloudwatchLogsClient> {
create_client::<CloudwatchLogsClientBuilder>(
&CloudwatchLogsClientBuilder {},
&self.auth,
self.region.region(),
self.region.endpoint(),
proxy,
&self.tls,
&None,
false,
)
.await
}
Expand Down
8 changes: 7 additions & 1 deletion src/sinks/aws_cloudwatch_logs/integration_tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -468,7 +468,13 @@ async fn create_client_test() -> CloudwatchLogsClient {
let proxy = ProxyConfig::default();

create_client::<CloudwatchLogsClientBuilder>(
&auth, region, endpoint, &proxy, &None, &None, false,
&CloudwatchLogsClientBuilder {},
&auth,
region,
endpoint,
&proxy,
&None,
&None,
)
.await
.unwrap()
Expand Down
8 changes: 2 additions & 6 deletions src/sinks/aws_cloudwatch_metrics/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -119,13 +119,9 @@ struct CloudwatchMetricsClientBuilder;
impl ClientBuilder for CloudwatchMetricsClientBuilder {
type Client = aws_sdk_cloudwatch::client::Client;

fn build(config: &aws_types::SdkConfig) -> Self::Client {
fn build(&self, config: &aws_types::SdkConfig) -> Self::Client {
aws_sdk_cloudwatch::client::Client::new(config)
}

fn build_and_force_path_style(config: &aws_types::SdkConfig) -> Self::Client {
CloudwatchMetricsClientBuilder::build(config)
}
}

#[async_trait::async_trait]
Expand Down Expand Up @@ -176,13 +172,13 @@ impl CloudWatchMetricsSinkConfig {
};

create_client::<CloudwatchMetricsClientBuilder>(
&CloudwatchMetricsClientBuilder {},
&self.auth,
region,
self.region.endpoint(),
proxy,
&self.tls,
&None,
false,
)
.await
}
Expand Down
8 changes: 2 additions & 6 deletions src/sinks/aws_kinesis/firehose/config.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,13 +38,9 @@ pub struct KinesisFirehoseClientBuilder;
impl ClientBuilder for KinesisFirehoseClientBuilder {
type Client = KinesisClient;

fn build(config: &aws_types::SdkConfig) -> Self::Client {
fn build(&self, config: &aws_types::SdkConfig) -> Self::Client {
Self::Client::new(config)
}

fn build_and_force_path_style(config: &aws_types::SdkConfig) -> Self::Client {
KinesisFirehoseClientBuilder::build(config)
}
}

// AWS Kinesis Firehose API accepts payloads up to 4MB or 500 events
Expand Down Expand Up @@ -106,13 +102,13 @@ impl KinesisFirehoseSinkConfig {

pub async fn create_client(&self, proxy: &ProxyConfig) -> crate::Result<KinesisClient> {
create_client::<KinesisFirehoseClientBuilder>(
&KinesisFirehoseClientBuilder {},
&self.base.auth,
self.base.region.region(),
self.base.region.endpoint(),
proxy,
&self.base.tls,
&None,
false,
)
.await
}
Expand Down
2 changes: 1 addition & 1 deletion src/sinks/aws_kinesis/firehose/integration_tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -255,13 +255,13 @@ async fn firehose_client() -> aws_sdk_firehose::Client {
let proxy = ProxyConfig::default();

create_client::<KinesisFirehoseClientBuilder>(
&KinesisFirehoseClientBuilder {},
&auth,
region_endpoint.region(),
region_endpoint.endpoint(),
&proxy,
&None,
&None,
false,
)
.await
.unwrap()
Expand Down
8 changes: 2 additions & 6 deletions src/sinks/aws_kinesis/streams/config.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,13 +37,9 @@ pub struct KinesisClientBuilder;
impl ClientBuilder for KinesisClientBuilder {
type Client = KinesisClient;

fn build(config: &aws_types::SdkConfig) -> Self::Client {
fn build(&self, config: &aws_types::SdkConfig) -> Self::Client {
KinesisClient::new(config)
}

fn build_and_force_path_style(config: &aws_types::SdkConfig) -> Self::Client {
KinesisClientBuilder::build(config)
}
}

pub const MAX_PAYLOAD_SIZE: usize = 5_000_000;
Expand Down Expand Up @@ -103,13 +99,13 @@ impl KinesisStreamsSinkConfig {

pub async fn create_client(&self, proxy: &ProxyConfig) -> crate::Result<KinesisClient> {
create_client::<KinesisClientBuilder>(
&KinesisClientBuilder {},
&self.base.auth,
self.base.region.region(),
self.base.region.endpoint(),
proxy,
&self.base.tls,
&None,
false,
)
.await
}
Expand Down
2 changes: 1 addition & 1 deletion src/sinks/aws_kinesis/streams/integration_tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -177,13 +177,13 @@ async fn client() -> aws_sdk_kinesis::Client {
let proxy = ProxyConfig::default();
let region = RegionOrEndpoint::with_both("us-east-1", kinesis_address());
create_client::<KinesisClientBuilder>(
&KinesisClientBuilder {},
&auth,
region.region(),
region.endpoint(),
&proxy,
&None,
&None,
false,
)
.await
.unwrap()
Expand Down
6 changes: 5 additions & 1 deletion src/sinks/aws_s3/integration_tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -490,14 +490,18 @@ async fn client() -> S3Client {
let region = RegionOrEndpoint::with_both("us-east-1", s3_address());
let proxy = ProxyConfig::default();
let tls_options = None;
let force_path_style_value: bool = true;

create_client::<S3ClientBuilder>(
&S3ClientBuilder {
force_path_style: Some(force_path_style_value),
},
&auth,
region.region(),
region.endpoint(),
&proxy,
&tls_options,
&None,
true,
)
.await
.unwrap()
Expand Down
8 changes: 2 additions & 6 deletions src/sinks/aws_s_s/sns/config.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,13 +48,13 @@ impl GenerateConfig for SnsSinkConfig {
impl SnsSinkConfig {
pub(super) async fn create_client(&self, proxy: &ProxyConfig) -> crate::Result<SnsClient> {
create_client::<SnsClientBuilder>(
&SnsClientBuilder {},
&self.base_config.auth,
self.region.region(),
self.region.endpoint(),
proxy,
&self.base_config.tls,
&None,
false,
)
.await
}
Expand Down Expand Up @@ -109,13 +109,9 @@ pub(super) struct SnsClientBuilder;
impl ClientBuilder for SnsClientBuilder {
type Client = aws_sdk_sns::client::Client;

fn build(config: &aws_types::SdkConfig) -> Self::Client {
fn build(&self, config: &aws_types::SdkConfig) -> Self::Client {
aws_sdk_sns::client::Client::new(config)
}

fn build_and_force_path_style(config: &aws_types::SdkConfig) -> Self::Client {
SnsClientBuilder::build(config)
}
}

pub(super) async fn healthcheck(client: SnsClient, topic_arn: String) -> crate::Result<()> {
Expand Down
4 changes: 2 additions & 2 deletions src/sinks/aws_s_s/sns/integration_tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,13 +32,13 @@ async fn create_sns_test_client() -> SnsClient {
let endpoint = sns_address();
let proxy = ProxyConfig::default();
create_client::<SnsClientBuilder>(
&SnsClientBuilder {},
&auth,
Some(Region::new("us-east-1")),
Some(endpoint),
&proxy,
&None,
&None,
false,
)
.await
.unwrap()
Expand All @@ -54,13 +54,13 @@ async fn create_sqs_test_client() -> SqsClient {
let endpoint = sqs_address();
let proxy = ProxyConfig::default();
create_client::<SqsClientBuilder>(
&SqsClientBuilder {},
&auth,
Some(Region::new("us-east-1")),
Some(endpoint),
&proxy,
&None,
&None,
false,
)
.await
.unwrap()
Expand Down
Loading
Loading