Bridge: variable expansion for config files

Config files can now use "dollar brace" notation to do variable
substitutions. See the readme for an example.

Furthermore, the variables available for substitution are exposed via a
simple HashMap meaning we can filter the vars exposed in the future, if
requested. This also has the benefit of being nicer for testing ;)

Tests are added to capture the initial behavior of the new dep, but also
to lock in the contract if we decided to "roll our own" in the future.

---

Open source review required.

Adds a dependency on [envsubst-rs](https://github.com/coreos/envsubst-rs)
to provide the base substitution capability.

This was selected based on the following criteria:

- support for the absolute minimum features (no expression evaluation
  for fallbacks, etc).
- Tiny. 170LOC (60+ of which are tests), easy to understand and fork if
  needed.
- Vaguely looks maintained, owned by coreos (part of redhat).

The big idea with adding this dep is, _"it already exists, but rolling
our own based on this if we want to customize behavior is easy."_

Author: svix-onelson

bridge/Cargo.lock

@@ -133,6 +133,27 @@ The config file itself does the heavy lifting.
 
 When unset, the current working directory is checked for a file named `svix-bridge.yaml`.
 
+## Variable Expansion
+
+`svix-bridge` supports environment variable expansion inside the config file.
+
+Using "dollar brace" notation, e.g. `${MY_VARIABLE}`, a substitution will be made from the environment.
+If the variable lookup fails, the raw variable text is left in the config as-is.
+
+As an example, here's a RabbitMQ sender configured with environment variables:
+
+```yaml
+senders:
+  - name: "send-webhooks-from-${QUEUE_NAME}"
+    input:
+      type: "rabbitmq"
+      uri: "${MQ_URI}"
+      queue_name: "${QUEUE_NAME}"
+    output:
+      type: "svix"
+      token: "${SVIX_TOKEN}"
+```
+
 Each sender and receiver can optionally specify a `transformation`.
 Transformations should define a function called `handler` that accepts an object and returns an object.
 

bridge/README.md

@@ -10,6 +10,7 @@ anyhow = "1"
 clap = { version = "4.2.4", features = ["env", "derive"] }
 axum = { version = "0.6", features = ["macros"] }
 enum_dispatch = "0.3"
+envsubst = "0.2.1"
 http = "0.2"
 hyper = { version = "0.14", features = ["full"] }
 lazy_static = "1.4"

bridge/svix-bridge/Cargo.toml

@@ -1,4 +1,7 @@
 use serde::Deserialize;
+use std::borrow::Cow;
+use std::collections::HashMap;
+use std::io::{Error, ErrorKind};
 use std::net::SocketAddr;
 use svix_bridge_plugin_queue::config::{
     into_receiver_output, QueueConsumerConfig, ReceiverOutputOpts as QueueOutOpts,
@@ -26,6 +29,28 @@ pub struct Config {
     pub http_listen_address: SocketAddr,
 }
 
+impl Config {
+    /// Build a Config from yaml source.
+    /// Optionally accepts a map to perform variable substitution with.
+    pub fn from_src(
+        raw_src: &str,
+        vars: Option<&HashMap<String, String>>,
+    ) -> std::io::Result<Self> {
+        let src = if let Some(vars) = vars {
+            Cow::Owned(envsubst::substitute(raw_src, vars).map_err(|e| {
+                Error::new(
+                    ErrorKind::Other,
+                    format!("Variable substitution failed: {e}"),
+                )
+            })?)
+        } else {
+            Cow::Borrowed(raw_src)
+        };
+        serde_yaml::from_str(&src)
+            .map_err(|e| Error::new(ErrorKind::Other, format!("Failed to parse config: {}", e)))
+    }
+}
+
 fn default_http_listen_address() -> SocketAddr {
     "0.0.0.0:5000".parse().expect("default http listen address")
 }

bridge/svix-bridge/src/config/mod.rs

@@ -1,5 +1,8 @@
 use super::Config;
 use crate::config::{LogFormat, LogLevel, SenderConfig};
+use std::collections::HashMap;
+use svix_bridge_plugin_queue::config::{QueueConsumerConfig, RabbitMqInputOpts, SenderInputOpts};
+use svix_bridge_types::{SenderOutputOpts, SvixSenderOutputOpts};
 
 /// This is meant to be a kitchen sink config, hitting as many possible
 /// configuration options as possible to ensure they parse correctly.
@@ -331,3 +334,154 @@ fn test_senders_example() {
     assert!(!conf.senders.is_empty());
     assert!(conf.receivers.is_empty());
 }
+
+#[test]
+fn test_variable_substitution_missing_vars() {
+    let src = r#"
+    opentelemetry: 
+        address: "${OTEL_ADDR}"
+    "#;
+    let vars = HashMap::new();
+    let cfg = Config::from_src(src, Some(&vars)).unwrap();
+    let otel = cfg.opentelemetry.unwrap();
+    // when lookups in the vars map fail, the original token text is preserved.
+    assert_eq!(&otel.address, "${OTEL_ADDR}");
+}
+
+#[test]
+fn test_variable_substitution_available_vars() {
+    let src = r#"
+    opentelemetry:
+        address: "${OTEL_ADDR}"
+        sample_ratio: ${OTEL_SAMPLE_RATIO}
+    "#;
+    let mut vars = HashMap::new();
+    vars.insert(
+        String::from("OTEL_ADDR"),
+        String::from("http://127.0.0.1:8080"),
+    );
+    vars.insert(String::from("OTEL_SAMPLE_RATIO"), String::from("0.25"));
+    let cfg = Config::from_src(src, Some(&vars)).unwrap();
+    // when lookups succeed, the token should be replaced.
+    let otel = cfg.opentelemetry.unwrap();
+    assert_eq!(&otel.address, "http://127.0.0.1:8080");
+    assert_eq!(otel.sample_ratio, Some(0.25));
+}
+
+#[test]
+fn test_variable_substitution_requires_braces() {
+    let src = r#"
+    opentelemetry:
+        # Neglecting to use ${} notation means the port number will not be substituted. 
+        address: "${OTEL_SCHEME}://${OTEL_HOST}:$OTEL_PORT"
+    "#;
+    let mut vars = HashMap::new();
+    vars.insert(String::from("OTEL_SCHEME"), String::from("https"));
+    vars.insert(String::from("OTEL_HOST"), String::from("127.0.0.1"));
+    vars.insert(String::from("OTEL_PORT"), String::from("9999"));
+    let cfg = Config::from_src(src, Some(&vars)).unwrap();
+    // when lookups succeed, the token should be replaced.
+    let otel = cfg.opentelemetry.unwrap();
+    // Not the user-intended outcome, but it simplifies the parsing requirements.
+    assert_eq!(&otel.address, "https://127.0.0.1:$OTEL_PORT");
+}
+
+#[test]
+fn test_variable_substitution_missing_numeric_var_is_err() {
+    // Unfortunate side-effect of templating yaml.
+    //
+    // If the variable is missing, usually you've got three options:
+    // - retain the token text that failed the lookup (envsubst-rs does this)
+    // - replace the token with an empty string (the CLI `envsubst` does this)
+    // - mark it an error (neither do this, but we can if we roll our own impl)
+    //
+    // For yaml, the field typings are heavily/poorly inferred so for an optional float like
+    // `sample_ratio` an empty string would parse as a `None`, which could be a bad fallback since
+    // otel considers this a 1.0 ratio (send everything).
+    //
+    // For this specific case, retaining the token text produces an error, which happens to be useful.
+    // For fields that happen to be strings anyway, errors may show up later (after the config parsing).
+    // Ex: using `${QUEUE_NAME}` in a rabbit sender input will surface in logs as an error when we
+    // try to connect: "no such queue '${QUEUE_NAME}'".
+
+    let src = r#"
+    opentelemetry:
+        address: "${OTEL_ADDR}"
+        # This var will be missing, causing the template token to
+        # be retained causing a parse failure :(
+        sample_ratio: ${OTEL_SAMPLE_RATIO}
+    "#;
+    let vars = HashMap::new();
+    let err = Config::from_src(src, Some(&vars)).err().unwrap();
+    let want = "Failed to parse config: opentelemetry.sample_ratio: invalid type: \
+                    string \"${OTEL_SAMPLE_RATIO}\", expected f64 at line 6 column 23";
+    assert_eq!(want, err.to_string());
+}
+
+#[test]
+fn test_variable_substitution_repeated_lookups() {
+    // This is probably a given, but we should expect a single variable can be referenced multiple
+    // times within the config.
+    // The concrete use case: auth tokens.
+
+    let src = r#"
+    senders:
+      - name: "rabbitmq-1"
+        input:
+          type: "rabbitmq"
+          uri: "${RABBIT_URI}"
+          queue_name: "${QUEUE_NAME_1}"
+        output:
+          type: "svix"
+          token: "${SVIX_TOKEN}"
+      - name: "rabbitmq-2"
+        input:
+          type: "rabbitmq"
+          uri: "${RABBIT_URI}"
+          queue_name: "${QUEUE_NAME_2}"
+        output:
+          type: "svix"
+          token: "${SVIX_TOKEN}"
+    "#;
+    let mut vars = HashMap::new();
+    vars.insert(
+        String::from("RABBIT_URI"),
+        String::from("amqp://guest:guest@localhost:5672/%2f"),
+    );
+    vars.insert(String::from("QUEUE_NAME_1"), String::from("one"));
+    vars.insert(String::from("QUEUE_NAME_2"), String::from("two"));
+    vars.insert(String::from("SVIX_TOKEN"), String::from("x"));
+    let cfg = Config::from_src(src, Some(&vars)).unwrap();
+
+    if let SenderConfig::QueueConsumer(QueueConsumerConfig {
+        input:
+            SenderInputOpts::RabbitMQ(RabbitMqInputOpts {
+                uri, queue_name, ..
+            }),
+        output: SenderOutputOpts::Svix(SvixSenderOutputOpts { token, .. }),
+        ..
+    }) = &cfg.senders[0]
+    {
+        assert_eq!(uri, "amqp://guest:guest@localhost:5672/%2f");
+        assert_eq!(queue_name, "one");
+        assert_eq!(token, "x");
+    } else {
+        panic!("sender did not match expected pattern");
+    }
+
+    if let SenderConfig::QueueConsumer(QueueConsumerConfig {
+        input:
+            SenderInputOpts::RabbitMQ(RabbitMqInputOpts {
+                uri, queue_name, ..
+            }),
+        output: SenderOutputOpts::Svix(SvixSenderOutputOpts { token, .. }),
+        ..
+    }) = &cfg.senders[1]
+    {
+        assert_eq!(uri, "amqp://guest:guest@localhost:5672/%2f");
+        assert_eq!(queue_name, "two");
+        assert_eq!(token, "x");
+    } else {
+        panic!("sender did not match expected pattern");
+    }
+}

bridge/svix-bridge/src/config/tests.rs

@@ -155,18 +155,23 @@ pub struct Args {
 async fn main() -> Result<()> {
     let args = Args::parse();
 
-    let config = args.cfg.unwrap_or_else(|| {
+    let config_path = args.cfg.unwrap_or_else(|| {
         std::env::current_dir()
             .expect("current dir")
             .join("svix-bridge.yaml")
     });
-    let cfg: Config = serde_yaml::from_str(&std::fs::read_to_string(&config).map_err(|e| {
-        let p = config.into_os_string().into_string().expect("config path");
+
+    let cfg_source = std::fs::read_to_string(&config_path).map_err(|e| {
+        let p = config_path
+            .into_os_string()
+            .into_string()
+            .expect("config path");
         Error::new(ErrorKind::Other, format!("Failed to read {p}: {e}"))
-    })?)
-    .map_err(|e| Error::new(ErrorKind::Other, format!("Failed to parse config: {}", e)))?;
-    setup_tracing(&cfg);
+    })?;
 
+    let vars = std::env::vars().collect();
+    let cfg = Config::from_src(&cfg_source, Some(vars).as_ref())?;
+    setup_tracing(&cfg);
     tracing::info!("starting");
 
     let (xform_tx, mut xform_rx) = tokio::sync::mpsc::unbounded_channel::<TransformerJob>();