fix(reduce transform): use the correct merge strategy for top level o…

…bjects (#21067)

* fix(issue 21065): use the correct merge strategy for top level objects

* changelog

* revert config changes to preserve backwards compatibility

* enhance test case

* update test with another merge strategy, note we cannot handle strategies for both 'a' and 'a.b'

* tweak changelog text

* workaround for issue 21077

* add nested path to the test case

* Address review points

* add dot to message

* more review points

changelog.d/21067_reduce.fix.md

@@ -0,0 +1 @@
+Fixes a Vector v0.40.0 regression where the `reduce` transform would not group top level objects correctly.

src/internal_events/reduce.rs

@@ -1,5 +1,7 @@
 use metrics::counter;
-use vector_lib::internal_event::InternalEvent;
+use vector_lib::internal_event::{error_stage, error_type, InternalEvent};
+use vrl::path::PathParseError;
+use vrl::value::KeyString;
 
 #[derive(Debug)]
 pub struct ReduceStaleEventFlushed;
@@ -9,3 +11,28 @@ impl InternalEvent for ReduceStaleEventFlushed {
         counter!("stale_events_flushed_total").increment(1);
     }
 }
+
+#[derive(Debug)]
+pub struct ReduceAddEventError {
+    pub error: PathParseError,
+    pub path: KeyString,
+}
+
+impl InternalEvent for ReduceAddEventError {
+    fn emit(self) {
+        error!(
+            message = "Event field could not be reduced.",
+            path = ?self.path,
+            error = ?self.error,
+            error_type = error_type::CONDITION_FAILED,
+            stage = error_stage::PROCESSING,
+            internal_log_rate_limit = true,
+        );
+        counter!(
+            "component_errors_total",
+            "error_type" => error_type::PARSER_FAILED,
+            "stage" => error_stage::PROCESSING,
+        )
+        .increment(1);
+    }
+}

src/transforms/reduce/merge_strategy.rs

@@ -5,6 +5,7 @@ use bytes::{Bytes, BytesMut};
 use chrono::{DateTime, Utc};
 use ordered_float::NotNan;
 use vector_lib::configurable::configurable_component;
+use vrl::path::OwnedTargetPath;
 
 /// Strategies for merging events.
 #[configurable_component]
@@ -67,7 +68,11 @@ impl ReduceValueMerger for DiscardMerger {
         Ok(())
     }
 
