vector/enrichment_tables/memory/
config.rs

1use std::{num::NonZeroU64, sync::Arc};
2
3use async_trait::async_trait;
4use futures::{FutureExt, future};
5use tokio::sync::Mutex;
6use vector_lib::{
7    config::{AcknowledgementsConfig, DataType, Input, LogNamespace},
8    configurable::configurable_component,
9    enrichment::Table,
10    id::ComponentKey,
11    lookup::lookup_v2::OptionalValuePath,
12    schema::{self},
13    sink::VectorSink,
14};
15use vrl::{path::OwnedTargetPath, value::Kind};
16
17use super::{Memory, internal_events::InternalMetricsConfig, source::EXPIRED_ROUTE};
18use crate::{
19    config::{
20        EnrichmentTableConfig, SinkConfig, SinkContext, SourceConfig, SourceContext, SourceOutput,
21    },
22    enrichment_tables::memory::{
23        bloom_table::{BloomMemoryConfig, BloomMemoryTable},
24        cuckoo_table::{CuckooMemoryConfig, CuckooMemoryTable},
25    },
26    sinks::Healthcheck,
27    sources::Source,
28};
29
30/// Configuration for the `memory` enrichment table.
31#[configurable_component(enrichment_table("memory"))]
32#[derive(Clone)]
33#[serde(deny_unknown_fields)]
34pub struct MemoryConfig {
35    /// TTL (time-to-live in seconds) is used to limit the lifetime of data stored in the cache.
36    /// When TTL expires, data behind a specific key in the cache is removed.
37    /// TTL is reset when the key is replaced.
38    #[serde(default = "default_ttl")]
39    pub ttl: u64,
40    /// The scan interval used to look for expired records. This is provided
41    /// as an optimization to ensure that TTL is updated, but without doing
42    /// too many cache scans.
43    #[serde(default = "default_scan_interval")]
44    pub scan_interval: NonZeroU64,
45    /// The interval used for making writes visible in the table.
46    /// Longer intervals might get better performance,
47    /// but there is a longer delay before the data is visible in the table.
48    /// Since every TTL scan makes its changes visible, only use this value
49    /// if it is shorter than the `scan_interval`.
50    ///
51    /// NOTE: For cuckoo filter, all writes are visible immediately. Flush interval still defines
52    /// when metrics for cuckoo filter are made visible.
53    ///
54    /// By default, all writes are made visible immediately.
55    #[serde(skip_serializing_if = "vector_lib::serde::is_default")]
56    pub flush_interval: Option<NonZeroU64>,
57    /// Maximum size of the table in bytes. All insertions that make
58    /// this table bigger than the maximum size are rejected.
59    ///
60    /// By default, there is no size limit.
61    #[serde(skip_serializing_if = "vector_lib::serde::is_default")]
62    pub max_byte_size: Option<u64>,
63    /// The namespace to use for logs. This overrides the global setting.
64    #[configurable(metadata(docs::hidden))]
65    #[serde(default)]
66    pub log_namespace: Option<bool>,
67    /// Configuration of internal metrics
68    #[configurable(derived)]
69    #[serde(default)]
70    pub internal_metrics: InternalMetricsConfig,
71    /// Configuration for source functionality.
72    #[configurable(derived)]
73    #[serde(skip_serializing_if = "vector_lib::serde::is_default")]
74    pub source_config: Option<MemorySourceConfig>,
75    /// Field in the incoming value used as the TTL override.
76    #[configurable(derived)]
77    #[serde(default)]
78    pub ttl_field: OptionalValuePath,
79    /// Behavior for memory table state on configuration reload.
80    #[configurable(derived)]
81    #[serde(default)]
82    pub reload_behavior: ReloadBehavior,
83
84    /// Set to make the table act as a probabilistic filter instead of storing original values. This
85    /// will prevent reading values from the table - found keys will have empty value.
86    #[configurable(derived)]
87    #[serde(default)]
88    pub filter: Option<TableFilter>,
89
90    #[serde(skip)]
91    memory: Arc<Mutex<Option<Box<Memory>>>>,
92    #[serde(skip)]
93    cuckoo: Arc<Mutex<Option<Box<CuckooMemoryTable>>>>,
94    #[serde(skip)]
95    bloom: Arc<Mutex<Option<Box<BloomMemoryTable>>>>,
96}
97
98/// Behavior for memory enrichment table state on configuration reload.
99#[configurable_component]
100#[derive(Clone, Default)]
101#[serde(rename_all = "kebab-case")]
102pub enum ReloadBehavior {
103    /// Always clear state on configuration reload.
104    #[default]
105    ClearState,
106    /// Try to preserve state when possible.
107    PreserveState,
108}
109
110/// Configuration for memory enrichment table source functionality.
111#[configurable_component]
112#[derive(Clone, Debug, PartialEq, Eq)]
113#[serde(deny_unknown_fields)]
114pub struct MemorySourceConfig {
115    /// Interval for exporting all data from the table when used as a source.
116    #[serde(skip_serializing_if = "vector_lib::serde::is_default")]
117    pub export_interval: Option<NonZeroU64>,
118    /// Batch size for data exporting. Used to prevent exporting entire table at
119    /// once and blocking the system.
120    ///
121    /// By default, batches are not used and entire table is exported.
122    #[serde(skip_serializing_if = "vector_lib::serde::is_default")]
123    pub export_batch_size: Option<u64>,
124    /// If set to true, all data will be removed from cache after exporting.
125    /// Only valid if used as a source and export_interval > 0
126    ///
127    /// By default, export will not remove data from cache
128    #[serde(default = "crate::serde::default_false")]
129    pub remove_after_export: bool,
130    /// Set to true to export expired items via the `expired` output port.
131    /// Expired items ignore other settings and are exported as they are flushed from the table.
132    #[serde(default = "crate::serde::default_false")]
133    pub export_expired_items: bool,
134    /// Key to use for this component when used as a source. This must be different from the
135    /// component key.
136    pub source_key: String,
137}
138
139/// Configuration for memory enrichment table filter functionality.
140#[configurable_component]
141#[derive(Clone, Debug, PartialEq, Eq)]
142#[serde(deny_unknown_fields, rename_all = "snake_case", tag = "type")]
143#[configurable(metadata(docs::enum_tag_description = "The probabilistic filter to use."))]
144pub enum TableFilter {
145    /// Cuckoo filter
146    ///
147    /// Supports removal by accepting null values for keys, as well as TTL and LRU.
148    Cuckoo(CuckooMemoryConfig),
149    /// Bloom filter
150    ///
151    /// Only supports insertion and presence check, no TTL
152    Bloom(BloomMemoryConfig),
153}
154
155impl PartialEq for MemoryConfig {
156    fn eq(&self, other: &Self) -> bool {
157        self.ttl == other.ttl
158            && self.scan_interval == other.scan_interval
159            && self.flush_interval == other.flush_interval
160    }
161}
162impl Eq for MemoryConfig {}
163
164impl Default for MemoryConfig {
165    fn default() -> Self {
166        Self {
167            ttl: default_ttl(),
168            scan_interval: default_scan_interval(),
169            flush_interval: None,
170            memory: Arc::new(Mutex::new(None)),
171            cuckoo: Arc::new(Mutex::new(None)),
172            bloom: Arc::new(Mutex::new(None)),
173            max_byte_size: None,
174            log_namespace: None,
175            source_config: None,
176            internal_metrics: InternalMetricsConfig::default(),
177            ttl_field: OptionalValuePath::none(),
178            reload_behavior: Default::default(),
179            filter: None,
180        }
181    }
182}
183
184const fn default_ttl() -> u64 {
185    600
186}
187
188const fn default_scan_interval() -> NonZeroU64 {
189    unsafe { NonZeroU64::new_unchecked(30) }
190}
191
192impl MemoryConfig {
193    pub(super) async fn get_or_build_memory(
194        &self,
195        prev_state: Option<Box<dyn std::any::Any + Send + Sync>>,
196    ) -> Memory {
197        let mut boxed_memory = self.memory.lock().await;
198        *boxed_memory
199            .get_or_insert_with(|| {
200                if let Some(prev) = prev_state {
201                    Box::new(Memory::from_previous_state(self.clone(), prev))
202                } else {
203                    Box::new(Memory::new(self.clone()))
204                }
205            })
206            .clone()
207    }
208
209    pub(super) async fn get_or_build_cuckoo(
210        &self,
211        prev_state: Option<Box<dyn std::any::Any + Send + Sync>>,
212    ) -> crate::Result<CuckooMemoryTable> {
213        let Some(TableFilter::Cuckoo(cuckoo)) = &self.filter else {
214            panic!("No cuckoo");
215        };
216        let mut boxed_cuckoo = self.cuckoo.lock().await;
217        if let Some(boxed_cuckoo) = boxed_cuckoo.as_ref() {
218            Ok(*boxed_cuckoo.clone())
219        } else {
220            Ok(*boxed_cuckoo
221                .insert(if let Some(prev) = prev_state {
222                    Box::new(CuckooMemoryTable::from_previous_state(
223                        self.clone(),
224                        cuckoo.clone(),
225                        prev,
226                    )?)
227                } else {
228                    Box::new(CuckooMemoryTable::new(self.clone(), cuckoo.clone())?)
229                })
230                .clone())
231        }
232    }
233
234    pub(super) async fn get_or_build_bloom(
235        &self,
236        prev_state: Option<Box<dyn std::any::Any + Send + Sync>>,
237    ) -> crate::Result<BloomMemoryTable> {
238        let mut boxed_bloom = self.bloom.lock().await;
239        let Some(TableFilter::Bloom(bloom)) = &self.filter else {
240            panic!("No bloom");
241        };
242        if let Some(boxed_bloom) = boxed_bloom.as_ref() {
243            Ok(*boxed_bloom.clone())
244        } else {
245            Ok(*boxed_bloom
246                .insert(if let Some(prev) = prev_state {
247                    Box::new(BloomMemoryTable::from_previous_state(
248                        self.clone(),
249                        bloom.clone(),
250                        prev,
251                    )?)
252                } else {
253                    Box::new(BloomMemoryTable::new(self.clone(), bloom.clone())?)
254                })
255                .clone())
256        }
257    }
258}
259
260impl EnrichmentTableConfig for MemoryConfig {
261    async fn build(
262        &self,
263        _globals: &crate::config::GlobalOptions,
264        prev_state: Option<Box<dyn std::any::Any + Send + Sync>>,
265    ) -> crate::Result<Box<dyn Table + Send + Sync>> {
266        match &self.filter {
267            Some(TableFilter::Cuckoo(_)) => {
268                if self.source_config.is_some() {
269                    return Err("Source functionality is not supported for cuckoo filter".into());
270                }
271                Ok(Box::new(self.get_or_build_cuckoo(prev_state).await?))
272            }
273            Some(TableFilter::Bloom(_)) => {
274                if self.source_config.is_some() {
275                    return Err("Source functionality is not supported for bloom filter".into());
276                }
277                if self.ttl_field.path.is_some() || self.ttl != default_ttl() {
278                    return Err("TTL functionality is not supported for bloom filter.".into());
279                }
280                if self.scan_interval != default_scan_interval() {
281                    return Err("`scan_interval` has no effect for bloom filter.".into());
282                }
283                Ok(Box::new(self.get_or_build_bloom(prev_state).await?))
284            }
285            None => Ok(Box::new(self.get_or_build_memory(prev_state).await)),
286        }
287    }
288
289    fn wants_previous_state(&self) -> bool {
290        matches!(self.reload_behavior, ReloadBehavior::PreserveState)
291    }
292
293    fn sink_config(
294        &self,
295        default_key: &ComponentKey,
296    ) -> Option<(ComponentKey, Box<dyn SinkConfig>)> {
297        Some((default_key.clone(), Box::new(self.clone())))
298    }
299
300    fn source_config(
301        &self,
302        _default_key: &ComponentKey,
303    ) -> Option<(ComponentKey, Box<dyn SourceConfig>)> {
304        let Some(source_config) = &self.source_config else {
305            return None;
306        };
307        // Filters can't be used as a source
308        if self.filter.is_some() {
309            return None;
310        }
311        Some((
312            source_config.source_key.clone().into(),
313            Box::new(self.clone()),
314        ))
315    }
316}
317
318#[async_trait]
319#[typetag::serde(name = "memory_enrichment_table")]
320impl SinkConfig for MemoryConfig {
321    async fn build(&self, _cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> {
322        let sink = match &self.filter {
323            Some(TableFilter::Cuckoo(_)) => {
324                VectorSink::from_event_streamsink(self.get_or_build_cuckoo(None).await?)
325            }
326            Some(TableFilter::Bloom(_)) => {
327                VectorSink::from_event_streamsink(self.get_or_build_bloom(None).await?)
328            }
329            None => VectorSink::from_event_streamsink(self.get_or_build_memory(None).await),
330        };
331
332        Ok((sink, future::ok(()).boxed()))
333    }
334
335    fn input(&self) -> Input {
336        Input::log()
337    }
338
339    fn acknowledgements(&self) -> &AcknowledgementsConfig {
340        &AcknowledgementsConfig::DEFAULT
341    }
342}
343
344#[async_trait]
345#[typetag::serde(name = "memory_enrichment_table")]
346impl SourceConfig for MemoryConfig {
347    async fn build(&self, cx: SourceContext) -> crate::Result<Source> {
348        let memory = self.get_or_build_memory(None).await;
349
350        let log_namespace = cx.log_namespace(self.log_namespace);
351
352        Ok(Box::pin(
353            memory.as_source(cx.shutdown, cx.out, log_namespace).run(),
354        ))
355    }
356
357    fn outputs(&self, global_log_namespace: LogNamespace) -> Vec<SourceOutput> {
358        let log_namespace = global_log_namespace.merge(self.log_namespace);
359        let schema_definition = match log_namespace {
360            LogNamespace::Legacy => schema::Definition::default_legacy_namespace(),
361            LogNamespace::Vector => {
362                schema::Definition::new_with_default_metadata(Kind::any_object(), [log_namespace])
363                    .with_meaning(OwnedTargetPath::event_root(), "message")
364            }
365        }
366        .with_standard_vector_source_metadata();
367
368        if self
369            .source_config
370            .as_ref()
371            .map(|c| c.export_expired_items)
372            .unwrap_or_default()
373        {
374            vec![
375                SourceOutput::new_maybe_logs(DataType::Log, schema_definition.clone()),
376                SourceOutput::new_maybe_logs(DataType::Log, schema_definition)
377                    .with_port(EXPIRED_ROUTE),
378            ]
379        } else {
380            vec![SourceOutput::new_maybe_logs(
381                DataType::Log,
382                schema_definition,
383            )]
384        }
385    }
386
387    fn can_acknowledge(&self) -> bool {
388        false
389    }
390}
391
392impl std::fmt::Debug for MemoryConfig {
393    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
394        f.debug_struct("MemoryConfig")
395            .field("ttl", &self.ttl)
396            .field("scan_interval", &self.scan_interval)
397            .field("flush_interval", &self.flush_interval)
398            .field("max_byte_size", &self.max_byte_size)
399            .finish()
400    }
401}
402
403impl_generate_config_from_default!(MemoryConfig);
vector/enrichment_tables/memory/config.rs

vector/enrichment_tables/memory/
config.rs
