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,9 +34,7 @@ use crate::sinks::{
         config::CloudwatchLogsSinkConfig, config::Retention, request, retry::CloudwatchRetryLogic,
         sink::BatchCloudwatchRequest, CloudwatchKey,
     },
-    util::{
-        retries::FibonacciRetryPolicy, EncodedLength, TowerRequestConfig, TowerRequestSettings,
-    },
+    util::{retries::FibonacciRetryPolicy, EncodedLength, TowerRequestSettings},
 };
 
 type Svc = Buffer<
@@ -138,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,13 @@ impl SinkBatchSettings for CloudWatchMetricsDefaultBatchSettings {
     const TIMEOUT_SECS: f64 = 1.0;
 }
 
+#[derive(Clone, Copy, Debug)]
+pub struct CloudWatchMetricsTowerRequestConfigDefaults;
+
+impl TowerRequestConfigDefaults for CloudWatchMetricsTowerRequestConfigDefaults {
+    const RATE_LIMIT_NUM: u64 = 150;
+}
+
 /// Configuration for the `aws_cloudwatch_metrics` sink.
 #[configurable_component(sink(
     "aws_cloudwatch_metrics",
@@ -79,7 +88,7 @@ pub struct CloudWatchMetricsSinkConfig {
 
     #[configurable(derived)]
     #[serde(default)]
-    pub request: TowerRequestConfig,
+    pub request: TowerRequestConfig<CloudWatchMetricsTowerRequestConfigDefaults>,
 
     #[configurable(derived)]
     pub tls: Option<TlsConfig>,
@@ -225,11 +234,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

@@ -201,7 +201,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/sink.rs

@@ -40,9 +40,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

@@ -31,9 +31,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;
 
@@ -232,9 +229,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()

src/sinks/datadog/traces/config.rs

@@ -41,9 +41,6 @@ pub const BATCH_DEFAULT_TIMEOUT_SECS: f64 = 10.0;
 
 pub const PAYLOAD_LIMIT: usize = 3_200_000;
 
-const DEFAULT_REQUEST_RETRY_ATTEMPTS: usize = 5;
-const DEFAULT_REQUEST_RETRY_MAX_DURATION_SECS: u64 = 300;
-
 #[derive(Clone, Copy, Debug, Default)]
 pub struct DatadogTracesDefaultBatchSettings;
 
@@ -130,11 +127,7 @@ impl DatadogTracesConfig {
 
     pub fn build_sink(&self, client: HttpClient) -> crate::Result<VectorSink> {
         let default_api_key: Arc<str> = Arc::from(self.dd_common.default_api_key.inner());
-        let request_limits = self.request.unwrap_with(
-            &TowerRequestConfig::default()
-                .retry_attempts(DEFAULT_REQUEST_RETRY_ATTEMPTS)
-                .retry_max_duration_secs(DEFAULT_REQUEST_RETRY_MAX_DURATION_SECS),
-        );
+        let request_limits = self.request.into_settings();
         let endpoints = self.generate_traces_endpoint_configuration()?;
 
         let batcher_settings = self

src/sinks/elasticsearch/common.rs

@@ -19,7 +19,7 @@ use crate::{
             ElasticsearchAuthConfig, ElasticsearchCommonMode, ElasticsearchConfig, ParseError,
         },
         util::auth::Auth,
-        util::{http::RequestConfig, TowerRequestConfig, UriSerde},
+        util::{http::RequestConfig, UriSerde},
         HealthcheckError,
     },
     tls::TlsSettings,
@@ -89,10 +89,7 @@ impl ElasticsearchCommon {
 
         let mode = config.common_mode()?;
 
-        let tower_request = config
-            .request
-            .tower
-            .unwrap_with(&TowerRequestConfig::default());
+        let tower_request = config.request.tower.into_settings();
 
         let mut query_params = config.query.clone().unwrap_or_default();
         query_params.insert(

src/sinks/elasticsearch/config.rs

@@ -23,7 +23,7 @@ use crate::{
         },
         util::{
             http::RequestConfig, service::HealthConfig, BatchConfig, Compression,
-            RealtimeSizeBasedDefaultBatchSettings, TowerRequestConfig,
+            RealtimeSizeBasedDefaultBatchSettings,
         },
         Healthcheck, VectorSink,
     },
@@ -482,10 +482,7 @@ impl SinkConfig for ElasticsearchConfig {
 
         let client = HttpClient::new(common.tls_settings.clone(), cx.proxy())?;
 
-        let request_limits = self
-            .request
-            .tower
-            .unwrap_with(&TowerRequestConfig::default());
+        let request_limits = self.request.tower.into_settings();
 
         let health_config = self.endpoint_health.clone().unwrap_or_default();
 

src/sinks/gcp/chronicle_unstructured.rs

@@ -22,6 +22,7 @@ use vector_lib::{
 };
 use vrl::value::Kind;
 
+use crate::sinks::util::service::TowerRequestConfigDefaults;
 use crate::{
     codecs::{self, EncodingConfig},
     config::{GenerateConfig, SinkConfig, SinkContext},
@@ -97,6 +98,12 @@ impl SinkBatchSettings for ChronicleUnstructuredDefaultBatchSettings {
     const TIMEOUT_SECS: f64 = 15.0;
 }
 
+#[derive(Clone, Copy, Debug)]
+pub struct ChronicleUnstructuredTowerRequestConfigDefaults;
+
+impl TowerRequestConfigDefaults for ChronicleUnstructuredTowerRequestConfigDefaults {
+    const RATE_LIMIT_NUM: u64 = 1_000;
+}
 /// Configuration for the `gcp_chronicle_unstructured` sink.
 #[configurable_component(sink(
     "gcp_chronicle_unstructured",
@@ -132,7 +139,7 @@ pub struct ChronicleUnstructuredConfig {
 
     #[configurable(derived)]
     #[serde(default)]
-    pub request: TowerRequestConfig,
+    pub request: TowerRequestConfig<ChronicleUnstructuredTowerRequestConfigDefaults>,
 
     #[configurable(derived)]
     pub tls: Option<TlsConfig>,
@@ -235,10 +242,7 @@ impl ChronicleUnstructuredConfig {
     ) -> crate::Result<VectorSink> {
         use crate::sinks::util::service::ServiceBuilderExt;
 
-        let request = self.request.unwrap_with(&TowerRequestConfig {
-            rate_limit_num: Some(1000),
-            ..Default::default()
-        });
+        let request = self.request.into_settings();
 
         let batch_settings = self.batch.into_batcher_settings()?;