This is an automated email from the ASF dual-hosted git repository. hubcio pushed a commit to branch fix/server-logging-config in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 7ec59e227ff713eab3af769675b567268f88f24a Author: Hubert Gruszecki <[email protected]> AuthorDate: Wed Sep 30 00:01:33 2026 +0200 fix(server)!: make logging honor retention, level, path and file_enabled Hourly rotation filled the max_total_size / max_file_size archive slots (8 by default) with small files. A server kept about 9 hours of logs, and retention never deleted anything. Rotation is now by size only. The rotation thread deletes a rotated file retention after its last write. A line can therefore stay for the fill time of one file plus retention, as config.toml now says. In time zones with daylight saving, the hourly trigger also panicked in the fall-back hour. rolling-file 0.2.0 unwraps the ambiguous local time of that hour, and the panic hook then stops the server. EnvFilter reads a bare word that is not a level as a target, so a typo in logging.level turned off every log line. LogFilter now parses the level strictly in the validator and in late_init, and an invalid value fails the boot. It trims spaces around each directive, because EnvFilter mis-parses them. "none" stays an alias of "off", as the old config.toml documented. file_enabled = false no longer buffers every log line in memory. A relative logging.path no longer depends on the working directory, and the server_logs snapshot reads the same folder as the logger. retention = 0 now turns off deletion by age, and max_total_size still deletes the oldest files. A retention above about 56 years no longer panics the rotation thread and stops the server. [span] directives now work while telemetry is off. Upgrade notes: log files no longer rotate every hour. A config fails the boot if logging.level holds a bare target name or a malformed directive. For example, "info,server=verbose" started at info before, because EnvFilter::new dropped the bad directive. If the working directory has a folder with the logging.path name, logs now go to logging.path under the top-level path instead. --- Cargo.lock | 1 + core/configs/src/common/validators.rs | 62 ++++- core/integration/src/harness/config/server.rs | 5 + core/integration/src/harness/handle/server.rs | 3 + .../server/scenarios/log_rotation_scenario.rs | 90 +++++++ core/server/config.toml | 13 +- core/server/src/snapshot.rs | 13 +- core/server_common/Cargo.toml | 1 + core/server_common/src/log/logger.rs | 294 ++++++++++++++++----- core/server_common/src/log/mod.rs | 2 +- core/server_common/src/log/settings.rs | 124 +++++++++ 11 files changed, 516 insertions(+), 92 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 60cc140f7..6146d01dd 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -12794,6 +12794,7 @@ dependencies = [ "argon2", "bytemuck", "bytes", + "chrono", "compio", "compio-buf", "crossbeam", diff --git a/core/configs/src/common/validators.rs b/core/configs/src/common/validators.rs index 7eba86490..e2d2fce3c 100644 --- a/core/configs/src/common/validators.rs +++ b/core/configs/src/common/validators.rs @@ -23,6 +23,7 @@ use crate::ConfigurationError; use cpu_allocation::{CpuAllocation, allowed_cpus}; use err_trail::ErrContext; use iggy_common::Validatable; +use server_common::log::LogFilter; use std::thread::available_parallelism; /// 1 GiB max segment size. @@ -93,14 +94,17 @@ impl Validatable<ConfigurationError> for PersonalAccessTokenConfig { impl Validatable<ConfigurationError> for LoggingConfig { fn validate(&self) -> Result<(), ConfigurationError> { - if self.level.is_empty() { - eprintln!("logging.level is supposed be configured"); + if let Err(error) = self.level.parse::<LogFilter>() { + eprintln!( + "Configured logging.level {:?} is invalid: {error}", + self.level + ); return Err(ConfigurationError::InvalidConfigurationValue); } - if self.retention.as_secs() < 1 { + if !self.retention.is_zero() && self.retention.as_secs() < 1 { eprintln!( - "Configured logging.retention {} is less than minimum 1 second", + "Configured logging.retention {} is less than minimum 1 second, use 0 to turn off deletion by age", self.retention ); return Err(ConfigurationError::InvalidConfigurationValue); @@ -302,7 +306,7 @@ mod cpu_allocation_tests { #[cfg(test)] mod logging_tests { use super::*; - use iggy_common::IggyDuration; + use iggy_common::{IggyByteSize, IggyDuration}; use std::time::Duration; fn with_sysinfo_print_interval(interval: Duration) -> LoggingConfig { @@ -329,4 +333,52 @@ mod logging_tests { let config = with_sysinfo_print_interval(Duration::ZERO); assert!(config.validate().is_ok()); } + + #[test] + fn level_that_is_not_a_level_is_rejected() { + let config = LoggingConfig { + level: "inof".to_owned(), + ..LoggingConfig::default() + }; + assert!(config.validate().is_err()); + } + + #[test] + fn zero_retention_is_accepted() { + let config = LoggingConfig { + retention: IggyDuration::new(Duration::ZERO), + ..LoggingConfig::default() + }; + assert!(config.validate().is_ok()); + } + + #[test] + fn sub_second_retention_and_check_interval_are_rejected() { + let half_second = IggyDuration::new(Duration::from_millis(500)); + let retention = LoggingConfig { + retention: half_second, + ..LoggingConfig::default() + }; + let check_interval = LoggingConfig { + rotation_check_interval: half_second, + ..LoggingConfig::default() + }; + assert!(retention.validate().is_err()); + assert!(check_interval.validate().is_err()); + } + + #[test] + fn max_total_size_below_max_file_size_is_rejected_unless_unlimited() { + let below = LoggingConfig { + max_file_size: IggyByteSize::from(100), + max_total_size: IggyByteSize::from(50), + ..LoggingConfig::default() + }; + let unlimited = LoggingConfig { + max_total_size: IggyByteSize::from(0), + ..below.clone() + }; + assert!(below.validate().is_err()); + assert!(unlimited.validate().is_ok()); + } } diff --git a/core/integration/src/harness/config/server.rs b/core/integration/src/harness/config/server.rs index 94d755011..d64be20e6 100644 --- a/core/integration/src/harness/config/server.rs +++ b/core/integration/src/harness/config/server.rs @@ -18,6 +18,7 @@ use super::common::{EncryptionConfig, IpAddrKind, TlsConfig}; use bon::Builder; use std::collections::HashMap; +use std::path::PathBuf; #[derive(Debug, Clone, Builder)] pub struct TestServerConfig { @@ -38,6 +39,10 @@ pub struct TestServerConfig { pub extra_envs: HashMap<String, String>, #[builder(into)] pub executable_path: Option<String>, + /// Working directory of the server process. Unset, the server inherits + /// the working directory of the test, which is inside the checkout. + #[builder(into)] + pub current_dir: Option<PathBuf>, /// Bind every enabled transport to port 0 and discover the bound addresses /// from `runtime/current_config.toml` instead of pre-reserving ports. /// Single node only: a cluster roster names every port before boot. diff --git a/core/integration/src/harness/handle/server.rs b/core/integration/src/harness/handle/server.rs index 6c22ab7b1..11972f475 100644 --- a/core/integration/src/harness/handle/server.rs +++ b/core/integration/src/harness/handle/server.rs @@ -979,6 +979,9 @@ impl TestBinary for ServerHandle { command.env_remove("RUST_LOG"); } command.envs(&self.envs); + if let Some(current_dir) = &self.config.current_dir { + command.current_dir(current_dir); + } if let Some(limit) = self.open_files_limit { let rlimit = libc::rlimit { diff --git a/core/integration/tests/server/scenarios/log_rotation_scenario.rs b/core/integration/tests/server/scenarios/log_rotation_scenario.rs index 42820e259..d752f412f 100644 --- a/core/integration/tests/server/scenarios/log_rotation_scenario.rs +++ b/core/integration/tests/server/scenarios/log_rotation_scenario.rs @@ -21,6 +21,7 @@ use iggy_common::{Identifier, IggyByteSize, IggyDuration, IggyExpiry, MaxTopicSi use integration::harness::{TestHarness, TestServerConfig}; use serial_test::parallel; use std::collections::HashMap; +use std::io::{Cursor, Read}; use std::path::Path; use std::sync::LazyLock; use std::time::Duration; @@ -28,12 +29,15 @@ use test_case::test_matrix; use tokio::fs; use tokio::sync::Mutex; use tokio::time::{sleep, timeout}; +use zip::ZipArchive; const RETENTION_SECS: u64 = 30; const OPERATION_TIMEOUT_SECS: u64 = 10; const OPERATION_LOOP_COUNT: usize = 300; const FROM_BYTES_TO_KB: u64 = 1000; const IGGY_LOG_BASE_NAME: &str = "iggy-server.log"; +const LOGGING_INITIALIZED_LINE: &str = "Logging initialized"; +const DECOY_LOG_DIR: &str = "decoy_logs"; static PRINT_LOCK: LazyLock<Mutex<()>> = LazyLock::new(|| Mutex::new(())); @@ -126,6 +130,92 @@ async fn log_rotation_should_be_valid(present_log_config: LogRotationTestConfig) run(&harness, &log_dir, present_log_config).await; } +#[tokio::test] +#[parallel] +async fn given_log_path_existing_in_working_directory_when_server_starts_should_log_under_data_path() + { + // The server runs in this directory, so the decoy has the same relative + // path as the configured log directory. + let working_dir = tempfile::tempdir().expect("create server working directory"); + let decoy = working_dir.path().join(DECOY_LOG_DIR); + std::fs::create_dir(&decoy).expect("create decoy directory in the working directory"); + let extra_envs = HashMap::from([("IGGY_LOGGING_PATH".to_string(), DECOY_LOG_DIR.to_string())]); + let mut harness = TestHarness::builder() + .server( + TestServerConfig::builder() + .extra_envs(extra_envs) + .current_dir(working_dir.path()) + .build(), + ) + .build() + .unwrap(); + + harness.start().await.unwrap(); + + let log_file = harness + .server() + .data_path() + .join(DECOY_LOG_DIR) + .join(IGGY_LOG_BASE_NAME); + assert!(log_file.is_file(), "no log file at {}", log_file.display()); + let leaked = std::fs::read_dir(&decoy) + .expect("read decoy directory") + .count(); + assert_eq!( + leaked, + 0, + "the server wrote logs into the working directory {}", + decoy.display() + ); + + let snapshot = harness + .root_client() + .await + .unwrap() + .snapshot( + SnapshotCompression::Stored, + vec![SystemSnapshotType::ServerLogs], + ) + .await + .unwrap(); + let mut server_logs = String::new(); + ZipArchive::new(Cursor::new(snapshot.0)) + .unwrap() + .by_name(&format!("{}.txt", SystemSnapshotType::ServerLogs)) + .unwrap() + .read_to_string(&mut server_logs) + .unwrap(); + assert!( + server_logs.contains(LOGGING_INITIALIZED_LINE), + "the snapshot read the empty working directory copy, not the logs under the data path" + ); +} + +#[tokio::test] +#[parallel] +async fn given_rust_log_and_config_level_when_server_starts_should_apply_rust_log() { + // Only the startup line from `server_common` tells the two filters apart; + // the harness needs the other info lines to see the cluster form. + let extra_envs = HashMap::from([ + ( + "IGGY_LOGGING_LEVEL".to_string(), + "info,server_common=warn".to_string(), + ), + ("RUST_LOG".to_string(), "info".to_string()), + ]); + let mut harness = TestHarness::builder() + .server(TestServerConfig::builder().extra_envs(extra_envs).build()) + .build() + .unwrap(); + + harness.start().await.unwrap(); + + assert!( + harness.server().stdout_contains("Log filter: info"), + "the server_common startup line is missing, so logging.level overrode RUST_LOG" + ); +} + async fn run(harness: &TestHarness, log_dir: &str, present_log_config: LogRotationTestConfig) { let done_status = false; let present_log_test_title = present_log_config.name.clone(); diff --git a/core/server/config.toml b/core/server/config.toml index 21e3c832b..b1d348b13 100644 --- a/core/server/config.toml +++ b/core/server/config.toml @@ -380,6 +380,8 @@ path = "logs" # Log filtering directive using the same syntax as the RUST_LOG environment variable. # Supports simple levels ("trace", "debug", "info", "warn", "error", "off" or "none") # as well as complex directives like "warn,server=debug,iggy=trace". +# A bare word must be a level: write "target=level" to filter one target. +# An invalid value fails the boot. # Note: RUST_LOG environment variable always takes precedence over this setting. level = "info" @@ -397,14 +399,17 @@ max_file_size = "500 MB" # Maximum total size of all log files. When this size is reached, # the oldest log files will be deleted first. Set it to 0 to allow -# an unlimited number of archived logs. This does not disable time -# based log rotation or per-log-file size limits. +# an unlimited number of archived logs. This does not disable +# retention or per-log-file size limits. max_total_size = "4 GB" -# Time interval for checking log rotation status. Avoid less than 1s. +# Time interval for checking log rotation status. Must be at least 1 s. rotation_check_interval = "1 h" -# Time to retain log files before deletion. Avoid less than 1s, too. +# Time to keep a rotated log file after its last write. Must be at least +# 1 s, or 0 to turn off deletion by age (max_total_size still deletes the +# oldest files). The log file in use is never deleted, so a line can stay +# on disk for up to the time to fill one file plus the retention. retention = "7 days" # Interval for printing process and host usage (CPU, memory, disk, clients, diff --git a/core/server/src/snapshot.rs b/core/server/src/snapshot.rs index 87122b8dc..ace0f9262 100644 --- a/core/server/src/snapshot.rs +++ b/core/server/src/snapshot.rs @@ -196,16 +196,9 @@ fn process_list() -> io::Result<Vec<u8>> { } fn server_logs(server_config: &ServerConfig) -> io::Result<Vec<u8>> { - // Mirror the logger's path derivation (server_common `Logging::late_init`): - // it canonicalizes the configured subdirectory before joining the system - // path, so a relative `logging.path` that already exists resolves against the - // CWD. Skipping the canonicalize here would read a different (often empty) - // directory than the one the logger actually writes to. - let logs_subdirectory = PathBuf::from(&server_config.logging.path); - let logs_subdirectory = logs_subdirectory - .canonicalize() - .unwrap_or(logs_subdirectory); - let logs_path = PathBuf::from(server_config.get_system_path()).join(logs_subdirectory); + // Must match the path that server_common `Logging::late_init` writes to. + let logs_path = + PathBuf::from(server_config.get_system_path()).join(&server_config.logging.path); let mut log_files = Vec::new(); for entry in std::fs::read_dir(&logs_path)? { let entry = entry?; diff --git a/core/server_common/Cargo.toml b/core/server_common/Cargo.toml index ce10ca165..d143081c2 100644 --- a/core/server_common/Cargo.toml +++ b/core/server_common/Cargo.toml @@ -69,6 +69,7 @@ twox-hash = { workspace = true } nix = { workspace = true } [dev-dependencies] +chrono = { workspace = true } serial_test = { workspace = true } tempfile = { workspace = true } diff --git a/core/server_common/src/log/logger.rs b/core/server_common/src/log/logger.rs index b1d203847..0f9adaa2b 100644 --- a/core/server_common/src/log/logger.rs +++ b/core/server_common/src/log/logger.rs @@ -16,7 +16,9 @@ // under the License. use super::runtime::CompioRuntime; -use super::settings::{LoggingSettings, TelemetrySettings, TelemetryTransport}; +use super::settings::{ + LogFilter, LogFilterError, LoggingSettings, TelemetrySettings, TelemetryTransport, +}; use iggy_common::{IggyByteSize, IggyDuration}; use opentelemetry::KeyValue; use opentelemetry::global; @@ -43,7 +45,7 @@ use tracing_subscriber::fmt::format::DefaultVisitor; use tracing_subscriber::layer::SubscriberExt; use tracing_subscriber::util::SubscriberInitExt; use tracing_subscriber::{ - EnvFilter, Layer, Registry, filter::LevelFilter, fmt, fmt::MakeWriter, fmt::format::Format, + EnvFilter, Layer, Registry, fmt, fmt::MakeWriter, fmt::format::Format, layer::Identity, layer::Layered, reload, reload::Handle, }; @@ -60,6 +62,9 @@ pub enum LogError { #[error("Logging file reload failure")] FileReloadFailure, + + #[error("Invalid logging level: {0}")] + InvalidLevel(#[from] LogFilterError), } // Writer that does nothing @@ -175,7 +180,7 @@ impl Logging { // - EnvFilter: Applied FIRST via .with() to filter all subsequent layers // - Stdout layer: writes to NullWriter (discarded), replaced with real stdout in late_init // - File layer: writes to in-memory buffer, replaced with rolling file in late_init - // - Telemetry layers: no-op placeholders (LevelFilter::OFF), replaced if telemetry enabled + // - Telemetry layers: no-op placeholders, replaced if telemetry enabled // // EnvFilter MUST be applied first. All subsequent layers are typed for FilteredRegistry. // This ensures the filter is evaluated before any output layer processes events. @@ -190,14 +195,13 @@ impl Logging { // All output layers are typed for FilteredRegistry (Registry + EnvFilter) let mut layers: Vec<Box<dyn Layer<FilteredRegistry> + Send + Sync>> = vec![]; - // Telemetry layers - no-op placeholders (LevelFilter::OFF) replaced in late_init if enabled - let (otel_logs_layer, otel_logs_reload_handle) = - reload::Layer::new(LevelFilter::OFF.boxed()); + // Telemetry layers - no-op placeholders replaced in late_init if enabled + let (otel_logs_layer, otel_logs_reload_handle) = reload::Layer::new(Self::inactive_layer()); self.otel_logs_reload_handle = Some(otel_logs_reload_handle); layers.push(Box::new(otel_logs_layer)); let (otel_traces_layer, otel_traces_reload_handle) = - reload::Layer::new(LevelFilter::OFF.boxed()); + reload::Layer::new(Self::inactive_layer()); self.otel_traces_reload_handle = Some(otel_traces_reload_handle); layers.push(Box::new(otel_traces_layer)); @@ -244,7 +248,7 @@ impl Logging { std::env::var("RUST_LOG").unwrap() } else { // Use config level as EnvFilter directive - let env_filter = EnvFilter::new(&config.level); + let env_filter: EnvFilter = config.level.parse::<LogFilter>()?.into(); self.env_filter_reload_handle .as_ref() .ok_or(LogError::FilterReloadFailure)? @@ -271,13 +275,8 @@ impl Logging { self.dump_to_stdout(); // Initialize file logging if enabled - let logs_path = if config.file_enabled { - let base_directory = PathBuf::from(base_directory); - let logs_subdirectory = PathBuf::from(config.path.clone()); - let logs_subdirectory = logs_subdirectory - .canonicalize() - .unwrap_or(logs_subdirectory); - let logs_path = base_directory.join(logs_subdirectory); + let (file_layer, logs_path) = if config.file_enabled { + let logs_path = PathBuf::from(base_directory).join(&config.path); if let Err(e) = fs::create_dir_all(&logs_path) { warn!("Failed to create logs directory {logs_path:?}: {e}"); @@ -297,21 +296,9 @@ impl Logging { } let max_files = Self::calculate_max_files(config.max_total_size, config.max_file_size); - - // If max_file_size == 0, then keep interpreting behavior as same - // as fn IggyByteSize::as_human_string_with_zero_as_unlimited do. - // This will cover all log rotations if expecting unlimited. - let mut condition_builder = RollingConditionBasic::new(); - let max_file_size_bytes = config.max_file_size.as_bytes_u64(); - - if max_file_size_bytes != 0 { - condition_builder = condition_builder.max_size(max_file_size_bytes).hourly(); - } - let condition = condition_builder; - let file_appender = BasicRollingFileAppender::new( logs_path.join(IGGY_LOG_FILE_PREFIX), - condition, + Self::rolling_condition(config.max_file_size), max_files, ) .map_err(|e| { @@ -332,16 +319,17 @@ impl Logging { .boxed(); self.file_guard = Some(file_guard); - self.file_reload_handle - .as_ref() - .ok_or(LogError::FileReloadFailure)? - .modify(|layer| *layer = file_layer) - .expect("Failed to modify file layer"); - - Some(logs_path) + (file_layer, Some(logs_path)) } else { - None + (Self::inactive_layer(), None) }; + self.file_reload_handle + .as_ref() + .ok_or(LogError::FileReloadFailure)? + .modify(|layer| *layer = file_layer) + .expect("Failed to modify file layer"); + // Both dumps are done and the early layer is replaced, so free its lines. + *self.early_logs_buffer.lock().unwrap() = Vec::new(); // Initialize telemetry if enabled if telemetry_config.enabled { @@ -463,6 +451,14 @@ impl Logging { Format::default().with_thread_names(true) } + /// Stands in for an output that is off. It must enable every event: the + /// `Vec` of output layers enables an event only when all of its layers do, + /// so a disabling placeholder such as `LevelFilter::OFF` drops the events + /// that a `[span]` directive in the filter enables. + fn inactive_layer() -> Box<dyn Layer<FilteredRegistry> + Send + Sync> { + Identity::new().boxed() + } + fn calculate_max_files( max_total_size_bytes: IggyByteSize, max_file_size_bytes: IggyByteSize, @@ -480,6 +476,17 @@ impl Logging { } } + /// Rotates on size only. A time trigger fills the `max_files` archive + /// slots with small files and deletes logs long before `retention` ends. + /// A zero `max_file_size` never rotates. + fn rolling_condition(max_file_size: IggyByteSize) -> RollingConditionBasic { + let condition = RollingConditionBasic::new(); + match max_file_size.as_bytes_u64() { + 0 => condition, + max_size => condition.max_size(max_size), + } + } + fn install_log_rotation_handler( &self, config: &LoggingSettings, @@ -635,7 +642,8 @@ impl Logging { let mut removed_files_count = 0; let cutoff = if !retention.is_zero() { match SystemTime::now().duration_since(UNIX_EPOCH) { - Ok(now) => Some(now - retention.get_duration()), + // A retention longer than the time since the epoch expires nothing. + Ok(now) => now.checked_sub(retention.get_duration()), Err(e) => { warn!("Failed to get current time: {e}"); return; @@ -781,30 +789,15 @@ impl<'writer> FormatFields<'writer> for NoAnsiFields { #[cfg(test)] mod tests { use super::*; + use crate::log::TelemetryEndpointSettings; + use chrono::{Local, TimeDelta}; + use rolling_file::RollingCondition; + use std::fs::File; + use std::path::Path; use tempfile::TempDir; + use tracing::info_span; - #[test] - fn test_log_directory_creation() { - let temp_dir = TempDir::new().expect("Failed to create temporary directory"); - let base_path = temp_dir.path().to_str().unwrap().to_string(); - let log_subdir = "test_logs".to_string(); - - let log_path = PathBuf::from(&base_path).join(&log_subdir); - assert!(!log_path.exists()); - fs::create_dir_all(&log_path).expect("Failed to create log directory"); - assert!(log_path.exists()); - } - - #[test] - fn test_disk_space_check() { - let temp_dir = TempDir::new().expect("Failed to create temporary directory"); - let log_path = temp_dir.path(); - let result = fs2::available_space(log_path); - assert!(result.is_ok()); - - let available_space = result.unwrap(); - assert!(available_space > 0); - } + const HOUR: Duration = Duration::from_secs(3600); #[test] fn test_calculate_max_files() { @@ -842,23 +835,180 @@ mod tests { } #[test] - fn test_cleanup_log_files_functions() { - use std::time::Duration; - let temp_dir = TempDir::new().expect("Failed to create temporary directory"); - let log_path = temp_dir.path().to_path_buf(); + fn given_max_file_size_when_hours_pass_should_rotate_only_on_size() { + let mut condition = Logging::rolling_condition(IggyByteSize::from(1_000)); + let start = Local::now(); + let two_hours_later = start + TimeDelta::hours(2); + + assert!(!condition.should_rollover(&start, 0)); + assert!( + !condition.should_rollover(&two_hours_later, 999), + "rotated on time while the file is below max_file_size" + ); + assert!(condition.should_rollover(&two_hours_later, 1_000)); + } + + #[test] + fn given_retention_when_cleaning_should_delete_only_expired_archives() { + let logs = TempDir::new().expect("create logs directory"); + let active = create_log_file(logs.path(), IGGY_LOG_FILE_PREFIX, 1, 2 * HOUR); + let expired = create_log_file(logs.path(), &archive_name(2), 1, 2 * HOUR); + let recent = create_log_file(logs.path(), &archive_name(1), 1, Duration::ZERO); + let unrelated = create_log_file(logs.path(), "notes.txt", 1, 2 * HOUR); + + Logging::cleanup_log_files( + &logs.path().to_path_buf(), + IggyDuration::new(HOUR), + IggyByteSize::from(0), + IggyByteSize::from(0), + ); + + assert!(!expired.exists(), "the expired archive was kept"); + for kept in [&active, &recent, &unrelated] { + assert!(kept.exists(), "{} was deleted", kept.display()); + } + } + + #[test] + fn given_retention_that_never_expires_when_cleaning_should_keep_old_archives() { + const LONGER_THAN_SINCE_EPOCH: Duration = Duration::from_secs(100 * 365 * 24 * 3600); + for retention in [Duration::ZERO, LONGER_THAN_SINCE_EPOCH] { + let logs = TempDir::new().expect("create logs directory"); + let archive = create_log_file(logs.path(), &archive_name(1), 1, 2 * HOUR); + + Logging::cleanup_log_files( + &logs.path().to_path_buf(), + IggyDuration::new(retention), + IggyByteSize::from(0), + IggyByteSize::from(0), + ); + + assert!( + archive.exists(), + "retention {retention:?} deleted an archive" + ); + } + } + + #[test] + fn given_archives_over_total_size_when_cleaning_should_delete_oldest_first() { + let logs = TempDir::new().expect("create logs directory"); + let minute = Duration::from_secs(60); + let active = create_log_file(logs.path(), IGGY_LOG_FILE_PREFIX, 100, Duration::ZERO); + let newest = create_log_file(logs.path(), &archive_name(1), 100, minute); + let middle = create_log_file(logs.path(), &archive_name(2), 100, 2 * minute); + let oldest = create_log_file(logs.path(), &archive_name(3), 100, 3 * minute); + + // The active file does not count, so deleting the oldest archive is enough. Logging::cleanup_log_files( - &log_path, - IggyDuration::new(Duration::from_secs(3600)), - IggyByteSize::from(2048 * 1024), - IggyByteSize::from(512 * 1024), + &logs.path().to_path_buf(), + IggyDuration::new(Duration::ZERO), + IggyByteSize::from(250), + IggyByteSize::from(100), + ); + + assert!(!oldest.exists(), "the oldest archive was kept"); + for kept in [&active, &newest, &middle] { + assert!(kept.exists(), "{} was deleted", kept.display()); + } + } + + #[test] + fn given_span_directive_when_event_is_inside_span_should_reach_outputs() { + let captured = Arc::new(Mutex::new(Vec::new())); + let filter: EnvFilter = "warn,[probe]=info" + .parse::<LogFilter>() + .expect("valid filter") + .into(); + let (filter_layer, _filter_handle) = reload::Layer::new(filter); + let outputs: Vec<Box<dyn Layer<FilteredRegistry> + Send + Sync>> = vec![ + Logging::inactive_layer(), + fmt::layer() + .with_ansi(false) + .with_writer(VecStringMakeWriter(captured.clone())) + .boxed(), + ]; + let subscriber = Registry::default().with(filter_layer).with(outputs); + + tracing::subscriber::with_default(subscriber, || { + info_span!("probe").in_scope(|| info!("inside the span")); + info!("outside the span"); + }); + + let lines = captured.lock().unwrap().concat(); + assert!( + lines.contains("inside the span"), + "the span directive was dropped: {lines:?}" + ); + assert!( + !lines.contains("outside the span"), + "the warn default was ignored: {lines:?}" ); } + // The only test here that calls `early_init`: it installs the process-wide subscriber. #[test] - fn test_logging_creation() { - let logging = Logging::new("0.0.0-test"); - assert!(logging.stdout_guard.is_none()); - assert!(logging.file_guard.is_none()); - assert!(logging.env_filter_reload_handle.is_none()); + fn given_file_output_disabled_when_logging_after_late_init_should_not_buffer_lines() { + let mut logging = Logging::new("0.0.0-test"); + logging.early_init(); + error!("line before late_init"); + assert!( + !logging.early_logs_buffer.lock().unwrap().is_empty(), + "the early layer did not capture the probe line; RUST_LOG may filter it out" + ); + + logging + .late_init( + String::new(), + &file_disabled_settings(), + &telemetry_disabled_settings(), + ) + .expect("late_init"); + error!("line after late_init"); + + let buffered = logging.early_logs_buffer.lock().unwrap(); + assert!( + buffered.is_empty(), + "the early buffer still holds {} lines", + buffered.len() + ); + } + + fn archive_name(index: usize) -> String { + format!("{IGGY_LOG_FILE_PREFIX}.{index}") + } + + fn create_log_file(directory: &Path, name: &str, size: u64, age: Duration) -> PathBuf { + let path = directory.join(name); + let file = File::create(&path).expect("create log file"); + file.set_len(size).expect("size log file"); + file.set_modified(SystemTime::now() - age) + .expect("age log file"); + path + } + + fn file_disabled_settings() -> LoggingSettings { + LoggingSettings { + path: "logs".to_owned(), + level: "info".to_owned(), + file_enabled: false, + max_file_size: IggyByteSize::from(0), + max_total_size: IggyByteSize::from(0), + rotation_check_interval: IggyDuration::new(HOUR), + retention: IggyDuration::new(Duration::ZERO), + } + } + + fn telemetry_disabled_settings() -> TelemetrySettings { + let endpoint = TelemetryEndpointSettings { + transport: TelemetryTransport::HTTP, + endpoint: String::new(), + }; + TelemetrySettings { + enabled: false, + service_name: String::new(), + logs: endpoint.clone(), + traces: endpoint, + } } } diff --git a/core/server_common/src/log/mod.rs b/core/server_common/src/log/mod.rs index c932681ae..e40b1c9ea 100644 --- a/core/server_common/src/log/mod.rs +++ b/core/server_common/src/log/mod.rs @@ -21,7 +21,7 @@ pub mod settings; pub use logger::{LogError, Logging}; pub use settings::{ - LoggingSettings, TelemetryEndpointSettings, TelemetrySettings, TelemetryTransport, + LogFilter, LoggingSettings, TelemetryEndpointSettings, TelemetrySettings, TelemetryTransport, }; /// Log the CI build fingerprint (or a developer-build notice). diff --git a/core/server_common/src/log/settings.rs b/core/server_common/src/log/settings.rs index a540dcd9e..d4c7c4c15 100644 --- a/core/server_common/src/log/settings.rs +++ b/core/server_common/src/log/settings.rs @@ -25,6 +25,8 @@ use derive_more::Display; use serde::{Deserialize, Serialize}; use std::str::FromStr; +use tracing_subscriber::EnvFilter; +use tracing_subscriber::filter::{LevelFilter, ParseError}; use iggy_common::{IggyByteSize, IggyDuration}; @@ -39,6 +41,67 @@ pub struct LoggingSettings { pub retention: IggyDuration, } +/// `logging.level` in the `RUST_LOG` syntax, parsed more strictly than +/// `EnvFilter::new` does. `EnvFilter` reads a bare word that is not a level +/// as a target name and enables only that target, so a typo such as `inof` +/// turned off every log line, errors included, without a warning. +#[derive(Debug)] +pub struct LogFilter(EnvFilter); + +#[derive(Debug, thiserror::Error)] +pub enum LogFilterError { + #[error("no filter directive")] + Empty, + + #[error( + "`{0}` is not a log level (trace, debug, info, warn, error or off); write `{0}=<level>` to filter one target" + )] + NotALevel(String), + + #[error(transparent)] + InvalidDirective(#[from] ParseError), +} + +/// The 0.8.0 and 0.9.0 `config.toml` listed `none` as a level, but `tracing` +/// has no such level. +const NONE_LEVEL_ALIAS: &str = "none"; +const OFF_LEVEL: &str = "off"; + +impl FromStr for LogFilter { + type Err = LogFilterError; + + fn from_str(level: &str) -> Result<Self, Self::Err> { + // `EnvFilter` mis-parses a directive with spaces around it. + let directives: Vec<&str> = level + .split(',') + .map(str::trim) + .filter(|directive| !directive.is_empty()) + .map(|directive| { + if directive.eq_ignore_ascii_case(NONE_LEVEL_ALIAS) { + OFF_LEVEL + } else { + directive + } + }) + .collect(); + if directives.is_empty() { + return Err(LogFilterError::Empty); + } + if let Some(word) = directives.iter().copied().find(|directive| { + !directive.contains(['=', '[']) && directive.parse::<LevelFilter>().is_err() + }) { + return Err(LogFilterError::NotALevel(word.to_owned())); + } + Ok(Self(EnvFilter::builder().parse(directives.join(","))?)) + } +} + +impl From<LogFilter> for EnvFilter { + fn from(filter: LogFilter) -> Self { + filter.0 + } +} + #[derive(Debug, Clone)] pub struct TelemetrySettings { pub enabled: bool, @@ -72,3 +135,64 @@ impl FromStr for TelemetryTransport { } } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn given_levels_and_directives_when_parsing_should_accept() { + for level in [ + "off", + "INFO", + "warn,server=debug,iggy=trace", + "info,[request]=debug", + ] { + assert!( + level.parse::<LogFilter>().is_ok(), + "rejected valid filter {level:?}" + ); + } + } + + #[test] + fn given_word_that_is_not_a_level_when_parsing_should_reject() { + for level in ["inof", "server", "warn,inof", "warn, inof"] { + assert!( + matches!( + level.parse::<LogFilter>(), + Err(LogFilterError::NotALevel(_)) + ), + "accepted {level:?}" + ); + } + } + + #[test] + fn given_none_when_parsing_should_mean_off() { + let none: EnvFilter = "none".parse::<LogFilter>().unwrap().into(); + let off: EnvFilter = "off".parse::<LogFilter>().unwrap().into(); + assert_eq!(none.to_string(), off.to_string()); + } + + #[test] + fn given_spaces_around_directives_when_parsing_should_keep_every_directive() { + let spaced: EnvFilter = " warn , server=debug ".parse::<LogFilter>().unwrap().into(); + let compact: EnvFilter = "warn,server=debug".parse::<LogFilter>().unwrap().into(); + assert_eq!(spaced.to_string(), compact.to_string()); + } + + #[test] + fn given_malformed_or_empty_filter_when_parsing_should_reject() { + assert!(matches!( + "server=verbose".parse::<LogFilter>(), + Err(LogFilterError::InvalidDirective(_)) + )); + for level in ["", ","] { + assert!( + matches!(level.parse::<LogFilter>(), Err(LogFilterError::Empty)), + "accepted {level:?}" + ); + } + } +}
