From 4feb30deffd124e3f2a494b64f355fa4b7e0c968 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Mon, 19 Oct 2026 17:01:19 -0700 Subject: [PATCH] Federated volume-id bound via heartbeat stamps (#11763) (#11819) * pb: heartbeat fields for the federated volume-id bound (#11763) Heartbeat carries max_volume_id_seen and last_seen_topology_id up, and HeartbeatResponse echoes max_volume_id and topology_id down, so the allocation bound is replicated to every volume server. * master: echo the volume-id bound and fold reported bounds in (#11763) Every heartbeat response carries the leader current max volume id and topology id; every heartbeat reported max_volume_id_seen is folded into the topology bound via UpAdjustMaxVolumeId. * volume: persist a per-dir vid stamp of the highest id ever seen (#11763) The store keeps max_volume_id_seen + last_seen_topology_id in a .vid_stamp file per volume dir, fed by heartbeat echoes, AllocateVolume, and local volume/EC creation, seeded at boot from stamps and local holdings, and reported in every heartbeat. A rebuilt master recovers the bound from the first surviving veteran report. * volume-rust: mirror the vid stamp and heartbeat echo handling (#11763) Same mechanism as the Go volume server: .vid_stamp per dir, bound folded in from add_location scans and allocate/mount calls, echoed fields applied on heartbeat responses, reported on every heartbeat. * volume: serialize vid stamp writes and fsync the directory (#11763) * volume-rust: serialize vid stamp writes and fsync the directory (#11763) * master: echo the allocation bound to connected servers on change The bound was only sent at connect time or on digest mismatch, so a healthy stream never learned later increases. If the high-id server then went offline and the master lost state, survivors restored a stale bound and the id could be reissued. Resend the heartbeat response whenever the bound or topology id moved. * volume: retry failed vid stamp saves on later updates A transient write failure left the in-memory bound raised but the stamp stale, and repeat updates at the same bound never retried. Track a dirty flag so any later update re-attempts the save. * volume: cover stamp bootstrap for pre-stamp dirs in test * volume-rust: retry failed vid stamp saves, mirroring Go * volume-rust: stamp every dir with the combined bound on add_location A dir whose own stamp sat below the combined bound kept its stale value after fold; losing the higher-stamped disk then restarted the server below ids it had already learned. Rewrite all stamps after each location folds in. * volume: stamp mounted volume and ec shard ids (#11763) * volume-rust: stamp mounted volume ids, mirroring Go (#11763) * volume: retry dirty vid stamp save on unchanged topology id (#11763) * volume-rust: retry dirty vid stamp save on unchanged topology id (#11763) * master: record echoed bound from the sent heartbeat response (#11763) --- seaweed-volume/proto/master.proto | 22 ++ seaweed-volume/src/server/heartbeat.rs | 10 + seaweed-volume/src/storage/mod.rs | 1 + seaweed-volume/src/storage/store.rs | 35 ++- seaweed-volume/src/storage/vid_stamp.rs | 232 ++++++++++++++++++++ weed/pb/master.proto | 12 + weed/pb/master_pb/master.pb.go | 61 ++++- weed/server/master_grpc_server.go | 19 ++ weed/server/volume_grpc_client_to_master.go | 3 + weed/storage/store.go | 54 +++-- weed/storage/store_ec.go | 1 + weed/storage/vid_stamp.go | 148 +++++++++++++ weed/storage/vid_stamp_test.go | 94 ++++++++ 13 files changed, 659 insertions(+), 33 deletions(-) create mode 100644 seaweed-volume/src/storage/vid_stamp.rs create mode 100644 weed/storage/vid_stamp.go create mode 100644 weed/storage/vid_stamp_test.go diff --git a/seaweed-volume/proto/master.proto b/seaweed-volume/proto/master.proto index 153ed085b3..9bc3022417 100644 --- a/seaweed-volume/proto/master.proto +++ b/seaweed-volume/proto/master.proto @@ -119,6 +119,12 @@ message Heartbeat { // place of `volumes`. A master that does not understand this never sets // volume_digest_supported, so it keeps being sent the whole list. repeated VolumeInformationMessage changed_volumes = 28; + + // Highest volume id this server has ever observed, persisted locally. Fed + // by HeartbeatResponse.max_volume_id echoes and by AllocateVolume requests. + uint32 max_volume_id_seen = 29; + // Topology id of the master cluster this server last heartbeated to. + string last_seen_topology_id = 30; } message HeartbeatResponse { @@ -135,6 +141,12 @@ message HeartbeatResponse { // The master compares volume digests, so a server that reports one may send // changed_volumes in place of its whole list. bool volume_digest_supported = 9; + // The leader's current allocation bound; servers persist it and report it + // back as max_volume_id_seen so the bound survives master state loss. + uint32 max_volume_id = 10; + // The leader's cluster identity; a different value reported back later + // means the master's state was rebuilt. + string topology_id = 11; } message VolumeInformationMessage { @@ -305,6 +317,11 @@ message VolumeGrowRequest { string data_node = 7; uint32 memory_map_max_size_mb = 8; string disk_type = 9; + // If > 0, grow only when the average fill ratio of the layout's writable + // volumes has reached this threshold (e.g. 0.6). Lets callers pre-grow on + // fill level with one call, instead of listing volumes to compute the + // ratio themselves. Ignored when there are no writable volumes. + float grow_avg_fill_threshold = 10; } message AssignResponse { @@ -611,4 +628,9 @@ message RaftLeadershipTransferResponse { } message VolumeGrowResponse { + // Average fill ratio of the layout's writable volumes at request time + // (0 when there are none). Reported whether or not growth was triggered. + float avg_fill = 1; + // Number of writable volumes at request time. + uint32 writable_count = 2; } diff --git a/seaweed-volume/src/server/heartbeat.rs b/seaweed-volume/src/server/heartbeat.rs index a3a7595192..081ff71a25 100644 --- a/seaweed-volume/src/server/heartbeat.rs +++ b/seaweed-volume/src/server/heartbeat.rs @@ -474,6 +474,14 @@ async fn do_heartbeat( ) .into()); } + let echo_max_volume_id = hb_resp.max_volume_id; + let echo_topology_id = hb_resp.topology_id.clone(); + off_runtime(config, state, move |_, state| { + let store = state.store.read().unwrap(); + store.update_max_volume_id_seen(echo_max_volume_id); + store.update_last_seen_topology_id(&echo_topology_id); + }) + .await?; let options_changed = volume_options_pass(config, state, &hb_resp).await?; if options_changed && maybe_adjust_volume_max(config, state).await? { @@ -1360,6 +1368,8 @@ fn build_heartbeat_with_ec_status( grpc_port: config.grpc_port as u32, location_uuids, disk_tags, + max_volume_id_seen: store.max_volume_id_seen(), + last_seen_topology_id: store.last_seen_topology_id(), ..Default::default() }; (heartbeat, volumes, actions) diff --git a/seaweed-volume/src/storage/mod.rs b/seaweed-volume/src/storage/mod.rs index cd7bd1df0e..c2d0561fc4 100644 --- a/seaweed-volume/src/storage/mod.rs +++ b/seaweed-volume/src/storage/mod.rs @@ -11,6 +11,7 @@ pub mod store_ec_mirror; pub mod store_ec_reconcile; pub mod super_block; pub mod types; +pub mod vid_stamp; pub mod volume; pub mod volume_idx_rebuild; pub mod volume_idx_repair; diff --git a/seaweed-volume/src/storage/store.rs b/seaweed-volume/src/storage/store.rs index ad41c25208..a7f0ab5b4b 100644 --- a/seaweed-volume/src/storage/store.rs +++ b/seaweed-volume/src/storage/store.rs @@ -6,7 +6,7 @@ use std::collections::HashSet; use std::io; -use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; use crate::config::MinFreeSpace; @@ -72,6 +72,10 @@ pub struct Store { /// reporting the heartbeat that saw it. pub reported_collections: Mutex>, pub reported_ec_collections: Mutex>, + pub max_volume_id_seen: AtomicU32, + pub last_seen_topology_id: Mutex, + pub vid_stamp_lock: Mutex<()>, + pub vid_stamp_dirty: std::sync::atomic::AtomicBool, } impl Store { @@ -91,6 +95,10 @@ impl Store { rack: String::new(), reported_collections: Mutex::new(HashSet::new()), reported_ec_collections: Mutex::new(HashSet::new()), + max_volume_id_seen: AtomicU32::new(0), + last_seen_topology_id: Mutex::new(String::new()), + vid_stamp_lock: Mutex::new(()), + vid_stamp_dirty: std::sync::atomic::AtomicBool::new(false), } } @@ -152,6 +160,12 @@ impl Store { // (read-only target, partial copy). self.reconcile_ec_shards_across_disks(); + self.fold_location_into_vid_stamp(self.locations.last().unwrap()); + // Write the combined bound back to every dir, including ones whose + // own stamp or volumes sit below it — a lost disk must not restart + // the server below the bound its remaining dirs once reached. + self.save_vid_stamps(); + Ok(()) } @@ -409,6 +423,7 @@ impl Store { disk_type: DiskType, spec: &VolumeSpec<'_>, ) -> Result<(), VolumeError> { + self.stamp_allocated_vid(vid); if self.find_volume(vid).is_some() { return Err(VolumeError::AlreadyExists); } @@ -506,7 +521,7 @@ impl Store { vid )))); } - return loc.create_volume( + let result = loc.create_volume( vid, self.needle_map_kind, &VolumeSpec { @@ -514,6 +529,10 @@ impl Store { ..Default::default() }, ); + if result.is_ok() { + self.stamp_allocated_vid(vid); + } + return result; } } Err(VolumeError::Io(io::Error::new( @@ -603,7 +622,10 @@ impl Store { ..Default::default() }, ) { - Ok(()) => return Ok(()), + Ok(()) => { + self.stamp_allocated_vid(vid); + return Ok(()); + } Err(e) => { last_err = Some(e); continue; @@ -659,7 +681,10 @@ impl Store { ..Default::default() }, ) { - Ok(()) => return Ok(()), + Ok(()) => { + self.stamp_allocated_vid(vid); + return Ok(()); + } Err(e) => { last_err = Some(e); continue; @@ -952,6 +977,7 @@ impl Store { collection: &str, shard_ids: &[ShardId], ) -> Result<(), VolumeError> { + self.stamp_allocated_vid(vid); // Find the location where the EC files live let loc_idx = self.find_ec_location(vid, collection).ok_or_else(|| { VolumeError::Io(io::Error::new( @@ -974,6 +1000,7 @@ impl Store { shard_id: ShardId, source_disk_type: &str, ) -> Result<(), VolumeError> { + self.stamp_allocated_vid(vid); // The .ecx may live on a different disk than the shard being mounted // (ec.balance / ec.rebuild can spread shards across sibling disks), so // look up its owner once and point EcVolume::new at the directory that diff --git a/seaweed-volume/src/storage/vid_stamp.rs b/seaweed-volume/src/storage/vid_stamp.rs new file mode 100644 index 0000000000..a98d4eaaf2 --- /dev/null +++ b/seaweed-volume/src/storage/vid_stamp.rs @@ -0,0 +1,232 @@ +//! Persisted volume-id high-water mark ("vid stamp"), mirroring Go's +//! `weed/storage/vid_stamp.go`. +//! +//! Each volume dir carries a `.vid_stamp` JSON file with the highest volume +//! id the server has ever observed plus the topology id it last heartbeated +//! to. The bound is fed by heartbeat echoes, AllocateVolume, and local +//! volume/EC creation, is reported back in `Heartbeat.max_volume_id_seen`, +//! and survives master state loss so rebuilt masters cannot reissue ids an +//! offline server still holds. + +use std::fs; +use std::io; +use std::path::Path; + +use serde::{Deserialize, Serialize}; + +use crate::storage::types::VolumeId; + +use super::store::Store; + +const VID_STAMP_FILE_NAME: &str = ".vid_stamp"; + +#[derive(Serialize, Deserialize, Default)] +struct VidStamp { + #[serde(rename = "maxVolumeId")] + max_volume_id: u32, + #[serde(rename = "topologyId", default, skip_serializing_if = "String::is_empty")] + topology_id: String, +} + +impl Store { + pub fn max_volume_id_seen(&self) -> u32 { + self.max_volume_id_seen.load(std::sync::atomic::Ordering::Relaxed) + } + + pub fn last_seen_topology_id(&self) -> String { + self.last_seen_topology_id + .lock() + .unwrap() + .clone() + } + + pub fn update_max_volume_id_seen(&self, vid: u32) { + use std::sync::atomic::Ordering::Relaxed; + let prev = self.max_volume_id_seen.fetch_max(vid, Relaxed); + if vid > prev || self.vid_stamp_dirty.load(Relaxed) { + self.save_vid_stamps(); + } + } + + pub fn update_last_seen_topology_id(&self, id: &str) { + if id.is_empty() { + return; + } + { + let mut guard = self.last_seen_topology_id.lock().unwrap(); + if *guard == id && !self.vid_stamp_dirty.load(std::sync::atomic::Ordering::Relaxed) { + return; + } + *guard = id.to_string(); + } + self.save_vid_stamps(); + } + + /// Seed the bound from a freshly added location: its persisted stamp, + /// its volumes, and its EC shards. Called per location from + /// `add_location` after the EC reconcile passes have run. + pub(crate) fn fold_location_into_vid_stamp(&self, loc: &super::disk_location::DiskLocation) { + if let Ok(stamp) = read_vid_stamp(Path::new(&loc.directory)) { + self.update_max_volume_id_seen(stamp.max_volume_id); + if !stamp.topology_id.is_empty() { + self.update_last_seen_topology_id(&stamp.topology_id); + } + } + let mut max_vid: u32 = 0; + for vid in loc.volume_ids() { + max_vid = max_vid.max(vid.0); + } + for (vid, _) in loc.ec_volumes() { + max_vid = max_vid.max(vid.0); + } + if max_vid > 0 { + self.update_max_volume_id_seen(max_vid); + } + } + + pub(crate) fn save_vid_stamps(&self) { + let _guard = self.vid_stamp_lock.lock().unwrap(); + let stamp = match serde_json::to_vec(&VidStamp { + max_volume_id: self.max_volume_id_seen(), + topology_id: self.last_seen_topology_id(), + }) { + Ok(s) => s, + Err(_) => return, + }; + let mut failed = false; + for loc in &self.locations { + if let Err(e) = write_vid_stamp(Path::new(&loc.directory), &stamp) { + tracing::warn!("write vid stamp in {}: {}", loc.directory, e); + failed = true; + } + } + self.vid_stamp_dirty + .store(failed, std::sync::atomic::Ordering::Relaxed); + } +} + +fn read_vid_stamp(dir: &Path) -> io::Result { + let data = fs::read(dir.join(VID_STAMP_FILE_NAME))?; + serde_json::from_slice(&data).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e)) +} + +fn write_vid_stamp(dir: &Path, stamp: &[u8]) -> io::Result<()> { + let tmp = dir.join(format!("{}.tmp", VID_STAMP_FILE_NAME)); + { + use std::io::Write; + let mut f = fs::File::create(&tmp)?; + f.write_all(stamp)?; + f.sync_all()?; + } + fs::rename(&tmp, dir.join(VID_STAMP_FILE_NAME))?; + if cfg!(unix) { + fs::File::open(dir)?.sync_all()?; + } + Ok(()) +} + +/// Stamps the volume id of a newly created volume or EC shard set. Called +/// from `Store::add_volume` / `Store::mount_ec_shards*` so the bound covers +/// ids before their files exist. +impl Store { + pub(crate) fn stamp_allocated_vid(&self, vid: VolumeId) { + self.update_max_volume_id_seen(vid.0); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::config::MinFreeSpace; + use crate::storage::needle_map::NeedleMapKind; + use crate::storage::types::DiskType; + use tempfile::TempDir; + + fn make_store(dir: &str) -> Store { + let mut store = Store::new(NeedleMapKind::InMemory); + store + .add_location(dir, dir, 10, DiskType::HardDrive, MinFreeSpace::Percent(1.0), Vec::new()) + .unwrap(); + store + } + + #[test] + fn stamp_written_on_bound_increase() { + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let store = make_store(dir); + + store.update_max_volume_id_seen(42); + + let stamp = read_vid_stamp(tmp.path()).unwrap(); + assert_eq!(stamp.max_volume_id, 42); + } + + #[test] + fn stamp_survives_restart() { + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + + let store = make_store(dir); + store.update_max_volume_id_seen(42); + store.update_last_seen_topology_id("cluster-a"); + drop(store); + + let restarted = make_store(dir); + assert_eq!(restarted.max_volume_id_seen(), 42); + assert_eq!(restarted.last_seen_topology_id(), "cluster-a"); + } + + #[test] + fn later_disk_gets_combined_stamp() { + let tmp1 = TempDir::new().unwrap(); + let tmp2 = TempDir::new().unwrap(); + // dir2 carries a stale stamp from an older run + write_vid_stamp( + tmp2.path(), + &serde_json::to_vec(&VidStamp { + max_volume_id: 10, + topology_id: String::new(), + }) + .unwrap(), + ) + .unwrap(); + + let mut store = Store::new(NeedleMapKind::InMemory); + store + .add_location( + tmp1.path().to_str().unwrap(), + tmp1.path().to_str().unwrap(), + 10, + DiskType::HardDrive, + MinFreeSpace::Percent(1.0), + Vec::new(), + ) + .unwrap(); + store.update_max_volume_id_seen(100); + store + .add_location( + tmp2.path().to_str().unwrap(), + tmp2.path().to_str().unwrap(), + 10, + DiskType::HardDrive, + MinFreeSpace::Percent(1.0), + Vec::new(), + ) + .unwrap(); + + let stamp2 = read_vid_stamp(tmp2.path()).unwrap(); + assert_eq!(stamp2.max_volume_id, 100); + } + + #[test] + fn bound_is_monotonic() { + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let store = make_store(dir); + + store.update_max_volume_id_seen(10); + store.update_max_volume_id_seen(5); + assert_eq!(store.max_volume_id_seen(), 10); + } +} diff --git a/weed/pb/master.proto b/weed/pb/master.proto index aafa558877..9bc3022417 100644 --- a/weed/pb/master.proto +++ b/weed/pb/master.proto @@ -119,6 +119,12 @@ message Heartbeat { // place of `volumes`. A master that does not understand this never sets // volume_digest_supported, so it keeps being sent the whole list. repeated VolumeInformationMessage changed_volumes = 28; + + // Highest volume id this server has ever observed, persisted locally. Fed + // by HeartbeatResponse.max_volume_id echoes and by AllocateVolume requests. + uint32 max_volume_id_seen = 29; + // Topology id of the master cluster this server last heartbeated to. + string last_seen_topology_id = 30; } message HeartbeatResponse { @@ -135,6 +141,12 @@ message HeartbeatResponse { // The master compares volume digests, so a server that reports one may send // changed_volumes in place of its whole list. bool volume_digest_supported = 9; + // The leader's current allocation bound; servers persist it and report it + // back as max_volume_id_seen so the bound survives master state loss. + uint32 max_volume_id = 10; + // The leader's cluster identity; a different value reported back later + // means the master's state was rebuilt. + string topology_id = 11; } message VolumeInformationMessage { diff --git a/weed/pb/master_pb/master.pb.go b/weed/pb/master_pb/master.pb.go index 49278c403e..bf219ed1da 100644 --- a/weed/pb/master_pb/master.pb.go +++ b/weed/pb/master_pb/master.pb.go @@ -130,8 +130,13 @@ type Heartbeat struct { // place of `volumes`. A master that does not understand this never sets // volume_digest_supported, so it keeps being sent the whole list. ChangedVolumes []*VolumeInformationMessage `protobuf:"bytes,28,rep,name=changed_volumes,json=changedVolumes,proto3" json:"changed_volumes,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + // Highest volume id this server has ever observed, persisted locally. Fed + // by HeartbeatResponse.max_volume_id echoes and by AllocateVolume requests. + MaxVolumeIdSeen uint32 `protobuf:"varint,29,opt,name=max_volume_id_seen,json=maxVolumeIdSeen,proto3" json:"max_volume_id_seen,omitempty"` + // Topology id of the master cluster this server last heartbeated to. + LastSeenTopologyId string `protobuf:"bytes,30,opt,name=last_seen_topology_id,json=lastSeenTopologyId,proto3" json:"last_seen_topology_id,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *Heartbeat) Reset() { @@ -339,6 +344,20 @@ func (x *Heartbeat) GetChangedVolumes() []*VolumeInformationMessage { return nil } +func (x *Heartbeat) GetMaxVolumeIdSeen() uint32 { + if x != nil { + return x.MaxVolumeIdSeen + } + return 0 +} + +func (x *Heartbeat) GetLastSeenTopologyId() string { + if x != nil { + return x.LastSeenTopologyId + } + return "" +} + type HeartbeatResponse struct { state protoimpl.MessageState `protogen:"open.v1"` VolumeSizeLimit uint64 `protobuf:"varint,1,opt,name=volume_size_limit,json=volumeSizeLimit,proto3" json:"volume_size_limit,omitempty"` @@ -354,8 +373,14 @@ type HeartbeatResponse struct { // The master compares volume digests, so a server that reports one may send // changed_volumes in place of its whole list. VolumeDigestSupported bool `protobuf:"varint,9,opt,name=volume_digest_supported,json=volumeDigestSupported,proto3" json:"volume_digest_supported,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + // The leader's current allocation bound; servers persist it and report it + // back as max_volume_id_seen so the bound survives master state loss. + MaxVolumeId uint32 `protobuf:"varint,10,opt,name=max_volume_id,json=maxVolumeId,proto3" json:"max_volume_id,omitempty"` + // The leader's cluster identity; a different value reported back later + // means the master's state was rebuilt. + TopologyId string `protobuf:"bytes,11,opt,name=topology_id,json=topologyId,proto3" json:"topology_id,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *HeartbeatResponse) Reset() { @@ -451,6 +476,20 @@ func (x *HeartbeatResponse) GetVolumeDigestSupported() bool { return false } +func (x *HeartbeatResponse) GetMaxVolumeId() uint32 { + if x != nil { + return x.MaxVolumeId + } + return 0 +} + +func (x *HeartbeatResponse) GetTopologyId() string { + if x != nil { + return x.TopologyId + } + return "" +} + type VolumeInformationMessage struct { state protoimpl.MessageState `protogen:"open.v1"` Id uint32 `protobuf:"varint,1,opt,name=id,proto3" json:"id,omitempty"` @@ -5058,7 +5097,7 @@ const file_master_proto_rawDesc = "" + "\adisk_id\x18\x01 \x01(\rR\x06diskId\x12\x12\n" + "\x04tags\x18\x02 \x03(\tR\x04tags\x12\x12\n" + "\x04type\x18\x03 \x01(\tR\x04type\x12(\n" + - "\x10max_volume_count\x18\x04 \x01(\x03R\x0emaxVolumeCount\"\xf0\v\n" + + "\x10max_volume_count\x18\x04 \x01(\x03R\x0emaxVolumeCount\"\xd0\f\n" + "\tHeartbeat\x12\x0e\n" + "\x02ip\x18\x01 \x01(\tR\x02ip\x12\x12\n" + "\x04port\x18\x02 \x01(\rR\x04port\x12\x1d\n" + @@ -5090,7 +5129,9 @@ const file_master_proto_rawDesc = "" + "\x10disk_total_bytes\x18\x19 \x03(\v2(.master_pb.Heartbeat.DiskTotalBytesEntryR\x0ediskTotalBytes\x12O\n" + "\x0fdisk_free_bytes\x18\x1a \x03(\v2'.master_pb.Heartbeat.DiskFreeBytesEntryR\rdiskFreeBytes\x12(\n" + "\rvolume_digest\x18\x1b \x01(\x04H\x00R\fvolumeDigest\x88\x01\x01\x12L\n" + - "\x0fchanged_volumes\x18\x1c \x03(\v2#.master_pb.VolumeInformationMessageR\x0echangedVolumes\x1aB\n" + + "\x0fchanged_volumes\x18\x1c \x03(\v2#.master_pb.VolumeInformationMessageR\x0echangedVolumes\x12+\n" + + "\x12max_volume_id_seen\x18\x1d \x01(\rR\x0fmaxVolumeIdSeen\x121\n" + + "\x15last_seen_topology_id\x18\x1e \x01(\tR\x12lastSeenTopologyId\x1aB\n" + "\x14MaxVolumeCountsEntry\x12\x10\n" + "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" + "\x05value\x18\x02 \x01(\rR\x05value:\x028\x01\x1aA\n" + @@ -5100,7 +5141,7 @@ const file_master_proto_rawDesc = "" + "\x12DiskFreeBytesEntry\x12\x10\n" + "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" + "\x05value\x18\x02 \x01(\x04R\x05value:\x028\x01B\x10\n" + - "\x0e_volume_digest\"\xbc\x03\n" + + "\x0e_volume_digest\"\x81\x04\n" + "\x11HeartbeatResponse\x12*\n" + "\x11volume_size_limit\x18\x01 \x01(\x04R\x0fvolumeSizeLimit\x12\x16\n" + "\x06leader\x18\x02 \x01(\tR\x06leader\x12'\n" + @@ -5110,7 +5151,11 @@ const file_master_proto_rawDesc = "" + "\x10duplicated_uuids\x18\x06 \x03(\tR\x0fduplicatedUuids\x12 \n" + "\vpreallocate\x18\a \x01(\bR\vpreallocate\x125\n" + "\x17resend_full_volume_list\x18\b \x01(\bR\x14resendFullVolumeList\x126\n" + - "\x17volume_digest_supported\x18\t \x01(\bR\x15volumeDigestSupported\"\xe2\x04\n" + + "\x17volume_digest_supported\x18\t \x01(\bR\x15volumeDigestSupported\x12\"\n" + + "\rmax_volume_id\x18\n" + + " \x01(\rR\vmaxVolumeId\x12\x1f\n" + + "\vtopology_id\x18\v \x01(\tR\n" + + "topologyId\"\xe2\x04\n" + "\x18VolumeInformationMessage\x12\x0e\n" + "\x02id\x18\x01 \x01(\rR\x02id\x12\x12\n" + "\x04size\x18\x02 \x01(\x04R\x04size\x12\x1e\n" + diff --git a/weed/server/master_grpc_server.go b/weed/server/master_grpc_server.go index 058ecacd58..ac9e0577ef 100644 --- a/weed/server/master_grpc_server.go +++ b/weed/server/master_grpc_server.go @@ -46,6 +46,8 @@ func (ms *MasterServer) heartbeatResponse() *master_pb.HeartbeatResponse { return &master_pb.HeartbeatResponse{ VolumeSizeLimit: uint64(ms.option.VolumeSizeLimitMB) * 1024 * 1024, Preallocate: ms.preallocateSize > 0, + MaxVolumeId: uint32(ms.Topo.GetMaxVolumeId()), + TopologyId: ms.Topo.GetTopologyId(), } } @@ -137,6 +139,8 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ } }() + var echoedVid needle.VolumeId + var echoedTopoId string for { heartbeat, err := stream.Recv() if err != nil { @@ -179,6 +183,7 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ } ms.Topo.Sequence.SetMax(heartbeat.MaxFileKey) + ms.Topo.UpAdjustMaxVolumeId(needle.VolumeId(heartbeat.MaxVolumeIdSeen)) if dn == nil { // Skip delta heartbeat for volume server versions better than 3.28 https://github.com/seaweedfs/seaweedfs/pull/3630 if heartbeat.Ip == "" { @@ -206,6 +211,7 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ glog.Warningf("SendHeartbeat.Send volume size to %s:%d %v", dn.Ip, dn.Port, err) return err } + echoedVid, echoedTopoId = needle.VolumeId(response.MaxVolumeId), response.TopologyId stats.MasterReceivedHeartbeatCounter.WithLabelValues("dataNode").Inc() dn.Counter++ } @@ -342,6 +348,19 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ glog.Warningf("SendHeartbeat.Send resend request to %s:%d %v", dn.Ip, dn.Port, err) return err } + echoedVid, echoedTopoId = needle.VolumeId(response.MaxVolumeId), response.TopologyId + } + + // Every connected server keeps a durable copy of the bound. Echo it + // whenever it moved so one veteran can restore it after state loss, + // even while the offline server that held the high ids never rejoins. + if vid, topoId := ms.Topo.GetMaxVolumeId(), ms.Topo.GetTopologyId(); vid != echoedVid || topoId != echoedTopoId { + response := ms.heartbeatResponse() + if err := stream.Send(response); err != nil { + glog.Warningf("SendHeartbeat.Send bound echo %v", err) + return err + } + echoedVid, echoedTopoId = needle.VolumeId(response.MaxVolumeId), response.TopologyId } } } diff --git a/weed/server/volume_grpc_client_to_master.go b/weed/server/volume_grpc_client_to_master.go index 13700fb802..5c17a3b0ca 100644 --- a/weed/server/volume_grpc_client_to_master.go +++ b/weed/server/volume_grpc_client_to_master.go @@ -19,6 +19,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/security" "github.com/seaweedfs/seaweedfs/weed/storage/backend" "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" @@ -217,6 +218,8 @@ func (vs *VolumeServer) doHeartbeatWithRetry(masterAddress pb.ServerAddress, grp os.Exit(1) } } + vs.store.UpdateMaxVolumeIdSeen(needle.VolumeId(in.MaxVolumeId)) + vs.store.UpdateLastSeenTopologyId(in.TopologyId) volumeOptsChanged := false if vs.store.GetPreallocate() != in.GetPreallocate() { vs.store.SetPreallocate(in.GetPreallocate()) diff --git a/weed/storage/store.go b/weed/storage/store.go index 06c87c8239..bc04dd2dac 100644 --- a/weed/storage/store.go +++ b/weed/storage/store.go @@ -97,6 +97,11 @@ type Store struct { // the heartbeat that saw it. Written only from the heartbeat goroutine. reportedCollections map[string]struct{} reportedEcCollections map[string]struct{} + maxVolumeIdSeen atomic.Uint32 + lastSeenTopologyId atomic.Value // string + vidStampReady atomic.Bool + vidStampDirty atomic.Bool + vidStampLock sync.Mutex } func (s *Store) String() (str string) { @@ -145,6 +150,7 @@ func NewStore( diskId := uint32(i) // Track disk ID location.ecShardNotifyHandler = func(collection string, vid needle.VolumeId, shardId erasure_coding.ShardId, ecVolume *erasure_coding.EcVolume) { + s.UpdateMaxVolumeIdSeen(vid) si := erasure_coding.NewShardsInfo() si.Set(erasure_coding.NewShardInfo(shardId, erasure_coding.ShardSize(ecVolume.ShardSize()))) @@ -211,6 +217,8 @@ func NewStore( glog.Fatalf("failed to resolve state for volume %s: %v", id, err) } + s.bootstrapMaxVolumeIdSeen() + return } @@ -320,6 +328,7 @@ func (s *Store) FindFreeLocation(filterFn func(location *DiskLocation) bool, rep return ret } func (s *Store) addVolume(vid needle.VolumeId, collection string, needleMapKind NeedleMapKind, replicaPlacement *super_block.ReplicaPlacement, ttl *needle.TTL, preallocate int64, ver needle.Version, memoryMapMaxSizeMb uint32, diskType DiskType, ldbTimeout int64) error { + s.UpdateMaxVolumeIdSeen(vid) if s.findVolume(vid) != nil { return fmt.Errorf("Volume Id %d already exists!", vid) } @@ -686,27 +695,29 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat { stats.VolumeServerIoQuarantineGauge.WithLabelValues("ec_shard").Set(float64(quarantinedEcShards)) return &master_pb.Heartbeat{ - Ip: s.Ip, - Port: uint32(s.Port), - GrpcPort: uint32(s.GrpcPort), - PublicUrl: s.PublicUrl, - Id: s.Id, - MaxVolumeCounts: maxVolumeCounts, - DiskTotalBytes: diskTotalBytes, - DiskFreeBytes: diskFreeBytes, - MaxFileKey: NeedleIdToUint64(maxFileKey), - DataCenter: s.dataCenter, - Rack: s.rack, - Volumes: heartbeatVolumes, - ChangedVolumes: changedVolumes, - DeletedVolumes: departedVolumes, - VolumeDigest: &volumeDigest, - DeletedEcShards: deletedEcVolumes, - HasNoVolumes: hasNoVolumes, - HasNoEcShards: len(ecVolumeMessages) == 0, - LocationUuids: uuidList, - DiskTags: diskTags, - State: state, + Ip: s.Ip, + Port: uint32(s.Port), + GrpcPort: uint32(s.GrpcPort), + PublicUrl: s.PublicUrl, + Id: s.Id, + MaxVolumeCounts: maxVolumeCounts, + DiskTotalBytes: diskTotalBytes, + DiskFreeBytes: diskFreeBytes, + MaxFileKey: NeedleIdToUint64(maxFileKey), + DataCenter: s.dataCenter, + Rack: s.rack, + Volumes: heartbeatVolumes, + ChangedVolumes: changedVolumes, + DeletedVolumes: departedVolumes, + VolumeDigest: &volumeDigest, + DeletedEcShards: deletedEcVolumes, + HasNoVolumes: hasNoVolumes, + HasNoEcShards: len(ecVolumeMessages) == 0, + LocationUuids: uuidList, + DiskTags: diskTags, + State: state, + MaxVolumeIdSeen: uint32(s.MaxVolumeIdSeen()), + LastSeenTopologyId: s.LastSeenTopologyId(), } } @@ -1027,6 +1038,7 @@ func (s *Store) mountVolume(i needle.VolumeId, collection *string, validators .. ReadOnly: readOnly, ReadOnlyCanDelete: readOnlyCanDelete, } + s.UpdateMaxVolumeIdSeen(i) return nil } } diff --git a/weed/storage/store_ec.go b/weed/storage/store_ec.go index 5ec1c9f877..396a1619a7 100644 --- a/weed/storage/store_ec.go +++ b/weed/storage/store_ec.go @@ -304,6 +304,7 @@ func (s *Store) MountEcShards(collection string, vid needle.VolumeId, shardId er DiskId: uint32(diskId), EncodeTsNs: ecVolume.EncodeTsNs, } + s.UpdateMaxVolumeIdSeen(vid) return nil } if errors.Is(err, os.ErrNotExist) { diff --git a/weed/storage/vid_stamp.go b/weed/storage/vid_stamp.go new file mode 100644 index 0000000000..5e304d87f3 --- /dev/null +++ b/weed/storage/vid_stamp.go @@ -0,0 +1,148 @@ +package storage + +import ( + "encoding/json" + "os" + "path/filepath" + + "github.com/seaweedfs/seaweedfs/weed/glog" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/util" +) + +const vidStampFileName = ".vid_stamp" + +type vidStamp struct { + MaxVolumeId uint32 `json:"maxVolumeId"` + TopologyId string `json:"topologyId,omitempty"` +} + +func (s *Store) MaxVolumeIdSeen() needle.VolumeId { + return needle.VolumeId(s.maxVolumeIdSeen.Load()) +} + +func (s *Store) LastSeenTopologyId() string { + if v := s.lastSeenTopologyId.Load(); v != nil { + return v.(string) + } + return "" +} + +// UpdateMaxVolumeIdSeen raises the server's persisted bound and stamps every +// volume dir when it grows. Feeds: heartbeat echoes, AllocateVolume, local +// volume and EC shard creation. +func (s *Store) UpdateMaxVolumeIdSeen(vid needle.VolumeId) { + grew := false + for { + current := s.maxVolumeIdSeen.Load() + if current >= uint32(vid) { + break + } + if s.maxVolumeIdSeen.CompareAndSwap(current, uint32(vid)) { + grew = true + break + } + } + if s.vidStampReady.Load() && (grew || s.vidStampDirty.Load()) { + s.saveVidStamps() + } +} + +func (s *Store) UpdateLastSeenTopologyId(id string) { + if id == "" || (s.LastSeenTopologyId() == id && !s.vidStampDirty.Load()) { + return + } + s.lastSeenTopologyId.Store(id) + if s.vidStampReady.Load() { + s.saveVidStamps() + } +} + +// bootstrapMaxVolumeIdSeen seeds the bound from persisted stamps plus the max +// of every volume and EC shard id on disk, so a server upgraded mid-incident +// still reports a bound covering its own holdings. +func (s *Store) bootstrapMaxVolumeIdSeen() { + var maxVid needle.VolumeId + for _, location := range s.Locations { + if stamp, err := readVidStamp(location.Directory); err == nil { + if needle.VolumeId(stamp.MaxVolumeId) > maxVid { + maxVid = needle.VolumeId(stamp.MaxVolumeId) + } + if stamp.TopologyId != "" && s.LastSeenTopologyId() == "" { + s.lastSeenTopologyId.Store(stamp.TopologyId) + } + } + location.volumesLock.RLock() + for vid := range location.volumes { + if vid > maxVid { + maxVid = vid + } + } + location.volumesLock.RUnlock() + location.ecVolumesLock.RLock() + for vid := range location.ecVolumes { + if vid > maxVid { + maxVid = vid + } + } + location.ecVolumesLock.RUnlock() + } + s.maxVolumeIdSeen.Store(uint32(maxVid)) + s.vidStampReady.Store(true) + s.saveVidStamps() +} + +func (s *Store) saveVidStamps() { + s.vidStampLock.Lock() + defer s.vidStampLock.Unlock() + stamp, err := json.Marshal(vidStamp{ + MaxVolumeId: s.maxVolumeIdSeen.Load(), + TopologyId: s.LastSeenTopologyId(), + }) + if err != nil { + return + } + failed := false + for _, location := range s.Locations { + if err := writeVidStamp(location.Directory, stamp); err != nil { + glog.Warningf("write vid stamp in %s: %v", location.Directory, err) + failed = true + } + } + s.vidStampDirty.Store(failed) +} + +func readVidStamp(dir string) (*vidStamp, error) { + data, err := os.ReadFile(filepath.Join(dir, vidStampFileName)) + if err != nil { + return nil, err + } + stamp := &vidStamp{} + if err := json.Unmarshal(data, stamp); err != nil { + return nil, err + } + return stamp, nil +} + +func writeVidStamp(dir string, stamp []byte) error { + tmp := filepath.Join(dir, vidStampFileName+".tmp") + f, err := os.OpenFile(tmp, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644) + if err != nil { + return err + } + if _, err := f.Write(stamp); err != nil { + f.Close() + return err + } + if err := f.Sync(); err != nil { + f.Close() + return err + } + if err := f.Close(); err != nil { + return err + } + if err := os.Rename(tmp, filepath.Join(dir, vidStampFileName)); err != nil { + return err + } + return util.FsyncDir(dir) +} diff --git a/weed/storage/vid_stamp_test.go b/weed/storage/vid_stamp_test.go new file mode 100644 index 0000000000..59e5251bc2 --- /dev/null +++ b/weed/storage/vid_stamp_test.go @@ -0,0 +1,94 @@ +package storage + +import ( + "os" + "path/filepath" + "testing" + + "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" +) + +func newVidStampStore(t *testing.T, dir string) *Store { + t.Helper() + return 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()) +} + +func TestVidStampWrittenOnBoundIncrease(t *testing.T) { + dir := t.TempDir() + store := newVidStampStore(t, dir) + defer store.Close() + + store.UpdateMaxVolumeIdSeen(needle.VolumeId(42)) + + data, err := os.ReadFile(filepath.Join(dir, vidStampFileName)) + if err != nil { + t.Fatalf("read stamp: %v", err) + } + stamp, err := readVidStamp(dir) + if err != nil { + t.Fatalf("parse stamp: %v", err) + } + if stamp.MaxVolumeId != 42 { + t.Fatalf("stamp max = %d, want 42", stamp.MaxVolumeId) + } + if len(data) == 0 { + t.Fatal("empty stamp") + } +} + +func TestVidStampSurvivesRestart(t *testing.T) { + dir := t.TempDir() + + store := newVidStampStore(t, dir) + store.UpdateMaxVolumeIdSeen(needle.VolumeId(42)) + store.UpdateLastSeenTopologyId("cluster-a") + store.Close() + + restarted := newVidStampStore(t, dir) + defer restarted.Close() + if got := restarted.MaxVolumeIdSeen(); got != needle.VolumeId(42) { + t.Fatalf("restored bound = %d, want 42", got) + } + if got := restarted.LastSeenTopologyId(); got != "cluster-a" { + t.Fatalf("restored topology id = %q, want cluster-a", got) + } +} + +func TestVidStampMonotonic(t *testing.T) { + dir := t.TempDir() + store := newVidStampStore(t, dir) + defer store.Close() + + store.UpdateMaxVolumeIdSeen(needle.VolumeId(10)) + store.UpdateMaxVolumeIdSeen(needle.VolumeId(5)) + if got := store.MaxVolumeIdSeen(); got != needle.VolumeId(10) { + t.Fatalf("bound regressed to %d, want 10", got) + } +} + +func TestVidStampBootstrapFromVolumes(t *testing.T) { + dir := t.TempDir() + + store := newVidStampStore(t, dir) + if err := store.AddVolume(17, "", NeedleMapInMemory, "000", "", 0, needle.GetCurrentVersion(), 0, types.HardDriveType, 0); err != nil { + t.Fatalf("add volume: %v", err) + } + store.Close() + + // upgrading a pre-stamp dir: volumes exist but no stamp yet + if err := os.Remove(filepath.Join(dir, vidStampFileName)); err != nil { + t.Fatalf("remove stamp: %v", err) + } + + restarted := newVidStampStore(t, dir) + defer restarted.Close() + if got := restarted.MaxVolumeIdSeen(); got != needle.VolumeId(17) { + t.Fatalf("bootstrapped bound = %d, want 17", got) + } +}