feat(transform): Add redis enrichment transformer

Cargo.toml

@@ -727,6 +727,7 @@ transforms-logs = [
   "transforms-lua",
   "transforms-metric_to_log",
   "transforms-reduce",
+  "transforms-redis",
   "transforms-remap",
   "transforms-route",
   "transforms-exclusive-route",
@@ -756,6 +757,7 @@ transforms-log_to_metric = []
 transforms-lua = ["dep:mlua", "vector-lib/lua"]
 transforms-metric_to_log = []
 transforms-reduce = ["transforms-impl-reduce"]
+transforms-redis = ["dep:redis"]
 transforms-remap = []
 transforms-route = []
 transforms-exclusive-route = []

changelog.d/add-redis-enrichment-transform.feature.md

@@ -0,0 +1,5 @@
+Adds a new transform that allows enrichment of events from Redis.  Allows for dynamic event enrichment from external sources while being fully decoupled from that source.  
+Example:  Feature Flags or Metadata Enrichment
+
+
+authors: akutta

src/internal_events/mod.rs

@@ -112,6 +112,7 @@ mod prometheus;
 #[cfg(any(feature = "sinks-pulsar", feature = "sources-pulsar"))]
 mod pulsar;
 #[cfg(feature = "sources-redis")]
+#[cfg(any(feature = "sources-redis", feature = "transforms-redis"))]
 mod redis;
 #[cfg(feature = "transforms-impl-reduce")]
 mod reduce;
@@ -263,7 +264,7 @@ pub(crate) use self::postgresql_metrics::*;
 pub(crate) use self::prometheus::*;
 #[cfg(any(feature = "sinks-pulsar", feature = "sources-pulsar"))]
 pub(crate) use self::pulsar::*;
-#[cfg(feature = "sources-redis")]
+#[cfg(any(feature = "sources-redis", feature = "transforms-redis"))]
 pub(crate) use self::redis::*;
 #[cfg(feature = "transforms-impl-reduce")]
 pub(crate) use self::reduce::*;

src/internal_events/redis.rs

@@ -33,3 +33,43 @@ impl InternalEvent for RedisReceiveEventError {
         .increment(1);
     }
 }
+
+#[derive(Debug, NamedInternalEvent)]
+pub struct RedisTransformLookupError {
+    pub error: String,
+}
+
+impl InternalEvent for RedisTransformLookupError {
+    fn emit(self) {
+        error!(
+            message = "Redis lookup failed.",
+            error = %self.error,
+            error_type = error_type::REQUEST_FAILED,
+            stage = error_stage::PROCESSING,
+        );
+        counter!(
+            "component_errors_total",
+            "error_type" => error_type::REQUEST_FAILED,
+            "stage" => error_stage::PROCESSING,
+        )
+        .increment(1);
+    }
+}
+
+#[derive(Debug, NamedInternalEvent)]
+pub struct RedisTransformLruCacheHit;
+
+impl InternalEvent for RedisTransformLruCacheHit {
+    fn emit(self) {
+        counter!("component_cache_hits_total", "cache" => "redis_transform").increment(1);
+    }
+}
+
+#[derive(Debug, NamedInternalEvent)]
+pub struct RedisTransformLruCacheMiss;
+
+impl InternalEvent for RedisTransformLruCacheMiss {
+    fn emit(self) {
+        counter!("component_cache_misses_total", "cache" => "redis_transform").increment(1);
+    }
+}

src/transforms/mod.rs

@@ -23,6 +23,8 @@ pub mod log_to_metric;
 pub mod lua;
 #[cfg(feature = "transforms-metric_to_log")]
 pub mod metric_to_log;
+#[cfg(feature = "transforms-redis")]
+pub mod redis;
 #[cfg(feature = "transforms-remap")]
 pub mod remap;
 #[cfg(feature = "transforms-route")]

src/transforms/redis/config.rs

