refactor(obs): enhance log rotation robustness and refine filter logic (#2155)

Signed-off-by: houseme <housemecn@gmail.com>
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

crates/obs/src/telemetry/filter.rs

@@ -17,8 +17,8 @@
 //! This module provides helper functions for building `EnvFilter` instances
 //! used across different logging backends (stdout, file, OpenTelemetry).
 
-use smallvec::SmallVec;
-use tracing_subscriber::EnvFilter;
+use std::env;
+use tracing_subscriber::{EnvFilter, filter::LevelFilter};
 
 /// Build an `EnvFilter` from the given log level string.
 ///
@@ -37,7 +37,8 @@ use tracing_subscriber::EnvFilter;
 /// # Returns
 /// A configured `EnvFilter` ready to be attached to a `tracing_subscriber` registry.
 fn is_verbose_level(level: &str) -> bool {
-    matches!(level.to_ascii_lowercase().as_str(), "trace" | "debug")
+    let s = level.trim().to_ascii_lowercase();
+    s.contains("trace") || s.contains("debug")
 }
 
 fn rust_log_requests_verbose(rust_log: &str) -> bool {
@@ -47,8 +48,14 @@ fn rust_log_requests_verbose(rust_log: &str) -> bool {
             return false;
         }
 
-        let level = directive.rsplit('=').next().unwrap_or("").trim();
-        is_verbose_level(level)
+        // If the directive is just a level (e.g. "debug"), check it.
+        // If it's a module=level (e.g. "my_crate=debug"), we split by '='.
+        // Note: rsplit keeps the last part as the first element of iterator if we take next().
+        if let Some(level_part) = directive.rsplit('=').next() {
+            is_verbose_level(level_part)
+        } else {
+            false
+        }
     })
 }
 
@@ -58,6 +65,13 @@ fn should_suppress_noisy_crates(logger_level: &str, default_level: Option<&str>,
     }
 
     if let Some(rust_log) = rust_log {
+        // If RUST_LOG is present, we check if ANY part of it requests verbose logging.
+        // If the user explicitly asks for debug/trace anywhere, we assume they are debugging
+        // and might want to see third-party logs unless they silence them.
+        // HOWEVER, standard practice is: if RUST_LOG is set, we respect it fully and
+        // adding extra suppressions might be confusing.
+        // But the original logic was: "For non-verbose levels... noisy crates are silenced".
+        // So if RUST_LOG="info", we suppress. If RUST_LOG="debug", we don't.
         return !rust_log_requests_verbose(rust_log);
     }
 
@@ -65,32 +79,53 @@ fn should_suppress_noisy_crates(logger_level: &str, default_level: Option<&str>,
 }
 
 pub(super) fn build_env_filter(logger_level: &str, default_level: Option<&str>) -> EnvFilter {
-    let level = default_level.unwrap_or(logger_level);
-    let rust_log = if default_level.is_none() {
-        std::env::var("RUST_LOG").ok()
-    } else {
-        None
-    };
-    let mut filter = default_level
-        .map(EnvFilter::new)
-        .unwrap_or_else(|| EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new(level)));
+    // 1. Determine the base filter source.
+    // If `default_level` is set (e.g. forced override), we use it.
+    // Otherwise, we look at `RUST_LOG`.
+    // If `RUST_LOG` is missing, we fallback to `logger_level` (from config/env var `RUSTFS_OBS_LOGGER_LEVEL`).
 
