Add support for stateful filters

Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>

Author: didier-wenzek

crates/extensions/tedge_wasm_mapper/Cargo.toml

@@ -10,7 +10,7 @@ repository.workspace = true
 
 [dependencies]
 async-trait = { workspace = true }
-camino = { workspace = true }
+camino = { workspace = true, features = ["serde1"] }
 serde = { workspace = true, features = ["derive"] }
 tedge_actors = { workspace = true }
 tedge_mqtt_ext = { workspace = true }

crates/extensions/tedge_wasm_mapper/components/Cargo.lock

@@ -7,9 +7,6 @@ edition = "2021"
 serde_json = { workspace = true }
 time = { workspace = true }
 wit-bindgen = { workspace = true }
-wit-bindgen-rt = { workspace = true }
 
 [lib]
 crate-type = ["cdylib"]
-
-

crates/extensions/tedge_wasm_mapper/components/rust/add_timestamp/Cargo.toml

@@ -46,10 +46,12 @@ impl Guest for Component {
         Ok(vec![updated_message])
     }
 
+    /// Not configurable
     fn update_config(_config: Message) -> Result<(), FilterError> {
         Ok(())
     }
 
+    /// Stateless
     fn tick(_timestamp: Datetime) -> Result<Vec<Message>, FilterError> {
         Ok(vec![])
     }

crates/extensions/tedge_wasm_mapper/components/rust/add_timestamp/src/lib.rs

@@ -0,0 +1,13 @@
+[package]
+name = "group_by_timestamp"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+serde = { workspace = true }
+serde_json = { workspace = true }
+time = { workspace = true }
+wit-bindgen = { workspace = true }
+
+[lib]
+crate-type = ["cdylib"]

crates/extensions/tedge_wasm_mapper/components/rust/group_by_timestamp/Cargo.toml

