chore(deps): Swap out bloom crate for bloomy (#17911)

Signed-off-by: Jesse Szwedko <[email protected]>

---------

Signed-off-by: Jesse Szwedko <[email protected]>

Author: jszwedko

Cargo.lock

@@ -233,8 +233,8 @@ arc-swap = { version = "1.6", default-features = false, optional = true }
 async-compression = { version = "0.4.0", default-features = false, features = ["tokio", "gzip", "zstd"], optional = true }
 apache-avro = { version = "0.14.0", default-features = false, optional = true }
 axum = { version = "0.6.18", default-features = false }
-base64 = { version = "0.21.1", default-features = false, optional = true }
-bloom = { version = "0.3.2", default-features = false, optional = true }
+base64 = { version = "0.21.2", default-features = false, optional = true }
+bloomy  = { version = "1.2.0", default-features = false, optional = true }
 bollard = { version = "0.14.0", default-features = false, features = ["ssl", "chrono"], optional = true }
 bytes = { version = "1.4.0", default-features = false, features = ["serde"] }
 bytesize = { version = "1.2.0", default-features = false }
@@ -587,7 +587,7 @@ transforms-reduce = []
 transforms-remap = []
 transforms-route = []
 transforms-sample = []
-transforms-tag_cardinality_limit = ["dep:bloom", "dep:hashbrown"]
+transforms-tag_cardinality_limit = ["dep:bloomy", "dep:hashbrown"]
 transforms-throttle = ["dep:governor"]
 
 # Sinks

Cargo.toml

@@ -80,7 +80,7 @@ bitvec,https://github.com/bitvecto-rs/bitvec,MIT,The bitvec Authors
 block-buffer,https://github.com/RustCrypto/utils,MIT OR Apache-2.0,RustCrypto Developers
 block-padding,https://github.com/RustCrypto/utils,MIT OR Apache-2.0,RustCrypto Developers
 blocking,https://github.com/smol-rs/blocking,Apache-2.0 OR MIT,Stjepan Glavina <[email protected]>
-bloom,https://github.com/nicklan/bloom-rs,GPL-2.0,Nick Lanham <[email protected]>
+bloomy,https://docs.rs/bloomy/,MIT,"Aleksandr Bezobchuk <[email protected]>, Alexis Sellier <[email protected]>"
 bollard,https://github.com/fussybeaver/bollard,Apache-2.0,Bollard contributors
 borsh,https://github.com/near/borsh-rs,MIT OR Apache-2.0,Near Inc <[email protected]>
 borsh-derive,https://github.com/nearprotocol/borsh,Apache-2.0,Near Inc <[email protected]>

LICENSE-3rdparty.csv

@@ -18,7 +18,7 @@ use vector_core::config::LogNamespace;
 pub struct TagCardinalityLimitConfig {
     /// How many distinct values to accept for any given key.
     #[serde(default = "default_value_limit")]
-    pub value_limit: u32,
+    pub value_limit: usize,
 
     #[configurable(derived)]
     #[serde(default = "default_limit_exceeded_action")]
@@ -80,7 +80,7 @@ const fn default_limit_exceeded_action() -> LimitExceededAction {
     LimitExceededAction::DropTag
 }
 
-const fn default_value_limit() -> u32 {
+const fn default_value_limit() -> usize {
     500
 }
 

src/transforms/tag_cardinality_limit/config.rs

@@ -55,11 +55,11 @@ impl TagCardinalityLimit {
         }
 
         // Tag value not yet part of the accepted set.
-        if tag_value_set.len() < self.config.value_limit as usize {
+        if tag_value_set.len() < self.config.value_limit {
             // accept the new value
             tag_value_set.insert(value.clone());
 
-            if tag_value_set.len() == self.config.value_limit as usize {
+            if tag_value_set.len() == self.config.value_limit {
                 emit!(TagCardinalityValueLimitReached { key });
             }
 
@@ -76,7 +76,7 @@ impl TagCardinalityLimit {
         self.accepted_tags
             .get(key)
             .map(|value_set| {
-                !value_set.contains(value) && value_set.len() >= self.config.value_limit as usize
+                !value_set.contains(value) && value_set.len() >= self.config.value_limit
             })
             .unwrap_or(false)
     }

src/transforms/tag_cardinality_limit/mod.rs

@@ -1,19 +1,18 @@
 use crate::event::metric::TagValueSet;
 use crate::transforms::tag_cardinality_limit::config::Mode;
-use bloom::{BloomFilter, ASMS};
+use bloomy::BloomFilter;
 use std::collections::HashSet;
 use std::fmt;
 
 /// Container for storing the set of accepted values for a given tag key.
 #[derive(Debug)]
 pub struct AcceptedTagValueSet {
     storage: TagValueSetStorage,
-    num_elements: usize,
 }
 
 enum TagValueSetStorage {
     Set(HashSet<TagValueSet>),
-    Bloom(BloomFilter),
+    Bloom(BloomFilter<TagValueSet>),
 }
 
 impl fmt::Debug for TagValueSetStorage {
@@ -26,40 +25,37 @@ impl fmt::Debug for TagValueSetStorage {
 }
 
 impl AcceptedTagValueSet {
-    pub fn new(value_limit: u32, mode: &Mode) -> Self {
+    pub fn new(value_limit: usize, mode: &Mode) -> Self {
         let storage = match &mode {
-            Mode::Exact => TagValueSetStorage::Set(HashSet::with_capacity(value_limit as usize)),
+            Mode::Exact => TagValueSetStorage::Set(HashSet::with_capacity(value_limit)),
             Mode::Probabilistic(config) => {
                 let num_bits = config.cache_size_per_key / 8; // Convert bytes to bits
-                let num_hashes = bloom::optimal_num_hashes(num_bits, value_limit);
-                TagValueSetStorage::Bloom(BloomFilter::with_size(num_bits, num_hashes))
+                TagValueSetStorage::Bloom(BloomFilter::with_size(num_bits))
             }
         };
-        Self {
-            storage,
-            num_elements: 0,
-        }
+        Self { storage }
     }
 
     pub fn contains(&self, value: &TagValueSet) -> bool {
         match &self.storage {
             TagValueSetStorage::Set(set) => set.contains(value),
-            TagValueSetStorage::Bloom(bloom) => bloom.contains(&value),
+            TagValueSetStorage::Bloom(bloom) => bloom.contains(value),
         }
     }
 
-    pub const fn len(&self) -> usize {
-        self.num_elements
+    pub fn len(&self) -> usize {
+        match &self.storage {
+            TagValueSetStorage::Set(set) => set.len(),
+            TagValueSetStorage::Bloom(bloom) => bloom.count(),
+        }
     }
 
-    pub fn insert(&mut self, value: TagValueSet) -> bool {
-        let inserted = match &mut self.storage {
-            TagValueSetStorage::Set(set) => set.insert(value),
+    pub fn insert(&mut self, value: TagValueSet) {
+        match &mut self.storage {
+            TagValueSetStorage::Set(set) => {
+                set.insert(value);
+            }
             TagValueSetStorage::Bloom(bloom) => bloom.insert(&value),
         };
-        if inserted {
-            self.num_elements += 1
-        }
-        inserted
     }
 }

src/transforms/tag_cardinality_limit/tag_value_set.rs

@@ -31,7 +31,7 @@ fn make_metric(tags: MetricTags) -> Event {
 }
 
 const fn make_transform_hashset(
-    value_limit: u32,
+    value_limit: usize,
     limit_exceeded_action: LimitExceededAction,
 ) -> TagCardinalityLimitConfig {
     TagCardinalityLimitConfig {
@@ -42,7 +42,7 @@ const fn make_transform_hashset(
 }
 
 const fn make_transform_bloom(
-    value_limit: u32,
+    value_limit: usize,
     limit_exceeded_action: LimitExceededAction,
 ) -> TagCardinalityLimitConfig {
     TagCardinalityLimitConfig {