feat(logging): use separate file per spawned node

Author: RolandSherwin

Cargo.lock

@@ -21,7 +21,7 @@ serde = { version = "1.0.133", features = ["derive", "rc"] }
 serde_json = { version = "1.0" }
 sysinfo = { version = "0.30.8", default-features = false, optional = true }
 thiserror = "1.0.23"
-tokio = { version = "1.43.1", optional = true }
+tokio = { version = "1.43.1", features = ["rt", "rt-multi-thread", "macros", "time"], optional = true }
 tracing = { version = "~0.1.26" }
 tracing-appender = "~0.2.0"
 tracing-core = "0.1.30"
@@ -31,6 +31,8 @@ tracing-subscriber = { version = "0.3.16", features = ["json"] }
 [dev-dependencies]
 color-eyre = "0.6.3"
 tracing-test = "0.2.4"
+tokio = { version = "1.43.1", features = ["rt", "macros", "rt-multi-thread", "time"] }
+tempfile = "3.8"
 
 [features]
 otlp = [

ant-logging/Cargo.toml

@@ -13,6 +13,7 @@ use crate::{
 };
 use std::collections::BTreeMap;
 use tracing_appender::non_blocking::WorkerGuard;
+use tracing_core::Metadata;
 use tracing_core::{Event, Level, Subscriber};
 use tracing_subscriber::{
     filter::Targets,
@@ -22,15 +23,15 @@ use tracing_subscriber::{
         time::{FormatTime, SystemTime},
         FmtContext, FormatEvent, FormatFields,
     },
-    layer::Filter,
+    layer::{Context, Filter},
     registry::LookupSpan,
     reload::{self, Handle},
     Layer, Registry,
 };
 
-const MAX_LOG_SIZE: usize = 20 * 1024 * 1024;
-const MAX_UNCOMPRESSED_LOG_FILES: usize = 10;
-const MAX_LOG_FILES: usize = 1000;
+pub(crate) const MAX_LOG_SIZE: usize = 20 * 1024 * 1024;
+pub(crate) const MAX_UNCOMPRESSED_LOG_FILES: usize = 10;
+pub(crate) const MAX_LOG_FILES: usize = 1000;
 // Everything is logged by default
 const ALL_ANT_LOGS: &str = "all";
 // Trace at nodes, clients, debug at networking layer
@@ -239,7 +240,7 @@ impl TracingLayers {
 /// `export ANT_LOG = libp2p=DEBUG, tokio=INFO, all, sn_client=ERROR`
 /// Custom keywords will take less precedence if the same target has been manually specified in the CSV.
 /// `sn_client=ERROR` in the above example will be used instead of the TRACE level set by "all" keyword.
-fn get_logging_targets(logging_env_value: &str) -> Result<Vec<(String, Level)>> {
+pub(crate) fn get_logging_targets(logging_env_value: &str) -> Result<Vec<(String, Level)>> {
     let mut targets = BTreeMap::new();
     let mut contains_keyword_all_sn_logs = false;
     let mut contains_keyword_verbose_sn_logs = false;
@@ -325,3 +326,85 @@ fn get_log_level_from_str(log_level: &str) -> Result<Level> {
         ))),
     }
 }
+
+use std::collections::HashMap;
+use std::sync::{Arc, Mutex};
+use tracing_appender::non_blocking::NonBlocking;
+
+/// Layer that routes events to different file appenders based on span context
+pub struct NodeRoutingLayer {
+    node_writers: Arc<Mutex<HashMap<String, NonBlocking>>>,
+    targets_filter: Targets,
+}
+
+impl NodeRoutingLayer {
+    pub fn new(targets: Vec<(String, Level)>) -> Self {
+        Self {
+            node_writers: Arc::new(Mutex::new(HashMap::new())),
+            targets_filter: Targets::new().with_targets(targets),
+        }
+    }
+
+    pub fn add_node_writer(&mut self, node_name: String, writer: NonBlocking) {
+        let mut writers = self
+            .node_writers
+            .lock()
+            .expect("Failed to acquire node writers lock");
+        writers.insert(node_name, writer);
+    }
+}
+
+impl<S> Layer<S> for NodeRoutingLayer
+where
+    S: Subscriber + for<'lookup> LookupSpan<'lookup>,
+{
+    fn enabled(&self, meta: &Metadata<'_>, ctx: Context<'_, S>) -> bool {
+        use tracing_subscriber::layer::Filter;
+        Filter::enabled(&self.targets_filter, meta, &ctx)
+    }
+
+    fn on_event(&self, event: &tracing::Event<'_>, ctx: Context<'_, S>) {
+        // Find which node this event belongs to based on span hierarchy
+        let mut target_node = None;
+
+        if let Some(span_ref) = ctx.lookup_current() {
+            let mut current = Some(span_ref);
+            while let Some(span) = current {
+                let span_name = span.name();
+
+                // Check for standard node spans: node_1, node_2, etc.
+                if span_name.starts_with("node_") {
+                    target_node = Some(span_name.to_string());
+                    break;
+                }
+
+                // Check for node_other spans (for nodes > 20)
+                if span_name == "node_other" {
+                    // For node_other, we'll route to a default "node_other" directory
+                    target_node = Some("node_other".to_string());
+                    break;
+                }
+
+                current = span.parent();
+            }
+        }
+
+        // Route to the appropriate writer
+        if let Some(node_name) = target_node {
+            let writers = self
+                .node_writers
+                .lock()
+                .expect("Failed to acquire node writers lock");
+            if let Some(writer) = writers.get(&node_name) {
+                // Create a temporary fmt layer to format and write the event
+                let temp_layer = tracing_fmt::layer()
+                    .with_ansi(false)
+                    .with_writer(writer.clone())
+                    .event_format(LogFormatter);
+
+                // Forward the event to the temporary layer for proper formatting
+                temp_layer.on_event(event, ctx);
+            }
+        }
+    }
+}

