new: added support for custom log message delimiters (#138)

pamburus · Feb 25, 2024 · 6abffc4 · 6abffc4
1 parent 0a2bdd8
commit 6abffc4
Show file tree

Hide file tree

Showing 8 changed files with 619 additions and 76 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -4,7 +4,7 @@ categories = ["command-line-utilities"]
 description = "Utility for viewing json-formatted log files."
 keywords = ["cli", "human", "log"]
 name = "hl"
-version = "0.25.3-alpha.4"
+version = "0.25.3-alpha.5"
 edition = "2021"
 build = "build.rs"
 

diff --git a/README.md b/README.md
@@ -471,6 +471,7 @@ Options:
       --tail <TAIL>                                      Number of last messages to preload from each file in --follow mode [default: 10]
       --sync-interval-ms <SYNC_INTERVAL_MS>              Synchronization interval for live streaming mode enabled by --follow option [default: 100]
   -o, --output <OUTPUT>                                  Output file
+      --delimiter <DELIMITER>                            Log message delimiter, [NUL, CR, LF, CRLF] or any custom string
       --dump-index                                       Dump index metadata and exit
       --help                                             Print help
   -V, --version                                          Print version

diff --git a/src/app.rs b/src/app.rs
@@ -36,7 +36,7 @@ use crate::index::{Indexer, Timestamp};
 use crate::input::{BlockLine, InputHolder, InputReference, Input};
 use crate::model::{Filter, Parser, ParserSettings, RawRecord, Record, RecordFilter, RecordWithSourceConstructor};
 use crate::query::Query;
-use crate::scanning::{BufFactory, Scanner, Segment, SegmentBuf, SegmentBufFactory};
+use crate::scanning::{BufFactory, Delimit, Delimiter, Scanner, SearchExt, Segment, SegmentBuf, SegmentBufFactory};
 use crate::serdex::StreamDeserializerWithOffsets;
 use crate::settings::{Fields, Formatting};
 use crate::theme::{Element, StylingPush, Theme};
@@ -68,6 +68,7 @@ pub struct Options {
     pub dump_index: bool,
     pub app_dirs: Option<AppDirs>,
     pub tail: u64,
+    pub delimiter: Delimiter,
 }
 
 impl Options {
@@ -135,7 +136,7 @@ impl App {
             // spawn reader thread
             let reader = scope.spawn(closure!(clone sfi, |_| -> Result<()> {
                 let mut tx = StripedSender::new(txi);
-                let scanner = Scanner::new(sfi, b'\n');
+                let scanner = Scanner::new(sfi, &self.options.delimiter);
                 for (i, mut input) in inputs.into_iter().enumerate() {
                     for item in scanner.items(&mut input.stream).with_max_segment_size(self.options.max_message_size.into()) {
                         if tx.send((i, item?)).is_none() {
@@ -204,6 +205,7 @@ impl App {
             NonZeroU32::try_from(self.options.max_message_size)?.try_into()?,
             cache_dir,
             &self.options.fields.settings.predefined,
+            self.options.delimiter.clone(),
         );
 
         let input_badges = self.input_badges(inputs.iter().map(|x| &x.reference));
@@ -403,7 +405,7 @@ impl App {
             let mut readers = Vec::with_capacity(m);
             for (i, input_ref) in inputs.into_iter().enumerate() {
                 let reader = scope.spawn(closure!(clone sfi, clone txi, |_| -> Result<()> {
-                    let scanner = Scanner::new(sfi.clone(), b'\n');
+                    let scanner = Scanner::new(sfi.clone(), &self.options.delimiter);
                     let mut meta = None;
                     if let InputReference::File(filename) = &input_ref { 
                         meta = Some(fs::metadata(filename)?);
@@ -703,6 +705,7 @@ impl App {
         let options = SegmentProcessorOptions{
             allow_prefix: self.options.allow_prefix, 
             allow_unparsed_data: self.options.filter.is_empty() && self.options.query.is_none(),
+            delimiter: self.options.delimiter.clone(),
         };
 
         SegmentProcessor::new(parser, self.formatter(), self.options.filter_and_query(), options)
@@ -721,6 +724,7 @@ pub trait SegmentProcess {
 pub struct SegmentProcessorOptions {
     pub allow_prefix: bool,
     pub allow_unparsed_data: bool,
+    pub delimiter: Delimiter,
 }
 
 // ---
@@ -753,7 +757,7 @@ impl<'a, Formatter: RecordWithSourceFormatter, Filter: RecordFilter> SegmentProc
     where
         O: RecordObserver,
     {
-        for data in rtrim(data, b'\n').split(|c| *c == b'\n') {
+        for data in self.options.delimiter.clone().into_searcher().split(data) {
             if data.len() == 0 {
                 continue;
             }
@@ -924,14 +928,6 @@ impl<T> StripedSender<T> {
 
 // ---
 
-fn rtrim<'a>(s: &'a [u8], c: u8) -> &'a [u8] {
-    if s.len() > 0 && s[s.len() - 1] == c {
-        &s[..s.len() - 1]
-    } else {
-        s
-    }
-}
-
 fn common_prefix_len<'a, V, I>(items: &'a Vec<I>) -> usize
 where
     V: 'a + Eq + PartialEq + Copy,

diff --git a/src/index.rs b/src/index.rs
@@ -37,7 +37,7 @@ use crate::index_capnp as schema;
 use crate::input::Input;
 use crate::level::Level;
 use crate::model::{Parser, ParserSettings, RawRecord};
-use crate::scanning::{Scanner, Segment, SegmentBuf, SegmentBufFactory};
+use crate::scanning::{Delimiter, Scanner, Segment, SegmentBuf, SegmentBufFactory};
 use crate::settings::PredefinedFields;
 
 // types
@@ -128,6 +128,7 @@ pub struct Indexer {
     max_message_size: u32,
     dir: PathBuf,
     parser: Parser,
+    delimiter: Delimiter,
 }
 
 impl Indexer {
@@ -138,13 +139,15 @@ impl Indexer {
         max_message_size: u32,
         dir: PathBuf,
         fields: &PredefinedFields,
+        delimiter: Delimiter,
     ) -> Self {
         Self {
             concurrency,
             buffer_size,
             max_message_size,
             dir,
             parser: Parser::new(ParserSettings::new(&fields, empty(), false)),
+            delimiter,
         }
     }
 
@@ -252,7 +255,7 @@ impl Indexer {
             // spawn reader thread
             let reader = scope.spawn(closure!(clone sfi, |_| -> Result<()> {
                 let mut sn: usize = 0;
-                let scanner = Scanner::new(sfi, b'\n');
+                let scanner = Scanner::new(sfi, &self.delimiter);
                 for item in scanner.items(input).with_max_segment_size(self.max_message_size.try_into()?) {
                     if let Err(_) = txi[sn % n].send((sn, item?)) {
                         break;

diff --git a/src/lib.rs b/src/lib.rs
@@ -43,6 +43,7 @@ pub use filtering::DefaultNormalizing;
 pub use formatting::RecordFormatter;
 pub use model::{FieldFilterSet, Filter, Level, Parser, ParserSettings, RecordFilter};
 pub use query::Query;
+pub use scanning::Delimiter;
 pub use settings::Settings;
 pub use theme::Theme;
 

diff --git a/src/main.rs b/src/main.rs
@@ -28,6 +28,7 @@ use hl::signal::SignalHandler;
 use hl::theme::{Theme, ThemeOrigin};
 use hl::timeparse::parse_time;
 use hl::timezone::Tz;
+use hl::Delimiter;
 use hl::{IncludeExcludeKeyFilter, KeyMatchOptions};
 
 // ---
@@ -203,6 +204,10 @@ struct Opt {
     #[arg(long, short = 'o', overrides_with = "output")]
     output: Option<String>,
 
+    /// Log message delimiter, [NUL, CR, LF, CRLF] or any custom string.
+    #[arg(long, overrides_with = "delimiter")]
+    delimiter: Option<String>,
+
     /// Dump index metadata and exit.
     #[arg(long)]
     dump_index: bool,
@@ -389,6 +394,25 @@ fn run() -> Result<()> {
         }
     }
 
+    let mut delimiter = Delimiter::default();
+    if let Some(d) = opt.delimiter {
+        delimiter = match d.to_lowercase().as_str() {
+            "nul" => Delimiter::Byte(0),
+            "lf" => Delimiter::Byte(b'\n'),
+            "cr" => Delimiter::Byte(b'\r'),
+            "crlf" => Delimiter::default(),
+            _ => {
+                if d.len() == 1 {
+                    Delimiter::Byte(d.as_bytes()[0])
+                } else if d.len() > 1 {
+                    Delimiter::Str(d)
+                } else {
+                    Delimiter::default()
+                }
+            }
+        };
+    }
+
     // Create app.
     let app = hl::App::new(hl::Options {
         theme: Arc::new(theme),
@@ -421,6 +445,7 @@ fn run() -> Result<()> {
         dump_index: opt.dump_index,
         app_dirs: Some(app_dirs),
         tail: opt.tail,
+        delimiter,
     });
 
     // Configure input.