vector/sinks/aws_cloudwatch_logs/
config.rs

1use std::collections::{BTreeMap, HashMap};
2
3use aws_sdk_cloudwatchlogs::Client as CloudwatchLogsClient;
4use futures::FutureExt;
5use http::HeaderValue;
6use serde::{Deserialize, Deserializer, de};
7use tower::ServiceBuilder;
8use vector_lib::{
9    codecs::JsonSerializerConfig, configurable::configurable_component, schema,
10    stream::BatcherSettings,
11};
12use vrl::value::Kind;
13
14use crate::{
15    aws::{AwsAuthentication, ClientBuilder, RegionOrEndpoint, create_client},
16    codecs::{Encoder, EncodingConfig},
17    config::{
18        AcknowledgementsConfig, DataType, DynValidatedSink, GenerateConfig, Input, ProxyConfig,
19        SinkConfig, SinkContext, ValidatedSink,
20    },
21    sinks::{
22        Healthcheck, VectorSink,
23        aws_cloudwatch_logs::{
24            healthcheck::healthcheck, request_builder::CloudwatchRequestBuilder,
25            retry::CloudwatchRetryLogic, service::CloudwatchLogsPartitionSvc, sink::CloudwatchSink,
26        },
27        util::{
28            BatchConfig, Compression, ServiceBuilderExt, SinkBatchSettings,
29            http::{OrderedHeaderName, RequestConfig, validate_headers},
30        },
31    },
32    template::{ConfinedTemplate, ConfinementConfig, Template, UnconfinedTemplate},
33    tls::TlsConfig,
34};
35
36pub struct CloudwatchLogsClientBuilder;
37
38impl ClientBuilder for CloudwatchLogsClientBuilder {
39    type Client = aws_sdk_cloudwatchlogs::client::Client;
40
41    fn build(&self, config: &aws_types::SdkConfig) -> Self::Client {
42        aws_sdk_cloudwatchlogs::client::Client::new(config)
43    }
44}
45
46#[configurable_component]
47#[derive(Clone, Debug, Default)]
48/// Retention policy configuration for AWS CloudWatch Log Group
49pub struct Retention {
50    /// Whether or not to set a retention policy when creating a new Log Group.
51    #[serde(default)]
52    pub enabled: bool,
53
54    /// If retention is enabled, the number of days to retain logs for.
55    #[serde(
56        default,
57        deserialize_with = "retention_days",
58        skip_serializing_if = "crate::serde::is_default"
59    )]
60    pub days: u32,
61}
62
63fn retention_days<'de, D>(deserializer: D) -> Result<u32, D::Error>
64where
65    D: Deserializer<'de>,
66{
67    let days: u32 = Deserialize::deserialize(deserializer)?;
68    const ALLOWED_VALUES: &[u32] = &[
69        1, 3, 5, 7, 14, 30, 60, 90, 120, 150, 180, 365, 400, 545, 731, 1096, 1827, 2192, 2557,
70        2922, 3288, 3653,
71    ];
72    if ALLOWED_VALUES.contains(&days) {
73        Ok(days)
74    } else {
75        let msg = format!("one of allowed values: {ALLOWED_VALUES:?}").to_owned();
76        let expected: &str = msg.as_str();
77        Err(de::Error::invalid_value(
78            de::Unexpected::Signed(days.into()),
79            &expected,
80        ))
81    }
82}
83
84/// Configuration for the `aws_cloudwatch_logs` sink.
85#[configurable_component(sink(
86    "aws_cloudwatch_logs",
87    "Publish log events to AWS CloudWatch Logs."
88))]
89#[derive(Clone, Debug)]
90#[serde(deny_unknown_fields)]
91pub struct CloudwatchLogsSinkConfig {
92    /// The [group name][group_name] of the target CloudWatch Logs stream.
93    ///
94    /// [group_name]: https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/Working-with-log-groups-and-streams.html
95    #[configurable(metadata(docs::examples = "group-name"))]
96    #[configurable(metadata(docs::examples = "group-{{ file }}"))]
97    pub group_name: Template,
98
99    /// The [stream name][stream_name] of the target CloudWatch Logs stream.
100    ///
101    /// There can only be one writer to a log stream at a time. If multiple instances are writing to
102    /// the same log group, the stream name must include an identifier that is guaranteed to be
103    /// unique per instance.
104    ///
105    /// [stream_name]: https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/Working-with-log-groups-and-streams.html
106    #[configurable(metadata(docs::examples = "stream-{{ host }}"))]
107    #[configurable(metadata(docs::examples = "%Y-%m-%d"))]
108    #[configurable(metadata(docs::examples = "stream-name"))]
109    pub stream_name: UnconfinedTemplate,
110
111    /// The [AWS region][aws_region] of the target service.
112    ///
113    /// [aws_region]: https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/Concepts.RegionsAndAvailabilityZones.html
114    #[serde(flatten)]
115    pub region: RegionOrEndpoint,
116
117    /// Dynamically create a [log group][log_group] if it does not already exist.
118    ///
119    /// This ignores `create_missing_stream` directly after creating the group and creates
120    /// the first stream.
121    ///
122    /// [log_group]: https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/Working-with-log-groups-and-streams.html
123    #[serde(default = "crate::serde::default_true")]
124    pub create_missing_group: bool,
125
126    /// Dynamically create a [log stream][log_stream] if it does not already exist.
127    ///
128    /// [log_stream]: https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/Working-with-log-groups-and-streams.html
129    #[serde(default = "crate::serde::default_true")]
130    pub create_missing_stream: bool,
131
132    #[configurable(derived)]
133    #[serde(default)]
134    pub retention: Retention,
135
136    #[configurable(derived)]
137    pub encoding: EncodingConfig,
138
139    #[configurable(derived)]
140    #[serde(default)]
141    pub compression: Compression,
142
143    #[configurable(derived)]
144    #[serde(default)]
145    pub batch: BatchConfig<CloudwatchLogsDefaultBatchSettings>,
146
147    #[configurable(derived)]
148    #[serde(default)]
149    pub request: RequestConfig,
150
151    #[configurable(derived)]
152    pub tls: Option<TlsConfig>,
153
154    /// The ARN of an [IAM role][iam_role] to assume at startup.
155    ///
156    /// [iam_role]: https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles.html
157    #[configurable(deprecated)]
158    #[configurable(metadata(docs::hidden))]
159    pub assume_role: Option<String>,
160
161    #[configurable(derived)]
162    #[serde(default)]
163    pub auth: AwsAuthentication,
164
165    #[configurable(derived)]
166    #[serde(
167        default,
168        deserialize_with = "crate::serde::bool_or_struct",
169        skip_serializing_if = "crate::serde::is_default"
170    )]
171    pub acknowledgements: AcknowledgementsConfig,
172
173    /// The [ARN][arn] (Amazon Resource Name) of the [KMS key][kms_key] to use when encrypting log data.
174    ///
175    /// [arn]: https://docs.aws.amazon.com/IAM/latest/UserGuide/reference-arns.html
176    /// [kms_key]: https://docs.aws.amazon.com/kms/latest/developerguide/overview.html
177    #[configurable(derived)]
178    #[serde(default)]
179    pub kms_key: Option<String>,
180
181    /// The Key-value pairs to be applied as [tags][tags] to the log group and stream.
182    ///
183    /// [tags]: https://docs.aws.amazon.com/whitepapers/latest/tagging-best-practices/what-are-tags.html
184    #[configurable(derived)]
185    #[serde(default)]
186    #[configurable(metadata(
187        docs::additional_props_description = "A tag represented as a key-value pair"
188    ))]
189    pub tags: Option<HashMap<String, String>>,
190
191    #[configurable(derived)]
192    #[serde(flatten)]
193    pub confinement: ConfinementConfig,
194}
195
196impl CloudwatchLogsSinkConfig {
197    pub async fn create_client(&self, proxy: &ProxyConfig) -> crate::Result<CloudwatchLogsClient> {
198        create_client::<CloudwatchLogsClientBuilder>(
199            &CloudwatchLogsClientBuilder {},
200            &self.auth,
201            self.region.region(),
202            self.region.endpoint(),
203            proxy,
204            self.tls.as_ref(),
205            None,
206        )
207        .await
208    }
209}
210
211#[async_trait::async_trait]
212#[typetag::serde(name = "aws_cloudwatch_logs")]
213impl SinkConfig for CloudwatchLogsSinkConfig {
214    fn confinement_config(&self) -> Option<&crate::template::ConfinementConfig> {
215        Some(&self.confinement)
216    }
217
218    fn input(&self) -> Input {
219        let requirement =
220            schema::Requirement::empty().optional_meaning("timestamp", Kind::timestamp());
221
222        Input::new(self.encoding.config().input_type() & DataType::Log)
223            .with_schema_requirement(requirement)
224    }
225
226    fn acknowledgements(&self) -> &AcknowledgementsConfig {
227        &self.acknowledgements
228    }
229
230    fn as_dyn_validated(&self) -> Option<&dyn DynValidatedSink> {
231        Some(self)
232    }
233}
234
235#[derive(Clone, Debug)]
236pub struct ValidatedCloudwatchLogs {
237    group_template: ConfinedTemplate,
238    batcher_settings: BatcherSettings,
239    headers: BTreeMap<OrderedHeaderName, HeaderValue>,
240}
241
242#[async_trait::async_trait]
243impl ValidatedSink for CloudwatchLogsSinkConfig {
244    type Validated = ValidatedCloudwatchLogs;
245
246    fn validate(&self) -> crate::Result<ValidatedCloudwatchLogs> {
247        let group_template =
248            self.group_name
249                .clone()
250                .confine(&self.confinement, Self::NAME, "group_name")?;
251        let batcher_settings = self.batch.into_batcher_settings()?;
252        let headers = validate_headers(&self.request.headers)?;
253
254        Ok(ValidatedCloudwatchLogs {
255            group_template,
256            batcher_settings,
257            headers,
258        })
259    }
260
261    async fn build(
262        &self,
263        validated: &ValidatedCloudwatchLogs,
264        cx: SinkContext,
265    ) -> crate::Result<(VectorSink, Healthcheck)> {
266        let ValidatedCloudwatchLogs {
267            group_template,
268            batcher_settings,
269            headers,
270        } = validated.clone();
271        let request_settings = self.request.tower.into_settings();
272        let client = self.create_client(cx.proxy()).await?;
273        let svc = ServiceBuilder::new()
274            .settings(request_settings, CloudwatchRetryLogic::new())
275            .service(CloudwatchLogsPartitionSvc::new(
276                self.clone(),
277                client.clone(),
278                headers.clone(),
279            ));
280        let transformer = self.encoding.transformer();
281        let serializer = self.encoding.build()?;
282        let encoder = Encoder::<()>::new(serializer);
283        let healthcheck = healthcheck(self.clone(), client).boxed();
284        let sink = CloudwatchSink {
285            batcher_settings,
286            request_builder: CloudwatchRequestBuilder {
287                group_template,
288                stream_template: self.stream_name.clone(),
289                transformer,
290                encoder,
291            },
292
293            service: svc,
294        };
295        Ok((VectorSink::from_event_streamsink(sink), healthcheck))
296    }
297}
298
299impl GenerateConfig for CloudwatchLogsSinkConfig {
300    fn generate_config() -> serde_json::Value {
301        serde_json::to_value(default_config(JsonSerializerConfig::default().into())).unwrap()
302    }
303}
304
305fn default_config(encoding: EncodingConfig) -> CloudwatchLogsSinkConfig {
306    CloudwatchLogsSinkConfig {
307        encoding,
308        group_name: Default::default(),
309        stream_name: Default::default(),
310        region: Default::default(),
311        create_missing_group: true,
312        create_missing_stream: true,
313        retention: Default::default(),
314        compression: Default::default(),
315        batch: Default::default(),
316        request: Default::default(),
317        tls: Default::default(),
318        assume_role: Default::default(),
319        auth: Default::default(),
320        acknowledgements: Default::default(),
321        kms_key: Default::default(),
322        tags: Default::default(),
323        confinement: Default::default(),
324    }
325}
326
327#[derive(Clone, Copy, Debug, Default)]
328pub struct CloudwatchLogsDefaultBatchSettings;
329
330impl SinkBatchSettings for CloudwatchLogsDefaultBatchSettings {
331    const MAX_EVENTS: Option<usize> = Some(10_000);
332    const MAX_BYTES: Option<usize> = Some(1_048_576);
333    const TIMEOUT_SECS: f64 = 1.0;
334}
335
336#[cfg(test)]
337mod tests {
338    use crate::config::ValidatedSink;
339    use crate::sinks::aws_cloudwatch_logs::config::CloudwatchLogsSinkConfig;
340    use crate::template::{ConfinementConfig, Template};
341    use vector_lib::codecs::JsonSerializerConfig;
342
343    #[test]
344    fn prepares_valid_config() {
345        let mut config = super::default_config(JsonSerializerConfig::default().into());
346        config.group_name = "group-{{ file }}".try_into().unwrap();
347        config.stream_name = "stream".try_into().unwrap();
348
349        let validated = config.validate().expect("preparation should succeed");
350        assert_eq!(validated.group_template.to_string(), "group-{{ file }}");
351        assert_eq!(validated.batcher_settings.item_limit, 10_000);
352    }
353
354    #[test]
355    fn test_generate_config() {
356        crate::test_util::test_generate_config::<CloudwatchLogsSinkConfig>();
357    }
358
359    #[test]
360    fn validate_rejects_invalid_header_name() {
361        let mut config = super::default_config(JsonSerializerConfig::default().into());
362        config.group_name = "group".try_into().unwrap();
363        config.stream_name = "stream".try_into().unwrap();
364        config
365            .request
366            .headers
367            .insert("invalid header name".to_string(), "value".to_string());
368
369        assert!(config.validate().is_err());
370    }
371
372    #[test]
373    fn validate_rejects_invalid_header_value() {
374        let mut config = super::default_config(JsonSerializerConfig::default().into());
375        config.group_name = "group".try_into().unwrap();
376        config.stream_name = "stream".try_into().unwrap();
377        config.request.headers.insert(
378            "valid-header".to_string(),
379            "value\nwith newline".to_string(),
380        );
381
382        assert!(config.validate().is_err());
383    }
384
385    #[test]
386    fn validate_retains_valid_headers() {
387        let mut config = super::default_config(JsonSerializerConfig::default().into());
388        config.group_name = "group".try_into().unwrap();
389        config.stream_name = "stream".try_into().unwrap();
390        config
391            .request
392            .headers
393            .insert("x-custom-header".to_string(), "custom-value".to_string());
394
395        let validated = config.validate().expect("preparation should succeed");
396        assert_eq!(validated.headers.len(), 1);
397        let (name, value) = validated.headers.iter().next().unwrap();
398        assert_eq!(name.inner(), "x-custom-header");
399        assert_eq!(value, "custom-value");
400    }
401
402    #[test]
403    fn confinement_rejects_unconfined_group_name() {
404        let template = Template::try_from("{{ group }}").unwrap();
405        let config = ConfinementConfig::default();
406        let result = template.confine(&config, "aws_cloudwatch_logs", "group_name");
407        assert!(result.is_err());
408    }
409
410    #[test]
411    fn confinement_opt_out_allows_unconfined_group_name() {
412        let template = Template::try_from("{{ group }}").unwrap();
413        let config = ConfinementConfig {
414            dangerously_allow_unconfined_template_resolution: true,
415        };
416        let result = template.confine(&config, "aws_cloudwatch_logs", "group_name");
417        assert!(result.is_ok());
418    }
419
420    #[test]
421    fn confinement_allows_prefixed_group_name() {
422        let template = Template::try_from("events-{{ env }}").unwrap();
423        let config = ConfinementConfig::default();
424        let result = template.confine(&config, "aws_cloudwatch_logs", "group_name");
425        assert!(result.is_ok());
426    }
427}
vector/sinks/aws_cloudwatch_logs/config.rs

vector/sinks/aws_cloudwatch_logs/
config.rs
