feat(redis sink): support input based on encoding type

ynachi · ynachi · commit 3c2a0bf55d49 · 2025-05-02T20:55:57.000+02:00
diff --git a/changelog.d/22909_redis_sink_supports_encoding_input.feature.md b/changelog.d/22909_redis_sink_supports_encoding_input.feature.md
@@ -0,0 +1,3 @@
+The redis sink now supports any input event type that the configured encoding supports. It previously only supported log events.
+
+authors: ynachi
diff --git a/src/sinks/redis/config.rs b/src/sinks/redis/config.rs
@@ -146,7 +146,7 @@ impl SinkConfig for RedisSinkConfig {
     }
 
     fn input(&self) -> Input {
-        Input::new(self.encoding.config().input_type() & DataType::Log)
+        Input::new(self.encoding.config().input_type())
     }
 
     fn acknowledgements(&self) -> &AcknowledgementsConfig {
diff --git a/src/sinks/redis/integration_tests.rs b/src/sinks/redis/integration_tests.rs
@@ -7,6 +7,10 @@ use vector_lib::{
     event::LogEvent,
 };
 
+use crate::event::{
+    BatchNotifier, BatchStatus, Event, Metric, MetricKind, MetricValue, TraceEvent,
+};
+
 use super::config::{DataTypeConfig, ListOption, Method, RedisSinkConfig};
 use crate::{
     sinks::prelude::*,
@@ -15,7 +19,7 @@ use crate::{
             assert_data_volume_sink_compliance, assert_sink_compliance, DATA_VOLUME_SINK_TAGS,
             SINK_TAGS,
         },
-        random_lines_with_stream, random_string, trace_init,
+        map_event_batch_stream, random_lines_with_stream, random_string, trace_init,
     },
 };
 
@@ -290,3 +294,165 @@ async fn redis_sink_channel_data_volume_tags() {
         }
     }
 }
+
+#[tokio::test]
+async fn redis_sink_metrics() {
+    trace_init();
+
+    let key = Template::try_from(format!("test-metrics-{}", random_string(10)))
+        .expect("should not fail to create key template");
+    debug!("Test key name: {}.", key);
+    let num_events = 1000;
+    debug!("Test events num: {}.", num_events);
+
+    let cnf = RedisSinkConfig {
+        endpoint: redis_server(),
+        key: key.clone(),
+        encoding: JsonSerializerConfig::default().into(),
+        data_type: DataTypeConfig::List,
+        list_option: Some(ListOption {
+            method: Method::RPush,
+        }),
+        batch: BatchConfig::default(),
+        request: TowerRequestConfig {
+            rate_limit_num: u64::MAX,
+            ..Default::default()
+        },
+        acknowledgements: Default::default(),
+    };
+
+    // Create a mix of counter and gauge metrics
+    let mut events: Vec<Event> = Vec::new();
+    for i in 0..num_events {
+        let metric = if i % 2 == 0 {
+            // Counter metrics
+            Metric::new(
+                format!("counter_{}", i),
+                MetricKind::Absolute,
+                MetricValue::Counter { value: i as f64 },
+            )
+        } else {
+            // Gauge metrics
+            Metric::new(
+                format!("gauge_{}", i),
+                MetricKind::Absolute,
+                MetricValue::Gauge { value: i as f64 },
+            )
+        };
+        events.push(metric.into());
+    }
+    let input = stream::iter(events.clone().into_iter().map(Into::into));
+
+    // Publish events
+    let cnf2 = cnf.clone();
+    assert_sink_compliance(&SINK_TAGS, async move {
+        let cx = SinkContext::default();
+        let (sink, _healthcheck) = cnf2.build(cx).await.unwrap();
+        sink.run(input).await
+    })
+    .await
+    .expect("Running sink failed");
+
+    // Verify metrics were stored correctly
+    let mut conn = cnf.build_client().await.unwrap();
+
+    let key_exists: bool = conn.exists(key.to_string()).await.unwrap();
+    debug!("Test key: {} exists: {}.", key, key_exists);
+    assert!(key_exists);
+
+    let llen: usize = conn.llen(key.clone().to_string()).await.unwrap();
+    debug!("Test key: {} len: {}.", key, llen);
+    assert_eq!(llen, num_events);
+
+    // Verify the content of each metric
+    for i in 0..num_events {
+        let original_event = events.get(i).unwrap().as_metric();
+        let payload: (String, String) = conn.blpop(key.clone().to_string(), 2000.0).await.unwrap();
+        let val = payload.1;
+
+        // Parse the JSON and verify key metric properties
+        let json: serde_json::Value = serde_json::from_str(&val).unwrap();
+
+        if i % 2 == 0 {
+            // Counter metrics
+            assert_eq!(json["name"], format!("counter_{}", i));
+            assert_eq!(json["kind"], "absolute");
+            assert_eq!(json["counter"]["value"], i as f64);
+        } else {
+            // Gauge metrics
+            assert_eq!(json["name"], format!("gauge_{}", i));
+            assert_eq!(json["kind"], "absolute");
+            assert_eq!(json["gauge"]["value"], i as f64);
+        }
+
+        // Verify that the name matches what we expect
+        assert_eq!(json["name"].as_str().unwrap(), original_event.name());
+    }
+}
+
+#[tokio::test]
+async fn redis_sink_traces() {
+    use crate::test_util::components::{assert_sink_compliance, SINK_TAGS};
+
+    trace_init();
+
+    assert_sink_compliance(&SINK_TAGS, async {
+        // Setup Redis sink config
+        let key = Template::try_from(format!("test-traces-{}", random_string(10))).unwrap();
+        let config = RedisSinkConfig {
+            endpoint: redis_server(),
+            key: key.clone(),
+            encoding: JsonSerializerConfig::default().into(),
+            data_type: DataTypeConfig::List,
+            list_option: Some(ListOption {
+                method: Method::RPush,
+            }),
+            batch: BatchConfig::default(),
+            request: TowerRequestConfig::default(),
+            acknowledgements: Default::default(),
+        };
+
+        // Build the sink
+        let cx = SinkContext::default();
+        let (sink, _) = config.build(cx).await.unwrap();
+
+        // Create a  trace event
+        let mut trace = TraceEvent::default();
+        trace.insert("name", "test_trace");
+        trace.insert("service", "redis_test");
+
+        // Set up batch notification for checking delivery status
+        let (batch, receiver) = BatchNotifier::new_with_receiver();
+        let trace_with_batch = trace.with_batch_notifier(&batch);
+
+        // Create event stream with proper conversion to EventArray
+        let events = vec![Event::Trace(trace_with_batch)];
+        let stream = map_event_batch_stream(stream::iter(events), Some(batch));
+
+        // Run the sink with the stream
+        sink.run(stream).await.unwrap();
+
+        // Check that events were delivered
+        assert_eq!(receiver.await, BatchStatus::Delivered);
+
+        // Verify data in Redis
+        let mut conn = redis::Client::open(redis_server())
+            .unwrap()
+            .get_async_connection()
+            .await
+            .unwrap();
+
+        let len: usize = conn.llen(key.to_string()).await.unwrap();
+        assert_eq!(len, 1);
+
+        // Check content
+        let payload: (String, String) = conn.blpop(key.to_string(), 2000.0).await.unwrap();
+        let json_str = payload.1;
+
+        // Verify the trace content
+        let json: serde_json::Value = serde_json::from_str(&json_str).unwrap();
+        assert_eq!(json["name"], "test_trace");
+        assert_eq!(json["service"], "redis_test");
+    })
+    .await;
+}
diff --git a/src/sinks/redis/tests.rs b/src/sinks/redis/tests.rs
@@ -1,7 +1,7 @@
 use std::collections::HashMap;
 
 use vector_lib::codecs::{JsonSerializerConfig, TextSerializerConfig};
