Merge pull request #3863 from didier-wenzek/feat/tedge-flows-timestamp

feat: Use JS Date for tedge flows timestamps

Author: didier-wenzek

crates/core/tedge/src/cli/flows/cli.rs

@@ -7,6 +7,7 @@ use anyhow::anyhow;
 use anyhow::Context;
 use anyhow::Error;
 use camino::Utf8PathBuf;
+use std::time::SystemTime;
 use tedge_config::TEdgeConfig;
 use tedge_flows::BaseFlowRegistry;
 use tedge_flows::Message;
@@ -47,6 +48,11 @@ pub enum TEdgeFlowsCli {
         #[clap(long = "final-on-interval")]
         final_on_interval: bool,
 
+        /// Processing time to be used for the test
+        #[clap(long = "processing-time")]
+        #[arg(value_parser = humantime::parse_rfc3339_weak)]
+        processing_time: Option<SystemTime>,
+
         /// The input payloads are base64 encoded and have to be decoded first
         #[clap(long = "base64-input")]
         base64_input: bool,
@@ -80,6 +86,7 @@ impl BuildCommand for TEdgeFlowsCli {
                 flows_dir,
                 flow,
                 final_on_interval,
+                processing_time,
                 base64_input,
                 base64_output,
                 topic,
@@ -97,6 +104,7 @@ impl BuildCommand for TEdgeFlowsCli {
                     flow,
                     message,
                     final_on_interval,
+                    processing_time,
                     base64_input,
                     base64_output,
                 }

crates/core/tedge/src/cli/flows/test.rs

@@ -5,9 +5,9 @@ use anyhow::Error;
 use base64::prelude::BASE64_STANDARD;
 use base64::prelude::*;
 use camino::Utf8PathBuf;
+use std::time::SystemTime;
 use tedge_config::TEdgeConfig;
 use tedge_flows::BaseFlowRegistry;
-use tedge_flows::DateTime;
 use tedge_flows::FlowResult;
 use tedge_flows::Message;
 use tedge_flows::MessageProcessor;
@@ -22,6 +22,7 @@ pub struct TestCommand {
     pub flow: Option<Utf8PathBuf>,
     pub message: Option<Message>,
     pub final_on_interval: bool,
+    pub processing_time: Option<SystemTime>,
     pub base64_input: bool,
     pub base64_output: bool,
 }
@@ -41,18 +42,18 @@ impl Command for TestCommand {
             Some(flow) => TEdgeFlowsCli::load_file(&self.flows_dir, flow).await?,
         };
         if let Some(message) = &self.message {
-            let timestamp = DateTime::now();
+            let timestamp = self.processing_time.unwrap_or_else(SystemTime::now);
             self.process(&mut processor, message.clone(), timestamp)
                 .await;
         } else {
             let mut stdin = BufReader::new(tokio::io::stdin());
             while let Some(message) = next_message(&mut stdin).await {
-                let timestamp = DateTime::now();
+                let timestamp = self.processing_time.unwrap_or_else(SystemTime::now);
                 self.process(&mut processor, message, timestamp).await;
             }
         }
         if self.final_on_interval {
-            let timestamp = DateTime::now();
+            let timestamp = SystemTime::now();
             let now = processor
                 .last_interval_deadline()
                 .unwrap_or_else(Instant::now);
@@ -67,7 +68,7 @@ impl TestCommand {
         &self,
         processor: &mut MessageProcessor<BaseFlowRegistry>,
         mut message: Message,
-        timestamp: DateTime,
+        timestamp: SystemTime,
     ) {
         if self.base64_input {
             match BASE64_STANDARD.decode(message.payload) {
@@ -89,7 +90,7 @@ impl TestCommand {
     async fn tick(
         &self,
         processor: &mut MessageProcessor<BaseFlowRegistry>,
-        timestamp: DateTime,
+        timestamp: SystemTime,
         now: Instant,
     ) {
         processor

crates/extensions/tedge_flows/src/actor.rs

@@ -1,5 +1,4 @@
 use crate::connected_flow::ConnectedFlowRegistry;
-use crate::flow::DateTime;
 use crate::flow::FlowError;
 use crate::flow::FlowOutput;
 use crate::flow::FlowResult;
@@ -16,6 +15,7 @@ use serde_json::json;
 use std::cmp::min;
 use std::collections::HashSet;
 use std::time::Duration;
+use std::time::SystemTime;
 use tedge_actors::Actor;
 use tedge_actors::DynSender;
 use tedge_actors::MessageReceiver;
@@ -207,7 +207,7 @@ impl FlowsMapper {
 
     async fn on_source_poll(&mut self) -> Result<(), RuntimeError> {
         let now = Instant::now();
-        let timestamp = DateTime::now();
+        let timestamp = SystemTime::now();
 
         let mut in_messages = vec![];
         for flow in self.processor.registry.flows_mut() {
@@ -241,7 +241,7 @@ impl FlowsMapper {
         source: SourceTag,
         message: Message,
     ) -> Result<(), RuntimeError> {
-        let timestamp = DateTime::now();
+        let timestamp = SystemTime::now();
         for messages in self
             .processor
             .on_message(timestamp, &source, &message)
@@ -255,7 +255,7 @@ impl FlowsMapper {
 
     async fn on_interval(&mut self) -> Result<(), RuntimeError> {
         let now = Instant::now();
-        let timestamp = DateTime::now();
+        let timestamp = SystemTime::now();
         if self.next_dump <= now {
             self.processor.dump_memory_stats().await;
             self.processor.dump_processing_stats().await;
@@ -295,7 +295,7 @@ impl FlowsMapper {
     ) -> Result<(), RuntimeError> {
         if let Some(flow) = self.processor.registry.flow(&flow_name) {
             let topic = flow.input_topic().to_string();
-            let timestamp = DateTime::now();
+            let timestamp = SystemTime::now();
             let message = Message::new(topic, line);
             if let Some(result) = self
                 .processor

crates/extensions/tedge_flows/src/connected_flow.rs

@@ -1,5 +1,4 @@
 use crate::config::ConfigError;
-use crate::flow::DateTime;
 use crate::flow::Flow;
 use crate::flow::FlowError;
 use crate::flow::FlowInput;
@@ -14,6 +13,7 @@ use crate::input_source::StreamingSource;
 use crate::registry::FlowRegistry;
 use crate::registry::FlowStore;
 use camino::Utf8Path;
+use std::time::SystemTime;
 use tedge_watch_ext::WatchRequest;
 use tokio::time::Instant;
 
@@ -65,14 +65,14 @@ impl ConnectedFlow {
         self.polling_source.as_ref().and_then(|p| p.next_deadline())
     }
 
-    pub async fn on_source_poll(&mut self, timestamp: DateTime, now: Instant) -> FlowResult {
+    pub async fn on_source_poll(&mut self, timestamp: SystemTime, now: Instant) -> FlowResult {
         let result = self.on_source_poll_steps(timestamp, now).await;
         self.flow.publish(result)
     }
 
     async fn on_source_poll_steps(
         &mut self,
-        timestamp: DateTime,
+        timestamp: SystemTime,
         now: Instant,
     ) -> Result<Vec<Message>, FlowError> {
         let Some(source) = &mut self.polling_source.as_mut() else {

crates/extensions/tedge_flows/src/flow.rs

@@ -10,11 +10,11 @@ use serde_json::Value;
 use std::fmt::Display;
 use std::fmt::Formatter;
 use std::time::Duration;
+use std::time::SystemTime;
 use tedge_mqtt_ext::MqttMessage;
 use tedge_mqtt_ext::Topic;
 use tedge_mqtt_ext::TopicFilter;
 use tedge_watch_ext::WatchError;
-use time::OffsetDateTime;
 use tokio::time::Instant;
 use tracing::error;
 use tracing::warn;
@@ -112,17 +112,11 @@ impl FlowResult {
     }
 }
 
-#[derive(Copy, Clone, Debug, serde::Deserialize, serde::Serialize, Eq, PartialEq)]
-pub struct DateTime {
-    pub seconds: u64,
-    pub nanoseconds: u32,
-}
-
 #[derive(Clone, serde::Deserialize, serde::Serialize, Eq, PartialEq)]
 pub struct Message {
     pub topic: String,
     pub payload: Vec<u8>,
-    pub timestamp: Option<DateTime>,
+    pub timestamp: Option<SystemTime>,
 }
 
 #[derive(thiserror::Error, Debug)]
@@ -198,7 +192,7 @@ impl Flow {
         &mut self,
         js_runtime: &JsRuntime,
         stats: &mut Counter,
-        timestamp: DateTime,
+        timestamp: SystemTime,
         message: &Message,
     ) -> FlowResult {
         let stated_at = stats.flow_on_message_start(self.name());
@@ -218,7 +212,7 @@ impl Flow {
         &mut self,
         js_runtime: &JsRuntime,
         stats: &mut Counter,
-        timestamp: DateTime,
+        timestamp: SystemTime,
         message: &Message,
     ) -> Result<Vec<Message>, FlowError> {
         let mut messages = vec![message.clone()];
@@ -246,7 +240,7 @@ impl Flow {
         &mut self,
         js_runtime: &JsRuntime,
         stats: &mut Counter,
-        timestamp: DateTime,
+        timestamp: SystemTime,
         now: Instant,
     ) -> FlowResult {
         let stated_at = stats.flow_on_interval_start(self.name());
@@ -266,7 +260,7 @@ impl Flow {
         &mut self,
         js_runtime: &JsRuntime,
         stats: &mut Counter,
-        timestamp: DateTime,
+        timestamp: SystemTime,
         now: Instant,
     ) -> Result<Vec<Message>, FlowError> {
         let mut messages = vec![];
@@ -399,36 +393,6 @@ impl FlowStep {
     }
 }
 
-impl DateTime {
-    pub fn now() -> Self {
-        DateTime::try_from(OffsetDateTime::now_utc()).unwrap()
-    }
-
-    pub fn tick_now(&self, tick_every: std::time::Duration) -> bool {
-        let tick_every_secs = tick_every.as_secs();
-        tick_every_secs != 0 && (self.seconds % tick_every_secs == 0)
-    }
-
-    pub fn json(&self) -> Value {
-        json!({"seconds": self.seconds, "nanoseconds": self.nanoseconds})
-    }
-}
-
-impl TryFrom<OffsetDateTime> for DateTime {
-    type Error = FlowError;
-
-    fn try_from(value: OffsetDateTime) -> Result<Self, Self::Error> {
-        let seconds = u64::try_from(value.unix_timestamp()).map_err(|err| {
-            FlowError::UnsupportedMessage(format!("failed to convert timestamp: {}", err))
-        })?;
-
-        Ok(DateTime {
-            seconds,
-            nanoseconds: value.nanosecond(),
-        })
-    }
-}
-
 impl Message {
     pub fn new(topic: impl ToString, payload: impl Into<Vec<u8>>) -> Self {
         Message {
@@ -441,7 +405,7 @@ impl Message {
     pub fn with_timestamp(
         topic: impl ToString,
         payload: impl Into<Vec<u8>>,
-        timestamp: DateTime,
+        timestamp: SystemTime,
     ) -> Self {
         Message {
             topic: topic.to_string(),
@@ -450,15 +414,9 @@ impl Message {
         }
     }
 
-    #[cfg(test)]
-    pub fn sent_now(mut self) -> Self {
-        self.timestamp = Some(DateTime::now());
-        self
-    }
-
     pub fn json(&self) -> Value {
         if let Some(timestamp) = &self.timestamp {
-            json!({"topic": self.topic, "payload": self.payload, "timestamp": timestamp.json()})
+            json!({"topic": self.topic, "payload": self.payload, "timestamp": epoch_ms(timestamp)})
         } else {
             json!({"topic": self.topic, "payload": self.payload, "timestamp": null})
         }
@@ -508,6 +466,13 @@ impl TryFrom<Message> for MqttMessage {
     }
 }
 
+pub(crate) fn epoch_ms(time: &SystemTime) -> u128 {
+    let duration = time
+        .duration_since(SystemTime::UNIX_EPOCH)
+        .expect("SystemTime after UNIX EPOCH");
+    duration.as_millis()
+}
+
 pub fn error_from_js(err: LoadError) -> FlowError {
     FlowError::IncorrectSetting(format!("{err:#}"))
 }

crates/extensions/tedge_flows/src/input_source.rs

@@ -1,8 +1,8 @@
-use crate::flow::DateTime;
 use crate::flow::Message;
 use async_trait::async_trait;
 use camino::Utf8PathBuf;
 use std::time::Duration;
+use std::time::SystemTime;
 use tedge_watch_ext::WatchRequest;
 use tokio::time::Instant;
 
@@ -16,7 +16,7 @@ pub trait StreamingSource: Send + Sync {
 #[async_trait]
 pub trait PollingSource: Send + Sync {
     /// Poll the source for any available messages at the given timestamp
-    async fn poll(&mut self, timestamp: DateTime) -> Result<Vec<Message>, PollingSourceError>;
+    async fn poll(&mut self, timestamp: SystemTime) -> Result<Vec<Message>, PollingSourceError>;
 
     /// Get the next deadline when this source should be polled
     /// Returns None if the source doesn't have scheduled polling
@@ -53,7 +53,7 @@ impl CommandPollingSource {
 
 #[async_trait]
 impl PollingSource for CommandPollingSource {
-    async fn poll(&mut self, timestamp: DateTime) -> Result<Vec<Message>, PollingSourceError> {
+    async fn poll(&mut self, timestamp: SystemTime) -> Result<Vec<Message>, PollingSourceError> {
         let output = tedge_watch_ext::command_output(&self.command)
             .await
             .map_err(|err| PollingSourceError::CannotPoll {
@@ -119,7 +119,7 @@ impl FilePollingSource {
 
 #[async_trait]
 impl PollingSource for FilePollingSource {
-    async fn poll(&mut self, timestamp: DateTime) -> Result<Vec<Message>, PollingSourceError> {
+    async fn poll(&mut self, timestamp: SystemTime) -> Result<Vec<Message>, PollingSourceError> {
         let output = tokio::fs::read_to_string(&self.path).await.map_err(|err| {
             PollingSourceError::CannotPoll {
                 resource: self.path.clone().to_string(),