enhancement(aws sinks): use validated config pattern

Author: thomasqueirozb

changelog.d/aws_s3_ssekms_key_id_timezone.enhancement.md

@@ -0,0 +1,4 @@
+The `ssekms_key_id` option in the `aws_s3` sink now respects the configured timezone when the
+value is a template containing time components, matching the existing behavior of `key_prefix`.
+
+authors: thomasqueirozb

src/sinks/aws_cloudwatch_logs/config.rs

@@ -1,18 +1,22 @@
-use std::collections::HashMap;
+use std::collections::{BTreeMap, HashMap};
 
 use aws_sdk_cloudwatchlogs::Client as CloudwatchLogsClient;
 use futures::FutureExt;
+use http::HeaderValue;
 use serde::{Deserialize, Deserializer, de};
 use tower::ServiceBuilder;
-use vector_lib::{codecs::JsonSerializerConfig, configurable::configurable_component, schema};
+use vector_lib::{
+    codecs::JsonSerializerConfig, configurable::configurable_component, schema,
+    stream::BatcherSettings,
+};
 use vrl::value::Kind;
 
 use crate::{
     aws::{AwsAuthentication, ClientBuilder, RegionOrEndpoint, create_client},
     codecs::{Encoder, EncodingConfig},
     config::{
-        AcknowledgementsConfig, DataType, GenerateConfig, Input, ProxyConfig, SinkConfig,
-        SinkContext,
+        AcknowledgementsConfig, DataType, DynValidatedSink, GenerateConfig, Input, ProxyConfig,
+        SinkConfig, SinkContext, ValidatedSink,
     },
     sinks::{
         Healthcheck, VectorSink,
@@ -21,10 +25,11 @@ use crate::{
             retry::CloudwatchRetryLogic, service::CloudwatchLogsPartitionSvc, sink::CloudwatchSink,
         },
         util::{
-            BatchConfig, Compression, ServiceBuilderExt, SinkBatchSettings, http::RequestConfig,
+            BatchConfig, Compression, ServiceBuilderExt, SinkBatchSettings,
+            http::{OrderedHeaderName, RequestConfig, validate_headers},
         },
     },
-    template::{ConfinementConfig, Template},
+    template::{ConfinedTemplate, ConfinementConfig, Template},
     tls::TlsConfig,
 };
 
