Support environment variable substitution

arroyo-connectors/src/lib.rs

@@ -3,7 +3,7 @@ use arroyo_rpc::api_types::connections::{
     ConnectionProfile, ConnectionSchema, ConnectionType, FieldType, SourceField, SourceFieldType,
 };
 use arroyo_rpc::primitive_to_sql;
-use arroyo_types::string_to_map;
+use arroyo_types::{string_to_map, substitute_env_vars};
 use axum::response::sse::Event;
 use blackhole::BlackholeConnector;
 use fluvio::FluvioConnector;
@@ -344,8 +344,10 @@ fn construct_http_client(endpoint: &str, headers: Option<&String>) -> anyhow::Re
                 Ok((
                     TryInto::<HeaderName>::try_into(&k)
                         .map_err(|_| anyhow!("invalid header name {}", k))?,
-                    TryInto::<HeaderValue>::try_into(&v)
-                        .map_err(|_| anyhow!("invalid header value {}", v))?,
+                    TryInto::<HeaderValue>::try_into(
+                        substitute_env_vars(&v).map_err(|e| anyhow!(e))?,
+                    )
+                    .map_err(|_| anyhow!("invalid header value {}", v))?,
                 ))
             })
             .collect();

arroyo-connectors/src/polling_http.rs

@@ -126,7 +126,7 @@ impl Connector for PollingHTTPConnector {
                 Err(err) => TestSourceMessage {
                     error: true,
                     done: true,
-                    message: format!("{:?}", err),
+                    message: format!("{:?}", err.root_cause()),
                 },
             };
 

arroyo-types/Cargo.toml

@@ -8,3 +8,4 @@ bincode = "2.0.0-rc.3"
 serde = { version = "1.0", features = ["derive"] }
 arrow = { workspace = true }
 arrow-array = { workspace = true }
+regex = "1.10.2"

arroyo-types/src/lib.rs

@@ -1,6 +1,7 @@
 use arrow::datatypes::SchemaRef;
 use arrow_array::RecordBatch;
 use bincode::{config, Decode, Encode};
+use regex::Regex;
 use serde::ser::SerializeStruct;
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
@@ -807,6 +808,7 @@ pub enum DateTruncPrecision {
 }
 
 use std::convert::TryFrom;
+use std::sync::OnceLock;
 
 impl TryFrom<&str> for DatePart {
     type Error = String;
@@ -867,6 +869,28 @@ pub fn range_for_server(i: usize, n: usize) -> RangeInclusive<u64> {
     start..=end
 }
 
+pub fn substitute_env_vars(input: &str) -> Result<String, String> {
+    // Regex to match patterns like {{ VAR_NAME }}
+    static RE: OnceLock<Regex> = OnceLock::new();
+    let re = RE.get_or_init(|| Regex::new(r"\{\{\s*(\w+)\s*\}\}").unwrap());
+
+    let mut result = input.to_string();
+
+    for caps in re.captures_iter(input) {
+        let var_name = caps.get(1).unwrap().as_str();
+        let full_match = caps.get(0).unwrap().as_str();
+
+        match env::var(var_name) {
+            Ok(value) => {
+                result = result.replace(full_match, &value);
+            }
+            Err(_) => return Err(format!("Environment variable {} not found", var_name)),
+        }
+    }
+
+    Ok(result)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -903,4 +927,27 @@ mod tests {
             "u64::MAX is not in the correct range"
         );
     }
+
+    #[test]
+    fn test_no_placeholders() {
+        let input = "This is a test string with no placeholders";
+        assert_eq!(substitute_env_vars(input).unwrap(), input);
+    }
+
+    #[test]
+    fn test_with_placeholders() {
+        env::set_var("TEST_VAR", "environment variable");
+        let input = "This is a {{ TEST_VAR }}";
+        let expected = "This is a environment variable";
+        assert_eq!(substitute_env_vars(input).unwrap(), expected);
+    }
+
+    #[test]
+    fn test_multiple_placeholders() {
+        env::set_var("VAR1", "first");
+        env::set_var("VAR2", "second");
+        let input = "Here is the {{ VAR1 }} and here is the {{ VAR2 }}";
+        let expected = "Here is the first and here is the second";
+        assert_eq!(substitute_env_vars(input).unwrap(), expected);
+    }
 }

arroyo-worker/src/connectors/polling_http.rs

@@ -9,7 +9,7 @@ use std::{marker::PhantomData, time::Duration};
 use arroyo_macro::source_fn;
 use arroyo_rpc::ControlMessage;
 use arroyo_rpc::{grpc::TableDescriptor, OperatorConfig};
-use arroyo_types::{string_to_map, Message, Record, UserError, Watermark};
+use arroyo_types::{string_to_map, substitute_env_vars, Message, Record, UserError, Watermark};
 
 use serde::{Deserialize, Serialize};
 use tokio::select;
@@ -73,7 +73,8 @@ where
                 (
                     (&k).try_into()
                         .expect(&format!("invalid header name {}", k)),
-                    (&v).try_into()
+                    (substitute_env_vars(&v).unwrap())
+                        .try_into()
                         .expect(&format!("invalid header value {}", v)),
                 )
             })

connector-schemas/polling_http/table.json

@@ -12,7 +12,7 @@
     "headers": {
       "title": "Headers",
       "type": "string",
-      "description": "Comma separated list of headers to send with the request",
+      "description": "Comma separated list of headers to send with the request. May contain [variables](https://doc.arroyo.dev/todo).",
       "pattern": "([a-zA-Z0-9-]+: ?.+,)*([a-zA-Z0-9-]+: ?.+)",
       "examples": ["Authentication: digest 1234,Content-Type: application/json"]
     },