feat: audit journal of commands and trace of actor messages

Cargo.toml

@@ -182,6 +182,7 @@ tokio-util = { version = "0.7", features = ["codec"] }
 toml = "0.8"
 tower = "0.4"
 tracing = { version = "0.1", features = ["attributes", "log"] }
+tracing-appender = "0.2"
 tracing-subscriber = { version = "0.3", features = ["time", "env-filter"] }
 try-traits = "0.1"
 tungstenite = "0.20"

crates/common/tedge_config/Cargo.toml

@@ -35,6 +35,7 @@ tedge_utils = { workspace = true, features = ["timestamp"] }
 thiserror = { workspace = true }
 toml = { workspace = true }
 tracing = { workspace = true }
+tracing-appender = { workspace = true }
 tracing-subscriber = { workspace = true }
 url = { workspace = true }
 which = { workspace = true }

crates/common/tedge_config/src/system_services/log_config.rs

@@ -5,6 +5,12 @@ use crate::system_services::SystemConfig;
 use crate::system_services::SystemServiceError;
 use std::io::IsTerminal;
 use std::str::FromStr;
+use tracing::metadata::LevelFilter;
+use tracing_appender::rolling::*;
+use tracing_subscriber::filter::filter_fn;
+use tracing_subscriber::layer::SubscriberExt;
+use tracing_subscriber::util::SubscriberInitExt;
+use tracing_subscriber::Layer;
 
 /// Configures and enables logging taking into account flags, env variables and file config.
 ///
@@ -19,7 +25,8 @@ pub fn log_init(
     flags: &LogConfigArgs,
     config_dir: &Utf8Path,
 ) -> Result<(), SystemServiceError> {
-    let subscriber = tracing_subscriber::fmt()
+    // General logging
+    let log_layer = tracing_subscriber::fmt::layer()
         .with_writer(std::io::stderr)
         .with_ansi(std::io::stderr().is_terminal())
         .with_timer(tracing_subscriber::fmt::time::UtcTime::rfc_3339());
@@ -28,22 +35,60 @@ pub fn log_init(
         .log_level
         .or(flags.debug.then_some(tracing::Level::DEBUG));
 
-    if let Some(log_level) = log_level {
-        subscriber.with_max_level(log_level).init();
-        return Ok(());
-    }
-
-    if std::env::var("RUST_LOG").is_ok() {
-        subscriber
-            .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
+    let log_layer = if let Some(log_level) = log_level {
+        log_layer
+            .with_filter(LevelFilter::from_level(log_level))
+            .boxed()
+    } else if std::env::var("RUST_LOG").is_ok() {
+        log_layer
             .with_file(true)
             .with_line_number(true)
-            .init();
-        return Ok(());
-    }
-
-    let log_level = get_log_level(sname, config_dir)?;
-    subscriber.with_max_level(log_level).init();
+            .with_filter(tracing_subscriber::EnvFilter::from_default_env())
+            .boxed()
+    } else {
+        let log_level = get_log_level(sname, config_dir)?;
+        log_layer
+            .with_filter(LevelFilter::from_level(log_level))
+            .boxed()
+    };
+
+    // Audit journal
+    let audit_appender = RollingFileAppender::builder()
+        .rotation(Rotation::DAILY)
+        .filename_prefix("tedge.audit.log")
+        .max_log_files(7);
+    let audit_layer = audit_appender
+        .build("/var/log/tedge")
+        .ok()
+        .map(|audit_appender| {
+            tracing_subscriber::fmt::layer()
+                .with_writer(audit_appender)
+                .with_timer(tracing_subscriber::fmt::time::UtcTime::rfc_3339())
+                .with_filter(LevelFilter::INFO)
+                .with_filter(filter_fn(|metadata| metadata.target() == "Audit"))
+        });
+
+    // Actor traces
+    let trace_appender = RollingFileAppender::builder()
+        .rotation(Rotation::DAILY)
+        .filename_prefix("tedge.actors.log")
+        .max_log_files(2);
+    let trace_layer = trace_appender
+        .build("/var/log/tedge")
+        .ok()
+        .map(|trace_appender| {
+            tracing_subscriber::fmt::layer()
+                .with_writer(trace_appender)
+                .with_timer(tracing_subscriber::fmt::time::UtcTime::rfc_3339())
+                .with_filter(LevelFilter::DEBUG)
+                .with_filter(filter_fn(|metadata| metadata.target() == "Actors"))
+        });
+
+    tracing_subscriber::registry()
+        .with(audit_layer)
+        .with(trace_layer)
+        .with(log_layer)
+        .init();
 
     Ok(())
 }

crates/core/tedge/src/cli/config/commands/add.rs

@@ -25,6 +25,7 @@ impl Command for AddConfigCommand {
                     .map_err(|e| e.into())
             })
             .map_err(anyhow::Error::new)?;
