gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-07-26 08:18:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(listobjects): add guarded metadata-fast listing (#4311)

Browse Source 
* feat(listobjects): add phase 0 observability metrics

* perf(listobjects): reduce gather metadata decoding

* test(listobjects): add adversarial listing coverage

* feat(listobjects): add source-aware cursor fields

* feat(listobjects): define index source modes

* feat(listobjects): model index rebuild health

* feat(listobjects): add opt-in index fallback gate

* feat(listobjects): verify index candidate pages

* docs(listobjects): add rollout runbook

* docs(listobjects): untrack baseline fixtures

* feat(listobjects): add opt-in key-only provider

* feat(listobjects): add index verification metrics

* perf(listobjects): gate list metrics on get stage switch

* feat(listobjects): bind provider health generation

* feat(listobjects): add persistent key-only provider

* feat(listobjects): bind persistent provider lifecycle

* feat(listobjects): track persistent index mutations

* feat(listobjects): persist index mutation checkpoints

* fix(listobjects): aggregate benchmark metric snapshots

* feat(listobjects): store journal in system namespace

* chore(listobjects): report fallback reasons in bench

* test(listobjects): verify metadata-fast stale fallbacks

* feat(listobjects): serve metadata-fast snapshots behind guardrails

* test(listobjects): add metadata-fast chaos bench

* fix(listobjects): attribute metadata-fast fallback metrics

* test(listobjects): add metadata-fast fallback chaos probes

* test(listobjects): add quorum journal chaos guardrails

* fix(listobjects): satisfy pre-pr clippy gate

* docs(listobjects): untrack rollout runbook

* fix(ecstore): remove redundant heal error conversion

* test(filemeta): satisfy replication info clippy

----

Co-Authored-By: heihutu <heihutu@gmail.com>
This commit is contained in:
houseme
2026-07-07 01:38:52 +08:00
committed by GitHub
parent 0271abc14d
commit 25193ee6cf
9 changed files with 6033 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/store/list_objects.rs
+3992 -55
View File
File diff suppressed because it is too large Load Diff
crates/ecstore/src/store/mod.rs
+37 -9
View File
@@ -352,7 +352,13 @@ impl crate::storage_api_contracts::object::ObjectIO for ECStore {
}
#[instrument(level = "debug", skip(self, data))]
async fn put_object(&self, bucket: &str, object: &str, data: &mut PutObjReader, opts: &ObjectOptions) -> Result<ObjectInfo> {
enqueue_transition_after_write(self.handle_put_object(bucket, object, data, opts).await, LcEventSrc::S3PutObject).await
let result =
enqueue_transition_after_write(self.handle_put_object(bucket, object, data, opts).await, LcEventSrc::S3PutObject)
.await;
if result.is_ok() {
list_objects::observe_list_objects_mutation(self, bucket).await;
}
result
}
}
@@ -418,22 +424,34 @@ impl crate::storage_api_contracts::object::ObjectOperations for ECStore {
src_opts: &ObjectOptions,
dst_opts: &ObjectOptions,
) -> Result<ObjectInfo> {
enqueue_transition_after_write(
let result = enqueue_transition_after_write(
self.handle_copy_object(src_bucket, src_object, dst_bucket, dst_object, src_info, src_opts, dst_opts)
.await,
LcEventSrc::S3CopyObject,
)
.await
.await;
if result.is_ok() {
list_objects::observe_list_objects_mutation(self, dst_bucket).await;
}
result
}
#[instrument(skip(self))]
async fn delete_object_version(&self, bucket: &str, object: &str, fi: &FileInfo, force_del_marker: bool) -> Result<()> {
self.handle_delete_object_version(bucket, object, fi, force_del_marker).await
let result = self.handle_delete_object_version(bucket, object, fi, force_del_marker).await;
if result.is_ok() {
list_objects::observe_list_objects_mutation(self, bucket).await;
}
result
}
#[instrument(skip(self))]
async fn delete_object(&self, bucket: &str, object: &str, opts: ObjectOptions) -> Result<ObjectInfo> {
self.handle_delete_object(bucket, object, opts).await
let result = self.handle_delete_object(bucket, object, opts).await;
if result.is_ok() {
list_objects::observe_list_objects_mutation(self, bucket).await;
}
result
}
#[instrument(skip(self))]
@@ -443,7 +461,12 @@ impl crate::storage_api_contracts::object::ObjectOperations for ECStore {
objects: Vec<ObjectToDelete>,
opts: ObjectOptions,
) -> (Vec<DeletedObject>, Vec<Option<Error>>) {
self.handle_delete_objects(bucket, objects, opts).await
let result = self.handle_delete_objects(bucket, objects, opts).await;
let success_count = result.1.iter().filter(|err| err.is_none()).count();
if success_count > 0 {
list_objects::observe_list_objects_mutations(self, bucket, success_count).await;
}
result
}
#[instrument(skip(self))]
@@ -660,12 +683,17 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for ECStore {
uploaded_parts: Vec<CompletePart>,
opts: &ObjectOptions,
) -> Result<ObjectInfo> {
enqueue_transition_after_write(
self.handle_complete_multipart_upload(bucket, object, upload_id, uploaded_parts, opts)
let result = enqueue_transition_after_write(
self.clone()
.handle_complete_multipart_upload(bucket, object, upload_id, uploaded_parts, opts)
.await,
LcEventSrc::S3CompleteMultipartUpload,
)
.await
.await;
if result.is_ok() {
list_objects::observe_list_objects_mutation(self.as_ref(), bucket).await;
}
result
}
}
crates/filemeta/src/fileinfo.rs
+14 -8
View File
@@ -896,22 +896,28 @@ mod tests {
assert!(base.replication_info_equals(&FileInfo::default()));
// Differing mark_deleted breaks equality.
let mut marked = FileInfo::default();
marked.mark_deleted = true;
let marked = FileInfo {
mark_deleted: true,
..Default::default()
};
assert!(!base.replication_info_equals(&marked));
// Differing replication_state_internal breaks equality (regression guard:
// this field used to be ignored by replication_info_equals).
let mut with_state = FileInfo::default();
with_state.replication_state_internal = Some(ReplicationState {
replicate_decision_str: "arn:aws:s3:::dest".to_string(),
let with_state = FileInfo {
replication_state_internal: Some(ReplicationState {
replicate_decision_str: "arn:aws:s3:::dest".to_string(),
..Default::default()
}),
..Default::default()
});
};
assert!(!base.replication_info_equals(&with_state));
// Equal replication states remain equal.
let mut with_state_clone = FileInfo::default();
with_state_clone.replication_state_internal = with_state.replication_state_internal.clone();
let with_state_clone = FileInfo {
replication_state_internal: with_state.replication_state_internal.clone(),
..Default::default()
};
assert!(with_state.replication_info_equals(&with_state_clone));
}
}
crates/io-metrics/src/lib.rs
+47
View File
@@ -95,6 +95,7 @@ pub mod config;
pub mod deadlock_metrics;
pub mod internode_metrics;
pub mod io_metrics;
pub mod list_objects_metrics;
pub mod lock_metrics;
pub mod performance;
pub mod process_lock_metrics;
@@ -129,6 +130,12 @@ pub use io_metrics::{
IoSchedulerStats, record_bandwidth_observation, record_buffer_size_adjustment, record_io_priority_decision,
record_io_scheduler_decision, record_load_level_change, record_queue_operation, record_starvation_event,
};
pub use list_objects_metrics::{
LIST_OBJECTS_GATHER_OUTCOME_INPUT_CLOSED, LIST_OBJECTS_GATHER_OUTCOME_LIMIT_REACHED, LIST_OBJECTS_SOURCE_WALKER,
ListObjectsGatherObservation, ListObjectsIndexPageObservation, init_list_objects_metrics, record_list_objects_gather,
record_list_objects_index_attempt, record_list_objects_index_fallback, record_list_objects_index_live_verify_failure,
record_list_objects_index_served, record_list_objects_merge,
};
// Backpressure metrics exports
pub use backpressure_metrics::{
@@ -1941,6 +1948,46 @@ pub fn record_cgroup_memory_split(
}
}
/// Allocator memory stats captured from the active process allocator.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct AllocatorMemoryObservation {
pub reserved_bytes: Option<u64>,
pub committed_bytes: Option<u64>,
pub page_committed_bytes: Option<u64>,
pub malloc_requested_bytes: Option<u64>,
pub malloc_requested_peak_bytes: Option<u64>,
pub malloc_requested_total_bytes: Option<u64>,
pub heap_count: Option<u64>,
}
/// Record allocator-level memory attribution when supported by the allocator.
#[inline(always)]
pub fn record_allocator_memory_observation(backend: &'static str, observation: AllocatorMemoryObservation) {
if let Some(reserved_bytes) = observation.reserved_bytes {
gauge!("rustfs_memory_allocator_reserved_bytes", "backend" => backend).set(reserved_bytes as f64);
}
if let Some(committed_bytes) = observation.committed_bytes {
gauge!("rustfs_memory_allocator_committed_bytes", "backend" => backend).set(committed_bytes as f64);
}
if let Some(page_committed_bytes) = observation.page_committed_bytes {
gauge!("rustfs_memory_allocator_page_committed_bytes", "backend" => backend).set(page_committed_bytes as f64);
}
if let Some(malloc_requested_bytes) = observation.malloc_requested_bytes {
gauge!("rustfs_memory_allocator_malloc_requested_bytes", "backend" => backend).set(malloc_requested_bytes as f64);
}
if let Some(malloc_requested_peak_bytes) = observation.malloc_requested_peak_bytes {
gauge!("rustfs_memory_allocator_malloc_requested_peak_bytes", "backend" => backend)
.set(malloc_requested_peak_bytes as f64);
}
if let Some(malloc_requested_total_bytes) = observation.malloc_requested_total_bytes {
gauge!("rustfs_memory_allocator_malloc_requested_total_bytes", "backend" => backend)
.set(malloc_requested_total_bytes as f64);
}
if let Some(heap_count) = observation.heap_count {
gauge!("rustfs_memory_allocator_heap_count", "backend" => backend).set(heap_count as f64);
}
}
/// Track encoded bytes currently queued between erasure encode and disk writers.
#[inline(always)]
pub fn add_ec_encode_inflight_bytes(bytes: usize) {
crates/io-metrics/src/list_objects_metrics.rs
+376
View File
@@ -0,0 +1,376 @@
// Copyright 2024 RustFS Team
//
// Licensed 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.
use metrics::{counter, describe_counter, describe_histogram, histogram};
use std::sync::OnceLock;
use crate::get_stage_metrics_enabled;
pub const LIST_OBJECTS_SOURCE_WALKER: &str = "walker";
pub const LIST_OBJECTS_GATHER_OUTCOME_LIMIT_REACHED: &str = "limit_reached";
pub const LIST_OBJECTS_GATHER_OUTCOME_INPUT_CLOSED: &str = "input_closed";
pub const LIST_OBJECTS_MERGE_OUTCOME_STARTED: &str = "started";
const LIST_OBJECTS_GATHER_TOTAL: &str = "rustfs_s3_list_objects_gather_total";
const LIST_OBJECTS_GATHER_DURATION_MS: &str = "rustfs_s3_list_objects_gather_duration_ms";
const LIST_OBJECTS_GATHER_SCANNED_ENTRIES: &str = "rustfs_s3_list_objects_gather_scanned_entries";
const LIST_OBJECTS_GATHER_RETURNED_ENTRIES: &str = "rustfs_s3_list_objects_gather_returned_entries";
const LIST_OBJECTS_GATHER_FILTERED_ENTRIES: &str = "rustfs_s3_list_objects_gather_filtered_entries";
const LIST_OBJECTS_GATHER_SCAN_AMPLIFICATION: &str = "rustfs_s3_list_objects_gather_scan_amplification";
const LIST_OBJECTS_GATHER_LIMIT: &str = "rustfs_s3_list_objects_gather_limit";
const LIST_OBJECTS_MERGE_FAN_IN: &str = "rustfs_s3_list_objects_merge_fan_in";
const LIST_OBJECTS_MERGE_READ_QUORUM: &str = "rustfs_s3_list_objects_merge_read_quorum";
const LIST_OBJECTS_INDEX_ATTEMPT_TOTAL: &str = "rustfs_s3_list_objects_index_attempt_total";
const LIST_OBJECTS_INDEX_FALLBACK_TOTAL: &str = "rustfs_s3_list_objects_index_fallback_total";
const LIST_OBJECTS_INDEX_SERVED_TOTAL: &str = "rustfs_s3_list_objects_index_served_total";
const LIST_OBJECTS_INDEX_CANDIDATE_KEYS: &str = "rustfs_s3_list_objects_index_candidate_keys";
const LIST_OBJECTS_INDEX_LIVE_VERIFY_ATTEMPTS: &str = "rustfs_s3_list_objects_index_live_verify_attempts";
const LIST_OBJECTS_INDEX_LIVE_VERIFY_HITS: &str = "rustfs_s3_list_objects_index_live_verify_hits";
const LIST_OBJECTS_INDEX_LIVE_VERIFY_MISSES: &str = "rustfs_s3_list_objects_index_live_verify_misses";
const LIST_OBJECTS_INDEX_LIVE_VERIFY_FAILURE_TOTAL: &str = "rustfs_s3_list_objects_index_live_verify_failure_total";
const LIST_OBJECTS_INDEX_RETURNED_OBJECTS: &str = "rustfs_s3_list_objects_index_returned_objects";
const LIST_OBJECTS_INDEX_RETURNED_PREFIXES: &str = "rustfs_s3_list_objects_index_returned_prefixes";
const LIST_OBJECTS_INDEX_VERIFICATION_IO_AMPLIFICATION: &str = "rustfs_s3_list_objects_index_verification_io_amplification";
#[derive(Clone, Copy, Debug, PartialEq)]
pub struct ListObjectsGatherObservation {
pub source: &'static str,
pub outcome: &'static str,
pub limit: i32,
pub scanned_entries: usize,
pub returned_entries: usize,
pub duration_ms: f64,
pub has_prefix: bool,
pub has_delimiter: bool,
pub has_marker: bool,
}
#[derive(Clone, Copy, Debug, PartialEq)]
pub struct ListObjectsIndexPageObservation {
pub source: &'static str,
pub provider: &'static str,
pub candidate_keys: usize,
pub live_verify_attempts: usize,
pub live_verify_hits: usize,
pub live_verify_misses: usize,
pub returned_objects: usize,
pub returned_prefixes: usize,
pub is_truncated: bool,
}
#[inline(always)]
fn bool_label(value: bool) -> &'static str {
if value { "true" } else { "false" }
}
#[inline(always)]
fn count_as_f64(value: usize) -> f64 {
value as f64
}
#[inline(always)]
fn limit_as_f64(value: i32) -> f64 {
value.max(0) as f64
}
pub fn init_list_objects_metrics() {
static METRICS_DESC_INIT: OnceLock<()> = OnceLock::new();
METRICS_DESC_INIT.get_or_init(|| {
describe_counter!(
LIST_OBJECTS_GATHER_TOTAL,
"Total number of ListObjects gather phases by source, outcome, and request shape."
);
describe_histogram!(LIST_OBJECTS_GATHER_DURATION_MS, "ListObjects gather phase duration in milliseconds.");
describe_histogram!(
LIST_OBJECTS_GATHER_SCANNED_ENTRIES,
"Number of entries consumed by ListObjects gather before producing a page."
);
describe_histogram!(
LIST_OBJECTS_GATHER_RETURNED_ENTRIES,
"Number of entries returned by ListObjects gather before pagination trimming."
);
describe_histogram!(
LIST_OBJECTS_GATHER_FILTERED_ENTRIES,
"Number of entries filtered by ListObjects gather before producing a page."
);
describe_histogram!(
LIST_OBJECTS_GATHER_SCAN_AMPLIFICATION,
"Ratio of scanned entries to returned entries in ListObjects gather."
);
describe_histogram!(LIST_OBJECTS_GATHER_LIMIT, "Requested internal ListObjects gather page limit.");
describe_histogram!(LIST_OBJECTS_MERGE_FAN_IN, "Number of input streams merged for ListObjects pages.");
describe_histogram!(LIST_OBJECTS_MERGE_READ_QUORUM, "Read quorum used while merging ListObjects entries.");
describe_counter!(
LIST_OBJECTS_INDEX_ATTEMPT_TOTAL,
"Total number of opt-in ListObjects index serving attempts by source, provider, and request shape."
);
describe_counter!(
LIST_OBJECTS_INDEX_FALLBACK_TOTAL,
"Total number of opt-in ListObjects index attempts that fell back to the live walker."
);
describe_counter!(
LIST_OBJECTS_INDEX_SERVED_TOTAL,
"Total number of ListObjects pages served by an opt-in index provider."
);
describe_histogram!(
LIST_OBJECTS_INDEX_CANDIDATE_KEYS,
"Number of candidate keys proposed by the opt-in ListObjects index provider per page."
);
describe_histogram!(
LIST_OBJECTS_INDEX_LIVE_VERIFY_ATTEMPTS,
"Number of live xl.meta verification attempts made by opt-in ListObjects index serving per page."
);
describe_histogram!(
LIST_OBJECTS_INDEX_LIVE_VERIFY_HITS,
"Number of opt-in ListObjects index candidates accepted by live xl.meta verification per page."
);
describe_histogram!(
LIST_OBJECTS_INDEX_LIVE_VERIFY_MISSES,
"Number of opt-in ListObjects index candidates rejected as stale or missing by live xl.meta verification per page."
);
describe_counter!(
LIST_OBJECTS_INDEX_LIVE_VERIFY_FAILURE_TOTAL,
"Total number of opt-in ListObjects index live xl.meta verification failures."
);
describe_histogram!(
LIST_OBJECTS_INDEX_RETURNED_OBJECTS,
"Number of objects returned by opt-in ListObjects index serving per page."
);
describe_histogram!(
LIST_OBJECTS_INDEX_RETURNED_PREFIXES,
"Number of common prefixes returned by opt-in ListObjects index serving per page."
);
describe_histogram!(
LIST_OBJECTS_INDEX_VERIFICATION_IO_AMPLIFICATION,
"Ratio of live verification attempts to returned objects for opt-in ListObjects index serving."
);
});
}
pub fn record_list_objects_gather(observation: ListObjectsGatherObservation) {
if !get_stage_metrics_enabled() {
return;
}
let filtered_entries = observation.scanned_entries.saturating_sub(observation.returned_entries);
let scan_amplification = if observation.returned_entries == 0 {
count_as_f64(observation.scanned_entries)
} else {
count_as_f64(observation.scanned_entries) / count_as_f64(observation.returned_entries)
};
counter!(
LIST_OBJECTS_GATHER_TOTAL,
"source" => observation.source,
"outcome" => observation.outcome,
"has_prefix" => bool_label(observation.has_prefix),
"has_delimiter" => bool_label(observation.has_delimiter),
"has_marker" => bool_label(observation.has_marker),
)
.increment(1);
histogram!(
LIST_OBJECTS_GATHER_DURATION_MS,
"source" => observation.source,
"outcome" => observation.outcome
)
.record(observation.duration_ms);
histogram!(LIST_OBJECTS_GATHER_SCANNED_ENTRIES, "source" => observation.source)
.record(count_as_f64(observation.scanned_entries));
histogram!(LIST_OBJECTS_GATHER_RETURNED_ENTRIES, "source" => observation.source)
.record(count_as_f64(observation.returned_entries));
histogram!(LIST_OBJECTS_GATHER_FILTERED_ENTRIES, "source" => observation.source).record(count_as_f64(filtered_entries));
histogram!(LIST_OBJECTS_GATHER_SCAN_AMPLIFICATION, "source" => observation.source).record(scan_amplification);
histogram!(LIST_OBJECTS_GATHER_LIMIT, "source" => observation.source).record(limit_as_f64(observation.limit));
}
pub fn record_list_objects_merge(source: &'static str, input_channels: usize, read_quorum: usize) {
if !get_stage_metrics_enabled() {
return;
}
histogram!(
LIST_OBJECTS_MERGE_FAN_IN,
"source" => source,
"outcome" => LIST_OBJECTS_MERGE_OUTCOME_STARTED
)
.record(count_as_f64(input_channels));
histogram!(
LIST_OBJECTS_MERGE_READ_QUORUM,
"source" => source,
"outcome" => LIST_OBJECTS_MERGE_OUTCOME_STARTED
)
.record(count_as_f64(read_quorum));
}
pub fn record_list_objects_index_fallback(source: &'static str, reason: &'static str) {
if !get_stage_metrics_enabled() {
return;
}
counter!(
LIST_OBJECTS_INDEX_FALLBACK_TOTAL,
"source" => source,
"reason" => reason
)
.increment(1);
}
pub fn record_list_objects_index_attempt(
source: &'static str,
provider: &'static str,
has_prefix: bool,
has_delimiter: bool,
has_marker: bool,
) {
if !get_stage_metrics_enabled() {
return;
}
counter!(
LIST_OBJECTS_INDEX_ATTEMPT_TOTAL,
"source" => source,
"provider" => provider,
"has_prefix" => bool_label(has_prefix),
"has_delimiter" => bool_label(has_delimiter),
"has_marker" => bool_label(has_marker),
)
.increment(1);
}
pub fn record_list_objects_index_live_verify_failure(source: &'static str, reason: &'static str) {
if !get_stage_metrics_enabled() {
return;
}
counter!(
LIST_OBJECTS_INDEX_LIVE_VERIFY_FAILURE_TOTAL,
"source" => source,
"reason" => reason,
)
.increment(1);
}
pub fn record_list_objects_index_served(observation: ListObjectsIndexPageObservation) {
if !get_stage_metrics_enabled() {
return;
}
let returned_objects = count_as_f64(observation.returned_objects);
let verification_io_amplification = if observation.returned_objects == 0 {
count_as_f64(observation.live_verify_attempts)
} else {
count_as_f64(observation.live_verify_attempts) / returned_objects
};
counter!(
LIST_OBJECTS_INDEX_SERVED_TOTAL,
"source" => observation.source,
"provider" => observation.provider,
"is_truncated" => bool_label(observation.is_truncated),
)
.increment(1);
histogram!(
LIST_OBJECTS_INDEX_CANDIDATE_KEYS,
"source" => observation.source,
"provider" => observation.provider,
)
.record(count_as_f64(observation.candidate_keys));
histogram!(
LIST_OBJECTS_INDEX_LIVE_VERIFY_ATTEMPTS,
"source" => observation.source,
"provider" => observation.provider,
)
.record(count_as_f64(observation.live_verify_attempts));
histogram!(
LIST_OBJECTS_INDEX_LIVE_VERIFY_HITS,
"source" => observation.source,
"provider" => observation.provider,
)
.record(count_as_f64(observation.live_verify_hits));
histogram!(
LIST_OBJECTS_INDEX_LIVE_VERIFY_MISSES,
"source" => observation.source,
"provider" => observation.provider,
)
.record(count_as_f64(observation.live_verify_misses));
histogram!(
LIST_OBJECTS_INDEX_RETURNED_OBJECTS,
"source" => observation.source,
"provider" => observation.provider,
)
.record(returned_objects);
histogram!(
LIST_OBJECTS_INDEX_RETURNED_PREFIXES,
"source" => observation.source,
"provider" => observation.provider,
)
.record(count_as_f64(observation.returned_prefixes));
histogram!(
LIST_OBJECTS_INDEX_VERIFICATION_IO_AMPLIFICATION,
"source" => observation.source,
"provider" => observation.provider,
)
.record(verification_io_amplification);
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn record_gather_observation_handles_empty_page() {
init_list_objects_metrics();
record_list_objects_gather(ListObjectsGatherObservation {
source: LIST_OBJECTS_SOURCE_WALKER,
outcome: LIST_OBJECTS_GATHER_OUTCOME_INPUT_CLOSED,
limit: 1001,
scanned_entries: 42,
returned_entries: 0,
duration_ms: 3.5,
has_prefix: true,
has_delimiter: false,
has_marker: true,
});
}
#[test]
fn record_merge_observation_accepts_zero_quorum() {
init_list_objects_metrics();
record_list_objects_merge(LIST_OBJECTS_SOURCE_WALKER, 4, 0);
}
#[test]
fn record_index_fallback_observation_accepts_reason() {
init_list_objects_metrics();
record_list_objects_index_fallback("index_key_only", "unsupported_request");
}
#[test]
fn record_index_attempt_and_served_observations_accept_counts() {
init_list_objects_metrics();
record_list_objects_index_attempt("index_key_only", "walker_key_only", true, true, false);
record_list_objects_index_served(ListObjectsIndexPageObservation {
source: "index_key_only",
provider: "walker_key_only",
candidate_keys: 1000,
live_verify_attempts: 700,
live_verify_hits: 650,
live_verify_misses: 50,
returned_objects: 600,
returned_prefixes: 10,
is_truncated: true,
});
record_list_objects_index_live_verify_failure("index_key_only", "read_error");
}
}
rustfs/src/memory_observability.rs
+140 -5
View File
@@ -13,11 +13,14 @@
// limitations under the License.
use rustfs_io_metrics::{
record_cgroup_memory_split, record_cpu_usage, record_memory_usage, record_process_memory_split,
snapshot_process_resource_and_system,
AllocatorMemoryObservation, record_allocator_memory_observation, record_cgroup_memory_split, record_cpu_usage,
record_memory_usage, record_process_memory_split, snapshot_process_resource_and_system,
};
use serde::Serialize;
use serde_json::Value;
use std::collections::HashMap;
#[cfg(not(target_os = "windows"))]
use std::ffi::CStr;
use std::path::Path;
use std::sync::{Mutex, OnceLock};
use std::time::Duration;
@@ -127,6 +130,12 @@ struct CgroupMemorySnapshot {
inactive_file_bytes: Option<u64>,
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
struct AllocatorMemorySnapshot {
backend: &'static str,
observation: AllocatorMemoryObservation,
}
fn memory_system() -> &'static Mutex<System> {
MEMORY_SYSTEM.get_or_init(|| Mutex::new(System::new()))
}
@@ -204,6 +213,94 @@ fn read_cgroup_memory_snapshot() -> Option<CgroupMemorySnapshot> {
read_cgroup_v2().or_else(read_cgroup_v1)
}
fn numeric_json_value(value: &Value) -> Option<u64> {
match value {
Value::Number(number) => number
.as_u64()
.or_else(|| number.as_i64().and_then(|value| u64::try_from(value).ok())),
Value::String(value) => value.parse::<u64>().ok(),
_ => None,
}
}
fn numeric_json_field(value: &Value, field: &str) -> Option<u64> {
match value {
Value::Object(fields) => fields
.get(field)
.and_then(numeric_json_value)
.or_else(|| fields.values().find_map(|value| numeric_json_field(value, field))),
Value::Array(values) => values.iter().find_map(|value| numeric_json_field(value, field)),
_ => None,
}
}
fn mimalloc_stat_field(value: &Value, metric: &str, field: &str) -> Option<u64> {
match value {
Value::Object(fields) => {
if let Some(metric_value) = fields.get(metric)
&& let Some(value) = numeric_json_value(metric_value).or_else(|| numeric_json_field(metric_value, field))
{
return Some(value);
}
fields.values().find_map(|value| mimalloc_stat_field(value, metric, field))
}
Value::Array(values) => values.iter().find_map(|value| mimalloc_stat_field(value, metric, field)),
_ => None,
}
}
fn mimalloc_stat_current(value: &Value, metric: &str) -> Option<u64> {
mimalloc_stat_field(value, metric, "current")
}
fn parse_mimalloc_stats_json(stats_json: &str) -> Option<AllocatorMemoryObservation> {
let value = serde_json::from_str::<Value>(stats_json).ok()?;
let observation = AllocatorMemoryObservation {
reserved_bytes: mimalloc_stat_current(&value, "reserved"),
committed_bytes: mimalloc_stat_current(&value, "committed"),
page_committed_bytes: mimalloc_stat_current(&value, "page_committed"),
malloc_requested_bytes: mimalloc_stat_current(&value, "malloc_requested"),
malloc_requested_peak_bytes: mimalloc_stat_field(&value, "malloc_requested", "peak"),
malloc_requested_total_bytes: mimalloc_stat_field(&value, "malloc_requested", "total"),
heap_count: mimalloc_stat_current(&value, "heaps").or_else(|| mimalloc_stat_current(&value, "heap_count")),
};
if observation == AllocatorMemoryObservation::default() {
None
} else {
Some(observation)
}
}
#[cfg(not(target_os = "windows"))]
#[allow(unsafe_code)]
fn read_allocator_memory_snapshot() -> Option<AllocatorMemorySnapshot> {
// SAFETY: `mi_stats_get_json` returns a null-terminated JSON buffer owned by
// mimalloc when called with a null input buffer. The mimalloc API requires
// freeing that buffer with `mi_free`, which is done before returning.
let stats = unsafe {
let stats_ptr = libmimalloc_sys::mi_stats_get_json(0, std::ptr::null_mut());
if stats_ptr.is_null() {
return None;
}
let stats = CStr::from_ptr(stats_ptr).to_str().ok().map(str::to_owned);
libmimalloc_sys::mi_free(stats_ptr.cast());
stats?
};
let observation = parse_mimalloc_stats_json(&stats)?;
Some(AllocatorMemorySnapshot {
backend: crate::allocator_reclaim::allocator_backend(),
observation,
})
}
#[cfg(target_os = "windows")]
fn read_allocator_memory_snapshot() -> Option<AllocatorMemorySnapshot> {
None
}
fn configured_memory_observability_interval_secs() -> u64 {
rustfs_utils::get_env_u64(ENV_MEMORY_OBSERVABILITY_INTERVAL_SECS, DEFAULT_MEMORY_OBSERVABILITY_INTERVAL_SECS).max(1)
}
@@ -276,11 +373,12 @@ async fn record_memory_snapshot() {
let (resource, process) = snapshot_process_resource_and_system();
let total_memory = refresh_total_memory();
let cgroup = read_cgroup_memory_snapshot();
(resource, process, total_memory, cgroup)
let allocator = read_allocator_memory_snapshot();
(resource, process, total_memory, cgroup, allocator)
})
.await
{
Ok((resource, process, total_memory, cgroup)) => {
Ok((resource, process, total_memory, cgroup, allocator)) => {
record_memory_usage(process.resident_memory_bytes, total_memory);
record_cpu_usage(resource.cpu_percent);
record_process_memory_split(process.resident_memory_bytes, process.virtual_memory_bytes);
@@ -295,6 +393,10 @@ async fn record_memory_snapshot() {
cgroup.inactive_file_bytes,
);
}
if let Some(allocator) = allocator {
record_allocator_memory_observation(allocator.backend, allocator.observation);
}
}
Err(err) => {
debug!(error = ?err, "memory observability sampler task failed");
@@ -330,7 +432,7 @@ mod tests {
CgroupMemorySnapshot, MEMORY_OBSERVABILITY_SERVICE_NAME, MemoryObservabilityCancellationSource,
MemoryObservabilityController, MemoryObservabilityDesiredState, MemoryObservabilityServiceState,
MemoryObservabilityShutdownHandle, MemoryObservabilityWorkerMutation, build_memory_observability_controller_snapshot,
build_memory_observability_status_snapshot, parse_kv_stats, read_optional_u64,
build_memory_observability_status_snapshot, parse_kv_stats, parse_mimalloc_stats_json, read_optional_u64,
};
use std::fs;
use std::path::PathBuf;
@@ -366,6 +468,39 @@ mod tests {
assert_eq!(snapshot.inactive_file_bytes, None);
}
#[test]
fn parse_mimalloc_stats_json_extracts_allocator_attribution() {
let parsed = parse_mimalloc_stats_json(
r#"{
"process": {
"reserved": { "current": 1048576 },
"committed": { "current": "524288" },
"page_committed": { "current": 262144 },
"malloc_requested": {
"current": 131072,
"peak": 196608,
"total": 10485760
},
"heaps": { "current": 8 }
}
}"#,
)
.expect("mimalloc stats should parse");
assert_eq!(parsed.reserved_bytes, Some(1_048_576));
assert_eq!(parsed.committed_bytes, Some(524_288));
assert_eq!(parsed.page_committed_bytes, Some(262_144));
assert_eq!(parsed.malloc_requested_bytes, Some(131_072));
assert_eq!(parsed.malloc_requested_peak_bytes, Some(196_608));
assert_eq!(parsed.malloc_requested_total_bytes, Some(10_485_760));
assert_eq!(parsed.heap_count, Some(8));
}
#[test]
fn parse_mimalloc_stats_json_rejects_unrecognized_payload() {
assert_eq!(parse_mimalloc_stats_json(r#"{ "allocator": "unknown" }"#), None);
}
#[test]
fn memory_observability_snapshot_reports_disabled_when_metrics_are_disabled() {
let snapshot = build_memory_observability_status_snapshot(false, 15, false);
rustfs/src/storage/ecfs.rs
+1
View File
@@ -78,6 +78,7 @@ impl Default for FS {
impl FS {
pub fn new() -> Self {
rustfs_io_metrics::init_s3_metrics();
rustfs_io_metrics::init_list_objects_metrics();
Self {}
}
scripts/run_listobjects_verified_bench.sh
Executable
+1324
View File
File diff suppressed because it is too large Load Diff
scripts/validate_issue_841_list_objects_observability.sh
Executable
+102
View File
@@ -0,0 +1,102 @@
#!/usr/bin/env bash
set -euo pipefail
# Acceptance runner for rustfs/backlog#841.
PROJECT_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
OUT_DIR="${OUT_DIR:-${PROJECT_ROOT}/target/issue-841-acceptance-$(date +%Y%m%d-%H%M%S)}"
log_info() { printf '[INFO] %s\n' "$*"; }
log_error() { printf '[ERROR] %s\n' "$*" >&2; }
usage() {
cat <<'USAGE'
Usage:
scripts/validate_issue_841_list_objects_observability.sh
Environment:
OUT_DIR output directory (default: target/issue-841-acceptance-<ts>)
USAGE
}
require_cmd() {
if ! command -v "$1" >/dev/null 2>&1; then
log_error "command not found: $1"
exit 1
fi
}
run_unit_checks() {
log_info "Running ListObjects observability unit checks"
local test_log="$OUT_DIR/issue-841-tests.log"
: > "$test_log"
(
cd "$PROJECT_ROOT"
cargo test -p rustfs-io-metrics list_objects_metrics -- --nocapture
cargo test -p rustfs-ecstore list_objects -- --nocapture
) 2>&1 | tee "$test_log"
log_info "Unit test log: $test_log"
}
run_static_checks() {
log_info "Running ListObjects observability static checks"
local static_log="$OUT_DIR/static-checks.log"
: > "$static_log"
local required_patterns=(
"rustfs_s3_list_objects_gather_total"
"rustfs_s3_list_objects_gather_scan_amplification"
"rustfs_s3_list_objects_merge_fan_in"
"record_list_objects_gather"
"record_list_objects_merge"
"init_list_objects_metrics"
"listobjects-v2-baseline-fixtures"
)
local pattern
for pattern in "${required_patterns[@]}"; do
if rg --no-ignore -n "$pattern" "$PROJECT_ROOT/crates" "$PROJECT_ROOT/rustfs" "$PROJECT_ROOT/docs" "$PROJECT_ROOT/scripts" >> "$static_log"; then
log_info "Found required symbol: $pattern"
else
log_error "Missing required symbol: $pattern"
log_error "Static check log: $static_log"
return 1
fi
done
log_info "Static check log: $static_log"
}
parse_args() {
while [[ $# -gt 0 ]]; do
case "$1" in
-h|--help)
usage
exit 0
;;
*)
log_error "unknown arg: $1"
usage
exit 1
;;
esac
done
}
main() {
parse_args "$@"
require_cmd cargo
require_cmd rg
require_cmd tee
mkdir -p "$OUT_DIR"
run_unit_checks
run_static_checks
log_info "Validation summary:"
log_info " - logs: $OUT_DIR"
log_info " - status: pass"
}
main "$@"