-    if should_suppress_noisy_crates(logger_level, default_level, rust_log.as_deref()) {
-        let directives: SmallVec<[(&str, &str); 6]> = smallvec::smallvec![
-            ("hyper", "off"),
-            ("tonic", "off"),
-            ("h2", "off"),
-            ("reqwest", "off"),
-            ("tower", "off"),
+    let rust_log_env = env::var("RUST_LOG").ok();
+
+    // Logic:
+    // - If default_level is Some, use EnvFilter::new(default_level).
+    // - Else if RUST_LOG is set, use EnvFilter::new(rust_log).
+    // - Else use EnvFilter::new(logger_level).
+
+    let mut filter = if let Some(lvl) = default_level {
+        EnvFilter::new(lvl)
+    } else if let Some(ref rust_log) = rust_log_env {
+        EnvFilter::new(rust_log)
+    } else {
+        EnvFilter::new(logger_level)
+    };
+
+    // 2. Apply noisy crate suppression if needed.
+    // We only suppress if the effective configuration is NOT verbose (i.e. not debug/trace).
+    if should_suppress_noisy_crates(logger_level, default_level, rust_log_env.as_deref()) {
+        let directives = [
+            ("hyper", LevelFilter::OFF),
+            ("tonic", LevelFilter::OFF),
+            ("h2", LevelFilter::OFF),
+            ("reqwest", LevelFilter::OFF),
+            ("tower", LevelFilter::OFF),
             // HTTP request logs are demoted to WARN to reduce volume in production.
-            ("rustfs::server::http", "warn"),
+            ("rustfs::server::http", LevelFilter::WARN),
         ];
+
         for (crate_name, level) in directives {
+            // We use `add_directive` which effectively appends to the filter.
+            // If RUST_LOG already specified `hyper=debug`, adding `hyper=off` later MIGHT override it
+            // depending on specificity, but usually the last directive wins or the most specific one.
+            // EnvFilter parsing order matters.
+
+            // To be safe and respectful of RUST_LOG, we should arguably NOT override if RUST_LOG is set?
+            // BUT, the requirement says: "For non-verbose levels... noisy internal crates... are silenced".
+            // So if RUST_LOG="info", we DO want to silence hyper.
+            // If RUST_LOG="hyper=debug", `should_suppress_noisy_crates` returns false, so we won't silence it.
+
             match format!("{crate_name}={level}").parse() {
                 Ok(directive) => filter = filter.add_directive(directive),
                 Err(e) => {
-                    // The directive strings are compile-time constants, so this
-                    // branch should never be reached; emit a diagnostic just in case.
                     eprintln!("obs: invalid log filter directive '{crate_name}={level}': {e}");
                 }
             }
@@ -105,46 +140,95 @@ mod tests {
     use super::*;
 
     #[test]
-    fn test_build_env_filter_default_level_overrides() {
-        // Ensure that providing a default_level uses it instead of logger_level.
-        let filter = build_env_filter("debug", Some("error"));
-        // The Debug output uses `LevelFilter::ERROR` for the error level directive.
-        let dbg = format!("{filter:?}");
-        assert!(
-            dbg.contains("LevelFilter::ERROR"),
-            "Expected 'LevelFilter::ERROR' in filter debug output: {dbg}"
-        );
+    fn test_is_verbose_level() {
+        assert!(is_verbose_level("debug"));
+        assert!(is_verbose_level("trace"));
+        assert!(is_verbose_level("DEBUG"));
+        assert!(is_verbose_level("info,rustfs=debug"));
+        assert!(!is_verbose_level("info"));
+        assert!(!is_verbose_level("warn"));
     }
 
     #[test]
-    fn test_build_env_filter_suppresses_noisy_crates() {
-        // For info level, hyper/tonic/etc. should be suppressed with OFF.
-        let filter = build_env_filter("debug", Some("info"));
-        let dbg = format!("{filter:?}");
-        // The Debug output uses `LevelFilter::OFF` for suppressed crates.
-        assert!(
-            dbg.contains("LevelFilter::OFF"),
-            "Expected 'LevelFilter::OFF' suppression directives in filter: {dbg}"
-        );
+    fn test_rust_log_requests_verbose() {
+        assert!(rust_log_requests_verbose("debug"));
+        assert!(rust_log_requests_verbose("rustfs=debug"));
+        assert!(rust_log_requests_verbose("info,rustfs=trace"));
+        assert!(!rust_log_requests_verbose("info"));
+        assert!(!rust_log_requests_verbose("rustfs=info"));
     }
 
     #[test]
-    fn test_build_env_filter_debug_no_suppression() {
-        // For debug level, our code does NOT inject any OFF directives.
-        let filter = build_env_filter("info", Some("debug"));
-        let dbg = format!("{filter:?}");
-        // Verify the filter builds without panicking and contains the debug level.
-        assert!(!dbg.is_empty());
-        assert!(
-            dbg.contains("LevelFilter::DEBUG"),
-            "Expected 'LevelFilter::DEBUG' in filter debug output: {dbg}"
-        );
-    }
+    fn test_should_suppress() {
+        // Case 1: logger_level="info", no RUST_LOG -> suppress
+        assert!(should_suppress_noisy_crates("info", None, None));
 
-    #[test]
-    fn test_should_suppress_noisy_crates_respects_rust_log_debug() {
+        // Case 2: logger_level="debug", no RUST_LOG -> no suppress
+        assert!(!should_suppress_noisy_crates("debug", None, None));
+
+        // Case 3: RUST_LOG="info" -> suppress
+        assert!(should_suppress_noisy_crates("debug", None, Some("info")));
+
+        // Case 4: RUST_LOG="debug" -> no suppress
         assert!(!should_suppress_noisy_crates("info", None, Some("debug")));
-        assert!(!should_suppress_noisy_crates("info", None, Some("s3=info,hyper=debug")));
-        assert!(should_suppress_noisy_crates("info", None, Some("info")));
+
+        // Case 5: RUST_LOG="rustfs=debug" -> no suppress
+        assert!(!should_suppress_noisy_crates("info", None, Some("rustfs=debug")));
+    }
+
+    #[test]
+    fn test_build_env_filter_injects_suppressions_without_rust_log() {
+        // When RUST_LOG is not set and the base level is non-verbose ("info"),
+        // build_env_filter should inject suppression directives for noisy crates.
+        temp_env::with_var("RUST_LOG", None::<&str>, || {
+            let filter = build_env_filter("info", None);
+            let filter_str = filter.to_string();
+
+            for noisy_crate in ["hyper", "tonic", "h2", "reqwest", "tower"] {
+                assert!(
+                    filter_str.contains(noisy_crate),
+                    "expected EnvFilter to contain suppression directive for `{}`; got `{}`",
+                    noisy_crate,
+                    filter_str
+                );
+            }
+        });
+    }
+
+    #[test]
+    fn test_build_env_filter_respects_verbose_rust_log() {
+        // When RUST_LOG requests a verbose level, automatic noisy-crate suppression
+        // should not be applied, even if the logger_level is non-verbose.
+        temp_env::with_var("RUST_LOG", Some("debug"), || {
+            let filter = build_env_filter("info", None);
+            let filter_str = filter.to_string();
+
+            // We assume "off" is used to silence noisy crates; absence of these
+            // directives indicates that suppression was not injected.
+            for noisy_crate in ["hyper", "tonic", "h2", "reqwest", "tower"] {
+                let directive = format!("{}=off", noisy_crate);
+                assert!(
+                    !filter_str.contains(&directive),
+                    "did not expect EnvFilter to contain `{}` when RUST_LOG is verbose; got `{}`",
+                    directive,
+                    filter_str
+                );
+            }
+        });
+    }
+
+    #[test]
+    fn test_build_env_filter_precedence_of_rust_log_over_logger_level() {
+        // When default_level is None, RUST_LOG should override logger_level.
+        temp_env::with_var("RUST_LOG", Some("warn"), || {
+            let filter = build_env_filter("debug", None);
+            let filter_str = filter.to_string();
+
+            assert!(
+                filter_str.contains("warn"),
+                "expected EnvFilter to reflect RUST_LOG=warn; got `{}`",
+                filter_str
+            );
+        });
     }
 }

crates/obs/src/telemetry/local.rs

@@ -286,7 +286,7 @@ fn init_file_logging_internal(
 /// This prevents world-writable log directories from being a security hazard.
 /// No-ops if permissions are already `0755` or stricter.
 #[cfg(unix)]
-fn ensure_dir_permissions(log_directory: &str) -> Result<(), TelemetryError> {
+pub(super) fn ensure_dir_permissions(log_directory: &str) -> Result<(), TelemetryError> {
     use std::fs::Permissions;
     use std::os::unix::fs::PermissionsExt;
 
@@ -335,7 +335,7 @@ fn ensure_dir_permissions(log_directory: &str) -> Result<(), TelemetryError> {
 ///
 /// # Returns
 /// A [`tokio::task::JoinHandle`] for the spawned cleanup loop.
-fn spawn_cleanup_task(
+pub(super) fn spawn_cleanup_task(
     config: &OtelConfig,
     log_directory: &str,
     log_filename: &str,

crates/obs/src/telemetry/mod.rs

@@ -76,7 +76,7 @@ use rustfs_utils::get_env_opt_str;
 /// application.  Dropping it triggers ordered shutdown of all providers.
 ///
 /// # Errors
-/// Returns [`TelemetryError`] when a backend fails to initialise (e.g., cannot
+/// Returns [`TelemetryError`] when a backend fails to initialize (e.g., cannot
 /// create the log directory, or an OTLP exporter cannot connect).
 pub(crate) fn init_telemetry(config: &OtelConfig) -> Result<OtelGuard, TelemetryError> {
     let environment = config.environment.as_deref().unwrap_or(ENVIRONMENT);

crates/obs/src/telemetry/otel.rs

@@ -33,12 +33,14 @@
 //! compression for efficiency over the wire.
 
 use crate::TelemetryError;
+use crate::cleaner::types::FileMatchMode;
 use crate::config::OtelConfig;
 use crate::global::OBSERVABILITY_METRIC_ENABLED;
 use crate::telemetry::filter::build_env_filter;
 use crate::telemetry::guard::OtelGuard;
 use crate::telemetry::recorder::Recorder;
 use crate::telemetry::resource::build_resource;
+use crate::telemetry::rolling::{RollingAppender, Rotation};
 use metrics::counter;
 use opentelemetry::{global, trace::TracerProvider};
 use opentelemetry_appender_tracing::layer::OpenTelemetryTracingBridge;
@@ -49,11 +51,12 @@ use opentelemetry_sdk::{
     metrics::{PeriodicReader, SdkMeterProvider},
     trace::{RandomIdGenerator, Sampler, SdkTracerProvider},
 };
+use rustfs_config::observability::DEFAULT_OBS_LOG_MAX_SINGLE_FILE_SIZE_BYTES;
 use rustfs_config::{
-    APP_NAME, DEFAULT_OBS_LOG_STDOUT_ENABLED, DEFAULT_OBS_LOGS_EXPORT_ENABLED, DEFAULT_OBS_METRICS_EXPORT_ENABLED,
-    DEFAULT_OBS_TRACES_EXPORT_ENABLED, METER_INTERVAL, SAMPLE_RATIO,
+    APP_NAME, DEFAULT_LOG_KEEP_FILES, DEFAULT_LOG_ROTATION_TIME, DEFAULT_OBS_LOG_STDOUT_ENABLED, DEFAULT_OBS_LOGS_EXPORT_ENABLED,
+    DEFAULT_OBS_METRICS_EXPORT_ENABLED, DEFAULT_OBS_TRACES_EXPORT_ENABLED, METER_INTERVAL, SAMPLE_RATIO,
 };
-use std::{io::IsTerminal, time::Duration};
+use std::{fs, io::IsTerminal, time::Duration};
 use tracing::info;
 use tracing_error::ErrorLayer;
 use tracing_opentelemetry::{MetricsLayer, OpenTelemetryLayer};
@@ -64,6 +67,9 @@ use tracing_subscriber::{
     util::SubscriberInitExt,
 };
 
+// Import helper functions from local.rs (sibling module)
+use super::local::{ensure_dir_permissions, spawn_cleanup_task};
+
 /// Initialize the full OpenTelemetry HTTP pipeline (traces + metrics + logs).
 ///
 /// This function is invoked when at least one OTLP endpoint has been
@@ -108,21 +114,41 @@ pub(super) fn init_observability_http(
         .as_deref()
         .filter(|s| !s.is_empty())
         .map(|s| s.to_string())
-        .unwrap_or_else(|| format!("{root_ep}/v1/traces"));
+        .unwrap_or_else(|| {
+            if !root_ep.is_empty() {
+                format!("{root_ep}/v1/traces")
+            } else {
+                String::new()
+            }
+        });
 
     let metric_ep: String = config
         .metric_endpoint
         .as_deref()
         .filter(|s| !s.is_empty())
         .map(|s| s.to_string())
-        .unwrap_or_else(|| format!("{root_ep}/v1/metrics"));
+        .unwrap_or_else(|| {
+            if !root_ep.is_empty() {
+                format!("{root_ep}/v1/metrics")
+            } else {
+                String::new()
+            }
+        });
 
+    // If log_endpoint is not explicitly set, fallback to root_ep/v1/logs ONLY if root_ep is present.
+    // If both are empty, log_ep is empty, which triggers the fallback to file logging logic.
     let log_ep: String = config
         .log_endpoint
         .as_deref()
         .filter(|s| !s.is_empty())
         .map(|s| s.to_string())
-        .unwrap_or_else(|| format!("{root_ep}/v1/logs"));
+        .unwrap_or_else(|| {
+            if !root_ep.is_empty() {
+                format!("{root_ep}/v1/logs")
+            } else {
+                String::new()
+            }
+        });
 
     // ── Tracer provider (HTTP) ────────────────────────────────────────────────
     let tracer_provider = build_tracer_provider(&trace_ep, config, res.clone(), sampler, use_stdout)?;
@@ -130,48 +156,132 @@ pub(super) fn init_observability_http(
     // ── Meter provider (HTTP) ─────────────────────────────────────────────────
     let meter_provider = build_meter_provider(&metric_ep, config, res.clone(), &service_name, use_stdout)?;
 
-    // ── Logger provider (HTTP) ────────────────────────────────────────────────
-    let logger_provider = build_logger_provider(&log_ep, config, res, use_stdout)?;
-
     #[cfg(unix)]
     let profiling_agent = init_profiler(config);
 
+    // ── Logger Logic ──────────────────────────────────────────────────────────
+    let mut logger_provider: Option<SdkLoggerProvider> = None;
+    let mut otel_bridge = None;
+    let mut file_layer_opt = None; // File layer (File mode)
+    let mut stdout_layer_opt = None; // Stdout layer (File mode)
+    let mut cleanup_handle = None;
+    let mut tracing_guard = None; // Guard for file writer
+    let mut stdout_guard = None; // Guard for stdout writer (File mode)
+
+    // ── Case 1: OTLP Logging
+    if !log_ep.is_empty() {
+        // Init OTLP logger logic.
+        // We initialize the OTLP collector and honor the configured stdout setting
+        // (e.g. via RUSTFS_OBS_USE_STDOUT / config.use_stdout) when building the provider.
+        logger_provider = build_logger_provider(&log_ep, config, res, false)?;
+
+        // Build bridge to capture `tracing` events.
+        otel_bridge = logger_provider
+            .as_ref()
+            .map(|p| OpenTelemetryTracingBridge::new(p).with_filter(build_env_filter(logger_level, None)));
+
+        // Note: We do NOT create a separate `fmt_layer_opt` here; stdout behavior is driven by the provider.
+    }
+    let span_events = if is_production { FmtSpan::CLOSE } else { FmtSpan::FULL };
+    // ── Case 2: File Logging
+    // Supplement: If log_directory is set and no OTLP log endpoint is configured, we enable file logging logic.
+    if let Some(log_directory) = config.log_directory.as_deref().filter(|s| !s.is_empty())
+        && logger_provider.is_none()
+    {
+        let log_filename = config.log_filename.as_deref().unwrap_or(&service_name);
+        let keep_files = config.log_keep_files.unwrap_or(DEFAULT_LOG_KEEP_FILES);
+
+        // 1. Ensure dir exists
+        if let Err(e) = fs::create_dir_all(log_directory) {
+            return Err(TelemetryError::Io(e.to_string()));
+        }
+        // 2. Permissions
+        #[cfg(unix)]
+        ensure_dir_permissions(log_directory)?;
+
+        // 3. Rotation
+        let rotation_str = config
+            .log_rotation_time
+            .as_deref()
+            .unwrap_or(DEFAULT_LOG_ROTATION_TIME)
+            .to_lowercase();
+        let match_mode = match config.log_match_mode.as_deref().map(|s| s.to_lowercase()).as_deref() {
+            Some("prefix") => FileMatchMode::Prefix,
+            _ => FileMatchMode::Suffix,
+        };
+        let rotation = match rotation_str.as_str() {
+            "minutely" => Rotation::Minutely,
+            "hourly" => Rotation::Hourly,
+            "daily" => Rotation::Daily,
+            _ => Rotation::Daily,
+        };
+        let max_single_file_size = config
+            .log_max_single_file_size_bytes
+            .unwrap_or(DEFAULT_OBS_LOG_MAX_SINGLE_FILE_SIZE_BYTES);
+
+        let file_appender =
+            RollingAppender::new(log_directory, log_filename.to_string(), rotation, max_single_file_size, match_mode)?;
+
+        let (non_blocking, guard) = tracing_appender::non_blocking(file_appender);
+        tracing_guard = Some(guard);
+
+        file_layer_opt = Some(
+            tracing_subscriber::fmt::layer()
+                .with_timer(LocalTime::rfc_3339())
+                .with_target(true)
+                .with_ansi(false)
+                .with_thread_names(true)
+                .with_thread_ids(true)
+                .with_file(true)
+                .with_line_number(true)
+                .with_writer(non_blocking)
+                .json()
+                .with_current_span(true)
+                .with_span_list(true)
+                .with_span_events(span_events.clone())
+                .with_filter(build_env_filter(logger_level, None)),
+        );
+
+        // Cleanup task
+        cleanup_handle = Some(spawn_cleanup_task(config, log_directory, log_filename, keep_files));
+
+        info!(
+            "Init file logging at '{}', rotation: {}, keep {} files",
+            log_directory, rotation_str, keep_files
+        );
+    }
+
     // ── Tracing subscriber registry ───────────────────────────────────────────
-    // Build an optional stdout formatting layer. When `log_stdout_enabled` is
-    // false the field is `None` and tracing-subscriber will skip it.
-    let fmt_layer_opt = if config.log_stdout_enabled.unwrap_or(DEFAULT_OBS_LOG_STDOUT_ENABLED) {
-        let enable_color = std::io::stdout().is_terminal();
-        let span_event = if is_production { FmtSpan::CLOSE } else { FmtSpan::FULL };
-        let layer = tracing_subscriber::fmt::layer()
-            .with_timer(LocalTime::rfc_3339())
-            .with_target(true)
-            .with_ansi(enable_color)
-            .with_thread_names(true)
-            .with_thread_ids(true)
-            .with_file(true)
-            .with_line_number(true)
-            .json()
-            .with_current_span(true)
-            .with_span_list(true)
-            .with_span_events(span_event)
-            .with_filter(build_env_filter(logger_level, None));
-        Some(layer)
-    } else {
-        None
-    };
-    let filter = build_env_filter(logger_level, None);
-    let otel_bridge = logger_provider
-        .as_ref()
-        .map(|p| OpenTelemetryTracingBridge::new(p).with_filter(build_env_filter(logger_level, None)));
     let tracer_layer = tracer_provider
         .as_ref()
         .map(|p| OpenTelemetryLayer::new(p.tracer(service_name.to_string())));
     let metrics_layer = meter_provider.as_ref().map(|p| MetricsLayer::new(p.clone()));
 
+    // Optional stdout mirror (matching init_file_logging_internal logic)
+    // This is separate from OTLP stdout logic. If file logging is enabled, we honor its stdout rules.
+    if config.log_stdout_enabled.unwrap_or(DEFAULT_OBS_LOG_STDOUT_ENABLED) || !is_production {
+        let (stdout_nb, stdout_g) = tracing_appender::non_blocking(std::io::stdout());
+        stdout_guard = Some(stdout_g);
+        stdout_layer_opt = Some(
+            tracing_subscriber::fmt::layer()
+                .with_timer(LocalTime::rfc_3339())
+                .with_target(true)
+                .with_ansi(std::io::stdout().is_terminal())
+                .with_thread_names(true)
+                .with_thread_ids(true)
+                .with_file(true)
+                .with_line_number(true)
+                .with_writer(stdout_nb)
+                .with_span_events(span_events)
+                .with_filter(build_env_filter(logger_level, None)),
+        );
+    }
+    let filter = build_env_filter(logger_level, None);
     tracing_subscriber::registry()
         .with(filter)
         .with(ErrorLayer::default())
-        .with(fmt_layer_opt)
+        .with(file_layer_opt) // File
+        .with(stdout_layer_opt) // Stdout (only if file logging enabled it)
         .with(tracer_layer)
         .with(otel_bridge)
         .with(metrics_layer)
@@ -189,9 +299,9 @@ pub(super) fn init_observability_http(
         logger_provider,
         #[cfg(unix)]
         profiling_agent,
-        tracing_guard: None,
-        stdout_guard: None,
-        cleanup_handle: None,
+        tracing_guard,
+        stdout_guard,
+        cleanup_handle,
     })
 }
 
@@ -283,7 +393,7 @@ fn build_meter_provider(
         })
         .build();
 
-    global::set_meter_provider(provider.clone() as SdkMeterProvider);
+    global::set_meter_provider(provider.clone());
     metrics::set_global_recorder(recorder).map_err(|e| TelemetryError::InstallMetricsRecorder(e.to_string()))?;
     OBSERVABILITY_METRIC_ENABLED.set(true).ok();
     Ok(Some(provider))

crates/obs/src/telemetry/rolling.rs

@@ -136,24 +136,39 @@ impl RollingAppender {
             fs::create_dir_all(parent)?;
         }
 
-        // Open in append mode
-        let file = fs::OpenOptions::new().create(true).append(true).open(&path)?;
+        // Add retry logic to handle transient FS issues (e.g. anti-virus, indexing, or quick rename race)
+        const MAX_RETRIES: u32 = 3;
+        let mut last_error = None;
 
-        let meta = file.metadata()?;
-        self.size = meta.len();
+        for i in 0..MAX_RETRIES {
+            // Open in append mode
+            match fs::OpenOptions::new().create(true).append(true).open(&path) {
+                Ok(file) => {
+                    let meta = file.metadata()?;
+                    self.size = meta.len();
 
-        // Seed `last_roll_ts` from the file's modification time so that a
-        // process restart correctly triggers time-based rotation if the active
-        // file belongs to a previous period.
-        if let Ok(modified) = meta.modified() {
-            // Convert SystemTime to jiff::Timestamp
-            if let Ok(ts) = jiff::Timestamp::try_from(modified) {
-                self.last_roll_ts = ts.as_second();
+                    // Seed `last_roll_ts` from the file's modification time so that a
+                    // process restart correctly triggers time-based rotation if the active
+                    // file belongs to a previous period.
+                    if let Ok(modified) = meta.modified() {
+                        // Convert SystemTime to jiff::Timestamp
+                        if let Ok(ts) = jiff::Timestamp::try_from(modified) {
+                            self.last_roll_ts = ts.as_second();
+                        }
+                    }
+
+                    self.file = Some(file);
+                    return Ok(());
+                }
+                Err(e) => {
+                    last_error = Some(e);
+                    // Exponential backoff: 10ms, 20ms, 40ms
+                    thread::sleep(Duration::from_millis(10 * (1 << i)));
+                }
             }
         }
 
-        self.file = Some(file);
-        Ok(())
+        Err(last_error.unwrap_or_else(|| io::Error::other("Failed to open log file after retries")))
     }
 
     fn should_roll(&self, write_len: u64) -> bool {
@@ -173,7 +188,12 @@ impl RollingAppender {
     fn roll(&mut self) -> io::Result<()> {
         // 1. Close current file first to ensure all buffers are flushed to OS (if any)
         // and handle released.
-        self.file = None;
+        if let Some(mut file) = self.file.take()
+            && let Err(e) = file.flush()
+        {
+            eprintln!("Failed to flush log file before rotation: {}", e);
+            return Err(e);
+        }
 
         let active_path = self.active_file_path();
         if !active_path.exists() {
@@ -222,10 +242,14 @@ impl RollingAppender {
                     // Success!
                     // 4. Reset state
                     self.size = 0;
-                    self.last_roll_ts = now.timestamp().as_second();
 
                     // 5. Re-open (creates new active file)
                     self.open_file()?;
+
+                    // Explicitly update last_roll_ts to the rotation time.
+                    // This overrides whatever open_file() derived from mtime, ensuring
+                    // we stick to the logical rotation time.
+                    self.last_roll_ts = now.timestamp().as_second();
                     return Ok(());
                 }
                 Err(e) => {