wip

Author: Bravo555

Cargo.lock

@@ -18,6 +18,7 @@ rumqttc = { workspace = true }
 serde = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true, features = ["rt", "time", "rt-multi-thread"] }
+tracing = { workspace = true }
 zeroize = { workspace = true }
 
 [dev-dependencies]

crates/common/mqtt_channel/Cargo.toml

@@ -8,8 +8,6 @@ use futures::channel::mpsc;
 use futures::channel::oneshot;
 use futures::SinkExt;
 use futures::StreamExt;
-use log::error;
-use log::info;
 use rumqttc::AsyncClient;
 use rumqttc::Event;
 use rumqttc::EventLoop;
@@ -21,6 +19,10 @@ use std::time::Duration;
 use tokio::sync::OwnedSemaphorePermit;
 use tokio::sync::Semaphore;
 use tokio::time::sleep;
+use tracing::error;
+use tracing::info;
+use tracing::instrument;
+use tracing::trace;
 
 /// A connection to some MQTT server
 pub struct Connection {
@@ -123,6 +125,7 @@ impl Connection {
         let _ = self.pub_done.await;
     }
 
+    #[instrument(level = "trace", skip(message_sender, error_sender))]
     async fn open(
         config: &Config,
         mut message_sender: mpsc::UnboundedSender<MqttMessage>,
@@ -201,6 +204,7 @@ impl Connection {
         Ok((mqtt_client, event_loop))
     }
 
+    #[instrument(skip_all, level = "trace")]
     async fn receiver_loop(
         mqtt_client: AsyncClient,
         config: Config,
@@ -306,17 +310,21 @@ impl Connection {
         let _ = message_sender.close().await;
         let _ = error_sender.close().await;
         let _ = done.send(());
+        trace!("terminating receiver loop");
         Ok(())
     }
 
+    #[instrument(skip_all, level = "trace")]
     async fn sender_loop(
         mqtt_client: AsyncClient,
         mut messages_receiver: mpsc::UnboundedReceiver<MqttMessage>,
         mut error_sender: mpsc::UnboundedSender<MqttError>,
         last_will: Option<MqttMessage>,
         _disconnect_permit: OwnedSemaphorePermit,
     ) {
+        trace!("waiting for message");
         while let Some(message) = messages_receiver.next().await {
+            trace!(msg = ?message, "received message");
             let payload = Vec::from(message.payload_bytes());
             if let Err(err) = mqtt_client
                 .publish(message.topic, message.qos, message.retain, payload)

crates/common/mqtt_channel/src/connection.rs

@@ -25,6 +25,7 @@ tokio = { workspace = true, default_features = false, features = [
     "macros",
     "time",
 ] }
+tracing = { workspace = true }
 
 [dev-dependencies]
 env_logger = { workspace = true } # TODO: remove me

crates/core/tedge_actors/Cargo.toml

@@ -5,6 +5,8 @@ use async_trait::async_trait;
 use futures::channel::mpsc;
 use futures::channel::oneshot;
 use futures::SinkExt;
+use tracing::debug;
+use tracing::instrument;
 
 /// A sender of messages of type `M`
 ///
@@ -75,13 +77,24 @@ impl<M: Message, N: Message + Into<M>> CloneSender<N> for DynSender<M> {
 #[async_trait]
 impl<M: Message, N: Message + Into<M>> Sender<N> for mpsc::Sender<M> {
     async fn send(&mut self, message: N) -> Result<(), ChannelError> {
-        Ok(SinkExt::send(&mut self, message.into()).await?)
+        let message = message.into();
+        if let Err(err) = self.try_send(message) {
+            if err.is_full() {
+                debug!("Sender is full");
+            }
+            let message = err.into_inner();
+            SinkExt::send(&mut self, message).await?;
+            debug!("Blocked send completed");
+        }
+
+        Ok(())
     }
 }
 
 /// An `mpsc::UnboundedSender<M>` is a `DynSender<N>` provided `N` implements `Into<M>`
 #[async_trait]
 impl<M: Message, N: Message + Into<M>> Sender<N> for mpsc::UnboundedSender<M> {
+    #[instrument(name = "mpsc::UnboundedSender::send", skip_all)]
     async fn send(&mut self, message: N) -> Result<(), ChannelError> {
         Ok(SinkExt::send(&mut self, message.into()).await?)
     }

crates/core/tedge_actors/src/channels.rs

@@ -95,6 +95,7 @@ use futures::channel::mpsc;
 use futures::StreamExt;
 use log::debug;
 use std::fmt::Debug;
+use tracing::instrument;
 
 #[async_trait]
 pub trait MessageReceiver<Input> {
@@ -163,13 +164,15 @@ impl<Input: Debug> LoggingReceiver<Input> {
 impl<Input: Send + Debug> MessageReceiver<Input> for LoggingReceiver<Input> {
     async fn try_recv(&mut self) -> Result<Option<Input>, RuntimeRequest> {
         let message = self.receiver.try_recv().await;
-        debug!(target: &self.name, "recv {:?}", message);
+        debug!("recv {:?}", message);
         message
     }
 
+    #[instrument(name = "LoggingReceiver::recv", skip(self), fields(name = self.name))]
     async fn recv(&mut self) -> Option<Input> {
+        debug!("attempting recv");
         let message = self.receiver.recv().await;
-        debug!(target: &self.name, "recv {:?}", message);
+        debug!("recv");
         message
     }
 
@@ -202,6 +205,7 @@ impl<Output> LoggingSender<Output> {
 
 #[async_trait]
 impl<Output: Message> Sender<Output> for LoggingSender<Output> {
+    #[instrument(name = "LoggingSender::send", skip(self, message), fields(name = self.name))]
     async fn send(&mut self, message: Output) -> Result<(), ChannelError> {
         log_message_sent(&self.name, &message);
         self.sender.send(message).await

crates/core/tedge_actors/src/message_boxes.rs

@@ -1,3 +1,6 @@
+use tracing::debug_span;
+use tracing::Instrument;
+
 use crate::Actor;
 use crate::Builder;
 use crate::DynSender;
@@ -39,7 +42,11 @@ impl RunActor {
     }
 
     pub async fn run(self) -> Result<(), RuntimeError> {
-        self.actor.run_boxed().await
+        let name = self.actor.name().to_string();
+        self.actor
+            .run_boxed()
+            .instrument(debug_span!("Actor::run", actor_name = name))
+            .await
     }
 }
 

crates/core/tedge_actors/src/run_actor.rs

@@ -11,14 +11,18 @@ use crate::RuntimeRequestSink;
 use futures::channel::mpsc;
 use futures::prelude::*;
 use futures::stream::FuturesUnordered;
-use log::debug;
-use log::error;
-use log::info;
 use std::collections::HashMap;
 use std::panic;
 use std::time::Duration;
 use tokio::task::JoinError;
 use tokio::task::JoinHandle;
+use tracing::debug;
+use tracing::error;
+use tracing::info;
+use tracing::instrument;
+
+// TODO: set back to 60
+const ACTORS_EXIT_TIMEOUT: Duration = Duration::from_secs(5);
 
 /// Actions sent by actors to the runtime
 #[derive(Debug)]
@@ -64,8 +68,7 @@ impl Runtime {
 
     fn with_events_sender(events_sender: Option<DynSender<RuntimeEvent>>) -> Runtime {
         let (actions_sender, actions_receiver) = mpsc::channel(16);
-        let runtime_actor =
-            RuntimeActor::new(actions_receiver, events_sender, Duration::from_secs(60));
+        let runtime_actor = RuntimeActor::new(actions_receiver, events_sender, ACTORS_EXIT_TIMEOUT);
 
         let runtime_task = tokio::spawn(runtime_actor.run());
         Runtime {
@@ -174,6 +177,7 @@ impl RuntimeActor {
         }
     }
 
+    #[instrument(name = "Runtime", level = "trace", skip_all)]
     async fn run(mut self) -> Result<(), RuntimeError> {
         info!(target: "Runtime", "Started");
         let mut aborting_error = None;

crates/core/tedge_actors/src/runtime.rs

@@ -13,6 +13,7 @@ use futures::channel::oneshot;
 use futures::StreamExt;
 use std::fmt::Debug;
 use std::ops::ControlFlow;
+use tracing::instrument;
 
 /// A message box for a request-response server
 pub type ServerMessageBox<Request, Response> = LoggingReceiver<RequestEnvelope<Request, Response>>;
@@ -146,6 +147,7 @@ impl<Request, Response> Clone for RequestSender<Request, Response> {
 
 #[async_trait]
 impl<Request: Message, Response: Message> Sender<Request> for RequestSender<Request, Response> {
+    #[instrument(name = "RequestSender::send", skip_all)]
     async fn send(&mut self, request: Request) -> Result<(), ChannelError> {
         let reply_to = self.reply_to.sender();
         self.sender

crates/core/tedge_actors/src/servers/message_boxes.rs

@@ -17,7 +17,7 @@ json-writer = { workspace = true }
 log = { workspace = true }
 mqtt_channel = { workspace = true }
 regex = { workspace = true }
-serde = { workspace = true }
+serde = { workspace = true, features = ["derive", "rc"] }
 serde_json = { workspace = true }
 shell-words = { workspace = true }
 tedge_utils = { workspace = true, features = ["timestamp"] }
@@ -30,6 +30,7 @@ time = { workspace = true, features = [
     "serde-well-known",
 ] }
 tokio = { workspace = true, features = ["fs", "process"] }
+tracing = { workspace = true }
 
 [dev-dependencies]
 anyhow = { workspace = true }