collect message transformation stats

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

Author: didier-wenzek

crates/extensions/tedge_gen_mapper/src/actor.rs

@@ -133,6 +133,7 @@ impl GenMapper {
         let timestamp = DateTime::now();
         if timestamp.seconds % 300 == 0 {
             self.processor.dump_memory_stats().await;
+            self.processor.dump_processing_stats().await;
         }
         for (pipeline_id, pipeline_messages) in self.processor.tick(&timestamp).await {
             match pipeline_messages {

crates/extensions/tedge_gen_mapper/src/js_filter.rs

@@ -72,6 +72,10 @@ impl JsFilter {
         &self.path
     }
 
+    pub fn source(&self) -> String {
+        format!("{}", self.path.display())
+    }
+
     /// Process a message returning zero, one or more messages
     ///
     /// The "process" function of the JS module is passed 3 arguments

crates/extensions/tedge_gen_mapper/src/lib.rs

@@ -4,6 +4,7 @@ mod js_filter;
 mod js_runtime;
 pub mod pipeline;
 mod runtime;
+mod stats;
 
 use crate::actor::GenMapper;
 pub use crate::runtime::MessageProcessor;

crates/extensions/tedge_gen_mapper/src/pipeline.rs

@@ -1,5 +1,6 @@
 use crate::js_filter::JsFilter;
 use crate::js_runtime::JsRuntime;
+use crate::stats::Counter;
 use crate::LoadError;
 use camino::Utf8Path;
 use camino::Utf8PathBuf;
@@ -76,6 +77,7 @@ impl Pipeline {
     pub async fn process(
         &mut self,
         js_runtime: &JsRuntime,
+        stats: &mut Counter,
         timestamp: &DateTime,
         message: &Message,
     ) -> Result<Vec<Message>, FilterError> {
@@ -84,38 +86,66 @@ impl Pipeline {
             return Ok(vec![]);
         }
 
+        let stated_at = stats.pipeline_process_start(self.source.as_str());
         let mut messages = vec![message.clone()];
         for stage in self.stages.iter() {
+            let js = stage.filter.source();
             let mut transformed_messages = vec![];
             for message in messages.iter() {
+                let filter_started_at = stats.filter_start(&js, "process");
                 let filter_output = stage.filter.process(js_runtime, timestamp, message).await;
+                match &filter_output {
+                    Ok(messages) => {
+                        stats.filter_done(&js, "process", filter_started_at, messages.len())
+                    }
+                    Err(_) => stats.filter_failed(&js, "process"),
+                }
                 transformed_messages.extend(filter_output?);
             }
             messages = transformed_messages;
         }
+
+        stats.pipeline_process_done(self.source.as_str(), stated_at, messages.len());
         Ok(messages)
     }
 
     pub async fn tick(
         &mut self,
         js_runtime: &JsRuntime,
+        stats: &mut Counter,
         timestamp: &DateTime,
     ) -> Result<Vec<Message>, FilterError> {
+        let stated_at = stats.pipeline_tick_start(self.source.as_str());
         let mut messages = vec![];
         for stage in self.stages.iter() {
+            let js = stage.filter.source();
             // Process first the messages triggered upstream by the tick
             let mut transformed_messages = vec![];
             for message in messages.iter() {
+                let filter_started_at = stats.filter_start(&js, "process");
                 let filter_output = stage.filter.process(js_runtime, timestamp, message).await;
+                match &filter_output {
+                    Ok(messages) => {
+                        stats.filter_done(&js, "process", filter_started_at, messages.len())
+                    }
+                    Err(_) => stats.filter_failed(&js, "process"),
+                }
                 transformed_messages.extend(filter_output?);
             }
 
             // Only then process the tick
-            transformed_messages.extend(stage.filter.tick(js_runtime, timestamp).await?);
+            let filter_started_at = stats.filter_start(&js, "tick");
+            let tick_output = stage.filter.tick(js_runtime, timestamp).await;
+            match &tick_output {
+                Ok(messages) => stats.filter_done(&js, "tick", filter_started_at, messages.len()),
+                Err(_) => stats.filter_failed(&js, "tick"),
+            }
+            transformed_messages.extend(tick_output?);
 
             // Iterate with all the messages collected at this stage
             messages = transformed_messages;
         }
+        stats.pipeline_tick_done(self.source.as_str(), stated_at, messages.len());
         Ok(messages)
     }
 }

crates/extensions/tedge_gen_mapper/src/runtime.rs

@@ -4,6 +4,7 @@ use crate::pipeline::DateTime;
 use crate::pipeline::FilterError;
 use crate::pipeline::Message;
 use crate::pipeline::Pipeline;
+use crate::stats::Counter;
 use crate::LoadError;
 use camino::Utf8Path;
 use camino::Utf8PathBuf;
@@ -21,6 +22,7 @@ pub struct MessageProcessor {
     pub config_dir: PathBuf,
     pub pipelines: HashMap<String, Pipeline>,
     pub(super) js_runtime: JsRuntime,
+    pub stats: Counter,
 }
 
 impl MessageProcessor {
@@ -34,11 +36,13 @@ impl MessageProcessor {
         let mut pipeline_specs = PipelineSpecs::default();
         pipeline_specs.load(&config_dir).await;
         let pipelines = pipeline_specs.compile(&mut js_runtime, &config_dir).await;
+        let stats = Counter::default();
 
         Ok(MessageProcessor {
             config_dir,
             pipelines,
             js_runtime,
+            stats,
         })
     }
 
@@ -52,10 +56,13 @@ impl MessageProcessor {
         let mut pipeline_specs = PipelineSpecs::default();
         pipeline_specs.load_single_pipeline(&pipeline).await;
         let pipelines = pipeline_specs.compile(&mut js_runtime, &config_dir).await;
+        let stats = Counter::default();
+
         Ok(MessageProcessor {
             config_dir,
             pipelines,
             js_runtime,
+            stats,
         })
     }
 
@@ -68,10 +75,13 @@ impl MessageProcessor {
         let mut pipeline_specs = PipelineSpecs::default();
         pipeline_specs.load_single_filter(&filter).await;
         let pipelines = pipeline_specs.compile(&mut js_runtime, &config_dir).await;
+        let stats = Counter::default();
+
         Ok(MessageProcessor {
             config_dir,
             pipelines,
             js_runtime,
+            stats,
         })
     }
 
@@ -88,11 +98,20 @@ impl MessageProcessor {
         timestamp: &DateTime,
         message: &Message,
     ) -> Vec<(String, Result<Vec<Message>, FilterError>)> {
+        let started_at = self.stats.runtime_process_start();
+
         let mut out_messages = vec![];
         for (pipeline_id, pipeline) in self.pipelines.iter_mut() {
-            let pipeline_output = pipeline.process(&self.js_runtime, timestamp, message).await;
+            let pipeline_output = pipeline
+                .process(&self.js_runtime, &mut self.stats, timestamp, message)
+                .await;
+            if pipeline_output.is_err() {
+                self.stats.pipeline_process_failed(pipeline_id);
+            }
             out_messages.push((pipeline_id.clone(), pipeline_output));
         }
+
+        self.stats.runtime_process_done(started_at);
         out_messages
     }
 
@@ -102,12 +121,21 @@ impl MessageProcessor {
     ) -> Vec<(String, Result<Vec<Message>, FilterError>)> {
         let mut out_messages = vec![];
         for (pipeline_id, pipeline) in self.pipelines.iter_mut() {
-            let pipeline_output = pipeline.tick(&self.js_runtime, timestamp).await;
+            let pipeline_output = pipeline
+                .tick(&self.js_runtime, &mut self.stats, timestamp)
+                .await;
+            if pipeline_output.is_err() {
+                self.stats.pipeline_tick_failed(pipeline_id);
+            }
             out_messages.push((pipeline_id.clone(), pipeline_output));
         }
         out_messages
     }
 
+    pub async fn dump_processing_stats(&self) {
+        self.stats.dump_processing_stats();
+    }
+
     pub async fn dump_memory_stats(&self) {
         self.js_runtime.dump_memory_stats().await;
     }