-    fn insert_into(self: Box<Self>, path: &str, v: &mut LogEvent) -> Result<(), String> {
+    fn insert_into(
+        self: Box<Self>,
+        path: &OwnedTargetPath,
+        v: &mut LogEvent,
+    ) -> Result<(), String> {
         v.insert(path, self.v);
         Ok(())
     }
@@ -93,7 +98,11 @@ impl ReduceValueMerger for RetainMerger {
         Ok(())
     }
 
-    fn insert_into(self: Box<Self>, path: &str, v: &mut LogEvent) -> Result<(), String> {
+    fn insert_into(
+        self: Box<Self>,
+        path: &OwnedTargetPath,
+        v: &mut LogEvent,
+    ) -> Result<(), String> {
         v.insert(path, self.v);
         Ok(())
     }
@@ -133,7 +142,11 @@ impl ReduceValueMerger for ConcatMerger {
         }
     }
 
-    fn insert_into(self: Box<Self>, path: &str, v: &mut LogEvent) -> Result<(), String> {
+    fn insert_into(
+        self: Box<Self>,
+        path: &OwnedTargetPath,
+        v: &mut LogEvent,
+    ) -> Result<(), String> {
         v.insert(path, Value::Bytes(self.v.into()));
         Ok(())
     }
@@ -160,7 +173,11 @@ impl ReduceValueMerger for ConcatArrayMerger {
         Ok(())
     }
 
-    fn insert_into(self: Box<Self>, path: &str, v: &mut LogEvent) -> Result<(), String> {
+    fn insert_into(
+        self: Box<Self>,
+        path: &OwnedTargetPath,
+        v: &mut LogEvent,
+    ) -> Result<(), String> {
         v.insert(path, Value::Array(self.v));
         Ok(())
     }
@@ -183,7 +200,11 @@ impl ReduceValueMerger for ArrayMerger {
         Ok(())
     }
 
-    fn insert_into(self: Box<Self>, path: &str, v: &mut LogEvent) -> Result<(), String> {
+    fn insert_into(
+        self: Box<Self>,
+        path: &OwnedTargetPath,
+        v: &mut LogEvent,
+    ) -> Result<(), String> {
         v.insert(path, Value::Array(self.v));
         Ok(())
     }
@@ -215,7 +236,11 @@ impl ReduceValueMerger for LongestArrayMerger {
         }
     }
 
-    fn insert_into(self: Box<Self>, path: &str, v: &mut LogEvent) -> Result<(), String> {
+    fn insert_into(
+        self: Box<Self>,
+        path: &OwnedTargetPath,
+        v: &mut LogEvent,
+    ) -> Result<(), String> {
         v.insert(path, Value::Array(self.v));
         Ok(())
     }
@@ -247,7 +272,11 @@ impl ReduceValueMerger for ShortestArrayMerger {
         }
     }
 
-    fn insert_into(self: Box<Self>, path: &str, v: &mut LogEvent) -> Result<(), String> {
+    fn insert_into(
+        self: Box<Self>,
+        path: &OwnedTargetPath,
+        v: &mut LogEvent,
+    ) -> Result<(), String> {
         v.insert(path, Value::Array(self.v));
         Ok(())
     }
@@ -292,7 +321,11 @@ impl ReduceValueMerger for FlatUniqueMerger {
         Ok(())
     }
 
-    fn insert_into(self: Box<Self>, path: &str, v: &mut LogEvent) -> Result<(), String> {
+    fn insert_into(
+        self: Box<Self>,
+        path: &OwnedTargetPath,
+        v: &mut LogEvent,
+    ) -> Result<(), String> {
         v.insert(path, Value::Array(self.v.into_iter().collect()));
         Ok(())
     }
@@ -326,7 +359,11 @@ impl ReduceValueMerger for TimestampWindowMerger {
         Ok(())
     }
 
-    fn insert_into(self: Box<Self>, path: &str, v: &mut LogEvent) -> Result<(), String> {
+    fn insert_into(
+        self: Box<Self>,
+        path: &OwnedTargetPath,
+        v: &mut LogEvent,
+    ) -> Result<(), String> {
         v.insert(
             format!("{}_end", path).as_str(),
             Value::Timestamp(self.latest),
@@ -390,7 +427,11 @@ impl ReduceValueMerger for AddNumbersMerger {
         Ok(())
     }
 
-    fn insert_into(self: Box<Self>, path: &str, v: &mut LogEvent) -> Result<(), String> {
+    fn insert_into(
+        self: Box<Self>,
+        path: &OwnedTargetPath,
+        v: &mut LogEvent,
+    ) -> Result<(), String> {
         match self.v {
             NumberMergerValue::Float(f) => v.insert(path, Value::Float(f)),
             NumberMergerValue::Int(i) => v.insert(path, Value::Integer(i)),
@@ -449,7 +490,11 @@ impl ReduceValueMerger for MaxNumberMerger {
         Ok(())
     }
 
-    fn insert_into(self: Box<Self>, path: &str, v: &mut LogEvent) -> Result<(), String> {
+    fn insert_into(
+        self: Box<Self>,
+        path: &OwnedTargetPath,
+        v: &mut LogEvent,
+    ) -> Result<(), String> {
         match self.v {
             NumberMergerValue::Float(f) => v.insert(path, Value::Float(f)),
             NumberMergerValue::Int(i) => v.insert(path, Value::Integer(i)),
@@ -508,7 +553,11 @@ impl ReduceValueMerger for MinNumberMerger {
         Ok(())
     }
 
-    fn insert_into(self: Box<Self>, path: &str, v: &mut LogEvent) -> Result<(), String> {
+    fn insert_into(
+        self: Box<Self>,
+        path: &OwnedTargetPath,
+        v: &mut LogEvent,
+    ) -> Result<(), String> {
         match self.v {
             NumberMergerValue::Float(f) => v.insert(path, Value::Float(f)),
             NumberMergerValue::Int(i) => v.insert(path, Value::Integer(i)),
@@ -519,7 +568,8 @@ impl ReduceValueMerger for MinNumberMerger {
 
 pub trait ReduceValueMerger: std::fmt::Debug + Send + Sync {
     fn add(&mut self, v: Value) -> Result<(), String>;
-    fn insert_into(self: Box<Self>, path: &str, v: &mut LogEvent) -> Result<(), String>;
+    fn insert_into(self: Box<Self>, path: &OwnedTargetPath, v: &mut LogEvent)
+        -> Result<(), String>;
 }
 
 impl From<Value> for Box<dyn ReduceValueMerger> {
@@ -612,10 +662,10 @@ pub(crate) fn get_value_merger(
 
 #[cfg(test)]
 mod test {
-    use serde_json::json;
-
     use super::*;
     use crate::event::LogEvent;
+    use serde_json::json;
+    use vrl::owned_event_path;
 
     #[test]
     fn initial_values() {
@@ -876,7 +926,8 @@ mod test {
         let mut merger = get_value_merger(initial, strategy)?;
         merger.add(additional)?;
         let mut output = LogEvent::default();
-        merger.insert_into("out", &mut output)?;
-        Ok(output.remove("out").unwrap())
+        let out_path = owned_event_path!("out");
+        merger.insert_into(&out_path, &mut output)?;
+        Ok(output.remove(&out_path).unwrap())
     }
 }