@@ -0,0 +1,207 @@
+wit_bindgen::generate!({
+    world: "tedge",
+    path: "../../../wit/filter.wit",
+});
+
+use crate::exports::tedge::filter::filtering::Datetime;
+use crate::exports::tedge::filter::filtering::Filter;
+use crate::exports::tedge::filter::filtering::FilterError;
+use crate::exports::tedge::filter::filtering::Guest;
+use crate::exports::tedge::filter::filtering::GuestFilter;
+use crate::exports::tedge::filter::filtering::Message;
+use serde::Deserialize;
+use serde::Serialize;
+use std::cell::RefCell;
+use std::collections::HashMap;
+use time::format_description::well_known::Rfc3339;
+use time::OffsetDateTime;
+
+pub struct Component;
+
+export!(Component);
+
+impl Guest for Component {
+    type Filter = GroupByTimestamp;
+
+    fn new_filter(config: Message) -> Filter {
+        Filter::new(GroupByTimestamp::new(config))
+    }
+}
+
+/// Group thin-edge measurements occurring the same time-window
+///
+/// This POC ignores duplicates and simply takes the latest value
+pub struct GroupByTimestamp {
+    time_window_secs: u64,
+    grouped_messages: RefCell<HashMap<u64, HashMap<Topic, Measurements>>>,
+}
+
+type Topic = String;
+
+impl GuestFilter for GroupByTimestamp {
+    fn new(_config: Message) -> Self {
+        GroupByTimestamp {
+            time_window_secs: 60,
+            grouped_messages: RefCell::new(HashMap::new()),
+        }
+    }
+
+    fn process(
+        &self,
+        _ingestion_timestamp: Datetime,
+        message: Message,
+    ) -> Result<Vec<Message>, FilterError> {
+        let Ok(measurements) = serde_json::from_str::<Measurements>(&message.payload) else {
+            return Err(FilterError::UnsupportedMessage(
+                "Expect thin-edge measurements".to_string(),
+            ));
+        };
+
+        let Some(measurement_timestamp) = &measurements.time else {
+            return Err(FilterError::UnsupportedMessage(
+                "Missing timestamp".to_string(),
+            ));
+        };
+
+        let Ok(unix_timestamp) = OffsetDateTime::parse(measurement_timestamp, &Rfc3339)
+            .map(|dt| dt.unix_timestamp() as u64)
+        else {
+            return Err(FilterError::UnsupportedMessage(
+                "Expect Rfc3339 timestamp".to_string(),
+            ));
+        };
+
+        let measurement_topic = message.topic.clone();
+        let time_window_secs = (unix_timestamp / self.time_window_secs) * self.time_window_secs;
+        self.grouped_messages
+            .borrow_mut()
+            .entry(time_window_secs)
+            .or_default()
+            .entry(measurement_topic)
+            .or_default()
+            .merge(measurements);
+
+        Ok(vec![])
+    }
+
+    fn update_config(&self, _config: Message) -> Result<(), FilterError> {
+        Ok(())
+    }
+
+    fn tick(&self, timestamp: Datetime) -> Result<Vec<Message>, FilterError> {
+        let time_window_secs = (timestamp.seconds / self.time_window_secs) * self.time_window_secs;
+        let mut elapsed = vec![];
+        for t in self.grouped_messages.borrow().keys() {
+            if *t < time_window_secs {
+                elapsed.push(*t);
+            }
+        }
+
+        let mut all_elapsed_messages = vec![];
+        for t in elapsed {
+            if let Some(elapsed_groups) = self.grouped_messages.borrow_mut().remove(&t) {
+                let mut elapsed_messages = GroupByTimestamp::into_messages(t, elapsed_groups)?;
+                all_elapsed_messages.append(&mut elapsed_messages);
+            }
+        }
+        Ok(all_elapsed_messages)
+    }
+}
+
+impl GroupByTimestamp {
+    fn into_messages(
+        unix_timestamp: u64,
+        group: HashMap<Topic, Measurements>,
+    ) -> Result<Vec<Message>, FilterError> {
+        let mut elapsed_messages = vec![];
+        let ingestion_time = time::OffsetDateTime::from_unix_timestamp(unix_timestamp as i64)
+            .map_err(|err| {
+                FilterError::IncorrectSetting(format!("Fail to format unix timestamp: {err}"))
+            })?
+            .format(&Rfc3339)
+            .map_err(|err| {
+                FilterError::IncorrectSetting(format!("Fail to format timestamp as Rfc3339: {err}"))
+            })?;
+
+        for (topic, mut measurements) in group {
+            measurements.time = Some(ingestion_time.clone());
+            let payload = serde_json::to_string(&measurements).map_err(|err| {
+                FilterError::IncorrectSetting(format!("Fail to format measurements as JSON: {err}"))
+            })?;
+            elapsed_messages.push(Message { topic, payload })
+        }
+        Ok(elapsed_messages)
+    }
+}
+
+#[derive(Deserialize, Serialize, Default)]
+struct Measurements {
+    #[serde(skip_serializing_if = "Option::is_none")]
+    time: Option<String>,
+
+    #[serde(flatten)]
+    pub extras: HashMap<String, Measurement>,
+}
+
+#[derive(Deserialize, Serialize)]
+#[serde(untagged)]
+enum Measurement {
+    Number(serde_json::Number),
+    Text(String),
+    Group(HashMap<String, serde_json::Number>),
+}
+
+impl Measurements {
+    /// Merge into this set of measurements a new set of measurements
+    ///
+    /// Returns all the measurements that cannot be merged, if any
+    pub fn merge(&mut self, other: Measurements) -> Option<Measurements> {
+        let mut rejected_vals = HashMap::new();
+        for (k, new_val) in other.extras {
+            match self.extras.get_mut(k.as_str()) {
+                None => {
+                    self.extras.insert(k, new_val);
+                }
+                Some(old_val) => {
+                    if let Some(rejected_val) = old_val.merge(new_val) {
+                        rejected_vals.insert(k, rejected_val);
+                    }
+                }
+            };
+        }
+
+        if rejected_vals.is_empty() {
+            None
+        } else {
+            Some(Measurements {
+                time: other.time.clone(),
+                extras: rejected_vals,
+            })
+        }
+    }
+}
+
+impl Measurement {
+    /// Merge this measurement with a new one
+    ///
+    /// Returns the new measurement if it cannot be merged
+    pub fn merge(&mut self, other: Measurement) -> Option<Measurement> {
+        match (self, other) {
+            (Measurement::Number(ref mut left), Measurement::Number(right)) => {
+                *left = right;
+                None
+            }
+            (Measurement::Text(ref mut left), Measurement::Text(right)) => {
+                *left = right;
+                None
+            }
+            (Measurement::Group(ref mut left), Measurement::Group(right)) => {
+                for (k, v) in right {
+                    left.insert(k, v);
+                }
+                None
+            }
+            (_, other) => Some(other),
+        }
+    }
+}

