feat(redis source): Use OptionalValuePath for redis_key (#15615)

Author: spencergilbert

src/sources/redis/mod.rs

@@ -5,16 +5,18 @@ use codecs::{
     StreamDecodingError,
 };
 use futures::StreamExt;
-use lookup::{lookup_v2::BorrowedSegment, owned_value_path, path};
+use lookup::{lookup_v2::OptionalValuePath, owned_value_path, path, OwnedValuePath};
 use snafu::{ResultExt, Snafu};
 use tokio_util::codec::FramedRead;
 use value::Kind;
 use vector_common::internal_event::{
     ByteSize, BytesReceived, CountByteSize, InternalEventHandle as _, Protocol, Registered,
 };
 use vector_config::{configurable_component, NamedComponent};
-use vector_core::config::{LegacyKey, LogNamespace};
-use vector_core::EstimatedJsonEncodedSizeOf;
+use vector_core::{
+    config::{LegacyKey, LogNamespace},
+    EstimatedJsonEncodedSizeOf,
+};
 
 use crate::{
     codecs::{Decoder, DecodingConfig},
@@ -116,7 +118,7 @@ pub struct RedisSourceConfig {
     /// The value will be the Redis key that the event was read from.
     ///
     /// By default, this is not set and the field will not be automatically added.
-    redis_key: Option<String>,
+    redis_key: Option<OptionalValuePath>,
 
     #[configurable(derived)]
     #[serde(default = "default_framing_message_based")]
@@ -158,6 +160,7 @@ impl SourceConfig for RedisSourceConfig {
         if self.key.is_empty() {
             return Err("`key` cannot be empty.".into());
         }
+        let redis_key = self.redis_key.clone().and_then(|k| k.path);
 
         let client = redis::Client::open(self.url.as_str()).context(ClientSnafu {})?;
         let connection_info = ConnectionInfo::from(client.get_connection_info());
@@ -173,7 +176,7 @@ impl SourceConfig for RedisSourceConfig {
             bytes_received: bytes_received.clone(),
             events_received: events_received.clone(),
             key: self.key.clone(),
-            redis_key: self.redis_key.clone(),
+            redis_key,
             decoder,
             cx,
             log_namespace,
@@ -193,8 +196,8 @@ impl SourceConfig for RedisSourceConfig {
 
         let redis_key_path = self
             .redis_key
-            .as_ref()
-            .map(|x| owned_value_path!(x))
+            .clone()
+            .and_then(|k| k.path)
             .map(LegacyKey::InsertIfEmpty);
 
         let schema_definition = self
@@ -222,7 +225,7 @@ pub(self) struct InputHandler {
     pub bytes_received: Registered<BytesReceived>,
     pub events_received: Registered<EventsReceived>,
     pub key: String,
-    pub redis_key: Option<String>,
+    pub redis_key: Option<OwnedValuePath>,
     pub decoder: Decoder,
     pub log_namespace: LogNamespace,
     pub cx: SourceContext,
@@ -257,15 +260,10 @@ impl InputHandler {
                                 now,
                             );
 
-                            let redis_key_path = self
-                                .redis_key
-                                .as_deref()
-                                .map(|x| [BorrowedSegment::from(x)]);
-
                             self.log_namespace.insert_source_metadata(
                                 RedisSourceConfig::NAME,
                                 log,
-                                redis_key_path.as_ref().map(LegacyKey::InsertIfEmpty),
+                                self.redis_key.as_ref().map(LegacyKey::InsertIfEmpty),
                                 path!("key"),
                                 self.key.as_str(),
                             );
@@ -307,10 +305,13 @@ mod integration_test {
     use redis::AsyncCommands;
 
     use super::*;
-    use crate::config::log_schema;
-    use crate::test_util::components::{run_and_assert_source_compliance_n, SOURCE_TAGS};
     use crate::{
-        test_util::{collect_n, random_string},
+        config::log_schema,
+        test_util::{
+            collect_n,
+            components::{run_and_assert_source_compliance_n, SOURCE_TAGS},
+            random_string,
+        },
         SourceSender,
     };
 
@@ -369,7 +370,7 @@ mod integration_test {
             }),
             url: REDIS_SERVER.to_owned(),
             key: key.clone(),
-            redis_key: Some("remapped_key".into()),
+            redis_key: Some(OptionalValuePath::from(owned_value_path!("remapped_key"))),
             framing: default_framing_message_based(),
             decoding: default_decoding(),
             log_namespace: Some(true),