Repository: ArroyoSystems/arroyo (language: rust)

Issue:
Support Redis username and password in RedisConnector
Right now if you want to connect to a Redis cluster that has a username and password you have to inline it into the connection string, e.g. `redis://username:password@host:ip`. It'd be nice to be able to specify this on the Redis connection. This'd be added as two optional properties [here](https://github.com/ArroyoSystems/arroyo/blob/master/connector-schemas/redis/connection.json#L11). I think there should probably a single username/password for the Clustered case as well. 

Then, to update the operators, you'll need to use a different IntoConnectionInfo, likely just ConnectionInfo. That would happen [here](https://github.com/ArroyoSystems/arroyo/blob/master/arroyo-worker/src/connectors/redis/sink.rs#L261-L268).

Will also need to pick the sql parameters for these options.



Discussion / hints:

Fix patch:
diff --git a/arroyo-api/src/pipelines.rs b/arroyo-api/src/pipelines.rs
index 6e4d2ad22..c28ce5a62 100644
--- a/arroyo-api/src/pipelines.rs
+++ b/arroyo-api/src/pipelines.rs
@@ -54,6 +54,7 @@ use crate::types::public::{PipelineType, RestartMode, StopMode};
 use crate::{connection_tables, to_micros};
 use crate::{handle_db_error, optimizations, AuthData};
 use create_pipeline_req::Config::Sql;
+
 const DEFAULT_CHECKPOINT_INTERVAL: Duration = Duration::from_secs(10);
 
 async fn compile_sql<'e, E>(
diff --git a/arroyo-connectors/src/redis.rs b/arroyo-connectors/src/redis.rs
index 4c951c54c..6db9bd5da 100644
--- a/arroyo-connectors/src/redis.rs
+++ b/arroyo-connectors/src/redis.rs
@@ -1,17 +1,21 @@
+use std::collections::HashMap;
+use std::convert::Infallible;
+
 use anyhow::{anyhow, bail};
-use arroyo_rpc::api_types::connections::{
-    ConnectionProfile, ConnectionSchema, ConnectionType, FieldType, PrimitiveType,
-    TestSourceMessage,
-};
-use arroyo_rpc::OperatorConfig;
+use arroyo_rpc::var_str::VarStr;
 use axum::response::sse::Event;
 use redis::cluster::ClusterClient;
+use redis::{Client, ConnectionInfo, IntoConnectionInfo};
 use serde::{Deserialize, Serialize};
-use std::collections::HashMap;
-use std::convert::Infallible;
 use tokio::sync::oneshot::Receiver;
 use typify::import_types;
 
+use arroyo_rpc::api_types::connections::{
+    ConnectionProfile, ConnectionSchema, ConnectionType, FieldType, PrimitiveType,
+    TestSourceMessage,
+};
+use arroyo_rpc::OperatorConfig;
+
 use crate::{pull_opt, pull_option_to_u64, Connection, Connector};
 
 pub struct RedisConnector {}
@@ -20,31 +24,64 @@ const CONFIG_SCHEMA: &str = include_str!("../../connector-schemas/redis/connecti
 const TABLE_SCHEMA: &str = include_str!("../../connector-schemas/redis/table.json");
 const ICON: &str = include_str!("../resources/redis.svg");
 
-import_types!(schema = "../connector-schemas/redis/connection.json",);
+import_types!(
+    schema = "../connector-schemas/redis/connection.json",
+    convert = {
+        {type = "string", format = "var-str"} = VarStr
+    }
+);
 import_types!(schema = "../connector-schemas/redis/table.json");
 
-fn validate_column(
-    schema: &ConnectionSchema,
-    column: String,
-    path: &str,
-) -> anyhow::Result<String> {
-    if schema
-        .fields
-        .iter()
-        .find(|f| {
-            f.field_name == column
-                && f.field_type.r#type == FieldType::Primitive(PrimitiveType::String)
+enum RedisClient {
+    Standard(Client),
+    Clustered(ClusterClient),
+}
+
+impl RedisClient {
+    pub fn new(config: &RedisConfig) -> anyhow::Result<Self> {
+        Ok(match &config.connection {
+            RedisConfigConnection::Address(address) => {
+                let info = from_address(&config, &address.0)?;
+
+                RedisClient::Standard(Client::open(info).map_err(|e| {
+                    anyhow!(
+                        "Failed to construct Redis client for {}: {:?}",
+                        address.0,
+                        e
+                    )
+                })?)
+            }
+            RedisConfigConnection::Addresses(addresses) => {
+                let infos: anyhow::Result<Vec<ConnectionInfo>> = addresses
+                    .iter()
+                    .map(|address| from_address(&config, address))
+                    .collect();
+
+                RedisClient::Clustered(
+                    ClusterClient::new(infos?).map_err(|e| {
+                        anyhow!("Failed to construct Redis Cluster client: {:?}", e)
+                    })?,
+                )
+            }
         })
-        .is_none()
-    {
-        bail!(
-            "invalid value '{}' for {}, must be the name of a TEXT column on the table",
-            column,
-            path
-        );
+    }
+}
+
+fn from_address(config: &RedisConfig, address: &str) -> anyhow::Result<ConnectionInfo> {
+    let mut info: ConnectionInfo = address
+        .to_string()
+        .into_connection_info()
+        .map_err(|e| anyhow!("invalid redis address: {:?}", e))?;
+
+    if let Some(username) = &config.username {
+        info.redis.username = Some(username.sub_env_vars().map_err(|e| anyhow!("{}", e))?);
+    }
+
+    if let Some(password) = &config.password {
+        info.redis.password = Some(password.sub_env_vars().map_err(|e| anyhow!("{}", e))?);
     }
 
-    Ok(column)
+    Ok(info)
 }
 
 async fn test_inner(
@@ -57,21 +94,14 @@ async fn test_inner(
         .await
         .unwrap();
 
-    match c.connection {
-        RedisConfigConnection::Address(address) => {
-            let client = redis::Client::open(address.0.clone()).map_err(|e| {
-                anyhow!(
-                    "Failed to construct Redis client for {}: {:?}",
-                    address.0,
-                    e
-                )
-            })?;
+    let client = RedisClient::new(&c)?;
 
+    match &client {
+        RedisClient::Standard(client) => {
             let mut connection = client
                 .get_async_connection()
                 .await
-                .map_err(|e| anyhow!("Failed to connect to Redis at {}: {:?}", address.0, e))?;
-
+                .map_err(|e| anyhow!("Failed to connect to to Redis Cluster: {:?}", e))?;
             tx.send(Ok(Event::default()
                 .json_data(TestSourceMessage::info(
                     "Connected successfully, sending PING",
@@ -85,15 +115,11 @@ async fn test_inner(
                 .await
                 .map_err(|e| anyhow!("Received error sending PING command: {:?}", e))?;
         }
-        RedisConfigConnection::Addresses(addresses) => {
-            let client = ClusterClient::new(addresses.into_iter().map(|a| a.0).collect())
-                .map_err(|e| anyhow!("Failed to construct Redis Cluster client: {:?}", e))?;
-
+        RedisClient::Clustered(client) => {
             let mut connection = client
                 .get_async_connection()
                 .await
                 .map_err(|e| anyhow!("Failed to connect to to Redis Cluster: {:?}", e))?;
-
             tx.send(Ok(Event::default()
                 .json_data(TestSourceMessage::info(
                     "Connected successfully, sending PING",
@@ -107,7 +133,7 @@ async fn test_inner(
                 .await
                 .map_err(|e| anyhow!("Received error sending PING command: {:?}", e))?;
         }
-    };
+    }
 
     Ok("Received PING response successfully".to_string())
 }
@@ -154,7 +180,7 @@ impl Connector for RedisConnector {
         let (tx, rx) = tokio::sync::oneshot::channel();
 
         tokio::spawn(async move {
-            let (itx, _) = tokio::sync::mpsc::channel(8);
+            let (itx, _rx) = tokio::sync::mpsc::channel(8);
             let message = match test_inner(profile, itx).await {
                 Ok(_) => TestSourceMessage::done("Successfully connected to Redis"),
                 Err(e) => TestSourceMessage::fail(format!("Failed to connect to Redis: {:?}", e)),
@@ -195,18 +221,16 @@ impl Connector for RedisConnector {
     ) -> anyhow::Result<Connection> {
         let connection_config = match profile {
             Some(connection_profile) => {
-                let config: RedisConfigConnection =
-                    serde_json::from_value(connection_profile.config.clone())
-                        .map_err(|e| anyhow!("Failed to parse connection config: {:?}", e))?;
-                RedisConfig { connection: config }
+                serde_json::from_value(connection_profile.config.clone())
+                    .map_err(|e| anyhow!("Failed to parse connection config: {:?}", e))?
             }
             None => {
                 let address = options.remove("address");
 
                 let cluster_addresses = options.remove("cluster.addresses");
-                let connection_config = match (address, cluster_addresses) {
+                let connection = match (address, cluster_addresses) {
                     (Some(address), None) => {
-                        RedisConfigConnection::Address(RedisConfigConnectionAddress(address))
+                        RedisConfigConnection::Address(Address(address.to_string()))
                     }
                     (None, Some(cluster_addresses)) => RedisConfigConnection::Addresses(
                         cluster_addresses
@@ -221,8 +245,14 @@ impl Connector for RedisConnector {
                         bail!("one of `address` or `cluster.addresses` must be set");
                     }
                 };
+
+                let username = options.remove("username").map(VarStr::new);
+                let password = options.remove("password").map(VarStr::new);
+
                 RedisConfig {
-                    connection: connection_config,
+                    connection,
+                    username,
+                    password,
                 }
             }
         };
@@ -339,51 +369,7 @@ impl Connector for RedisConnector {
             .map(|t| t.to_owned())
             .ok_or_else(|| anyhow!("'format' must be set for Redis connection"))?;
 
-        match &config.connection {
-            RedisConfigConnection::Address(address) => {
-                let _ = redis::Client::open(address.0.clone()).map_err(|e| {
-                    anyhow!(
-                        "Failed to construct Redis client for {}: {:?}",
-                        address.0,
-                        e
-                    )
-                })?;
-            }
-            RedisConfigConnection::Addresses(addresses) => {
-                let _ = ClusterClient::new(addresses.into_iter().map(|a| a.0.clone()).collect())
-                    .map_err(|e| anyhow!("Failed to construct Redis Cluster client: {:?}", e))?;
-            }
-        }
-
-        match &table.connector_type {
-            TableType::Target(Target::StringTable { key_column, .. }) => {
-                if let Some(key_column) = key_column {
-                    validate_column(&schema, key_column.clone(), "connector_type.key_column")?;
-                }
-            }
-            TableType::Target(Target::ListTable {
-                list_key_column, ..
-            }) => {
-                if let Some(n) = list_key_column {
-                    validate_column(&schema, n.clone(), "connector_type.list_key_column")?;
-                }
-            }
-            TableType::Target(Target::HashTable {
-                hash_key_column,
-                hash_field_column,
-                ..
-            }) => {
-                if let Some(n) = hash_key_column {
-                    validate_column(&schema, n.clone(), "connector_type.hash_key_column")?;
-                }
-
-                validate_column(
-                    &schema,
-                    hash_field_column.clone(),
-                    "connector_type.hash_field_column",
-                )?;
-            }
-        };
+        let _ = RedisClient::new(&config)?;
 
         let config = OperatorConfig {
             connection: serde_json::to_value(config).unwrap(),
diff --git a/arroyo-worker/src/connectors/redis/mod.rs b/arroyo-worker/src/connectors/redis/mod.rs
index e2f86e178..644a8ffda 100644
--- a/arroyo-worker/src/connectors/redis/mod.rs
+++ b/arroyo-worker/src/connectors/redis/mod.rs
@@ -1,6 +1,80 @@
+use crate::connectors::redis::sink::GeneralConnection;
+use anyhow::anyhow;
+use arroyo_rpc::var_str::VarStr;
+use redis::aio::ConnectionManager;
+use redis::cluster::ClusterClient;
+use redis::{Client, ConnectionInfo, IntoConnectionInfo};
 use serde::{Deserialize, Serialize};
 use typify::import_types;
+
 pub mod sink;
 
-import_types!(schema = "../connector-schemas/redis/connection.json");
+import_types!(schema = "../connector-schemas/redis/connection.json",
+    convert = {
+        {type = "string", format = "var-str"} = VarStr
+    }
+);
 import_types!(schema = "../connector-schemas/redis/table.json");
+
+pub(crate) enum RedisClient {
+    Standard(Client),
+    Clustered(ClusterClient),
+}
+
+impl RedisClient {
+    pub fn new(config: &RedisConfig) -> anyhow::Result<Self> {
+        Ok(match &config.connection {
+            RedisConfigConnection::Address(address) => {
+                let info = from_address(&config, &address.0)?;
+
+                RedisClient::Standard(Client::open(info).map_err(|e| {
+                    anyhow!(
+                        "Failed to construct Redis client for {}: {:?}",
+                        address.0,
+                        e
+                    )
+                })?)
+            }
+            RedisConfigConnection::Addresses(addresses) => {
+                let infos: anyhow::Result<Vec<ConnectionInfo>> = addresses
+                    .iter()
+                    .map(|address| from_address(&config, address))
+                    .collect();
+
+                RedisClient::Clustered(
+                    ClusterClient::new(infos?).map_err(|e| {
+                        anyhow!("Failed to construct Redis Cluster client: {:?}", e)
+                    })?,
+                )
+            }
+        })
+    }
+
+    async fn get_connection(&self) -> Result<GeneralConnection, redis::RedisError> {
+        Ok(match self {
+            RedisClient::Standard(c) => {
+                GeneralConnection::Standard(ConnectionManager::new(c.clone()).await?)
+            }
+            RedisClient::Clustered(c) => {
+                GeneralConnection::Clustered(c.get_async_connection().await?)
+            }
+        })
+    }
+}
+
+fn from_address(config: &RedisConfig, address: &str) -> anyhow::Result<ConnectionInfo> {
+    let mut info: ConnectionInfo = address
+        .to_string()
+        .into_connection_info()
+        .map_err(|e| anyhow!("invalid redis address: {:?}", e))?;
+
+    if let Some(username) = &config.username {
+        info.redis.username = Some(username.sub_env_vars().map_err(|e| anyhow!("{}", e))?);
+    }
+
+    if let Some(password) = &config.password {
+        info.redis.password = Some(password.sub_env_vars().map_err(|e| anyhow!("{}", e))?);
+    }
+
+    Ok(info)
+}
diff --git a/arroyo-worker/src/connectors/redis/sink.rs b/arroyo-worker/src/connectors/redis/sink.rs
index ebbcbf449..8207416fc 100644
--- a/arroyo-worker/src/connectors/redis/sink.rs
+++ b/arroyo-worker/src/connectors/redis/sink.rs
@@ -1,14 +1,14 @@
-use crate::connectors::redis::{RedisConfig, RedisConfigConnection, RedisTable, TableType, Target};
+use crate::connectors::redis::{RedisClient, RedisConfig, RedisTable, TableType, Target};
 use crate::engine::{Context, ErrorReporter, StreamNode};
+
 use arroyo_formats::DataSerializer;
 use arroyo_formats::SchemaData;
 use arroyo_macro::process_fn;
 use arroyo_rpc::OperatorConfig;
 use arroyo_types::{CheckpointBarrier, Key, Record};
 use redis::aio::{ConnectionLike, ConnectionManager};
-use redis::cluster::ClusterClient;
 use redis::cluster_async::ClusterConnection;
-use redis::{Client, Cmd, Pipeline, RedisFuture};
+use redis::{Cmd, Pipeline, RedisFuture};
 use serde::Serialize;
 use serde_json::Value;
 use std::collections::HashSet;
@@ -31,7 +31,7 @@ where
 {
     serializer: DataSerializer<T>,
     table: RedisTable,
-    client: Clients,
+    client: RedisClient,
     cmd_q: Option<(Sender<u32>, Receiver<RedisCmd>)>,
 
     rx: Receiver<u32>,
@@ -61,22 +61,6 @@ enum RedisCmd {
     Flush(u32),
 }
 
-enum Clients {
-    Standard(Client),
-    Clustered(ClusterClient),
-}
-
-impl Clients {
-    async fn get_connection(&self) -> Result<GeneralConnection, redis::RedisError> {
-        Ok(match self {
-            Clients::Standard(c) => {
-                GeneralConnection::Standard(ConnectionManager::new(c.clone()).await?)
-            }
-            Clients::Clustered(c) => GeneralConnection::Clustered(c.get_async_connection().await?),
-        })
-    }
-}
-
 pub enum GeneralConnection {
     Standard(ConnectionManager),
     Clustered(ClusterConnection),
@@ -257,15 +241,7 @@ where
         let table: RedisTable =
             serde_json::from_value(config.table).expect("Invalid table config for Redis");
 
-        let client = match profile.connection {
-            RedisConfigConnection::Address(address) => {
-                Clients::Standard(Client::open(address.0).expect("invalid address"))
-            }
-            RedisConfigConnection::Addresses(addresses) => Clients::Clustered(
-                ClusterClient::new(addresses.into_iter().map(|e| e.0).collect())
-                    .expect("failed to construct cluster client"),
-            ),
-        };
+        let client = RedisClient::new(&profile).expect("Unable to construct redis client");
 
         let (tx, cmd_rx) = tokio::sync::mpsc::channel(128);
         let (cmd_tx, rx) = tokio::sync::mpsc::channel(128);
diff --git a/connector-schemas/redis/connection.json b/connector-schemas/redis/connection.json
index 751d7a40d..0851e2040 100644
--- a/connector-schemas/redis/connection.json
+++ b/connector-schemas/redis/connection.json
@@ -11,6 +11,7 @@
                     "properties": {
                         "address": {
                             "type": "string",
+                            "title": "address",
                             "description": "The address of your Redis server",
                             "examples": ["redis://localhost:6379"],
                             "pattern": "^redis://.*$"
@@ -43,8 +44,24 @@
                     "additionalProperties": false
                 }
             ]
+        },
+
+        "username": {
+            "title": "Username",
+            "type": "string",
+            "description": "The username for your redis cluster (if using auth)",
+            "format": "var-str"
+        },
+        "password": {
+            "title": "Password",
+            "type": "string",
+            "description": "The password for your redis cluster (if using auth)",
+            "format": "var-str"
         }
     },
+    "sensitive": [
+        "password"
+    ],
     "required": [
         "connection"
     ]


Test patch: