Add ljx tool for working with logjet files

Author: Bo Maryniuk

ljx/Cargo.toml

@@ -0,0 +1,11 @@
+[package]
+name = "ljx"
+version = "0.1.0"
+edition = "2024"
+license = "Apache-2.0"
+
+[dependencies]
+clap = { version = "4.5", features = ["derive", "wrap_help"] }
+colored = "3"
+logjet = { path = ".." }
+regex = "1.12"

ljx/src/cli.rs

@@ -0,0 +1,151 @@
+use std::path::PathBuf;
+
+use clap::builder::styling;
+use clap::{Args, CommandFactory, Parser, Subcommand, ValueEnum};
+use colored::Colorize;
+
+use crate::predicate::PredicateArgs;
+
+#[derive(Debug, Parser)]
+#[command(
+    name = "ljx",
+    version,
+    about = "Offline toolbox for .logjet files",
+    long_about = None
+)]
+pub struct Cli {
+    #[command(subcommand)]
+    pub command: Command,
+}
+
+pub fn build_cli() -> clap::Command {
+    let appname = "ljx";
+    let styles = styling::Styles::styled()
+        .header(styling::AnsiColor::Yellow.on_default())
+        .usage(styling::AnsiColor::Yellow.on_default())
+        .literal(styling::AnsiColor::BrightGreen.on_default())
+        .placeholder(styling::AnsiColor::BrightMagenta.on_default());
+
+    Cli::command()
+        .styles(styles)
+        .about(format!(
+            "{} - {}",
+            appname.bright_magenta().bold(),
+            "offline toolbox for .logjet streams"
+        ))
+        .override_usage(format!(
+            "{appname} <COMMAND> [OPTIONS] [ARGS]"
+        ))
+        .after_help(
+            "Examples:\n  ljx count telemetry.logjet -F error -i\n  ljx filter telemetry.logjet -o only-logs.logjet -e 'java\\..*\\.bs'",
+        )
+}
+
+#[derive(Debug, Subcommand)]
+pub enum Command {
+    Split(SplitArgs),
+    Join(JoinArgs),
+    Filter(FilterArgs),
+    Count(CountArgs),
+    Stats(StatsArgs),
+    Cat(CatArgs),
+}
+
+#[derive(Debug, Clone, Args)]
+pub struct CountArgs {
+    #[arg(value_name = "INPUT", help = "Input .logjet file or - for stdin")]
+    pub input: PathBuf,
+
+    #[command(flatten)]
+    pub predicate: PredicateArgs,
+}
+
+#[derive(Debug, Clone, Args)]
+pub struct FilterArgs {
+    #[arg(value_name = "INPUT", help = "Input .logjet file or - for stdin")]
+    pub input: PathBuf,
+
+    #[arg(
+        short,
+        long,
+        value_name = "OUTPUT",
+        help = "Output .logjet file or - for stdout"
+    )]
+    pub output: PathBuf,
+
+    #[arg(long, value_enum, default_value_t = OutputCodec::Lz4)]
+    pub codec: OutputCodec,
+
+    #[arg(
+        long,
+        default_value_t = logjet::DEFAULT_BLOCK_TARGET_SIZE,
+        help = "Target uncompressed bytes per output block"
+    )]
+    pub block_target_size: usize,
+
+    #[command(flatten)]
+    pub predicate: PredicateArgs,
+}
+
+#[derive(Debug, Clone, Args)]
+pub struct SplitArgs {
+    #[arg(value_name = "INPUT")]
+    pub input: PathBuf,
+
+    #[arg(value_name = "OUTPUT_PREFIX")]
+    pub output_prefix: PathBuf,
+
+    #[arg(long)]
+    pub max_bytes: Option<u64>,
+
+    #[arg(long)]
+    pub max_records: Option<u64>,
+
+    #[arg(long, help = "RFC 3339 or unix-ns range support to be added")]
+    pub timestamp_range: Option<String>,
+}
+
+#[derive(Debug, Clone, Args)]
+pub struct JoinArgs {
+    #[arg(value_name = "INPUT", required = true)]
+    pub inputs: Vec<PathBuf>,
+
+    #[arg(short, long, value_name = "OUTPUT")]
+    pub output: PathBuf,
+
+    #[arg(long)]
+    pub validate_sequence_continuity: bool,
+}
+
+#[derive(Debug, Clone, Args)]
+pub struct StatsArgs {
+    #[arg(value_name = "INPUT")]
+    pub input: PathBuf,
+
+    #[arg(long, help = "Compute payload size summaries by record type")]
+    pub field_stats: bool,
+}
+
+#[derive(Debug, Clone, Args)]
+pub struct CatArgs {
+    #[arg(value_name = "INPUT")]
+    pub input: PathBuf,
+
+    #[arg(long, default_value_t = false)]
+    pub hex_payload: bool,
+}
+
+#[derive(Debug, Clone, Copy, ValueEnum)]
+pub enum OutputCodec {
+    None,
+    Lz4,
+}
+
+impl From<OutputCodec> for logjet::Codec {
+    fn from(value: OutputCodec) -> Self {
+        match value {
+            OutputCodec::None => Self::None,
+            OutputCodec::Lz4 => Self::Lz4,
+        }
+    }
+}

