Bridge: variable expansion for config files (#986)

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

~~Since variable expansion might not be assumed, this behavior is gated
by
a CLI flag/env var and is **off by default**.~~

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>();