Merge pull request #4012 from didier-wenzek/feat/configuration-shared-by-the-steps-of-a-flow

feat: The steps of a flow can share configuration values

Author: didier-wenzek

crates/extensions/c8y_mapper_ext/src/flows.rs

@@ -110,10 +110,12 @@ topic = "{errors_topic}"
         format!(
             r#"input.mqtt.topics = {input_topics:?}
 
+config = {{ topic_root = "{topic_prefix}" }}
+
 steps = [
     {{ builtin = "add-timestamp", config = {{ property = "time", format = "unix", reformat = false }} }},
-    {{ builtin = "cache-early-messages", config = {{ topic_root = "{topic_prefix}" }} }},
-    {{ builtin = "into-c8y-measurements", config = {{ topic_root = "{topic_prefix}" }} }},
+    {{ builtin = "cache-early-messages" }},
+    {{ builtin = "into-c8y-measurements" }},
     {{ builtin = "limit-payload-size", config = {{ max_size = {max_size} }} }},
 ]
 
@@ -141,10 +143,12 @@ topic = "{errors_topic}"
         format!(
             r#"input.mqtt.topics = {input_topics:?}
 
+config = {{ topic_root = "{topic_prefix}", c8y_prefix = "{c8y_prefix}" }}
+
 steps = [
     {{ builtin = "add-timestamp", config = {{ property = "time", format = "rfc3339", reformat = false }} }},
-    {{ builtin = "cache-early-messages", config = {{ topic_root = "{topic_prefix}" }} }},
-    {{ builtin = "into-c8y-events", config = {{ topic_root = "{topic_prefix}", c8y_prefix = "{c8y_prefix}", max_mqtt_payload_size = {max_mqtt_payload_size} }} }},
+    {{ builtin = "cache-early-messages" }},
+    {{ builtin = "into-c8y-events", config = {{ max_mqtt_payload_size = {max_mqtt_payload_size} }} }},
 ]
 
 [output.mqtt]
@@ -173,10 +177,12 @@ topic = "{errors_topic}"
         format!(
             r#"input.mqtt.topics = {input_topics:?}
 
+config = {{ topic_root = "{topic_prefix}", c8y_prefix = "{c8y_prefix}" }}
+
 steps = [
     {{ builtin = "add-timestamp", config = {{ property = "time", format = "rfc3339", reformat = false }} }},
-    {{ builtin = "cache-early-messages", config = {{ topic_root = "{topic_prefix}" }} }},
-    {{ builtin = "into-c8y-alarms", interval = "{alarm_interval}", config = {{ topic_root = "{topic_prefix}", c8y_prefix = "{c8y_prefix}" }} }},
+    {{ builtin = "cache-early-messages" }},
+    {{ builtin = "into-c8y-alarms", interval = "{alarm_interval}" }},
     {{ builtin = "limit-payload-size", config = {{ max_size = {max_size} }} }},
 ]
 
@@ -204,9 +210,11 @@ topic = "{errors_topic}"
         format!(
             r#"input.mqtt.topics = {input_topics:?}
 
+config = {{ topic_root = "{topic_prefix}", c8y_prefix = "{c8y_prefix}" }}
+
 steps = [
-    {{ builtin = "cache-early-messages", config = {{ topic_root = "{topic_prefix}" }} }},
-    {{ builtin = "into-c8y-health-status", config = {{ topic_root = "{topic_prefix}", main_device = "{main_device}", c8y_prefix = "{c8y_prefix}" }} }},
+    {{ builtin = "cache-early-messages" }},
+    {{ builtin = "into-c8y-health-status", config = {{ main_device = "{main_device}" }} }},
 ]
 
 [output.mqtt]

crates/extensions/tedge_flows/src/config.rs

@@ -10,6 +10,7 @@ use camino::Utf8Path;
 use camino::Utf8PathBuf;
 use glob::glob;
 use serde::Deserialize;
+use serde_json::Map;
 use serde_json::Value;
 use std::collections::HashMap;
 use std::fmt::Debug;
@@ -28,6 +29,10 @@ pub struct FlowConfig {
     description: Option<String>,
     tags: Option<Vec<String>>,
 
+    /// configuration shared by the steps of this flow
+    #[serde(default)]
+    config: Map<String, Value>,
+
     input: InputConfig,
     #[serde(default)]
     steps: Vec<StepConfig>,
@@ -43,7 +48,7 @@ pub struct StepConfig {
     step: StepSpec,
 
     #[serde(default)]
-    config: Option<Value>,
+    config: Map<String, Value>,
 
     #[serde(default)]
     #[serde(deserialize_with = "parse_optional_human_duration")]
@@ -187,14 +192,15 @@ impl FlowConfig {
         let input_topic = "#".to_string();
         let step = StepConfig {
             step: StepSpec::JavaScript(script),
-            config: None,
+            config: Map::new(),
             interval: None,
         };
         Self {
             name: None,
             version: None,
             description: None,
             tags: None,
+            config: Map::new(),
             input: InputConfig::Mqtt {
                 topics: vec![input_topic],
             },
@@ -216,6 +222,8 @@ impl FlowConfig {
         let mut steps = vec![];
         for (i, step) in self.steps.into_iter().enumerate() {
             let step = step
+                .with_shared_config(&self.config)
+                .with_interval_as_config()
                 .compile(rs_transformers, js_runtime, i, &source)
                 .await?;
             steps.push(step);
@@ -238,6 +246,25 @@ impl FlowConfig {
 }
 
 impl StepConfig {
+    pub fn with_shared_config(mut self, shared_config: &Map<String, Value>) -> Self {
+        for (k, v) in shared_config.iter() {
+            if !self.config.contains_key(k) {
+                self.config.insert(k.clone(), v.clone());
+            }
+        }
+        self
+    }
+
+    pub fn with_interval_as_config(mut self) -> Self {
+        let key = "interval";
+        let interval = self.interval.unwrap_or(Duration::from_secs(1));
+        if !self.config.contains_key(key) {
+            self.config
+                .insert(key.to_string(), interval.as_secs().into());
+        }
+        self
+    }
+
     pub async fn compile(
         &self,
         rs_transformers: &BuiltinTransformers,
@@ -253,8 +280,13 @@ impl StepConfig {
                 Self::instantiate_builtin(rs_transformers, flow, name, index)?
             }
         };
+        let config = if self.config.is_empty() {
+            None
+        } else {
+            Some(Value::Object(self.config.clone()))
+        };
         let step = step
-            .with_config(self.config.clone())?
+            .with_config(config)?
             .with_interval(self.interval, flow.as_str());
         Ok(step)
     }
@@ -388,3 +420,77 @@ fn default_errors() -> OutputConfig {
         topic: Some("te/error".to_string()),
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use serde_json::json;
+
+    #[test]
+    fn inherit_shared_config() {
+        for (shared_config, step_config, merged_config) in [
+            (json!({}), json!({"x": 1, "y": 2}), json!({"x": 1, "y": 2})),
+            (
+                json!({"z": 3}),
+                json!({"x": 1, "y": 2}),
+                json!({"x": 1, "y": 2, "z": 3}),
+            ),
+            (
+                json!({"z": 3, "x": 4}),
+                json!({"x": 1, "y": 2}),
+                json!({"x": 1, "y": 2, "z": 3}),
+            ),
+            (
+                json!({"x": 4}),
+                json!({"x": 1, "y": 2}),
+                json!({"x": 1, "y": 2}),
+            ),
+            (json!({"x": 4}), json!({}), json!({"x": 4})),
+        ] {
+            let shared_config = shared_config.as_object().unwrap();
+            let step_config = step_config.as_object().unwrap();
+            let merged_config = merged_config.as_object().unwrap();
+            let step = StepConfig {
+                step: StepSpec::Transformer("some-step".to_string()),
+                config: step_config.clone(),
+                interval: None,
+            };
+            assert_eq!(
+                &step.with_shared_config(shared_config).config,
+                merged_config
+            );
+        }
+    }
+
+    #[test]
+    fn with_interval_as_config() {
+        for (interval, config, merged_config) in [
+            (None, json!({}), json!({"interval": 1})),
+            (
+                Some(Duration::from_secs(5)),
+                json!({}),
+                json!({"interval": 5}),
+            ),
+            (None, json!({"x": 42}), json!({"interval": 1, "x": 42})),
+            (
+                Some(Duration::from_secs(5)),
+                json!({"x": 42}),
+                json!({"interval": 5, "x": 42}),
+            ),
+            (
+                Some(Duration::from_secs(5)),
+                json!({"interval": 33, "x": 42}),
+                json!({"interval": 33, "x": 42}),
+            ),
+        ] {
+            let config = config.as_object().unwrap();
+            let merged_config = merged_config.as_object().unwrap();
+            let step = StepConfig {
+                step: StepSpec::Transformer("some-step".to_string()),
+                config: config.clone(),
+                interval,
+            };
+            assert_eq!(&step.with_interval_as_config().config, merged_config);
+        }
+    }
+}

crates/extensions/tedge_flows/src/js_value.rs

@@ -84,20 +84,18 @@ impl JsonValue {
         })
     }
 
-    pub fn strings_property(&self, property: &str) -> Vec<&str> {
-        self.property(property)
-            .map(|v| match v {
-                JsonValue::String(string) => vec![string.as_str()],
-                JsonValue::Array(props) => props
-                    .iter()
-                    .filter_map(|v| match v {
-                        JsonValue::String(s) => Some(s.as_str()),
-                        _ => None,
-                    })
-                    .collect(),
-                _ => vec![],
-            })
-            .unwrap_or_default()
+    pub fn strings_property(&self, property: &str) -> Option<Vec<&str>> {
+        self.property(property).map(|v| match v {
+            JsonValue::String(string) => vec![string.as_str()],
+            JsonValue::Array(props) => props
+                .iter()
+                .filter_map(|v| match v {
+                    JsonValue::String(s) => Some(s.as_str()),
+                    _ => None,
+                })
+                .collect(),
+            _ => vec![],
+        })
     }
 
     pub fn number_property(&self, property: &str) -> Option<&Number> {

crates/extensions/tedge_flows/src/transformers/ignore_topics.rs

@@ -18,8 +18,9 @@ impl Transformer for IgnoreTopics {
     }
 
     fn set_config(&mut self, config: JsonValue) -> Result<(), ConfigError> {
-        let topics = config.strings_property("topics");
-        self.topics = crate::config::topic_filters(topics)?;
+        if let Some(topics) = config.strings_property("topics") {
+            self.topics = crate::config::topic_filters(topics)?;
+        }
         Ok(())
     }
 

crates/extensions/tedge_flows/src/transformers/update_context.rs

@@ -26,8 +26,9 @@ impl Transformer for UpdateContext {
     }
 
     fn set_config(&mut self, config: JsonValue) -> Result<(), ConfigError> {
-        let topics = config.strings_property("topics");
-        self.topics = crate::config::topic_filters(topics)?;
+        if let Some(topics) = config.strings_property("topics") {
+            self.topics = crate::config::topic_filters(topics)?;
+        }
         Ok(())
     }
 

docs/src/references/mappers/flows.md

@@ -132,18 +132,18 @@ type Context = {
   script: KVStore,
 
   // A value provided by the flow configuration of that step
-  config: any,
+  config: unknown,
 }
 
 type KVStore = {
   // List the keys for which this store holds a value  
   keys(): string[]
 
   // Get the value attached to a key (returning null, if none)
-  get(key: string): any,
+  get(key: string): unknown,
 
   // Set the value attached to a key (removing the key if the provided value is null)
-  set(key: string, value: any),
+  set(key: string, value: unknown),
 
   // Remove any value attache to a key
   remove(key: string),
@@ -177,11 +177,17 @@ A flow script can also export a `onInterval` function
   - This can also be a TypeScript module with a `.ts` extension.
 - The definition of flow defines its input, output and error sink as well as a list of transformation steps.
 - Each step is built either from a `script` or a `builtin` transformation
-- A step possibly given a config (arbitrary json that will be passed to the transformation script)
+- A step can possibly be given a config (an arbitrary json object that will be passed to the transformation script)
+- Configuration values can also be defined at the flow level,
+  these values will be used as default configuration values by all the steps.
+- The pace at which a step `onInterval` function is called defaults to one second,
+  and can be configured `{ script = "my-script.js", interval = "60s", config = { .. } }`.
 
 ```toml
 input.mqtt.topics = ["te/+/+/+/+/m/+"]
 
+config = { format = "rfc3339" }
+
 steps = [
     { builtin = "add-timestamp" },
     { script = "drop_stragglers.js", config = { max_delay = 60 } },