diff --git a/Cargo.lock b/Cargo.lock index bba969e0..ab233218 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1171,6 +1171,7 @@ dependencies = [ "tracing", "tracing-subscriber", "webpki-roots", + "zstd 0.12.3+zstd.1.5.2", ] [[package]] @@ -3112,7 +3113,7 @@ dependencies = [ "pbkdf2", "sha1", "time 0.3.17", - "zstd", + "zstd 0.11.2+zstd.1.5.2", ] [[package]] @@ -3121,7 +3122,16 @@ version = "0.11.2+zstd.1.5.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "20cc960326ece64f010d2d2107537f26dc589a6573a316bd5b1dba685fa5fde4" dependencies = [ - "zstd-safe", + "zstd-safe 5.0.2+zstd.1.5.2", +] + +[[package]] +name = "zstd" +version = "0.12.3+zstd.1.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "76eea132fb024e0e13fd9c2f5d5d595d8a967aa72382ac2f9d39fcc95afd0806" +dependencies = [ + "zstd-safe 6.0.4+zstd.1.5.4", ] [[package]] @@ -3134,6 +3144,16 @@ dependencies = [ "zstd-sys", ] +[[package]] +name = "zstd-safe" +version = "6.0.4+zstd.1.5.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7afb4b54b8910cf5447638cb54bf4e8a65cbedd783af98b98c62ffe91f185543" +dependencies = [ + "libc", + "zstd-sys", +] + [[package]] name = "zstd-sys" version = "2.0.7+zstd.1.5.4" diff --git a/crates/kumod/Cargo.toml b/crates/kumod/Cargo.toml index 309adcb0..2129ac8a 100644 --- a/crates/kumod/Cargo.toml +++ b/crates/kumod/Cargo.toml @@ -40,6 +40,7 @@ tokio-rustls = "0.23" tracing = "0.1" tracing-subscriber = {version="0.3", features=["env-filter"]} webpki-roots = "0.22" +zstd = "0.12" [dev-dependencies] k9 = "0.11" diff --git a/crates/kumod/src/logging.rs b/crates/kumod/src/logging.rs index ab9798e7..cc2c7a8f 100644 --- a/crates/kumod/src/logging.rs +++ b/crates/kumod/src/logging.rs @@ -11,6 +11,7 @@ use std::io::Write; use std::path::PathBuf; use std::sync::Mutex; use std::thread::JoinHandle; +use zstd::stream::write::{AutoFinishEncoder, Encoder}; static LOGGER: OnceCell = OnceCell::new(); @@ -18,13 +19,22 @@ static LOGGER: OnceCell = OnceCell::new(); pub struct LogFileParams { /// Where to place the log files pub log_dir: PathBuf, + /// How many uncompressed bytes to allow per file segment #[serde(default = "LogFileParams::default_max_file_size")] pub max_file_size: u64, + /// Maximum number of outstanding items to be logged before + /// the submission will block; helps to avoid runaway issues + /// spiralling out of control. + #[serde(default = "LogFileParams::default_back_pressure")] + pub back_pressure: u64, } impl LogFileParams { fn default_max_file_size() -> u64 { - 128_000_000 + 1_000_000_000 + } + fn default_back_pressure() -> u64 { + 128_000 } } @@ -47,7 +57,7 @@ impl Logger { std::fs::create_dir_all(¶ms.log_dir) .with_context(|| format!("creating log directory {}", params.log_dir.display()))?; - let (sender, receiver) = async_channel::bounded(128000); + let (sender, receiver) = async_channel::bounded(params.back_pressure); let thread = std::thread::spawn(move || { let runtime = tokio::runtime::Builder::new_current_thread() .enable_time() @@ -69,7 +79,7 @@ impl Logger { async fn logger_thread(params: LogFileParams, receiver: Receiver) { struct OpenedFile { - file: File, + file: AutoFinishEncoder<'static, File>, name: PathBuf, written: u64, } @@ -92,7 +102,9 @@ impl Logger { .with_context(|| format!("open log file {name:?}"))?; file.replace(OpenedFile { - file: f, + file: Encoder::new(f, 0) + .context("set up zstd encoder")? + .auto_finish(), name, written: 0, }); @@ -188,6 +200,10 @@ pub struct JsonLogRecord { /// The time at which the message was initially received and created #[serde(with = "chrono::serde::ts_seconds")] pub created: DateTime, + /// The number of delivery attempts that have been made. + /// Note that this may be approximate after a restart; use the + /// number of logged events to determine the true number + pub num_attempts: u16, } pub async fn log_disposition( @@ -218,6 +234,7 @@ pub async fn log_disposition( response, timestamp: Utc::now(), created: msg.id().created(), + num_attempts: msg.get_num_attempts(), }; if let Err(err) = logger.log(record).await { tracing::error!("failed to log: {err:#}");