feat(sinks): introduce RetryStrategy / Config for http based sinks

benches/http.rs

@@ -63,6 +63,7 @@ fn benchmark_http(c: &mut Criterion) {
                                 request: Default::default(),
                                 tls: Default::default(),
                                 acknowledgements: Default::default(),
+                                retry_strategy: Default::default(),
                             },
                         );
 

changelog.d/10870_http_retry_strategy.feature.md

@@ -0,0 +1,7 @@
+HTTP-based sinks that use the shared retry helpers now support a `retry_strategy` configuration
+option to control which HTTP response codes are retried. The `http` sink also includes a new
+example showing how to retry only specific transient status codes.
+
+Issue: https://github.com/vectordotdev/vector/issues/10870
+
+authors: ndrsg

config/examples/http_sink_custom_retry.yaml

@@ -0,0 +1,42 @@
+# HTTP sink example with a custom retry strategy
+# ----------------------------------------------------
+# Sends demo logs to an HTTP endpoint and only retries the response codes
+# that the upstream API documents as transient.
+
+data_dir: "/var/lib/vector"
+
+sources:
+  demo_logs:
+    type: "demo_logs"
+    format: "json"
+    interval: 1
+
+sinks:
+  http_out:
+    type: "http"
+    inputs: [ "demo_logs" ]
+    uri: "https://example.com/ingest"
+    method: "post"
+
+    # Skip the startup probe so the example can be adapted locally.
+    healthcheck:
+      enabled: false
+
+    # Send newline-delimited JSON in the request body.
+    framing:
+      method: "newline_delimited"
+    encoding:
+      codec: "json"
+
+    # Control how many retries are made and how quickly backoff grows.
+    request:
+      timeout_secs: 60
+      retry_attempts: 8
+      retry_initial_backoff_secs: 2
+      retry_max_duration_secs: 30
+
+    # Retry only on the exact HTTP status codes that this destination
+    # treats as temporary failures.
+    retry_strategy:
+      type: "custom"
+      status_codes: [ 408, 425, 429, 503 ]

src/sinks/appsignal/config.rs

@@ -21,7 +21,7 @@ use crate::{
         prelude::{SinkConfig, SinkContext},
         util::{
             BatchConfig, Compression, ServiceBuilderExt, SinkBatchSettings, TowerRequestConfig,
-            http::HttpStatusRetryLogic,
+            http::{HttpStatusRetryLogic, RetryStrategy},
         },
     },
 };
@@ -67,6 +67,10 @@ pub(super) struct AppsignalConfig {
         skip_serializing_if = "crate::serde::is_default"
     )]
     acknowledgements: AcknowledgementsConfig,
+
+    #[configurable(derived)]
+    #[serde(default)]
+    retry_strategy: RetryStrategy,
 }
 
 pub(super) fn default_endpoint() -> String {
@@ -99,7 +103,10 @@ impl AppsignalConfig {
 
         let request_opts = self.request;
         let request_settings = request_opts.into_settings();
-        let retry_logic = HttpStatusRetryLogic::new(|req: &AppsignalResponse| req.http_status);
+        let retry_logic = HttpStatusRetryLogic::new(
+            |req: &AppsignalResponse| req.http_status,
+            self.retry_strategy.clone(),
+        );
 
         let service = ServiceBuilder::new()
             .settings(request_settings, retry_logic)

src/sinks/axiom/config.rs

@@ -15,7 +15,8 @@ use crate::{
         Healthcheck, VectorSink,
         http::config::{HttpMethod, HttpSinkConfig},
         util::{
-            BatchConfig, Compression, RealtimeSizeBasedDefaultBatchSettings, http::RequestConfig,
+            BatchConfig, Compression, RealtimeSizeBasedDefaultBatchSettings,
+            http::{RequestConfig, RetryStrategy},
         },
     },
     tls::TlsConfig,
@@ -124,6 +125,10 @@ pub struct AxiomConfig {
         skip_serializing_if = "crate::serde::is_default"
     )]
     pub acknowledgements: AcknowledgementsConfig,
+
+    #[configurable(derived)]
+    #[serde(default)]
+    pub retry_strategy: RetryStrategy,
 }
 
 impl GenerateConfig for AxiomConfig {
@@ -180,6 +185,7 @@ impl SinkConfig for AxiomConfig {
             ),
             payload_prefix: "".into(), // Always newline delimited JSON
             payload_suffix: "".into(), // Always newline delimited JSON
+            retry_strategy: self.retry_strategy.clone(),
         };
 
         http_sink_config.build(cx).await

src/sinks/azure_logs_ingestion/config.rs

@@ -16,7 +16,10 @@ use crate::{
     http::{HttpClient, get_http_scheme_from_uri},
     sinks::{
         prelude::*,
-        util::{RealtimeSizeBasedDefaultBatchSettings, UriSerde, http::HttpStatusRetryLogic},
+        util::{
+            RealtimeSizeBasedDefaultBatchSettings, UriSerde,
+            http::{HttpStatusRetryLogic, RetryStrategy},
+        },
     },
 };
 
