Track bytes written by WAL

Author: guilload

quickwit/Cargo.lock

@@ -164,7 +164,7 @@ md5 = "0.8"
 mime_guess = "2.0"
 mini-moka = "0.10.3"
 mockall = "0.14"
-mrecordlog = { git = "https://github.com/quickwit-oss/mrecordlog", rev = "306c0a7" }
+mrecordlog = { git = "https://github.com/quickwit-oss/mrecordlog", branch = "guilload/io-outcome" }
 new_string_template = "1.5"
 nom = "8.0"
 numfmt = "1.2"

quickwit/Cargo.toml

@@ -15,8 +15,9 @@
 use std::time::Duration;
 
 use quickwit_common::metrics::MEMORY_METRICS;
+use quickwit_common::rate_limited_warn;
 use tikv_jemallocator::Jemalloc;
-use tracing::{error, warn};
+use tracing::error;
 
 #[cfg(feature = "jemalloc-profiled")]
 #[global_allocator]
@@ -39,28 +40,32 @@ pub async fn jemalloc_metrics_loop() -> tikv_jemalloc_ctl::Result<()> {
     let resident_mib = tikv_jemalloc_ctl::stats::resident::mib()?;
 
     let mut poll_interval = tokio::time::interval(JEMALLOC_METRICS_POLLING_INTERVAL);
+    poll_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
 
     loop {
         poll_interval.tick().await;
 
-        // Many statistics are cached and only updated when the epoch is advanced. A transient
-        // failure here would not be expected, but we log and retry on the next tick rather than
-        // exiting the loop and leaving the metrics frozen forever.
         if let Err(error) = epoch_mib.advance() {
-            warn!(%error, "failed to advance jemalloc epoch");
+            rate_limited_warn!(limit_per_min = 1, %error, "failed to advance jemalloc epoch");
             continue;
         }
         match active_mib.read() {
             Ok(active) => memory_metrics.active_bytes.set(active as i64),
-            Err(error) => warn!(%error, "failed to read jemalloc stats.active"),
+            Err(error) => {
+                rate_limited_warn!(limit_per_min = 1, %error, "failed to read jemalloc stats.active");
+            }
         }
         match allocated_mib.read() {
             Ok(allocated) => memory_metrics.allocated_bytes.set(allocated as i64),
-            Err(error) => warn!(%error, "failed to read jemalloc stats.allocated"),
+            Err(error) => {
+                rate_limited_warn!(limit_per_min = 1, %error, "failed to read jemalloc stats.allocated");
+            }
         }
         match resident_mib.read() {
             Ok(resident) => memory_metrics.resident_bytes.set(resident as i64),
-            Err(error) => warn!(%error, "failed to read jemalloc stats.resident"),
+            Err(error) => {
+                rate_limited_warn!(limit_per_min = 1, %error, "failed to read jemalloc stats.resident");
+            }
         }
     }
 }

quickwit/quickwit-cli/src/jemalloc.rs

@@ -52,7 +52,7 @@ pub mod jemalloc;
 pub mod logger;
 pub mod metrics;
 #[cfg(target_os = "linux")]
-pub mod procfs;
+pub mod proc_io;
 pub mod service;
 pub mod source;
 pub mod split;

quickwit/quickwit-cli/src/lib.rs

@@ -24,7 +24,7 @@ use quickwit_cli::cli::{CliCommand, build_cli};
 use quickwit_cli::jemalloc::start_jemalloc_metrics_loop;
 use quickwit_cli::logger::setup_logging_and_tracing;
 #[cfg(target_os = "linux")]
-use quickwit_cli::procfs::start_proc_io_metrics_loop;
+use quickwit_cli::proc_io::start_proc_io_metrics_loop;
 use quickwit_cli::{busy_detector, install_default_crypto_ring_provider};
 use quickwit_common::runtimes::scrape_tokio_runtime_metrics;
 use quickwit_serve::BuildInfo;

quickwit/quickwit-cli/src/main.rs

@@ -17,7 +17,8 @@ use std::time::Duration;
 use procfs::ProcResult;
 use procfs::process::Process;
 use quickwit_common::metrics::{IO_METRICS, IntCounter};
-use tracing::{error, warn};
+use quickwit_common::rate_limited_tracing::rate_limited_warn;
+use tracing::error;
 
 const PROC_IO_METRICS_POLLING_INTERVAL: Duration = Duration::from_secs(5);
 
