Tedge flows statistics are published over MQTT

Signed-off-by: Didier Wenzek <[email protected]>

Author: didier-wenzek

crates/extensions/tedge_flows/src/actor.rs

@@ -6,6 +6,7 @@ use crate::flow::Message;
 use crate::flow::SourceTag;
 use crate::registry::FlowRegistryExt;
 use crate::runtime::MessageProcessor;
+use crate::stats::MqttStatsPublisher;
 use crate::InputMessage;
 use crate::Tick;
 use async_trait::async_trait;
@@ -48,6 +49,7 @@ pub struct FlowsMapper {
     pub(super) watched_commands: HashSet<String>,
     pub(super) processor: MessageProcessor<ConnectedFlowRegistry>,
     pub(super) next_dump: Instant,
+    pub(super) stats_publisher: MqttStatsPublisher,
 }
 
 #[async_trait]
@@ -258,7 +260,13 @@ impl FlowsMapper {
         let timestamp = SystemTime::now();
         if self.next_dump <= now {
             self.processor.dump_memory_stats().await;
-            self.processor.dump_processing_stats().await;
+            for record in self
+                .processor
+                .dump_processing_stats(&self.stats_publisher)
+                .await
+            {
+                self.mqtt_sender.send(record).await?;
+            }
             self.next_dump = now + STATS_DUMP_INTERVAL;
         }
         for messages in self.processor.on_interval(timestamp, now).await {

crates/extensions/tedge_flows/src/lib.rs

@@ -18,6 +18,7 @@ pub use crate::flow::*;
 pub use crate::registry::BaseFlowRegistry;
 pub use crate::registry::FlowRegistryExt;
 pub use crate::runtime::MessageProcessor;
+use crate::stats::MqttStatsPublisher;
 use camino::Utf8Path;
 use std::collections::HashSet;
 use std::convert::Infallible;
@@ -131,6 +132,9 @@ impl Builder<FlowsMapper> for FlowsMapperBuilder {
     fn build(self) -> FlowsMapper {
         let subscriptions = self.topics().clone();
         let watched_commands = HashSet::new();
+        let stats_publisher = MqttStatsPublisher {
+            topic_prefix: "te/device/main/service/tedge-flows/stats".to_string(),
+        };
         FlowsMapper {
             messages: self.message_box.build(),
             mqtt_sender: self.mqtt_sender,
@@ -139,6 +143,7 @@ impl Builder<FlowsMapper> for FlowsMapperBuilder {
             watched_commands,
             processor: self.processor,
             next_dump: Instant::now() + STATS_DUMP_INTERVAL,
+            stats_publisher,
         }
     }
 }

crates/extensions/tedge_flows/src/runtime.rs

@@ -5,6 +5,7 @@ use crate::js_runtime::JsRuntime;
 use crate::registry::BaseFlowRegistry;
 use crate::registry::FlowRegistryExt;
 use crate::stats::Counter;
+use crate::stats::StatsPublisher;
 use crate::LoadError;
 use camino::Utf8Path;
 use camino::Utf8PathBuf;
@@ -135,8 +136,8 @@ impl<Registry: FlowRegistryExt + Send> MessageProcessor<Registry> {
         out_messages
     }
 
-    pub async fn dump_processing_stats(&self) {
-        self.stats.dump_processing_stats();
+    pub async fn dump_processing_stats<P: StatsPublisher>(&self, publisher: &P) -> Vec<P::Record> {
+        self.stats.dump_processing_stats(publisher)
     }
 
     pub async fn dump_memory_stats(&self) {

crates/extensions/tedge_flows/src/stats.rs

@@ -1,7 +1,10 @@
+use serde_json::Value;
 use std::collections::HashMap;
 use std::fmt::Display;
 use std::time::Duration;
 use std::time::Instant;
+use tedge_mqtt_ext::MqttMessage;
+use tedge_mqtt_ext::Topic;
 
 #[derive(Default)]
 pub struct Counter {
@@ -113,11 +116,12 @@ impl Counter {
         self.from_start.entry(dim).or_default().add(sample);
     }
 
-    pub fn dump_processing_stats(&self) {
+    pub fn dump_processing_stats<P: StatsPublisher>(&self, publisher: &P) -> Vec<P::Record> {
         tracing::info!(target: "flows", "Processing statistics:");
-        for (dim, stats) in &self.from_start {
-            stats.dump_statistics(dim)
-        }
+        self.from_start
+            .iter()
+            .filter_map(|(dim, stats)| stats.dump_statistics(dim, publisher))
+            .collect()
     }
 }
 
@@ -140,15 +144,27 @@ impl Stats {
         }
     }
 
-    pub fn dump_statistics(&self, dim: &Dimension) {
-        tracing::info!(target: "flows", "    - {dim}");
-        tracing::info!(target: "flows", "         - input count: {}", self.messages_in);
-        tracing::info!(target: "flows", "         - output count: {}", self.messages_out);
-        tracing::info!(target: "flows", "         - error count: {}", self.error_raised);
-        if let Some(duration_stats) = &self.processing_time {
-            tracing::info!(target: "flows", "         - min processing time: {:?}", duration_stats.min);
-            tracing::info!(target: "flows", "         - max processing time: {:?}", duration_stats.max);
-        }
+    pub fn dump_statistics<P: StatsPublisher>(
+        &self,
+        dim: &Dimension,
+        publisher: &P,
+    ) -> Option<P::Record> {
+        let stats = match self.processing_time.as_ref() {
+            None => serde_json::json!({
+                "input": self.messages_in,
+                "output": self.messages_out,
+                "error": self.error_raised,
+            }),
+            Some(duration_stats) => serde_json::json!({
+                "input": self.messages_in,
+                "output": self.messages_out,
+                "error": self.error_raised,
+                "cpu-min": format!("{:?}", duration_stats.min),
+                "cpu-max": format!("{:?}", duration_stats.max),
+            }),
+        };
+
+        publisher.publish(dim, stats)
     }
 }
 
@@ -192,3 +208,59 @@ impl Dimension {
         }
     }
 }
+
+pub trait StatsPublisher {
+    type Record;
+
+    fn publish(&self, dim: &Dimension, stats: serde_json::Value) -> Option<Self::Record>;
+}
+
+pub struct TracingStatsPublisher;
+
+impl StatsPublisher for TracingStatsPublisher {
+    type Record = ();
+
+    fn publish(&self, dim: &Dimension, stats: Value) -> Option<()> {
+        tracing::info!(target: "flows", "  - {dim}");
+        if let Some(stats) = stats.as_object() {
+            for (k, v) in stats {
+                tracing::info!(target: "flows", "    - {k}: {v}");
+            }
+        }
+        None
+    }
+}
+
+pub struct MqttStatsPublisher {
+    pub topic_prefix: String,
+}
+
+impl StatsPublisher for MqttStatsPublisher {
+    type Record = MqttMessage;
+
+    fn publish(&self, dim: &Dimension, stats: Value) -> Option<Self::Record> {
+        match dim {
+            Dimension::Flow(path) | Dimension::OnMessage(path) => {
+                self.topic_for(path).map(|topic| {
+                    let payload = stats.to_string();
+                    MqttMessage::new(&topic, payload)
+                })
+            }
+
+            Dimension::Runtime => self.topic_for("runtime").map(|topic| {
+                let payload = stats.to_string();
+                MqttMessage::new(&topic, payload)
+            }),
+
+            _ => None,
+        }
+    }
+}
+
+impl MqttStatsPublisher {
+    pub fn topic_for(&self, path: &str) -> Option<Topic> {
+        let name = path.split('/').last().unwrap();
+        let topic = format!("{}/{}", self.topic_prefix, name);
+        Topic::new(&topic).ok()
+    }
+}

crates/extensions/tedge_flows/tests/interval.rs

@@ -370,6 +370,7 @@ async fn interval_executes_when_time_exceeds_interval() {
     let count = || {
         captured_messages
             .retain(|msg| !msg.topic.as_ref().contains("status"))
+            .retain(|msg| !msg.topic.as_ref().contains("stats"))
             .count()
     };