Merge remote-tracking branch 'origin/master' into dependabot/cargo/li…

…bc-0.2.117

Cargo.toml

@@ -213,7 +213,6 @@ vrl-stdlib = { path = "lib/vrl/stdlib" }
 lookup = { path = "lib/lookup" }
 
 # External libs
-arc-swap = { version = "1.5", default-features = false }
 async-compression = { version = "0.3.7", default-features = false, features = ["tokio", "gzip", "zstd"] }
 avro-rs = { version = "0.13.0", default-features = false, optional = true }
 base64 = { version = "0.13.0", default-features = false, optional = true }

lib/datadog/grok/build.rs

@@ -9,14 +9,13 @@ use std::{
 };
 
 fn main() {
+    read_grok_patterns();
+
+    println!("cargo:rerun-if-changed=src/parser.lalrpop");
     lalrpop::Configuration::new()
         .always_use_colors()
         .process_current_dir()
         .unwrap();
-
-    println!("cargo:rerun-if-changed=src/parser.lalrpop");
-
-    read_grok_patterns();
 }
 
 /// Reads grok patterns defined in the `patterns` folder into the static `PATTERNS` variable

lib/vector-common/Cargo.toml

@@ -49,5 +49,6 @@ metrics = { version = "0.17.0", default-features = false, features = ["std"] }
 nom = { version = "7", optional = true }
 serde_json = { version = "1.0.78", default-features = false, features = ["std", "raw_value"] }
 serde = { version = "1.0.136", optional = true, features = ["derive"] }
+smallvec = { version = "1", default-features = false }
 snafu = { version = "0.7", optional = true }
 tracing = { version = "0.1.29", default-features = false }

lib/vector-common/src/byte_size_of.rs

@@ -4,6 +4,7 @@ use std::{
 };
 
 use serde_json::{value::RawValue, Value};
+use smallvec::SmallVec;
 
 pub trait ByteSizeOf {
     /// Returns the in-memory size of this type
@@ -62,6 +63,15 @@ where
     }
 }
 
+impl<A: smallvec::Array> ByteSizeOf for SmallVec<A>
+where
+    A::Item: ByteSizeOf,
+{
+    fn allocated_bytes(&self) -> usize {
+        self.iter().map(ByteSizeOf::size_of).sum()
+    }
+}
+
 impl<T> ByteSizeOf for &[T]
 where
     T: ByteSizeOf,

lib/vector-core/Cargo.toml

@@ -38,7 +38,7 @@ pin-project = { version = "1.0.10", default-features = false }
 prost = { version = "0.9", default-features = false, features = ["std"] }
 prost-types = { version = "0.9", default-features = false }
 regex = { version = "1.5.4", default-features = false, features = ["std", "perf"] }
-serde = { version = "1.0.136", default-features = false, features = ["derive"] }
+serde = { version = "1.0.136", default-features = false, features = ["derive", "rc"] }
 serde_json = { version = "1.0.78", default-features = false }
 snafu = { version = "0.7.0", default-features = false }
 substring = { version = "1.4", default-features = false }

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