@@ -36,42 +37,51 @@ async fn proc_io_metrics_loop() -> ProcResult<()> {
     let mut previous_write_syscalls: u64 = 0;
 
     let mut poll_interval = tokio::time::interval(PROC_IO_METRICS_POLLING_INTERVAL);
+    poll_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
 
     loop {
         poll_interval.tick().await;
 
         let io = match process.io() {
             Ok(io) => io,
             Err(error) => {
-                warn!(%error, "failed to read /proc/self/io");
+                rate_limited_warn!(
+                    limit_per_min = 1,
+                    %error,
+                    "failed to read /proc/self/io"
+                );
                 continue;
             }
         };
-
-        increment_counter_by_delta(
+        increment_counter(
             &IO_METRICS.read_bytes_total,
             io.read_bytes,
             &mut previous_read_bytes,
         );
-        increment_counter_by_delta(
+        increment_counter(
             &IO_METRICS.write_bytes_total,
             io.write_bytes,
             &mut previous_write_bytes,
         );
-        increment_counter_by_delta(
+        increment_counter(
             &IO_METRICS.read_syscalls_total,
             io.syscr,
             &mut previous_read_syscalls,
         );
-        increment_counter_by_delta(
+        increment_counter(
             &IO_METRICS.write_syscalls_total,
             io.syscw,
             &mut previous_write_syscalls,
         );
     }
 }
 
-fn increment_counter_by_delta(counter: &IntCounter, current: u64, previous: &mut u64) {
+fn increment_counter(counter: &IntCounter, current: u64, previous: &mut u64) {
+    debug_assert!(
+        current >= *previous,
+        "/proc/self/io counters should be monotonic for a given PID"
+    );
+
     if current >= *previous {
         counter.inc_by(current - *previous);
     }

quickwit/quickwit-cli/src/procfs.rs quickwit/quickwit-cli/src/proc_io.rsquickwit/quickwit-cli/src/procfs.rs renamed to quickwit/quickwit-cli/src/proc_io.rs

@@ -467,15 +467,15 @@ impl Default for IoMetrics {
                 "read_bytes_total",
                 "Cumulative bytes read from storage by the process, as reported by \
                  `/proc/self/io` `read_bytes`. Reflects block-layer I/O after page cache absorbs \
-                 reads served from RAM",
+                 reads served from memory",
                 "io",
                 &[],
             ),
             write_bytes_total: new_counter(
                 "write_bytes_total",
                 "Cumulative bytes written to storage by the process, as reported by \
                  `/proc/self/io` `write_bytes`. Reflects block-layer I/O after page cache \
-                 coalescing — what EBS `VolumeWriteOps` ultimately bills",
+                 coalescing",
                 "io",
                 &[],
             ),

quickwit/quickwit-common/src/metrics.rs

@@ -189,7 +189,7 @@ fn list_split_files(
     segment_metas: &[SegmentMeta],
     scratch_directory: &TempDirectory,
 ) -> io::Result<Vec<PathBuf>> {
-    let mut index_files = vec![scratch_directory.path().join("meta.json")];
+    let mut split_files = vec![scratch_directory.path().join("meta.json")];
 
     // list the segment files
     for segment_meta in segment_metas {
@@ -199,12 +199,12 @@ fn list_split_files(
                 // If the file is missing, this is fine.
                 // segment_meta.list_files() may actually returns files that
                 // may not exist.
-                index_files.push(filepath);
+                split_files.push(filepath);
             }
         }
     }
-    index_files.sort();
-    Ok(index_files)
+    split_files.sort();
+    Ok(split_files)
 }
 
 fn build_hotcache<W: io::Write>(split_path: &Path, out: &mut W) -> anyhow::Result<()> {

quickwit/quickwit-indexing/src/actors/packager.rs

@@ -72,7 +72,7 @@ impl Default for IngestResultMetrics {
     }
 }
 
-pub(super) struct IngestV2Metrics {
+pub(crate) struct IngestV2Metrics {
     pub reset_shards_operations_total: IntCounterVec<1>,
     pub open_shards: IntGauge,
     pub closed_shards: IntGauge,
@@ -83,6 +83,7 @@ pub(super) struct IngestV2Metrics {
     pub wal_lock_hold_duration_secs: HistogramVec<2>,
     pub wal_disk_used_bytes: IntGauge,
     pub wal_memory_used_bytes: IntGauge,
+    pub wal_bytes_written_total: IntCounterVec<1>,
     pub ingest_results: IngestResultMetrics,
     pub ingest_attempts: IntCounterVec<1>,
 }
@@ -164,6 +165,14 @@ impl Default for IngestV2Metrics {
                 "ingest",
                 &[],
             ),
+            wal_bytes_written_total: new_counter_vec(
+                "wal_bytes_written_total",
+                "Total number of bytes written to the WAL by each mutating operation, including \
+                 frame headers and end-of-block padding.",
+                "ingest",
+                &[],
+                ["operation"],
+            ),
         }
     }
 }
@@ -181,5 +190,5 @@ pub(super) fn report_wal_usage(wal_usage: ResourceUsage) {
         .set(wal_usage.memory_used_bytes as i64);
 }
 
-pub(super) static INGEST_V2_METRICS: LazyLock<IngestV2Metrics> =
+pub(crate) static INGEST_V2_METRICS: LazyLock<IngestV2Metrics> =
     LazyLock::new(IngestV2Metrics::default);

quickwit/quickwit-ingest/src/ingest_v2/metrics.rs

@@ -19,7 +19,7 @@ mod fetch;
 mod helpers;
 mod idle;
 mod ingester;
-mod metrics;
+pub(crate) mod metrics;
 mod models;
 mod mrecord;
 mod mrecordlog_utils;