enhancement(reduce transform): New setting for reduce transform: end_every_period_ms (#20440)

* New setting for reduce transform: end_every_period_ms

* Correct changelog name to match pattern

Signed-off-by: Jesse Szwedko <[email protected]>

* regenerate component docs

Signed-off-by: Jesse Szwedko <[email protected]>

---------

Signed-off-by: Jesse Szwedko <[email protected]>
Co-authored-by: Jesse Szwedko <[email protected]>

Author: charlesconnell

changelog.d/reduce_end_every_period.feature.md

@@ -0,0 +1,5 @@
+A new configuration option `end_every_period_ms` is available on reduce transforms
+If supplied, every time this interval elapses for a given grouping, the reduced value
+for that grouping is flushed. Checked every `flush_period_ms`.
+
+authors: charlesconnell

lib/vector-config/src/external/serde_with.rs

@@ -2,6 +2,7 @@ use std::cell::RefCell;
 
 use vector_config_common::{attributes::CustomAttribute, constants};
 
+use crate::schema::generate_optional_schema;
 use crate::{
     num::NumberClass,
     schema::{generate_number_schema, SchemaGenerator, SchemaObject},
@@ -132,3 +133,12 @@ impl Configurable for serde_with::DurationMilliSeconds<u64, serde_with::formats:
         Ok(generate_number_schema::<u64>())
     }
 }
+
+impl Configurable for Option<serde_with::DurationMilliSeconds<u64, serde_with::formats::Strict>> {
+    fn generate_schema(gen: &RefCell<SchemaGenerator>) -> Result<SchemaObject, GenerateError>
+    where
+        Self: Sized,
+    {
+        generate_optional_schema(&u64::as_configurable_ref(), gen)
+    }
+}

src/transforms/reduce/config.rs

@@ -36,6 +36,13 @@ pub struct ReduceConfig {
     #[configurable(metadata(docs::human_name = "Expire After"))]
     pub expire_after_ms: Duration,
 
+    /// If supplied, every time this interval elapses for a given grouping, the reduced value
+    /// for that grouping is flushed. Checked every flush_period_ms.
+    #[serde_as(as = "Option<serde_with::DurationMilliSeconds<u64>>")]
+    #[derivative(Default(value = "Option::None"))]
+    #[configurable(metadata(docs::human_name = "End-Every Period"))]
+    pub end_every_period_ms: Option<Duration>,
+
     /// The interval to check for and flush any expired events, in milliseconds.
     #[serde(default = "default_flush_period_ms")]
     #[serde_as(as = "serde_with::DurationMilliSeconds<u64>")]

src/transforms/reduce/transform.rs

@@ -24,6 +24,7 @@ struct ReduceState {
     events: usize,
     fields: HashMap<KeyString, Box<dyn ReduceValueMerger>>,
     stale_since: Instant,
+    creation: Instant,
     metadata: EventMetadata,
 }
 
@@ -35,6 +36,7 @@ impl ReduceState {
         Self {
             events: 0,
             stale_since: Instant::now(),
+            creation: Instant::now(),
             fields,
             metadata,
         }
@@ -93,6 +95,7 @@ impl ReduceState {
 pub struct Reduce {
     expire_after: Duration,
     flush_period: Duration,
+    end_every_period: Option<Duration>,
     group_by: Vec<String>,
     merge_strategies: IndexMap<KeyString, MergeStrategy>,
     reduce_merge_states: HashMap<Discriminant, ReduceState>,
@@ -126,6 +129,7 @@ impl Reduce {
         Ok(Reduce {
             expire_after: config.expire_after_ms,
             flush_period: config.flush_period_ms,
+            end_every_period: config.end_every_period_ms,
             group_by,
             merge_strategies: config.merge_strategies.clone(),
             reduce_merge_states: HashMap::new(),
@@ -139,6 +143,12 @@ impl Reduce {
         let mut flush_discriminants = Vec::new();
         let now = Instant::now();
         for (k, t) in &self.reduce_merge_states {
+            if let Some(period) = self.end_every_period {
+                if (now - t.creation) >= period {
+                    flush_discriminants.push(k.clone());
+                }
+            }
+
             if (now - t.stale_since) >= self.expire_after {
                 flush_discriminants.push(k.clone());
             }

website/cue/reference/components/transforms/base/reduce.cue

@@ -1,6 +1,14 @@
 package metadata
 
 base: components: transforms: reduce: configuration: {
+	end_every_period_ms: {
+		description: """
+			If supplied, every time this interval elapses for a given grouping, the reduced value
+			for that grouping is flushed. Checked every flush_period_ms.
+			"""
+		required: false
+		type: uint: {}
+	}
 	ends_when: {
 		description: """
 			A condition used to distinguish the final event of a transaction.