From 6d08b08f371a9a4d5d094b92bb48bc3e9b5544e4 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 7 Aug 2026 14:46:34 -0700 Subject: [PATCH] heartbeat: carry a volume digest and verify it (#10627) * pb: carry a volume digest on the heartbeat The full volume list is the only way a master notices a volume that vanished without a delta, so it cannot simply be dropped. A digest gives the same guarantee without the list, and a way back to the list when they disagree. The digest has explicit presence: a server holding no volumes reports 0, which has to stay distinguishable from a server that does not compute one at all. * volume: report a digest of the volumes each heartbeat carries Digests exactly what goes on the wire: volumes skipped as quarantined, phantom or expired are absent from both the list and the digest, so the master compares against the same set the server meant to report. Runs the master's own hash over the master's own conversion of the message, so the two ends cannot drift into disagreeing about a field. * master: check the reported volume digest and ask for the list on a mismatch Compared after everything the heartbeat carried has been applied, so agreement means the master is current rather than that nothing changed. Servers reporting no digest are untouched, and a mismatch on a heartbeat that already carried the full list is reported rather than answered: there is nothing further to ask for, so asking again would loop. Nodes reporting one volume id twice are skipped for the same reason. * rust: report the heartbeat volume digest Mirrors the Go volume server. The master compares this against a digest it computes itself, so the hash has to agree byte for byte across the two implementations, not merely be a hash of the same fields: report_hash_vectors pins it against values generated by the Go side, and the ttl and replica placement narrowing the master applies when it decodes a message is applied here too rather than assumed away. A drift there would not corrupt anything, but every volume server on this implementation would report a digest the master can never match and fall back to sending its whole volume list forever, which is the cost the digest exists to avoid. * master: pin what the digest check does to each kind of report The upgrade story rests on these: a server that reports no digest is never asked for anything, so the two sides can be upgraded in either order, and a disagreement that resending cannot fix is reported rather than re-asked, so it cannot loop. * topology: enumerate the digest coverage test from the message The list of fields was written out by hand, so a field added to VolumeInformationMessage later would fall outside the digest while the test went on passing, and a change to it would never reach the master. Walk the message descriptor instead. Some fields are narrowed or normalised on the way into VolumeInfo, so the smallest change to the wire value can land back on the stored one; the test offers several values per field and asks only that some change is visible. --- seaweed-volume/Cargo.lock | 7 + seaweed-volume/Cargo.toml | 3 + seaweed-volume/proto/master.proto | 14 ++ seaweed-volume/src/server/heartbeat.rs | 62 ++++++++- seaweed-volume/src/storage/mod.rs | 1 + .../src/storage/volume_report_hash.rs | 104 +++++++++++++++ weed/pb/master.proto | 9 ++ weed/pb/master_pb/master.pb.go | 43 +++++-- weed/server/master_grpc_server.go | 46 +++++++ weed/server/master_grpc_server_digest_test.go | 90 +++++++++++++ weed/storage/store.go | 18 +++ weed/storage/store_heartbeat_digest_test.go | 81 ++++++++++++ weed/topology/volume_digest_test.go | 120 +++++++++++++----- 13 files changed, 557 insertions(+), 41 deletions(-) create mode 100644 seaweed-volume/src/storage/volume_report_hash.rs create mode 100644 weed/server/master_grpc_server_digest_test.go create mode 100644 weed/storage/store_heartbeat_digest_test.go diff --git a/seaweed-volume/Cargo.lock b/seaweed-volume/Cargo.lock index 150b88856..a4c875258 100644 --- a/seaweed-volume/Cargo.lock +++ b/seaweed-volume/Cargo.lock @@ -4556,6 +4556,7 @@ dependencies = [ "tracing-subscriber", "uuid", "x509-parser", + "xxhash-rust", ] [[package]] @@ -4989,6 +4990,12 @@ version = "0.13.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "66fee0b777b0f5ac1c69bb06d361268faafa61cd4682ae064a171c16c433e9e4" +[[package]] +name = "xxhash-rust" +version = "0.8.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "aee1b19627c7c60102ab80d3a9cbe18de90bfe03bfa6c3715447681f0e8c8af6" + [[package]] name = "yoke" version = "0.8.2" diff --git a/seaweed-volume/Cargo.toml b/seaweed-volume/Cargo.toml index 41bb8fa31..481f7bc12 100644 --- a/seaweed-volume/Cargo.toml +++ b/seaweed-volume/Cargo.toml @@ -75,6 +75,9 @@ serde_urlencoded = "0.7" crc32c = "0.6" crc32fast = "1" +# xxhash64 - must match Go's cespare/xxhash for the heartbeat volume digest +xxhash-rust = { version = "0.8", features = ["xxh64"] } + # Memory-mapped files memmap2 = "0.9" diff --git a/seaweed-volume/proto/master.proto b/seaweed-volume/proto/master.proto index 0fbfe198a..afffc7d06 100644 --- a/seaweed-volume/proto/master.proto +++ b/seaweed-volume/proto/master.proto @@ -105,6 +105,12 @@ message Heartbeat { // physical disk capacity per disk type, in bytes, from the underlying filesystem map disk_total_bytes = 25; map disk_free_bytes = 26; + + // Digest of every volume in this heartbeat's view of the server, letting the + // 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. + optional uint64 volume_digest = 27; } message HeartbeatResponse { @@ -115,6 +121,9 @@ message HeartbeatResponse { repeated StorageBackend storage_backends = 5; repeated string duplicated_uuids = 6; bool preallocate = 7; + // 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; } message VolumeInformationMessage { @@ -294,6 +303,11 @@ message StatisticsResponse { uint64 total_size = 4; uint64 used_size = 5; uint64 file_count = 6; + // sizes counting one copy of the data: a single replica of a regular volume, + // the data shards of an ec volume. logical_total_size scales the free space + // by the copies the requested replication makes. + uint64 logical_total_size = 7; + uint64 logical_used_size = 8; } // diff --git a/seaweed-volume/src/server/heartbeat.rs b/seaweed-volume/src/server/heartbeat.rs index c0fcb3cdc..0a9181220 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_hash::report_hash; use crate::storage::types::NeedleId; const DUPLICATE_UUID_RETRY_MESSAGE: &str = "duplicate UUIDs detected, retrying connection"; @@ -800,6 +801,10 @@ 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. + let mut volume_digest: u64 = 0; let mut max_file_key = NeedleId(0); let mut max_volume_counts: HashMap = HashMap::new(); let mut disk_total_bytes: HashMap = HashMap::new(); @@ -875,7 +880,7 @@ fn build_heartbeat_with_ec_status( } let (remote_storage_name, remote_storage_key) = vol.remote_storage_name_key(); - volumes.push(master_pb::VolumeInformationMessage { + let volume_message = master_pb::VolumeInformationMessage { id: vol.id.0, size: volume_size, collection: vol.collection.clone(), @@ -892,8 +897,9 @@ fn build_heartbeat_with_ec_status( disk_id: disk_id as u32, remote_storage_name, remote_storage_key, - ..Default::default() - }); + }; + volume_digest ^= report_hash(&volume_message); + volumes.push(volume_message); } else if vol.is_expired_long_enough(MAX_TTL_VOLUME_REMOVAL_DELAY) { delete_vids.push(vol.id); should_delete_volume = true; @@ -967,6 +973,7 @@ fn build_heartbeat_with_ec_status( rack: config.rack.clone(), admin_port: config.port as u32, volumes, + volume_digest: Some(volume_digest), deleted_ec_shards, has_no_volumes, has_no_ec_shards, @@ -1246,6 +1253,55 @@ mod tests { assert!(heartbeat.has_no_volumes); } + // The digest must cover exactly the volumes the heartbeat carries. A volume + // 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. + #[test] + fn test_build_heartbeat_digests_exactly_what_it_reports() { + let temp_dir = tempfile::tempdir().unwrap(); + let dir = temp_dir.path().to_str().unwrap(); + + let mut store = Store::new(NeedleMapKind::InMemory); + store + .add_location( + dir, + dir, + 8, + DiskType::HardDrive, + MinFreeSpace::Percent(1.0), + Vec::new(), + ) + .unwrap(); + + let empty = build_heartbeat(&test_config(), &mut store); + assert_eq!(empty.volume_digest, Some(0)); + + for vid in [VolumeId(1), VolumeId(2)] { + store + .add_volume( + vid, + "pics", + None, + None, + 0, + DiskType::HardDrive, + Version::current(), + ) + .unwrap(); + } + + let heartbeat = build_heartbeat(&test_config(), &mut store); + assert_eq!(heartbeat.volumes.len(), 2); + + let expected = heartbeat + .volumes + .iter() + .fold(0u64, |acc, m| acc ^ crate::storage::volume_report_hash::report_hash(m)); + assert_eq!(heartbeat.volume_digest, Some(expected)); + assert_ne!(heartbeat.volume_digest, Some(0)); + } + #[test] fn test_build_heartbeat_tracks_go_read_only_labels_and_disk_id() { let temp_dir = tempfile::tempdir().unwrap(); diff --git a/seaweed-volume/src/storage/mod.rs b/seaweed-volume/src/storage/mod.rs index 57e195ba5..abe8ba285 100644 --- a/seaweed-volume/src/storage/mod.rs +++ b/seaweed-volume/src/storage/mod.rs @@ -10,3 +10,4 @@ pub mod super_block; pub mod types; pub mod volume; pub mod volume_idx_repair; +pub mod volume_report_hash; diff --git a/seaweed-volume/src/storage/volume_report_hash.rs b/seaweed-volume/src/storage/volume_report_hash.rs new file mode 100644 index 000000000..c661d4d7f --- /dev/null +++ b/seaweed-volume/src/storage/volume_report_hash.rs @@ -0,0 +1,104 @@ +//! Mirror of `weed/storage/volume_report_hash.go`. +//! +//! The master compares the digest a volume server reports against one it +//! computes itself, so this has to agree with the Go implementation +//! byte-for-byte. `report_hash_vectors` pins that against values produced by +//! the Go side; do not change the layout without regenerating them there. + +use xxhash_rust::xxh64::xxh64; + +use crate::pb::master_pb; + +/// Digests everything a volume server reports about a volume. +/// +/// It must cover every field of `VolumeInformationMessage`: a change the hash +/// misses is a change the master would never be told about. +pub fn report_hash(m: &master_pb::VolumeInformationMessage) -> u64 { + let mut buf = [0u8; 57]; + buf[0..4].copy_from_slice(&m.id.to_le_bytes()); + buf[4..12].copy_from_slice(&m.size.to_le_bytes()); + buf[12..20].copy_from_slice(&m.file_count.to_le_bytes()); + buf[20..28].copy_from_slice(&m.delete_count.to_le_bytes()); + buf[28..36].copy_from_slice(&m.deleted_byte_count.to_le_bytes()); + // The master stores these narrowed, so hash what it will hold, not what the + // wire type could carry. + buf[36..40].copy_from_slice(&((m.replica_placement as u8) as u32).to_le_bytes()); + buf[40..44].copy_from_slice(&((m.version as u8) as u32).to_le_bytes()); + buf[44..48].copy_from_slice(&normalize_ttl(m.ttl).to_le_bytes()); + buf[48..52].copy_from_slice(&m.compact_revision.to_le_bytes()); + buf[52..56].copy_from_slice(&m.disk_id.to_le_bytes()); + if m.read_only { + buf[56] = 1; + } + let mut h = xxh64(&buf, 0); + + h = fold(h, xxh64(&(m.modified_at_second as u64).to_le_bytes(), 0)); + h = fold(h, xxh64(m.collection.as_bytes(), 0)); + h = fold(h, xxh64(m.disk_type.as_bytes(), 0)); + h = fold(h, xxh64(m.remote_storage_name.as_bytes(), 0)); + h = fold(h, xxh64(m.remote_storage_key.as_bytes(), 0)); + h +} + +/// A ttl whose count is zero encodes as zero however the unit is set, matching +/// what the master stores after decoding it. +fn normalize_ttl(ttl: u32) -> u32 { + let count = (ttl >> 8) & 0xff; + if count == 0 { + return 0; + } + (count << 8) | (ttl & 0xff) +} + +/// Combines two hashes order-dependently, so swapping two string fields is not +/// invisible. +fn fold(h: u64, x: u64) -> u64 { + let h = (h ^ x).wrapping_mul(0x9E37_79B9_7F4A_7C15); + h ^ (h >> 29) +} + +#[cfg(test)] +mod tests { + use super::*; + + // Produced by the Go implementation. If these drift, every volume server + // running this build reports a digest the master can never match, and falls + // back to sending its whole volume list forever. + #[test] + fn report_hash_vectors() { + let empty = master_pb::VolumeInformationMessage::default(); + assert_eq!(report_hash(&empty), 17122085700329870549); + + let mut one = master_pb::VolumeInformationMessage::default(); + one.id = 1; + assert_eq!(report_hash(&one), 12867601919960834066); + + let full = master_pb::VolumeInformationMessage { + id: 42, + size: 1 << 30, + collection: "c".to_string(), + file_count: 7, + delete_count: 2, + deleted_byte_count: 99, + read_only: true, + replica_placement: 10, + version: 3, + ttl: 3 << 8, + compact_revision: 5, + modified_at_second: 1700000000, + remote_storage_name: "s3".to_string(), + remote_storage_key: "k/1.dat".to_string(), + disk_type: "ssd".to_string(), + disk_id: 2, + }; + assert_eq!(report_hash(&full), 12500327696413250175); + } + + #[test] + fn ttl_with_no_count_is_dropped() { + assert_eq!(normalize_ttl(0), 0); + assert_eq!(normalize_ttl(3), 0); + assert_eq!(normalize_ttl(3 << 8), 3 << 8); + assert_eq!(normalize_ttl((3 << 8) | 4), (3 << 8) | 4); + } +} diff --git a/weed/pb/master.proto b/weed/pb/master.proto index e53d6676a..afffc7d06 100644 --- a/weed/pb/master.proto +++ b/weed/pb/master.proto @@ -105,6 +105,12 @@ message Heartbeat { // physical disk capacity per disk type, in bytes, from the underlying filesystem map disk_total_bytes = 25; map disk_free_bytes = 26; + + // Digest of every volume in this heartbeat's view of the server, letting the + // 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. + optional uint64 volume_digest = 27; } message HeartbeatResponse { @@ -115,6 +121,9 @@ message HeartbeatResponse { repeated StorageBackend storage_backends = 5; repeated string duplicated_uuids = 6; bool preallocate = 7; + // 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; } message VolumeInformationMessage { diff --git a/weed/pb/master_pb/master.pb.go b/weed/pb/master_pb/master.pb.go index 56bea92a4..03075dca8 100644 --- a/weed/pb/master_pb/master.pb.go +++ b/weed/pb/master_pb/master.pb.go @@ -121,8 +121,13 @@ type Heartbeat struct { // physical disk capacity per disk type, in bytes, from the underlying filesystem DiskTotalBytes map[string]uint64 `protobuf:"bytes,25,rep,name=disk_total_bytes,json=diskTotalBytes,proto3" json:"disk_total_bytes,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"varint,2,opt,name=value"` DiskFreeBytes map[string]uint64 `protobuf:"bytes,26,rep,name=disk_free_bytes,json=diskFreeBytes,proto3" json:"disk_free_bytes,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"varint,2,opt,name=value"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + // Digest of every volume in this heartbeat's view of the server, letting the + // 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 } func (x *Heartbeat) Reset() { @@ -316,6 +321,13 @@ func (x *Heartbeat) GetDiskFreeBytes() map[string]uint64 { return nil } +func (x *Heartbeat) GetVolumeDigest() uint64 { + if x != nil && x.VolumeDigest != nil { + return *x.VolumeDigest + } + return 0 +} + 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"` @@ -325,8 +337,11 @@ type HeartbeatResponse struct { StorageBackends []*StorageBackend `protobuf:"bytes,5,rep,name=storage_backends,json=storageBackends,proto3" json:"storage_backends,omitempty"` DuplicatedUuids []string `protobuf:"bytes,6,rep,name=duplicated_uuids,json=duplicatedUuids,proto3" json:"duplicated_uuids,omitempty"` Preallocate bool `protobuf:"varint,7,opt,name=preallocate,proto3" json:"preallocate,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + // 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 } func (x *HeartbeatResponse) Reset() { @@ -408,6 +423,13 @@ func (x *HeartbeatResponse) GetPreallocate() bool { return false } +func (x *HeartbeatResponse) GetResendFullVolumeList() bool { + if x != nil { + return x.ResendFullVolumeList + } + return false +} + type VolumeInformationMessage struct { state protoimpl.MessageState `protogen:"open.v1"` Id uint32 `protobuf:"varint,1,opt,name=id,proto3" json:"id,omitempty"` @@ -4569,8 +4591,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\"\xe6\n" + - "\n" + + "\x10max_volume_count\x18\x04 \x01(\x03R\x0emaxVolumeCount\"\xa2\v\n" + "\tHeartbeat\x12\x0e\n" + "\x02ip\x18\x01 \x01(\tR\x02ip\x12\x12\n" + "\x04port\x18\x02 \x01(\rR\x04port\x12\x1d\n" + @@ -4600,7 +4621,8 @@ const file_master_proto_rawDesc = "" + "\x05state\x18\x17 \x01(\v2#.volume_server_pb.VolumeServerStateR\x05state\x12/\n" + "\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\x1aB\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" + "\x14MaxVolumeCountsEntry\x12\x10\n" + "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" + "\x05value\x18\x02 \x01(\rR\x05value:\x028\x01\x1aA\n" + @@ -4609,7 +4631,8 @@ const file_master_proto_rawDesc = "" + "\x05value\x18\x02 \x01(\x04R\x05value:\x028\x01\x1a@\n" + "\x12DiskFreeBytesEntry\x12\x10\n" + "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" + - "\x05value\x18\x02 \x01(\x04R\x05value:\x028\x01\"\xcd\x02\n" + + "\x05value\x18\x02 \x01(\x04R\x05value:\x028\x01B\x10\n" + + "\x0e_volume_digest\"\x84\x03\n" + "\x11HeartbeatResponse\x12*\n" + "\x11volume_size_limit\x18\x01 \x01(\x04R\x0fvolumeSizeLimit\x12\x16\n" + "\x06leader\x18\x02 \x01(\tR\x06leader\x12'\n" + @@ -4617,7 +4640,8 @@ const file_master_proto_rawDesc = "" + "\x18metrics_interval_seconds\x18\x04 \x01(\rR\x16metricsIntervalSeconds\x12D\n" + "\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\"\xb1\x04\n" + + "\vpreallocate\x18\a \x01(\bR\vpreallocate\x125\n" + + "\x17resend_full_volume_list\x18\b \x01(\bR\x14resendFullVolumeList\"\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" + @@ -5203,6 +5227,7 @@ func file_master_proto_init() { if File_master_proto != nil { return } + file_master_proto_msgTypes[1].OneofWrappers = []any{} type x struct{} out := protoimpl.TypeBuilder{ File: protoimpl.DescBuilder{ diff --git a/weed/server/master_grpc_server.go b/weed/server/master_grpc_server.go index 728835345..5ee4d49fc 100644 --- a/weed/server/master_grpc_server.go +++ b/weed/server/master_grpc_server.go @@ -275,9 +275,55 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ if len(message.NewVids) > 0 || len(message.DeletedVids) > 0 || len(message.NewEcVids) > 0 || len(message.DeletedEcVids) > 0 { ms.broadcastToClients(&master_pb.KeepConnectedResponse{VolumeLocation: message}) } + + // 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 { + glog.Warningf("SendHeartbeat.Send resend request to %s:%d %v", dn.Ip, dn.Port, err) + return err + } + } } } +// checkVolumeDigest compares the digest a volume server reported against the +// master's own, and reports whether the master needs the full volume list to +// recover. Servers that report no digest are left alone: they still send the +// whole list every time. +func (ms *MasterServer) checkVolumeDigest(heartbeat *master_pb.Heartbeat, dn *topology.DataNode) bool { + 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 { + stats.MasterReceivedHeartbeatCounter.WithLabelValues("volumeDigestMatch").Inc() + 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. + 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) + return false + } + glog.V(0).Infof("volume server %s reported digest %d, master holds %d: requesting the full volume list", + dn.Url(), reported, held) + return true +} + // KeepConnected keep a stream gRPC call to the master. Used by clients to know the master is up. // And clients gets the up-to-date list of volume locations func (ms *MasterServer) KeepConnected(stream master_pb.Seaweed_KeepConnectedServer) error { diff --git a/weed/server/master_grpc_server_digest_test.go b/weed/server/master_grpc_server_digest_test.go new file mode 100644 index 000000000..25847358e --- /dev/null +++ b/weed/server/master_grpc_server_digest_test.go @@ -0,0 +1,90 @@ +package weed_server + +import ( + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "github.com/seaweedfs/seaweedfs/weed/sequence" + "github.com/seaweedfs/seaweedfs/weed/topology" +) + +func digestTestCluster(t *testing.T) (*MasterServer, *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 &MasterServer{Topo: topo}, dn +} + +func digestTestVolumeMessage(id uint32) *master_pb.VolumeInformationMessage { + return &master_pb.VolumeInformationMessage{ + Id: id, Size: 1024, Collection: "c", Version: 3, + } +} + +// A volume server that predates the digest keeps sending its whole list and +// must never be asked for anything, whatever the master computes. This is what +// lets the two sides be upgraded in either order. +func TestDigestCheckIgnoresServersThatReportNone(t *testing.T) { + ms, dn := digestTestCluster(t) + ms.Topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{digestTestVolumeMessage(1)}, dn) + + if ms.checkVolumeDigest(&master_pb.Heartbeat{ + Volumes: []*master_pb.VolumeInformationMessage{digestTestVolumeMessage(1)}, + }, dn) { + t.Error("a server reporting no digest was asked to resend") + } +} + +func TestDigestCheckAcceptsAMatchingReport(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{Volumes: volumes, VolumeDigest: &digest}, dn) { + t.Error("a matching digest was asked to resend") + } +} + +// A heartbeat that already carried the whole list has nothing further to give, +// so a mismatch there is a genuine disagreement to report rather than something +// to ask about again. +func TestDigestCheckDoesNotReaskAfterAFullList(t *testing.T) { + ms, dn := digestTestCluster(t) + volumes := []*master_pb.VolumeInformationMessage{digestTestVolumeMessage(1)} + ms.Topo.SyncDataNodeRegistration(volumes, dn) + + wrong := dn.VolumeDigest() ^ 1 + if ms.checkVolumeDigest(&master_pb.Heartbeat{Volumes: volumes, VolumeDigest: &wrong}, dn) { + t.Error("a full volume list that still disagreed was asked to resend, which would repeat forever") + } +} + +// The case the request exists for: a heartbeat carrying no list whose digest +// disagrees means the master has drifted and needs the list back. +func TestDigestCheckAsksForTheListWhenADeltaDisagrees(t *testing.T) { + ms, dn := digestTestCluster(t) + ms.Topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{digestTestVolumeMessage(1)}, dn) + + wrong := dn.VolumeDigest() ^ 1 + if !ms.checkVolumeDigest(&master_pb.Heartbeat{VolumeDigest: &wrong}, dn) { + t.Error("a disagreeing digest with no list to fall back on was not asked to resend") + } +} + +// 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) { + ms, dn := digestTestCluster(t) + duplicated := digestTestVolumeMessage(1) + duplicated.DiskId = 1 + ms.Topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{ + 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") + } +} diff --git a/weed/storage/store.go b/weed/storage/store.go index f898d535c..ab6cc6b99 100644 --- a/weed/storage/store.go +++ b/weed/storage/store.go @@ -414,6 +414,10 @@ 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. + var volumeDigest uint64 maxVolumeCounts := make(map[string]uint32) // Per-disk effective max for DiskTag, captured alongside the per-type sum. diskMaxByID := make(map[int]int32) @@ -487,6 +491,7 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat { shouldDeleteVolume := false if !v.expired(volumeMessage.Size, s.GetVolumeSizeLimit()) { volumeMessages = append(volumeMessages, volumeMessage) + volumeDigest ^= reportHashOf(volumeMessage) } else { if v.expiredLongEnough(MAX_TTL_VOLUME_REMOVAL_DELAY) { deleteVids = append(deleteVids, v.Id) @@ -594,6 +599,7 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat { DataCenter: s.dataCenter, Rack: s.rack, Volumes: volumeMessages, + VolumeDigest: &volumeDigest, DeletedEcShards: deletedEcVolumes, HasNoVolumes: len(volumeMessages) == 0, HasNoEcShards: len(ecVolumeMessages) == 0, @@ -603,6 +609,18 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat { } +// reportHashOf digests a volume exactly as the master will digest what it +// stores for that volume, by running the master's own hash over the same +// conversion the master applies to the message. +func reportHashOf(m *master_pb.VolumeInformationMessage) uint64 { + vi, err := NewVolumeInfo(m) + if err != nil { + glog.Warningf("volume %d: cannot digest heartbeat report: %v", m.Id, err) + return 0 + } + return vi.ReportHash() +} + 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_heartbeat_digest_test.go b/weed/storage/store_heartbeat_digest_test.go new file mode 100644 index 000000000..c58e9469e --- /dev/null +++ b/weed/storage/store_heartbeat_digest_test.go @@ -0,0 +1,81 @@ +package storage + +import ( + "testing" + + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/super_block" +) + +func mountTestVolume(t *testing.T, loc *DiskLocation, vid needle.VolumeId) { + t.Helper() + v, err := NewVolume(loc.Directory, loc.IdxDirectory, "", vid, NeedleMapInMemory, + &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0) + if err != nil { + t.Fatal(err) + } + loc.SetVolume(vid, v) +} + +// The digest has to cover exactly the volumes the heartbeat carries. A volume +// reported but left out of the digest, or the reverse, makes the master's +// comparison disagree forever. +func TestCollectHeartbeatDigestsExactlyWhatItReports(t *testing.T) { + store := newTestStore(t, 2) + mountTestVolume(t, store.Locations[0], 1) + mountTestVolume(t, store.Locations[0], 2) + mountTestVolume(t, store.Locations[1], 3) + + heartbeat := store.CollectHeartbeat() + if heartbeat.VolumeDigest == nil { + t.Fatal("heartbeat carried no digest") + } + if len(heartbeat.Volumes) != 3 { + t.Fatalf("expected 3 volumes reported, got %d", len(heartbeat.Volumes)) + } + + var want uint64 + for _, m := range heartbeat.Volumes { + vi, err := NewVolumeInfo(m) + if err != nil { + t.Fatal(err) + } + want ^= vi.ReportHash() + } + if got := heartbeat.GetVolumeDigest(); got != want { + t.Errorf("digest %d does not cover the reported volumes (%d)", got, want) + } +} + +// A server holding nothing reports a digest of 0, which is why the field needs +// explicit presence: it must stay distinguishable from a server that computes +// no digest at all. +func TestCollectHeartbeatDigestsAnEmptyStore(t *testing.T) { + store := newTestStore(t, 1) + + heartbeat := store.CollectHeartbeat() + if heartbeat.VolumeDigest == nil { + t.Fatal("an empty store still has to report a digest, or the master cannot tell it from an old server") + } + if got := heartbeat.GetVolumeDigest(); got != 0 { + t.Errorf("expected an empty store to digest to 0, got %d", got) + } + if !heartbeat.HasNoVolumes { + t.Error("expected has_no_volumes on an empty store") + } +} + +func TestCollectHeartbeatDigestFollowsVolumeChanges(t *testing.T) { + store := newTestStore(t, 1) + mountTestVolume(t, store.Locations[0], 1) + first := store.CollectHeartbeat().GetVolumeDigest() + + if second := store.CollectHeartbeat().GetVolumeDigest(); second != first { + t.Errorf("an unchanged store reported a different digest: %d then %d", first, second) + } + + mountTestVolume(t, store.Locations[0], 2) + if grown := store.CollectHeartbeat().GetVolumeDigest(); grown == first { + t.Error("mounting a volume left the digest unchanged") + } +} diff --git a/weed/topology/volume_digest_test.go b/weed/topology/volume_digest_test.go index 9d2ab2585..70ded92c8 100644 --- a/weed/topology/volume_digest_test.go +++ b/weed/topology/volume_digest_test.go @@ -3,11 +3,16 @@ package topology import ( "testing" + "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/reflect/protoreflect" + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "github.com/seaweedfs/seaweedfs/weed/stats" "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/util" ) func digestTestNode(t *testing.T) (*Topology, *DataNode) { @@ -66,46 +71,63 @@ func TestVolumeDigestIsIndependentOfReportOrder(t *testing.T) { } } +// Every field of VolumeInformationMessage has to reach the digest: one the hash +// skips is a change the master would never be told about. Enumerated from the +// message rather than listed here, so a field added later cannot quietly fall +// outside the digest while this still passes. func TestVolumeDigestTracksEveryReportedField(t *testing.T) { base := digestTestVolume(1) - mutations := map[string]func(*master_pb.VolumeInformationMessage){ - "Id": func(m *master_pb.VolumeInformationMessage) { m.Id = 2 }, - "Size": func(m *master_pb.VolumeInformationMessage) { m.Size++ }, - "Collection": func(m *master_pb.VolumeInformationMessage) { m.Collection = "other" }, - "FileCount": func(m *master_pb.VolumeInformationMessage) { m.FileCount++ }, - "DeleteCount": func(m *master_pb.VolumeInformationMessage) { m.DeleteCount++ }, - "DeletedByteCount": func(m *master_pb.VolumeInformationMessage) { m.DeletedByteCount++ }, - "ReadOnly": func(m *master_pb.VolumeInformationMessage) { m.ReadOnly = true }, - "ReplicaPlacement": func(m *master_pb.VolumeInformationMessage) { m.ReplicaPlacement = 10 }, - "Version": func(m *master_pb.VolumeInformationMessage) { m.Version = 2 }, - "Ttl": func(m *master_pb.VolumeInformationMessage) { m.Ttl = 3 << 8 }, - "CompactRevision": func(m *master_pb.VolumeInformationMessage) { m.CompactRevision++ }, - "ModifiedAtSecond": func(m *master_pb.VolumeInformationMessage) { m.ModifiedAtSecond++ }, - "RemoteStorageName": func(m *master_pb.VolumeInformationMessage) { m.RemoteStorageName = "s3" }, - "RemoteStorageKey": func(m *master_pb.VolumeInformationMessage) { m.RemoteStorageKey = "k" }, - "DiskType": func(m *master_pb.VolumeInformationMessage) { m.DiskType = "ssd" }, - "DiskId": func(m *master_pb.VolumeInformationMessage) { m.DiskId = 1 }, - } - baseInfo, err := storage.NewVolumeInfo(base) if err != nil { t.Fatal(err) } - for name, mutate := range mutations { - t.Run(name, func(t *testing.T) { - changed := digestTestVolume(1) - mutate(changed) - changedInfo, err := storage.NewVolumeInfo(changed) - if err != nil { - t.Fatal(err) - } - if baseInfo.ReportHash() == changedInfo.ReportHash() { - t.Errorf("a change to %s is invisible to the digest, so the master would never be told about it", name) + + fields := base.ProtoReflect().Descriptor().Fields() + for i := 0; i < fields.Len(); i++ { + fd := fields.Get(i) + t.Run(string(fd.Name()), func(t *testing.T) { + candidates := distinctValuesFor(t, fd, base.ProtoReflect().Get(fd)) + for _, candidate := range candidates { + changed := proto.Clone(base).(*master_pb.VolumeInformationMessage) + changed.ProtoReflect().Set(fd, candidate) + changedInfo, err := storage.NewVolumeInfo(changed) + if err != nil { + t.Fatal(err) + } + if baseInfo.ReportHash() != changedInfo.ReportHash() { + return + } } + t.Errorf("no change to %s moves the digest, so the master would never be told about one", fd.Name()) }) } } +// distinctValuesFor offers values that differ from current. Several, because +// some fields are narrowed or normalised on the way into VolumeInfo and the +// smallest change to the wire value can land back on the stored one. +func distinctValuesFor(t *testing.T, fd protoreflect.FieldDescriptor, current protoreflect.Value) []protoreflect.Value { + t.Helper() + switch fd.Kind() { + case protoreflect.BoolKind: + return []protoreflect.Value{protoreflect.ValueOfBool(!current.Bool())} + case protoreflect.Uint32Kind: + return []protoreflect.Value{ + protoreflect.ValueOfUint32(uint32(current.Uint()) + 1), + protoreflect.ValueOfUint32(uint32(current.Uint()) + 1<<8), + protoreflect.ValueOfUint32(uint32(current.Uint()) + 1<<16), + } + case protoreflect.Uint64Kind: + return []protoreflect.Value{protoreflect.ValueOfUint64(current.Uint() + 1)} + case protoreflect.Int64Kind: + return []protoreflect.Value{protoreflect.ValueOfInt64(current.Int() + 1)} + case protoreflect.StringKind: + return []protoreflect.Value{protoreflect.ValueOfString(current.String() + "x")} + } + t.Fatalf("field %s has kind %s, which this test does not know how to vary", fd.Name(), fd.Kind()) + return nil +} + func TestVolumeDigestFollowsVolumeChanges(t *testing.T) { topo, dn := digestTestNode(t) volumes := []*master_pb.VolumeInformationMessage{digestTestVolume(1), digestTestVolume(2)} @@ -408,3 +430,43 @@ func TestVolumeIndexDigestFollowsRemovedLookupEntry(t *testing.T) { t.Error("the node that was merely passed in should never have gained the entry") } } + +// The two ends must agree on real heartbeat data, not just on hand-built +// messages: the volume server hashes what it is about to send, the master +// hashes what it stored from it. +func TestMasterDigestMatchesWhatAVolumeServerReports(t *testing.T) { + dir := t.TempDir() + loc := storage.NewDiskLocation(dir, 100, util.MinFreeSpace{}, "", types.HardDriveType, nil, + stats.DefaultDiskIOProbeConfig()) + for _, vid := range []needle.VolumeId{1, 2, 3} { + v, err := storage.NewVolume(loc.Directory, loc.IdxDirectory, "", vid, storage.NeedleMapInMemory, + &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0) + if err != nil { + t.Fatal(err) + } + loc.SetVolume(vid, v) + } + + reported := make([]*master_pb.VolumeInformationMessage, 0, 3) + var serverDigest uint64 + for _, vid := range []needle.VolumeId{1, 2, 3} { + v, _ := loc.FindVolume(vid) + _, m := v.ToVolumeInformationMessage() + if m == nil { + t.Fatalf("volume %d reported nothing", vid) + } + vi, err := storage.NewVolumeInfo(m) + if err != nil { + t.Fatal(err) + } + serverDigest ^= vi.ReportHash() + reported = append(reported, m) + } + + topo, dn := digestTestNode(t) + topo.SyncDataNodeRegistration(reported, dn) + + if got := dn.VolumeDigest(); got != serverDigest { + t.Errorf("master digest %d does not match the reporting server's %d", got, serverDigest) + } +}