feat(aws_cloudwatch_logs sink): allow setting type of log class to cr…

…eate

Problem: Prior to this commit it was not possible to specify the log
group's class type. The method prior always created a Standard type.

------------------------------------------------------------

Solution: Allow specifying the log group class type via a new field,
`group_class`.

Initial Issue Report: vectordotdev#22008

Closes vectordotdev#22008

changelog.d/22008-specify-cloudwatch-log-class.enhancement.md

@@ -0,0 +1,3 @@
+The Cloudwatch Logs Sink now supports specifying the type of log class to create.
+
+authors: PriceHiller

src/sinks/aws_cloudwatch_logs/config.rs

@@ -76,6 +76,30 @@ where
     }
 }
 
+/// Defines the log class to create
+///
+/// See https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/CloudWatch_Logs_Log_Classes.html
+#[configurable_component]
+#[derive(Clone, Debug, Default)]
+pub enum LogGroupClassDef {
+    /// Logs that require real-time monitoring or frequently accessed logs
+    #[default]
+    Standard,
+    /// Log class that can be used to cost-effectively consolidate logs
+    InfrequentAccess,
+}
+
+impl From<LogGroupClassDef> for aws_sdk_cloudwatchlogs::types::LogGroupClass {
+    fn from(value: LogGroupClassDef) -> Self {
+        match value {
+            LogGroupClassDef::Standard => aws_sdk_cloudwatchlogs::types::LogGroupClass::Standard,
+            LogGroupClassDef::InfrequentAccess => {
+                aws_sdk_cloudwatchlogs::types::LogGroupClass::InfrequentAccess
+            }
+        }
+    }
+}
+
 /// Configuration for the `aws_cloudwatch_logs` sink.
 #[configurable_component(sink(
     "aws_cloudwatch_logs",