-use vector_lib::event::LogEvent;
+use vector_lib::event::{LogEvent, Metric, MetricKind, MetricValue};
 use vector_lib::request_metadata::GroupedCountByteSize;
 
 use super::{config::RedisSinkConfig, request_builder::encode_event};
@@ -16,7 +16,7 @@ fn generate_config() {
 }
 
 #[test]
-fn redis_event_json() {
+fn redis_log_event_json() {
     let msg = "hello_world".to_owned();
     let mut byte_size = GroupedCountByteSize::new_untagged();
     let mut evt = LogEvent::from(msg.clone());
@@ -35,7 +35,7 @@ fn redis_event_json() {
 }
 
 #[test]
-fn redis_event_text() {
+fn redis_log_event_text() {
     let msg = "hello_world".to_owned();
     let evt = LogEvent::from(msg.clone());
     let mut byte_size = GroupedCountByteSize::new_untagged();
@@ -52,7 +52,7 @@ fn redis_event_text() {
 }
 
 #[test]
-fn redis_encode_event() {
+fn redis_log_encode_event() {
     let msg = "hello_world";
     let mut evt = LogEvent::from(msg);
     let mut byte_size = GroupedCountByteSize::new_untagged();
@@ -71,3 +71,29 @@ fn redis_encode_event() {
     let map: HashMap<String, String> = serde_json::from_slice(&result[..]).unwrap();
     assert!(!map.contains_key("key"));
 }
+
+#[test]
+fn redis_metric_encode_event() {
+    let mut byte_size = GroupedCountByteSize::new_untagged();
+    let metric = Metric::new(
+        "test_counter",
+        MetricKind::Absolute,
+        MetricValue::Counter { value: 42.0 },
+    );
+
+    let result = encode_event(
+        metric.into(),
+        "metrics.counter".to_string(),
+        &Default::default(),
+        &mut Encoder::<()>::new(JsonSerializerConfig::default().build().into()),
+        &mut byte_size,
+    )
+    .unwrap()
+    .value;
+
+    let json: serde_json::Value = serde_json::from_slice(&result).unwrap();
+
+    assert_eq!(json["name"], "test_counter");
+    assert_eq!(json["kind"], "absolute");
+    assert_eq!(json["counter"]["value"], 42.0);
+}

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,3 @@`
	`1`	`+The redis sink now supports any input event type that the configured encoding supports. It previously only supported log events.`
	`2`	`+`
	`3`	`+authors: ynachi`
Original file line number	Diff line number	Diff line change
`@@ -146,7 +146,7 @@ impl SinkConfig for RedisSinkConfig {`
`146`	`146`	`}`
`147`	`147`
`148`	`148`	`fn input(&self) -> Input {`
`149`		`- Input::new(self.encoding.config().input_type() & DataType::Log)`
	`149`	`+ Input::new(self.encoding.config().input_type())`
`150`	`150`	`}`
`151`	`151`
`152`	`152`	`fn acknowledgements(&self) -> &AcknowledgementsConfig {`