ant-logging/src/layers.rs

@@ -188,6 +188,123 @@ impl LogBuilder {
         Ok((reload_handle, layers.log_appender_guard))
     }
 
+    /// Initialize multi-node logging with per-node log files.
+    /// Each node gets its own log directory and rotation.
+    ///
+    /// # Arguments
+    /// * `node_count` - Number of nodes to create logging for
+    ///
+    /// # Returns
+    /// * `ReloadHandle` - Handle to modify log levels
+    /// * `Vec<WorkerGuard>` - Guards for all node appenders
+    pub fn initialize_with_multi_node_logging(
+        self,
+        node_count: usize,
+    ) -> Result<(ReloadHandle, Vec<WorkerGuard>)> {
+        use crate::appender;
+        use crate::layers::{NodeRoutingLayer, TracingLayers};
+
+        if node_count == 1 {
+            // Fall back to existing single-node implementation
+            let (handle, guard) = self.initialize()?;
+            return Ok((
+                handle,
+                vec![guard.unwrap_or_else(|| {
+                    // Create a dummy guard if none exists
+                    let (_, guard) = tracing_appender::non_blocking(std::io::sink());
+                    guard
+                })],
+            ));
+        }
+
+        // Multi-node logging requires file output
+        let base_log_dir = match &self.output_dest {
+            LogOutputDest::Path(path) => path.clone(),
+            _ => {
+                return Err(Error::LoggingConfiguration(
+                    "Multi-node logging requires file output".to_string(),
+                ))
+            }
+        };
+
+        // Get logging targets
+        let targets = match std::env::var("ANT_LOG") {
+            Ok(sn_log_val) => {
+                if self.print_updates_to_stdout {
+                    println!("Using ANT_LOG={sn_log_val}");
+                }
+                crate::layers::get_logging_targets(&sn_log_val)?
+            }
+            Err(_) => self.default_logging_targets.clone(),
+        };
+
+        // Create NodeRoutingLayer and set up per-node appenders
+        let mut routing_layer = NodeRoutingLayer::new(targets);
+        let mut guards = Vec::new();
+
+        for i in 1..=node_count {
+            let node_name = format!("node_{i}");
+
+            let node_log_dir = base_log_dir.join(&node_name);
+            std::fs::create_dir_all(&node_log_dir)?;
+
+            if self.print_updates_to_stdout {
+                println!("Logging for {node_name} to directory: {node_log_dir:?}");
+            }
+
+            let (appender, guard) = appender::file_rotater(
+                &node_log_dir,
+                crate::layers::MAX_LOG_SIZE,
+                self.max_log_files
+                    .unwrap_or(crate::layers::MAX_UNCOMPRESSED_LOG_FILES),
+                self.max_archived_log_files
+                    .map(|max_archived| {
+                        max_archived
+                            + self
+                                .max_log_files
+                                .unwrap_or(crate::layers::MAX_UNCOMPRESSED_LOG_FILES)
+                    })
+                    .unwrap_or(crate::layers::MAX_LOG_FILES),
+            );
+
+            routing_layer.add_node_writer(node_name, appender);
+            guards.push(guard);
+        }
+
+        let mut layers = TracingLayers::default();
+        layers.layers.push(Box::new(routing_layer));
+
+        #[cfg(feature = "otlp")]
+        {
+            match std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT") {
+                Ok(_) => layers.otlp_layer(self.default_logging_targets)?,
+                Err(_) => println!(
+                    "The OTLP feature is enabled but the OTEL_EXPORTER_OTLP_ENDPOINT variable is not \
+                    set, so traces will not be submitted."
+                ),
+            }
+        }
+
+        if tracing_subscriber::registry()
+            .with(layers.layers)
+            .try_init()
+            .is_err()
+        {
+            return Err(Error::LoggingConfiguration(
+                "Global subscriber already initialized".to_string(),
+            ));
+        }
+
+        // Create reload handle for log level changes
+        let targets_filter: Box<
+            dyn tracing_subscriber::layer::Filter<tracing_subscriber::Registry> + Send + Sync,
+        > = Box::new(tracing_subscriber::filter::Targets::new());
+        let (_, reload_handle) = tracing_subscriber::reload::Layer::new(targets_filter);
+        let reload_handle = ReloadHandle(reload_handle);
+
+        Ok((reload_handle, guards))
+    }
+
     /// Logs to the data_dir. Should be called from a single threaded tokio/non-tokio context.
     /// Provide the test file name to capture tracings from the test.
     ///

