hubcio commented on code in PR #4488:
URL: https://github.com/apache/iggy/pull/4488#discussion_r4249012271
##########
core/journal/src/durable_storage.rs:
##########
@@ -64,6 +65,10 @@ pub struct StorageEntry {
pub trait DurableStorage {
type File: DurableFile;
+ /// Count `bytes` just allocated on disk towards the process-wide disk
+ /// usage estimate. Backends that do not write to the data volume ignore
it.
+ fn charge_allocation(&self, _bytes: u64) {}
Review Comment:
warning: `persist_offset_retained` and `persist_offset_with_storage` create
offset files without charging this counter, so usage is understated between
samples. charge new offset allocations without counting existing blocks again.
##########
core/journal/src/partition_journal/segments.rs:
##########
@@ -423,10 +425,12 @@ impl<S: DurableStorage> PartitionPrepareJournal<S> {
buffers.push(prepares[index].slice(size_of::<PrepareHeader>()..));
bodies.next();
}
+ let body_bytes = buffers.iter().map(Frozen::len).sum::<usize>() as
u64;
self.segment_writing_file(first)
.await?
.write_frozen_vectored(first.position, buffers)
.await?;
+ self.storage.charge_allocation(body_bytes);
Review Comment:
warning: writes into preallocated space charge blocks that the reservation
already counted, so `disk_used_bytes` grows without equivalent allocation.
charge only new allocation here and in `MessagesWriter`.
##########
core/server_common/src/fs_utils.rs:
##########
@@ -257,4 +278,36 @@ mod tests {
});
assert_eq!(std::fs::read(&path).unwrap(), b"messages");
}
+
+ #[cfg(target_os = "linux")]
+ #[test]
+ fn
given_a_reserved_file_when_reserving_again_should_report_nothing_newly_allocated()
{
Review Comment:
warning: this test calls `reserve` directly and never checks `DISK_USAGE`,
so it still passes if the production charge disappears. assert counter deltas
for fresh reservations, repeated reservations, and writes into reserved space.
##########
core/partitions/src/state_transfer.rs:
##########
@@ -4537,11 +4537,13 @@ pub(crate) async fn reclaim_retry_checkpoint_files(
}
async fn write_staging_file(path: &Path, payload: Vec<u8>) ->
std::io::Result<()> {
+ let len = payload.len() as u64;
let mut file = compio::fs::File::create(path)
.await
.note_descriptor_exhaustion(|| format!("creating {}",
path.display()))?;
let (result, _) = file.write_all_at(payload, 0).await.into();
result?;
+ DISK_USAGE.charge(len);
Review Comment:
warning: a partial write can allocate bytes before returning an error, but
this path skips the charge and leaves those bytes uncounted. reconcile
allocation after failed writes here and in the journal, message, and index
writers.
##########
core/server/src/http/metrics.rs:
##########
@@ -270,6 +298,23 @@ mod tests {
);
}
+ /// Shares the process-wide estimate with every other test in this binary,
+ /// so this asserts the series and its type once a sample exists.
+ #[test]
+ fn disk_used_bytes_lands_in_the_exposition_once_measured() {
Review Comment:
warning: this test checks the metric name and type but never its value, so a
collector that emits a constant passes. assert the sampled value and its change
after a charge.
##########
core/integration/tests/server/disk_usage.rs:
##########
@@ -0,0 +1,187 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! The disk usage estimate behind `disk_used_bytes` against `statvfs` of the
+//! data volume. Both read the whole volume, so these tests must run alone: a
+//! `#[serial]` excludes the harness tests under `cargo test`, and
+//! `.config/nextest.toml` gives them every thread under nextest.
+
+use std::time::{Duration, Instant};
+
+use iggy::prelude::*;
+use integration::harness::TestHarness;
+use integration::harness::disk::volume_used_bytes;
+use integration::harness::seeds::{self, names};
+use serial_test::serial;
+use tokio::time::sleep;
+
+use super::http_client::HttpClient;
+
+const KIB: u64 = 1024;
+const MIB: u64 = 1024 * KIB;
+/// Per round the server writes about 2 MiB of messages, and more than that
+/// once the WAL and the segment both hold them. An estimate that missed the
+/// write sites would be off by at least that much until the next sample.
+const MESSAGES_PER_ROUND: usize = 16;
+const MESSAGE_BYTES: usize = 128 * KIB as usize;
+const ROUNDS: usize = 8;
+/// Block rounding, file metadata and the server's own log lines, none of
+/// which the write sites charge.
+const TOLERANCE: u64 = MIB;
+const FOREIGN_FILE_BYTES: usize = 16 * MIB as usize;
+/// Comfortably more than one sampling interval (1 s).
+const SAMPLE_DEADLINE: Duration = Duration::from_secs(5);
+const POLL_INTERVAL: Duration = Duration::from_millis(100);
+
+#[tokio::test]
+#[serial]
+async fn
given_iggy_writes_between_samples_when_scraping_should_track_the_volume_within_tolerance()
+{
+ let harness = start_server().await;
+ let client = harness.tcp_root_client().await.expect("tcp root client");
+ let http = HttpClient::login_root(&harness).await;
+ let data_path = harness.server().data_path();
+ let stream = Identifier::named(names::STREAM).expect("stream identifier");
+ let topic = Identifier::named(names::TOPIC).expect("topic identifier");
+
+ let mut rounds = Vec::with_capacity(ROUNDS);
+ for round in 0..ROUNDS {
+ let mut messages: Vec<IggyMessage> = (0..MESSAGES_PER_ROUND)
+ .map(|_| {
+ IggyMessage::builder()
+ .payload(vec![round as u8; MESSAGE_BYTES].into())
+ .build()
+ .expect("build message")
+ })
+ .collect();
+ client
+ .send_messages(
+ &stream,
+ &topic,
+ &Partitioning::partition_id(0),
+ &mut messages,
+ )
+ .await
+ .expect("send messages");
+
+ // Bracketing the scrape absorbs writes the server finishes while it
+ // runs, such as a segment flush that follows the WAL write.
+ let before = volume_used_bytes(&data_path);
+ let estimate = scrape_estimate(&http).await;
+ let after = volume_used_bytes(&data_path);
+ rounds.push((before, estimate, after));
+ }
+
+ let outside: Vec<_> = rounds
+ .iter()
+ .enumerate()
+ .filter(|(_, (before, estimate, after))| {
+ *estimate + TOLERANCE < *before || *estimate > *after + TOLERANCE
+ })
+ .collect();
+ assert!(
+ outside.is_empty(),
+ "the estimate must stay within {TOLERANCE} bytes of statvfs, rounds
outside \
+ (round, (statvfs before, estimate, statvfs after)): {outside:?}"
+ );
+}
+
+#[tokio::test]
+#[serial]
+async fn
given_a_file_written_outside_iggy_when_the_next_sample_runs_should_count_it_until_deleted()
+{
+ let harness = start_server().await;
+ let http = HttpClient::login_root(&harness).await;
+ let data_path = harness.server().data_path();
+ // Beside the data directory, so on the same volume but never charged.
+ let foreign_path = data_path.with_extension("foreign");
+ let idle = scrape_estimate(&http).await;
+
+ std::fs::write(&foreign_path, vec![0u8; FOREIGN_FILE_BYTES]).expect("write
foreign file");
+ let foreign_bytes = FOREIGN_FILE_BYTES as u64;
+ let with_file = wait_for_estimate(&http, |estimate| {
+ estimate + TOLERANCE >= idle + foreign_bytes
Review Comment:
warning: these one-sided bounds accept exaggerated counter changes and never
bound the error against `statvfs`. compare counter and filesystem deltas for
both file creation and removal.
##########
core/server_common/src/disk_usage.rs:
##########
@@ -0,0 +1,176 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Estimated usage of the volume holding the data directory.
+//!
+//! A periodic `statvfs` sample sets the base and the sites that allocate disk
+//! space add what they write, so the estimate follows a burst between two
+//! samples without a syscall on the write path. Each sample re-bases the
+//! count, so anything the write sites cannot see (hard links, temporary files
+//! replaced by rename, logs, other processes, deleted segments) is wrong for
+//! at most one sampling interval.
+
+use crossbeam::utils::CachePadded;
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+
+/// The process-wide estimate. It counts from zero until the first sample.
+pub static DISK_USAGE: DiskUsage = DiskUsage::new();
+
+#[derive(Debug)]
+pub struct DiskUsage {
+ // Every shard adds to it, so it gets a cache line of its own.
+ used_bytes: CachePadded<AtomicU64>,
+ measured: AtomicBool,
+}
+
+impl DiskUsage {
+ #[must_use]
+ pub const fn new() -> Self {
+ Self {
+ used_bytes: CachePadded::new(AtomicU64::new(0)),
+ measured: AtomicBool::new(false),
+ }
+ }
+
+ /// Adds `bytes` that Iggy has just written to, or reserved on, the volume.
+ #[inline]
+ pub fn charge(&self, bytes: u64) {
+ self.used_bytes.fetch_add(bytes, Ordering::Relaxed);
+ }
+
+ /// Estimated bytes used on the volume: the last sample plus the bytes
+ /// charged since.
+ #[must_use]
+ #[inline]
+ pub fn used_bytes(&self) -> u64 {
+ self.used_bytes.load(Ordering::Relaxed)
+ }
+
+ /// Whether a sample has set the base. Until then [`Self::used_bytes`]
+ /// holds only the bytes charged since the process started.
+ #[must_use]
+ pub fn is_measured(&self) -> bool {
+ self.measured.load(Ordering::Acquire)
+ }
+
+ /// Runs `measure` and re-bases the estimate on the used bytes it returns,
+ /// passing its sample through. Call it from one thread only.
+ ///
+ /// The correction is added instead of stored, so a charge that lands after
+ /// `measure` returns is kept. A charge that lands while `measure` runs is
+ /// counted twice until the next sample, which errs towards a fuller disk.
+ /// When `measure` returns `None` the estimate keeps counting from the
+ /// previous base.
+ pub fn rebase<T>(&self, measure: impl FnOnce() -> Option<(T, u64)>) ->
Option<T> {
+ let counted = self.used_bytes.load(Ordering::Relaxed);
+ let (sample, used_bytes) = measure()?;
+ self.used_bytes
+ .fetch_add(used_bytes.wrapping_sub(counted), Ordering::Relaxed);
+ self.measured.store(true, Ordering::Release);
+ Some(sample)
+ }
+}
+
+impl Default for DiskUsage {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ const MIB: u64 = 1024 * 1024;
+
+ #[test]
+ fn given_no_sample_when_charging_should_count_from_zero() {
+ let usage = DiskUsage::new();
+
+ usage.charge(3 * MIB);
+
+ assert_eq!(usage.used_bytes(), 3 * MIB);
+ assert!(!usage.is_measured());
+ }
+
+ #[test]
+ fn given_charges_when_rebasing_should_replace_them_with_the_sample() {
+ let usage = DiskUsage::new();
+ usage.charge(5 * MIB);
+
+ let sample = usage.rebase(|| Some(("sample", 100 * MIB)));
+
+ assert_eq!(sample, Some("sample"));
+ assert_eq!(usage.used_bytes(), 100 * MIB);
+ assert!(usage.is_measured());
+ }
+
+ #[test]
+ fn given_a_sample_below_the_estimate_when_rebasing_should_lower_it() {
+ let usage = DiskUsage::new();
+ usage.rebase(|| Some(((), 100 * MIB)));
+ usage.charge(40 * MIB);
+
+ usage.rebase(|| Some(((), 90 * MIB)));
+
+ assert_eq!(
+ usage.used_bytes(),
+ 90 * MIB,
+ "space freed since the last sample must lower the estimate"
+ );
+ }
+
+ #[test]
+ fn given_a_charge_after_the_sample_when_rebasing_should_keep_it() {
Review Comment:
warning: this test charges only after `rebase` returns, so it misses updates
between measurement and correction. use barriers to place a concurrent charge
in that interval.
##########
core/journal/src/partition_journal/segments.rs:
##########
@@ -423,10 +425,12 @@ impl<S: DurableStorage> PartitionPrepareJournal<S> {
buffers.push(prepares[index].slice(size_of::<PrepareHeader>()..));
bodies.next();
}
+ let body_bytes = buffers.iter().map(Frozen::len).sum::<usize>() as
u64;
Review Comment:
simplification: this loop sums buffer lengths even though `end` already
contains their combined extent. use `end - first.position` after the successful
write.
##########
core/integration/tests/server/disk_usage.rs:
##########
@@ -0,0 +1,187 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! The disk usage estimate behind `disk_used_bytes` against `statvfs` of the
+//! data volume. Both read the whole volume, so these tests must run alone: a
+//! `#[serial]` excludes the harness tests under `cargo test`, and
+//! `.config/nextest.toml` gives them every thread under nextest.
+
+use std::time::{Duration, Instant};
+
+use iggy::prelude::*;
+use integration::harness::TestHarness;
+use integration::harness::disk::volume_used_bytes;
+use integration::harness::seeds::{self, names};
+use serial_test::serial;
+use tokio::time::sleep;
+
+use super::http_client::HttpClient;
+
+const KIB: u64 = 1024;
+const MIB: u64 = 1024 * KIB;
+/// Per round the server writes about 2 MiB of messages, and more than that
+/// once the WAL and the segment both hold them. An estimate that missed the
+/// write sites would be off by at least that much until the next sample.
+const MESSAGES_PER_ROUND: usize = 16;
+const MESSAGE_BYTES: usize = 128 * KIB as usize;
+const ROUNDS: usize = 8;
+/// Block rounding, file metadata and the server's own log lines, none of
+/// which the write sites charge.
+const TOLERANCE: u64 = MIB;
+const FOREIGN_FILE_BYTES: usize = 16 * MIB as usize;
+/// Comfortably more than one sampling interval (1 s).
+const SAMPLE_DEADLINE: Duration = Duration::from_secs(5);
+const POLL_INTERVAL: Duration = Duration::from_millis(100);
+
+#[tokio::test]
+#[serial]
+async fn
given_iggy_writes_between_samples_when_scraping_should_track_the_volume_within_tolerance()
+{
+ let harness = start_server().await;
+ let client = harness.tcp_root_client().await.expect("tcp root client");
+ let http = HttpClient::login_root(&harness).await;
+ let data_path = harness.server().data_path();
+ let stream = Identifier::named(names::STREAM).expect("stream identifier");
+ let topic = Identifier::named(names::TOPIC).expect("topic identifier");
+
+ let mut rounds = Vec::with_capacity(ROUNDS);
+ for round in 0..ROUNDS {
+ let mut messages: Vec<IggyMessage> = (0..MESSAGES_PER_ROUND)
+ .map(|_| {
+ IggyMessage::builder()
+ .payload(vec![round as u8; MESSAGE_BYTES].into())
+ .build()
+ .expect("build message")
+ })
+ .collect();
+ client
+ .send_messages(
+ &stream,
+ &topic,
+ &Partitioning::partition_id(0),
+ &mut messages,
+ )
+ .await
+ .expect("send messages");
+
+ // Bracketing the scrape absorbs writes the server finishes while it
+ // runs, such as a segment flush that follows the WAL write.
+ let before = volume_used_bytes(&data_path);
+ let estimate = scrape_estimate(&http).await;
+ let after = volume_used_bytes(&data_path);
+ rounds.push((before, estimate, after));
+ }
+
+ let outside: Vec<_> = rounds
+ .iter()
+ .enumerate()
+ .filter(|(_, (before, estimate, after))| {
+ *estimate + TOLERANCE < *before || *estimate > *after + TOLERANCE
Review Comment:
warning: the sampler can refresh before each assertion, so this test can
pass with missing write charges. freeze the sampler and compare counter and
filesystem deltas across the same interval.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
