diff --git a/.docker/compose/docker-compose.cluster.local-build.yml b/.docker/compose/docker-compose.cluster.local-build.yml index a41ec96d2..6e2208ce9 100644 --- a/.docker/compose/docker-compose.cluster.local-build.yml +++ b/.docker/compose/docker-compose.cluster.local-build.yml @@ -20,13 +20,16 @@ services: dockerfile: Dockerfile.source hostname: node1 environment: - - RUSTFS_VOLUMES=http://node{1...4}:9000/data/rustfs{0...3} + - RUSTFS_VOLUMES=${RUSTFS_VOLUMES:-http://node{1...4}:9000/data/rustfs{0...3}} - RUSTFS_ADDRESS=:9000 - RUSTFS_CONSOLE_ENABLE=true - RUSTFS_ACCESS_KEY=${RUSTFS_ACCESS_KEY:-rustfs-cluster-admin} - RUSTFS_SECRET_KEY=${RUSTFS_SECRET_KEY:-rustfs-cluster-secret} - RUSTFS_OBS_ENDPOINT=${RUSTFS_OBS_ENDPOINT:-http://host.docker.internal:4318} - RUSTFS_OBS_LOGGER_LEVEL=${RUSTFS_OBS_LOGGER_LEVEL:-info} + - RUSTFS_OBS_USE_STDOUT=${RUSTFS_OBS_USE_STDOUT:-false} + - RUSTFS_OBS_LOG_STDOUT_ENABLED=${RUSTFS_OBS_LOG_STDOUT_ENABLED:-false} + - RUSTFS_ISSUE3031_DIAG_ENABLE=${RUSTFS_ISSUE3031_DIAG_ENABLE:-false} - RUSTFS_UNSAFE_BYPASS_DISK_CHECK=${RUSTFS_UNSAFE_BYPASS_DISK_CHECK:-true} extra_hosts: - "host.docker.internal:host-gateway" @@ -47,13 +50,16 @@ services: dockerfile: Dockerfile.source hostname: node2 environment: - - RUSTFS_VOLUMES=http://node{1...4}:9000/data/rustfs{0...3} + - RUSTFS_VOLUMES=${RUSTFS_VOLUMES:-http://node{1...4}:9000/data/rustfs{0...3}} - RUSTFS_ADDRESS=:9000 - RUSTFS_CONSOLE_ENABLE=true - RUSTFS_ACCESS_KEY=${RUSTFS_ACCESS_KEY:-rustfs-cluster-admin} - RUSTFS_SECRET_KEY=${RUSTFS_SECRET_KEY:-rustfs-cluster-secret} - RUSTFS_OBS_ENDPOINT=${RUSTFS_OBS_ENDPOINT:-http://host.docker.internal:4318} - RUSTFS_OBS_LOGGER_LEVEL=${RUSTFS_OBS_LOGGER_LEVEL:-info} + - RUSTFS_OBS_USE_STDOUT=${RUSTFS_OBS_USE_STDOUT:-false} + - RUSTFS_OBS_LOG_STDOUT_ENABLED=${RUSTFS_OBS_LOG_STDOUT_ENABLED:-false} + - RUSTFS_ISSUE3031_DIAG_ENABLE=${RUSTFS_ISSUE3031_DIAG_ENABLE:-false} - RUSTFS_UNSAFE_BYPASS_DISK_CHECK=${RUSTFS_UNSAFE_BYPASS_DISK_CHECK:-true} extra_hosts: - "host.docker.internal:host-gateway" @@ -74,13 +80,16 @@ services: dockerfile: Dockerfile.source hostname: node3 environment: - - RUSTFS_VOLUMES=http://node{1...4}:9000/data/rustfs{0...3} + - RUSTFS_VOLUMES=${RUSTFS_VOLUMES:-http://node{1...4}:9000/data/rustfs{0...3}} - RUSTFS_ADDRESS=:9000 - RUSTFS_CONSOLE_ENABLE=true - RUSTFS_ACCESS_KEY=${RUSTFS_ACCESS_KEY:-rustfs-cluster-admin} - RUSTFS_SECRET_KEY=${RUSTFS_SECRET_KEY:-rustfs-cluster-secret} - RUSTFS_OBS_ENDPOINT=${RUSTFS_OBS_ENDPOINT:-http://host.docker.internal:4318} - RUSTFS_OBS_LOGGER_LEVEL=${RUSTFS_OBS_LOGGER_LEVEL:-info} + - RUSTFS_OBS_USE_STDOUT=${RUSTFS_OBS_USE_STDOUT:-false} + - RUSTFS_OBS_LOG_STDOUT_ENABLED=${RUSTFS_OBS_LOG_STDOUT_ENABLED:-false} + - RUSTFS_ISSUE3031_DIAG_ENABLE=${RUSTFS_ISSUE3031_DIAG_ENABLE:-false} - RUSTFS_UNSAFE_BYPASS_DISK_CHECK=${RUSTFS_UNSAFE_BYPASS_DISK_CHECK:-true} extra_hosts: - "host.docker.internal:host-gateway" @@ -101,13 +110,16 @@ services: dockerfile: Dockerfile.source hostname: node4 environment: - - RUSTFS_VOLUMES=http://node{1...4}:9000/data/rustfs{0...3} + - RUSTFS_VOLUMES=${RUSTFS_VOLUMES:-http://node{1...4}:9000/data/rustfs{0...3}} - RUSTFS_ADDRESS=:9000 - RUSTFS_CONSOLE_ENABLE=true - RUSTFS_ACCESS_KEY=${RUSTFS_ACCESS_KEY:-rustfs-cluster-admin} - RUSTFS_SECRET_KEY=${RUSTFS_SECRET_KEY:-rustfs-cluster-secret} - RUSTFS_OBS_ENDPOINT=${RUSTFS_OBS_ENDPOINT:-http://host.docker.internal:4318} - RUSTFS_OBS_LOGGER_LEVEL=${RUSTFS_OBS_LOGGER_LEVEL:-info} + - RUSTFS_OBS_USE_STDOUT=${RUSTFS_OBS_USE_STDOUT:-false} + - RUSTFS_OBS_LOG_STDOUT_ENABLED=${RUSTFS_OBS_LOG_STDOUT_ENABLED:-false} + - RUSTFS_ISSUE3031_DIAG_ENABLE=${RUSTFS_ISSUE3031_DIAG_ENABLE:-false} - RUSTFS_UNSAFE_BYPASS_DISK_CHECK=${RUSTFS_UNSAFE_BYPASS_DISK_CHECK:-true} extra_hosts: - "host.docker.internal:host-gateway" diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index fd3261002..bf13e3eb0 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -1183,6 +1183,11 @@ impl LocalDisk { } } + // Ensure Tokio's file write is driven to completion before callers expose + // metadata paths to subsequent reads. + f.flush().await.map_err(to_file_error)?; + f.shutdown().await.map_err(to_file_error)?; + Ok(()) } @@ -1993,11 +1998,7 @@ impl DiskAPI for LocalDisk { let meta_op = match lstat_std(&src_file_path).map_err(|e| to_file_error(e).into()) { Ok(meta) => Some(meta), Err(e) => { - if e != DiskError::FileNotFound { - return Err(e); - } - - None + return Err(e); } }; @@ -2009,10 +2010,22 @@ impl DiskAPI for LocalDisk { } remove_std(&dst_file_path).map_err(to_file_error)?; + } else { + let meta = lstat_std(&src_file_path).map_err(|e| -> DiskError { to_file_error(e).into() })?; + if meta.is_dir() { + warn!("rename_part src is dir {:?}", &src_file_path); + return Err(DiskError::FileAccessDenied); + } } rename_all(&src_file_path, &dst_file_path, &dst_volume_dir).await?; + let dst_meta = lstat_std(&dst_file_path).map_err(|e| -> DiskError { to_file_error(e).into() })?; + if src_is_dir != dst_meta.is_dir() { + warn!("rename_part dst type changed after rename {:?}", &dst_file_path); + return Err(DiskError::FileAccessDenied); + } + self.write_all(dst_volume, format!("{dst_path}.meta").as_str(), meta).await?; if let Some(parent) = src_file_path.parent() { diff --git a/crates/ecstore/src/set_disk.rs b/crates/ecstore/src/set_disk.rs index 03d99e803..4c5155a26 100644 --- a/crates/ecstore/src/set_disk.rs +++ b/crates/ecstore/src/set_disk.rs @@ -130,6 +130,7 @@ pub const DEFAULT_READ_BUFFER_SIZE: usize = MI_B; // 1 MiB = 1024 * 1024; pub const MAX_PARTS_COUNT: usize = 10000; pub(crate) const RUSTFS_MULTIPART_BUCKET_KEY: &str = "x-rustfs-internal-multipart-bucket"; pub(crate) const RUSTFS_MULTIPART_OBJECT_KEY: &str = "x-rustfs-internal-multipart-object"; +const ENV_ISSUE3031_DIAG_ENABLE: &str = "RUSTFS_ISSUE3031_DIAG_ENABLE"; pub(crate) fn strip_internal_multipart_metadata(metadata: &mut HashMap) { metadata.remove(RUSTFS_MULTIPART_BUCKET_KEY); @@ -270,6 +271,10 @@ fn record_lock_release(bucket: &str, object: &str, lock_id: &str, lock_type: &st ); } +fn issue3031_diag_enabled() -> bool { + rustfs_utils::get_env_bool(ENV_ISSUE3031_DIAG_ENABLE, false) +} + fn build_tiered_decommission_file_info( bucket: &str, object: &str, @@ -3262,6 +3267,7 @@ impl MultipartOperations for SetDisks { let upload_id_path = Self::get_upload_id_dir(bucket, object, upload_id); let write_quorum = fi.write_quorum(self.default_write_quorum()); + let read_quorum = fi.read_quorum(self.default_read_quorum()); let disks = self.disks.read().await; @@ -3278,7 +3284,7 @@ impl MultipartOperations for SetDisks { let part_numbers = uploaded_parts.iter().map(|v| v.part_num).collect::>(); let object_parts = - Self::read_parts(&disks, RUSTFS_META_MULTIPART_BUCKET, &part_meta_paths, &part_numbers, write_quorum).await?; + Self::read_parts(&disks, RUSTFS_META_MULTIPART_BUCKET, &part_meta_paths, &part_numbers, read_quorum).await?; if object_parts.len() != uploaded_parts.len() { return Err(Error::other("part result number err")); @@ -3302,6 +3308,21 @@ impl MultipartOperations for SetDisks { for (i, part) in object_parts.iter().enumerate() { if let Some(err) = &part.error { error!("complete_multipart_upload part error: {:?}", &err); + if issue3031_diag_enabled() { + warn!( + target: "rustfs_ecstore::set_disk", + op = "complete_multipart_upload", + bucket = %bucket, + object = %object, + upload_id = %upload_id, + uploaded_part_num = uploaded_parts[i].part_num, + observed_part_num = part.number, + read_quorum = read_quorum, + write_quorum = write_quorum, + error = %err, + "issue3031_complete_part_error" + ); + } } if uploaded_parts[i].part_num != part.number { diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs index 35a2f9102..f0236c7ff 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/crates/ecstore/src/set_disk/read.rs @@ -144,18 +144,24 @@ impl SetDisks { for (part_idx, part_info) in part_meta_paths.iter().enumerate() { let mut part_meta_quorum = HashMap::new(); let mut part_infos = Vec::new(); - for (j, parts) in object_parts.iter().enumerate() { + let mut present_count = 0usize; + let mut missing_or_empty_count = 0usize; + let mut mismatched_response_count = 0usize; + for parts in object_parts.iter() { if parts.len() != part_meta_paths.len() { + mismatched_response_count += 1; *part_meta_quorum.entry(part_info.clone()).or_insert(0) += 1; continue; } if !parts[part_idx].etag.is_empty() { + present_count += 1; *part_meta_quorum.entry(parts[part_idx].etag.clone()).or_insert(0) += 1; part_infos.push(parts[part_idx].clone()); continue; } + missing_or_empty_count += 1; *part_meta_quorum.entry(part_info.clone()).or_insert(0) += 1; } @@ -194,6 +200,22 @@ impl SetDisks { { ret[part_idx] = found.clone(); } else { + if issue3031_diag_enabled() { + warn!( + target: "rustfs_ecstore::set_disk", + bucket = %bucket, + part_meta_path = %part_info, + part_id = part_numbers[part_idx], + read_quorum = read_quorum, + max_quorum = max_quorum, + disk_response_count = object_parts.len(), + present_count = present_count, + missing_or_empty_count = missing_or_empty_count, + mismatched_response_count = mismatched_response_count, + max_vote_is_missing_marker = max_etag.map(|etag| etag == part_info).unwrap_or(false), + "issue3031_read_parts_part_quorum" + ); + } ret[part_idx] = ObjectPartInfo { number: part_numbers[part_idx], error: Some(format!("part.{} not found", part_numbers[part_idx])), diff --git a/crates/ecstore/src/set_disk/write.rs b/crates/ecstore/src/set_disk/write.rs index f317ed94d..9ce0fe277 100644 --- a/crates/ecstore/src/set_disk/write.rs +++ b/crates/ecstore/src/set_disk/write.rs @@ -202,6 +202,9 @@ impl SetDisks { #[tracing::instrument(skip(self))] pub(super) async fn cleanup_multipart_path(&self, paths: &[String]) { + if paths.is_empty() { + return; + } let disks = self.get_disks_internal().await; let mut errs = Vec::with_capacity(disks.len()); @@ -290,6 +293,27 @@ impl SetDisks { } } + if issue3031_diag_enabled() { + let success_count = errs.iter().filter(|err| err.is_none()).count(); + let error_count = errs.len().saturating_sub(success_count); + let disk_not_found_count = errs.iter().filter(|err| matches!(err, Some(DiskError::DiskNotFound))).count(); + let file_not_found_count = errs.iter().filter(|err| matches!(err, Some(DiskError::FileNotFound))).count(); + warn!( + target: "rustfs_ecstore::set_disk", + src_bucket = %src_bucket, + src_object = %src_object, + dst_bucket = %dst_bucket, + dst_object = %dst_object, + write_quorum = write_quorum, + disk_count = errs.len(), + success_count = success_count, + error_count = error_count, + disk_not_found_count = disk_not_found_count, + file_not_found_count = file_not_found_count, + "issue3031_rename_part_context" + ); + } + if let Some(err) = reduce_write_quorum_errs(&errs, OBJECT_OP_IGNORED_ERRS, write_quorum) { warn!("rename_part errs {:?}", &errs); self.cleanup_multipart_path(&[dst_object.to_string(), format!("{dst_object}.meta")])