@@ -88,7 +93,7 @@ pub struct CloudwatchLogsSinkConfig {
     ///
     /// [group_name]: https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/Working-with-log-groups-and-streams.html
     #[configurable(metadata(docs::examples = "group-name"))]
-    #[configurable(metadata(docs::examples = "{{ file }}"))]
+    #[configurable(metadata(docs::examples = "group-{{ file }}"))]
     pub group_name: Template,
 
     /// The [stream name][stream_name] of the target CloudWatch Logs stream.
@@ -98,7 +103,7 @@ pub struct CloudwatchLogsSinkConfig {
     /// unique per instance.
     ///
     /// [stream_name]: https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/Working-with-log-groups-and-streams.html
-    #[configurable(metadata(docs::examples = "{{ host }}"))]
+    #[configurable(metadata(docs::examples = "stream-{{ host }}"))]
     #[configurable(metadata(docs::examples = "%Y-%m-%d"))]
     #[configurable(metadata(docs::examples = "stream-name"))]
     pub stream_name: Template,
@@ -206,7 +211,40 @@ impl CloudwatchLogsSinkConfig {
 #[async_trait::async_trait]
 #[typetag::serde(name = "aws_cloudwatch_logs")]
 impl SinkConfig for CloudwatchLogsSinkConfig {
-    async fn build(&self, cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> {
+    fn confinement_config(&self) -> Option<&crate::template::ConfinementConfig> {
+        Some(&self.confinement)
+    }
+
+    fn input(&self) -> Input {
+        let requirement =
+            schema::Requirement::empty().optional_meaning("timestamp", Kind::timestamp());
+
+        Input::new(self.encoding.config().input_type() & DataType::Log)
+            .with_schema_requirement(requirement)
+    }
+
+    fn acknowledgements(&self) -> &AcknowledgementsConfig {
+        &self.acknowledgements
+    }
+
+    fn as_dyn_validated(&self) -> Option<&dyn DynValidatedSink> {
+        Some(self)
+    }
+}
+
+#[derive(Clone, Debug)]
+pub struct ValidatedCloudwatchLogs {
+    group_template: ConfinedTemplate,
+    stream_template: ConfinedTemplate,
+    batcher_settings: BatcherSettings,
+    headers: BTreeMap<OrderedHeaderName, HeaderValue>,
+}
+
+#[async_trait::async_trait]
+impl ValidatedSink for CloudwatchLogsSinkConfig {
+    type Validated = ValidatedCloudwatchLogs;
+
+    fn validate(&self) -> crate::Result<ValidatedCloudwatchLogs> {
         let group_template =
             self.group_name
                 .clone()
@@ -215,16 +253,37 @@ impl SinkConfig for CloudwatchLogsSinkConfig {
             self.stream_name
                 .clone()
                 .confine(&self.confinement, Self::NAME, "stream_name")?;
-
         let batcher_settings = self.batch.into_batcher_settings()?;
+        let headers = validate_headers(&self.request.headers)?;
+
+        Ok(ValidatedCloudwatchLogs {
+            group_template,
+            stream_template,
+            batcher_settings,
+            headers,
+        })
+    }
+
+    async fn build(
+        &self,
+        validated: &ValidatedCloudwatchLogs,
+        cx: SinkContext,
+    ) -> crate::Result<(VectorSink, Healthcheck)> {
+        let ValidatedCloudwatchLogs {
+            group_template,
+            stream_template,
+            batcher_settings,
+            headers,
+        } = validated.clone();
         let request_settings = self.request.tower.into_settings();
         let client = self.create_client(cx.proxy()).await?;
         let svc = ServiceBuilder::new()
             .settings(request_settings, CloudwatchRetryLogic::new())
             .service(CloudwatchLogsPartitionSvc::new(
                 self.clone(),
                 client.clone(),
-            )?);
+                headers.clone(),
+            ));
         let transformer = self.encoding.transformer();
         let serializer = self.encoding.build()?;
         let encoder = Encoder::<()>::new(serializer);
@@ -242,22 +301,6 @@ impl SinkConfig for CloudwatchLogsSinkConfig {
         };
         Ok((VectorSink::from_event_streamsink(sink), healthcheck))
     }
-
-    fn confinement_config(&self) -> Option<&crate::template::ConfinementConfig> {
-        Some(&self.confinement)
-    }
-
-    fn input(&self) -> Input {
-        let requirement =
-            schema::Requirement::empty().optional_meaning("timestamp", Kind::timestamp());
-
-        Input::new(self.encoding.config().input_type() & DataType::Log)
-            .with_schema_requirement(requirement)
-    }
-
-    fn acknowledgements(&self) -> &AcknowledgementsConfig {
-        &self.acknowledgements
-    }
 }
 
 impl GenerateConfig for CloudwatchLogsSinkConfig {
@@ -299,14 +342,71 @@ impl SinkBatchSettings for CloudwatchLogsDefaultBatchSettings {
 
 #[cfg(test)]
 mod tests {
+    use crate::config::ValidatedSink;
     use crate::sinks::aws_cloudwatch_logs::config::CloudwatchLogsSinkConfig;
     use crate::template::{ConfinementConfig, Template};
+    use vector_lib::codecs::JsonSerializerConfig;
+
+    #[test]
+    fn prepares_valid_config() {
+        let mut config = super::default_config(JsonSerializerConfig::default().into());
+        config.group_name = "group-{{ file }}".try_into().unwrap();
+        config.stream_name = "stream".try_into().unwrap();
+
+        let validated = config.validate().expect("preparation should succeed");
+        assert_eq!(validated.group_template.to_string(), "group-{{ file }}");
+        assert_eq!(validated.stream_template.to_string(), "stream");
+        assert_eq!(validated.batcher_settings.item_limit, 10_000);
+    }
 
     #[test]
     fn test_generate_config() {
         crate::test_util::test_generate_config::<CloudwatchLogsSinkConfig>();
     }
 
+    #[test]
+    fn validate_rejects_invalid_header_name() {
+        let mut config = super::default_config(JsonSerializerConfig::default().into());
+        config.group_name = "group".try_into().unwrap();
+        config.stream_name = "stream".try_into().unwrap();
+        config
+            .request
+            .headers
+            .insert("invalid header name".to_string(), "value".to_string());
+
+        assert!(config.validate().is_err());
+    }
+
+    #[test]
+    fn validate_rejects_invalid_header_value() {
+        let mut config = super::default_config(JsonSerializerConfig::default().into());
+        config.group_name = "group".try_into().unwrap();
+        config.stream_name = "stream".try_into().unwrap();
+        config.request.headers.insert(
+            "valid-header".to_string(),
+            "value\nwith newline".to_string(),
+        );
+
+        assert!(config.validate().is_err());
+    }
+
+    #[test]
+    fn validate_retains_valid_headers() {
+        let mut config = super::default_config(JsonSerializerConfig::default().into());
+        config.group_name = "group".try_into().unwrap();
+        config.stream_name = "stream".try_into().unwrap();
+        config
+            .request
+            .headers
+            .insert("x-custom-header".to_string(), "custom-value".to_string());
+
+        let validated = config.validate().expect("preparation should succeed");
+        assert_eq!(validated.headers.len(), 1);
+        let (name, value) = validated.headers.iter().next().unwrap();
+        assert_eq!(name.inner(), "x-custom-header");
+        assert_eq!(value, "custom-value");
+    }
+
     #[test]
     fn confinement_rejects_unconfined_group_name() {
         let template = Template::try_from("{{ group }}").unwrap();

src/sinks/aws_cloudwatch_logs/request.rs

@@ -1,5 +1,5 @@
 use std::{
-    collections::HashMap,
+    collections::{BTreeMap, HashMap},
     future::Future,
     pin::Pin,
     task::{Context, Poll, ready},
@@ -18,11 +18,13 @@ use aws_sdk_cloudwatchlogs::{
 };
 use aws_smithy_runtime_api::client::{orchestrator::HttpResponse, result::SdkError};
 use futures::{FutureExt, future::BoxFuture};
-use http::{HeaderValue, header::HeaderName};
-use indexmap::IndexMap;
+use http::HeaderValue;
 use tokio::sync::oneshot;
 
-use crate::sinks::aws_cloudwatch_logs::{config::Retention, service::CloudwatchError};
+use crate::sinks::{
+    aws_cloudwatch_logs::{config::Retention, service::CloudwatchError},
+    util::http::OrderedHeaderName,
+};
 
 pub struct CloudwatchFuture {
     client: Client,
@@ -38,7 +40,7 @@ struct Client {
     client: CloudwatchLogsClient,
     stream_name: String,
     group_name: String,
-    headers: IndexMap<HeaderName, HeaderValue>,
+    headers: BTreeMap<OrderedHeaderName, HeaderValue>,
     retention_days: u32,
     kms_key: Option<String>,
     tags: Option<HashMap<String, String>>,
@@ -59,7 +61,7 @@ impl CloudwatchFuture {
     #[allow(clippy::too_many_arguments)]
     pub(super) fn new(
         client: CloudwatchLogsClient,
-        headers: IndexMap<HeaderName, HeaderValue>,
+        headers: BTreeMap<OrderedHeaderName, HeaderValue>,
         stream_name: String,
         group_name: String,
         create_missing_group: bool,
@@ -267,7 +269,8 @@ impl Client {
                 .customize()
                 .mutate_request(move |req| {
                     for (header, value) in headers.iter() {
-                        req.headers_mut().insert(header.clone(), value.clone());
+                        req.headers_mut()
+                            .insert(header.inner().clone(), value.clone());
                     }
                 })
                 .send()

src/sinks/aws_cloudwatch_logs/service.rs

@@ -1,5 +1,5 @@
 use std::{
-    collections::HashMap,
+    collections::{BTreeMap, HashMap},
     fmt,
     task::{Context, Poll, ready},
 };
@@ -17,12 +17,7 @@ use aws_smithy_runtime_api::client::{orchestrator::HttpResponse, result::SdkErro
 use chrono::Duration;
 use futures::{FutureExt, future::BoxFuture};
 use futures_util::TryFutureExt;
-use http::{
-    HeaderValue,
-    header::{HeaderName, InvalidHeaderName, InvalidHeaderValue},
-};
-use indexmap::IndexMap;
-use snafu::{ResultExt, Snafu};
+use http::HeaderValue;
 use tokio::sync::oneshot;
 use tower::{
     Service, ServiceBuilder, ServiceExt,
@@ -45,7 +40,9 @@ use crate::sinks::{
         retry::CloudwatchRetryLogic,
         sink::BatchCloudwatchRequest,
     },
-    util::{EncodedLength, TowerRequestSettings, retries::FibonacciRetryPolicy},
+    util::{
+        EncodedLength, TowerRequestSettings, http::OrderedHeaderName, retries::FibonacciRetryPolicy,
+    },
 };
 
 type Svc = Buffer<
@@ -133,40 +130,21 @@ impl DriverResponse for CloudwatchResponse {
     }
 }
 
-#[derive(Snafu, Debug)]
-enum HeaderError {
-    #[snafu(display("invalid header name {source}"))]
-    InvalidName { source: InvalidHeaderName },
-    #[snafu(display("invalid header value {source}"))]
-    InvalidValue { source: InvalidHeaderValue },
-}
-
 impl CloudwatchLogsPartitionSvc {
     pub fn new(
         config: CloudwatchLogsSinkConfig,
         client: CloudwatchLogsClient,
-    ) -> crate::Result<Self> {
+        headers: BTreeMap<OrderedHeaderName, HeaderValue>,
+    ) -> Self {
         let request_settings = config.request.tower.into_settings();
 
-        let headers = config
-            .request
-            .headers
-            .iter()
-            .map(|(name, value)| {
-                Ok((
-                    HeaderName::from_bytes(name.as_bytes()).context(InvalidNameSnafu {})?,
-                    HeaderValue::from_str(value.as_str()).context(InvalidValueSnafu {})?,
-                ))
-            })
-            .collect::<Result<IndexMap<_, _>, HeaderError>>()?;
-
-        Ok(Self {
+        Self {
             config,
             clients: HashMap::new(),
             request_settings,
             client,
             headers,
-        })
+        }
     }
 }
 
@@ -236,7 +214,7 @@ impl CloudwatchLogsSvc {
         config: CloudwatchLogsSinkConfig,
         key: &CloudwatchKey,
         client: CloudwatchLogsClient,
-        headers: IndexMap<HeaderName, HeaderValue>,
+        headers: BTreeMap<OrderedHeaderName, HeaderValue>,
     ) -> Self {
         let group_name = key.group.clone();
         let stream_name = key.stream.clone();
@@ -347,7 +325,7 @@ impl Service<Vec<InputLogEvent>> for CloudwatchLogsSvc {
 
 pub struct CloudwatchLogsSvc {
     client: CloudwatchLogsClient,
-    headers: IndexMap<HeaderName, HeaderValue>,
+    headers: BTreeMap<OrderedHeaderName, HeaderValue>,
     stream_name: String,
     group_name: String,
     create_missing_group: bool,
@@ -368,7 +346,7 @@ impl EncodedLength for InputLogEvent {
 #[derive(Clone)]
 pub struct CloudwatchLogsPartitionSvc {
     config: CloudwatchLogsSinkConfig,
-    headers: IndexMap<HeaderName, HeaderValue>,
+    headers: BTreeMap<OrderedHeaderName, HeaderValue>,
     clients: HashMap<CloudwatchKey, Svc>,
     request_settings: TowerRequestSettings,
     client: CloudwatchLogsClient,