[Merged by Bors] - feat: add support to string interpolation

connector/sink-test-connector/Connector.toml

@@ -7,12 +7,8 @@ fluvio = "0.10.0"
 description = "Test Sink Connector"
 license = "Apache-2.0"
 
-
 [direction]
 dest = true
 
 [deployment]
 binary = "sink-test-connector"
-
-[secret.TEST_API_KEY]
-type = "env"

connector/sink-test-connector/config-example.yaml

@@ -5,10 +5,12 @@ meta:
   topic: test-topic
   secrets:
     - name: TEST_API_KEY
+    - name: TEST_API_CLIENT_ID
 custom:
   api_key:
     secret:
       name: TEST_API_KEY
+  client_id: ${{ secrets.TEST_API_CLIENT_ID }}
 transforms:
   - uses: infinyon/jolt@0.1.0
     with:

connector/sink-test-connector/secrets.txt

@@ -1 +1,2 @@
 TEST_API_KEY=test value
+TEST_API_CLIENT_ID=client1

connector/sink-test-connector/src/main.rs

@@ -18,4 +18,5 @@ async fn start(config: CustomConfig, mut stream: impl ConsumerStream) -> Result<
 #[connector(config)]
 struct CustomConfig {
     api_key: SecretString,
+    client_id: String,
 }

connector/sink-test-connector/src/sink.rs

@@ -14,7 +14,9 @@ impl TestSink {
     pub(crate) fn new(config: &CustomConfig) -> Result<Self> {
         debug!(?config.api_key);
         let resolved = config.api_key.resolve()?;
+        let resolved_2 = config.client_id.clone();
         debug!(resolved);
+        debug!(resolved_2);
         Ok(Self {})
     }
 }

crates/fluvio-connector-common/src/config.rs

@@ -1,15 +1,14 @@
-pub use fluvio_connector_package::config::ConnectorConfig;
-use tracing::trace;
+use std::io::Read;
 
-use std::{path::PathBuf, fs::File};
-
-use serde::de::DeserializeOwned;
 use anyhow::{Result, Context};
+use serde::de::DeserializeOwned;
 use serde_yaml::Value;
+use tracing::trace;
+
+pub use fluvio_connector_package::config::ConnectorConfig;
 
-pub fn value_from_file<P: Into<PathBuf>>(path: P) -> Result<Value> {
-    let file = File::open(path.into())?;
-    serde_yaml::from_reader(file).context("unable to parse config file into YAML")
+pub fn value_from_reader<R: Read>(reader: R) -> Result<Value> {
+    serde_yaml::from_reader(reader).context("unable to parse config file into YAML")
 }
 
 pub fn from_value<T: DeserializeOwned>(value: Value, root: Option<&str>) -> Result<T> {

crates/fluvio-connector-common/src/lib.rs

@@ -4,6 +4,7 @@ pub mod monitoring;
 pub mod consumer;
 pub mod config;
 
+pub use fluvio_connector_package::render_config_str;
 pub use fluvio_connector_package::secret;
 
 #[cfg(feature = "derive")]

crates/fluvio-connector-derive/src/generator.rs

@@ -113,11 +113,15 @@ fn init_and_parse_config(config_type_path: &Path) -> TokenStream {
 
         ::fluvio_connector_common::tracing::info!("Reading config file from: {}", opts.config.to_string_lossy());
 
-        let config_value = ::fluvio_connector_common::config::value_from_file(opts.config.as_path())?;
-        ::fluvio_connector_common::tracing::trace!("{:#?}", config_value);
+        let config_str = ::std::fs::read_to_string(opts.config.as_path())?;
+        ::fluvio_connector_common::tracing::debug!(%config_str, "input config");
+
+        /// Resolve any secrets/env in the config
+        let config_str_resolved =::fluvio_connector_common::render_config_str(&config_str)?;
+
+        let config_value = ::fluvio_connector_common::config::value_from_reader(config_str_resolved.as_bytes())?;
 
         let common_config = ::fluvio_connector_common::config::ConnectorConfig::from_value(config_value.clone())?;
-        ::fluvio_connector_common::tracing::debug!("{:#?}", common_config);
 
         let user_config: #config_type_path = ::fluvio_connector_common::config::from_value(config_value, Some(#config_type_path::__config_name()))?;
 
@@ -129,6 +133,7 @@ pub(crate) fn generate_connector_config(item: &ConnectorConfigStruct) -> TokenSt
     let config_struct = item.item_struct;
     let ident = &item.item_struct.ident;
     let config_name = &item.config_name;
+
     quote! {
         #[derive(serde::Deserialize)]
         #config_struct
@@ -137,6 +142,7 @@ pub(crate) fn generate_connector_config(item: &ConnectorConfigStruct) -> TokenSt
             pub fn __config_name() -> &'static str {
                 #config_name
             }
+
         }
     }
 }

