Github-Mirror/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-09-08 15:41:15 +02:00
Code Issues Packages Projects Releases Wiki Activity

trim the comments on this change to the parts that are not evident

Browse Source
This commit is contained in:
Chris Lu
2026-08-07 17:32:18 -07:00
parent 092f955d12
commit bf6b249a05
10 changed files with 47 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files
seaweed-volume/proto/master.proto
+4 -6
View File
@@ -106,10 +106,9 @@ message Heartbeat {
map<string, uint64> disk_total_bytes = 25;
map<string, uint64> 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.
// Lets the master check its copy is current without being sent the whole
// list. Optional because a server holding no volumes reports 0, which must
// stay distinct from a server that computes none.
optional uint64 volume_digest = 27;
}
@@ -121,8 +120,7 @@ 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.
// The reported digest disagrees, so changes alone are not enough.
bool resend_full_volume_list = 8;
}
seaweed-volume/src/server/heartbeat.rs
+4 -7
View File
@@ -801,9 +801,8 @@ 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 exactly what is reported: volumes skipped below 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<String, u32> = HashMap::new();
@@ -1253,10 +1252,8 @@ 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.
// A volume reported but left out of the digest, or the reverse, makes the
// master's comparison disagree forever.
#[test]
fn test_build_heartbeat_digests_exactly_what_it_reports() {
let temp_dir = tempfile::tempdir().unwrap();
seaweed-volume/src/storage/volume_report_hash.rs
+9 -18
View File
@@ -1,17 +1,12 @@
//! 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.
//! Mirror of `weed/storage/volume_report_hash.go`. Has to agree with it byte
//! for byte, so `report_hash_vectors` pins the layout against values produced
//! there; regenerate them before changing it.
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
/// 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];
@@ -20,8 +15,7 @@ pub fn report_hash(m: &master_pb::VolumeInformationMessage) -> u64 {
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.
// The master stores these narrowed, so hash what it will hold.
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());
@@ -40,8 +34,7 @@ pub fn report_hash(m: &master_pb::VolumeInformationMessage) -> u64 {
h
}
/// A ttl whose count is zero encodes as zero however the unit is set, matching
/// what the master stores after decoding it.
/// A zero count encodes as zero however the unit is set, as the master stores it.
fn normalize_ttl(ttl: u32) -> u32 {
let count = (ttl >> 8) & 0xff;
if count == 0 {
@@ -50,8 +43,7 @@ fn normalize_ttl(ttl: u32) -> u32 {
(count << 8) | (ttl & 0xff)
}
/// Combines two hashes order-dependently, so swapping two string fields is not
/// invisible.
/// Order-dependent, 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)
@@ -61,9 +53,8 @@ fn fold(h: u64, x: u64) -> u64 {
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.
// Produced by the Go implementation. Drift here means every volume server on
// this build silently falls back to sending its whole volume list forever.
#[test]
fn report_hash_vectors() {
let empty = master_pb::VolumeInformationMessage::default();
weed/pb/master.proto
+4 -6
View File
@@ -106,10 +106,9 @@ message Heartbeat {
map<string, uint64> disk_total_bytes = 25;
map<string, uint64> 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.
// Lets the master check its copy is current without being sent the whole
// list. Optional because a server holding no volumes reports 0, which must
// stay distinct from a server that computes none.
optional uint64 volume_digest = 27;
}
@@ -121,8 +120,7 @@ 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.
// The reported digest disagrees, so changes alone are not enough.
bool resend_full_volume_list = 8;
}
weed/pb/master_pb/master.pb.go
+4 -6
View File
@@ -121,10 +121,9 @@ 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"`
// 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.
// Lets the master check its copy is current without being sent the whole
// list. Optional because a server holding no volumes reports 0, which must
// stay distinct from a server that computes none.
VolumeDigest *uint64 `protobuf:"varint,27,opt,name=volume_digest,json=volumeDigest,proto3,oneof" json:"volume_digest,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
@@ -337,8 +336,7 @@ 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"`
// 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.
// The reported digest disagrees, so changes alone are not enough.
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
weed/server/master_grpc_server.go
+6 -12
View File
@@ -276,8 +276,8 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ
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.
// After everything the heartbeat carried is 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)
@@ -287,17 +287,13 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ
}
}
// 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.
// checkVolumeDigest reports whether the master needs the full volume list to
// recover. Servers reporting no digest are left alone; they send it anyway.
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.
// Reported twice but stored once, so no amount of resending can agree.
if dn.HasDuplicateVolumeIds() {
stats.MasterReceivedHeartbeatCounter.WithLabelValues("volumeDigestNotComparable").Inc()
return false
@@ -311,9 +307,7 @@ func (ms *MasterServer) checkVolumeDigest(heartbeat *master_pb.Heartbeat, dn *to
}
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.
// Nothing further to ask for, so say so rather than repeat the request.
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)
weed/server/master_grpc_server_digest_test.go
+1 -10
View File
@@ -22,9 +22,7 @@ func digestTestVolumeMessage(id uint32) *master_pb.VolumeInformationMessage {
}
}
// 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.
// 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)
@@ -47,9 +45,6 @@ func TestDigestCheckAcceptsAMatchingReport(t *testing.T) {
}
}
// 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)}
@@ -61,8 +56,6 @@ func TestDigestCheckDoesNotReaskAfterAFullList(t *testing.T) {
}
}
// 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)
@@ -73,8 +66,6 @@ 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) {
ms, dn := digestTestCluster(t)
duplicated := digestTestVolumeMessage(1)
weed/storage/store.go
+4 -6
View File
@@ -414,9 +414,8 @@ 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 exactly what is reported: volumes skipped below 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.
@@ -609,9 +608,8 @@ 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.
// Runs the master's own hash over the master's own conversion of the message,
// so the two ends cannot drift.
func reportHashOf(m *master_pb.VolumeInformationMessage) uint64 {
vi, err := NewVolumeInfo(m)
if err != nil {
weed/storage/store_heartbeat_digest_test.go
+3 -6
View File
@@ -17,9 +17,8 @@ func mountTestVolume(t *testing.T, loc *DiskLocation, vid needle.VolumeId) {
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.
// 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)
@@ -47,9 +46,7 @@ func TestCollectHeartbeatDigestsExactlyWhatItReports(t *testing.T) {
}
}
// 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.
// 0 must stay distinguishable from a server that computes no digest at all.
func TestCollectHeartbeatDigestsAnEmptyStore(t *testing.T) {
store := newTestStore(t, 1)
weed/topology/volume_digest_test.go
+8 -15
View File
@@ -71,10 +71,8 @@ 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.
// 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)
baseInfo, err := storage.NewVolumeInfo(base)
@@ -103,8 +101,7 @@ func TestVolumeDigestTracksEveryReportedField(t *testing.T) {
}
}
// distinctValuesFor offers values that differ from current. Several, because
// some fields are narrowed or normalised on the way into VolumeInfo and the
// Several, because some fields are narrowed 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()
@@ -186,10 +183,9 @@ func TestVolumeDigestFollowsDeltaRegistration(t *testing.T) {
}
}
// The point of the digest is not to detect that volumes changed -- in any live
// cluster some always have. It is to confirm that after applying the changes a
// heartbeat did carry, the master holds what the volume server holds. So a
// heartbeat reporting only the volumes that moved must still reconcile.
// The digest is not a change detector: in a live cluster some volumes always
// have changed. It asks whether the master is current once the heartbeat's own
// changes are applied.
func TestVolumeDigestMatchesAfterApplyingOnlyChangedVolumes(t *testing.T) {
const total = 50
full := make([]*master_pb.VolumeInformationMessage, 0, total)
@@ -232,8 +228,7 @@ func TestVolumeDigestMatchesAfterApplyingOnlyChangedVolumes(t *testing.T) {
}
}
// A volume that disappears without a delta is exactly what the full list exists
// to catch, and is the case the digest has to keep catching.
// What the full list exists to catch, and the digest has to keep catching.
func TestVolumeDigestCatchesASilentlyLostVolume(t *testing.T) {
full := []*master_pb.VolumeInformationMessage{
digestTestVolume(1), digestTestVolume(2), digestTestVolume(3),
@@ -431,9 +426,7 @@ func TestVolumeIndexDigestFollowsRemovedLookupEntry(t *testing.T) {
}
}
// 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.
// Real heartbeat data, not hand-built messages.
func TestMasterDigestMatchesWhatAVolumeServerReports(t *testing.T) {
dir := t.TempDir()
loc := storage.NewDiskLocation(dir, 100, util.MinFreeSpace{}, "", types.HardDriveType, nil,