feat(enriching): add memory enrichment table

Cargo.toml

@@ -310,6 +310,8 @@ dirs-next = { version = "2.0.0", default-features = false, optional = true }
 dyn-clone = { version = "1.0.17", default-features = false }
 encoding_rs = { version = "0.8.35", default-features = false, features = ["serde"] }
 enum_dispatch = { version = "0.3.13", default-features = false }
+evmap = { version = "10.0.2", default-features = false, optional = true }
+evmap-derive = { version = "0.2.0", default-features = false, optional = true }
 exitcode = { version = "1.1.2", default-features = false }
 flate2.workspace = true
 futures-util = { version = "0.3.29", default-features = false }
@@ -371,6 +373,7 @@ tokio-tungstenite = { version = "0.20.1", default-features = false, features = [
 toml.workspace = true
 tonic = { workspace = true, optional = true }
 hickory-proto = { workspace = true, optional = true }
+thread_local = { version = "1.1.8", default-features = false, optional = true }
 typetag = { version = "0.2.18", default-features = false }
 url = { version = "2.5.4", default-features = false, features = ["serde"] }
 warp = { version = "0.3.7", default-features = false }
@@ -517,9 +520,10 @@ protobuf-build = ["dep:tonic-build", "dep:prost-build"]
 gcp = ["dep:base64", "dep:goauth", "dep:smpl_jwt"]
 
 # Enrichment Tables
-enrichment-tables = ["enrichment-tables-geoip", "enrichment-tables-mmdb"]
+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"]
 
 # Codecs
 codecs-syslog = ["vector-lib/syslog"]
@@ -972,7 +976,7 @@ remap-benches = ["transforms-remap"]
 transform-benches = ["transforms-filter", "transforms-dedupe", "transforms-reduce", "transforms-route"]
 codecs-benches = []
 loki-benches = ["sinks-loki"]
-enrichment-tables-benches = ["enrichment-tables-geoip", "enrichment-tables-mmdb"]
+enrichment-tables-benches = ["enrichment-tables-geoip", "enrichment-tables-mmdb", "enrichment-tables-memory"]
 proptest = ["dep:proptest", "dep:proptest-derive", "vrl/proptest"]
 
 [[bench]]

src/api/schema/components/mod.rs

@@ -301,7 +301,14 @@ pub fn update_config(config: &Config) {
     }
 
     // Sinks
-    for (component_key, sink) in config.sinks() {
+    let table_sinks = config
+        .enrichment_tables()
+        .filter_map(|(k, e)| e.as_sink().map(|s| (k, s)))
+        .collect::<Vec<_>>();
+    for (component_key, sink) in config
+        .sinks()
+        .chain(table_sinks.iter().map(|(key, sink)| (*key, sink)))
+    {
         new_components.insert(
             component_key.clone(),
             Component::Sink(sink::Sink(sink::Data {

src/config/builder.rs

@@ -38,7 +38,7 @@ pub struct ConfigBuilder {
 
     /// All configured enrichment tables.
     #[serde(default)]
-    pub enrichment_tables: IndexMap<ComponentKey, EnrichmentTableOuter>,
+    pub enrichment_tables: IndexMap<ComponentKey, EnrichmentTableOuter<String>>,
 
     /// All configured sources.
     #[serde(default)]
@@ -106,6 +106,11 @@ impl From<Config> for ConfigBuilder {
             .map(|(key, sink)| (key, sink.map_inputs(ToString::to_string)))
             .collect();
 
+        let enrichment_tables = enrichment_tables
+            .into_iter()
+            .map(|(key, table)| (key, table.map_inputs(ToString::to_string)))
+            .collect();
+
         let tests = tests.into_iter().map(TestDefinition::stringify).collect();
 
         ConfigBuilder {
@@ -145,11 +150,16 @@ impl ConfigBuilder {
     pub fn add_enrichment_table<K: Into<String>, E: Into<EnrichmentTables>>(
         &mut self,
         key: K,
+        inputs: &[&str],
         enrichment_table: E,
     ) {
+        let inputs = inputs
+            .iter()
+            .map(|value| value.to_string())
+            .collect::<Vec<_>>();
         self.enrichment_tables.insert(
             ComponentKey::from(key.into()),
-            EnrichmentTableOuter::new(enrichment_table),
+            EnrichmentTableOuter::new(inputs, enrichment_table),
         );
     }
 

src/config/compiler.rs

@@ -3,7 +3,7 @@ use super::{
     OutputId,
 };
 
-use indexmap::IndexSet;
+use indexmap::{IndexMap, IndexSet};
 use vector_lib::id::Inputs;
 
 pub fn compile(mut builder: ConfigBuilder) -> Result<(Config, Vec<String>), Vec<String>> {
@@ -52,8 +52,17 @@ pub fn compile(mut builder: ConfigBuilder) -> Result<(Config, Vec<String>), Vec<
         graceful_shutdown_duration,
         allow_empty: _,
     } = builder;
+    let sinks_and_table_sinks = sinks
+        .clone()
+        .into_iter()
+        .chain(
+            enrichment_tables
+                .iter()
+                .filter_map(|(key, table)| table.as_sink().map(|s| (key.clone(), s))),
+        )
+        .collect::<IndexMap<_, _>>();
 
-    let graph = match Graph::new(&sources, &transforms, &sinks, schema) {
+    let graph = match Graph::new(&sources, &transforms, &sinks_and_table_sinks, schema) {
         Ok(graph) => graph,
         Err(graph_errors) => {
             errors.extend(graph_errors);
@@ -85,6 +94,13 @@ pub fn compile(mut builder: ConfigBuilder) -> Result<(Config, Vec<String>), Vec<
             (key, transform.with_inputs(inputs))
         })
         .collect();
+    let enrichment_tables = enrichment_tables
+        .into_iter()
+        .map(|(key, table)| {
+            let inputs = graph.inputs_for(&key);
+            (key, table.with_inputs(inputs))
+        })
+        .collect();
     let tests = tests
         .into_iter()
         .map(|test| test.resolve_outputs(&graph))

src/config/diff.rs

@@ -31,26 +31,32 @@ impl ConfigDiff {
         self.sources.flip();
         self.transforms.flip();
         self.sinks.flip();
+        self.enrichment_tables.flip();
         self
     }
 
     /// Checks whether or not the given component is present at all.
     pub fn contains(&self, key: &ComponentKey) -> bool {
-        self.sources.contains(key) || self.transforms.contains(key) || self.sinks.contains(key)
+        self.sources.contains(key)
+            || self.transforms.contains(key)
+            || self.sinks.contains(key)
+            || self.enrichment_tables.contains(key)
     }
 
     /// Checks whether or not the given component is changed.
     pub fn is_changed(&self, key: &ComponentKey) -> bool {
         self.sources.is_changed(key)
             || self.transforms.is_changed(key)
             || self.sinks.is_changed(key)
+            || self.enrichment_tables.contains(key)
     }
 
     /// Checks whether or not the given component is removed.
     pub fn is_removed(&self, key: &ComponentKey) -> bool {
         self.sources.is_removed(key)
             || self.transforms.is_removed(key)
             || self.sinks.is_removed(key)
+            || self.enrichment_tables.contains(key)
     }
 }
 

src/config/enrichment_table.rs

@@ -1,21 +1,79 @@
 use enum_dispatch::enum_dispatch;
+use serde::Serialize;
 use vector_lib::config::GlobalOptions;
-use vector_lib::configurable::{configurable_component, NamedComponent};
+use vector_lib::configurable::{configurable_component, Configurable, NamedComponent, ToValue};
+use vector_lib::id::Inputs;
 
 use crate::enrichment_tables::EnrichmentTables;
 
+use super::dot_graph::GraphConfig;
+use super::{SinkConfig, SinkOuter};
+
 /// Fully resolved enrichment table component.
 #[configurable_component]
 #[derive(Clone, Debug)]
-pub struct EnrichmentTableOuter {
+pub struct EnrichmentTableOuter<T>
+where
+    T: Configurable + Serialize + 'static + ToValue + Clone,
+{
     #[serde(flatten)]
     pub inner: EnrichmentTables,
+    #[configurable(derived)]
+    #[serde(default, skip_serializing_if = "vector_lib::serde::is_default")]
+    pub graph: GraphConfig,
+    #[configurable(derived)]
+    #[serde(
+        default = "Inputs::<T>::default",
+        skip_serializing_if = "Inputs::is_empty"
+    )]
+    pub inputs: Inputs<T>,
 }
 
-impl EnrichmentTableOuter {
-    pub fn new<I: Into<EnrichmentTables>>(inner: I) -> Self {
+impl<T> EnrichmentTableOuter<T>
+where
+    T: Configurable + Serialize + 'static + ToValue + Clone,
+{
+    pub fn new<I, IET>(inputs: I, inner: IET) -> Self
+    where
+        I: IntoIterator<Item = T>,
+        IET: Into<EnrichmentTables>,
+    {
         Self {
             inner: inner.into(),
+            graph: Default::default(),
+            inputs: Inputs::from_iter(inputs),
+        }
+    }
+
+    pub fn as_sink(&self) -> Option<SinkOuter<T>> {
+        self.inner.sink_config().map(|sink| SinkOuter {
+            graph: self.graph.clone(),
+            inputs: self.inputs.clone(),
+            healthcheck_uri: None,
+            healthcheck: Default::default(),
+            buffer: Default::default(),
+            proxy: Default::default(),
+            inner: sink,
+        })
+    }
+
+    pub(super) fn map_inputs<U>(self, f: impl Fn(&T) -> U) -> EnrichmentTableOuter<U>
+    where
+        U: Configurable + Serialize + 'static + ToValue + Clone,
+    {
+        let inputs = self.inputs.iter().map(f).collect::<Vec<_>>();
+        self.with_inputs(inputs)
+    }
+
+    pub(crate) fn with_inputs<I, U>(self, inputs: I) -> EnrichmentTableOuter<U>
+    where
+        I: IntoIterator<Item = U>,
+        U: Configurable + Serialize + 'static + ToValue + Clone,
+    {
+        EnrichmentTableOuter {
+            inputs: Inputs::from_iter(inputs),
+            inner: self.inner,
+            graph: self.graph,
         }
     }
 }
@@ -36,4 +94,8 @@ pub trait EnrichmentTableConfig: NamedComponent + core::fmt::Debug + Send + Sync
         &self,
         globals: &GlobalOptions,
     ) -> crate::Result<Box<dyn vector_lib::enrichment::Table + Send + Sync>>;
+
+    fn sink_config(&self) -> Option<Box<dyn SinkConfig>> {
+        None
+    }
 }

src/config/graph.rs

@@ -112,7 +112,7 @@ impl Graph {
             );
         }
 
-        for (id, config) in sinks.iter() {
+        for (id, config) in sinks {
             graph.nodes.insert(
                 id.clone(),
                 Node::Sink {
@@ -133,7 +133,7 @@ impl Graph {
             }
         }
 
-        for (id, config) in sinks.iter() {
+        for (id, config) in sinks {
             for input in config.inputs.iter() {
                 if let Err(e) = graph.add_input(input, id, &available_inputs) {
                     errors.push(e);

src/config/loading/config_builder.rs

@@ -63,7 +63,7 @@ impl Process for ConfigBuilderLoader {
             }
             Some(ComponentHint::EnrichmentTable) => {
                 self.builder.enrichment_tables.extend(deserialize_table::<
-                    IndexMap<ComponentKey, EnrichmentTableOuter>,
+                    IndexMap<ComponentKey, EnrichmentTableOuter<_>>,
                 >(table)?);
             }
             Some(ComponentHint::Test) => {