feat(enrichment tables): add cuckoo filter to memory table

Cargo.toml

@@ -157,6 +157,7 @@ const-str = { version = "1.1.0", default-features = false }
 convert_case = { version = "0.8", default-features = false }
 criterion = "0.8"
 crossbeam-utils = { version = "0.8.21", default-features = false }
+cuckoo-clock = { version = "0.1.5" , default-features = false }
 darling = { version = "0.20.11", default-features = false, features = ["suggestions"] }
 dashmap = { version = "6.1.0", default-features = false }
 derivative = { version = "2.2.0", default-features = false }
@@ -383,6 +384,7 @@ chrono.workspace = true
 chrono-tz.workspace = true
 colored.workspace = true
 csv = { version = "1.3", default-features = false }
+cuckoo-clock = { workspace = true, optional = true }
 databend-client = { version = "0.28.0", default-features = false, features = ["rustls"], optional = true }
 derivative.workspace = true
 dirs-next = { version = "2.0.0", default-features = false, optional = true }
@@ -448,6 +450,7 @@ sqlx = { version = "0.8.6", default-features = false, features = ["derive", "pos
 stream-cancel = { version = "0.8.2", default-features = false }
 strip-ansi-escapes = { version = "0.2.1", default-features = false }
 syslog = { version = "6.1.1", default-features = false, optional = true }
+tempfile = { workspace = true, optional = true }
 tokio-postgres = { version = "0.7.18", default-features = false, features = ["runtime", "with-chrono-0_4"], optional = true }
 tokio-tungstenite = { workspace = true, features = ["connect"], optional = true }
 toml.workspace = true
@@ -659,7 +662,7 @@ gcp = ["dep:base64", "dep:goauth", "dep:smpl_jwt"]
 enrichment-tables = ["enrichment-tables-geoip", "enrichment-tables-mmdb", "enrichment-tables-memory"]
 enrichment-tables-geoip = ["dep:maxminddb"]
 enrichment-tables-mmdb = ["dep:maxminddb"]
-enrichment-tables-memory = ["dep:evmap", "dep:evmap-derive", "dep:thread_local"]
+enrichment-tables-memory = ["dep:cuckoo-clock", "dep:evmap", "dep:evmap-derive", "dep:thread_local", "dep:tempfile"]
 
 # Codecs
 codecs-arrow = ["dep:arrow", "dep:arrow-schema", "vector-lib/arrow"]

LICENSE-3rdparty.csv

@@ -209,6 +209,7 @@ csv,https://github.com/BurntSushi/rust-csv,Unlicense OR MIT,Andrew Gallant <jams
 csv-core,https://github.com/BurntSushi/rust-csv,Unlicense OR MIT,Andrew Gallant <jamslam@gmail.com>
 ctr,https://github.com/RustCrypto/block-modes,MIT OR Apache-2.0,RustCrypto Developers
 ctutils,https://github.com/RustCrypto/utils,Apache-2.0 OR MIT,RustCrypto Developers
+cuckoo-clock,https://github.com/Quad9DNS/cuckoo-clock,MIT,"John Todd <jtodd@quad9.net>, Ensar Sarajčić <dev@ensarsarajcic.com>"
 curl-sys,https://github.com/alexcrichton/curl-rust,MIT,Alex Crichton <alex@alexcrichton.com>
 curve25519-dalek,https://github.com/dalek-cryptography/curve25519-dalek/tree/main/curve25519-dalek,BSD-3-Clause,"Isis Lovecruft <isis@patternsinthevoid.net>, Henry de Valence <hdevalence@hdevalence.ca>"
 curve25519-dalek-derive,https://github.com/dalek-cryptography/curve25519-dalek,MIT OR Apache-2.0,The curve25519-dalek-derive Authors

changelog.d/25143_enrichment_table_memory_cuckoo_filter.feature.md

@@ -0,0 +1,3 @@
+Added cuckoo filter support for `memory` enrichment table, to provide an efficient way to store and check presence of keys with a low memory footprint at the cost of false positives.
+
+authors: esensar Quad9DNS

lib/vector-common/src/internal_event/metric_name.rs

@@ -98,6 +98,7 @@ pub enum CounterName {
     MemoryEnrichmentTableFlushesTotal,
     MemoryEnrichmentTableInsertionsTotal,
     MemoryEnrichmentTableReadsTotal,
+    MemoryEnrichmentTableRemovedTotal,
     MemoryEnrichmentTableTtlExpirations,
     ComponentCpuUsageNsTotal,
 }
@@ -364,6 +365,7 @@ impl CounterName {
                 "memory_enrichment_table_insertions_total"
             }
             Self::MemoryEnrichmentTableReadsTotal => "memory_enrichment_table_reads_total",
+            Self::MemoryEnrichmentTableRemovedTotal => "memory_enrichment_table_removed_total",
             Self::MemoryEnrichmentTableTtlExpirations => "memory_enrichment_table_ttl_expirations",
             Self::ComponentCpuUsageNsTotal => "component_cpu_usage_ns_total",
         }

src/enrichment_tables/memory/config.rs

@@ -19,13 +19,15 @@ use crate::{
     config::{
         EnrichmentTableConfig, SinkConfig, SinkContext, SourceConfig, SourceContext, SourceOutput,
     },
+    enrichment_tables::memory::cuckoo_table::{CuckooMemoryConfig, CuckooMemoryTable},
     sinks::Healthcheck,
     sources::Source,
 };
 
 /// Configuration for the `memory` enrichment table.
 #[configurable_component(enrichment_table("memory"))]
 #[derive(Clone)]
+#[serde(deny_unknown_fields)]
 pub struct MemoryConfig {
     /// TTL (time-to-live in seconds) is used to limit the lifetime of data stored in the cache.
     /// When TTL expires, data behind a specific key in the cache is removed.
@@ -43,6 +45,9 @@ pub struct MemoryConfig {
     /// Since every TTL scan makes its changes visible, only use this value
     /// if it is shorter than the `scan_interval`.
     ///
+    /// NOTE: For cuckoo filter, all writes are visible immediately. Flush interval still defines
+    /// when metrics for cuckoo filter are made visible.
+    ///
     /// By default, all writes are made visible immediately.
     #[serde(skip_serializing_if = "vector_lib::serde::is_default")]
     pub flush_interval: Option<u64>,
@@ -69,8 +74,16 @@ pub struct MemoryConfig {
     #[serde(default)]
     pub ttl_field: OptionalValuePath,
 
+    /// Set to make the table act as a probabilistic filter instead of storing original values. This
+    /// will prevent reading values from the table - found keys will have empty value.
+    #[configurable(derived)]
+    #[serde(default)]
+    pub filter: Option<TableFilter>,
+
     #[serde(skip)]
     memory: Arc<Mutex<Option<Box<Memory>>>>,
+    #[serde(skip)]
+    cuckoo: Arc<Mutex<Option<Box<CuckooMemoryTable>>>>,
 }
 
 /// Configuration for memory enrichment table source functionality.
@@ -102,6 +115,17 @@ pub struct MemorySourceConfig {
     pub source_key: String,
 }
 
+/// Configuration for memory enrichment table filter functionality.
+#[configurable_component]
+#[derive(Clone, Debug, PartialEq, Eq)]
+#[serde(deny_unknown_fields, rename_all = "snake_case", tag = "type")]
+pub enum TableFilter {
+    /// Cuckoo filter
+    ///
+    /// Supports removal by accepting null values for keys, as well as TTL and LRU.
+    Cuckoo(CuckooMemoryConfig),
+}
+
 impl PartialEq for MemoryConfig {
     fn eq(&self, other: &Self) -> bool {
         self.ttl == other.ttl
@@ -118,11 +142,13 @@ impl Default for MemoryConfig {
             scan_interval: default_scan_interval(),
             flush_interval: None,
             memory: Arc::new(Mutex::new(None)),
+            cuckoo: Arc::new(Mutex::new(None)),
             max_byte_size: None,
             log_namespace: None,
             source_config: None,
             internal_metrics: InternalMetricsConfig::default(),
             ttl_field: OptionalValuePath::none(),
+            filter: None,
         }
     }
 }
@@ -142,14 +168,39 @@ impl MemoryConfig {
             .get_or_insert_with(|| Box::new(Memory::new(self.clone())))
             .clone()
     }
+
+    pub(super) async fn get_or_build_cuckoo(&self) -> crate::Result<CuckooMemoryTable> {
+        let mut boxed_cuckoo = self.cuckoo.lock().await;
+        let Some(TableFilter::Cuckoo(cuckoo)) = &self.filter else {
+            panic!("No cuckoo");
+        };
+        if let Some(boxed_cuckoo) = boxed_cuckoo.as_ref() {
+            Ok(*boxed_cuckoo.clone())
+        } else {
+            Ok(*boxed_cuckoo
+                .insert(Box::new(CuckooMemoryTable::new(
+                    self.clone(),
+                    cuckoo.clone(),
+                )?))
+                .clone())
+        }
+    }
 }
 
 impl EnrichmentTableConfig for MemoryConfig {
     async fn build(
         &self,
         _globals: &crate::config::GlobalOptions,
     ) -> crate::Result<Box<dyn Table + Send + Sync>> {
-        Ok(Box::new(self.get_or_build_memory().await))
+        match &self.filter {
+            Some(TableFilter::Cuckoo(_)) => {
+                if self.source_config.is_some() {
+                    return Err("Source functionality is not supported for cuckoo filter".into());
+                }
+                Ok(Box::new(self.get_or_build_cuckoo().await?))
+            }
+            None => Ok(Box::new(self.get_or_build_memory().await)),
+        }
     }
 
     fn sink_config(
@@ -166,6 +217,10 @@ impl EnrichmentTableConfig for MemoryConfig {
         let Some(source_config) = &self.source_config else {
             return None;
         };
+        // Filters can't be used as a source
+        if self.filter.is_some() {
+            return None;
+        }
         Some((
             source_config.source_key.clone().into(),
             Box::new(self.clone()),
@@ -177,7 +232,12 @@ impl EnrichmentTableConfig for MemoryConfig {
 #[typetag::serde(name = "memory_enrichment_table")]
 impl SinkConfig for MemoryConfig {
     async fn build(&self, _cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> {
-        let sink = VectorSink::from_event_streamsink(self.get_or_build_memory().await);
+        let sink = match &self.filter {
+            Some(TableFilter::Cuckoo(_)) => {
+                VectorSink::from_event_streamsink(self.get_or_build_cuckoo().await?)
+            }
+            None => VectorSink::from_event_streamsink(self.get_or_build_memory().await),
+        };
 
         Ok((sink, future::ok(()).boxed()))
     }