feat: LogSchema metadata key refacoring (#18099)

pront · web-flow · commit a8bb9f45867a · 2023-07-27T20:52:23.000Z
* feat: LogSchema metadata key refacoring

* fix failing test
diff --git a/lib/codecs/src/decoding/format/syslog.rs b/lib/codecs/src/decoding/format/syslog.rs
@@ -483,9 +483,8 @@ fn insert_fields_from_syslog(
 
 #[cfg(test)]
 mod tests {
-    use vector_core::config::{init_log_schema, log_schema, LogSchema};
-
     use super::*;
+    use vector_core::config::{init_log_schema, log_schema, LogSchema};
 
     #[test]
     fn deserialize_syslog_legacy_namespace() {
@@ -522,8 +521,12 @@ mod tests {
 
     fn init() {
         let mut schema = LogSchema::default();
-        schema.set_message_key(Some(owned_value_path!("legacy_message")));
-        schema.set_message_key(Some(owned_value_path!("legacy_timestamp")));
+        schema.set_message_key(Some(OwnedTargetPath::event(owned_value_path!(
+            "legacy_message"
+        ))));
+        schema.set_message_key(Some(OwnedTargetPath::event(owned_value_path!(
+            "legacy_timestamp"
+        ))));
         init_log_schema(schema, false);
     }
 }
diff --git a/lib/vector-core/src/config/log_schema.rs b/lib/vector-core/src/config/log_schema.rs
@@ -1,8 +1,7 @@
-use lookup::lookup_v2::{OptionalTargetPath, OptionalValuePath};
+use lookup::lookup_v2::OptionalTargetPath;
 use lookup::{OwnedTargetPath, OwnedValuePath};
 use once_cell::sync::{Lazy, OnceCell};
 use vector_config::configurable_component;
-use vrl::path::PathPrefix;
 
 static LOG_SCHEMA: OnceCell<LogSchema> = OnceCell::new();
 static LOG_SCHEMA_DEFAULT: Lazy<LogSchema> = Lazy::new(LogSchema::default);
@@ -74,7 +73,7 @@ pub struct LogSchema {
     /// Generally, this field will be set by Vector to hold event-specific metadata, such as
     /// annotations by the `remap` transform when an error or abort is encountered.
     #[serde(default = "LogSchema::default_metadata_key")]
-    metadata_key: OptionalValuePath,
+    metadata_key: OptionalTargetPath,
 }
 
 impl Default for LogSchema {
@@ -106,8 +105,8 @@ impl LogSchema {
         OptionalTargetPath::event(SOURCE_TYPE)
     }
 
-    fn default_metadata_key() -> OptionalValuePath {
-        OptionalValuePath::new(METADATA)
+    fn default_metadata_key() -> OptionalTargetPath {
+        OptionalTargetPath::event(METADATA)
     }
 
     pub fn message_key(&self) -> Option<&OwnedValuePath> {
@@ -140,7 +139,7 @@ impl LogSchema {
     }
 
     pub fn metadata_key(&self) -> Option<&OwnedValuePath> {
-        self.metadata_key.path.as_ref()
+        self.metadata_key.as_ref().map(|key| &key.path)
     }
 
     pub fn message_key_target_path(&self) -> Option<&OwnedTargetPath> {
@@ -159,24 +158,28 @@ impl LogSchema {
         self.source_type_key.as_ref()
     }
 
-    pub fn set_message_key(&mut self, path: Option<OwnedValuePath>) {
-        self.message_key = OptionalTargetPath::from(PathPrefix::Event, path);
+    pub fn metadata_key_target_path(&self) -> Option<&OwnedTargetPath> {
+        self.metadata_key.as_ref()
     }
 
-    pub fn set_timestamp_key(&mut self, path: Option<OwnedValuePath>) {
-        self.timestamp_key = OptionalTargetPath::from(PathPrefix::Event, path);
+    pub fn set_message_key(&mut self, path: Option<OwnedTargetPath>) {
+        self.message_key = OptionalTargetPath { path };
     }
 
-    pub fn set_host_key(&mut self, path: Option<OwnedValuePath>) {
-        self.host_key = OptionalTargetPath::from(PathPrefix::Event, path);
+    pub fn set_timestamp_key(&mut self, path: Option<OwnedTargetPath>) {
+        self.timestamp_key = OptionalTargetPath { path };
     }
 
-    pub fn set_source_type_key(&mut self, path: Option<OwnedValuePath>) {
-        self.source_type_key = OptionalTargetPath::from(PathPrefix::Event, path);
+    pub fn set_host_key(&mut self, path: Option<OwnedTargetPath>) {
+        self.host_key = OptionalTargetPath { path };
     }
 
-    pub fn set_metadata_key(&mut self, path: Option<OwnedValuePath>) {
-        self.metadata_key = OptionalValuePath { path };
+    pub fn set_source_type_key(&mut self, path: Option<OwnedTargetPath>) {
+        self.source_type_key = OptionalTargetPath { path };
+    }
+
+    pub fn set_metadata_key(&mut self, path: Option<OwnedTargetPath>) {
+        self.metadata_key = OptionalTargetPath { path };
     }
 
     /// Merge two `LogSchema` instances together.
@@ -195,35 +198,35 @@ impl LogSchema {
             {
                 errors.push("conflicting values for 'log_schema.host_key' found".to_owned());
             } else {
-                self.set_host_key(other.host_key().cloned());
+                self.set_host_key(other.host_key_target_path().cloned());
             }
             if self.message_key() != LOG_SCHEMA_DEFAULT.message_key()
                 && self.message_key() != other.message_key()
             {
                 errors.push("conflicting values for 'log_schema.message_key' found".to_owned());
             } else {
-                self.set_message_key(other.message_key().cloned());
+                self.set_message_key(other.message_key_target_path().cloned());
             }
             if self.timestamp_key() != LOG_SCHEMA_DEFAULT.timestamp_key()
                 && self.timestamp_key() != other.timestamp_key()
             {
                 errors.push("conflicting values for 'log_schema.timestamp_key' found".to_owned());
             } else {
-                self.set_timestamp_key(other.timestamp_key().cloned());
+                self.set_timestamp_key(other.timestamp_key_target_path().cloned());
             }
             if self.source_type_key() != LOG_SCHEMA_DEFAULT.source_type_key()
                 && self.source_type_key() != other.source_type_key()
             {
                 errors.push("conflicting values for 'log_schema.source_type_key' found".to_owned());
             } else {
-                self.set_source_type_key(other.source_type_key().cloned());
+                self.set_source_type_key(other.source_type_key_target_path().cloned());
             }
             if self.metadata_key() != LOG_SCHEMA_DEFAULT.metadata_key()
                 && self.metadata_key() != other.metadata_key()
             {
                 errors.push("conflicting values for 'log_schema.metadata_key' found".to_owned());
             } else {
-                self.set_metadata_key(other.metadata_key().cloned());
+                self.set_metadata_key(other.metadata_key_target_path().cloned());
             }
         }
 
diff --git a/lib/vector-core/src/config/mod.rs b/lib/vector-core/src/config/mod.rs
@@ -551,15 +551,14 @@ mod test {
     use chrono::Utc;
     use lookup::{event_path, owned_value_path, OwnedTargetPath};
     use vector_common::btreemap;
-    use vrl::path::OwnedValuePath;
     use vrl::value::Kind;
 
     #[test]
     fn test_insert_standard_vector_source_metadata() {
-        let nested_path = "a.b.c.d".to_string();
-
         let mut schema = LogSchema::default();
-        schema.set_source_type_key(Some(OwnedValuePath::try_from(nested_path).unwrap()));
+        schema.set_source_type_key(Some(OwnedTargetPath::event(owned_value_path!(
+            "a", "b", "c", "d"
+        ))));
         init_log_schema(schema, false);
 
         let namespace = LogNamespace::Legacy;
diff --git a/lib/vector-core/src/event/trace.rs b/lib/vector-core/src/event/trace.rs
@@ -7,7 +7,6 @@ use vector_common::{
     internal_event::TaggedEventsSent, json_size::JsonSize, request_metadata::GetEventCountTags,
     EventDataEq,
 };
-use vrl::path::{PathPrefix, ValuePath};
 
 use super::{
     BatchNotifier, EstimatedJsonEncodedSizeOf, EventFinalizer, EventFinalizers, EventMetadata,
@@ -95,12 +94,11 @@ impl TraceEvent {
 
     pub fn maybe_insert<'a, F: FnOnce() -> Value>(
         &mut self,
-        prefix: PathPrefix,
-        path: Option<impl ValuePath<'a>>,
+        path: Option<impl TargetPath<'a>>,
         value_callback: F,
     ) -> Option<Value> {
         if let Some(path) = path {
-            return self.0.insert((prefix, path), value_callback());
+            return self.0.insert(path, value_callback());
         }
         None
     }
diff --git a/src/sources/http_client/client.rs b/src/sources/http_client/client.rs
@@ -350,9 +350,9 @@ impl http_client::HttpClientContext for HttpClientContext {
                     }
                 }
                 Event::Trace(ref mut trace) => {
-                    if let Some(source_type_key) = log_schema().source_type_key_target_path() {
-                        trace.insert(source_type_key, Bytes::from(HttpClientConfig::NAME));
-                    }
+                    trace.maybe_insert(log_schema().source_type_key_target_path(), || {
+                        Bytes::from(HttpClientConfig::NAME).into()
+                    });
                 }
             }
         }
diff --git a/src/transforms/remap.rs b/src/transforms/remap.rs
@@ -487,7 +487,7 @@ where
                 }
             }
             Event::Trace(ref mut trace) => {
-                trace.maybe_insert(PathPrefix::Event, log_schema().metadata_key(), || {
+                trace.maybe_insert(log_schema().metadata_key_target_path(), || {
                     self.dropped_data(reason, error).into()
                 });
             }

Original file line number	Diff line number	Diff line change
`@@ -1,8 +1,7 @@`
`1`		`-use lookup::lookup_v2::{OptionalTargetPath, OptionalValuePath};`
	`1`	`+use lookup::lookup_v2::OptionalTargetPath;`
`2`	`2`	`use lookup::{OwnedTargetPath, OwnedValuePath};`
`3`	`3`	`use once_cell::sync::{Lazy, OnceCell};`
`4`	`4`	`use vector_config::configurable_component;`
`5`		`-use vrl::path::PathPrefix;`
`6`	`5`
`7`	`6`	`static LOG_SCHEMA: OnceCell<LogSchema> = OnceCell::new();`
`8`	`7`	`static LOG_SCHEMA_DEFAULT: Lazy<LogSchema> = Lazy::new(LogSchema::default);`
`@@ -74,7 +73,7 @@ pub struct LogSchema {`
`74`	`73`	`/// Generally, this field will be set by Vector to hold event-specific metadata, such as`
`75`	`74`	/// annotations by the `remap` transform when an error or abort is encountered.
`76`	`75`	`#[serde(default = "LogSchema::default_metadata_key")]`
`77`		`- metadata_key: OptionalValuePath,`
	`76`	`+ metadata_key: OptionalTargetPath,`
`78`	`77`	`}`
`79`	`78`
`80`	`79`	`impl Default for LogSchema {`
`@@ -106,8 +105,8 @@ impl LogSchema {`
`106`	`105`	`OptionalTargetPath::event(SOURCE_TYPE)`
`107`	`106`	`}`
`108`	`107`
`109`		`- fn default_metadata_key() -> OptionalValuePath {`
`110`		`- OptionalValuePath::new(METADATA)`
	`108`	`+ fn default_metadata_key() -> OptionalTargetPath {`
	`109`	`+ OptionalTargetPath::event(METADATA)`
`111`	`110`	`}`
`112`	`111`
`113`	`112`	`pub fn message_key(&self) -> Option<&OwnedValuePath> {`
`@@ -140,7 +139,7 @@ impl LogSchema {`
`140`	`139`	`}`
`141`	`140`
`142`	`141`	`pub fn metadata_key(&self) -> Option<&OwnedValuePath> {`
`143`		`- self.metadata_key.path.as_ref()`
	`142`	`+ self.metadata_key.as_ref().map(\|key\| &key.path)`
`144`	`143`	`}`
`145`	`144`
`146`	`145`	`pub fn message_key_target_path(&self) -> Option<&OwnedTargetPath> {`
`@@ -159,24 +158,28 @@ impl LogSchema {`
`159`	`158`	`self.source_type_key.as_ref()`
`160`	`159`	`}`
`161`	`160`
`162`		`- pub fn set_message_key(&mut self, path: Option<OwnedValuePath>) {`
`163`		`- self.message_key = OptionalTargetPath::from(PathPrefix::Event, path);`
	`161`	`+ pub fn metadata_key_target_path(&self) -> Option<&OwnedTargetPath> {`
	`162`	`+ self.metadata_key.as_ref()`
`164`	`163`	`}`
`165`	`164`
`166`		`- pub fn set_timestamp_key(&mut self, path: Option<OwnedValuePath>) {`
`167`		`- self.timestamp_key = OptionalTargetPath::from(PathPrefix::Event, path);`
	`165`	`+ pub fn set_message_key(&mut self, path: Option<OwnedTargetPath>) {`
	`166`	`+ self.message_key = OptionalTargetPath { path };`
`168`	`167`	`}`
`169`	`168`
`170`		`- pub fn set_host_key(&mut self, path: Option<OwnedValuePath>) {`
`171`		`- self.host_key = OptionalTargetPath::from(PathPrefix::Event, path);`
	`169`	`+ pub fn set_timestamp_key(&mut self, path: Option<OwnedTargetPath>) {`
	`170`	`+ self.timestamp_key = OptionalTargetPath { path };`
`172`	`171`	`}`
`173`	`172`
`174`		`- pub fn set_source_type_key(&mut self, path: Option<OwnedValuePath>) {`
`175`		`- self.source_type_key = OptionalTargetPath::from(PathPrefix::Event, path);`
	`173`	`+ pub fn set_host_key(&mut self, path: Option<OwnedTargetPath>) {`
	`174`	`+ self.host_key = OptionalTargetPath { path };`
`176`	`175`	`}`
`177`	`176`
`178`		`- pub fn set_metadata_key(&mut self, path: Option<OwnedValuePath>) {`
`179`		`- self.metadata_key = OptionalValuePath { path };`
	`177`	`+ pub fn set_source_type_key(&mut self, path: Option<OwnedTargetPath>) {`
	`178`	`+ self.source_type_key = OptionalTargetPath { path };`
	`179`	`+ }`
	`180`	`+`
	`181`	`+ pub fn set_metadata_key(&mut self, path: Option<OwnedTargetPath>) {`
	`182`	`+ self.metadata_key = OptionalTargetPath { path };`
`180`	`183`	`}`
`181`	`184`
`182`	`185`	/// Merge two `LogSchema` instances together.
`@@ -195,35 +198,35 @@ impl LogSchema {`
`195`	`198`	`{`
`196`	`199`	`errors.push("conflicting values for 'log_schema.host_key' found".to_owned());`
`197`	`200`	`} else {`
`198`		`- self.set_host_key(other.host_key().cloned());`
	`201`	`+ self.set_host_key(other.host_key_target_path().cloned());`
`199`	`202`	`}`
`200`	`203`	`if self.message_key() != LOG_SCHEMA_DEFAULT.message_key()`
`201`	`204`	`&& self.message_key() != other.message_key()`
`202`	`205`	`{`
`203`	`206`	`errors.push("conflicting values for 'log_schema.message_key' found".to_owned());`
`204`	`207`	`} else {`
`205`		`- self.set_message_key(other.message_key().cloned());`
	`208`	`+ self.set_message_key(other.message_key_target_path().cloned());`
`206`	`209`	`}`
`207`	`210`	`if self.timestamp_key() != LOG_SCHEMA_DEFAULT.timestamp_key()`
`208`	`211`	`&& self.timestamp_key() != other.timestamp_key()`
`209`	`212`	`{`
`210`	`213`	`errors.push("conflicting values for 'log_schema.timestamp_key' found".to_owned());`
`211`	`214`	`} else {`
`212`		`- self.set_timestamp_key(other.timestamp_key().cloned());`
	`215`	`+ self.set_timestamp_key(other.timestamp_key_target_path().cloned());`
`213`	`216`	`}`
`214`	`217`	`if self.source_type_key() != LOG_SCHEMA_DEFAULT.source_type_key()`
`215`	`218`	`&& self.source_type_key() != other.source_type_key()`
`216`	`219`	`{`
`217`	`220`	`errors.push("conflicting values for 'log_schema.source_type_key' found".to_owned());`
`218`	`221`	`} else {`
`219`		`- self.set_source_type_key(other.source_type_key().cloned());`
	`222`	`+ self.set_source_type_key(other.source_type_key_target_path().cloned());`
`220`	`223`	`}`
`221`	`224`	`if self.metadata_key() != LOG_SCHEMA_DEFAULT.metadata_key()`
`222`	`225`	`&& self.metadata_key() != other.metadata_key()`
`223`	`226`	`{`
`224`	`227`	`errors.push("conflicting values for 'log_schema.metadata_key' found".to_owned());`
`225`	`228`	`} else {`
`226`		`- self.set_metadata_key(other.metadata_key().cloned());`
	`229`	`+ self.set_metadata_key(other.metadata_key_target_path().cloned());`
`227`	`230`	`}`
`228`	`231`	`}`
`229`	`232`
Original file line number	Diff line number	Diff line change
`@@ -350,9 +350,9 @@ impl http_client::HttpClientContext for HttpClientContext {`
`350`	`350`	`}`
`351`	`351`	`}`
`352`	`352`	`Event::Trace(ref mut trace) => {`
`353`		`- if let Some(source_type_key) = log_schema().source_type_key_target_path() {`
`354`		`- trace.insert(source_type_key, Bytes::from(HttpClientConfig::NAME));`
`355`		`- }`
	`353`	`+ trace.maybe_insert(log_schema().source_type_key_target_path(), \|\| {`
	`354`	`+ Bytes::from(HttpClientConfig::NAME).into()`
	`355`	`+ });`
`356`	`356`	`}`
`357`	`357`	`}`
`358`	`358`	`}`
Original file line number	Diff line number	Diff line change
`@@ -487,7 +487,7 @@ where`
`487`	`487`	`}`
`488`	`488`	`}`
`489`	`489`	`Event::Trace(ref mut trace) => {`
`490`		`- trace.maybe_insert(PathPrefix::Event, log_schema().metadata_key(), \|\| {`
	`490`	`+ trace.maybe_insert(log_schema().metadata_key_target_path(), \|\| {`
`491`	`491`	`self.dropped_data(reason, error).into()`
`492`	`492`	`});`
`493`	`493`	`}`