+        tracing::info!(target: "Audit", "tedge config add {} {}", &self.key, &self.value);
         Ok(())
     }
 }

crates/core/tedge/src/cli/config/commands/remove.rs

@@ -21,6 +21,7 @@ impl Command for RemoveConfigCommand {
                     .map_err(|e| e.into())
             })
             .map_err(anyhow::Error::new)?;
+        tracing::info!(target: "Audit", "tedge config remove {} {}", &self.key, &self.value);
         Ok(())
     }
 }

crates/core/tedge/src/cli/config/commands/set.rs

@@ -25,6 +25,7 @@ impl Command for SetConfigCommand {
                     .map_err(|e| e.into())
             })
             .map_err(anyhow::Error::new)?;
+        tracing::info!(target: "Audit", "tedge config set {} {}", &self.key, &self.value);
         Ok(())
     }
 }

crates/core/tedge/src/cli/config/commands/unset.rs

@@ -17,6 +17,7 @@ impl Command for UnsetConfigCommand {
         self.config_location
             .update_toml(&|dto, _reader| Ok(dto.try_unset_key(&self.key)?))
             .map_err(anyhow::Error::new)?;
+        tracing::info!(target: "Audit", "tedge config unset {}", &self.key);
         Ok(())
     }
 }

crates/core/tedge_actors/Cargo.toml

@@ -17,17 +17,16 @@ test-helpers = []
 [dependencies]
 async-trait = { workspace = true }
 futures = { workspace = true }
-log = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true, default_features = false, features = [
     "sync",
     "rt",
     "macros",
     "time",
 ] }
+tracing = { workspace = true }
 
 [dev-dependencies]
-env_logger = { workspace = true } # TODO: remove me
 tokio = { workspace = true, default_features = false, features = [
     "rt-multi-thread",
 ] }

crates/core/tedge_actors/src/message_boxes.rs

@@ -93,7 +93,6 @@ use crate::RuntimeRequest;
 use async_trait::async_trait;
 use futures::channel::mpsc;
 use futures::StreamExt;
-use log::debug;
 use std::fmt::Debug;
 
 #[async_trait]
@@ -160,22 +159,22 @@ impl<Input: Debug> LoggingReceiver<Input> {
 }
 
 #[async_trait]
-impl<Input: Send + Debug> MessageReceiver<Input> for LoggingReceiver<Input> {
+impl<Input: Send + Debug + Sync> 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);
+        log_message_received(&self.name, &message);
         message
     }
 
     async fn recv(&mut self) -> Option<Input> {
         let message = self.receiver.recv().await;
-        debug!(target: &self.name, "recv {:?}", message);
+        log_message_received(&self.name, &message);
         message
     }
 
     async fn recv_signal(&mut self) -> Option<RuntimeRequest> {
         let message = self.receiver.recv_signal().await;
-        debug!(target: &self.name, "recv {:?}", message);
+        log_message_received(&self.name, &message);
         message
     }
 }
@@ -208,8 +207,14 @@ impl<Output: Message> Sender<Output> for LoggingSender<Output> {
     }
 }
 
-pub fn log_message_sent<I: Debug>(target: &str, message: I) {
-    debug!(target: target, "send {message:?}");
+#[inline]
+pub fn log_message_received<I: Debug>(actor: &str, message: &I) {
+    tracing::debug!(target: "Actors", actor, recv = ?message);
+}
+
+#[inline]
+pub fn log_message_sent<I: Debug>(actor: &str, message: &I) {
+    tracing::debug!(target: "Actors", actor, send = ?message);
 }
 
 /// An unbounded receiver
@@ -251,10 +256,10 @@ impl<Input: Debug> UnboundedLoggingReceiver<Input> {
 }
 
 #[async_trait]
-impl<Input: Send + Debug> MessageReceiver<Input> for UnboundedLoggingReceiver<Input> {
+impl<Input: Send + Debug + Sync> MessageReceiver<Input> for UnboundedLoggingReceiver<Input> {
     async fn try_recv(&mut self) -> Result<Option<Input>, RuntimeRequest> {
         let message = self.next_message().await;
-        debug!(target: &self.name, "recv {:?}", message);
+        log_message_received(&self.name, &message);
         message
     }
 
@@ -263,13 +268,13 @@ impl<Input: Send + Debug> MessageReceiver<Input> for UnboundedLoggingReceiver<In
             Ok(Some(message)) => Some(message),
             _ => None,
         };
-        debug!(target: &self.name, "recv {:?}", message);
+        log_message_received(&self.name, &message);
         message
     }
 
     async fn recv_signal(&mut self) -> Option<RuntimeRequest> {
         let message = self.signal_receiver.next().await;
-        debug!(target: &self.name, "recv {:?}", message);
+        log_message_received(&self.name, &message);
         message
     }
 }