From ce7d38863905c7f90f06f6af52983c96c4f430e5 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 7 Aug 2026 23:36:28 -0700 Subject: [PATCH] heartbeat: send only the volumes that changed (#10640) * pb: let a heartbeat carry only the volumes that changed A partial list cannot travel in volumes: a master that did not understand it would read the absences as deletions. So changes get their own field, used only once the master has said it compares digests and can tell when it has fallen behind. * master: apply the volumes a heartbeat reports as changed Only the named volumes are touched. A full report says the server holds exactly these; a changed report says nothing about the ones it leaves out, so absence must not read as removal. Also advertises that the master compares digests, which is what lets a server stop sending its whole list. Advertising it once per connection means a server reconnecting to a master that does not is back to full lists straight away. * volume: send only the volumes that changed once the master accepts them The whole list goes on every heartbeat until the master says it compares digests, and again whenever it asks, so a master that cannot tell when it has fallen behind never has to. has_no_volumes stays derived from a full list alone. Deriving it from what a heartbeat happens to carry would make a quiet one read as a server that had lost every volume, and the master would drop them all. The digest still covers every volume held rather than the ones sent, which is what lets the master confirm that applying the changes left it current. Reporting state is per-connection: a server that reconnects, or reaches a different master, starts again from the full list. * volume: let the zero reporting state stand for having told no master anything A Store built as a literal, which tests do, left the reporting state nil and panicked on the first heartbeat. As a value its zero form already means nothing has been reported to anyone, which is exactly the state that sends the whole list. * rust: send only the volumes that changed once the master accepts them Mirrors the Go volume server, with one hazard the Go side does not have: mount and unmount deltas here are derived by diffing successive heartbeats, so a heartbeat that carries a partial list would report every volume it left out as unmounted. Collecting now returns the full set alongside the message, and every site that diffs uses that rather than what went on the wire. * volume: do not let a full-list request be lost to the heartbeat it raced The request arrived while a heartbeat was already being built as a delta, and committing that heartbeat cleared it, so the master waited for another digest mismatch before asking again. Count the requests and clear only the one the heartbeat answered. * rust: stop marking volumes reported by a heartbeat that is thrown away The state-notify path collected a heartbeat only to diff its volume list, then sent a delta message of its own and dropped the one it had collected. Once collecting recorded what the master had been told, every mount or unmount silently marked the changed volumes as sent, and the master learned of them only after a digest mismatch. Snapshotting no longer records anything, and no longer expires ec volumes whose deletion that path was already discarding. * master: announce only the volumes a change actually brought Every changed volume was broadcast as a new location. Volumes grow constantly and growth moves no location, so on a busy cluster that told every connected client about volumes it could already reach, filling bounded broadcast queues and pushing out the topology updates that matter. * master: ask for the full list when only one can repair the master Delta heartbeats stop the full report, and with it the only thing that re-registers a volume the lookup index lost. The volume server cannot see that divergence and its digest cannot show it, so the master now checks its own two indexes agree and asks for the list when they do not. A node reporting one volume id twice is kept on full lists for the same reason rather than merely skipped: its digest can never be verified, so nothing else would tell the master what it had stopped holding. * master: keep the volume options on every heartbeat response A volume server takes them from whatever response arrives, and preallocate is a bare bool with no way to tell off from unmentioned. A response sent to ask for the volume list therefore turned preallocation off until the server reconnected. Responses sent mid-stream now start from the configured options rather than being built field by field. * master: announce a volume the lookup index had lost Repairing the index makes the volume servable again, but clients were told it went when the node dropped out and nothing told them otherwise: the disk map still held it, so it did not count as an arrival. Reaching the lookup index is what makes a volume servable, so recovering an entry there is an arrival as far as clients are concerned, on both the full report and the changed-volume path. --- seaweed-volume/proto/master.proto | 7 + seaweed-volume/src/server/heartbeat.rs | 255 ++++++++++++++++-- seaweed-volume/src/storage/mod.rs | 1 + seaweed-volume/src/storage/store.rs | 2 + seaweed-volume/src/storage/volume_report.rs | 73 +++++ weed/pb/master.proto | 7 + weed/pb/master_pb/master.pb.go | 214 ++++++++------- weed/server/master_grpc_server.go | 71 +++-- ...master_grpc_server_changed_volumes_test.go | 147 ++++++++++ weed/server/master_grpc_server_digest_test.go | 64 ++++- weed/server/volume_grpc_client_to_master.go | 11 + weed/storage/store.go | 52 +++- weed/storage/store_volume_report.go | 87 ++++++ weed/storage/store_volume_report_test.go | 155 +++++++++++ weed/topology/topology.go | 51 ++++ 15 files changed, 1046 insertions(+), 151 deletions(-) create mode 100644 seaweed-volume/src/storage/volume_report.rs create mode 100644 weed/server/master_grpc_server_changed_volumes_test.go create mode 100644 weed/storage/store_volume_report.go create mode 100644 weed/storage/store_volume_report_test.go diff --git a/seaweed-volume/proto/master.proto b/seaweed-volume/proto/master.proto index afffc7d06..841924727 100644 --- a/seaweed-volume/proto/master.proto +++ b/seaweed-volume/proto/master.proto @@ -111,6 +111,10 @@ message Heartbeat { // from servers that do not compute it, and distinct from a digest of 0, which // is what a server holding no volumes reports. optional uint64 volume_digest = 27; + // Volumes whose reported state changed since the last heartbeat, sent in + // 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; } message HeartbeatResponse { @@ -124,6 +128,9 @@ message HeartbeatResponse { // The master's view of this server's volumes disagrees with the reported // digest, so it needs the full volume list rather than changes alone. bool resend_full_volume_list = 8; + // 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; } message VolumeInformationMessage { diff --git a/seaweed-volume/src/server/heartbeat.rs b/seaweed-volume/src/server/heartbeat.rs index 0a9181220..3af2adb42 100644 --- a/seaweed-volume/src/server/heartbeat.rs +++ b/seaweed-volume/src/server/heartbeat.rs @@ -19,6 +19,7 @@ use crate::pb::master_pb::seaweed_client::SeaweedClient; use crate::pb::volume_server_pb; use crate::remote_storage::s3_tier::{S3TierBackend, S3TierConfig}; use crate::storage::store::Store; +use crate::storage::volume_report::VolumeReportKey; use crate::storage::volume_report_hash::report_hash; use crate::storage::types::NeedleId; @@ -274,6 +275,13 @@ fn duplicate_directories(store: &Store, duplicated_uuids: &[String]) -> Vec bool { + if hb_resp.volume_digest_supported { + store.volume_report.accept_deltas(); + } + if hb_resp.resend_full_volume_list { + info!("master asked for the full volume list"); + store.volume_report.request_full_list(); + } let mut volume_opts_changed = false; if store.get_preallocate() != hb_resp.preallocate { store.set_preallocate(hb_resp.preallocate); @@ -384,13 +392,14 @@ async fn do_heartbeat( let (tx, rx) = tokio::sync::mpsc::channel::(32); + // This master may know nothing about this server, and has not yet said + // whether it understands digests, so start from the whole list. + state.store.read().unwrap().volume_report.reset(); + // Keep track of what we sent, to generate delta updates - let initial_hb = collect_heartbeat(config, state); - let mut last_volumes: HashMap = initial_hb - .volumes - .iter() - .map(|v| (v.id, v.clone())) - .collect(); + let (initial_hb, initial_volumes) = collect_heartbeat_with_snapshot(config, state); + let mut last_volumes: HashMap = + initial_volumes.iter().map(|v| (v.id, v.clone())).collect(); let mut last_ec_shards = { let store = state.store.read().unwrap(); collect_ec_shard_delta_messages(&store) @@ -455,9 +464,10 @@ async fn do_heartbeat( apply_master_volume_options(&s, &hb_resp) }; if changed { - let adjusted_hb = collect_heartbeat(config, state); + let (adjusted_hb, adjusted_volumes) = + collect_heartbeat_with_snapshot(config, state); last_volumes = - adjusted_hb.volumes.iter().map(|v| (v.id, v.clone())).collect(); + adjusted_volumes.iter().map(|v| (v.id, v.clone())).collect(); last_ec_shards = { let store = state.store.read().unwrap(); collect_ec_shard_delta_messages(&store) @@ -490,8 +500,8 @@ async fn do_heartbeat( let s = state.store.read().unwrap(); s.maybe_adjust_volume_max(); } - let current_hb = collect_heartbeat(config, state); - last_volumes = current_hb.volumes.iter().map(|v| (v.id, v.clone())).collect(); + let (current_hb, current_volumes) = collect_heartbeat_with_snapshot(config, state); + last_volumes = current_volumes.iter().map(|v| (v.id, v.clone())).collect(); last_ec_shards = { let store = state.store.read().unwrap(); collect_ec_shard_delta_messages(&store) @@ -519,8 +529,8 @@ async fn do_heartbeat( info!("Heartbeat stopping"); return Ok(None); } - let current_hb = collect_heartbeat(config, state); - let current_volumes: HashMap = current_hb.volumes.iter().map(|v| (v.id, v.clone())).collect(); + let held_volumes = collect_volume_snapshot(config, state); + let current_volumes: HashMap = held_volumes.iter().map(|v| (v.id, v.clone())).collect(); let current_ec_shards = { let store = state.store.read().unwrap(); collect_ec_shard_delta_messages(&store) @@ -735,10 +745,10 @@ fn parse_bool_property(value: Option<&String>) -> bool { } /// Collect volume information into a Heartbeat message. -fn collect_heartbeat( +fn collect_heartbeat_with_snapshot( config: &HeartbeatConfig, state: &Arc, -) -> master_pb::Heartbeat { +) -> (master_pb::Heartbeat, Vec) { let mut store = state.store.write().unwrap(); let (ec_shards, deleted_ec_shards) = store.delete_expired_ec_volumes(); build_heartbeat_with_ec_status( @@ -746,9 +756,28 @@ fn collect_heartbeat( &mut store, deleted_ec_shards, ec_shards.is_empty(), + true, ) } +/// Lists the volumes the server holds without touching reporting state or +/// expiring anything, for callers that only need to diff against a previous +/// snapshot and send a message of their own. +fn collect_volume_snapshot( + config: &HeartbeatConfig, + state: &Arc, +) -> Vec { + let mut store = state.store.write().unwrap(); + build_heartbeat_with_ec_status(config, &mut store, Vec::new(), true, false).1 +} + +fn collect_heartbeat( + config: &HeartbeatConfig, + state: &Arc, +) -> master_pb::Heartbeat { + collect_heartbeat_with_snapshot(config, state).0 +} + fn collect_location_metadata( store: &Store, disk_max_by_id: &[i32], @@ -781,15 +810,19 @@ fn collect_location_metadata( #[cfg(test)] fn build_heartbeat(config: &HeartbeatConfig, store: &mut Store) -> master_pb::Heartbeat { let has_no_ec_shards = collect_live_ec_shards(store, false).is_empty(); - build_heartbeat_with_ec_status(config, store, Vec::new(), has_no_ec_shards) + build_heartbeat_with_ec_status(config, store, Vec::new(), has_no_ec_shards, true).0 } +/// Returns the heartbeat to send and, separately, every volume held. The +/// caller derives mount and unmount deltas by diffing successive snapshots, so +/// it must not be handed the partial list a heartbeat may carry. fn build_heartbeat_with_ec_status( config: &HeartbeatConfig, store: &mut Store, deleted_ec_shards: Vec, has_no_ec_shards: bool, -) -> master_pb::Heartbeat { + commit_report: bool, +) -> (master_pb::Heartbeat, Vec) { const MAX_TTL_VOLUME_REMOVAL_DELAY: u32 = 10; #[derive(Default)] @@ -801,10 +834,13 @@ fn build_heartbeat_with_ec_status( } let mut volumes = Vec::new(); - // Digest of exactly what this heartbeat reports, so the master can tell - // whether its copy is current. Volumes skipped above -- quarantined, - // phantom, expired -- are absent from both the list and the digest. + // Covers every volume held, whether or not this heartbeat names it, so the + // master can tell whether applying what it was sent leaves it current. + // Volumes skipped below -- quarantined, phantom, expired -- are in neither. let mut volume_digest: u64 = 0; + let (send_full_list, report_generation) = store.volume_report.begin(); + let mut reported_hashes: HashMap = HashMap::new(); + let mut changed_volumes = Vec::new(); let mut max_file_key = NeedleId(0); let mut max_volume_counts: HashMap = HashMap::new(); let mut disk_total_bytes: HashMap = HashMap::new(); @@ -898,7 +934,13 @@ fn build_heartbeat_with_ec_status( remote_storage_name, remote_storage_key, }; - volume_digest ^= report_hash(&volume_message); + let hash = report_hash(&volume_message); + volume_digest ^= hash; + let key: VolumeReportKey = (volume_message.disk_id, volume_message.id); + reported_hashes.insert(key, hash); + if send_full_list || store.volume_report.changed(key, hash) { + changed_volumes.push(volume_message.clone()); + } volumes.push(volume_message); } else if vol.is_expired_long_enough(MAX_TTL_VOLUME_REMOVAL_DELAY) { delete_vids.push(vol.id); @@ -960,10 +1002,23 @@ fn build_heartbeat_with_ec_status( let total_max: i64 = max_volume_counts.values().map(|v| *v as i64).sum(); crate::metrics::MAX_VOLUMES.set(total_max); - let has_no_volumes = volumes.is_empty(); + // Only when this heartbeat is going to be sent: marking volumes reported + // and then discarding the message would leave the master never told. + if commit_report { + store.volume_report.commit(reported_hashes, report_generation); + } + + // has_no_volumes says the server holds nothing, so it may only be derived + // from a full list. Deriving it from a changed-only heartbeat would make a + // quiet one read as an empty server and drop every volume on it. + let (heartbeat_volumes, changed_volumes, has_no_volumes) = if send_full_list { + (changed_volumes, Vec::new(), volumes.is_empty()) + } else { + (Vec::new(), changed_volumes, false) + }; let (location_uuids, disk_tags) = collect_location_metadata(store, &disk_max_by_id); - master_pb::Heartbeat { + let heartbeat = master_pb::Heartbeat { id: store.id.clone(), ip: config.ip.clone(), port: config.port as u32, @@ -972,7 +1027,8 @@ fn build_heartbeat_with_ec_status( data_center: config.data_center.clone(), rack: config.rack.clone(), admin_port: config.port as u32, - volumes, + volumes: heartbeat_volumes, + changed_volumes, volume_digest: Some(volume_digest), deleted_ec_shards, has_no_volumes, @@ -984,7 +1040,8 @@ fn build_heartbeat_with_ec_status( location_uuids, disk_tags, ..Default::default() - } + }; + (heartbeat, volumes) } fn collect_live_ec_shards( @@ -1257,6 +1314,156 @@ mod tests { // reported but left out of the digest, or the reverse, makes the master's // comparison disagree forever. An empty store still reports a digest, so // the master can tell it from a server that computes none. + fn reporting_store(dir: &str, count: u32) -> Store { + let mut store = Store::new(NeedleMapKind::InMemory); + store + .add_location( + dir, + dir, + 16, + DiskType::HardDrive, + MinFreeSpace::Percent(1.0), + Vec::new(), + ) + .unwrap(); + for id in 1..=count { + store + .add_volume( + VolumeId(id), + "pics", + None, + None, + 0, + DiskType::HardDrive, + Version::current(), + ) + .unwrap(); + } + store.volume_report.reset(); + store + } + + // Until the master says it compares digests it may be one that reads a + // partial list as the whole truth, so it keeps getting the whole list. + #[test] + fn test_heartbeat_sends_full_list_until_the_master_accepts() { + let temp_dir = tempfile::tempdir().unwrap(); + let mut store = reporting_store(temp_dir.path().to_str().unwrap(), 2); + + for _ in 0..3 { + let heartbeat = build_heartbeat(&test_config(), &mut store); + assert_eq!(heartbeat.volumes.len(), 2); + assert!(heartbeat.changed_volumes.is_empty()); + } + } + + // The one that would be catastrophic: a heartbeat with nothing to report + // must not look like a server that has lost every volume. + #[test] + fn test_quiet_heartbeat_does_not_look_like_an_empty_server() { + let temp_dir = tempfile::tempdir().unwrap(); + let mut store = reporting_store(temp_dir.path().to_str().unwrap(), 2); + store.volume_report.accept_deltas(); + let full = build_heartbeat(&test_config(), &mut store); + + let quiet = build_heartbeat(&test_config(), &mut store); + assert!(quiet.volumes.is_empty()); + assert!(quiet.changed_volumes.is_empty()); + assert!(!quiet.has_no_volumes); + // The digest still covers everything held, not just what was sent. + assert_eq!(quiet.volume_digest, full.volume_digest); + } + + #[test] + fn test_heartbeat_reports_only_what_changed() { + let temp_dir = tempfile::tempdir().unwrap(); + let mut store = reporting_store(temp_dir.path().to_str().unwrap(), 2); + store.volume_report.accept_deltas(); + build_heartbeat(&test_config(), &mut store); + + store + .add_volume( + VolumeId(3), + "pics", + None, + None, + 0, + DiskType::HardDrive, + Version::current(), + ) + .unwrap(); + + let heartbeat = build_heartbeat(&test_config(), &mut store); + assert!(heartbeat.volumes.is_empty()); + assert_eq!(heartbeat.changed_volumes.len(), 1); + assert_eq!(heartbeat.changed_volumes[0].id, 3); + } + + #[test] + fn test_heartbeat_returns_to_the_full_list_on_request() { + let temp_dir = tempfile::tempdir().unwrap(); + let mut store = reporting_store(temp_dir.path().to_str().unwrap(), 2); + store.volume_report.accept_deltas(); + build_heartbeat(&test_config(), &mut store); + + store.volume_report.request_full_list(); + let resent = build_heartbeat(&test_config(), &mut store); + assert_eq!(resent.volumes.len(), 2); + assert!(resent.changed_volumes.is_empty()); + + let next = build_heartbeat(&test_config(), &mut store); + assert!(next.volumes.is_empty()); + } + + // A request that lands while a heartbeat is being built asked about a later + // state than that heartbeat carries, so it must survive being committed over. + #[test] + fn test_full_list_request_during_collection_survives() { + let temp_dir = tempfile::tempdir().unwrap(); + let mut store = reporting_store(temp_dir.path().to_str().unwrap(), 2); + store.volume_report.accept_deltas(); + build_heartbeat(&test_config(), &mut store); + + let (full, generation) = store.volume_report.begin(); + assert!(!full); + store.volume_report.request_full_list(); + store.volume_report.commit(HashMap::new(), generation); + + let heartbeat = build_heartbeat(&test_config(), &mut store); + assert_eq!(heartbeat.volumes.len(), 2); + } + + // Taking a snapshot must not mark volumes as told to a master that is + // getting a different message. + #[test] + fn test_snapshot_does_not_mark_volumes_reported() { + let temp_dir = tempfile::tempdir().unwrap(); + let mut store = reporting_store(temp_dir.path().to_str().unwrap(), 2); + store.volume_report.accept_deltas(); + build_heartbeat(&test_config(), &mut store); + + store + .add_volume( + VolumeId(3), + "pics", + None, + None, + 0, + DiskType::HardDrive, + Version::current(), + ) + .unwrap(); + + // What the notify path does: collect a snapshot, send a message of its own. + let snapshot = + build_heartbeat_with_ec_status(&test_config(), &mut store, Vec::new(), true, false).1; + assert_eq!(snapshot.len(), 3); + + let heartbeat = build_heartbeat(&test_config(), &mut store); + assert_eq!(heartbeat.changed_volumes.len(), 1); + assert_eq!(heartbeat.changed_volumes[0].id, 3); + } + #[test] fn test_build_heartbeat_digests_exactly_what_it_reports() { let temp_dir = tempfile::tempdir().unwrap(); diff --git a/seaweed-volume/src/storage/mod.rs b/seaweed-volume/src/storage/mod.rs index abe8ba285..9ded27ac0 100644 --- a/seaweed-volume/src/storage/mod.rs +++ b/seaweed-volume/src/storage/mod.rs @@ -10,4 +10,5 @@ pub mod super_block; pub mod types; pub mod volume; pub mod volume_idx_repair; +pub mod volume_report; pub mod volume_report_hash; diff --git a/seaweed-volume/src/storage/store.rs b/seaweed-volume/src/storage/store.rs index 48567d12a..beeca7e83 100644 --- a/seaweed-volume/src/storage/store.rs +++ b/seaweed-volume/src/storage/store.rs @@ -31,6 +31,7 @@ pub struct Store { pub public_url: String, pub data_center: String, pub rack: String, + pub volume_report: crate::storage::volume_report::VolumeReportState, } impl Store { @@ -45,6 +46,7 @@ impl Store { port: 0, grpc_port: 0, public_url: String::new(), + volume_report: Default::default(), data_center: String::new(), rack: String::new(), } diff --git a/seaweed-volume/src/storage/volume_report.rs b/seaweed-volume/src/storage/volume_report.rs new file mode 100644 index 000000000..aedf24f23 --- /dev/null +++ b/seaweed-volume/src/storage/volume_report.rs @@ -0,0 +1,73 @@ +//! Mirror of `weed/storage/store_volume_report.go`. + +use std::collections::HashMap; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::Mutex; + +/// Identifies one reported copy. Keyed by disk as well as id because a volume +/// id can be mounted on two disks, and reporting one of them would leave the +/// other's changes untold. +pub type VolumeReportKey = (u32, u32); + +/// Remembers what the master was last told about each volume, so a heartbeat +/// can carry only what moved since. +/// +/// Per-connection: a server that reconnects, or reaches a different master, +/// knows nothing about what that master holds and starts again from the full +/// list. The default has told no master anything, so it sends the whole list +/// until one accepts changes. +#[derive(Default)] +pub struct VolumeReportState { + /// Set once the master says it compares digests. Until then the whole list + /// goes every time, which is what an older master needs. + deltas_accepted: AtomicBool, + full_list_needed: AtomicBool, + /// Counts requests for the whole list, so one arriving while a heartbeat is + /// being built is not marked satisfied by it. + full_list_generation: AtomicU64, + last_reported: Mutex>, +} + +impl VolumeReportState { + /// Drops everything known about the master's view. + pub fn reset(&self) { + self.deltas_accepted.store(false, Ordering::Relaxed); + self.full_list_needed.store(true, Ordering::Relaxed); + self.full_list_generation.fetch_add(1, Ordering::Relaxed); + self.last_reported.lock().unwrap().clear(); + } + + pub fn accept_deltas(&self) { + self.deltas_accepted.store(true, Ordering::Relaxed); + } + + pub fn request_full_list(&self) { + self.full_list_needed.store(true, Ordering::Relaxed); + self.full_list_generation.fetch_add(1, Ordering::Relaxed); + } + + /// Reports whether this heartbeat must carry the whole list, and the + /// request it answers. + pub fn begin(&self) -> (bool, u64) { + let full = self.full_list_needed.load(Ordering::Relaxed) + || !self.deltas_accepted.load(Ordering::Relaxed); + (full, self.full_list_generation.load(Ordering::Relaxed)) + } + + /// Reports whether the master needs telling about this volume, given what + /// it was last told. + pub fn changed(&self, key: VolumeReportKey, hash: u64) -> bool { + self.last_reported.lock().unwrap().get(&key) != Some(&hash) + } + + /// Records what this heartbeat told the master. Volumes absent from + /// `reported` are forgotten, so one that comes back is reported again. + pub fn commit(&self, reported: HashMap, generation: u64) { + *self.last_reported.lock().unwrap() = reported; + // A request that arrived while this heartbeat was being built asked + // about a later state than it carries, so it stands. + if self.full_list_generation.load(Ordering::Relaxed) == generation { + self.full_list_needed.store(false, Ordering::Relaxed); + } + } +} diff --git a/weed/pb/master.proto b/weed/pb/master.proto index afffc7d06..841924727 100644 --- a/weed/pb/master.proto +++ b/weed/pb/master.proto @@ -111,6 +111,10 @@ message Heartbeat { // from servers that do not compute it, and distinct from a digest of 0, which // is what a server holding no volumes reports. optional uint64 volume_digest = 27; + // Volumes whose reported state changed since the last heartbeat, sent in + // 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; } message HeartbeatResponse { @@ -124,6 +128,9 @@ message HeartbeatResponse { // The master's view of this server's volumes disagrees with the reported // digest, so it needs the full volume list rather than changes alone. bool resend_full_volume_list = 8; + // 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; } message VolumeInformationMessage { diff --git a/weed/pb/master_pb/master.pb.go b/weed/pb/master_pb/master.pb.go index 03075dca8..ae3794697 100644 --- a/weed/pb/master_pb/master.pb.go +++ b/weed/pb/master_pb/master.pb.go @@ -125,9 +125,13 @@ type Heartbeat struct { // master check its copy is current without being sent the whole list. Absent // from servers that do not compute it, and distinct from a digest of 0, which // is what a server holding no volumes reports. - VolumeDigest *uint64 `protobuf:"varint,27,opt,name=volume_digest,json=volumeDigest,proto3,oneof" json:"volume_digest,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + VolumeDigest *uint64 `protobuf:"varint,27,opt,name=volume_digest,json=volumeDigest,proto3,oneof" json:"volume_digest,omitempty"` + // Volumes whose reported state changed since the last heartbeat, sent in + // 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 } func (x *Heartbeat) Reset() { @@ -328,6 +332,13 @@ func (x *Heartbeat) GetVolumeDigest() uint64 { return 0 } +func (x *Heartbeat) GetChangedVolumes() []*VolumeInformationMessage { + if x != nil { + return x.ChangedVolumes + } + return nil +} + 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"` @@ -340,8 +351,11 @@ type HeartbeatResponse struct { // The master's view of this server's volumes disagrees with the reported // digest, so it needs the full volume list rather than changes alone. ResendFullVolumeList bool `protobuf:"varint,8,opt,name=resend_full_volume_list,json=resendFullVolumeList,proto3" json:"resend_full_volume_list,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + // 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 } func (x *HeartbeatResponse) Reset() { @@ -430,6 +444,13 @@ func (x *HeartbeatResponse) GetResendFullVolumeList() bool { return false } +func (x *HeartbeatResponse) GetVolumeDigestSupported() bool { + if x != nil { + return x.VolumeDigestSupported + } + return false +} + type VolumeInformationMessage struct { state protoimpl.MessageState `protogen:"open.v1"` Id uint32 `protobuf:"varint,1,opt,name=id,proto3" json:"id,omitempty"` @@ -4591,7 +4612,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\"\xa2\v\n" + + "\x10max_volume_count\x18\x04 \x01(\x03R\x0emaxVolumeCount\"\xf0\v\n" + "\tHeartbeat\x12\x0e\n" + "\x02ip\x18\x01 \x01(\tR\x02ip\x12\x12\n" + "\x04port\x18\x02 \x01(\rR\x04port\x12\x1d\n" + @@ -4622,7 +4643,8 @@ const file_master_proto_rawDesc = "" + "\tdisk_tags\x18\x18 \x03(\v2\x12.master_pb.DiskTagR\bdiskTags\x12R\n" + "\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\x1aB\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" + "\x14MaxVolumeCountsEntry\x12\x10\n" + "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" + "\x05value\x18\x02 \x01(\rR\x05value:\x028\x01\x1aA\n" + @@ -4632,7 +4654,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\"\x84\x03\n" + + "\x0e_volume_digest\"\xbc\x03\n" + "\x11HeartbeatResponse\x12*\n" + "\x11volume_size_limit\x18\x01 \x01(\x04R\x0fvolumeSizeLimit\x12\x16\n" + "\x06leader\x18\x02 \x01(\tR\x06leader\x12'\n" + @@ -4641,7 +4663,8 @@ const file_master_proto_rawDesc = "" + "\x10storage_backends\x18\x05 \x03(\v2\x19.master_pb.StorageBackendR\x0fstorageBackends\x12)\n" + "\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\"\xb1\x04\n" + + "\x17resend_full_volume_list\x18\b \x01(\bR\x14resendFullVolumeList\x126\n" + + "\x17volume_digest_supported\x18\t \x01(\bR\x15volumeDigestSupported\"\xb1\x04\n" + "\x18VolumeInformationMessage\x12\x0e\n" + "\x02id\x18\x01 \x01(\rR\x02id\x12\x12\n" + "\x04size\x18\x02 \x01(\x04R\x04size\x12\x1e\n" + @@ -5134,92 +5157,93 @@ var file_master_proto_depIdxs = []int32{ 0, // 8: master_pb.Heartbeat.disk_tags:type_name -> master_pb.DiskTag 66, // 9: master_pb.Heartbeat.disk_total_bytes:type_name -> master_pb.Heartbeat.DiskTotalBytesEntry 67, // 10: master_pb.Heartbeat.disk_free_bytes:type_name -> master_pb.Heartbeat.DiskFreeBytesEntry - 6, // 11: master_pb.HeartbeatResponse.storage_backends:type_name -> master_pb.StorageBackend - 68, // 12: master_pb.StorageBackend.properties:type_name -> master_pb.StorageBackend.PropertiesEntry - 69, // 13: master_pb.SuperBlockExtra.erasure_coding:type_name -> master_pb.SuperBlockExtra.ErasureCoding - 10, // 14: master_pb.KeepConnectedResponse.volume_location:type_name -> master_pb.VolumeLocation - 11, // 15: master_pb.KeepConnectedResponse.cluster_node_update:type_name -> master_pb.ClusterNodeUpdate - 13, // 16: master_pb.KeepConnectedResponse.lock_ring_update:type_name -> master_pb.LockRingUpdate - 70, // 17: master_pb.LookupVolumeResponse.volume_id_locations:type_name -> master_pb.LookupVolumeResponse.VolumeIdLocation - 16, // 18: master_pb.AssignResponse.replicas:type_name -> master_pb.Location - 16, // 19: master_pb.AssignResponse.location:type_name -> master_pb.Location - 22, // 20: master_pb.CollectionListResponse.collections:type_name -> master_pb.Collection - 3, // 21: master_pb.DiskInfo.volume_infos:type_name -> master_pb.VolumeInformationMessage - 5, // 22: master_pb.DiskInfo.ec_shard_infos:type_name -> master_pb.VolumeEcShardInformationMessage - 71, // 23: master_pb.DiskInfo.max_volume_count_by_disk:type_name -> master_pb.DiskInfo.MaxVolumeCountByDiskEntry - 72, // 24: master_pb.DataNodeInfo.diskInfos:type_name -> master_pb.DataNodeInfo.DiskInfosEntry - 28, // 25: master_pb.RackInfo.data_node_infos:type_name -> master_pb.DataNodeInfo - 73, // 26: master_pb.RackInfo.diskInfos:type_name -> master_pb.RackInfo.DiskInfosEntry - 29, // 27: master_pb.DataCenterInfo.rack_infos:type_name -> master_pb.RackInfo - 74, // 28: master_pb.DataCenterInfo.diskInfos:type_name -> master_pb.DataCenterInfo.DiskInfosEntry - 30, // 29: master_pb.TopologyInfo.data_center_infos:type_name -> master_pb.DataCenterInfo - 75, // 30: master_pb.TopologyInfo.diskInfos:type_name -> master_pb.TopologyInfo.DiskInfosEntry - 31, // 31: master_pb.VolumeListResponse.topology_info:type_name -> master_pb.TopologyInfo - 76, // 32: master_pb.LookupEcVolumeResponse.shard_id_locations:type_name -> master_pb.LookupEcVolumeResponse.EcShardIdLocation - 6, // 33: master_pb.GetMasterConfigurationResponse.storage_backends:type_name -> master_pb.StorageBackend - 77, // 34: master_pb.ListClusterNodesResponse.cluster_nodes:type_name -> master_pb.ListClusterNodesResponse.ClusterNode - 78, // 35: master_pb.RaftListClusterServersResponse.cluster_servers:type_name -> master_pb.RaftListClusterServersResponse.ClusterServers - 16, // 36: master_pb.LookupVolumeResponse.VolumeIdLocation.locations:type_name -> master_pb.Location - 27, // 37: master_pb.DataNodeInfo.DiskInfosEntry.value:type_name -> master_pb.DiskInfo - 27, // 38: master_pb.RackInfo.DiskInfosEntry.value:type_name -> master_pb.DiskInfo - 27, // 39: master_pb.DataCenterInfo.DiskInfosEntry.value:type_name -> master_pb.DiskInfo - 27, // 40: master_pb.TopologyInfo.DiskInfosEntry.value:type_name -> master_pb.DiskInfo - 16, // 41: master_pb.LookupEcVolumeResponse.EcShardIdLocation.locations:type_name -> master_pb.Location - 1, // 42: master_pb.Seaweed.SendHeartbeat:input_type -> master_pb.Heartbeat - 9, // 43: master_pb.Seaweed.KeepConnected:input_type -> master_pb.KeepConnectedRequest - 14, // 44: master_pb.Seaweed.LookupVolume:input_type -> master_pb.LookupVolumeRequest - 17, // 45: master_pb.Seaweed.Assign:input_type -> master_pb.AssignRequest - 17, // 46: master_pb.Seaweed.StreamAssign:input_type -> master_pb.AssignRequest - 20, // 47: master_pb.Seaweed.Statistics:input_type -> master_pb.StatisticsRequest - 23, // 48: master_pb.Seaweed.CollectionList:input_type -> master_pb.CollectionListRequest - 25, // 49: master_pb.Seaweed.CollectionDelete:input_type -> master_pb.CollectionDeleteRequest - 32, // 50: master_pb.Seaweed.VolumeList:input_type -> master_pb.VolumeListRequest - 34, // 51: master_pb.Seaweed.LookupEcVolume:input_type -> master_pb.LookupEcVolumeRequest - 36, // 52: master_pb.Seaweed.VacuumVolume:input_type -> master_pb.VacuumVolumeRequest - 38, // 53: master_pb.Seaweed.DisableVacuum:input_type -> master_pb.DisableVacuumRequest - 40, // 54: master_pb.Seaweed.EnableVacuum:input_type -> master_pb.EnableVacuumRequest - 42, // 55: master_pb.Seaweed.VolumeMarkReadonly:input_type -> master_pb.VolumeMarkReadonlyRequest - 44, // 56: master_pb.Seaweed.GetMasterConfiguration:input_type -> master_pb.GetMasterConfigurationRequest - 46, // 57: master_pb.Seaweed.ListClusterNodes:input_type -> master_pb.ListClusterNodesRequest - 48, // 58: master_pb.Seaweed.LeaseAdminToken:input_type -> master_pb.LeaseAdminTokenRequest - 50, // 59: master_pb.Seaweed.ReleaseAdminToken:input_type -> master_pb.ReleaseAdminTokenRequest - 52, // 60: master_pb.Seaweed.GetAdminLockStatus:input_type -> master_pb.GetAdminLockStatusRequest - 54, // 61: master_pb.Seaweed.Ping:input_type -> master_pb.PingRequest - 60, // 62: master_pb.Seaweed.RaftListClusterServers:input_type -> master_pb.RaftListClusterServersRequest - 56, // 63: master_pb.Seaweed.RaftAddServer:input_type -> master_pb.RaftAddServerRequest - 58, // 64: master_pb.Seaweed.RaftRemoveServer:input_type -> master_pb.RaftRemoveServerRequest - 62, // 65: master_pb.Seaweed.RaftLeadershipTransfer:input_type -> master_pb.RaftLeadershipTransferRequest - 18, // 66: master_pb.Seaweed.VolumeGrow:input_type -> master_pb.VolumeGrowRequest - 2, // 67: master_pb.Seaweed.SendHeartbeat:output_type -> master_pb.HeartbeatResponse - 12, // 68: master_pb.Seaweed.KeepConnected:output_type -> master_pb.KeepConnectedResponse - 15, // 69: master_pb.Seaweed.LookupVolume:output_type -> master_pb.LookupVolumeResponse - 19, // 70: master_pb.Seaweed.Assign:output_type -> master_pb.AssignResponse - 19, // 71: master_pb.Seaweed.StreamAssign:output_type -> master_pb.AssignResponse - 21, // 72: master_pb.Seaweed.Statistics:output_type -> master_pb.StatisticsResponse - 24, // 73: master_pb.Seaweed.CollectionList:output_type -> master_pb.CollectionListResponse - 26, // 74: master_pb.Seaweed.CollectionDelete:output_type -> master_pb.CollectionDeleteResponse - 33, // 75: master_pb.Seaweed.VolumeList:output_type -> master_pb.VolumeListResponse - 35, // 76: master_pb.Seaweed.LookupEcVolume:output_type -> master_pb.LookupEcVolumeResponse - 37, // 77: master_pb.Seaweed.VacuumVolume:output_type -> master_pb.VacuumVolumeResponse - 39, // 78: master_pb.Seaweed.DisableVacuum:output_type -> master_pb.DisableVacuumResponse - 41, // 79: master_pb.Seaweed.EnableVacuum:output_type -> master_pb.EnableVacuumResponse - 43, // 80: master_pb.Seaweed.VolumeMarkReadonly:output_type -> master_pb.VolumeMarkReadonlyResponse - 45, // 81: master_pb.Seaweed.GetMasterConfiguration:output_type -> master_pb.GetMasterConfigurationResponse - 47, // 82: master_pb.Seaweed.ListClusterNodes:output_type -> master_pb.ListClusterNodesResponse - 49, // 83: master_pb.Seaweed.LeaseAdminToken:output_type -> master_pb.LeaseAdminTokenResponse - 51, // 84: master_pb.Seaweed.ReleaseAdminToken:output_type -> master_pb.ReleaseAdminTokenResponse - 53, // 85: master_pb.Seaweed.GetAdminLockStatus:output_type -> master_pb.GetAdminLockStatusResponse - 55, // 86: master_pb.Seaweed.Ping:output_type -> master_pb.PingResponse - 61, // 87: master_pb.Seaweed.RaftListClusterServers:output_type -> master_pb.RaftListClusterServersResponse - 57, // 88: master_pb.Seaweed.RaftAddServer:output_type -> master_pb.RaftAddServerResponse - 59, // 89: master_pb.Seaweed.RaftRemoveServer:output_type -> master_pb.RaftRemoveServerResponse - 63, // 90: master_pb.Seaweed.RaftLeadershipTransfer:output_type -> master_pb.RaftLeadershipTransferResponse - 64, // 91: master_pb.Seaweed.VolumeGrow:output_type -> master_pb.VolumeGrowResponse - 67, // [67:92] is the sub-list for method output_type - 42, // [42:67] is the sub-list for method input_type - 42, // [42:42] is the sub-list for extension type_name - 42, // [42:42] is the sub-list for extension extendee - 0, // [0:42] is the sub-list for field type_name + 3, // 11: master_pb.Heartbeat.changed_volumes:type_name -> master_pb.VolumeInformationMessage + 6, // 12: master_pb.HeartbeatResponse.storage_backends:type_name -> master_pb.StorageBackend + 68, // 13: master_pb.StorageBackend.properties:type_name -> master_pb.StorageBackend.PropertiesEntry + 69, // 14: master_pb.SuperBlockExtra.erasure_coding:type_name -> master_pb.SuperBlockExtra.ErasureCoding + 10, // 15: master_pb.KeepConnectedResponse.volume_location:type_name -> master_pb.VolumeLocation + 11, // 16: master_pb.KeepConnectedResponse.cluster_node_update:type_name -> master_pb.ClusterNodeUpdate + 13, // 17: master_pb.KeepConnectedResponse.lock_ring_update:type_name -> master_pb.LockRingUpdate + 70, // 18: master_pb.LookupVolumeResponse.volume_id_locations:type_name -> master_pb.LookupVolumeResponse.VolumeIdLocation + 16, // 19: master_pb.AssignResponse.replicas:type_name -> master_pb.Location + 16, // 20: master_pb.AssignResponse.location:type_name -> master_pb.Location + 22, // 21: master_pb.CollectionListResponse.collections:type_name -> master_pb.Collection + 3, // 22: master_pb.DiskInfo.volume_infos:type_name -> master_pb.VolumeInformationMessage + 5, // 23: master_pb.DiskInfo.ec_shard_infos:type_name -> master_pb.VolumeEcShardInformationMessage + 71, // 24: master_pb.DiskInfo.max_volume_count_by_disk:type_name -> master_pb.DiskInfo.MaxVolumeCountByDiskEntry + 72, // 25: master_pb.DataNodeInfo.diskInfos:type_name -> master_pb.DataNodeInfo.DiskInfosEntry + 28, // 26: master_pb.RackInfo.data_node_infos:type_name -> master_pb.DataNodeInfo + 73, // 27: master_pb.RackInfo.diskInfos:type_name -> master_pb.RackInfo.DiskInfosEntry + 29, // 28: master_pb.DataCenterInfo.rack_infos:type_name -> master_pb.RackInfo + 74, // 29: master_pb.DataCenterInfo.diskInfos:type_name -> master_pb.DataCenterInfo.DiskInfosEntry + 30, // 30: master_pb.TopologyInfo.data_center_infos:type_name -> master_pb.DataCenterInfo + 75, // 31: master_pb.TopologyInfo.diskInfos:type_name -> master_pb.TopologyInfo.DiskInfosEntry + 31, // 32: master_pb.VolumeListResponse.topology_info:type_name -> master_pb.TopologyInfo + 76, // 33: master_pb.LookupEcVolumeResponse.shard_id_locations:type_name -> master_pb.LookupEcVolumeResponse.EcShardIdLocation + 6, // 34: master_pb.GetMasterConfigurationResponse.storage_backends:type_name -> master_pb.StorageBackend + 77, // 35: master_pb.ListClusterNodesResponse.cluster_nodes:type_name -> master_pb.ListClusterNodesResponse.ClusterNode + 78, // 36: master_pb.RaftListClusterServersResponse.cluster_servers:type_name -> master_pb.RaftListClusterServersResponse.ClusterServers + 16, // 37: master_pb.LookupVolumeResponse.VolumeIdLocation.locations:type_name -> master_pb.Location + 27, // 38: master_pb.DataNodeInfo.DiskInfosEntry.value:type_name -> master_pb.DiskInfo + 27, // 39: master_pb.RackInfo.DiskInfosEntry.value:type_name -> master_pb.DiskInfo + 27, // 40: master_pb.DataCenterInfo.DiskInfosEntry.value:type_name -> master_pb.DiskInfo + 27, // 41: master_pb.TopologyInfo.DiskInfosEntry.value:type_name -> master_pb.DiskInfo + 16, // 42: master_pb.LookupEcVolumeResponse.EcShardIdLocation.locations:type_name -> master_pb.Location + 1, // 43: master_pb.Seaweed.SendHeartbeat:input_type -> master_pb.Heartbeat + 9, // 44: master_pb.Seaweed.KeepConnected:input_type -> master_pb.KeepConnectedRequest + 14, // 45: master_pb.Seaweed.LookupVolume:input_type -> master_pb.LookupVolumeRequest + 17, // 46: master_pb.Seaweed.Assign:input_type -> master_pb.AssignRequest + 17, // 47: master_pb.Seaweed.StreamAssign:input_type -> master_pb.AssignRequest + 20, // 48: master_pb.Seaweed.Statistics:input_type -> master_pb.StatisticsRequest + 23, // 49: master_pb.Seaweed.CollectionList:input_type -> master_pb.CollectionListRequest + 25, // 50: master_pb.Seaweed.CollectionDelete:input_type -> master_pb.CollectionDeleteRequest + 32, // 51: master_pb.Seaweed.VolumeList:input_type -> master_pb.VolumeListRequest + 34, // 52: master_pb.Seaweed.LookupEcVolume:input_type -> master_pb.LookupEcVolumeRequest + 36, // 53: master_pb.Seaweed.VacuumVolume:input_type -> master_pb.VacuumVolumeRequest + 38, // 54: master_pb.Seaweed.DisableVacuum:input_type -> master_pb.DisableVacuumRequest + 40, // 55: master_pb.Seaweed.EnableVacuum:input_type -> master_pb.EnableVacuumRequest + 42, // 56: master_pb.Seaweed.VolumeMarkReadonly:input_type -> master_pb.VolumeMarkReadonlyRequest + 44, // 57: master_pb.Seaweed.GetMasterConfiguration:input_type -> master_pb.GetMasterConfigurationRequest + 46, // 58: master_pb.Seaweed.ListClusterNodes:input_type -> master_pb.ListClusterNodesRequest + 48, // 59: master_pb.Seaweed.LeaseAdminToken:input_type -> master_pb.LeaseAdminTokenRequest + 50, // 60: master_pb.Seaweed.ReleaseAdminToken:input_type -> master_pb.ReleaseAdminTokenRequest + 52, // 61: master_pb.Seaweed.GetAdminLockStatus:input_type -> master_pb.GetAdminLockStatusRequest + 54, // 62: master_pb.Seaweed.Ping:input_type -> master_pb.PingRequest + 60, // 63: master_pb.Seaweed.RaftListClusterServers:input_type -> master_pb.RaftListClusterServersRequest + 56, // 64: master_pb.Seaweed.RaftAddServer:input_type -> master_pb.RaftAddServerRequest + 58, // 65: master_pb.Seaweed.RaftRemoveServer:input_type -> master_pb.RaftRemoveServerRequest + 62, // 66: master_pb.Seaweed.RaftLeadershipTransfer:input_type -> master_pb.RaftLeadershipTransferRequest + 18, // 67: master_pb.Seaweed.VolumeGrow:input_type -> master_pb.VolumeGrowRequest + 2, // 68: master_pb.Seaweed.SendHeartbeat:output_type -> master_pb.HeartbeatResponse + 12, // 69: master_pb.Seaweed.KeepConnected:output_type -> master_pb.KeepConnectedResponse + 15, // 70: master_pb.Seaweed.LookupVolume:output_type -> master_pb.LookupVolumeResponse + 19, // 71: master_pb.Seaweed.Assign:output_type -> master_pb.AssignResponse + 19, // 72: master_pb.Seaweed.StreamAssign:output_type -> master_pb.AssignResponse + 21, // 73: master_pb.Seaweed.Statistics:output_type -> master_pb.StatisticsResponse + 24, // 74: master_pb.Seaweed.CollectionList:output_type -> master_pb.CollectionListResponse + 26, // 75: master_pb.Seaweed.CollectionDelete:output_type -> master_pb.CollectionDeleteResponse + 33, // 76: master_pb.Seaweed.VolumeList:output_type -> master_pb.VolumeListResponse + 35, // 77: master_pb.Seaweed.LookupEcVolume:output_type -> master_pb.LookupEcVolumeResponse + 37, // 78: master_pb.Seaweed.VacuumVolume:output_type -> master_pb.VacuumVolumeResponse + 39, // 79: master_pb.Seaweed.DisableVacuum:output_type -> master_pb.DisableVacuumResponse + 41, // 80: master_pb.Seaweed.EnableVacuum:output_type -> master_pb.EnableVacuumResponse + 43, // 81: master_pb.Seaweed.VolumeMarkReadonly:output_type -> master_pb.VolumeMarkReadonlyResponse + 45, // 82: master_pb.Seaweed.GetMasterConfiguration:output_type -> master_pb.GetMasterConfigurationResponse + 47, // 83: master_pb.Seaweed.ListClusterNodes:output_type -> master_pb.ListClusterNodesResponse + 49, // 84: master_pb.Seaweed.LeaseAdminToken:output_type -> master_pb.LeaseAdminTokenResponse + 51, // 85: master_pb.Seaweed.ReleaseAdminToken:output_type -> master_pb.ReleaseAdminTokenResponse + 53, // 86: master_pb.Seaweed.GetAdminLockStatus:output_type -> master_pb.GetAdminLockStatusResponse + 55, // 87: master_pb.Seaweed.Ping:output_type -> master_pb.PingResponse + 61, // 88: master_pb.Seaweed.RaftListClusterServers:output_type -> master_pb.RaftListClusterServersResponse + 57, // 89: master_pb.Seaweed.RaftAddServer:output_type -> master_pb.RaftAddServerResponse + 59, // 90: master_pb.Seaweed.RaftRemoveServer:output_type -> master_pb.RaftRemoveServerResponse + 63, // 91: master_pb.Seaweed.RaftLeadershipTransfer:output_type -> master_pb.RaftLeadershipTransferResponse + 64, // 92: master_pb.Seaweed.VolumeGrow:output_type -> master_pb.VolumeGrowResponse + 68, // [68:93] is the sub-list for method output_type + 43, // [43:68] is the sub-list for method input_type + 43, // [43:43] is the sub-list for extension type_name + 43, // [43:43] is the sub-list for extension extendee + 0, // [0:43] is the sub-list for field type_name } func init() { file_master_proto_init() } diff --git a/weed/server/master_grpc_server.go b/weed/server/master_grpc_server.go index 37b5e468c..bd632699e 100644 --- a/weed/server/master_grpc_server.go +++ b/weed/server/master_grpc_server.go @@ -38,6 +38,16 @@ func shouldBroadcastVolumeRemoval(dn *topology.DataNode, vid needle.VolumeId) bo return err != nil } +// heartbeatResponse carries the options a volume server takes from every +// response it receives. A response that left them out would be read as the +// master turning them off, so anything sent mid-stream has to start here. +func (ms *MasterServer) heartbeatResponse() *master_pb.HeartbeatResponse { + return &master_pb.HeartbeatResponse{ + VolumeSizeLimit: uint64(ms.option.VolumeSizeLimitMB) * 1024 * 1024, + Preallocate: ms.preallocateSize > 0, + } +} + func (ms *MasterServer) RegisterUuids(heartbeat *master_pb.Heartbeat) (duplicated_uuids []string, err error) { ms.Topo.UuidAccessLock.Lock() defer ms.Topo.UuidAccessLock.Unlock() @@ -170,10 +180,9 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ return err } - if err := stream.Send(&master_pb.HeartbeatResponse{ - VolumeSizeLimit: uint64(ms.option.VolumeSizeLimitMB) * 1024 * 1024, - Preallocate: ms.preallocateSize > 0, - }); err != nil { + response := ms.heartbeatResponse() + response.VolumeDigestSupported = true + if err := stream.Send(response); err != nil { glog.Warningf("SendHeartbeat.Send volume size to %s:%d %v", dn.Ip, dn.Port, err) return err } @@ -232,6 +241,13 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ } } + if len(heartbeat.ChangedVolumes) > 0 { + stats.MasterReceivedHeartbeatCounter.WithLabelValues("changedVolumes").Inc() + for _, v := range ms.Topo.ApplyVolumeChanges(heartbeat.ChangedVolumes, dn) { + message.NewVids = append(message.NewVids, uint32(v.Id)) + } + } + if len(heartbeat.Volumes) > 0 || heartbeat.HasNoVolumes { if heartbeat.Ip != "" { dcName, rackName := ms.Topo.Configuration.Locate(heartbeat.Ip, heartbeat.DataCenter, heartbeat.Rack) @@ -296,7 +312,9 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ // Checked after everything the heartbeat carried has been applied, so a // match means the master is current, not that nothing changed. if resend := ms.checkVolumeDigest(heartbeat, dn); resend { - if err := stream.Send(&master_pb.HeartbeatResponse{ResendFullVolumeList: true}); err != nil { + response := ms.heartbeatResponse() + response.ResendFullVolumeList = true + if err := stream.Send(response); err != nil { glog.Warningf("SendHeartbeat.Send resend request to %s:%d %v", dn.Ip, dn.Port, err) return err } @@ -312,32 +330,43 @@ func (ms *MasterServer) checkVolumeDigest(heartbeat *master_pb.Heartbeat, dn *to if heartbeat.VolumeDigest == nil { return false } - // One volume id mounted on two disks is reported twice but stored once, so - // the digests cannot agree however often the list is resent. Asking would - // loop forever. - if dn.HasDuplicateVolumeIds() { - stats.MasterReceivedHeartbeatCounter.WithLabelValues("volumeDigestNotComparable").Inc() - return false - } reported := heartbeat.GetVolumeDigest() held := dn.VolumeDigest() - if held == reported { + needsFullList, reason := true, "" + switch { + case dn.HasDuplicateVolumeIds(): + // Reported twice but stored once, so the digests can never agree. The + // server has to keep sending its whole list, since nothing else would + // tell the master what it had stopped holding. + stats.MasterReceivedHeartbeatCounter.WithLabelValues("volumeDigestNotComparable").Inc() + case !dn.HasConsistentVolumeIndex(): + // The lookup index has drifted from the disks, which the server cannot + // see and its digest cannot show. Only a full report re-registers the + // volumes that stopped being servable. + stats.MasterReceivedHeartbeatCounter.WithLabelValues("volumeIndexInconsistent").Inc() + reason = "lookup index disagrees with the volumes held" + case held != reported: + stats.MasterReceivedHeartbeatCounter.WithLabelValues("volumeDigestMismatch").Inc() + reason = fmt.Sprintf("reported digest %d, master holds %d", reported, held) + default: stats.MasterReceivedHeartbeatCounter.WithLabelValues("volumeDigestMatch").Inc() + needsFullList = false + } + if !needsFullList { return false } - stats.MasterReceivedHeartbeatCounter.WithLabelValues("volumeDigestMismatch").Inc() - // A heartbeat that already carried the full list has nothing more to give; - // asking again would just repeat. Say so instead, because at this point the - // two ends genuinely disagree about what the server holds. + // A heartbeat that already carried the full list has nothing more to give. if len(heartbeat.Volumes) > 0 || heartbeat.HasNoVolumes { - glog.Warningf("volume server %s reported digest %d after a full volume list, master holds %d", - dn.Url(), reported, held) + if reason != "" { + glog.Warningf("volume server %s still disagrees after a full volume list: %s", dn.Url(), reason) + } return false } - glog.V(0).Infof("volume server %s reported digest %d, master holds %d: requesting the full volume list", - dn.Url(), reported, held) + if reason != "" { + glog.V(0).Infof("volume server %s: %s, requesting the full volume list", dn.Url(), reason) + } return true } diff --git a/weed/server/master_grpc_server_changed_volumes_test.go b/weed/server/master_grpc_server_changed_volumes_test.go new file mode 100644 index 000000000..42fee2ded --- /dev/null +++ b/weed/server/master_grpc_server_changed_volumes_test.go @@ -0,0 +1,147 @@ +package weed_server + +import ( + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "github.com/seaweedfs/seaweedfs/weed/sequence" + "github.com/seaweedfs/seaweedfs/weed/storage" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/super_block" + "github.com/seaweedfs/seaweedfs/weed/storage/types" + "github.com/seaweedfs/seaweedfs/weed/topology" +) + +func changedTestCluster(t *testing.T) (*topology.Topology, *topology.DataNode) { + t.Helper() + topo := topology.NewTopology("test", sequence.NewMemorySequencer(), 32*1024*1024*1024, 5, false) + dn := topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1"). + GetOrCreateDataNode("127.0.0.1", 8080, 18080, "", "", map[string]uint32{"": 100}) + return topo, dn +} + +func changedTestVolume(id uint32, size uint64) *master_pb.VolumeInformationMessage { + return &master_pb.VolumeInformationMessage{ + Id: id, Size: size, Collection: "c", Version: 3, FileCount: 1, + } +} + +// Applying only the volumes a heartbeat named has to leave the master holding +// what the server holds, or the digest it is checked against means nothing. +func TestChangedVolumesBringTheMasterCurrent(t *testing.T) { + topo, dn := changedTestCluster(t) + topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{ + changedTestVolume(1, 1024), changedTestVolume(2, 1024), changedTestVolume(3, 1024), + }, dn) + + grown := changedTestVolume(2, 8192) + topo.ApplyVolumeChanges([]*master_pb.VolumeInformationMessage{grown}, dn) + + reference, referenceNode := changedTestCluster(t) + reference.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{ + changedTestVolume(1, 1024), grown, changedTestVolume(3, 1024), + }, referenceNode) + + if dn.VolumeDigest() != referenceNode.VolumeDigest() { + t.Errorf("after applying the change the master digests %d, the server reports %d", + dn.VolumeDigest(), referenceNode.VolumeDigest()) + } +} + +// Silence about a volume in a changed-only heartbeat says nothing about whether +// the server still has it, unlike a full report. +func TestChangedVolumesDoNotRemoveUnmentionedVolumes(t *testing.T) { + topo, dn := changedTestCluster(t) + topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{ + changedTestVolume(1, 1024), changedTestVolume(2, 1024), + }, dn) + + topo.ApplyVolumeChanges([]*master_pb.VolumeInformationMessage{changedTestVolume(1, 4096)}, dn) + + if _, err := dn.GetVolumesById(needle.VolumeId(2)); err != nil { + t.Errorf("a volume the heartbeat did not mention was dropped: %v", err) + } + if !dn.HasConsistentVolumeIndex() { + t.Error("applying changes left the lookup index disagreeing with the disks") + } +} + +// A volume the master has never seen can arrive as a change. +func TestChangedVolumesRegisterUnknownVolumes(t *testing.T) { + topo, dn := changedTestCluster(t) + topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{changedTestVolume(1, 1024)}, dn) + + topo.ApplyVolumeChanges([]*master_pb.VolumeInformationMessage{changedTestVolume(9, 1024)}, dn) + + if locations := topo.Lookup("c", needle.VolumeId(9)); len(locations) != 1 { + t.Errorf("a volume first seen as a change is not servable: %v", locations) + } + if !dn.HasConsistentVolumeIndex() { + t.Error("applying changes left the lookup index disagreeing with the disks") + } +} + +// Volumes grow constantly, and a growth moves no location. Telling every +// connected client about each one would flood bounded broadcast queues and push +// out the topology updates that do matter. +func TestChangedVolumesAnnounceOnlyArrivals(t *testing.T) { + topo, dn := changedTestCluster(t) + topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{ + changedTestVolume(1, 1024), changedTestVolume(2, 1024), + }, dn) + + grown := topo.ApplyVolumeChanges([]*master_pb.VolumeInformationMessage{ + changedTestVolume(1, 8192), changedTestVolume(2, 9216), + }, dn) + if len(grown) != 0 { + t.Errorf("volumes that only grew were announced as new locations: %v", grown) + } + + arrived := topo.ApplyVolumeChanges([]*master_pb.VolumeInformationMessage{ + changedTestVolume(1, 16384), changedTestVolume(7, 1024), + }, dn) + if len(arrived) != 1 || arrived[0].Id != needle.VolumeId(7) { + t.Errorf("expected only the volume that arrived, got %v", arrived) + } +} + +// A node dropping out tells clients its volumes went. If the lookup index then +// loses one while the disk map keeps it, repairing the index has to announce it +// as well: nothing else will, and the clients that heard it go would never hear +// otherwise. +func TestRepairedLookupEntryIsAnnounced(t *testing.T) { + for _, tc := range []struct { + name string + apply func(*topology.Topology, *topology.DataNode, *master_pb.VolumeInformationMessage) []storage.VolumeInfo + }{ + {"ViaChanges", func(topo *topology.Topology, dn *topology.DataNode, v *master_pb.VolumeInformationMessage) []storage.VolumeInfo { + return topo.ApplyVolumeChanges([]*master_pb.VolumeInformationMessage{v}, dn) + }}, + {"ViaFullList", func(topo *topology.Topology, dn *topology.DataNode, v *master_pb.VolumeInformationMessage) []storage.VolumeInfo { + announced, _ := topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{v}, dn) + return announced + }}, + } { + t.Run(tc.name, func(t *testing.T) { + topo, dn := changedTestCluster(t) + volume := changedTestVolume(1, 1024) + topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{volume}, dn) + + rp, _ := super_block.NewReplicaPlacementFromString("000") + vl := topo.GetVolumeLayout("c", rp, needle.EMPTY_TTL, types.HardDriveType) + vl.SetVolumeUnavailable(dn, needle.VolumeId(1)) + if locations := topo.Lookup("c", needle.VolumeId(1)); locations != nil { + t.Fatalf("expected the volume to be unservable, got %v", locations) + } + + announced := tc.apply(topo, dn, changedTestVolume(1, 1024)) + + if locations := topo.Lookup("c", needle.VolumeId(1)); len(locations) != 1 { + t.Fatalf("the repair did not make the volume servable again: %v", locations) + } + if len(announced) != 1 || announced[0].Id != needle.VolumeId(1) { + t.Errorf("the repair was not announced to clients, so those told it went stay stale: %v", announced) + } + }) + } +} diff --git a/weed/server/master_grpc_server_digest_test.go b/weed/server/master_grpc_server_digest_test.go index 25847358e..2736f4189 100644 --- a/weed/server/master_grpc_server_digest_test.go +++ b/weed/server/master_grpc_server_digest_test.go @@ -5,6 +5,9 @@ import ( "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" "github.com/seaweedfs/seaweedfs/weed/sequence" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/super_block" + "github.com/seaweedfs/seaweedfs/weed/storage/types" "github.com/seaweedfs/seaweedfs/weed/topology" ) @@ -75,7 +78,10 @@ func TestDigestCheckAsksForTheListWhenADeltaDisagrees(t *testing.T) { // A node reporting one volume id twice is stored once, so the digests cannot // agree however often the list is resent. -func TestDigestCheckSkipsNodesWithDuplicateVolumeIds(t *testing.T) { +// A node reporting one volume id twice is stored once, so no digest can ever +// agree. It has to keep sending its whole list: nothing else would tell the +// master what it stopped holding. +func TestDigestCheckKeepsDuplicateNodesOnFullLists(t *testing.T) { ms, dn := digestTestCluster(t) duplicated := digestTestVolumeMessage(1) duplicated.DiskId = 1 @@ -83,8 +89,58 @@ func TestDigestCheckSkipsNodesWithDuplicateVolumeIds(t *testing.T) { digestTestVolumeMessage(1), duplicated, }, dn) - wrong := dn.VolumeDigest() ^ 1 - if ms.checkVolumeDigest(&master_pb.Heartbeat{VolumeDigest: &wrong}, dn) { - t.Error("a node the master cannot represent was asked to resend, which would repeat forever") + digest := dn.VolumeDigest() + if !ms.checkVolumeDigest(&master_pb.Heartbeat{VolumeDigest: &digest}, dn) { + t.Error("a node whose digest can never be verified was left sending only changes") + } + // And is not asked again for a list it just sent. + if ms.checkVolumeDigest(&master_pb.Heartbeat{ + Volumes: []*master_pb.VolumeInformationMessage{digestTestVolumeMessage(1), duplicated}, + VolumeDigest: &digest, + }, dn) { + t.Error("a node that just sent its whole list was asked for it again") + } +} + +// The lookup index can drift from the disks without the volume server seeing +// anything, so its digest still matches. Only a full report re-registers the +// volumes that stopped being servable, and in delta mode nothing else asks for +// one. +func TestDigestCheckAsksForTheListWhenTheLookupIndexDrifts(t *testing.T) { + ms, dn := digestTestCluster(t) + volumes := []*master_pb.VolumeInformationMessage{digestTestVolumeMessage(1), digestTestVolumeMessage(2)} + ms.Topo.SyncDataNodeRegistration(volumes, dn) + + digest := dn.VolumeDigest() + if ms.checkVolumeDigest(&master_pb.Heartbeat{VolumeDigest: &digest}, dn) { + t.Fatal("a healthy node was asked to resend") + } + + rp, _ := super_block.NewReplicaPlacementFromString("000") + vl := ms.Topo.GetVolumeLayout("c", rp, needle.EMPTY_TTL, types.HardDriveType) + vl.SetVolumeUnavailable(dn, needle.VolumeId(1)) + + if dn.VolumeDigest() != digest { + t.Fatal("expected the reported digest to be unaffected, which is why the index has to be checked") + } + if !ms.checkVolumeDigest(&master_pb.Heartbeat{VolumeDigest: &digest}, dn) { + t.Error("a volume that stopped being servable left the node sending only changes, so nothing would repair it") + } +} + +// A volume server takes the options from every response it receives, and +// preallocate is a bare bool with no way to tell "off" from "not mentioned". A +// response that left it out would turn preallocation off until reconnect. +func TestHeartbeatResponsesCarryTheVolumeOptions(t *testing.T) { + ms := &MasterServer{option: &MasterOption{VolumeSizeLimitMB: 1024}, preallocateSize: 1} + + resend := ms.heartbeatResponse() + resend.ResendFullVolumeList = true + + if !resend.Preallocate { + t.Error("a resend request would turn off preallocation on the volume server") + } + if resend.VolumeSizeLimit != 1024*1024*1024 { + t.Errorf("a resend request carried volume size limit %d, want the configured one", resend.VolumeSizeLimit) } } diff --git a/weed/server/volume_grpc_client_to_master.go b/weed/server/volume_grpc_client_to_master.go index e55a6ca21..13700fb80 100644 --- a/weed/server/volume_grpc_client_to_master.go +++ b/weed/server/volume_grpc_client_to_master.go @@ -234,6 +234,13 @@ func (vs *VolumeServer) doHeartbeatWithRetry(masterAddress pb.ServerAddress, grp } } } + if in.GetVolumeDigestSupported() { + vs.store.AcceptVolumeChanges() + } + if in.GetResendFullVolumeList() { + glog.V(0).Infof("master %s asked for the full volume list", masterAddress) + vs.store.RequestFullVolumeList() + } if in.GetLeader() != "" { current := vs.getCurrentMaster() if !current.Equals(pb.ServerAddress(in.GetLeader())) { @@ -246,6 +253,10 @@ func (vs *VolumeServer) doHeartbeatWithRetry(masterAddress pb.ServerAddress, grp } }() + // This master may know nothing about this server, and has not yet said + // whether it understands digests, so start from the whole list. + vs.store.ResetVolumeReporting() + if err = stream.Send(vs.store.CollectHeartbeat()); err != nil { glog.V(0).Infof("Volume Server Failed to talk with master %s: %v", masterAddress, err) return "", err diff --git a/weed/storage/store.go b/weed/storage/store.go index ab6cc6b99..aa66ce46b 100644 --- a/weed/storage/store.go +++ b/weed/storage/store.go @@ -81,6 +81,7 @@ type Store struct { NewEcShardsChan chan *master_pb.VolumeEcShardInformationMessage DeletedEcShardsChan chan *master_pb.VolumeEcShardInformationMessage isStopping bool + volumeReport volumeReportState } func (s *Store) String() (str string) { @@ -414,10 +415,12 @@ func (s *Store) GetRack() string { func (s *Store) CollectHeartbeat() *master_pb.Heartbeat { var volumeMessages []*master_pb.VolumeInformationMessage - // Digest of exactly what this heartbeat reports, so the master can tell - // whether its copy is current. Volumes skipped above -- quarantined, - // phantom, expired -- are absent from both the list and the digest. + // Covers every volume held, whether or not this heartbeat names it, so the + // master can tell whether applying what it was sent leaves it current. + // Volumes skipped below -- quarantined, phantom, expired -- are in neither. var volumeDigest uint64 + sendFullList, reportGeneration := s.volumeReport.begin() + reportedHashes := make(map[volumeReportKey]uint64) maxVolumeCounts := make(map[string]uint32) // Per-disk effective max for DiskTag, captured alongside the per-type sum. diskMaxByID := make(map[int]int32) @@ -490,8 +493,12 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat { shouldDeleteVolume := false if !v.expired(volumeMessage.Size, s.GetVolumeSizeLimit()) { - volumeMessages = append(volumeMessages, volumeMessage) - volumeDigest ^= reportHashOf(volumeMessage) + reportHash := reportHashOf(volumeMessage) + volumeDigest ^= reportHash + reportedHashes[volumeReportKey{diskId: volumeMessage.DiskId, volumeId: volumeMessage.Id}] = reportHash + if sendFullList || s.volumeReport.changed(volumeMessage, reportHash) { + volumeMessages = append(volumeMessages, volumeMessage) + } } else { if v.expiredLongEnough(MAX_TTL_VOLUME_REMOVAL_DELAY) { deleteVids = append(deleteVids, v.Id) @@ -586,6 +593,18 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat { } } + s.volumeReport.commit(reportedHashes, reportGeneration) + + // has_no_volumes says the server holds nothing, so it may only be derived + // from a full list. Deriving it from a changed-only heartbeat would make a + // quiet one read as an empty server and drop every volume on it. + heartbeatVolumes, changedVolumes := volumeMessages, []*master_pb.VolumeInformationMessage(nil) + hasNoVolumes := len(volumeMessages) == 0 + if !sendFullList { + heartbeatVolumes, changedVolumes = nil, volumeMessages + hasNoVolumes = false + } + return &master_pb.Heartbeat{ Ip: s.Ip, Port: uint32(s.Port), @@ -598,10 +617,11 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat { MaxFileKey: NeedleIdToUint64(maxFileKey), DataCenter: s.dataCenter, Rack: s.rack, - Volumes: volumeMessages, + Volumes: heartbeatVolumes, + ChangedVolumes: changedVolumes, VolumeDigest: &volumeDigest, DeletedEcShards: deletedEcVolumes, - HasNoVolumes: len(volumeMessages) == 0, + HasNoVolumes: hasNoVolumes, HasNoEcShards: len(ecVolumeMessages) == 0, LocationUuids: uuidList, DiskTags: diskTags, @@ -621,6 +641,24 @@ func reportHashOf(m *master_pb.VolumeInformationMessage) uint64 { return vi.ReportHash() } +// ResetVolumeReporting forgets what the master was told, so the next heartbeat +// carries the whole list. Called when a connection is established, since a +// reconnect may reach a master that knows nothing about this server. +func (s *Store) ResetVolumeReporting() { + s.volumeReport.reset() +} + +// AcceptVolumeChanges records that the master compares digests, so heartbeats +// may carry only what changed. +func (s *Store) AcceptVolumeChanges() { + s.volumeReport.acceptDeltas() +} + +// RequestFullVolumeList makes the next heartbeat carry the whole list. +func (s *Store) RequestFullVolumeList() { + s.volumeReport.requestFullList() +} + func (s *Store) deleteExpiredEcVolumes() (ecShards, deleted []*master_pb.VolumeEcShardInformationMessage) { for diskId, location := range s.Locations { if location.isDiskUnavailable.Load() { diff --git a/weed/storage/store_volume_report.go b/weed/storage/store_volume_report.go new file mode 100644 index 000000000..11e4dd326 --- /dev/null +++ b/weed/storage/store_volume_report.go @@ -0,0 +1,87 @@ +package storage + +import ( + "sync" + + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" +) + +// volumeReportKey identifies one reported copy. Keyed by disk as well as id +// because a volume id can be mounted on two disks, and reporting one of them +// would leave the other's changes untold. +type volumeReportKey struct { + diskId uint32 + volumeId uint32 +} + +// volumeReportState remembers what the master was last told about each volume, +// so a heartbeat can carry only what moved since. +// +// It is per-connection: a server that reconnects, or reaches a different +// master, knows nothing about what that master holds and starts again from the +// full list. The zero value has told no master anything, so it sends the whole +// list until one accepts changes. +type volumeReportState struct { + mu sync.Mutex + // deltasAccepted is set once the master says it compares digests. Until + // then the whole list goes every time, which is what an older master needs. + deltasAccepted bool + fullListNeeded bool + // fullListGeneration counts requests for the whole list, so one arriving + // while a heartbeat is being built is not marked satisfied by it. + fullListGeneration uint64 + lastReported map[volumeReportKey]uint64 +} + +// reset drops everything known about the master's view. +func (s *volumeReportState) reset() { + s.mu.Lock() + defer s.mu.Unlock() + s.deltasAccepted = false + s.fullListNeeded = true + s.fullListGeneration++ + s.lastReported = nil +} + +func (s *volumeReportState) acceptDeltas() { + s.mu.Lock() + defer s.mu.Unlock() + s.deltasAccepted = true +} + +func (s *volumeReportState) requestFullList() { + s.mu.Lock() + defer s.mu.Unlock() + s.fullListNeeded = true + s.fullListGeneration++ +} + +// begin reports whether this heartbeat must carry the whole list, and the +// request it answers. +func (s *volumeReportState) begin() (full bool, generation uint64) { + s.mu.Lock() + defer s.mu.Unlock() + return s.fullListNeeded || !s.deltasAccepted, s.fullListGeneration +} + +// changed reports whether the master needs telling about this volume, given +// what it was last told. +func (s *volumeReportState) changed(m *master_pb.VolumeInformationMessage, hash uint64) bool { + s.mu.Lock() + defer s.mu.Unlock() + previous, known := s.lastReported[volumeReportKey{diskId: m.DiskId, volumeId: m.Id}] + return !known || previous != hash +} + +// commit records what this heartbeat told the master. Volumes absent from +// reported are forgotten, so one that comes back is reported again. +func (s *volumeReportState) commit(reported map[volumeReportKey]uint64, generation uint64) { + s.mu.Lock() + defer s.mu.Unlock() + s.lastReported = reported + // A request that arrived while this heartbeat was being built asked about a + // later state than it carries, so it stands. + if s.fullListGeneration == generation { + s.fullListNeeded = false + } +} diff --git a/weed/storage/store_volume_report_test.go b/weed/storage/store_volume_report_test.go new file mode 100644 index 000000000..dbbee7a43 --- /dev/null +++ b/weed/storage/store_volume_report_test.go @@ -0,0 +1,155 @@ +package storage + +import ( + "testing" + + "github.com/seaweedfs/seaweedfs/weed/storage/needle" +) + +func reportingStore(t *testing.T, vids ...needle.VolumeId) *Store { + t.Helper() + store := newTestStore(t, 1) + for _, vid := range vids { + mountTestVolume(t, store.Locations[0], vid) + } + return store +} + +// Until the master says it compares digests it may be one that reads a partial +// list as the whole truth, so it keeps getting the whole list. +func TestHeartbeatSendsFullListUntilTheMasterAccepts(t *testing.T) { + store := reportingStore(t, 1, 2) + store.ResetVolumeReporting() + + for i := 0; i < 3; i++ { + heartbeat := store.CollectHeartbeat() + if len(heartbeat.Volumes) != 2 { + t.Fatalf("heartbeat %d carried %d volumes, want the full list of 2", i, len(heartbeat.Volumes)) + } + if len(heartbeat.ChangedVolumes) != 0 { + t.Fatalf("heartbeat %d sent changes to a master that never accepted them", i) + } + } +} + +// The one that would be catastrophic: a heartbeat with nothing to report must +// not look like a server that has lost every volume. +func TestQuietHeartbeatDoesNotLookLikeAnEmptyServer(t *testing.T) { + store := reportingStore(t, 1, 2) + store.ResetVolumeReporting() + store.AcceptVolumeChanges() + store.CollectHeartbeat() + + heartbeat := store.CollectHeartbeat() + if len(heartbeat.Volumes) != 0 || len(heartbeat.ChangedVolumes) != 0 { + t.Fatalf("expected nothing to report, got %d volumes and %d changes", + len(heartbeat.Volumes), len(heartbeat.ChangedVolumes)) + } + if heartbeat.HasNoVolumes { + t.Error("a heartbeat with nothing to report claimed the server holds no volumes") + } +} + +// The digest covers everything held, not just what was sent, or a master that +// applied the changes could never confirm it is current. +func TestQuietHeartbeatStillDigestsEverythingHeld(t *testing.T) { + store := reportingStore(t, 1, 2) + store.ResetVolumeReporting() + full := store.CollectHeartbeat() + store.AcceptVolumeChanges() + + quiet := store.CollectHeartbeat() + if quiet.GetVolumeDigest() != full.GetVolumeDigest() { + t.Errorf("digest changed with nothing to report: %d then %d", + full.GetVolumeDigest(), quiet.GetVolumeDigest()) + } +} + +func TestHeartbeatReportsOnlyWhatChanged(t *testing.T) { + store := reportingStore(t, 1, 2) + store.ResetVolumeReporting() + store.AcceptVolumeChanges() + store.CollectHeartbeat() + + mountTestVolume(t, store.Locations[0], 3) + heartbeat := store.CollectHeartbeat() + + if len(heartbeat.Volumes) != 0 { + t.Errorf("a changed-only heartbeat carried a full list of %d", len(heartbeat.Volumes)) + } + if len(heartbeat.ChangedVolumes) != 1 || heartbeat.ChangedVolumes[0].Id != 3 { + t.Errorf("expected only the new volume, got %v", heartbeat.ChangedVolumes) + } +} + +func TestHeartbeatReturnsToTheFullListOnRequest(t *testing.T) { + store := reportingStore(t, 1, 2) + store.ResetVolumeReporting() + store.AcceptVolumeChanges() + store.CollectHeartbeat() + + store.RequestFullVolumeList() + heartbeat := store.CollectHeartbeat() + if len(heartbeat.Volumes) != 2 { + t.Errorf("after a resend request the heartbeat carried %d volumes, want 2", len(heartbeat.Volumes)) + } + if len(heartbeat.ChangedVolumes) != 0 { + t.Error("a resend should carry the list, not changes") + } + + if next := store.CollectHeartbeat(); len(next.Volumes) != 0 { + t.Errorf("the resend repeated instead of returning to changes: %d volumes", len(next.Volumes)) + } +} + +// A reconnect may reach a master that knows nothing about this server. +func TestReconnectingSendsTheFullListAgain(t *testing.T) { + store := reportingStore(t, 1, 2) + store.ResetVolumeReporting() + store.AcceptVolumeChanges() + store.CollectHeartbeat() + + store.ResetVolumeReporting() + heartbeat := store.CollectHeartbeat() + if len(heartbeat.Volumes) != 2 { + t.Errorf("after reconnecting the heartbeat carried %d volumes, want the full list of 2", len(heartbeat.Volumes)) + } +} + +// A volume that goes and comes back has to be reported again, so forgetting it +// while it is away is what makes that work. +func TestRemountedVolumeIsReportedAgain(t *testing.T) { + store := reportingStore(t, 1) + store.ResetVolumeReporting() + store.AcceptVolumeChanges() + store.CollectHeartbeat() + + store.Locations[0].UnloadVolume(needle.VolumeId(1)) + store.CollectHeartbeat() + + mountTestVolume(t, store.Locations[0], 1) + heartbeat := store.CollectHeartbeat() + if len(heartbeat.ChangedVolumes) != 1 { + t.Errorf("a remounted volume was not reported: %v", heartbeat.ChangedVolumes) + } +} + +// A request that lands while a heartbeat is being built asked about a later +// state than that heartbeat carries, so it must survive being committed over. +func TestFullListRequestDuringCollectionSurvives(t *testing.T) { + store := reportingStore(t, 1, 2) + store.ResetVolumeReporting() + store.AcceptVolumeChanges() + store.CollectHeartbeat() + + full, generation := store.volumeReport.begin() + if full { + t.Fatal("expected to be past the first full list") + } + store.RequestFullVolumeList() + store.volumeReport.commit(map[volumeReportKey]uint64{}, generation) + + if heartbeat := store.CollectHeartbeat(); len(heartbeat.Volumes) != 2 { + t.Errorf("a resend request made during collection was lost: %d volumes sent", len(heartbeat.Volumes)) + } +} diff --git a/weed/topology/topology.go b/weed/topology/topology.go index 1374ed4eb..ce74977d8 100644 --- a/weed/topology/topology.go +++ b/weed/topology/topology.go @@ -630,6 +630,11 @@ func (t *Topology) SyncDataNodeRegistration(volumes []*master_pb.VolumeInformati // RegisterVolumeLayout, which would repeat the GetVolumeLayout lookup. vl.RegisterVolume(&v, dn) vl.EnsureCorrectWritables(&v) + // Volumes new to the disk map were registered above, so reaching + // here means only the lookup index had lost it. Clients were told + // it went when the node dropped out, so the repair has to tell them + // it is back. + newVolumes = append(newVolumes, v) } if vl.UpdateVolumeSize(v.Id, v.Size, v.CompactRevision) { vl.AdjustActiveVolumeCountAfterRecovery(v.Id) @@ -668,6 +673,52 @@ func (t *Topology) IncrementalSyncDataNodeRegistration(newVolumes, deletedVolume return } +// ApplyVolumeChanges records the volumes a heartbeat reported as changed and +// returns the ones the node did not already have. Only the named volumes are +// touched: unlike a full report, silence about a volume says nothing about +// whether the server still has it. +// +// Most changes are a volume growing, which moves no location, so returning +// only the arrivals keeps a busy cluster from telling every client about +// volumes they can already reach. +func (t *Topology) ApplyVolumeChanges(changed []*master_pb.VolumeInformationMessage, dn *DataNode) (newVolumes []storage.VolumeInfo) { + volumeInfos := make([]storage.VolumeInfo, 0, len(changed)) + for _, v := range changed { + vi, err := storage.NewVolumeInfo(v) + if err != nil { + glog.V(0).Infof("Fail to convert changed volume information: %v", err) + continue + } + volumeInfos = append(volumeInfos, vi) + } + + for _, vi := range volumeInfos { + isNew, _ := dn.AddOrUpdateVolume(vi) + if vi.ReplicaPlacement == nil { + if isNew { + newVolumes = append(newVolumes, vi) + } + continue + } + vl := t.GetVolumeLayout(vi.Collection, vi.ReplicaPlacement, vi.Ttl, types.ToDiskType(vi.DiskType)) + // Reaching the lookup index is what makes a volume servable, so a + // volume only that index had lost is an arrival as far as clients are + // concerned: they were told it went when the node dropped out. + becameServable := !vl.HasDataNode(vi.Id, dn) + if becameServable { + vl.RegisterVolume(&vi, dn) + } + if isNew || becameServable { + newVolumes = append(newVolumes, vi) + } + vl.EnsureCorrectWritables(&vi) + if vl.UpdateVolumeSize(vi.Id, vi.Size, vi.CompactRevision) { + vl.AdjustActiveVolumeCountAfterRecovery(vi.Id) + } + } + return newVolumes +} + func (t *Topology) DataNodeRegistration(dcName, rackName string, dn *DataNode) { if dn.Parent() != nil { return