@@ -109,6 +112,10 @@ pub struct AzureLogsIngestionConfig {
         skip_serializing_if = "crate::serde::is_default"
     )]
     pub acknowledgements: AcknowledgementsConfig,
+
+    #[configurable(derived)]
+    #[serde(default)]
+    pub retry_strategy: RetryStrategy,
 }
 
 impl Default for AzureLogsIngestionConfig {
@@ -125,6 +132,7 @@ impl Default for AzureLogsIngestionConfig {
             request: Default::default(),
             tls: None,
             acknowledgements: Default::default(),
+            retry_strategy: Default::default(),
         }
     }
 }
@@ -348,8 +356,10 @@ impl AzureLogsIngestionConfig {
         )?;
         let healthcheck = service.healthcheck();
 
-        let retry_logic =
-            HttpStatusRetryLogic::new(|res: &AzureLogsIngestionResponse| res.http_status);
+        let retry_logic = HttpStatusRetryLogic::new(
+            |res: &AzureLogsIngestionResponse| res.http_status,
+            self.retry_strategy.clone(),
+        );
         let request_settings = self.request.into_settings();
         let service = ServiceBuilder::new()
             .settings(request_settings, retry_logic)

src/sinks/azure_monitor_logs/config.rs

@@ -16,7 +16,10 @@ use crate::{
     http::{HttpClient, get_http_scheme_from_uri},
     sinks::{
         prelude::*,
-        util::{RealtimeSizeBasedDefaultBatchSettings, UriSerde, http::HttpStatusRetryLogic},
+        util::{
+            RealtimeSizeBasedDefaultBatchSettings, UriSerde,
+            http::{HttpStatusRetryLogic, RetryStrategy},
+        },
     },
 };
 
@@ -113,6 +116,10 @@ pub struct AzureMonitorLogsConfig {
         skip_serializing_if = "crate::serde::is_default"
     )]
     pub acknowledgements: AcknowledgementsConfig,
+
+    #[configurable(derived)]
+    #[serde(default)]
+    pub retry_strategy: RetryStrategy,
 }
 
 impl Default for AzureMonitorLogsConfig {
@@ -129,6 +136,7 @@ impl Default for AzureMonitorLogsConfig {
             time_generated_key: None,
             tls: None,
             acknowledgements: Default::default(),
+            retry_strategy: Default::default(),
         }
     }
 }
@@ -181,8 +189,10 @@ impl AzureMonitorLogsConfig {
         )?;
         let healthcheck = service.healthcheck();
 
-        let retry_logic =
-            HttpStatusRetryLogic::new(|res: &AzureMonitorLogsResponse| res.http_status);
+        let retry_logic = HttpStatusRetryLogic::new(
+            |res: &AzureMonitorLogsResponse| res.http_status,
+            self.retry_strategy.clone(),
+        );
         let request_settings = self.request.into_settings();
         let service = ServiceBuilder::new()
             .settings(request_settings, retry_logic)

src/sinks/datadog/events/config.rs

@@ -14,7 +14,10 @@ use crate::{
     sinks::{
         Healthcheck, VectorSink,
         datadog::{DatadogCommonConfig, LocalDatadogCommonConfig},
-        util::{ServiceBuilderExt, TowerRequestConfig, http::HttpStatusRetryLogic},
+        util::{
+            ServiceBuilderExt, TowerRequestConfig,
+            http::{HttpStatusRetryLogic, RetryStrategy},
+        },
     },
     tls::MaybeTlsSettings,
 };
