feat(logger): implement logs truncation when line size limit exceedes

elfo-logger/Cargo.toml

@@ -15,6 +15,7 @@ workspace = true
 
 [features]
 tracing-log = [ "dep:tracing-log", "log" ]
+docsrs = []
 
 [dependencies]
 elfo-core = { version = "0.2.0-alpha.15", path = "../elfo-core", features = ["unstable"] }
@@ -37,6 +38,7 @@ metrics = "0.17"
 dashmap = "5"
 fxhash = "0.2.1"
 humantime = "2.1.0"
+bytesize = { version = "1.2.0", features = ["serde"] }
 
 [dev-dependencies]
 elfo-core = { version = "0.2.0-alpha.15", path = "../elfo-core", features = ["test-util"] }

elfo-logger/src/actor.rs

@@ -19,14 +19,16 @@ use crate::{
     config::{Config, Sink},
     filtering_layer::FilteringLayer,
     formatters::Formatter,
+    line_buffer::LineBuffer,
+    line_transaction::{Line as _, LineFactory, TryDirectWrite, UseSlowPath},
     theme, PreparedEvent, Shared,
 };
 
 pub(crate) struct Logger {
     ctx: Context<Config>,
     shared: Arc<Shared>,
     filtering_layer: FilteringLayer,
-    buffer: String,
+    buffer: LineBuffer,
 }
 
 /// Reload a log file, usually after rotation.