ant-logging/src/lib.rs

@@ -0,0 +1,126 @@
+// Integration test for multi-node logging functionality
+
+use ant_logging::{LogBuilder, LogOutputDest};
+use std::path::PathBuf;
+use std::time::Duration;
+use tempfile::TempDir;
+use tracing::{info, Instrument};
+
+#[tokio::test]
+async fn test_multi_node_logging_e2e() {
+    let temp_dir = TempDir::new().expect("Failed to create temp directory");
+    let log_dir = temp_dir.path().to_path_buf();
+
+    // Test multi-node logging with 2 nodes
+    let mut log_builder = LogBuilder::new(vec![(
+        "test_multi_node_e2e".to_string(),
+        tracing::Level::INFO,
+    )]);
+    log_builder.output_dest(LogOutputDest::Path(log_dir.clone()));
+
+    let (_reload_handle, guards) = log_builder
+        .initialize_with_multi_node_logging(2)
+        .expect("Failed to initialize multi-node logging");
+
+    // Log messages from different nodes
+    let node_1_span = tracing::info_span!("node_1");
+    let task1 = async {
+        info!("Message from node 1");
+        info!("Another message from node 1");
+
+        // Test nested spans
+        let inner_span = tracing::info_span!("inner_task");
+        let inner_task = async {
+            info!("Inner message from node 1");
+        }
+        .instrument(inner_span);
+        inner_task.await;
+    }
+    .instrument(node_1_span);
+
+    let node_2_span = tracing::info_span!("node_2");
+    let task2 = async {
+        info!("Message from node 2");
+    }
+    .instrument(node_2_span);
+
+    // Run tasks concurrently
+    tokio::join!(task1, task2);
+
+    // Allow time for logs to be written and flushed
+    tokio::time::sleep(Duration::from_millis(200)).await;
+    drop(guards);
+    tokio::time::sleep(Duration::from_millis(200)).await;
+
+    // Verify node directories were created
+    let node_1_dir = log_dir.join("node_1");
+    let node_2_dir = log_dir.join("node_2");
+
+    assert!(node_1_dir.exists(), "Node 1 directory should exist");
+    assert!(node_2_dir.exists(), "Node 2 directory should exist");
+
+    // Verify each node has its own log file with correct content
+    let node_1_content = read_log_content(&node_1_dir).expect("Failed to read node 1 logs");
+    let node_2_content = read_log_content(&node_2_dir).expect("Failed to read node 2 logs");
+
+    // Check node 1 logs contain all its messages
+    assert!(
+        node_1_content.contains("Message from node 1"),
+        "Node 1 logs should contain its messages"
+    );
+    assert!(
+        node_1_content.contains("Another message from node 1"),
+        "Node 1 logs should contain all its messages"
+    );
+    assert!(
+        node_1_content.contains("Inner message from node 1"),
+        "Node 1 logs should contain nested span messages"
+    );
+    assert!(
+        !node_1_content.contains("Message from node 2"),
+        "Node 1 logs should not contain node 2 messages"
+    );
+
+    // Check node 2 logs contain only its messages
+    assert!(
+        node_2_content.contains("Message from node 2"),
+        "Node 2 logs should contain its messages"
+    );
+    assert!(
+        !node_2_content.contains("Message from node 1"),
+        "Node 2 logs should not contain node 1 messages"
+    );
+
+    // Verify proper log formatting
+    assert!(
+        node_1_content.contains("test_multi_node_e2e"),
+        "Should contain target name"
+    );
+    assert!(
+        node_1_content.contains("node_1"),
+        "Should contain span information"
+    );
+    assert!(
+        node_2_content.contains("node_2"),
+        "Should contain span information"
+    );
+}
+
+/// Helper function to read log content from a node directory
+fn read_log_content(node_dir: &PathBuf) -> Result<String, Box<dyn std::error::Error>> {
+    let mut content = String::new();
+
+    for entry in std::fs::read_dir(node_dir)? {
+        let entry = entry?;
+        if entry.path().extension().is_some_and(|ext| ext == "log") {
+            let file_content = std::fs::read_to_string(entry.path())?;
+            content.push_str(&file_content);
+        }
+    }
+
+    if content.is_empty() {
+        return Err("No log content found".into());
+    }
+
+    Ok(content)
+}