fix(ARC, networking): improve request settings

src/components/validation/runner/config.rs

@@ -141,7 +141,7 @@ fn build_output_edge() -> (OutputEdge, impl Into<BoxedSink>) {
     // we don't want to waste time performing retries, especially when the test
     // harness is shutting down.
     output_sink.batch.timeout_secs = Some(0.1);
-    output_sink.request.retry_attempts = Some(0);
+    output_sink.request.retry_attempts = 0;
 
     let output_edge = OutputEdge::from_address(output_listen_addr);
 

src/components/validation/runner/telemetry.rs

@@ -52,7 +52,7 @@ impl Telemetry {
         // disable retries, as we don't want to waste time performing retries,
         // especially when the test harness is shutting down.
         vector_sink.batch.timeout_secs = Some(0.1);
-        vector_sink.request.retry_attempts = Some(0);
+        vector_sink.request.retry_attempts = 0;
 
         config_builder.add_source(INTERNAL_LOGS_KEY, internal_logs);
         config_builder.add_source(INTERNAL_METRICS_KEY, internal_metrics);

src/sinks/appsignal/config.rs

@@ -102,7 +102,7 @@ impl AppsignalConfig {
         let service = AppsignalService::new(http_client, endpoint, push_api_key, compression);
 
         let request_opts = self.request;
-        let request_settings = request_opts.unwrap_with(&TowerRequestConfig::default());
+        let request_settings = request_opts.into_settings();
         let retry_logic = HttpStatusRetryLogic::new(|req: &AppsignalResponse| req.http_status);
 
         let service = ServiceBuilder::new()

src/sinks/aws_cloudwatch_logs/config.rs

@@ -25,7 +25,6 @@ use crate::{
         },
         util::{
             http::RequestConfig, BatchConfig, Compression, ServiceBuilderExt, SinkBatchSettings,
-            TowerRequestConfig,
         },
         Healthcheck, VectorSink,
     },
@@ -211,10 +210,7 @@ impl CloudwatchLogsSinkConfig {
 impl SinkConfig for CloudwatchLogsSinkConfig {
     async fn build(&self, cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> {
         let batcher_settings = self.batch.into_batcher_settings()?;
-        let request_settings = self
-            .request
-            .tower
-            .unwrap_with(&TowerRequestConfig::default());
+        let request_settings = self.request.tower.into_settings();
         let client = self.create_client(cx.proxy()).await?;
         let smithy_client = self.create_smithy_client(cx.proxy()).await?;
         let svc = ServiceBuilder::new()

src/sinks/aws_cloudwatch_logs/service.rs

@@ -34,7 +34,7 @@ use crate::sinks::{
         config::CloudwatchLogsSinkConfig, config::Retention, request, retry::CloudwatchRetryLogic,
         sink::BatchCloudwatchRequest, CloudwatchKey,
     },
-    util::{retries::FixedRetryPolicy, EncodedLength, TowerRequestConfig, TowerRequestSettings},
+    util::{retries::FixedRetryPolicy, EncodedLength, TowerRequestSettings},
 };
 
 type Svc = Buffer<
@@ -136,10 +136,7 @@ impl CloudwatchLogsPartitionSvc {
         // https://github.com/awslabs/aws-sdk-rust/issues/537
         smithy_client: SmithyClient,
     ) -> Self {
-        let request_settings = config
-            .request
-            .tower
-            .unwrap_with(&TowerRequestConfig::default());
+        let request_settings = config.request.tower.into_settings();
 
         Self {
             config,

src/sinks/aws_cloudwatch_metrics/mod.rs

@@ -36,6 +36,8 @@ use crate::{
     tls::TlsConfig,
 };
 
+use super::util::service::TowerRequestConfigDefaults;
+
 #[derive(Clone, Copy, Debug, Default)]
 pub struct CloudWatchMetricsDefaultBatchSettings;
 
@@ -45,6 +47,14 @@ impl SinkBatchSettings for CloudWatchMetricsDefaultBatchSettings {
     const TIMEOUT_SECS: f64 = 1.0;
 }
 
+#[derive(Clone, Copy, Debug)]
+pub struct CloudWatchMetricsTowerRequestConfigDefaults;
+
+impl TowerRequestConfigDefaults for CloudWatchMetricsTowerRequestConfigDefaults {
+    const TIMEOUT_SECS: u64 = 30;
+    const RATE_LIMIT_NUM: u64 = 150;
+}
+
 /// Configuration for the `aws_cloudwatch_metrics` sink.
 #[configurable_component(sink(
     "aws_cloudwatch_metrics",
@@ -79,7 +89,7 @@ pub struct CloudWatchMetricsSinkConfig {
 
     #[configurable(derived)]
     #[serde(default)]
-    pub request: TowerRequestConfig,
+    pub request: TowerRequestConfig<CloudWatchMetricsTowerRequestConfigDefaults>,
 
     #[configurable(derived)]
     pub tls: Option<TlsConfig>,
@@ -225,11 +235,7 @@ impl CloudWatchMetricsSvc {
     ) -> crate::Result<VectorSink> {
         let default_namespace = config.default_namespace.clone();
         let batch = config.batch.into_batch_settings()?;
-        let request_settings = config.request.unwrap_with(
-            &TowerRequestConfig::default()
-                .timeout_secs(30)
-                .rate_limit_num(150),
-        );
+        let request_settings = config.request.into_settings();
 
         let service = CloudWatchMetricsSvc { client };
         let buffer = PartitionBuffer::new(MetricsBuffer::new(batch.size));

src/sinks/aws_kinesis/config.rs

@@ -95,7 +95,7 @@ where
     E: Send + 'static,
     RT: RetryLogic<Response = KinesisResponse> + Default,
 {
-    let request_limits = config.request.unwrap_with(&TowerRequestConfig::default());
+    let request_limits = config.request.into_settings();
 
     let region = config.region.region();
     let service = ServiceBuilder::new()

src/sinks/aws_kinesis/firehose/integration_tests.rs

@@ -52,8 +52,8 @@ async fn firehose_put_records() {
         encoding: JsonSerializerConfig::default().into(), // required for ES destination w/ localstack
         compression: Compression::None,
         request: TowerRequestConfig {
-            timeout_secs: Some(10),
-            retry_attempts: Some(0),
+            timeout_secs: 10,
+            retry_attempts: 0,
             ..Default::default()
         },
         tls: None,

src/sinks/aws_s3/config.rs

@@ -193,7 +193,7 @@ impl S3SinkConfig {
         // requests into in order to ship files to S3.  We build this here in
         // order to configure the client/service with retries, concurrency
         // limits, rate limits, and whatever else the client should have.
-        let request_limits = self.request.unwrap_with(&Default::default());
+        let request_limits = self.request.into_settings();
         let service = ServiceBuilder::new()
             .settings(request_limits, S3RetryLogic)
             .service(service);

src/sinks/aws_s_s/config.rs

@@ -13,6 +13,8 @@ use crate::{
     tls::TlsConfig,
 };
 
+use super::sink::SqsTowerRequestConfigDefaults;
+
 #[derive(Debug, Snafu)]
 pub(super) enum BuildError {
     #[snafu(display("`message_group_id` should be defined for FIFO queue."))]
@@ -51,7 +53,7 @@ pub(super) struct BaseSSSinkConfig {
 
     #[configurable(derived)]
     #[serde(default)]
-    pub(super) request: TowerRequestConfig,
+    pub(super) request: TowerRequestConfig<SqsTowerRequestConfigDefaults>,
 
     #[configurable(derived)]
     pub(super) tls: Option<TlsConfig>,

src/sinks/aws_s_s/sink.rs

@@ -1,6 +1,7 @@
 use super::{client::Client, request_builder::SSRequestBuilder, service::SSService};
 use crate::sinks::aws_s_s::retry::SSRetryLogic;
 use crate::sinks::prelude::*;
+use crate::sinks::util::service::TowerRequestConfigDefaults;
 
 #[derive(Clone, Copy, Debug, Default)]
 pub(crate) struct SqsSinkDefaultBatchSettings;
@@ -11,6 +12,13 @@ impl SinkBatchSettings for SqsSinkDefaultBatchSettings {
     const TIMEOUT_SECS: f64 = 1.0;
 }
 
+#[derive(Clone, Copy, Debug)]
+pub(crate) struct SqsTowerRequestConfigDefaults;
+
+impl TowerRequestConfigDefaults for SqsTowerRequestConfigDefaults {
+    const TIMEOUT_SECS: u64 = 30;
+}
+
 #[derive(Clone)]
 pub(super) struct SSSink<C, E>
 where
@@ -19,7 +27,7 @@ where
 {
     request_builder: SSRequestBuilder,
     service: SSService<C, E>,
-    request: TowerRequestConfig,
+    request: TowerRequestConfig<SqsTowerRequestConfigDefaults>,
 }
 
 impl<C, E> SSSink<C, E>
@@ -29,7 +37,7 @@ where
 {
     pub(super) fn new(
         request_builder: SSRequestBuilder,
-        request: TowerRequestConfig,
+        request: TowerRequestConfig<SqsTowerRequestConfigDefaults>,
         publisher: C,
     ) -> crate::Result<Self> {
         Ok(SSSink {
@@ -40,9 +48,7 @@ where
     }
 
     async fn run_inner(self: Box<Self>, input: BoxStream<'_, Event>) -> Result<(), ()> {
-        let request = self
-            .request
-            .unwrap_with(&TowerRequestConfig::default().timeout_secs(30));
+        let request = self.request.into_settings();
         let retry_logic: SSRetryLogic<E> = super::retry::SSRetryLogic::new();
         let service = tower::ServiceBuilder::new()
             .settings(request, retry_logic)

src/sinks/azure_blob/config.rs

@@ -7,6 +7,7 @@ use vector_lib::configurable::configurable_component;
 use vector_lib::sensitive_string::SensitiveString;
 
 use super::request_builder::AzureBlobRequestOptions;
+use crate::sinks::util::service::TowerRequestConfigDefaults;
 use crate::{
     codecs::{Encoder, EncodingConfigWithFraming, SinkType},
     config::{AcknowledgementsConfig, DataType, GenerateConfig, Input, SinkConfig, SinkContext},
@@ -24,6 +25,13 @@ use crate::{
     Result,
 };
 
+#[derive(Clone, Copy, Debug)]
+pub struct AzureBlobTowerRequestConfigDefaults;
+
+impl TowerRequestConfigDefaults for AzureBlobTowerRequestConfigDefaults {
+    const RATE_LIMIT_NUM: u64 = 250;
+}
+
 /// Configuration for the `azure_blob` sink.
 #[configurable_component(sink(
     "azure_blob",
@@ -131,7 +139,7 @@ pub struct AzureBlobSinkConfig {
 
     #[configurable(derived)]
     #[serde(default)]
-    pub request: TowerRequestConfig,
+    pub request: TowerRequestConfig<AzureBlobTowerRequestConfigDefaults>,
 
     #[configurable(derived)]
     #[serde(
@@ -202,9 +210,7 @@ const DEFAULT_FILENAME_APPEND_UUID: bool = true;
 
 impl AzureBlobSinkConfig {
     pub fn build_processor(&self, client: Arc<ContainerClient>) -> crate::Result<VectorSink> {
-        let request_limits = self
-            .request
-            .unwrap_with(&TowerRequestConfig::default().rate_limit_num(250));
+        let request_limits = self.request.into_settings();
         let service = ServiceBuilder::new()
             .settings(request_limits, AzureBlobRetryLogic)
             .service(AzureBlobService::new(client));

src/sinks/azure_monitor_logs/config.rs

@@ -185,7 +185,7 @@ impl AzureMonitorLogsConfig {
 
         let retry_logic =
             HttpStatusRetryLogic::new(|res: &AzureMonitorLogsResponse| res.http_status);
-        let request_settings = self.request.unwrap_with(&Default::default());
+        let request_settings = self.request.into_settings();
         let service = ServiceBuilder::new()
             .settings(request_settings, retry_logic)
             .service(service);

src/sinks/clickhouse/config.rs

@@ -95,7 +95,7 @@ impl SinkConfig for ClickhouseConfig {
             self.date_time_best_effort,
         );
 
-        let request_limits = self.request.unwrap_with(&Default::default());
+        let request_limits = self.request.into_settings();
         let service = ServiceBuilder::new()
             .settings(request_limits, ClickhouseRetryLogic::default())
             .service(service);

src/sinks/clickhouse/integration_tests.rs

@@ -50,7 +50,7 @@ async fn insert_events() {
         compression: Compression::None,
         batch,
         request: TowerRequestConfig {
-            retry_attempts: Some(1),
+            retry_attempts: 1,
             ..Default::default()
         },
         ..Default::default()
@@ -100,7 +100,7 @@ async fn skip_unknown_fields() {
         compression: Compression::None,
         batch,
         request: TowerRequestConfig {
-            retry_attempts: Some(1),
+            retry_attempts: 1,
             ..Default::default()
         },
         ..Default::default()
@@ -146,7 +146,7 @@ async fn insert_events_unix_timestamps() {
         encoding: Transformer::new(None, None, Some(TimestampFormat::Unix)).unwrap(),
         batch,
         request: TowerRequestConfig {
-            retry_attempts: Some(1),
+            retry_attempts: 1,
             ..Default::default()
         },
         ..Default::default()

src/sinks/databend/config.rs

@@ -116,7 +116,7 @@ impl SinkConfig for DatabendConfig {
             DatabendAPIClient::new(self.build_client(&cx)?, endpoint.clone(), auth.clone());
         let healthcheck = select_one(health_client).boxed();
 
-        let request_settings = self.request.unwrap_with(&TowerRequestConfig::default());
+        let request_settings = self.request.into_settings();
         let batch_settings = self.batch.into_batcher_settings()?;
 
         let database = config.database;

src/sinks/datadog/events/config.rs

@@ -72,7 +72,7 @@ impl DatadogEventsConfig {
         );
 
         let request_opts = self.request;
-        let request_settings = request_opts.unwrap_with(&TowerRequestConfig::default());
+        let request_settings = request_opts.into_settings();
         let retry_logic = HttpStatusRetryLogic::new(|req: &DatadogEventsResponse| req.http_status);
 
         let service = ServiceBuilder::new()

src/sinks/datadog/logs/config.rs

@@ -104,7 +104,7 @@ impl DatadogLogsConfig {
         dd_evp_origin: String,
     ) -> crate::Result<VectorSink> {
         let default_api_key: Arc<str> = Arc::from(self.dd_common.default_api_key.inner());
-        let request_limits = self.request.tower.unwrap_with(&Default::default());
+        let request_limits = self.request.tower.into_settings();
 
         // We forcefully cap the provided batch configuration to the size/log line limits imposed by
         // the Datadog Logs API, but we still allow them to be lowered if need be.

src/sinks/datadog/metrics/config.rs

@@ -16,7 +16,10 @@ use crate::{
     http::HttpClient,
     sinks::{
         datadog::DatadogCommonConfig,
-        util::{batch::BatchConfig, ServiceBuilderExt, SinkBatchSettings, TowerRequestConfig},
+        util::{
+            batch::BatchConfig, service::TowerRequestConfigDefaults, ServiceBuilderExt,
+            SinkBatchSettings, TowerRequestConfig,
+        },
         Healthcheck, UriParseSnafu, VectorSink,
     },
     tls::{MaybeTlsSettings, TlsEnableableConfig},
@@ -31,9 +34,6 @@ use crate::{
 pub const MAXIMUM_PAYLOAD_COMPRESSED_SIZE: usize = 3_200_000;
 pub const MAXIMUM_PAYLOAD_SIZE: usize = 62_914_560;
 
-// TODO: revisit our concurrency and batching defaults
-const DEFAULT_REQUEST_RETRY_ATTEMPTS: usize = 5;
-
 #[derive(Clone, Copy, Debug, Default)]
 pub struct DatadogMetricsDefaultBatchSettings;
 
@@ -43,6 +43,13 @@ impl SinkBatchSettings for DatadogMetricsDefaultBatchSettings {
     const TIMEOUT_SECS: f64 = 2.0;
 }
 
+#[derive(Clone, Copy, Debug)]
+pub struct DatadogMetricsTowerRequestConfigDefaults;
+
+impl TowerRequestConfigDefaults for DatadogMetricsTowerRequestConfigDefaults {
+    const RETRY_ATTEMPTS: usize = 5;
+}
+
 pub(super) const SERIES_V1_PATH: &str = "/api/v1/series";
 pub(super) const SERIES_V2_PATH: &str = "/api/v2/series";
 pub(super) const SKETCHES_PATH: &str = "/api/beta/sketches";
@@ -149,7 +156,7 @@ pub struct DatadogMetricsConfig {
 
     #[configurable(derived)]
     #[serde(default)]
-    pub request: TowerRequestConfig,
+    pub request: TowerRequestConfig<DatadogMetricsTowerRequestConfigDefaults>,
 }
 
 impl_generate_config_from_default!(DatadogMetricsConfig);
@@ -232,9 +239,7 @@ impl DatadogMetricsConfig {
         let batcher_settings = self.batch.into_batcher_settings()?;
 
         // TODO: revisit our concurrency and batching defaults
-        let request_limits = self.request.unwrap_with(
-            &TowerRequestConfig::default().retry_attempts(DEFAULT_REQUEST_RETRY_ATTEMPTS),
-        );
+        let request_limits = self.request.into_settings();
 
         let endpoint_configuration = self.generate_metrics_endpoint_configuration()?;
         let service = ServiceBuilder::new()