From c6e1387f59dd995848f14ce73fbc3ff899533db1 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Mon, 10 Aug 2026 16:31:26 -0700 Subject: [PATCH] shell: multi-target fs.mergeVolumes and volume.mark -readonlyCanDelete (#10706) * shell: fs.mergeVolumes distributes one volume across multiple -toVolumeId targets * volume: volume.mark -readonlyCanDelete rejects writes but keeps accepting deletes * seaweed-volume: mirror readonlyCanDelete volume state --- seaweed-volume/proto/volume_server.proto | 3 + seaweed-volume/src/server/grpc_server.rs | 16 +- seaweed-volume/src/storage/volume.rs | 265 ++++++++++++++++-- weed/pb/volume_server.proto | 3 + weed/pb/volume_server_pb/volume_server.pb.go | 56 ++-- weed/server/volume_grpc_admin.go | 6 +- weed/server/volume_grpc_scrub.go | 2 +- weed/shell/command_fs_merge_volumes.go | 254 ++++++++++++----- weed/shell/command_fs_merge_volumes_test.go | 130 ++++++++- weed/shell/command_volume_mark.go | 29 +- weed/shell/command_volume_move.go | 10 +- weed/storage/store.go | 25 +- .../store_mark_readonly_can_delete_test.go | 122 ++++++++ weed/storage/volume.go | 7 +- weed/storage/volume_loading.go | 6 +- .../volume_loading_corrupt_idx_test.go | 2 +- weed/storage/volume_mark_writable_test.go | 2 +- weed/storage/volume_write_test.go | 2 +- 18 files changed, 782 insertions(+), 158 deletions(-) create mode 100644 weed/storage/store_mark_readonly_can_delete_test.go diff --git a/seaweed-volume/proto/volume_server.proto b/seaweed-volume/proto/volume_server.proto index 39346899e..8cce9d66a 100644 --- a/seaweed-volume/proto/volume_server.proto +++ b/seaweed-volume/proto/volume_server.proto @@ -257,6 +257,8 @@ message VolumeDeleteResponse { message VolumeMarkReadonlyRequest { uint32 volume_id = 1; bool persist = 2; + // reject writes but keep accepting deletes, so expiring data can drain the volume + bool can_delete = 3; } message VolumeMarkReadonlyResponse { } @@ -594,6 +596,7 @@ message VolumeInfo { uint64 expire_at_sec = 6; // expiration time of ec volume bool read_only = 7; EcShardConfig ec_shard_config = 8; // EC shard configuration (optional, null = use default 10+4) + bool read_only_can_delete = 9; // with read_only: writes are rejected but deletes still land } // EcShardConfig specifies erasure coding shard configuration diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index 2eb70165c..d307abc6c 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/seaweed-volume/src/server/grpc_server.rs @@ -237,12 +237,17 @@ impl VolumeGrpcService { Ok(()) } - /// Shared helper matching Go's `makeVolumeReadonly(ctx, v, persist)`. + /// Shared helper matching Go's `makeVolumeReadonly(ctx, v, canDelete, persist)`. /// 1. Check maintenance mode /// 2. Notify master (readonly=true) /// 3. Mark local volume readonly /// 4. Notify master again (cover heartbeat race) - async fn make_volume_readonly(&self, vid: VolumeId, persist: bool) -> Result<(), Status> { + async fn make_volume_readonly( + &self, + vid: VolumeId, + can_delete: bool, + persist: bool, + ) -> Result<(), Status> { self.state.check_maintenance()?; let info = { @@ -268,7 +273,7 @@ impl VolumeGrpcService { { let mut store = self.state.store.write().unwrap(); if let Some((_, vol)) = store.find_volume_mut(vid) { - vol.set_read_only_persist(persist) + vol.set_read_only_persist(can_delete, persist) .map_err(|e| Status::internal(e.to_string()))?; } self.state.volume_state_notify.notify_one(); @@ -977,7 +982,8 @@ impl VolumeServer for VolumeGrpcService { .find_volume(vid) .ok_or_else(|| Status::not_found(format!("volume {} not found", vid)))?; } - self.make_volume_readonly(vid, req.persist).await?; + self.make_volume_readonly(vid, req.can_delete, req.persist) + .await?; Ok(Response::new( volume_server_pb::VolumeMarkReadonlyResponse {}, )) @@ -4067,7 +4073,7 @@ impl VolumeServer for VolumeGrpcService { let mut errs: Vec = Vec::new(); if req.mark_broken_volumes_readonly { for vid in &broken_vids { - match self.make_volume_readonly(*vid, true).await { + match self.make_volume_readonly(*vid, false, true).await { Ok(()) => { details.push(format!("volume {} is now read-only", vid.0)); } diff --git a/seaweed-volume/src/storage/volume.rs b/seaweed-volume/src/storage/volume.rs index 67a03eef7..b4b479ff9 100644 --- a/seaweed-volume/src/storage/volume.rs +++ b/seaweed-volume/src/storage/volume.rs @@ -227,6 +227,7 @@ impl OldVersionVifVolumeInfo { expire_at_sec: self.destroy_time, read_only: self.read_only, ec_shard_config: None, + read_only_can_delete: false, } } } @@ -255,6 +256,8 @@ pub struct VifVolumeInfo { skip_serializing_if = "Option::is_none" )] pub ec_shard_config: Option, + #[serde(default, rename = "readOnlyCanDelete")] + pub read_only_can_delete: bool, } impl VifVolumeInfo { @@ -285,6 +288,7 @@ impl VifVolumeInfo { parity_shards: c.parity_shards, encode_ts_ns: c.encode_ts_ns, }), + read_only_can_delete: pb.read_only_can_delete, } } @@ -317,6 +321,7 @@ impl VifVolumeInfo { encode_ts_ns: c.encode_ts_ns, } }), + read_only_can_delete: self.read_only_can_delete, } } } @@ -684,7 +689,11 @@ impl Volume { let has_volume_info_file = self.load_vif()?; if self.volume_info.read_only && !self.has_remote_file { - self.no_write_or_delete = true; + if self.volume_info.read_only_can_delete { + self.no_write_can_delete = true; + } else { + self.no_write_or_delete = true; + } } if self.has_remote_file { @@ -2299,34 +2308,39 @@ impl Volume { Ok(()) } - /// Mark this volume as read-only (no writes or deletes). - /// If `persist` is true, the readonly state is saved to the .vif file. + /// Mark this volume as read-only (no writes or deletes) and persist. pub fn set_read_only(&mut self) -> Result<(), VolumeError> { - self.no_write_or_delete = true; - self.save_vif() + self.set_read_only_persist(false, true) } /// Mark this volume as read-only, optionally persisting to .vif. - pub fn set_read_only_persist(&mut self, persist: bool) -> Result<(), VolumeError> { - self.no_write_or_delete = true; + /// With `can_delete`, deletes still land so expiring data drains the volume. + pub fn set_read_only_persist( + &mut self, + can_delete: bool, + persist: bool, + ) -> Result<(), VolumeError> { + if can_delete && !self.has_remote_file { + // deletes append tombstones to .idx; a read-only boot attached no writer + self.attach_idx_writer_if_missing()?; + } + self.no_write_or_delete = !can_delete; + if can_delete { + self.no_write_can_delete = true; + } else if !self.has_remote_file { + // downgrading a canDelete mark; remote volumes keep their derived flag + self.no_write_can_delete = false; + } if persist { self.save_vif()?; } Ok(()) } - /// Mark this volume as writable (allow writes and deletes). - /// - /// If the volume booted with .vif ReadOnly=true, `load_index` built the - /// needle map without an .idx writer attached, so subsequent puts would - /// silently skip the on-disk append and only mutate in-memory state — - /// surviving until the next restart, then vanishing. Re-attach a writer - /// here so writes persist again. - pub fn set_writable(&mut self) -> Result<(), VolumeError> { - // Attach the writer (if missing) before flipping the flag — otherwise - // a transient open/metadata failure would leave the volume marked - // writable with no .idx writer, and subsequent puts would silently - // skip the on-disk append and vanish on the next restart. + /// Attach an append-only .idx writer when missing (read-only boot). Runs + /// before any flag flips so a failure cannot leave puts silently skipping + /// the on-disk append. + fn attach_idx_writer_if_missing(&mut self) -> Result<(), VolumeError> { let needs_idx_writer = self .nm .as_ref() @@ -2344,7 +2358,23 @@ impl Volume { nm.set_idx_file(Box::new(write_file), idx_size); } } + Ok(()) + } + + /// Mark this volume as writable (allow writes and deletes). + /// + /// If the volume booted with .vif ReadOnly=true, `load_index` built the + /// needle map without an .idx writer attached, so subsequent puts would + /// silently skip the on-disk append and only mutate in-memory state — + /// surviving until the next restart, then vanishing. Re-attach a writer + /// here so writes persist again. + pub fn set_writable(&mut self) -> Result<(), VolumeError> { + self.attach_idx_writer_if_missing()?; self.no_write_or_delete = false; + // Remote-tiered volumes must stay no_write_can_delete regardless of marks. + if !self.has_remote_file { + self.no_write_can_delete = false; + } self.save_vif() } @@ -2354,7 +2384,8 @@ impl Volume { if self.has_remote_file { self.no_write_can_delete = true; self.no_write_or_delete = false; - } else { + } else if !self.volume_info.read_only_can_delete { + // only clear the remoteness-derived flag, not an operator mark self.no_write_can_delete = false; } } @@ -2389,7 +2420,11 @@ impl Volume { if let Ok(vif_info) = serde_json::from_str::(&content) { let pb_info = vif_info.to_pb(); if pb_info.read_only { - self.no_write_or_delete = true; + if pb_info.read_only_can_delete { + self.no_write_can_delete = true; + } else { + self.no_write_or_delete = true; + } } self.volume_info = pb_info; self.refresh_remote_write_mode(); @@ -2454,7 +2489,7 @@ impl Volume { /// Save volume info to .vif file in protobuf-JSON format (Go-compatible). /// Matches Go's SaveVolumeInfo: checks writability before writing and propagates errors. - fn save_vif(&self) -> Result<(), VolumeError> { + fn save_vif(&mut self) -> Result<(), VolumeError> { let vif_path = self.vif_path(); // Match Go: if file exists but is not writable, return an error @@ -2469,8 +2504,13 @@ impl Volume { } } + // remoteness-derived no_write_can_delete is not an operator mark; sync + // volume_info so refresh_remote_write_mode sees a real mark + let marked_can_delete = + self.no_write_can_delete && !self.has_remote_file && !self.no_write_or_delete; + self.volume_info.read_only = self.no_write_or_delete || marked_can_delete; + self.volume_info.read_only_can_delete = marked_can_delete; let mut vif = VifVolumeInfo::from_pb(&self.volume_info); - vif.read_only = self.no_write_or_delete; // Match Go's SaveVolumeInfo: compute ExpireAtSec from TTL let ttl_seconds = self.super_block.ttl.to_seconds(); @@ -2491,7 +2531,10 @@ impl Volume { /// Save full VolumeInfo to .vif file (for tiered storage). /// Matches Go's SaveVolumeInfo which computes ExpireAtSec from TTL. pub fn save_volume_info(&mut self) -> Result<(), VolumeError> { - self.volume_info.read_only = self.no_write_or_delete; + let marked_can_delete = + self.no_write_can_delete && !self.has_remote_file && !self.no_write_or_delete; + self.volume_info.read_only = self.no_write_or_delete || marked_can_delete; + self.volume_info.read_only_can_delete = marked_can_delete; // Compute ExpireAtSec from TTL (matches Go's SaveVolumeInfo) let ttl_seconds = self.super_block.ttl.to_seconds(); @@ -5096,7 +5139,7 @@ mod tests { ..Needle::default() }; v.write_needle(&mut n, true).unwrap(); - v.set_read_only_persist(true).unwrap(); + v.set_read_only_persist(false, true).unwrap(); v.sync_to_disk().unwrap(); } @@ -5161,6 +5204,178 @@ mod tests { assert_eq!(std::str::from_utf8(&probe.data).unwrap(), "after-mark-writable"); } + // readOnlyCanDelete rejects writes, keeps accepting deletes, and survives + // a restart. + #[test] + fn test_read_only_can_delete_persists_and_keeps_deletes_working() { + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + + { + let mut v = make_test_volume(dir); + for id in 1..=2u64 { + let mut n = Needle { + id: NeedleId(id), + cookie: Cookie(id as u32), + data: b"payload".to_vec(), + data_size: 7, + ..Needle::default() + }; + v.write_needle(&mut n, true).unwrap(); + } + v.set_read_only_persist(true, true).unwrap(); + assert!(v.no_write_can_delete); + assert!(!v.no_write_or_delete); + assert!(v.is_read_only()); + + let err = v + .write_needle( + &mut Needle { + id: NeedleId(3), + cookie: Cookie(3), + data: b"blocked".to_vec(), + data_size: 7, + ..Needle::default() + }, + true, + ) + .unwrap_err(); + assert!(matches!(err, VolumeError::ReadOnly)); + + let deleted = v + .delete_needle(&mut Needle { + id: NeedleId(1), + cookie: Cookie(1), + ..Needle::default() + }) + .unwrap(); + assert!(deleted.0 > 0, "deletes must still land"); + v.sync_to_disk().unwrap(); + } + + let mut v = Volume::new( + dir, + dir, + "", + VolumeId(1), + NeedleMapKind::InMemory, + None, + None, + 0, + Version::current(), + ) + .unwrap(); + assert!(v.no_write_can_delete, "state must survive a restart"); + assert!(!v.no_write_or_delete); + assert!( + v.nm.as_ref().unwrap().has_idx_writer(), + "canDelete boot must attach an .idx writer for delete tombstones" + ); + + let err = v + .write_needle( + &mut Needle { + id: NeedleId(4), + cookie: Cookie(4), + data: b"blocked".to_vec(), + data_size: 7, + ..Needle::default() + }, + true, + ) + .unwrap_err(); + assert!(matches!(err, VolumeError::ReadOnly)); + + let deleted = v + .delete_needle(&mut Needle { + id: NeedleId(2), + cookie: Cookie(2), + ..Needle::default() + }) + .unwrap(); + assert!(deleted.0 > 0, "deletes must still land after restart"); + + // downgrading to plain readonly clears the canDelete flag + v.set_read_only_persist(false, true).unwrap(); + assert!(v.no_write_or_delete); + assert!(!v.no_write_can_delete); + + // -writable clears the state again. + v.set_writable().unwrap(); + assert!(!v.is_read_only()); + let mut n = Needle { + id: NeedleId(5), + cookie: Cookie(5), + data: b"writable-again".to_vec(), + data_size: 14, + ..Needle::default() + }; + v.write_needle(&mut n, true).unwrap(); + } + + // Upgrading a volume that booted plain persisted-readonly to canDelete + // must attach the missing .idx writer, or delete tombstones cannot append. + #[test] + fn test_read_only_can_delete_upgrade_after_readonly_boot() { + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + + { + let mut v = make_test_volume(dir); + let mut n = Needle { + id: NeedleId(1), + cookie: Cookie(1), + data: b"payload".to_vec(), + data_size: 7, + ..Needle::default() + }; + v.write_needle(&mut n, true).unwrap(); + v.set_read_only_persist(false, true).unwrap(); + v.sync_to_disk().unwrap(); + } + + let mut v = Volume::new( + dir, + dir, + "", + VolumeId(1), + NeedleMapKind::InMemory, + None, + None, + 0, + Version::current(), + ) + .unwrap(); + assert!(v.no_write_or_delete); + let err = v + .delete_needle(&mut Needle { + id: NeedleId(1), + cookie: Cookie(1), + ..Needle::default() + }) + .unwrap_err(); + assert!( + matches!(err, VolumeError::ReadOnly), + "plain readonly must reject deletes" + ); + + v.set_read_only_persist(true, true).unwrap(); + assert!(!v.no_write_or_delete); + assert!(v.no_write_can_delete); + assert!( + v.nm.as_ref().unwrap().has_idx_writer(), + "upgrade to canDelete must attach the .idx writer" + ); + let deleted = v + .delete_needle(&mut Needle { + id: NeedleId(1), + cookie: Cookie(1), + ..Needle::default() + }) + .unwrap(); + assert!(deleted.0 > 0, "deletes must land once canDelete is set"); + } + #[test] fn test_load_vif_defaults_local_version_and_bytes_offset() { let tmp = TempDir::new().unwrap(); diff --git a/weed/pb/volume_server.proto b/weed/pb/volume_server.proto index 72c0b7877..5abaa06f9 100644 --- a/weed/pb/volume_server.proto +++ b/weed/pb/volume_server.proto @@ -265,6 +265,8 @@ message VolumeDeleteResponse { message VolumeMarkReadonlyRequest { uint32 volume_id = 1; bool persist = 2; + // reject writes but keep accepting deletes, so expiring data can drain the volume + bool can_delete = 3; } message VolumeMarkReadonlyResponse { } @@ -604,6 +606,7 @@ message VolumeInfo { uint64 expire_at_sec = 6; // expiration time of ec volume bool read_only = 7; EcShardConfig ec_shard_config = 8; // EC shard configuration (optional, null = use default 10+4) + bool read_only_can_delete = 9; // with read_only: writes are rejected but deletes still land } // EcShardConfig specifies erasure coding shard configuration diff --git a/weed/pb/volume_server_pb/volume_server.pb.go b/weed/pb/volume_server_pb/volume_server.pb.go index 056d41a29..368d16e4b 100644 --- a/weed/pb/volume_server_pb/volume_server.pb.go +++ b/weed/pb/volume_server_pb/volume_server.pb.go @@ -1549,9 +1549,11 @@ func (*VolumeDeleteResponse) Descriptor() ([]byte, []int) { } type VolumeMarkReadonlyRequest struct { - state protoimpl.MessageState `protogen:"open.v1"` - VolumeId uint32 `protobuf:"varint,1,opt,name=volume_id,json=volumeId,proto3" json:"volume_id,omitempty"` - Persist bool `protobuf:"varint,2,opt,name=persist,proto3" json:"persist,omitempty"` + state protoimpl.MessageState `protogen:"open.v1"` + VolumeId uint32 `protobuf:"varint,1,opt,name=volume_id,json=volumeId,proto3" json:"volume_id,omitempty"` + Persist bool `protobuf:"varint,2,opt,name=persist,proto3" json:"persist,omitempty"` + // reject writes but keep accepting deletes, so expiring data can drain the volume + CanDelete bool `protobuf:"varint,3,opt,name=can_delete,json=canDelete,proto3" json:"can_delete,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -1600,6 +1602,13 @@ func (x *VolumeMarkReadonlyRequest) GetPersist() bool { return false } +func (x *VolumeMarkReadonlyRequest) GetCanDelete() bool { + if x != nil { + return x.CanDelete + } + return false +} + type VolumeMarkReadonlyResponse struct { state protoimpl.MessageState `protogen:"open.v1"` unknownFields protoimpl.UnknownFields @@ -5012,17 +5021,18 @@ func (x *RemoteFile) GetExtension() string { } type VolumeInfo struct { - state protoimpl.MessageState `protogen:"open.v1"` - Files []*RemoteFile `protobuf:"bytes,1,rep,name=files,proto3" json:"files,omitempty"` - Version uint32 `protobuf:"varint,2,opt,name=version,proto3" json:"version,omitempty"` - Replication string `protobuf:"bytes,3,opt,name=replication,proto3" json:"replication,omitempty"` - BytesOffset uint32 `protobuf:"varint,4,opt,name=bytes_offset,json=bytesOffset,proto3" json:"bytes_offset,omitempty"` - DatFileSize int64 `protobuf:"varint,5,opt,name=dat_file_size,json=datFileSize,proto3" json:"dat_file_size,omitempty"` // store the original dat file size - ExpireAtSec uint64 `protobuf:"varint,6,opt,name=expire_at_sec,json=expireAtSec,proto3" json:"expire_at_sec,omitempty"` // expiration time of ec volume - ReadOnly bool `protobuf:"varint,7,opt,name=read_only,json=readOnly,proto3" json:"read_only,omitempty"` - EcShardConfig *EcShardConfig `protobuf:"bytes,8,opt,name=ec_shard_config,json=ecShardConfig,proto3" json:"ec_shard_config,omitempty"` // EC shard configuration (optional, null = use default 10+4) - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + state protoimpl.MessageState `protogen:"open.v1"` + Files []*RemoteFile `protobuf:"bytes,1,rep,name=files,proto3" json:"files,omitempty"` + Version uint32 `protobuf:"varint,2,opt,name=version,proto3" json:"version,omitempty"` + Replication string `protobuf:"bytes,3,opt,name=replication,proto3" json:"replication,omitempty"` + BytesOffset uint32 `protobuf:"varint,4,opt,name=bytes_offset,json=bytesOffset,proto3" json:"bytes_offset,omitempty"` + DatFileSize int64 `protobuf:"varint,5,opt,name=dat_file_size,json=datFileSize,proto3" json:"dat_file_size,omitempty"` // store the original dat file size + ExpireAtSec uint64 `protobuf:"varint,6,opt,name=expire_at_sec,json=expireAtSec,proto3" json:"expire_at_sec,omitempty"` // expiration time of ec volume + ReadOnly bool `protobuf:"varint,7,opt,name=read_only,json=readOnly,proto3" json:"read_only,omitempty"` + EcShardConfig *EcShardConfig `protobuf:"bytes,8,opt,name=ec_shard_config,json=ecShardConfig,proto3" json:"ec_shard_config,omitempty"` // EC shard configuration (optional, null = use default 10+4) + ReadOnlyCanDelete bool `protobuf:"varint,9,opt,name=read_only_can_delete,json=readOnlyCanDelete,proto3" json:"read_only_can_delete,omitempty"` // with read_only: writes are rejected but deletes still land + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *VolumeInfo) Reset() { @@ -5111,6 +5121,13 @@ func (x *VolumeInfo) GetEcShardConfig() *EcShardConfig { return nil } +func (x *VolumeInfo) GetReadOnlyCanDelete() bool { + if x != nil { + return x.ReadOnlyCanDelete + } + return false +} + // EcShardConfig specifies erasure coding shard configuration type EcShardConfig struct { state protoimpl.MessageState `protogen:"open.v1"` @@ -7258,10 +7275,12 @@ const file_volume_server_proto_rawDesc = "" + "\n" + "only_empty\x18\x02 \x01(\bR\tonlyEmpty\x12(\n" + "\x10keep_remote_data\x18\x03 \x01(\bR\x0ekeepRemoteData\"\x16\n" + - "\x14VolumeDeleteResponse\"R\n" + + "\x14VolumeDeleteResponse\"q\n" + "\x19VolumeMarkReadonlyRequest\x12\x1b\n" + "\tvolume_id\x18\x01 \x01(\rR\bvolumeId\x12\x18\n" + - "\apersist\x18\x02 \x01(\bR\apersist\"\x1c\n" + + "\apersist\x18\x02 \x01(\bR\apersist\x12\x1d\n" + + "\n" + + "can_delete\x18\x03 \x01(\bR\tcanDelete\"\x1c\n" + "\x1aVolumeMarkReadonlyResponse\"8\n" + "\x19VolumeMarkWritableRequest\x12\x1b\n" + "\tvolume_id\x18\x01 \x01(\rR\bvolumeId\"\x1c\n" + @@ -7540,7 +7559,7 @@ const file_volume_server_proto_rawDesc = "" + "\x06offset\x18\x04 \x01(\x04R\x06offset\x12\x1b\n" + "\tfile_size\x18\x05 \x01(\x04R\bfileSize\x12#\n" + "\rmodified_time\x18\x06 \x01(\x04R\fmodifiedTime\x12\x1c\n" + - "\textension\x18\a \x01(\tR\textension\"\xcd\x02\n" + + "\textension\x18\a \x01(\tR\textension\"\xfe\x02\n" + "\n" + "VolumeInfo\x122\n" + "\x05files\x18\x01 \x03(\v2\x1c.volume_server_pb.RemoteFileR\x05files\x12\x18\n" + @@ -7550,7 +7569,8 @@ const file_volume_server_proto_rawDesc = "" + "\rdat_file_size\x18\x05 \x01(\x03R\vdatFileSize\x12\"\n" + "\rexpire_at_sec\x18\x06 \x01(\x04R\vexpireAtSec\x12\x1b\n" + "\tread_only\x18\a \x01(\bR\breadOnly\x12G\n" + - "\x0fec_shard_config\x18\b \x01(\v2\x1f.volume_server_pb.EcShardConfigR\recShardConfig\"w\n" + + "\x0fec_shard_config\x18\b \x01(\v2\x1f.volume_server_pb.EcShardConfigR\recShardConfig\x12/\n" + + "\x14read_only_can_delete\x18\t \x01(\bR\x11readOnlyCanDelete\"w\n" + "\rEcShardConfig\x12\x1f\n" + "\vdata_shards\x18\x01 \x01(\rR\n" + "dataShards\x12#\n" + diff --git a/weed/server/volume_grpc_admin.go b/weed/server/volume_grpc_admin.go index 31b43cd2c..f4d330508 100644 --- a/weed/server/volume_grpc_admin.go +++ b/weed/server/volume_grpc_admin.go @@ -256,7 +256,7 @@ func (vs *VolumeServer) VolumeConfigure(ctx context.Context, req *volume_server_ } -func (vs *VolumeServer) makeVolumeReadonly(ctx context.Context, v *storage.Volume, persist bool) error { +func (vs *VolumeServer) makeVolumeReadonly(ctx context.Context, v *storage.Volume, canDelete bool, persist bool) error { if err := vs.CheckMaintenanceMode(); err != nil { return err } @@ -269,7 +269,7 @@ func (vs *VolumeServer) makeVolumeReadonly(ctx context.Context, v *storage.Volum // rare case 1.5: it will be unlucky if heartbeat happened between step 1 and 2. // step 2: mark local volume as readonly - if err := vs.store.MarkVolumeReadonly(v.Id, persist); err != nil { + if err := vs.store.MarkVolumeReadonly(v.Id, canDelete, persist); err != nil { glog.Errorf("mark volume %d readonly: %v", v.Id, err) return err } else { @@ -357,7 +357,7 @@ func (vs *VolumeServer) VolumeMarkReadonly(ctx context.Context, req *volume_serv return resp, fmt.Errorf("volume %d not found", req.VolumeId) } - if err := vs.makeVolumeReadonly(ctx, v, req.GetPersist()); err != nil { + if err := vs.makeVolumeReadonly(ctx, v, req.GetCanDelete(), req.GetPersist()); err != nil { return resp, err } diff --git a/weed/server/volume_grpc_scrub.go b/weed/server/volume_grpc_scrub.go index 412f0bbd2..c2e7a3bcf 100644 --- a/weed/server/volume_grpc_scrub.go +++ b/weed/server/volume_grpc_scrub.go @@ -66,7 +66,7 @@ func (vs *VolumeServer) ScrubVolume(ctx context.Context, req *volume_server_pb.S errs := []error{} if req.GetMarkBrokenVolumesReadonly() { for _, v := range brokenVolumes { - if err := vs.makeVolumeReadonly(ctx, v, true); err != nil { + if err := vs.makeVolumeReadonly(ctx, v, false, true); err != nil { errs = append(errs, err) details = append(details, err.Error()) } else { diff --git a/weed/shell/command_fs_merge_volumes.go b/weed/shell/command_fs_merge_volumes.go index 88953b285..ef82a6ee6 100644 --- a/weed/shell/command_fs_merge_volumes.go +++ b/weed/shell/command_fs_merge_volumes.go @@ -8,6 +8,7 @@ import ( "io" "net/http" "sort" + "strconv" "strings" "sync" "time" @@ -47,7 +48,11 @@ func (c *commandFsMergeVolumes) Help() string { This would help clear half-full volumes and let vacuum system to delete them later. - fs.mergeVolumes [-toVolumeId=y] [-fromVolumeId=x] [-collection="*"] [-dir=/] [-apply] + fs.mergeVolumes [-toVolumeId=y[,z]] [-fromVolumeId=x] [-collection="*"] [-dir=/] [-apply] + + -toVolumeId accepts a comma-separated list. With -fromVolumeId, the source + chunks are distributed across the listed volumes by remaining capacity, so a + volume that does not fit into any single target can still be cleared. ` } @@ -60,7 +65,7 @@ func (c *commandFsMergeVolumes) Do(args []string, commandEnv *CommandEnv, writer fsMergeVolumesCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError) dirArg := fsMergeVolumesCommand.String("dir", "/", "base directory to find and update files") fromVolumeArg := fsMergeVolumesCommand.Uint("fromVolumeId", 0, "move chunks with this volume id") - toVolumeArg := fsMergeVolumesCommand.Uint("toVolumeId", 0, "change chunks to this volume id") + toVolumeArg := fsMergeVolumesCommand.String("toVolumeId", "", "change chunks to this volume id, or distribute across a comma-separated list of volume ids") collectionArg := fsMergeVolumesCommand.String("collection", "*", "Name of collection to merge") apply := fsMergeVolumesCommand.Bool("apply", false, "applying the metadata changes") if err = fsMergeVolumesCommand.Parse(args); err != nil { @@ -79,12 +84,12 @@ func (c *commandFsMergeVolumes) Do(args []string, commandEnv *CommandEnv, writer if *fromVolumeArg > maxVolumeID { return fmt.Errorf("fromVolumeId %d exceeds max volume id %d", *fromVolumeArg, maxVolumeID) } - if *toVolumeArg > maxVolumeID { - return fmt.Errorf("toVolumeId %d exceeds max volume id %d", *toVolumeArg, maxVolumeID) - } fromVolumeId := needle.VolumeId(*fromVolumeArg) - toVolumeId := needle.VolumeId(*toVolumeArg) + toVolumeIds, err := parseTargetVolumeIds(*toVolumeArg) + if err != nil { + return err + } if err = c.reloadVolumesInfo(commandEnv.MasterClient); err != nil { return fmt.Errorf("reload volumes info: %w", err) @@ -99,43 +104,20 @@ func (c *commandFsMergeVolumes) Do(args []string, commandEnv *CommandEnv, writer return fmt.Errorf("fromVolumeId %d not found on master", fromVolumeId) } } - if toVolumeId != 0 { + for _, toVolumeId := range toVolumeIds { if _, err := c.getVolumeInfoById(toVolumeId); err != nil { return fmt.Errorf("toVolumeId %d not found on master", toVolumeId) } } - if fromVolumeId != 0 && toVolumeId != 0 { - if fromVolumeId == toVolumeId { - return fmt.Errorf("no volume id changes, %d == %d", fromVolumeId, toVolumeId) - } - compatible, err := c.volumesAreCompatible(fromVolumeId, toVolumeId) - if err != nil { - return fmt.Errorf("cannot determine volumes are compatible: %d and %d", fromVolumeId, toVolumeId) - } - if !compatible { - return fmt.Errorf("volume %d is not compatible with volume %d", fromVolumeId, toVolumeId) - } - fromSize := c.getVolumeSizeById(fromVolumeId) - toSize := c.getVolumeSizeById(toVolumeId) - if fromSize+toSize > c.volumeSizeLimit { - return fmt.Errorf( - "volume %d (%d MB) cannot merge into volume %d (%d MB_ due to volume size limit (%d MB)", - fromVolumeId, fromSize/1024/1024, - toVolumeId, toSize/1024/1024, - c.volumeSizeLimit/1024/1024, - ) - } - } - - plan, err := c.createMergePlan(*collectionArg, toVolumeId, fromVolumeId) + plan, err := c.createMergePlan(*collectionArg, toVolumeIds, fromVolumeId) if err != nil { return err } c.printPlan(plan) - if len(plan) == 0 { + if len(plan.targets) == 0 { return nil } @@ -204,19 +186,24 @@ func (c *commandFsMergeVolumes) Do(args []string, commandEnv *CommandEnv, writer } chunkVolumeId := needle.VolumeId(chunk.Fid.VolumeId) - toVolumeId, found := plan[chunkVolumeId] - if !found { + if !plan.isSource(chunkVolumeId) { continue } oldFid := chunk.GetFileIdString() oldVid := chunk.Fid.VolumeId - fmt.Printf("move %s(%s)\n", entryPath, oldFid) + toVolumeId, ok := plan.allocate(chunkVolumeId, chunk.Size) + if !ok { + fmt.Printf("skip %s(%s): no target volume has room\n", entryPath, oldFid) + continue + } + fmt.Printf("move %s(%s) => volume %d\n", entryPath, oldFid, toVolumeId) if !*apply { continue } if mvErr := moveChunk(chunk, toVolumeId, commandEnv.MasterClient); mvErr != nil { fmt.Printf("failed to move %s(%s): %v\n", entryPath, oldFid, mvErr) + plan.release(chunkVolumeId, toVolumeId, chunk.Size) continue } entryChanged = true @@ -357,16 +344,107 @@ func (c *commandFsMergeVolumes) reloadVolumesInfo(masterClient *wdclient.MasterC }) } -func (c *commandFsMergeVolumes) createMergePlan(collection string, toVolumeId needle.VolumeId, fromVolumeId needle.VolumeId) (map[needle.VolumeId]needle.VolumeId, error) { +// mergePlan maps each source volume to candidate targets: a single-target +// source sends every chunk there (historic behavior), a multi-target source +// allocates per chunk under mu since TraverseBfs callbacks run in parallel. +type mergePlan struct { + mu sync.Mutex + targets map[needle.VolumeId][]needle.VolumeId + plannedSize map[needle.VolumeId]uint64 + volumeSizeLimit uint64 +} + +func newMergePlan(volumeSizeLimit uint64) *mergePlan { + return &mergePlan{ + targets: make(map[needle.VolumeId][]needle.VolumeId), + plannedSize: make(map[needle.VolumeId]uint64), + volumeSizeLimit: volumeSizeLimit, + } +} + +func (p *mergePlan) isSource(vid needle.VolumeId) bool { + _, found := p.targets[vid] + return found +} + +// allocate picks the candidate with the most remaining capacity that still +// fits the chunk, and reserves the chunk size against it. +func (p *mergePlan) allocate(src needle.VolumeId, size uint64) (needle.VolumeId, bool) { + p.mu.Lock() + defer p.mu.Unlock() + candidates := p.targets[src] + if len(candidates) == 0 { + return 0, false + } + if len(candidates) == 1 { + return candidates[0], true + } + var best needle.VolumeId + var bestRemaining uint64 + found := false + for _, t := range candidates { + used := p.plannedSize[t] + if used+size > p.volumeSizeLimit { + continue + } + if remaining := p.volumeSizeLimit - used; !found || remaining > bestRemaining { + best, bestRemaining, found = t, remaining, true + } + } + if !found { + return 0, false + } + p.plannedSize[best] += size + return best, true +} + +// release returns a failed move's reservation so later chunks can use it. +// Single-target sources reserve at plan time, not per chunk. +func (p *mergePlan) release(src, target needle.VolumeId, size uint64) { + p.mu.Lock() + defer p.mu.Unlock() + if len(p.targets[src]) <= 1 { + return + } + if p.plannedSize[target] >= size { + p.plannedSize[target] -= size + } +} + +// Empty or "0" means unset, matching the old numeric flag's default. +func parseTargetVolumeIds(arg string) ([]needle.VolumeId, error) { + arg = strings.TrimSpace(arg) + if arg == "" || arg == "0" { + return nil, nil + } + var ids []needle.VolumeId + seen := make(map[needle.VolumeId]bool) + for _, part := range strings.Split(arg, ",") { + part = strings.TrimSpace(part) + v, err := strconv.ParseUint(part, 10, 32) + if err != nil || v == 0 { + return nil, fmt.Errorf("invalid toVolumeId %q", part) + } + vid := needle.VolumeId(v) + if seen[vid] { + return nil, fmt.Errorf("duplicate toVolumeId %d", vid) + } + seen[vid] = true + ids = append(ids, vid) + } + return ids, nil +} + +func (c *commandFsMergeVolumes) createMergePlan(collection string, toVolumeIds []needle.VolumeId, fromVolumeId needle.VolumeId) (*mergePlan, error) { // When the user names both endpoints, honor that exact direction. The // heuristic below only ever merges a smaller volume into a larger one, so // an explicit "merge larger into smaller" request would otherwise yield an // empty plan and silently do nothing. - if fromVolumeId != 0 && toVolumeId != 0 { - return c.createDirectedMergePlan(collection, fromVolumeId, toVolumeId) + if fromVolumeId != 0 && len(toVolumeIds) > 0 { + return c.createDirectedMergePlan(collection, fromVolumeId, toVolumeIds) } - plan := make(map[needle.VolumeId]needle.VolumeId) + plan := newMergePlan(c.volumeSizeLimit) volumeIds := maps.Keys(c.volumes) sort.Slice(volumeIds, func(a, b int) bool { return c.volumes[volumeIds[b]].Size < c.volumes[volumeIds[a]].Size @@ -377,7 +455,7 @@ func (c *commandFsMergeVolumes) createMergePlan(collection string, toVolumeId ne volume := c.volumes[volumeIds[i]] if volume.GetReadOnly() || c.getVolumeSize(volume) == 0 || (collection != "*" && collection != volume.GetCollection()) { - if fromVolumeId != 0 && volumeIds[i] == fromVolumeId || toVolumeId != 0 && volumeIds[i] == toVolumeId { + if fromVolumeId != 0 && volumeIds[i] == fromVolumeId || slices.Contains(toVolumeIds, volumeIds[i]) { if volume.GetReadOnly() { return nil, fmt.Errorf("volume %d is readonly", volumeIds[i]) } @@ -397,10 +475,10 @@ func (c *commandFsMergeVolumes) createMergePlan(collection string, toVolumeId ne } for j := 0; j < i; j++ { candidate := volumeIds[j] - if toVolumeId != 0 && candidate != toVolumeId { + if len(toVolumeIds) > 0 && !slices.Contains(toVolumeIds, candidate) { continue } - if _, moving := plan[candidate]; moving { + if _, moving := plan.targets[candidate]; moving { continue } compatible, err := c.volumesAreCompatible(src, candidate) @@ -411,7 +489,10 @@ func (c *commandFsMergeVolumes) createMergePlan(collection string, toVolumeId ne fmt.Printf("volume %d is not compatible with volume %d\n", src, candidate) continue } - candidatePlannedSize := c.getVolumeSizeBasedOnPlan(plan, candidate) + if _, tracked := plan.plannedSize[candidate]; !tracked { + plan.plannedSize[candidate] = c.getVolumeSizeById(candidate) + } + candidatePlannedSize := plan.plannedSize[candidate] if candidatePlannedSize+c.getVolumeSizeById(src) > c.volumeSizeLimit { fmt.Printf("volume %d (%d MB) merge into volume %d (%d MB, %d MB with plan) exceeds volume size limit (%d MB)\n", src, c.getVolumeSizeById(src)/1024/1024, @@ -419,7 +500,8 @@ func (c *commandFsMergeVolumes) createMergePlan(collection string, toVolumeId ne c.volumeSizeLimit/1024/1024) continue } - plan[src] = candidate + plan.targets[src] = []needle.VolumeId{candidate} + plan.plannedSize[candidate] += c.getVolumeSizeById(src) break } } @@ -427,16 +509,13 @@ func (c *commandFsMergeVolumes) createMergePlan(collection string, toVolumeId ne return plan, nil } -// createDirectedMergePlan builds the single-pair plan {from: to} exactly as the -// user requested, skipping the smaller-into-larger ordering the heuristic -// planner uses. Compatibility and the combined size limit are already checked -// by Do before this runs; here we only reject endpoints that cannot -// participate (read-only, empty, or outside the requested collection). -func (c *commandFsMergeVolumes) createDirectedMergePlan(collection string, from, to needle.VolumeId) (map[needle.VolumeId]needle.VolumeId, error) { - if from == to { - return nil, fmt.Errorf("no volume id changes, %d == %d", from, to) +// createDirectedMergePlan honors the exact direction the user named, skipping +// the heuristic planner's smaller-into-larger ordering. +func (c *commandFsMergeVolumes) createDirectedMergePlan(collection string, from needle.VolumeId, toIds []needle.VolumeId) (*mergePlan, error) { + if slices.Contains(toIds, from) { + return nil, fmt.Errorf("no volume id changes, %d is both source and target", from) } - for _, vid := range []needle.VolumeId{from, to} { + for _, vid := range append([]needle.VolumeId{from}, toIds...) { volume, err := c.getVolumeInfoById(vid) if err != nil { return nil, err @@ -452,18 +531,37 @@ func (c *commandFsMergeVolumes) createDirectedMergePlan(collection string, from, if vid == from && c.getVolumeSize(volume) == 0 { return nil, fmt.Errorf("volume %d is empty", vid) } - } - return map[needle.VolumeId]needle.VolumeId{from: to}, nil -} - -func (c *commandFsMergeVolumes) getVolumeSizeBasedOnPlan(plan map[needle.VolumeId]needle.VolumeId, vid needle.VolumeId) uint64 { - size := c.getVolumeSizeById(vid) - for src, dest := range plan { - if dest == vid { - size += c.getVolumeSizeById(src) + if vid != from { + compatible, err := c.volumesAreCompatible(from, vid) + if err != nil { + return nil, err + } + if !compatible { + return nil, fmt.Errorf("volume %d is not compatible with volume %d", from, vid) + } } } - return size + + plan := newMergePlan(c.volumeSizeLimit) + fromSize := c.getVolumeSizeById(from) + var totalFree uint64 + for _, to := range toIds { + toSize := c.getVolumeSizeById(to) + plan.plannedSize[to] = toSize + if toSize < c.volumeSizeLimit { + totalFree += c.volumeSizeLimit - toSize + } + } + if fromSize > totalFree { + return nil, fmt.Errorf( + "volume %d (%d MB) cannot merge into volumes %v (%d MB free) due to volume size limit (%d MB)", + from, fromSize/1024/1024, + toIds, totalFree/1024/1024, + c.volumeSizeLimit/1024/1024, + ) + } + plan.targets[from] = toIds + return plan, nil } // getVolumeSize is the volume's live data size, clamped since @@ -479,11 +577,16 @@ func (c *commandFsMergeVolumes) getVolumeSizeById(vid needle.VolumeId) uint64 { return c.getVolumeSize(c.volumes[vid]) } -func (c *commandFsMergeVolumes) printPlan(plan map[needle.VolumeId]needle.VolumeId) { +func (c *commandFsMergeVolumes) printPlan(plan *mergePlan) { fmt.Printf("max volume size: %d MB\n", c.volumeSizeLimit/1024/1024) reversePlan := make(map[needle.VolumeId][]needle.VolumeId) - for src, dest := range plan { - reversePlan[dest] = append(reversePlan[dest], src) + for src, dests := range plan.targets { + if len(dests) > 1 { + fmt.Printf("volume %d (%d MB) distribute across volumes %v by remaining capacity\n", + src, c.getVolumeSizeById(src)/1024/1024, dests) + continue + } + reversePlan[dests[0]] = append(reversePlan[dests[0]], src) } for dest, srcs := range reversePlan { currentSize := c.getVolumeSizeById(dest) @@ -518,7 +621,7 @@ func (c *commandFsMergeVolumes) rewriteManifestChunk( ctx context.Context, commandEnv *CommandEnv, lookupFn wdclient.LookupFileIdFunctionType, - plan map[needle.VolumeId]needle.VolumeId, + plan *mergePlan, entryPath util.FullPath, chunk *filer_pb.FileChunk, apply bool, @@ -555,19 +658,24 @@ func (c *commandFsMergeVolumes) rewriteManifestChunk( continue } subVid := needle.VolumeId(sub.Fid.VolumeId) - toVid, ok := plan[subVid] - if !ok { + if !plan.isSource(subVid) { continue } oldSubFid := sub.GetFileIdString() oldSubVid := sub.Fid.VolumeId - fmt.Printf("move %s(%s) [inside manifest %s]\n", entryPath, oldSubFid, chunk.GetFileIdString()) + toVid, ok := plan.allocate(subVid, sub.Size) + if !ok { + fmt.Printf("skip %s(%s) [inside manifest %s]: no target volume has room\n", entryPath, oldSubFid, chunk.GetFileIdString()) + continue + } + fmt.Printf("move %s(%s) => volume %d [inside manifest %s]\n", entryPath, oldSubFid, toVid, chunk.GetFileIdString()) if !apply { anySubChanged = true continue } if mErr := moveChunk(sub, toVid, commandEnv.MasterClient); mErr != nil { fmt.Printf("failed to move %s(%s): %v\n", entryPath, oldSubFid, mErr) + plan.release(subVid, toVid, sub.Size) continue } anySubChanged = true @@ -575,7 +683,7 @@ func (c *commandFsMergeVolumes) rewriteManifestChunk( } manifestVid := needle.VolumeId(chunk.Fid.VolumeId) - _, manifestMustMove := plan[manifestVid] + manifestMustMove := plan.isSource(manifestVid) if !anySubChanged && !manifestMustMove { return chunk, false, nil, nil @@ -626,7 +734,7 @@ func (c *commandFsMergeVolumes) uploadManifestChunk( commandEnv *CommandEnv, entryPath util.FullPath, collection string, - plan map[needle.VolumeId]needle.VolumeId, + plan *mergePlan, data []byte, ) (*filer_pb.FileChunk, error) { const manifestAssignAttempts = 10 @@ -653,7 +761,7 @@ func (c *commandFsMergeVolumes) uploadManifestChunk( if parseErr != nil { return fmt.Errorf("parse assigned fid %q: %w", resp.FileId, parseErr) } - if _, isSource := plan[needle.VolumeId(fid.VolumeId)]; !isSource { + if !plan.isSource(needle.VolumeId(fid.VolumeId)) { assignResp = resp return nil } diff --git a/weed/shell/command_fs_merge_volumes_test.go b/weed/shell/command_fs_merge_volumes_test.go index 21a71a377..859ba80b4 100644 --- a/weed/shell/command_fs_merge_volumes_test.go +++ b/weed/shell/command_fs_merge_volumes_test.go @@ -28,21 +28,21 @@ func TestCreateMergePlan_HonorsExplicitDirection(t *testing.T) { smaller := &master_pb.VolumeInformationMessage{Id: 83, Size: 7088822248} c := newMergeCmd(250000, larger, smaller) - plan, err := c.createMergePlan("*", needle.VolumeId(83), needle.VolumeId(87)) + plan, err := c.createMergePlan("*", []needle.VolumeId{83}, needle.VolumeId(87)) if err != nil { t.Fatalf("unexpected error: %v", err) } - if got := plan[needle.VolumeId(87)]; got != needle.VolumeId(83) { - t.Fatalf("expected 87->83, got plan=%v", plan) + if got := plan.targets[needle.VolumeId(87)]; len(got) != 1 || got[0] != needle.VolumeId(83) { + t.Fatalf("expected 87->83, got plan=%v", plan.targets) } // The reverse direction keeps working too. - plan, err = c.createMergePlan("*", needle.VolumeId(87), needle.VolumeId(83)) + plan, err = c.createMergePlan("*", []needle.VolumeId{87}, needle.VolumeId(83)) if err != nil { t.Fatalf("unexpected error: %v", err) } - if got := plan[needle.VolumeId(83)]; got != needle.VolumeId(87) { - t.Fatalf("expected 83->87, got plan=%v", plan) + if got := plan.targets[needle.VolumeId(83)]; len(got) != 1 || got[0] != needle.VolumeId(87) { + t.Fatalf("expected 83->87, got plan=%v", plan.targets) } } @@ -53,12 +53,12 @@ func TestCreateMergePlan_DirectedAllowsEmptyTarget(t *testing.T) { emptyTo := &master_pb.VolumeInformationMessage{Id: 83, Size: 0} c := newMergeCmd(250000, from, emptyTo) - plan, err := c.createMergePlan("*", needle.VolumeId(83), needle.VolumeId(87)) + plan, err := c.createMergePlan("*", []needle.VolumeId{83}, needle.VolumeId(87)) if err != nil { t.Fatalf("unexpected error merging into empty target: %v", err) } - if got := plan[needle.VolumeId(87)]; got != needle.VolumeId(83) { - t.Fatalf("expected 87->83, got plan=%v", plan) + if got := plan.targets[needle.VolumeId(87)]; len(got) != 1 || got[0] != needle.VolumeId(83) { + t.Fatalf("expected 87->83, got plan=%v", plan.targets) } } @@ -68,7 +68,7 @@ func TestCreateMergePlan_DirectedRejectsSelfMap(t *testing.T) { v := &master_pb.VolumeInformationMessage{Id: 87, Size: 100} c := newMergeCmd(250000, v) - _, err := c.createMergePlan("*", needle.VolumeId(87), needle.VolumeId(87)) + _, err := c.createMergePlan("*", []needle.VolumeId{87}, needle.VolumeId(87)) if err == nil || !strings.Contains(err.Error(), "no volume id changes") { t.Fatalf("expected self-map rejection, got %v", err) } @@ -113,7 +113,7 @@ func TestCreateMergePlan_DirectedRejectsIneligible(t *testing.T) { for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { c := newMergeCmd(250000, tc.from, tc.to) - _, err := c.createMergePlan(tc.coll, needle.VolumeId(tc.to.Id), needle.VolumeId(tc.from.Id)) + _, err := c.createMergePlan(tc.coll, []needle.VolumeId{needle.VolumeId(tc.to.Id)}, needle.VolumeId(tc.from.Id)) if err == nil { t.Fatalf("expected error %q, got nil", tc.wantErr) } @@ -124,6 +124,114 @@ func TestCreateMergePlan_DirectedRejectsIneligible(t *testing.T) { } } +// A source that fits into no single target distributes across several, each +// chunk going to the target with the most remaining capacity. +func TestCreateMergePlan_DistributesAcrossMultipleTargets(t *testing.T) { + mb := uint64(1024 * 1024) + from := &master_pb.VolumeInformationMessage{Id: 112, Size: 60 * mb} + to1 := &master_pb.VolumeInformationMessage{Id: 111, Size: 60 * mb} + to2 := &master_pb.VolumeInformationMessage{Id: 107, Size: 70 * mb} + c := newMergeCmd(100, from, to1, to2) + + plan, err := c.createMergePlan("*", []needle.VolumeId{111, 107}, needle.VolumeId(112)) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if got := plan.targets[needle.VolumeId(112)]; len(got) != 2 { + t.Fatalf("expected two targets, got %v", got) + } + + // 111 has 40 MB free, 107 has 30 MB free. The first chunk goes to the + // emptier 111; allocations keep balancing remaining capacity after that. + vid, ok := plan.allocate(needle.VolumeId(112), 5*mb) + if !ok || vid != needle.VolumeId(111) { + t.Fatalf("expected first chunk on 111, got %v ok=%v", vid, ok) + } + vid, ok = plan.allocate(needle.VolumeId(112), 10*mb) + if !ok || vid != needle.VolumeId(111) { + t.Fatalf("expected second chunk on 111 (35 MB free vs 30), got %v ok=%v", vid, ok) + } + vid, ok = plan.allocate(needle.VolumeId(112), 5*mb) + if !ok || vid != needle.VolumeId(107) { + t.Fatalf("expected third chunk on 107 (30 MB free vs 25), got %v ok=%v", vid, ok) + } + // A chunk larger than any remaining capacity is refused. + if vid, ok = plan.allocate(needle.VolumeId(112), 45*mb); ok { + t.Fatalf("expected no room for oversized chunk, got %v", vid) + } + // Smaller chunks still fit afterwards. + if _, ok = plan.allocate(needle.VolumeId(112), 20*mb); !ok { + t.Fatal("expected room for 20 MB chunk") + } + + // A failed move's release makes the reservation reusable. + vid2, ok := plan.allocate(needle.VolumeId(112), 20*mb) + if !ok { + t.Fatal("expected room for second 20 MB chunk") + } + plan.release(needle.VolumeId(112), vid2, 20*mb) + if vid3, ok := plan.allocate(needle.VolumeId(112), 20*mb); !ok || vid3 != vid2 { + t.Fatalf("released capacity must be reusable, got %v ok=%v", vid3, ok) + } +} + +// The combined free capacity of all targets must cover the source's live data. +func TestCreateMergePlan_MultiTargetRejectsInsufficientCapacity(t *testing.T) { + mb := uint64(1024 * 1024) + from := &master_pb.VolumeInformationMessage{Id: 112, Size: 90 * mb} + to1 := &master_pb.VolumeInformationMessage{Id: 111, Size: 60 * mb} + to2 := &master_pb.VolumeInformationMessage{Id: 107, Size: 60 * mb} + c := newMergeCmd(100, from, to1, to2) + + _, err := c.createMergePlan("*", []needle.VolumeId{111, 107}, needle.VolumeId(112)) + if err == nil || !strings.Contains(err.Error(), "cannot merge into volumes") { + t.Fatalf("expected capacity rejection, got %v", err) + } +} + +// Without -fromVolumeId, a -toVolumeId list restricts the heuristic planner's +// target candidates to the listed volumes. +func TestCreateMergePlan_TargetListRestrictsHeuristic(t *testing.T) { + mb := uint64(1024 * 1024) + big := &master_pb.VolumeInformationMessage{Id: 1, Size: 80 * mb} + mid := &master_pb.VolumeInformationMessage{Id: 2, Size: 50 * mb} + small := &master_pb.VolumeInformationMessage{Id: 3, Size: 10 * mb} + c := newMergeCmd(100, big, mid, small) + + plan, err := c.createMergePlan("*", []needle.VolumeId{2}, 0) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if got := plan.targets[needle.VolumeId(3)]; len(got) != 1 || got[0] != needle.VolumeId(2) { + t.Fatalf("expected 3->2, got %v", plan.targets) + } + if plan.isSource(needle.VolumeId(1)) { + t.Fatalf("volume 1 must not merge anywhere, got %v", plan.targets) + } +} + +func TestParseTargetVolumeIds(t *testing.T) { + if ids, err := parseTargetVolumeIds(""); err != nil || ids != nil { + t.Fatalf("empty arg: got %v, %v", ids, err) + } + if ids, err := parseTargetVolumeIds("0"); err != nil || ids != nil { + t.Fatalf("zero arg: got %v, %v", ids, err) + } + ids, err := parseTargetVolumeIds("111, 107") + if err != nil || len(ids) != 2 || ids[0] != needle.VolumeId(111) || ids[1] != needle.VolumeId(107) { + t.Fatalf("list arg: got %v, %v", ids, err) + } + if _, err = parseTargetVolumeIds("111,111"); err == nil { + t.Fatal("expected duplicate rejection") + } + if _, err = parseTargetVolumeIds("111,x"); err == nil { + t.Fatal("expected invalid id rejection") + } + if _, err = parseTargetVolumeIds("4294967297"); err == nil { + t.Fatal("expected uint32 overflow rejection") + } +} + // A volume reporting more deleted bytes than it holds must read as empty. func TestGetVolumeSize_ClampsDeletedOverSize(t *testing.T) { c := newMergeCmd(250000) diff --git a/weed/shell/command_volume_mark.go b/weed/shell/command_volume_mark.go index 80e0e90fb..f8c630af8 100644 --- a/weed/shell/command_volume_mark.go +++ b/weed/shell/command_volume_mark.go @@ -27,10 +27,14 @@ func (c *commandVolumeMark) Name() string { func (c *commandVolumeMark) Help() string { return `Mark volume writable or readonly from one volume server, or all volume replicas in one collection - volume.mark -node -volumeId -writable or -readonly - volume.mark -collection -writable or -readonly + volume.mark -node -volumeId -writable, -readonly or -readonlyCanDelete + volume.mark -collection -writable, -readonly or -readonlyCanDelete Use -collection ` + CollectionDefault + ` to target volumes that belong to no named collection. + + -readonlyCanDelete rejects new writes but keeps accepting deletes, so a + volume holding expiring data drains over time until it can be vacuumed + away or merged into other volumes with fs.mergeVolumes. ` } @@ -46,6 +50,7 @@ func (c *commandVolumeMark) Do(args []string, commandEnv *CommandEnv, writer io. collection := volMarkCommand.String("collection", "", "the collection name") writable := volMarkCommand.Bool("writable", false, "volume mark writable") readonly := volMarkCommand.Bool("readonly", false, "volume mark readonly") + readonlyCanDelete := volMarkCommand.Bool("readonlyCanDelete", false, "volume mark readonly but still accepting deletes") if err = volMarkCommand.Parse(args); err != nil { return nil } @@ -60,12 +65,16 @@ func (c *commandVolumeMark) Do(args []string, commandEnv *CommandEnv, writer io. volumeIdSet = true } }) - markWritable := false - if (*writable && *readonly) || (!*writable && !*readonly) { - return fmt.Errorf("use -readonly or -writable") - } else if *writable { - markWritable = true + modeCount := 0 + for _, set := range []bool{*writable, *readonly, *readonlyCanDelete} { + if set { + modeCount++ + } } + if modeCount != 1 { + return fmt.Errorf("use exactly one of -writable, -readonly or -readonlyCanDelete") + } + markWritable := *writable if collectionSet { if *collection == "" { @@ -94,10 +103,12 @@ func (c *commandVolumeMark) Do(args []string, commandEnv *CommandEnv, writer io. state := "readonly" if markWritable { state = "writable" + } else if *readonlyCanDelete { + state = "readonly (can delete)" } var failures []error for _, target := range targets { - if err := markVolumeWritable(context.Background(), commandEnv.option.GrpcDialOption, target.volumeId, target.sourceVolumeServer, markWritable, true); err != nil { + if err := markVolumeState(context.Background(), commandEnv.option.GrpcDialOption, target.volumeId, target.sourceVolumeServer, markWritable, *readonlyCanDelete, true); err != nil { failures = append(failures, fmt.Errorf("mark volume %d on %s: %w", target.volumeId, target.sourceVolumeServer, err)) fmt.Fprintf(writer, "volume %d on %s: %v\n", target.volumeId, target.sourceVolumeServer, err) continue @@ -114,7 +125,7 @@ func (c *commandVolumeMark) Do(args []string, commandEnv *CommandEnv, writer io. volumeId := needle.VolumeId(*volumeIdInt) - return markVolumeWritable(context.Background(), commandEnv.option.GrpcDialOption, volumeId, sourceVolumeServer, markWritable, true) + return markVolumeState(context.Background(), commandEnv.option.GrpcDialOption, volumeId, sourceVolumeServer, markWritable, *readonlyCanDelete, true) } type volumeMarkTarget struct { diff --git a/weed/shell/command_volume_move.go b/weed/shell/command_volume_move.go index 3a218f1db..d5e5c327c 100644 --- a/weed/shell/command_volume_move.go +++ b/weed/shell/command_volume_move.go @@ -259,6 +259,11 @@ func deleteVolume(ctx context.Context, grpcDialOption grpc.DialOption, volumeId } func markVolumeWritable(ctx context.Context, grpcDialOption grpc.DialOption, volumeId needle.VolumeId, sourceVolumeServer pb.ServerAddress, writable, persist bool) (err error) { + return markVolumeState(ctx, grpcDialOption, volumeId, sourceVolumeServer, writable, false, persist) +} + +// canDelete: readonly that still accepts deletes, so expiring data drains the volume. +func markVolumeState(ctx context.Context, grpcDialOption grpc.DialOption, volumeId needle.VolumeId, sourceVolumeServer pb.ServerAddress, writable, canDelete, persist bool) (err error) { return operation.WithVolumeServerClient(false, sourceVolumeServer, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error { if writable { _, err = volumeServerClient.VolumeMarkWritable(ctx, &volume_server_pb.VolumeMarkWritableRequest{ @@ -266,8 +271,9 @@ func markVolumeWritable(ctx context.Context, grpcDialOption grpc.DialOption, vol }) } else { _, err = volumeServerClient.VolumeMarkReadonly(ctx, &volume_server_pb.VolumeMarkReadonlyRequest{ - VolumeId: uint32(volumeId), - Persist: persist, + VolumeId: uint32(volumeId), + Persist: persist, + CanDelete: canDelete, }) } return err diff --git a/weed/storage/store.go b/weed/storage/store.go index 6f5fa458c..528f4c1d9 100644 --- a/weed/storage/store.go +++ b/weed/storage/store.go @@ -807,15 +807,28 @@ func (s *Store) HasVolume(i needle.VolumeId) bool { return v != nil } -func (s *Store) MarkVolumeReadonly(i needle.VolumeId, persist bool) error { +func (s *Store) MarkVolumeReadonly(i needle.VolumeId, canDelete bool, persist bool) error { v := s.findVolume(i) if v == nil { return fmt.Errorf("volume %d not found", i) } + if canDelete && !v.HasRemoteFile() { + // deletes append tombstones to .idx, which a readonly boot opened + // O_RDONLY; remote volumes already delete through a RDWR idx + if err := v.reopenIdxForWrite(); err != nil { + return fmt.Errorf("volume %d reopen idx for write: %v", i, err) + } + } v.noWriteLock.Lock() - v.noWriteOrDelete = true + v.noWriteOrDelete = !canDelete + if canDelete { + v.noWriteCanDelete = true + } else if !v.HasRemoteFile() { + // downgrading a canDelete mark; remote volumes keep their derived flag + v.noWriteCanDelete = false + } if persist { - v.PersistReadOnly(true) + v.PersistReadOnly(true, canDelete) } v.noWriteLock.Unlock() return nil @@ -835,7 +848,11 @@ func (s *Store) MarkVolumeWritable(i needle.VolumeId) error { } v.noWriteLock.Lock() v.noWriteOrDelete = false - v.PersistReadOnly(false) + // Remote-tiered volumes must stay noWriteCanDelete regardless of marks. + if !v.HasRemoteFile() { + v.noWriteCanDelete = false + } + v.PersistReadOnly(false, false) v.noWriteLock.Unlock() // Clear the EIO streak and the sticky quarantine flag so the next // CollectHeartbeat can announce the volume again. If the disk is diff --git a/weed/storage/store_mark_readonly_can_delete_test.go b/weed/storage/store_mark_readonly_can_delete_test.go new file mode 100644 index 000000000..12f0c9bdb --- /dev/null +++ b/weed/storage/store_mark_readonly_can_delete_test.go @@ -0,0 +1,122 @@ +package storage + +import ( + "os" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/seaweedfs/seaweedfs/weed/stats" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/types" + "github.com/seaweedfs/seaweedfs/weed/util" +) + +// newSingleDirStore builds a single-disk store over dir, draining every notify +// channel so mount and unmount never block. +func newSingleDirStore(t *testing.T, dir string) *Store { + t.Helper() + require.NoError(t, os.MkdirAll(dir, 0o755)) + store := NewStore(nil, "localhost", 8080, 18080, "http://localhost:8080", "store-id", + []string{dir}, []int32{100}, []util.MinFreeSpace{{}}, "", + NeedleMapInMemory, []types.DiskType{types.HardDriveType}, nil, 3, + stats.DefaultDiskIOProbeConfig()) + done := make(chan struct{}) + go func() { + for { + select { + case <-store.NewVolumesChan: + case <-store.DeletedVolumesChan: + case <-store.NewEcShardsChan: + case <-store.DeletedEcShardsChan: + case <-store.StateUpdateChan: + case <-done: + return + } + } + }() + t.Cleanup(func() { + close(done) + }) + return store +} + +// canDelete rejects writes, keeps accepting deletes, and survives a restart. +func TestMarkVolumeReadonlyCanDelete(t *testing.T) { + dir := t.TempDir() + store := newSingleDirStore(t, dir) + const vid = needle.VolumeId(11) + require.NoError(t, store.AddVolume(vid, "", NeedleMapInMemory, "000", "", 0, needle.GetCurrentVersion(), 0, types.HardDriveType, 0)) + + n1, n2 := newRandomNeedle(1), newRandomNeedle(2) + _, err := store.WriteVolumeNeedle(vid, n1, true, false) + require.NoError(t, err) + _, err = store.WriteVolumeNeedle(vid, n2, true, false) + require.NoError(t, err) + + require.NoError(t, store.MarkVolumeReadonly(vid, true, true)) + v := store.GetVolume(vid) + require.True(t, v.noWriteCanDelete) + require.False(t, v.noWriteOrDelete) + require.True(t, v.IsReadOnly()) + + _, err = store.WriteVolumeNeedle(vid, newRandomNeedle(3), true, false) + require.Error(t, err, "writes must be rejected") + _, err = store.DeleteVolumeNeedle(vid, n1) + require.NoError(t, err, "deletes must still land") + + store.Close() + store2 := newSingleDirStore(t, dir) + v2 := store2.GetVolume(vid) + require.NotNil(t, v2) + require.True(t, v2.noWriteCanDelete) + require.False(t, v2.noWriteOrDelete) + + _, err = store2.WriteVolumeNeedle(vid, newRandomNeedle(4), true, false) + require.Error(t, err, "writes must stay rejected after restart") + _, err = store2.DeleteVolumeNeedle(vid, n2) + require.NoError(t, err, "deletes must still land after restart") + + // Downgrading to plain readonly clears the canDelete flag. + require.NoError(t, store2.MarkVolumeReadonly(vid, false, true)) + require.True(t, v2.noWriteOrDelete) + require.False(t, v2.noWriteCanDelete) + _, err = store2.DeleteVolumeNeedle(vid, n2) + require.Error(t, err, "plain readonly must reject deletes") + + // -writable clears the state again. + require.NoError(t, store2.MarkVolumeWritable(vid)) + require.False(t, v2.IsReadOnly()) + _, err = store2.WriteVolumeNeedle(vid, newRandomNeedle(5), true, false) + require.NoError(t, err) + store2.Close() +} + +// Upgrading a volume that booted plain persisted-readonly to canDelete must +// reopen the O_RDONLY .idx for writing, or delete tombstones cannot append. +func TestMarkVolumeReadonlyCanDelete_AfterReadonlyBoot(t *testing.T) { + dir := t.TempDir() + store := newSingleDirStore(t, dir) + const vid = needle.VolumeId(12) + require.NoError(t, store.AddVolume(vid, "", NeedleMapInMemory, "000", "", 0, needle.GetCurrentVersion(), 0, types.HardDriveType, 0)) + + n := newRandomNeedle(1) + _, err := store.WriteVolumeNeedle(vid, n, true, false) + require.NoError(t, err) + require.NoError(t, store.MarkVolumeReadonly(vid, false, true)) + store.Close() + + store2 := newSingleDirStore(t, dir) + v := store2.GetVolume(vid) + require.NotNil(t, v) + require.True(t, v.noWriteOrDelete) + _, err = store2.DeleteVolumeNeedle(vid, n) + require.Error(t, err, "plain readonly must reject deletes") + + require.NoError(t, store2.MarkVolumeReadonly(vid, true, true)) + require.False(t, v.noWriteOrDelete) + require.True(t, v.noWriteCanDelete) + _, err = store2.DeleteVolumeNeedle(vid, n) + require.NoError(t, err, "deletes must land once canDelete is set") + store2.Close() +} diff --git a/weed/storage/volume.go b/weed/storage/volume.go index cd06c64e4..d91fcca0b 100644 --- a/weed/storage/volume.go +++ b/weed/storage/volume.go @@ -536,9 +536,10 @@ func (v *Volume) IsReadOnly() bool { return v.noWriteOrDelete || v.noWriteCanDelete || v.location.isDiskSpaceLow.Load() } -func (v *Volume) PersistReadOnly(readOnly bool) { - v.volumeInfoRWLock.RLock() - defer v.volumeInfoRWLock.RUnlock() +func (v *Volume) PersistReadOnly(readOnly bool, canDelete bool) { + v.volumeInfoRWLock.Lock() + defer v.volumeInfoRWLock.Unlock() v.volumeInfo.ReadOnly = readOnly + v.volumeInfo.ReadOnlyCanDelete = readOnly && canDelete v.SaveVolumeInfo() } diff --git a/weed/storage/volume_loading.go b/weed/storage/volume_loading.go index 8bcbf4aeb..74217bc90 100644 --- a/weed/storage/volume_loading.go +++ b/weed/storage/volume_loading.go @@ -154,7 +154,11 @@ func (v *Volume) load(alsoLoadIndex bool, createDatIfMissing bool, needleMapKind if v.volumeInfo.ReadOnly && !v.HasRemoteFile() { // this covers the case where the volume is marked as read-only and has no remote file - v.noWriteOrDelete = true + if v.volumeInfo.ReadOnlyCanDelete { + v.noWriteCanDelete = true + } else { + v.noWriteOrDelete = true + } } if v.HasRemoteFile() { diff --git a/weed/storage/volume_loading_corrupt_idx_test.go b/weed/storage/volume_loading_corrupt_idx_test.go index ca9d02e1b..61772aea2 100644 --- a/weed/storage/volume_loading_corrupt_idx_test.go +++ b/weed/storage/volume_loading_corrupt_idx_test.go @@ -20,7 +20,7 @@ func TestLoad_CorruptIdx_NoSegfault(t *testing.T) { if _, _, _, err := v.writeNeedle2(newRandomNeedle(1), true, false); err != nil { t.Fatalf("seed write: %v", err) } - v.PersistReadOnly(true) // reload goes through SortedFileNeedleMap + v.PersistReadOnly(true, false) // reload goes through SortedFileNeedleMap v.Close() // Truncate .idx to a non-aligned size so the walk rejects it. diff --git a/weed/storage/volume_mark_writable_test.go b/weed/storage/volume_mark_writable_test.go index 3d6959de6..fa4fe50c0 100644 --- a/weed/storage/volume_mark_writable_test.go +++ b/weed/storage/volume_mark_writable_test.go @@ -27,7 +27,7 @@ func TestMarkVolumeWritable_ReopensPersistedReadOnly(t *testing.T) { // Persist read-only state into .vif, then simulate a server restart by // closing and re-opening the volume from the same directory. - v.PersistReadOnly(true) + v.PersistReadOnly(true, false) v.Close() v2, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0) diff --git a/weed/storage/volume_write_test.go b/weed/storage/volume_write_test.go index b060d6728..4d2874d5c 100644 --- a/weed/storage/volume_write_test.go +++ b/weed/storage/volume_write_test.go @@ -187,7 +187,7 @@ func TestWriteNeedleBlobRejectedOnReadOnlyVolume(t *testing.T) { if err != nil { t.Fatalf("read needle blob: %v", err) } - v.PersistReadOnly(true) + v.PersistReadOnly(true, false) v.Close() v, err = NewVolume(dir, dir, "", 7, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0)