chore: Introduce a rename_key on LogEvent (#8907)

* Introduce a `rename_key` on `LogEvent`

This commit introduces a new function to `LogEvent` that allows for in-place key
renames. This resolves #8491 and is an extract from the ongoing #8825.

Signed-off-by: Brian L. Troutwine <brian@troutwine.us>

* feedback

Signed-off-by: Brian L. Troutwine <brian@troutwine.us>

* Correct function reference name

Signed-off-by: Brian L. Troutwine <brian@troutwine.us>

lib/vector-core/src/event/log_event.rs

@@ -121,6 +121,28 @@ impl LogEvent {
         util::log::insert_path(self.as_map_mut(), key, value.into())
     }
 
+    /// Rename a key in place without reference to pathing
+    ///
+    /// The function will rename a key in place without reference to any path
+    /// information in the key, much as if you were to call [`remove`] and then
+    /// [`insert_flat`].
+    ///
+    /// This function is a no-op if `from_key` and `to_key` are identical. If
+    /// `to_key` already exists in the structure its value will be overwritten
+    /// silently.
+    #[instrument(level = "trace", skip(self, from_key, to_key), fields(key = %from_key))]
+    #[inline]
+    pub fn rename_key_flat<K>(&mut self, from_key: K, to_key: K)
+    where
+        K: AsRef<str> + Into<String> + PartialEq + Display,
+    {
+        if from_key != to_key {
+            if let Some(val) = self.remove(from_key) {
+                self.insert_flat(to_key, val);
+            }
+        }
+    }
+
     #[instrument(level = "trace", skip(self, key), fields(key = %key))]
     pub fn insert_flat<K, V>(&mut self, key: K, value: V)
     where
@@ -441,11 +463,95 @@ mod test {
     use serde_json::json;
     use std::str::FromStr;
 
+    // The following two tests assert that renaming a key has no effect if the
+    // keys are equivalent, whether the key exists in the log or not.
+    #[test]
+    fn rename_key_flat_equiv_exists() {
+        let mut fields = BTreeMap::new();
+        fields.insert("one".to_string(), Value::Integer(1_i64));
+        fields.insert("two".to_string(), Value::Integer(2_i64));
+        let expected_fields = fields.clone();
+
+        let mut base = LogEvent::from_parts(fields, EventMetadata::default());
+        base.rename_key_flat("one", "one");
+        let (actual_fields, _) = base.into_parts();
+
+        assert_eq!(expected_fields, actual_fields);
+    }
+    #[test]
+    fn rename_key_flat_equiv_not_exists() {
+        let mut fields = BTreeMap::new();
+        fields.insert("one".to_string(), Value::Integer(1_i64));
+        fields.insert("two".to_string(), Value::Integer(2_i64));
+        let expected_fields = fields.clone();
+
+        let mut base = LogEvent::from_parts(fields, EventMetadata::default());
+        base.rename_key_flat("three", "three");
+        let (actual_fields, _) = base.into_parts();
+
+        assert_eq!(expected_fields, actual_fields);
+    }
+    // Assert that renaming a key has no effect if the key does not originally
+    // exist in the log, when the to -> from keys are not identical.
+    #[test]
+    fn rename_key_flat_not_exists() {
+        let mut fields = BTreeMap::new();
+        fields.insert("one".to_string(), Value::Integer(1_i64));
+        fields.insert("two".to_string(), Value::Integer(2_i64));
+        let expected_fields = fields.clone();
+
+        let mut base = LogEvent::from_parts(fields, EventMetadata::default());
+        base.rename_key_flat("three", "four");
+        let (actual_fields, _) = base.into_parts();
+
+        assert_eq!(expected_fields, actual_fields);
+    }
+    // Assert that renaming a key has the effect of moving the value from one
+    // key name to another if the key exists.
+    #[test]
+    fn rename_key_flat_no_overlap() {
+        let mut fields = BTreeMap::new();
+        fields.insert("one".to_string(), Value::Integer(1_i64));
+        fields.insert("two".to_string(), Value::Integer(2_i64));
+
+        let mut expected_fields = fields.clone();
+        let val = expected_fields.remove("one").unwrap();
+        expected_fields.insert("three".to_string(), val);
+
+        let mut base = LogEvent::from_parts(fields, EventMetadata::default());
+        base.rename_key_flat("one", "three");
+        let (actual_fields, _) = base.into_parts();
+
+        assert_eq!(expected_fields, actual_fields);
+    }
+    // Assert that renaming a key has the effect of moving the value from one
+    // key name to another if the key exists and will overwrite another key if
+    // it exists.
+    #[test]
+    fn rename_key_flat_overlap() {
+        let mut fields = BTreeMap::new();
+        fields.insert("one".to_string(), Value::Integer(1_i64));
+        fields.insert("two".to_string(), Value::Integer(2_i64));
+
+        let mut expected_fields = fields.clone();
+        let val = expected_fields.remove("one").unwrap();
+        expected_fields.insert("two".to_string(), val);
+
+        let mut base = LogEvent::from_parts(fields, EventMetadata::default());
+        base.rename_key_flat("one", "two");
+        let (actual_fields, _) = base.into_parts();
+
+        assert_eq!(expected_fields, actual_fields);
+    }
+
     // This test iterates over the `tests/data/fixtures/log_event` folder and:
-    //   * Ensures the EventLog parsed from bytes and turned into a serde_json::Value are equal to the
-    //     item being just plain parsed as json.
     //
-    // Basically: This test makes sure we aren't mutilating any content users might be sending.
+    //   * Ensures the EventLog parsed from bytes and turned into a
+    //   serde_json::Value are equal to the item being just plain parsed as
+    //   json.
+    //
+    // Basically: This test makes sure we aren't mutilating any content users
+    // might be sending.
     #[test]
     fn json_value_to_vector_log_event_to_json_value() {
         const FIXTURE_ROOT: &str = "tests/data/fixtures/log_event";

src/sinks/datadog/logs/service.rs

@@ -100,25 +100,9 @@ impl HttpSink for Service {
 
     fn encode_event(&self, mut event: Event) -> Option<Self::Input> {
         let log = event.as_mut_log();
-
-        if self.log_schema_message_key != "message" {
-            if let Some(message) = log.remove(self.log_schema_message_key) {
-                log.insert_flat("message", message);
-            }
-        }
-
-        if self.log_schema_timestamp_key != "date" {
-            if let Some(timestamp) = log.remove(self.log_schema_timestamp_key) {
-                log.insert_flat("date", timestamp);
-            }
-        }
-
-        if self.log_schema_host_key != "host" {
-            if let Some(host) = log.remove(self.log_schema_host_key) {
-                log.insert_flat("host", host);
-            }
-        }
-
+        log.rename_key_flat(self.log_schema_message_key, "message");
+        log.rename_key_flat(self.log_schema_timestamp_key, "date");
+        log.rename_key_flat(self.log_schema_host_key, "host");
         self.encoding.apply_rules(&mut event);
 
         let (fields, metadata) = event.into_log().into_parts();