diff --git a/changelog.d/22008_cloudwatch_logs_group_class.enhancement.md b/changelog.d/22008_cloudwatch_logs_group_class.enhancement.md new file mode 100644 index 0000000000000..677e7d5cf8d6a --- /dev/null +++ b/changelog.d/22008_cloudwatch_logs_group_class.enhancement.md @@ -0,0 +1,4 @@ +The CloudWatch Logs sink now accepts a `group_class` option that selects which +log class to create (`STANDARD` or `INFREQUENT_ACCESS`). Defaults to `STANDARD`. + +authors: vivshaw \ No newline at end of file diff --git a/src/sinks/aws_cloudwatch_logs/config.rs b/src/sinks/aws_cloudwatch_logs/config.rs index 6ec24d7e093e2..ff07a1531de2c 100644 --- a/src/sinks/aws_cloudwatch_logs/config.rs +++ b/src/sinks/aws_cloudwatch_logs/config.rs @@ -1,6 +1,6 @@ use std::collections::HashMap; -use aws_sdk_cloudwatchlogs::Client as CloudwatchLogsClient; +use aws_sdk_cloudwatchlogs::{Client as CloudwatchLogsClient, types::LogGroupClass}; use futures::FutureExt; use serde::{Deserialize, Deserializer, de}; use tower::ServiceBuilder; @@ -55,6 +55,30 @@ pub struct Retention { pub days: u32, } +/// CloudWatch Logs [log class][log_class] used when creating a new log group. +/// +/// [log_class]: https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/CloudWatch_Logs_Log_Classes.html +#[configurable_component] +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +#[serde(rename_all = "SCREAMING_SNAKE_CASE")] +pub enum CloudwatchLogsGroupClass { + /// For real-time monitoring and frequently-accessed logs. + #[default] + Standard, + + /// For cost-effective consolidation of logs that are queried only occasionally. + InfrequentAccess, +} + +impl From for LogGroupClass { + fn from(value: CloudwatchLogsGroupClass) -> Self { + match value { + CloudwatchLogsGroupClass::Standard => LogGroupClass::Standard, + CloudwatchLogsGroupClass::InfrequentAccess => LogGroupClass::InfrequentAccess, + } + } +} + fn retention_days<'de, D>(deserializer: D) -> Result where D: Deserializer<'de>, @@ -109,7 +133,8 @@ pub struct CloudwatchLogsSinkConfig { #[serde(flatten)] pub region: RegionOrEndpoint, - /// Dynamically create a [log group][log_group] if it does not already exist. + /// Dynamically create a [log group][log_group] if it does not already exist. Its group + /// class is determined by `group_class`. /// /// This ignores `create_missing_stream` directly after creating the group and creates /// the first stream. @@ -118,6 +143,14 @@ pub struct CloudwatchLogsSinkConfig { #[serde(default = "crate::serde::default_true")] pub create_missing_group: bool, + /// The [log class][log_class] used when dynamically creating a log group via + /// `create_missing_group`. + /// + /// [log_class]: https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/CloudWatch_Logs_Log_Classes.html + #[serde(default)] + #[configurable(derived)] + pub group_class: CloudwatchLogsGroupClass, + /// Dynamically create a [log stream][log_stream] if it does not already exist. /// /// [log_stream]: https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/Working-with-log-groups-and-streams.html @@ -257,6 +290,7 @@ fn default_config(encoding: EncodingConfig) -> CloudwatchLogsSinkConfig { stream_name: Default::default(), region: Default::default(), create_missing_group: true, + group_class: Default::default(), create_missing_stream: true, retention: Default::default(), compression: Default::default(), diff --git a/src/sinks/aws_cloudwatch_logs/integration_tests.rs b/src/sinks/aws_cloudwatch_logs/integration_tests.rs index f25a849552348..eb0dc64479103 100644 --- a/src/sinks/aws_cloudwatch_logs/integration_tests.rs +++ b/src/sinks/aws_cloudwatch_logs/integration_tests.rs @@ -54,6 +54,7 @@ async fn cloudwatch_insert_log_event() { region: RegionOrEndpoint::with_both("us-east-1", cloudwatch_address().as_str()), encoding: TextSerializerConfig::default().into(), create_missing_group: true, + group_class: Default::default(), create_missing_stream: true, retention: Default::default(), compression: Default::default(), @@ -107,6 +108,7 @@ async fn cloudwatch_insert_log_events_sorted() { region: RegionOrEndpoint::with_both("us-east-1", cloudwatch_address().as_str()), encoding: TextSerializerConfig::default().into(), create_missing_group: true, + group_class: Default::default(), create_missing_stream: true, retention: Default::default(), compression: Default::default(), @@ -185,6 +187,7 @@ async fn cloudwatch_insert_out_of_range_timestamp() { region: RegionOrEndpoint::with_both("us-east-1", cloudwatch_address().as_str()), encoding: TextSerializerConfig::default().into(), create_missing_group: true, + group_class: Default::default(), create_missing_stream: true, retention: Default::default(), compression: Default::default(), @@ -264,6 +267,7 @@ async fn cloudwatch_dynamic_group_and_stream_creation() { region: RegionOrEndpoint::with_both("us-east-1", cloudwatch_address().as_str()), encoding: TextSerializerConfig::default().into(), create_missing_group: true, + group_class: Default::default(), create_missing_stream: true, retention: Default::default(), compression: Default::default(), @@ -317,6 +321,7 @@ async fn cloudwatch_dynamic_group_and_stream_creation_with_kms_key_and_tags() { region: RegionOrEndpoint::with_both("us-east-1", cloudwatch_address().as_str()), encoding: TextSerializerConfig::default().into(), create_missing_group: true, + group_class: Default::default(), create_missing_stream: true, retention: Default::default(), compression: Default::default(), @@ -406,6 +411,7 @@ async fn cloudwatch_insert_log_event_batched() { region: RegionOrEndpoint::with_both("us-east-1", cloudwatch_address().as_str()), encoding: TextSerializerConfig::default().into(), create_missing_group: true, + group_class: Default::default(), create_missing_stream: true, retention: Default::default(), compression: Default::default(), @@ -459,6 +465,7 @@ async fn cloudwatch_insert_log_event_partitioned() { region: RegionOrEndpoint::with_both("us-east-1", cloudwatch_address().as_str()), encoding: TextSerializerConfig::default().into(), create_missing_group: true, + group_class: Default::default(), create_missing_stream: true, retention: Default::default(), compression: Default::default(), @@ -554,6 +561,7 @@ async fn cloudwatch_healthcheck() { region: RegionOrEndpoint::with_both("us-east-1", cloudwatch_address().as_str()), encoding: TextSerializerConfig::default().into(), create_missing_group: true, + group_class: Default::default(), create_missing_stream: true, retention: Default::default(), compression: Default::default(), diff --git a/src/sinks/aws_cloudwatch_logs/request.rs b/src/sinks/aws_cloudwatch_logs/request.rs index de3b03da25396..3440d65b9a7a6 100644 --- a/src/sinks/aws_cloudwatch_logs/request.rs +++ b/src/sinks/aws_cloudwatch_logs/request.rs @@ -22,7 +22,10 @@ use http::{HeaderValue, header::HeaderName}; use indexmap::IndexMap; use tokio::sync::oneshot; -use crate::sinks::aws_cloudwatch_logs::{config::Retention, service::CloudwatchError}; +use crate::sinks::aws_cloudwatch_logs::{ + config::{CloudwatchLogsGroupClass, Retention}, + service::CloudwatchError, +}; pub struct CloudwatchFuture { client: Client, @@ -42,6 +45,7 @@ struct Client { retention_days: u32, kms_key: Option, tags: Option>, + group_class: CloudwatchLogsGroupClass, } type ClientResult = BoxFuture<'static, Result>>; @@ -67,6 +71,7 @@ impl CloudwatchFuture { retention: Retention, kms_key: Option, tags: Option>, + group_class: CloudwatchLogsGroupClass, mut events: Vec>, token: Option, token_tx: oneshot::Sender>, @@ -80,6 +85,7 @@ impl CloudwatchFuture { retention_days, kms_key, tags, + group_class, }; let state = if let Some(token) = token { @@ -296,12 +302,14 @@ impl Client { let group_name = self.group_name.clone(); let kms_key = self.kms_key.clone(); let tags = self.tags.clone(); + let log_group_class = self.group_class.into(); Box::pin(async move { client .create_log_group() .log_group_name(group_name) .set_kms_key_id(kms_key) .set_tags(tags) + .log_group_class(log_group_class) .send() .await?; Ok(()) diff --git a/src/sinks/aws_cloudwatch_logs/service.rs b/src/sinks/aws_cloudwatch_logs/service.rs index cdb34702aa413..7f6f059a579b9 100644 --- a/src/sinks/aws_cloudwatch_logs/service.rs +++ b/src/sinks/aws_cloudwatch_logs/service.rs @@ -40,7 +40,7 @@ use vector_lib::{ use crate::sinks::{ aws_cloudwatch_logs::{ CloudwatchKey, - config::{CloudwatchLogsSinkConfig, Retention}, + config::{CloudwatchLogsGroupClass, CloudwatchLogsSinkConfig, Retention}, request, retry::CloudwatchRetryLogic, sink::BatchCloudwatchRequest, @@ -248,6 +248,7 @@ impl CloudwatchLogsSvc { let kms_key = config.kms_key.clone(); let tags = config.tags.clone(); + let group_class = config.group_class; CloudwatchLogsSvc { headers, @@ -259,6 +260,7 @@ impl CloudwatchLogsSvc { retention, kms_key, tags, + group_class, token: None, token_rx: None, } @@ -335,6 +337,7 @@ impl Service> for CloudwatchLogsSvc { self.retention.clone(), self.kms_key.clone(), self.tags.clone(), + self.group_class, event_batches, self.token.take(), tx, @@ -355,6 +358,7 @@ pub struct CloudwatchLogsSvc { retention: Retention, kms_key: Option, tags: Option>, + group_class: CloudwatchLogsGroupClass, token: Option, token_rx: Option>>, } diff --git a/website/cue/reference/components/sinks/generated/aws_cloudwatch_logs.cue b/website/cue/reference/components/sinks/generated/aws_cloudwatch_logs.cue index 9a76aa1a730aa..de1919d28a252 100644 --- a/website/cue/reference/components/sinks/generated/aws_cloudwatch_logs.cue +++ b/website/cue/reference/components/sinks/generated/aws_cloudwatch_logs.cue @@ -220,7 +220,8 @@ generated: components: sinks: aws_cloudwatch_logs: configuration: { } create_missing_group: { description: """ - Dynamically create a [log group][log_group] if it does not already exist. + Dynamically create a [log group][log_group] if it does not already exist. Its group + class is determined by `group_class`. This ignores `create_missing_stream` directly after creating the group and creates the first stream. @@ -679,6 +680,22 @@ generated: components: sinks: aws_cloudwatch_logs: configuration: { required: false type: string: examples: ["http://127.0.0.0:5000/path/to/service"] } + group_class: { + description: """ + The [log class][log_class] used when dynamically creating a log group via + `create_missing_group`. + + [log_class]: https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/CloudWatch_Logs_Log_Classes.html + """ + required: false + type: string: { + default: "STANDARD" + enum: { + INFREQUENT_ACCESS: "For cost-effective consolidation of logs that are queried only occasionally." + STANDARD: "For real-time monitoring and frequently-accessed logs." + } + } + } group_name: { description: """ The [group name][group_name] of the target CloudWatch Logs stream.