@@ -24,9 +24,9 @@ use crate::{config::log_schema, event::MaybeAsLogMut, ByteSizeOf};
 #[derive(Clone, Debug, Getters, MutGetters, PartialEq, PartialOrd, Derivative, Deserialize)]
 pub struct LogEvent {
     // **IMPORTANT:** Due to numerous legacy reasons this **must** be a Map variant.
-    #[derivative(Default(value = "Value::from(BTreeMap::default())"))]
+    #[derivative(Default(value = "Arc::new(Value::from(BTreeMap::default()))"))]
     #[serde(flatten)]
-    fields: Value,
+    fields: Arc<Value>,
 
     #[getset(get = "pub", get_mut = "pub")]
     #[serde(skip)]
@@ -36,7 +36,7 @@ pub struct LogEvent {
 impl Default for LogEvent {
     fn default() -> Self {
         Self {
-            fields: Value::Map(BTreeMap::new()),
+            fields: Arc::new(Value::Map(BTreeMap::new())),
             metadata: EventMetadata::default(),
         }
     }
@@ -58,25 +58,30 @@ impl LogEvent {
     #[must_use]
     pub fn new_with_metadata(metadata: EventMetadata) -> Self {
         Self {
-            fields: Value::Map(Default::default()),
+            fields: Arc::new(Value::Map(Default::default())),
             metadata,
         }
     }
 
     ///  Create a `LogEvent` into a tuple of its components
     pub fn from_parts(map: BTreeMap<String, Value>, metadata: EventMetadata) -> Self {
         let fields = Value::Map(map);
-        Self { fields, metadata }
+        Self {
+            fields: Arc::new(fields),
+            metadata,
+        }
     }
 
     /// Convert a `LogEvent` into a tuple of its components
     ///
     /// # Panics
     ///
     /// Panics if the fields of the `LogEvent` are not a `Value::Map`.
-    pub fn into_parts(self) -> (BTreeMap<String, Value>, EventMetadata) {
+    pub fn into_parts(mut self) -> (BTreeMap<String, Value>, EventMetadata) {
+        Arc::make_mut(&mut self.fields);
         (
-            self.fields
+            Arc::try_unwrap(self.fields)
+                .expect("already cloned")
                 .into_map()
                 .unwrap_or_else(|| unreachable!("fields must be a map")),
             self.metadata,
@@ -158,7 +163,10 @@ impl LogEvent {
         K: AsRef<str> + Into<String> + PartialEq + Display,
     {
         if from_key != to_key {
-            if let Some(val) = self.fields.as_map_mut().remove(from_key.as_ref()) {
+            if let Some(val) = Arc::make_mut(&mut self.fields)
+                .as_map_mut()
+                .remove(from_key.as_ref())
+            {
                 self.insert_flat(to_key, val);
             }
         }
@@ -198,7 +206,7 @@ impl LogEvent {
 
     #[instrument(level = "trace", skip(self))]
     pub fn keys<'a>(&'a self) -> impl Iterator<Item = String> + 'a {
-        match &self.fields {
+        match self.fields.as_ref() {
             Value::Map(map) => util::log::keys(map),
             _ => unreachable!(),
         }
@@ -216,15 +224,15 @@ impl LogEvent {
 
     #[instrument(level = "trace", skip(self))]
     pub fn as_map(&self) -> &BTreeMap<String, Value> {
-        match &self.fields {
+        match self.fields.as_ref() {
             Value::Map(map) => map,
             _ => unreachable!(),
         }
     }
 
     #[instrument(level = "trace", skip(self))]
     pub fn as_map_mut(&mut self) -> &mut BTreeMap<String, Value> {
-        match self.fields {
+        match Arc::make_mut(&mut self.fields) {
             Value::Map(ref mut map) => map,
             _ => unreachable!(),
         }
@@ -324,15 +332,16 @@ impl From<String> for LogEvent {
 impl From<BTreeMap<String, Value>> for LogEvent {
     fn from(map: BTreeMap<String, Value>) -> Self {
         LogEvent {
-            fields: Value::Map(map),
+            fields: Arc::new(Value::Map(map)),
             metadata: EventMetadata::default(),
         }
     }
 }
 
 impl From<LogEvent> for BTreeMap<String, Value> {
-    fn from(event: LogEvent) -> BTreeMap<String, Value> {
-        match event.fields {
+    fn from(mut event: LogEvent) -> BTreeMap<String, Value> {
+        Arc::make_mut(&mut event.fields);
+        match Arc::try_unwrap(event.fields).expect("already cloned") {
             Value::Map(map) => map,
             _ => unreachable!(),
         }
@@ -342,7 +351,7 @@ impl From<LogEvent> for BTreeMap<String, Value> {
 impl From<HashMap<String, Value>> for LogEvent {
     fn from(map: HashMap<String, Value>) -> Self {
         LogEvent {
-            fields: map.into_iter().collect(),
+            fields: Arc::new(map.into_iter().collect()),
             metadata: EventMetadata::default(),
         }
     }
@@ -380,7 +389,7 @@ impl TryInto<serde_json::Value> for LogEvent {
     type Error = crate::Error;
 
     fn try_into(self) -> Result<serde_json::Value, Self::Error> {
-        Ok(serde_json::to_value(self.fields)?)
+        Ok(serde_json::to_value(self.fields.as_ref())?)
     }
 }
 

scripts/integration/docker-compose.aws.yml

@@ -1,5 +1,3 @@
-version: "3"
-
 services:
   mock-ec2-metadata:
     image: timberiodev/mock-ec2-metadata:latest
@@ -74,3 +72,4 @@ volumes:
 
 networks:
   backend: {}
+

soaks/bin/detect_regressions

@@ -79,7 +79,7 @@ print("")
 print(results.to_markdown(index=False, tablefmt='github'))
 
 p_value_violation = results['p-value'] < args.p_value
-drift_filter = results['Δ mean %'].abs() > args.mean_drift_percentage
+drift_filter = results['Δ mean %'] > args.mean_drift_percentage
 declared_erratic = results.experiment.isin(known_erratic_soaks)
 
 changes = results[p_value_violation & drift_filter & ~declared_erratic]