@@ -0,0 +1,193 @@
+use std::{num::NonZeroUsize, time::Duration};
+
+use redis::{AsyncTypedCommands, RedisError};
+use serde_with::serde_as;
+use snafu::{ResultExt, Snafu};
+use vector_lib::{
+    config::{LogNamespace, clone_input_definitions},
+    configurable::configurable_component,
+    lookup::lookup_v2::OptionalValuePath,
+};
+
+use crate::{
+    config::{
+        DataType, GenerateConfig, Input, OutputId, TransformConfig, TransformContext,
+        TransformOutput,
+    },
+    schema,
+    template::Template,
+};
+
+use super::transform::RedisTransform;
+
+#[derive(Debug, Snafu)]
+enum BuildError {
+    #[snafu(display("Failed to build redis client: {}", source))]
+    Client { source: redis::RedisError },
+    #[snafu(display("Failed to create connection: {}", source))]
+    Connection { source: RedisError },
+    #[snafu(display("Connection to Redis timed out after {} seconds", timeout_secs))]
+    ConnectionTimeout { timeout_secs: u64 },
+}
+
+/// Configuration for the `redis` transform.
+#[serde_as]
+#[configurable_component(transform("redis", "Enrich events with data from Redis lookups."))]
+#[derive(Clone, Debug)]
+#[serde(deny_unknown_fields)]
+pub struct RedisTransformConfig {
+    /// The Redis URL to connect to.
+    ///
+    /// The URL must take the form of `protocol://server:port/db` where the `protocol` can either be `redis` or `rediss` for connections secured using TLS.
+    #[configurable(metadata(docs::examples = "redis://127.0.0.1:6379/0"))]
+    pub url: String,
+
+    /// The Redis key template to use for lookups.
+    ///
+    /// This template is evaluated for each event to determine the Redis key to look up.
+    /// The template can use event fields using the `{{ field_name }}` syntax, for example: `user:{{ user_id }}`.
+    #[configurable(metadata(
+        docs::examples = "user:{{ user_id }}",
+        docs::examples = "session:{{ session_id }}"
+    ))]
+    pub key: Template,
+
+    /// The field path where the Redis lookup value will be stored.
+    ///
+    /// If the Redis key is not found, the field will not be added to the event.
+    #[configurable(metadata(
+        docs::examples = "redis_data",
+        docs::examples = "enrichment.user_data"
+    ))]
+    pub output_field: OptionalValuePath,
+
+    /// The default value to use if the Redis key is not found.
+    ///
+    /// If not specified, the field will not be added when the key is missing.
+    #[configurable(metadata(docs::examples = "default_value", docs::examples = ""))]
+    pub default_value: Option<String>,
+
+    /// Maximum number of Redis lookup results to cache in memory.
+    ///
+    /// When set, Redis lookup results are cached to reduce round-trips to Redis.
+    /// The cache uses an LRU (Least Recently Used) eviction policy.
+    ///
+    /// If not specified, caching is disabled and every lookup will query Redis.
+    #[configurable(metadata(docs::examples = 1000, docs::examples = 10000))]
+    pub cache_max_size: Option<NonZeroUsize>,
+
+    /// Time-to-live (TTL) for cached Redis lookup results.
+    ///
+    /// When set, cached entries will expire after the specified duration.
+    /// Expired entries are automatically refreshed from Redis on the next lookup
+    /// instead of being evicted from the cache.
+    ///
+    /// If not specified, cached entries do not expire.
+    #[serde_as(as = "Option<serde_with::DurationMilliSeconds<u64>>")]
+    #[configurable(metadata(docs::examples = 300000, docs::examples = 3600000))]
+    #[configurable(metadata(docs::type_unit = "milliseconds"))]
+    pub cache_ttl: Option<Duration>,
+
+    /// Maximum number of concurrent Redis lookups.
+    ///
+    /// This limits the number of Redis lookups that can be in-flight simultaneously.
+    /// Higher values allow more parallelism but may increase Redis connection pressure.
+    ///
+    /// Defaults to 100 if not specified.
+    #[configurable(metadata(docs::examples = 50, docs::examples = 200))]
+    pub concurrency_limit: Option<NonZeroUsize>,
+
+    /// Timeout for establishing connection to Redis and verifying connectivity during startup.
+    ///
+    /// If Redis is unavailable or doesn't respond within this timeout, Vector will fail to start.
+    /// This ensures Vector fails fast during startup rather than starting and failing later when events need to be enriched.
+    ///
+    /// Defaults to 5 seconds if not specified.
+    #[serde_as(as = "serde_with::DurationMilliSeconds<u64>")]
+    #[configurable(metadata(docs::examples = 5000, docs::examples = 10000))]
+    #[configurable(metadata(docs::type_unit = "milliseconds"))]
+    #[serde(default = "default_connection_timeout")]
+    pub connection_timeout: Duration,
+}
+
+const fn default_concurrency_limit() -> NonZeroUsize {
+    NonZeroUsize::new(100).expect("concurrency limit must be at least 1")
+}
+
+const fn default_connection_timeout() -> Duration {
+    Duration::from_secs(5)
+}
+
+impl GenerateConfig for RedisTransformConfig {
+    fn generate_config() -> toml::Value {
+        toml::from_str(
+            r#"
+            url = "redis://127.0.0.1:6379/0"
+            key = "user:{{ user_id }}"
+            output_field = "redis_data"
+            "#,
+        )
+        .unwrap()
+    }
+}
+
+#[async_trait::async_trait]
+#[typetag::serde(name = "redis")]
+impl TransformConfig for RedisTransformConfig {
+    async fn build(
+        &self,
+        _context: &TransformContext,
+    ) -> crate::Result<crate::transforms::Transform> {
+        let redis_client = redis::Client::open(self.url.as_str()).context(ClientSnafu {})?;
+
+        // Establish connection with timeout to fail fast if Redis is unavailable
+        let connection_timeout = self.connection_timeout;
+        let mut conn = tokio::time::timeout(connection_timeout, redis_client.get_connection_manager())
+            .await
+            .map_err(|_| BuildError::ConnectionTimeout {
+                timeout_secs: connection_timeout.as_secs(),
+            })?
+            .context(ConnectionSnafu {})?;
+
+        // Test the connection with a ping, also with timeout
+        tokio::time::timeout(connection_timeout, conn.ping())
+            .await
+            .map_err(|_| BuildError::ConnectionTimeout {
+                timeout_secs: connection_timeout.as_secs(),
+            })?
+            .context(ConnectionSnafu {})?;
+
+        Ok(crate::transforms::Transform::event_task(
+            RedisTransform::new(
+                conn,
+                self.key.clone(),
+                self.output_field.clone(),
+                self.default_value.clone(),
+                self.cache_max_size,
+                self.cache_ttl,
+                self.concurrency_limit
+                    .unwrap_or_else(default_concurrency_limit),
+            ),
+        ))
+    }
+
+    fn input(&self) -> Input {
+        Input::all()
+    }
+
+    fn outputs(
+        &self,
+        _enrichment_tables: vector_lib::enrichment::TableRegistry,
+        input_definitions: &[(OutputId, schema::Definition)],
+        _: LogNamespace,
+    ) -> Vec<TransformOutput> {
+        vec![TransformOutput::new(
+            DataType::all_bits(),
+            clone_input_definitions(input_definitions),
+        )]
+    }
+
+    fn enable_concurrency(&self) -> bool {
+        true
+    }
+}

src/transforms/redis/mod.rs

@@ -0,0 +1,12 @@
+//! `redis` transform.
+//!
+//! Enriches events with data from Redis lookups.
+
+mod config;
+mod transform;
+
+#[cfg(test)]
+mod tests;
+
+pub use config::*;
+pub use transform::*;