crates/extensions/tedge_wasm_mapper/components/rust/group_by_timestamp/src/lib.rs

@@ -7,7 +7,6 @@ edition = "2021"
 serde = { workspace = true }
 serde_json = { workspace = true }
 wit-bindgen = { workspace = true }
-wit-bindgen-rt = { workspace = true }
 
 [lib]
 crate-type = ["cdylib"]

crates/extensions/tedge_wasm_mapper/components/rust/te_to_c8y/Cargo.toml

@@ -1,5 +1,4 @@
 use serde::Deserialize;
-use serde::Serialize;
 use serde_json::json;
 use std::collections::HashMap;
 
@@ -96,7 +95,7 @@ impl Guest for Component {
     }
 }
 
-#[derive(Deserialize, Serialize)]
+#[derive(Deserialize)]
 struct Measurements {
     #[serde(skip_serializing_if = "Option::is_none")]
     time: Option<String>,
@@ -105,7 +104,7 @@ struct Measurements {
     pub extras: HashMap<String, Measurement>,
 }
 
-#[derive(Deserialize, Serialize)]
+#[derive(Deserialize)]
 #[serde(untagged)]
 enum Measurement {
     Number(serde_json::Number),

crates/extensions/tedge_wasm_mapper/components/rust/te_to_c8y/src/lib.rs

@@ -54,7 +54,7 @@ impl StageConfig {
         };
         let config_topics = topic_filters(&self.config_topics)?;
         Ok(Stage {
-            filter: Box::new(filter),
+            filter,
             config_topics,
         })
     }

crates/extensions/tedge_wasm_mapper/src/config.rs

@@ -1,11 +1,15 @@
+use crate::pipeline::Filter;
 use crate::wasm::Datetime;
 use crate::wasm::Message;
 use crate::wasm::TransformedMessages;
 use crate::wasm::WasmFilter;
+use crate::wasm_filter::WasmFilterResource;
 use crate::LoadError;
 use camino::Utf8Path;
 use camino::Utf8PathBuf;
 use std::collections::HashMap;
+use tedge_mqtt_ext::MqttMessage;
+use tedge_mqtt_ext::Topic;
 use wasmtime::component::Component;
 use wasmtime::component::Linker;
 use wasmtime::component::ResourceTable;
@@ -43,7 +47,19 @@ impl HostEngine {
         Ok(())
     }
 
-    pub fn instantiate(&self, component: &Utf8Path) -> Result<WasmFilter, LoadError> {
+    pub fn instantiate(&self, component: &Utf8Path) -> Result<Box<dyn Filter>, LoadError> {
+        self.instantiate_filter(component)
+            .map(WasmFilter::into_dyn)
+            .or_else(|_| {
+                // FIXME: should read the config file
+                let config_topic = Topic::new_unchecked("config");
+                let no_config = MqttMessage::new(&config_topic, "");
+                self.instantiate_resource(component, &no_config)
+                    .map(WasmFilterResource::into_dyn)
+            })
+    }
+
+    pub fn instantiate_filter(&self, component: &Utf8Path) -> Result<WasmFilter, LoadError> {
         let Some(component) = self.components.get(component) else {
             return Err(LoadError::FileNotFound {
                 path: component.into(),
@@ -62,6 +78,22 @@ impl HostEngine {
 
         Ok(WasmFilter::new(store, process_func))
     }
+
+    pub fn instantiate_resource(
+        &self,
+        component: &Utf8Path,
+        config: &MqttMessage,
+    ) -> Result<WasmFilterResource, LoadError> {
+        let Some(component) = self.components.get(component) else {
+            return Err(LoadError::FileNotFound {
+                path: component.into(),
+            });
+        };
+
+        let state = HostState::default();
+        let store = Store::new(&self.engine, state);
+        WasmFilterResource::try_new(store, component, &self.linker, config)
+    }
 }
 
 pub struct HostState {