@@ -52,11 +54,13 @@ impl Logger {
 
     fn new(ctx: Context<Config>, shared: Arc<Shared>, filtering_layer: FilteringLayer) -> Self {
         filtering_layer.configure(&ctx.config().targets);
+        let buffer = LineBuffer::with_buffer_capacity(1024, ctx.config().max_line_size.0 as _);
+
         Self {
             ctx,
             shared,
             filtering_layer,
-            buffer: String::with_capacity(1024),
+            buffer,
         }
     }
 
@@ -75,20 +79,19 @@ impl Logger {
             tokio::select! {
                 event = self.shared.channel.receive() => {
                     let event = ward!(event, break);
-
                     self.buffer.clear();
 
                     if use_colors {
-                        self.format_event::<theme::ColoredTheme>(event);
+                        self.format_event::<theme::ColoredTheme, TryDirectWrite>(event)
                     } else {
-                        self.format_event::<theme::PlainTheme>(event);
+                        self.format_event::<theme::PlainTheme, TryDirectWrite>(event)
                     }
 
                     if let Some(file) = file.as_mut() {
                         // TODO: what about performance here?
-                        file.write_all(self.buffer.as_ref()).await.expect("cannot write to the config file");
+                        file.write_all(self.buffer.buffer.as_bytes()).await.expect("cannot write to the config file");
                     } else {
-                        print!("{}", self.buffer);
+                        print!("{}", self.buffer.buffer);
                     }
 
                     increment_counter!("elfo_written_events_total");
@@ -104,6 +107,7 @@ impl Logger {
                             file = open_file(self.ctx.config()).await;
                             use_colors = can_use_colors(self.ctx.config());
                             self.filtering_layer.configure(&self.ctx.config().targets);
+                            self.buffer.max_line_size = self.ctx.config().max_line_size.0 as _;
                         },
                         Terminate => {
                             // Close the channel and wait for the rest of the events.
@@ -120,61 +124,79 @@ impl Logger {
         }
     }
 
-    pub(super) fn format_event<T: theme::Theme>(&mut self, event: PreparedEvent) {
-        let out = &mut self.buffer;
+    pub(super) fn format_event<T: theme::Theme, F: LineFactory>(&mut self, event: PreparedEvent) {
         let config = self.ctx.config();
-
-        let payload = self
-            .shared
-            .pool
-            .get(event.payload_id)
-            .expect("unknown string");
-        self.shared.pool.clear(event.payload_id);
-
-        // <timestamp> <level> [<trace_id>] <object> - <message>\t<fields>
-
-        T::Timestamp::fmt(out, &event.timestamp);
-        out.push(' ');
-        T::Level::fmt(out, event.metadata.level());
-        out.push_str(" [");
-        T::TraceId::fmt(out, &event.trace_id);
-        out.push_str("] ");
-        T::ActorMeta::fmt(out, &event.object);
-        out.push_str(" - ");
-        T::Payload::fmt(out, &payload);
-
-        // Add ancestors' fields.
-        let mut span_id = event.span_id;
-        while let Some(data) = span_id
-            .as_ref()
-            .and_then(|span_id| self.shared.spans.get(span_id))
-        {
-            span_id.clone_from(&data.parent_id);
+        let try_slow_path = {
+            let mut line = F::create_line(&mut self.buffer);
 
             let payload = self
                 .shared
                 .pool
-                .get(data.payload_id)
+                .get(event.payload_id)
                 .expect("unknown string");
 
-            T::Payload::fmt(out, &payload);
-        }
+            // <timestamp> <level> [<trace_id>] <object> - <message>\t<fields>
+
+            T::Timestamp::fmt(line.meta_mut(), &event.timestamp);
+            line.meta_mut().push(' ');
+            T::Level::fmt(line.meta_mut(), event.metadata.level());
+            line.meta_mut().push_str(" [");
+            T::TraceId::fmt(line.meta_mut(), &event.trace_id);
+            line.meta_mut().push_str("] ");
+            T::ActorMeta::fmt(line.payload_mut(), &event.object);
+            line.payload_mut().push_str(" - ");
+            T::Payload::fmt(line.payload_mut(), &payload);
+
+            // Add ancestors' fields.
+            let mut span_id = event.span_id.clone();
+
+            {
+                let payload_buffer = line.payload_mut();
+                while let Some(data) = span_id
+                    .as_ref()
+                    .and_then(|span_id| self.shared.spans.get(span_id))
+                {
+                    span_id.clone_from(&data.parent_id);
+
+                    let payload = self
+                        .shared
+                        .pool
+                        .get(data.payload_id)
+                        .expect("unknown string");
+
+                    T::Payload::fmt(payload_buffer, &payload);
+                }
+            }
 
-        if config.format.with_location {
-            if let Some(location) = extract_location(event.metadata) {
-                out.push('\t');
-                T::Location::fmt(out, &location);
+            if config.format.with_location {
+                if let Some(location) = extract_location(event.metadata) {
+                    let fields_buffer = line.fields_mut();
+                    fields_buffer.push('\t');
+                    T::Location::fmt(line.fields_mut(), &location);
+                }
             }
-        }
 
-        if config.format.with_module {
-            if let Some(module) = event.metadata.module_path() {
-                out.push('\t');
-                T::Module::fmt(out, module);
+            if config.format.with_module {
+                if let Some(module) = event.metadata.module_path() {
+                    let fields_buffer = line.fields_mut();
+                    fields_buffer.push('\t');
+                    T::Module::fmt(fields_buffer, module);
+                }
             }
-        }
 
-        out.push('\n');
+            let try_slow_path =
+                !F::STOP && (line.total_wrote() > self.ctx.config().max_line_size.0 as _);
+            if try_slow_path {
+                line.discard();
+            } else {
+                self.shared.pool.clear(event.payload_id);
+            }
+            try_slow_path
+        };
+
+        if try_slow_path {
+            self.format_event::<T, UseSlowPath>(event);
+        }
     }
 }
 

elfo-logger/src/config.rs

@@ -4,6 +4,8 @@ use fxhash::FxHashMap;
 use serde::{Deserialize, Deserializer};
 use tracing::metadata::LevelFilter;
 
+use bytesize::ByteSize;
+
 #[derive(Debug, Deserialize)]
 pub(crate) struct Config {
     #[serde(default)]
@@ -12,6 +14,15 @@ pub(crate) struct Config {
     #[serde(default)]
     pub(crate) format: Format,
 
+    /// Size limit for each line wrote in the log. The limit is applied in the
+    /// following order:
+    ///
+    /// 1. Message (logged message)
+    /// 2. Fields (location and module)
+    /// 3. Meta-info (timestamp, log level, trace id)
+    #[serde(default = "default_max_line_size")]
+    pub(crate) max_line_size: ByteSize,
+
     #[serde(default)]
     pub(crate) targets: FxHashMap<String, LoggingTargetConfig>,
 }
@@ -39,6 +50,10 @@ pub(crate) struct Format {
     // TODO: colors
 }
 
+fn default_max_line_size() -> ByteSize {
+    ByteSize(u64::MAX)
+}
+
 // TODO: deduplicate with core
 fn deserialize_level_filter<'de, D>(deserializer: D) -> Result<LevelFilter, D::Error>
 where

elfo-logger/src/formatters.rs

@@ -8,6 +8,26 @@ pub(crate) trait Formatter<T: ?Sized> {
     fn fmt(dest: &mut String, v: &T);
 }
 
+// DoNothing
+
+pub(crate) struct DoNothing;
+
+impl<T> Formatter<T> for DoNothing {
+    fn fmt(_dest: &mut String, _v: &T) {
+        // Apparently does nothing
+    }
+}
+
+// ResetStyle
+
+pub(crate) struct ResetStyle;
+
+impl Formatter<()> for ResetStyle {
+    fn fmt(dest: &mut String, _v: &()) {
+        dest.push_str("\x1b[0m");
+    }
+}
+
 // Rfc3339Weak
 
 pub(crate) struct Rfc3339Weak;

elfo-logger/src/lib.rs

@@ -18,6 +18,9 @@ use elfo_core::{tracing::TraceId, ActorMeta, Blueprint};
 
 use crate::{actor::Logger, filtering_layer::FilteringLayer, printing_layer::PrintingLayer};
 
+#[cfg(feature = "docsrs")]
+pub use crate::config::Config;
+
 pub use crate::actor::ReopenLogFile;
 
 mod actor;
@@ -28,6 +31,9 @@ mod printing_layer;
 mod stats;
 mod theme;
 
+mod line_buffer;
+mod line_transaction;
+
 const CHANNEL_CAPACITY: usize = 128 * 1024;
 
 type StringId = usize;