@@ -33,6 +36,10 @@ pub struct DatadogEventsConfig {
     #[configurable(derived)]
     #[serde(default)]
     pub request: TowerRequestConfig,
+
+    #[configurable(derived)]
+    #[serde(default)]
+    pub retry_strategy: RetryStrategy,
 }
 
 impl GenerateConfig for DatadogEventsConfig {
@@ -64,7 +71,10 @@ impl DatadogEventsConfig {
 
         let request_opts = self.request;
         let request_settings = request_opts.into_settings();
-        let retry_logic = HttpStatusRetryLogic::new(|req: &DatadogEventsResponse| req.http_status);
+        let retry_logic = HttpStatusRetryLogic::new(
+            |req: &DatadogEventsResponse| req.http_status,
+            self.retry_strategy.clone(),
+        );
 
         let service = ServiceBuilder::new()
             .settings(request_settings, retry_logic)

src/sinks/gcp/stackdriver/logs/config.rs

@@ -21,7 +21,7 @@ use crate::{
         prelude::*,
         util::{
             BoxedRawValue, RealtimeSizeBasedDefaultBatchSettings,
-            http::{HttpService, http_response_retry_logic},
+            http::{HttpService, RetryStrategy, http_response_retry_logic},
             service::TowerRequestConfigDefaults,
         },
     },
@@ -106,6 +106,10 @@ pub(super) struct StackdriverConfig {
         skip_serializing_if = "crate::serde::is_default"
     )]
     acknowledgements: AcknowledgementsConfig,
+
+    #[configurable(derived)]
+    #[serde(default)]
+    pub retry_strategy: RetryStrategy,
 }
 
 pub(super) fn default_endpoint() -> String {
@@ -269,7 +273,10 @@ impl SinkConfig for StackdriverConfig {
         let service = HttpService::new(client.clone(), stackdriver_logs_service_request_builder);
 
         let service = ServiceBuilder::new()
-            .settings(request_limits, http_response_retry_logic())
+            .settings(
+                request_limits,
+                http_response_retry_logic(self.retry_strategy.clone()),
+            )
             .service(service);
 
         let sink = StackdriverLogsSink::new(service, batch_settings, request_builder);

src/sinks/gcp/stackdriver/metrics/config.rs

@@ -15,7 +15,8 @@ use crate::{
         prelude::*,
         util::{
             http::{
-                HttpRequest, HttpService, HttpServiceRequestBuilder, http_response_retry_logic,
+                HttpRequest, HttpService, HttpServiceRequestBuilder, RetryStrategy,
+                http_response_retry_logic,
             },
             service::TowerRequestConfigDefaults,
         },
@@ -77,6 +78,10 @@ pub struct StackdriverConfig {
         skip_serializing_if = "crate::serde::is_default"
     )]
     pub(super) acknowledgements: AcknowledgementsConfig,
+
+    #[configurable(derived)]
+    #[serde(default)]
+    pub retry_strategy: RetryStrategy,
 }
 
 fn default_metric_namespace_value() -> String {
@@ -126,7 +131,10 @@ impl SinkConfig for StackdriverConfig {
         let service = HttpService::new(client, stackdriver_metrics_service_request_builder);
 
         let service = ServiceBuilder::new()
-            .settings(request_limits, http_response_retry_logic())
+            .settings(
+                request_limits,
+                http_response_retry_logic(self.retry_strategy.clone()),
+            )
             .service(service);
 
         let sink = StackdriverMetricsSink::new(service, batch_settings, request_builder);

src/sinks/honeycomb/config.rs

@@ -16,7 +16,7 @@ use crate::{
         prelude::*,
         util::{
             BatchConfig, BoxedRawValue,
-            http::{HttpService, http_response_retry_logic},
+            http::{HttpService, RetryStrategy, http_response_retry_logic},
         },
     },
 };
@@ -71,6 +71,10 @@ pub struct HoneycombConfig {
         skip_serializing_if = "crate::serde::is_default"
     )]
     acknowledgements: AcknowledgementsConfig,
+
+    #[configurable(derived)]
+    #[serde(default)]
+    pub retry_strategy: RetryStrategy,
 }
 
 fn default_endpoint() -> String {
@@ -124,7 +128,10 @@ impl SinkConfig for HoneycombConfig {
         let request_limits = self.request.into_settings();
 
         let service = ServiceBuilder::new()
-            .settings(request_limits, http_response_retry_logic())
+            .settings(
+                request_limits,
+                http_response_retry_logic(self.retry_strategy.clone()),
+            )
             .service(service);
 
         let sink = HoneycombSink::new(service, batch_settings, request_builder);

src/sinks/http/config.rs

@@ -30,7 +30,10 @@ use crate::{
         prelude::*,
         util::{
             RealtimeSizeBasedDefaultBatchSettings, UriSerde,
-            http::{HttpService, OrderedHeaderName, RequestConfig, http_response_retry_logic},
+            http::{
+                HttpService, OrderedHeaderName, RequestConfig, RetryStrategy,
+                http_response_retry_logic,
+            },
         },
     },
 };
@@ -100,6 +103,10 @@ pub struct HttpSinkConfig {
         skip_serializing_if = "crate::serde::is_default"
     )]
     pub acknowledgements: AcknowledgementsConfig,
+
+    #[configurable(derived)]
+    #[serde(default)]
+    pub retry_strategy: RetryStrategy,
 }
 
 /// HTTP method.
@@ -330,7 +337,10 @@ impl SinkConfig for HttpSinkConfig {
         let request_limits = self.request.tower.into_settings();
 
         let service = ServiceBuilder::new()
-            .settings(request_limits, http_response_retry_logic())
+            .settings(
+                request_limits,
+                http_response_retry_logic(self.retry_strategy.clone()),
+            )
             .service(service);
 
         let sink = HttpSink::new(
@@ -405,6 +415,7 @@ mod tests {
                 acknowledgements: AcknowledgementsConfig::default(),
                 payload_prefix: String::new(),
                 payload_suffix: String::new(),
+                retry_strategy: RetryStrategy::default(),
             };
 
             let external_resource = ExternalResource::new(