Compare commits

..

1 Commits

Author SHA1 Message Date
overtrue d33e2fdcf1 chore(madmin): remove dead trace structs, keep TraceType bitflag helper only
TraceInfo, TraceInfoLegacy, TraceHTTPStats, TraceCallStats, TraceRequestInfo,
TraceResponseInfo, StorageStats, and OSStats are unreferenced outside trace.rs.
Trim to TraceType + its bitflag operations which are actively used by
service_commands.rs and profile_admin.rs.

-139 lines (215 -> 76 lines)
2026-08-21 23:54:02 +08:00
3 changed files with 36 additions and 187 deletions
-7
View File
@@ -57,13 +57,6 @@ pub const DEFAULT_MAX_IO_EVENTS_PER_TICK: usize = 1024;
pub const DEFAULT_EVENT_INTERVAL: u32 = 61;
pub const DEFAULT_RNG_SEED: Option<u64> = None; // None means random
/// Dedicated blocking thread pool for fsync/fdatasync operations.
/// When > 1, fsync operations are isolated from the main blocking pool to
/// prevent device-bound fsync from starving read operations (pread/stat/open).
/// Default 0 means auto (no isolation, use main runtime).
pub const ENV_FSYNC_BLOCKING_THREADS: &str = "RUSTFS_RUNTIME_FSYNC_BLOCKING_THREADS";
pub const DEFAULT_FSYNC_BLOCKING_THREADS: usize = 0;
// Dial9 Tokio Telemetry Default values
pub const DEFAULT_RUNTIME_DIAL9_ENABLED: bool = false; // Disabled by default
pub const DEFAULT_RUNTIME_DIAL9_OUTPUT_DIR: &str = "/var/log/rustfs/telemetry";
+4 -42
View File
@@ -315,7 +315,7 @@ pub async fn fsync_dir(dir: impl AsRef<Path>) -> io::Result<()> {
#[cfg(unix)]
{
let dir = dir.as_ref().to_path_buf();
fsync_spawn_blocking(move || fsync_dir_std(dir)).await?
tokio::task::spawn_blocking(move || fsync_dir_std(dir)).await?
}
#[cfg(not(unix))]
@@ -683,7 +683,7 @@ async fn fsync_open_dst_dir_group(group: &DstDirFsyncGroup) -> io::Result<()> {
#[cfg(test)]
let dir = group.dir.clone();
let dir_file = group.dir_file.clone();
fsync_spawn_blocking(move || {
tokio::task::spawn_blocking(move || {
#[cfg(test)]
{
if let Some(kind) = fsync_dir_recorder::take_grouped_failure(&dir) {
@@ -1080,44 +1080,6 @@ const TEST_GLOBAL_FILE_SYNCS: usize = 64;
static FILE_SYNC_PERMITS: LazyLock<Semaphore> = LazyLock::new(|| Semaphore::new(global_file_sync_limit()));
static DISK_FILE_SYNC_LIMITERS: LazyLock<Mutex<HashMap<PathBuf, Weak<Semaphore>>>> = LazyLock::new(|| Mutex::new(HashMap::new()));
/// Dedicated tokio runtime for fsync/fdatasync blocking operations. When
/// configured with >1 threads, isolates device-bound fsync from the main
/// blocking pool so reads (pread/stat/open) are not starved. `None` means
/// fall back to the main runtime (zero behavior change).
static FSYNC_RUNTIME: LazyLock<Option<tokio::runtime::Runtime>> = LazyLock::new(|| {
let threads =
rustfs_utils::get_env_usize(rustfs_config::ENV_FSYNC_BLOCKING_THREADS, rustfs_config::DEFAULT_FSYNC_BLOCKING_THREADS);
if threads <= 1 {
return None;
}
let mut builder = tokio::runtime::Builder::new_multi_thread();
builder
.worker_threads(2)
.max_blocking_threads(threads)
.thread_name("rustfs-fsync")
.thread_stack_size(512 * 1024)
.enable_all();
match builder.build() {
Ok(rt) => {
tracing::info!(threads, "fsync dedicated blocking pool enabled");
Some(rt)
}
Err(err) => {
tracing::warn!(%err, "failed to build fsync runtime, falling back to main pool");
None
}
}
});
/// Spawn a blocking task on the fsync-dedicated runtime if configured,
/// otherwise fall back to the main tokio blocking pool.
fn fsync_spawn_blocking<T: Send + 'static>(f: impl FnOnce() -> T + Send + 'static) -> tokio::task::JoinHandle<T> {
match FSYNC_RUNTIME.as_ref() {
Some(rt) => rt.spawn_blocking(f),
None => tokio::task::spawn_blocking(f),
}
}
static DISK_VOLUME_MUTATION_LOCKS: LazyLock<Mutex<HashMap<PathBuf, Weak<RwLock<()>>>>> =
LazyLock::new(|| Mutex::new(HashMap::new()));
type NamespaceMutationLock = AsyncMutex<()>;
@@ -1255,7 +1217,7 @@ where
F: FnOnce() -> io::Result<T> + Send + 'static,
{
let (disk_permit, global_permit) = acquire_file_sync_permits(disk_permits).await?;
let result = fsync_spawn_blocking(move || {
let result = tokio::task::spawn_blocking(move || {
let _disk_permit = disk_permit;
work()
})
@@ -2184,7 +2146,7 @@ async fn run_blocking_namespace_file_sync_operation_with_global<T: Send + 'stati
wait_started,
);
let disk_permit = admission.disk_permit.clone();
let result = fsync_spawn_blocking(move || {
let result = tokio::task::spawn_blocking(move || {
let _lease = lease;
let _disk_permit = disk_permit;
operation()
+32 -138
View File
@@ -12,18 +12,16 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use std::{collections::HashMap, time::Duration};
use jiff::Timestamp;
use serde::{Deserialize, Serialize};
use crate::heal_commands::HealResultItem;
/// Bitflag helper for service trace categories.
///
/// Each variant occupies a single bit so that a `TraceType` value can represent
/// an arbitrary combination of categories via bitwise OR.
#[derive(Debug, Clone, Copy, Serialize, Deserialize, Default)]
pub struct TraceType(u64);
impl TraceType {
// Define some constants
pub const OS: TraceType = TraceType(1 << 0);
pub const STORAGE: TraceType = TraceType(1 << 1);
pub const S3: TraceType = TraceType(1 << 2);
@@ -40,15 +38,13 @@ impl TraceType {
pub const FTP: TraceType = TraceType(1 << 13);
pub const ILM: TraceType = TraceType(1 << 14);
// MetricsAll must be last.
/// All trace categories combined. Must be updated when adding new variants.
pub const ALL: TraceType = TraceType((1 << 15) - 1);
pub fn new(t: u64) -> Self {
Self(t)
}
}
impl TraceType {
pub fn contains(&self, x: &TraceType) -> bool {
(self.0 & x.0) == x.0
}
@@ -76,140 +72,38 @@ impl TraceType {
}
}
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
pub struct TraceInfo {
#[serde(rename = "type")]
trace_type: u64,
#[serde(rename = "nodename")]
node_name: String,
#[serde(rename = "funcname")]
func_name: String,
#[serde(rename = "time")]
time: Timestamp,
#[serde(rename = "path")]
path: String,
#[serde(rename = "dur")]
duration: Duration,
#[serde(rename = "bytes", skip_serializing_if = "Option::is_none")]
bytes: Option<i64>,
#[serde(rename = "msg", skip_serializing_if = "Option::is_none")]
message: Option<String>,
#[serde(rename = "error", skip_serializing_if = "Option::is_none")]
error: Option<String>,
#[serde(rename = "custom", skip_serializing_if = "Option::is_none")]
custom: Option<HashMap<String, String>>,
#[serde(rename = "http", skip_serializing_if = "Option::is_none")]
http: Option<TraceHTTPStats>,
#[serde(rename = "healResult", skip_serializing_if = "Option::is_none")]
heal_result: Option<HealResultItem>,
}
impl TraceInfo {
pub fn mask(&self) -> u64 {
TraceType::new(self.trace_type).mask()
}
}
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
pub struct TraceInfoLegacy {
trace_info: TraceInfo,
#[serde(rename = "request")]
req_info: Option<TraceRequestInfo>,
#[serde(rename = "response")]
resp_info: Option<TraceResponseInfo>,
#[serde(rename = "stats")]
call_stats: Option<TraceCallStats>,
#[serde(rename = "storageStats")]
storage_stats: Option<StorageStats>,
#[serde(rename = "osStats")]
os_stats: Option<OSStats>,
}
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
pub struct StorageStats {
path: String,
duration: Duration,
}
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
pub struct OSStats {
path: String,
duration: Duration,
}
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
pub struct TraceHTTPStats {
req_info: TraceRequestInfo,
resp_info: TraceResponseInfo,
call_stats: TraceCallStats,
}
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
pub struct TraceCallStats {
input_bytes: i32,
output_bytes: i32,
latency: Duration,
time_to_first_byte: Duration,
}
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
pub struct TraceRequestInfo {
time: Timestamp,
proto: String,
method: String,
#[serde(skip_serializing_if = "Option::is_none")]
path: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
raw_query: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
headers: Option<HashMap<String, String>>,
#[serde(skip_serializing_if = "Option::is_none")]
body: Option<Vec<u8>>,
client: String,
}
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
pub struct TraceResponseInfo {
time: Timestamp,
#[serde(skip_serializing_if = "Option::is_none")]
headers: Option<HashMap<String, String>>,
#[serde(skip_serializing_if = "Option::is_none")]
body: Option<Vec<u8>>,
#[serde(skip_serializing_if = "Option::is_none")]
status_code: Option<i32>,
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn trace_timestamps_serialize_as_rfc3339_utc() {
let timestamp = Timestamp::constant(1_700_000_000, 123_456_000);
let trace = TraceInfo {
time: timestamp,
http: Some(TraceHTTPStats {
req_info: TraceRequestInfo {
time: timestamp,
..Default::default()
},
resp_info: TraceResponseInfo {
time: timestamp,
..Default::default()
},
..Default::default()
}),
..Default::default()
};
fn trace_type_contains_and_overlaps() {
let mut combined = TraceType::default();
combined.merge(&TraceType::S3);
combined.merge(&TraceType::HEALING);
let value = serde_json::to_value(trace).expect("trace should serialize");
assert_eq!(value["time"], "2023-11-14T22:13:20.123456Z");
assert_eq!(value["http"]["req_info"]["time"], "2023-11-14T22:13:20.123456Z");
assert_eq!(value["http"]["resp_info"]["time"], "2023-11-14T22:13:20.123456Z");
let trace: TraceInfo = serde_json::from_value(value).expect("trace should deserialize");
assert_eq!(trace.time, timestamp);
let http = trace.http.expect("http trace should deserialize");
assert_eq!(http.req_info.time, timestamp);
assert_eq!(http.resp_info.time, timestamp);
assert!(combined.contains(&TraceType::S3));
assert!(combined.contains(&TraceType::HEALING));
assert!(!combined.contains(&TraceType::SCANNER));
assert!(combined.overlaps(&TraceType::S3));
assert!(combined.overlaps(&TraceType::HEALING));
assert!(!combined.overlaps(&TraceType::SCANNER));
}
#[test]
fn trace_type_set_if() {
let mut tt = TraceType::default();
tt.set_if(true, &TraceType::OS);
tt.set_if(false, &TraceType::S3);
assert!(tt.contains(&TraceType::OS));
assert!(!tt.contains(&TraceType::S3));
}
#[test]
fn trace_type_single_type() {
assert!(TraceType::S3.single_type());
let mut combined = TraceType::S3;
combined.merge(&TraceType::HEALING);
assert!(!combined.single_type());
}
}