ljx/src/commands/cat.rs

@@ -0,0 +1,8 @@
+use crate::cli::CatArgs;
+use crate::error::{Error, Result};
+
+pub fn run(_args: CatArgs) -> Result<()> {
+    Err(Error::Unimplemented(
+        "cat is not implemented yet; the CLI shape is defined but record rendering still needs to be added",
+    ))
+}

ljx/src/commands/count.rs

@@ -0,0 +1,23 @@
+use logjet::LogjetReader;
+
+use crate::cli::CountArgs;
+use crate::error::Result;
+use crate::input::InputHandle;
+
+pub fn run(args: CountArgs) -> Result<()> {
+    let predicate = args.predicate.build()?;
+    let input = InputHandle::open(&args.input)?;
+    let mut reader = LogjetReader::new(input.into_buf_reader());
+
+    let mut count = 0u64;
+    while let Some(record) = reader.next_record()? {
+        if predicate.matches(&record) {
+            count = count
+                .checked_add(1)
+                .ok_or(logjet::Error::NumericOverflow("count"))?;
+        }
+    }
+
+    println!("{count}");
+    Ok(())
+}

ljx/src/commands/filter.rs

@@ -0,0 +1,36 @@
+use std::io::Write;
+
+use logjet::{LogjetReader, LogjetWriter, WriterConfig};
+
+use crate::cli::FilterArgs;
+use crate::error::Result;
+use crate::input::{InputHandle, open_output};
+
+pub fn run(args: FilterArgs) -> Result<()> {
+    let predicate = args.predicate.build()?;
+    let input = InputHandle::open(&args.input)?;
+    let output = open_output(&args.output)?;
+    let config = WriterConfig {
+        block_target_size: args.block_target_size,
+        codec: args.codec.into(),
+        sync_marker: logjet::DEFAULT_SYNC_MARKER,
+    };
+
+    let mut reader = LogjetReader::new(input.into_buf_reader());
+    let mut writer = LogjetWriter::with_config(output, config);
+
+    while let Some(record) = reader.next_record()? {
+        if predicate.matches(&record) {
+            writer.push(
+                record.record_type,
+                record.seq,
+                record.ts_unix_ns,
+                &record.payload,
+            )?;
+        }
+    }
+
+    let mut output = writer.into_inner()?;
+    output.flush()?;
+    Ok(())
+}

ljx/src/commands/join.rs

@@ -0,0 +1,8 @@
+use crate::cli::JoinArgs;
+use crate::error::{Error, Result};
+
+pub fn run(_args: JoinArgs) -> Result<()> {
+    Err(Error::Unimplemented(
+        "join is not implemented yet; the CLI shape is defined but ordered multi-segment merging still needs to be added",
+    ))
+}

ljx/src/commands/mod.rs

@@ -0,0 +1,6 @@
+pub mod cat;
+pub mod count;
+pub mod filter;
+pub mod join;
+pub mod split;
+pub mod stats;

ljx/src/commands/split.rs

@@ -0,0 +1,8 @@
+use crate::cli::SplitArgs;
+use crate::error::{Error, Result};
+
+pub fn run(_args: SplitArgs) -> Result<()> {
+    Err(Error::Unimplemented(
+        "split is not implemented yet; the CLI shape is defined but block-preserving file partitioning still needs to be added",
+    ))
+}

ljx/src/commands/stats.rs

@@ -0,0 +1,8 @@
+use crate::cli::StatsArgs;
+use crate::error::{Error, Result};
+
+pub fn run(_args: StatsArgs) -> Result<()> {
+    Err(Error::Unimplemented(
+        "stats is not implemented yet; the CLI shape is defined but streaming summaries still need to be added",
+    ))
+}

ljx/src/error.rs

@@ -0,0 +1,37 @@
+use std::fmt::{Display, Formatter};
+use std::io;
+
+pub type Result<T> = std::result::Result<T, Error>;
+
+#[derive(Debug)]
+pub enum Error {
+    Io(io::Error),
+    Logjet(logjet::Error),
+    Usage(String),
+    Unimplemented(&'static str),
+}
+
+impl Display for Error {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::Io(err) => write!(f, "{err}"),
+            Self::Logjet(err) => write!(f, "{err}"),
+            Self::Usage(msg) => write!(f, "{msg}"),
+            Self::Unimplemented(msg) => write!(f, "{msg}"),
+        }
+    }
+}
+
+impl std::error::Error for Error {}
+
+impl From<io::Error> for Error {
+    fn from(value: io::Error) -> Self {
+        Self::Io(value)
+    }
+}
+
+impl From<logjet::Error> for Error {
+    fn from(value: logjet::Error) -> Self {
+        Self::Logjet(value)
+    }
+}