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

Author: ynachi

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

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 {

src/sinks/redis/integration_tests.rs

@@ -7,6 +7,8 @@ use vector_lib::{
     event::LogEvent,
 };
 
+use crate::event::{Event, Metric, MetricKind, MetricValue};
+
 use super::config::{DataTypeConfig, ListOption, Method, RedisSinkConfig};
 use crate::{
     sinks::prelude::*,
@@ -290,3 +292,98 @@ 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());
+    }
+}

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);
+}