@@ -110,6 +134,7 @@ pub struct CloudwatchLogsSinkConfig {
     pub region: RegionOrEndpoint,
 
     /// Dynamically create a [log group][log_group] if it does not already exist.
+    /// This will create the log group with the group class specified by the `group_class` option.
     ///
     /// 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,
 
+    /// Specifies the specific [group class][group_class] to create when
+    /// `create_missing_group` is enabled.
+    ///
+    /// [group_class]: https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/CloudWatch_Logs_Log_Classes.html
+    #[configurable(derived)]
+    #[serde(default)]
+    pub group_class: LogGroupClassDef,
+
     /// 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
@@ -236,6 +269,7 @@ fn default_config(encoding: EncodingConfig) -> CloudwatchLogsSinkConfig {
     CloudwatchLogsSinkConfig {
         encoding,
         group_name: Default::default(),
+        group_class: Default::default(),
         stream_name: Default::default(),
         region: Default::default(),
         create_missing_group: true,

src/sinks/aws_cloudwatch_logs/integration_tests.rs

@@ -39,6 +39,7 @@ async fn cloudwatch_insert_log_event() {
     let config = CloudwatchLogsSinkConfig {
         stream_name: Template::try_from(stream_name.as_str()).unwrap(),
         group_name: Template::try_from(GROUP_NAME).unwrap(),
+        group_class: Default::default(),
         region: RegionOrEndpoint::with_both("us-east-1", cloudwatch_address().as_str()),
         encoding: TextSerializerConfig::default().into(),
         create_missing_group: true,
@@ -90,6 +91,7 @@ async fn cloudwatch_insert_log_events_sorted() {
     let config = CloudwatchLogsSinkConfig {
         stream_name: Template::try_from(stream_name.as_str()).unwrap(),
         group_name: Template::try_from(GROUP_NAME).unwrap(),
+        group_class: Default::default(),
         region: RegionOrEndpoint::with_both("us-east-1", cloudwatch_address().as_str()),
         encoding: TextSerializerConfig::default().into(),
         create_missing_group: true,
@@ -166,6 +168,7 @@ async fn cloudwatch_insert_out_of_range_timestamp() {
     let config = CloudwatchLogsSinkConfig {
         stream_name: Template::try_from(stream_name.as_str()).unwrap(),
         group_name: Template::try_from(GROUP_NAME).unwrap(),
+        group_class: Default::default(),
         region: RegionOrEndpoint::with_both("us-east-1", cloudwatch_address().as_str()),
         encoding: TextSerializerConfig::default().into(),
         create_missing_group: true,
@@ -243,6 +246,7 @@ async fn cloudwatch_dynamic_group_and_stream_creation() {
     let config = CloudwatchLogsSinkConfig {
         stream_name: Template::try_from(stream_name.as_str()).unwrap(),
         group_name: Template::try_from(group_name.as_str()).unwrap(),
+        group_class: Default::default(),
         region: RegionOrEndpoint::with_both("us-east-1", cloudwatch_address().as_str()),
         encoding: TextSerializerConfig::default().into(),
         create_missing_group: true,
@@ -299,6 +303,7 @@ async fn cloudwatch_insert_log_event_batched() {
     let config = CloudwatchLogsSinkConfig {
         stream_name: Template::try_from(stream_name.as_str()).unwrap(),
         group_name: Template::try_from(group_name.as_str()).unwrap(),
+        group_class: Default::default(),
         region: RegionOrEndpoint::with_both("us-east-1", cloudwatch_address().as_str()),
         encoding: TextSerializerConfig::default().into(),
         create_missing_group: true,
@@ -350,6 +355,7 @@ async fn cloudwatch_insert_log_event_partitioned() {
     let config = CloudwatchLogsSinkConfig {
         group_name: Template::try_from(GROUP_NAME).unwrap(),
         stream_name: Template::try_from(format!("{}-{{{{key}}}}", stream_name)).unwrap(),
+        group_class: Default::default(),
         region: RegionOrEndpoint::with_both("us-east-1", cloudwatch_address().as_str()),
         encoding: TextSerializerConfig::default().into(),
         create_missing_group: true,
@@ -443,6 +449,7 @@ async fn cloudwatch_healthcheck() {
     let config = CloudwatchLogsSinkConfig {
         stream_name: Template::try_from("test-stream").unwrap(),
         group_name: Template::try_from(GROUP_NAME).unwrap(),
+        group_class: Default::default(),
         region: RegionOrEndpoint::with_both("us-east-1", cloudwatch_address().as_str()),
         encoding: TextSerializerConfig::default().into(),
         create_missing_group: true,

src/sinks/aws_cloudwatch_logs/request.rs

@@ -12,7 +12,7 @@ use aws_sdk_cloudwatchlogs::{
         put_log_events::{PutLogEventsError, PutLogEventsOutput},
         put_retention_policy::PutRetentionPolicyError,
     },
-    types::InputLogEvent,
+    types::{InputLogEvent, LogGroupClass},
     Client as CloudwatchLogsClient,
 };
 use aws_smithy_runtime_api::client::{orchestrator::HttpResponse, result::SdkError};
@@ -38,6 +38,7 @@ struct Client {
     client: CloudwatchLogsClient,
     stream_name: String,
     group_name: String,
+    group_class: LogGroupClass,
     headers: IndexMap<HeaderName, HeaderValue>,
     retention_days: u32,
 }
@@ -60,6 +61,7 @@ impl CloudwatchFuture {
         headers: IndexMap<HeaderName, HeaderValue>,
         stream_name: String,
         group_name: String,
+        group_class: LogGroupClass,
         create_missing_group: bool,
         create_missing_stream: bool,
         retention: Retention,
@@ -72,6 +74,7 @@ impl CloudwatchFuture {
             client,
             stream_name,
             group_name,
+            group_class,
             headers,
             retention_days,
         };
@@ -288,10 +291,12 @@ impl Client {
     pub fn create_log_group(&self) -> ClientResult<(), CreateLogGroupError> {
         let client = self.client.clone();
         let group_name = self.group_name.clone();
+        let group_class = self.group_class.clone();
         Box::pin(async move {
             client
                 .create_log_group()
                 .log_group_name(group_name)
+                .log_group_class(group_class)
                 .send()
                 .await?;
             Ok(())

src/sinks/aws_cloudwatch_logs/service.rs

@@ -10,7 +10,7 @@ use aws_sdk_cloudwatchlogs::{
         describe_log_streams::DescribeLogStreamsError, put_log_events::PutLogEventsError,
         put_retention_policy::PutRetentionPolicyError,
     },
-    types::InputLogEvent,
+    types::{InputLogEvent, LogGroupClass},
     Client as CloudwatchLogsClient,
 };
 use aws_smithy_runtime_api::client::{orchestrator::HttpResponse, result::SdkError};
@@ -240,11 +240,14 @@ impl CloudwatchLogsSvc {
 
         let retention = config.retention.clone();
 
+        let group_class = config.group_class.into();
+
         CloudwatchLogsSvc {
             headers,
             client,
             stream_name,
             group_name,
+            group_class,
             create_missing_group,
             create_missing_stream,
             retention,
@@ -319,6 +322,7 @@ impl Service<Vec<InputLogEvent>> for CloudwatchLogsSvc {
                 self.headers.clone(),
                 self.stream_name.clone(),
                 self.group_name.clone(),
+                self.group_class.clone(),
                 self.create_missing_group,
                 self.create_missing_stream,
                 self.retention.clone(),
@@ -337,6 +341,7 @@ pub struct CloudwatchLogsSvc {
     headers: IndexMap<HeaderName, HeaderValue>,
     stream_name: String,
     group_name: String,
+    group_class: LogGroupClass,
     create_missing_group: bool,
     create_missing_stream: bool,
     retention: Retention,