chore(core): Implement LogEvent::get_mut_by_meaning (#20358)

* chore(core): Implement `LogEvent::get_mut_by_meaning`

Trying to get mutable access to a log event field by meaning currently requires
cloning the owned path returned by `find_key_by_meaning` because of the
reference requiring another borrow to the log event. Instead, we can clone the
underlying schema `Arc` and use that to index into the mutable log.

* Drop unused added method

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

@@ -318,14 +318,22 @@ impl LogEvent {
     /// aware that if the field has been dropped and then somehow re-added, we still fetch
     /// the dropped value here.
     pub fn get_by_meaning(&self, meaning: impl AsRef<str>) -> Option<&Value> {
-        if let Some(dropped) = self.metadata().dropped_field(&meaning) {
-            Some(dropped)
-        } else {
+        self.metadata().dropped_field(&meaning).or_else(|| {
             self.metadata()
                 .schema_definition()
                 .meaning_path(meaning.as_ref())
                 .and_then(|path| self.get(path))
-        }
+        })
+    }
+
+    /// Retrieves the mutable value of a field based on it's meaning.
+    /// Note that this does _not_ check the dropped fields, unlike `get_by_meaning`, since the
+    /// purpose of the mutable reference is to be able to modify the value and modifying the dropped
+    /// fields has no effect on the resulting event.
+    pub fn get_mut_by_meaning(&mut self, meaning: impl AsRef<str>) -> Option<&mut Value> {
+        Arc::clone(self.metadata.schema_definition())
+            .meaning_path(meaning.as_ref())
+            .and_then(|path| self.get_mut(path))
     }
 
     /// Retrieves the target path of a field based on the specified `meaning`.

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

@@ -337,8 +337,8 @@ impl EventMetadata {
     }
 
     /// Get the schema definition.
-    pub fn schema_definition(&self) -> &schema::Definition {
-        self.schema_definition.as_ref()
+    pub fn schema_definition(&self) -> &Arc<schema::Definition> {
+        &self.schema_definition
     }
 
     /// Set the schema definition.

src/sources/datadog_agent/tests.rs

@@ -118,7 +118,7 @@ fn test_decode_log_body() {
             assert_eq!(log["ddtags"], msg.ddtags.into());
 
             assert_eq!(
-                event.metadata().schema_definition(),
+                event.metadata().schema_definition().as_ref(),
                 &test_logs_schema_definition()
             );
         }
@@ -320,7 +320,7 @@ async fn full_payload_v1() {
             assert!(event.metadata().datadog_api_key().is_none());
             assert_eq!(*log.get_source_type().unwrap(), "datadog_agent".into());
             assert_eq!(
-                event.metadata().schema_definition(),
+                event.metadata().schema_definition().as_ref(),
                 &test_logs_schema_definition()
             );
         }
@@ -382,7 +382,7 @@ async fn full_payload_v2() {
             assert!(event.metadata().datadog_api_key().is_none());
             assert_eq!(*log.get_source_type().unwrap(), "datadog_agent".into());
             assert_eq!(
-                event.metadata().schema_definition(),
+                event.metadata().schema_definition().as_ref(),
                 &test_logs_schema_definition()
             );
         }
@@ -444,7 +444,7 @@ async fn no_api_key() {
             assert!(event.metadata().datadog_api_key().is_none());
             assert_eq!(*log.get_source_type().unwrap(), "datadog_agent".into());
             assert_eq!(
-                event.metadata().schema_definition(),
+                event.metadata().schema_definition().as_ref(),
                 &test_logs_schema_definition()
             );
         }
@@ -509,7 +509,7 @@ async fn api_key_in_url() {
                 "12345678abcdefgh12345678abcdefgh"
             );
             assert_eq!(
-                event.metadata().schema_definition(),
+                event.metadata().schema_definition().as_ref(),
                 &test_logs_schema_definition()
             );
         }
@@ -574,7 +574,7 @@ async fn api_key_in_query_params() {
                 "12345678abcdefgh12345678abcdefgh"
             );
             assert_eq!(
-                event.metadata().schema_definition(),
+                event.metadata().schema_definition().as_ref(),
                 &test_logs_schema_definition()
             );
         }
@@ -645,7 +645,7 @@ async fn api_key_in_header() {
                 "12345678abcdefgh12345678abcdefgh"
             );
             assert_eq!(
-                event.metadata().schema_definition(),
+                event.metadata().schema_definition().as_ref(),
                 &test_logs_schema_definition()
             );
         }
@@ -789,7 +789,7 @@ async fn ignores_api_key() {
             assert_eq!(*log.get_source_type().unwrap(), "datadog_agent".into());
             assert!(event.metadata().datadog_api_key().is_none());
             assert_eq!(
-                event.metadata().schema_definition(),
+                event.metadata().schema_definition().as_ref(),
                 &test_logs_schema_definition()
             );
         }
@@ -1498,7 +1498,7 @@ async fn split_outputs() {
                 "12345678abcdefgh12345678abcdefgh"
             );
             assert_eq!(
-                event.metadata().schema_definition(),
+                event.metadata().schema_definition().as_ref(),
                 &test_logs_schema_definition()
             );
         }

src/transforms/remap.rs

@@ -1238,9 +1238,8 @@ mod tests {
                     MetricValue::Counter { value: 1.0 },
                     // The schema definition is set in the topology, which isn't used in this test. Setting the definition
                     // to the actual value to skip the assertion here
-                    EventMetadata::default().with_schema_definition(&Arc::new(
-                        output.metadata().schema_definition().clone()
-                    )),
+                    EventMetadata::default()
+                        .with_schema_definition(output.metadata().schema_definition()),
                 )
                 .with_tags(Some(metric_tags! {
                     "hello" => "world",
@@ -1259,9 +1258,8 @@ mod tests {
                     MetricValue::Counter { value: 1.0 },
                     // The schema definition is set in the topology, which isn't used in this test. Setting the definition
                     // to the actual value to skip the assertion here
-                    EventMetadata::default().with_schema_definition(&Arc::new(
-                        output.metadata().schema_definition().clone()
-                    )),
+                    EventMetadata::default()
+                        .with_schema_definition(output.metadata().schema_definition()),
                 )
                 .with_tags(Some(metric_tags! {
                     "hello" => "goodbye",
@@ -1283,9 +1281,8 @@ mod tests {
                     MetricValue::Counter { value: 1.0 },
                     // The schema definition is set in the topology, which isn't used in this test. Setting the definition
                     // to the actual value to skip the assertion here
-                    EventMetadata::default().with_schema_definition(&Arc::new(
-                        output.metadata().schema_definition().clone()
-                    )),
+                    EventMetadata::default()
+                        .with_schema_definition(output.metadata().schema_definition()),
                 )
                 .with_tags(Some(metric_tags! {
                     "not_hello" => "oops",