crates/fluvio-connector-package/Cargo.toml

@@ -14,6 +14,7 @@ default = ["toml"]
 anyhow = { workspace = true }
 bytesize = { workspace = true}
 humantime-serde = "1.1.1"
+minijinja = { git = "https://github.com/mitsuhiko/minijinja" , rev = "c4b48dc5be9de0143e06bef97618c7ed029fb155", default-features = false, features = ["custom_syntax", "fuel"] }
 openapiv3 = { git = "https://github.com/galibey/openapiv3", rev = "bdd22f046d2bc19ede257504645d31f835545222", default-features = false }
 once_cell = { workspace = true }
 serde = { workspace = true, default-features = false, features = ["derive"] }

crates/fluvio-connector-package/src/config.rs

@@ -47,6 +47,12 @@ pub struct MetaConfig {
     pub secrets: Option<Vec<SecretConfig>>,
 }
 
+impl MetaConfig {
+    fn secrets(&self) -> HashSet<SecretConfig> {
+        HashSet::from_iter(self.secrets.clone().unwrap_or_default().into_iter())
+    }
+}
+
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
 pub struct ConsumerParameters {
@@ -173,7 +179,7 @@ impl ConnectorConfig {
     }
 
     pub fn secrets(&self) -> HashSet<SecretConfig> {
-        HashSet::from_iter(self.meta.secrets.clone().unwrap_or_default().into_iter())
+        self.meta.secrets()
     }
 
     pub fn from_value(value: serde_yaml::Value) -> Result<Self> {

crates/fluvio-connector-package/src/lib.rs

@@ -1,3 +1,6 @@
 pub mod metadata;
 pub mod config;
 pub mod secret;
+mod render;
+
+pub use render::render_config_str;

crates/fluvio-connector-package/src/render/context.rs

@@ -0,0 +1,83 @@
+use std::collections::HashMap;
+
+use serde::Serialize;
+use minijinja::value::Value;
+
+use crate::{secret::SecretStore, config::ConnectorConfig};
+
+/// Context for the template engine
+/// This is the data that is available to the template engine
+/// when it is rendering the template.
+#[derive(Serialize, Default)]
+pub(crate) struct Context(pub(crate) HashMap<&'static str, Value>);
+
+/// Template engine does not allow lazy loading of context values.
+/// This trait allows us to load context values on demand.
+pub(crate) trait ContextStore {
+    /// values to add to the context
+    fn extract_context_values(&self, input: &str) -> anyhow::Result<Value>;
+    /// key name for the context
+    fn context_name(&self) -> &'static str;
+}
+impl dyn ContextStore {
+    pub(crate) fn add_to_context(&self, context: &mut Context, input: &str) -> anyhow::Result<()> {
+        let value = self.extract_context_values(input)?;
+
+        context.0.insert(self.context_name(), value);
+        Ok(())
+    }
+}
+impl ContextStore for &dyn SecretStore {
+    fn extract_context_values(&self, input: &str) -> anyhow::Result<Value> {
+        let connector_config: ConnectorConfig = serde_yaml::from_reader(input.as_bytes())?;
+        let mut values = HashMap::default();
+
+        for secret in connector_config.secrets().iter() {
+            let secret_value = self.read(secret.name())?;
+            values.insert(secret.name().to_owned(), secret_value);
+        }
+        Ok(values.into())
+    }
+
+    fn context_name(&self) -> &'static str {
+        "secrets"
+    }
+}
+
+#[cfg(test)]
+mod test {
+
+    use minijinja::value::Value;
+
+    use super::ContextStore;
+
+    pub struct TestStore;
+
+    impl ContextStore for TestStore {
+        fn extract_context_values(&self, _input: &str) -> anyhow::Result<Value> {
+            Ok([("test", "value")].into_iter().collect())
+        }
+
+        fn context_name(&self) -> &'static str {
+            "test"
+        }
+    }
+
+    #[test]
+    fn test_context_store() {
+        let store = &TestStore as &dyn ContextStore;
+        let mut context = super::Context::default();
+        store.add_to_context(&mut context, "").unwrap();
+        assert_eq!(
+            context
+                .0
+                .get("test")
+                .unwrap()
+                .get_attr("test")
+                .unwrap()
+                .as_str()
+                .unwrap(),
+            "value"
+        );
+    }
+}