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) + } +}