diff --git a/weed/ec/ec_encode.go b/weed/ec/ec_encode.go index 8dfe9beb8..5ce8f4a9f 100644 --- a/weed/ec/ec_encode.go +++ b/weed/ec/ec_encode.go @@ -9,6 +9,7 @@ import ( "slices" "sort" "strconv" + "strings" "sync" "time" @@ -576,6 +577,58 @@ func CollectEcShardBitsByNode(topoInfo *master_pb.TopologyInfo, vid needle.Volum return res } +// collectNewestGenerationShardsInfo answers the same question as +// CollectEcShardBitsByNode -- which shards belong to the newest encode +// generation -- and keeps the sizes with them. +// +// The two have to agree on the generation. A re-encode can change the ratio, +// so an orphaned older generation's shards are a different length by nature: +// merging them into the size comparison makes a healthy current set look +// inconsistent, and since the orphan keeps being reported, every retry fails +// and the encode is left holding both the volume and its shards forever. +func collectNewestGenerationShardsInfo(topoInfo *master_pb.TopologyInfo, vid needle.VolumeId) map[pb.ServerAddress]*erasure_coding.ShardsInfo { + type shardEntry struct { + addr pb.ServerAddress + ts int64 + info *erasure_coding.ShardsInfo + } + var entries []shardEntry + var newestTs int64 + EachDataNode(topoInfo, func(dc DataCenterId, rack RackId, dn *master_pb.DataNodeInfo) { + for _, diskInfo := range dn.DiskInfos { + if diskInfo == nil { + continue + } + for _, ecInfo := range diskInfo.EcShardInfos { + if ecInfo.Id != uint32(vid) { + continue + } + entries = append(entries, shardEntry{ + addr: pb.NewServerAddressFromDataNode(dn), + ts: ecInfo.EncodeTsNs, + info: erasure_coding.ShardsInfoFromVolumeEcShardInformationMessage(ecInfo), + }) + if ecInfo.EncodeTsNs > newestTs { + newestTs = ecInfo.EncodeTsNs + } + } + } + }) + + res := make(map[pb.ServerAddress]*erasure_coding.ShardsInfo) + for _, e := range entries { + if e.ts != newestTs { + continue + } + if existing, ok := res[e.addr]; ok { + existing.Add(e.info) + } else { + res[e.addr] = e.info + } + } + return res +} + // waitForEcShardsToRegister polls the master topology until every given volume // reports a full EC shard set. Mounting shards notifies the master // asynchronously (mount -> NewEcShardsChan -> delta heartbeat), so a topology @@ -642,6 +695,55 @@ func ecShardsClumpedOnOneNode(topoInfo *master_pb.TopologyInfo, vid needle.Volum // sorted so the message is stable. It names the ids and not just the count: a // set holding shards 0-9 and one holding 4-13 are both "10 shards", and which // ones survived is what says whether the set is recoverable and from where. +// requireUniformShardSizes reports shards whose size disagrees with the rest. +// Every shard of a volume takes one piece of each block row, so they are all +// written to the same length -- an odd one out is a shard that was truncated, +// half copied, or written to a disk that filled up. Counting shards cannot see +// that, and the encode is about to delete the volume they were made from. +// +// Sizes the cluster does not report (zero) are skipped rather than treated as +// a disagreement: an older volume server, or one that has not yet heartbeated +// its shard sizes, must not block an encode that is otherwise sound. +func requireUniformShardSizes(vid needle.VolumeId, byNode map[pb.ServerAddress]*erasure_coding.ShardsInfo) error { + type holder struct { + server pb.ServerAddress + shard erasure_coding.ShardId + } + sizes := make(map[int64][]holder) + for server, si := range byNode { + if si == nil { + continue + } + for _, id := range si.Ids() { + size := int64(si.Size(id)) + if size == 0 { + continue + } + sizes[size] = append(sizes[size], holder{server: server, shard: id}) + } + } + if len(sizes) <= 1 { + return nil + } + + described := make([]string, 0, len(sizes)) + for size, holders := range sizes { + sort.Slice(holders, func(i, j int) bool { + if holders[i].server == holders[j].server { + return holders[i].shard < holders[j].shard + } + return holders[i].server < holders[j].server + }) + shards := make([]string, 0, len(holders)) + for _, h := range holders { + shards = append(shards, fmt.Sprintf("%s.%d", h.server, h.shard)) + } + described = append(described, fmt.Sprintf("%d bytes: %s", size, strings.Join(shards, " "))) + } + sort.Strings(described) + return fmt.Errorf("volume %d ec shards disagree on size, so at least one is incomplete (%s)", vid, strings.Join(described, "; ")) +} + func ecShardSummaryByNode(byNode map[pb.ServerAddress]erasure_coding.ShardBits) []string { summary := make([]string, 0, len(byNode)) for node, bits := range byNode { @@ -700,6 +802,14 @@ func verifyEcShardsBeforeDelete(env *Env, volumeIds []needle.VolumeId, diskType lastErr = fmt.Errorf("volume %d: %w (observed: %v)", vid, err, ecShardSummaryByNode(byNode)) break } + // Counting the shards says they exist, not that they are whole. The + // source volume is about to be deleted on their word, so also require + // the sizes to agree -- judging the same generation the count above + // judged, or an orphaned older encode would veto a healthy set. + if err := requireUniformShardSizes(vid, collectNewestGenerationShardsInfo(topoInfo, vid)); err != nil { + lastErr = err + break + } if expectSpread { if holder, clumped := ecShardsClumpedOnOneNode(topoInfo, vid, diskType); clumped { lastClumped = append(lastClumped, fmt.Sprintf("volume %d: all shards on %s", vid, holder)) diff --git a/weed/ec/ec_encode_shard_sizes_test.go b/weed/ec/ec_encode_shard_sizes_test.go new file mode 100644 index 000000000..cfe06b129 --- /dev/null +++ b/weed/ec/ec_encode_shard_sizes_test.go @@ -0,0 +1,214 @@ +package ec + +import ( + "strings" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" +) + +// sizedShards builds one holder's inventory as {shard id: size}. +func sizedShards(idToSize map[int]int64) *erasure_coding.ShardsInfo { + si := erasure_coding.NewShardsInfo() + for id, size := range idToSize { + si.Set(erasure_coding.NewShardInfo(erasure_coding.ShardId(id), erasure_coding.ShardSize(size))) + } + return si +} + +// An encode deletes the volume the shards were made from, on the strength of +// having counted them. Every shard takes one piece of each block row, so they +// are written to the same length: one that disagrees was truncated or half +// copied, and counting cannot see it. +func TestRequireUniformShardSizes(t *testing.T) { + tests := []struct { + name string + byNode map[pb.ServerAddress]*erasure_coding.ShardsInfo + wantErr bool + errContains []string + }{ + { + name: "one holder, all shards the same length", + byNode: map[pb.ServerAddress]*erasure_coding.ShardsInfo{ + "server1:8080": sizedShards(map[int]int64{0: 1048576, 1: 1048576, 2: 1048576}), + }, + }, + { + name: "spread across holders, still one length", + byNode: map[pb.ServerAddress]*erasure_coding.ShardsInfo{ + "server1:8080": sizedShards(map[int]int64{0: 1048576, 1: 1048576}), + "server2:8080": sizedShards(map[int]int64{2: 1048576, 3: 1048576}), + }, + }, + { + name: "a truncated shard is named with its holder", + byNode: map[pb.ServerAddress]*erasure_coding.ShardsInfo{ + "server1:8080": sizedShards(map[int]int64{0: 1048576, 1: 1048576}), + "server2:8080": sizedShards(map[int]int64{2: 4096}), + }, + wantErr: true, + errContains: []string{"disagree on size", "server2:8080.2", "4096"}, + }, + { + name: "two copies of one shard that disagree are still a disagreement", + byNode: map[pb.ServerAddress]*erasure_coding.ShardsInfo{ + "server1:8080": sizedShards(map[int]int64{5: 1048576}), + "server2:8080": sizedShards(map[int]int64{5: 1048570}), + }, + wantErr: true, + errContains: []string{"disagree on size"}, + }, + + // Sizes the cluster does not report must never block an encode: an + // older volume server, or one that has not heartbeated its sizes yet, + // reports zero, and refusing to delete on that would strand every + // encode against such a cluster in the hybrid state this check exists + // to avoid. + { + name: "unreported sizes are skipped, not read as a disagreement", + byNode: map[pb.ServerAddress]*erasure_coding.ShardsInfo{ + "server1:8080": sizedShards(map[int]int64{0: 1048576, 1: 0}), + "server2:8080": sizedShards(map[int]int64{2: 0}), + }, + }, + { + name: "nothing reported at all verifies vacuously", + byNode: map[pb.ServerAddress]*erasure_coding.ShardsInfo{ + "server1:8080": sizedShards(map[int]int64{0: 0, 1: 0}), + }, + }, + { + name: "no holders is not a size problem", + byNode: map[pb.ServerAddress]*erasure_coding.ShardsInfo{}, + }, + { + name: "a nil inventory is skipped rather than panicking", + byNode: map[pb.ServerAddress]*erasure_coding.ShardsInfo{ + "server1:8080": nil, + "server2:8080": sizedShards(map[int]int64{0: 1048576}), + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := requireUniformShardSizes(needle.VolumeId(7), tt.byNode) + if !tt.wantErr { + if err != nil { + t.Fatalf("want the encode to proceed, got %v", err) + } + return + } + if err == nil { + t.Fatal("want the deletion held back, got no error") + } + for _, want := range tt.errContains { + if !strings.Contains(err.Error(), want) { + t.Errorf("error %q does not mention %q", err, want) + } + } + }) + } +} + +// ecShardReport builds one heartbeat entry: a generation stamp and the shards +// that generation put on this node, all of one size. +func ecShardReport(vid uint32, encodeTsNs int64, size int64, ids ...int) *master_pb.VolumeEcShardInformationMessage { + si := erasure_coding.NewShardsInfo() + for _, id := range ids { + si.Set(erasure_coding.NewShardInfo(erasure_coding.ShardId(id), erasure_coding.ShardSize(size))) + } + return &master_pb.VolumeEcShardInformationMessage{ + Id: vid, + Collection: "ectest", + EcIndexBits: si.Bitmap(), + ShardSizes: si.SizesInt64(), + EncodeTsNs: encodeTsNs, + } +} + +func topologyWith(reports map[string][]*master_pb.VolumeEcShardInformationMessage) *master_pb.TopologyInfo { + var nodes []*master_pb.DataNodeInfo + for addr, shards := range reports { + nodes = append(nodes, &master_pb.DataNodeInfo{ + Id: addr, + DiskInfos: map[string]*master_pb.DiskInfo{ + "hdd": {EcShardInfos: shards}, + }, + }) + } + return &master_pb.TopologyInfo{ + DataCenterInfos: []*master_pb.DataCenterInfo{{ + Id: "dc1", + RackInfos: []*master_pb.RackInfo{{Id: "rack1", DataNodeInfos: nodes}}, + }}, + } +} + +// A re-encode can change the ratio, so an older generation's shards are a +// different length by nature. Recoverability already judges only the newest +// generation; if the size check did not, an orphaned older encode -- one the +// pre-encode sweep could not reach, but the master still hears about -- would +// disagree on size at every retry and strand the volume beside its shards for +// good. +func TestShardSizeCheckIgnoresOlderEncodeGenerations(t *testing.T) { + const vid = 9 + + topo := topologyWith(map[string][]*master_pb.VolumeEcShardInformationMessage{ + // the current encode: one length, spread over two nodes + "server1:8080": {ecShardReport(vid, 2000, 1048576, 0, 1, 2)}, + "server2:8080": {ecShardReport(vid, 2000, 1048576, 3, 4, 5)}, + // an orphan from an earlier encode, at a different length + "server3:8080": {ecShardReport(vid, 1000, 524288, 0, 1)}, + }) + + byNode := collectNewestGenerationShardsInfo(topo, needle.VolumeId(vid)) + if _, stale := byNode[pb.ServerAddress("server3:8080")]; stale { + t.Error("the older generation's holder must not be collected") + } + if err := requireUniformShardSizes(needle.VolumeId(vid), byNode); err != nil { + t.Fatalf("a healthy current generation must not be vetoed by an orphan: %v", err) + } +} + +// Fencing to the newest generation must not blind the check: a short shard +// inside that generation is exactly what it exists to catch. +func TestShardSizeCheckStillCatchesShortShardInNewestGeneration(t *testing.T) { + const vid = 9 + + topo := topologyWith(map[string][]*master_pb.VolumeEcShardInformationMessage{ + "server1:8080": {ecShardReport(vid, 2000, 1048576, 0, 1)}, + "server2:8080": {ecShardReport(vid, 2000, 4096, 2)}, + "server3:8080": {ecShardReport(vid, 1000, 524288, 0)}, + }) + + err := requireUniformShardSizes(needle.VolumeId(vid), collectNewestGenerationShardsInfo(topo, needle.VolumeId(vid))) + if err == nil { + t.Fatal("a truncated shard in the newest generation must hold the deletion back") + } + if !strings.Contains(err.Error(), "server2:8080.2") { + t.Errorf("error %q does not name the short shard", err) + } +} + +// Volumes encoded before generation stamping report zero, which is a single +// legacy generation rather than an orphan to drop. +func TestShardSizeCheckHandlesUnstampedGenerations(t *testing.T) { + const vid = 9 + + topo := topologyWith(map[string][]*master_pb.VolumeEcShardInformationMessage{ + "server1:8080": {ecShardReport(vid, 0, 1048576, 0, 1)}, + "server2:8080": {ecShardReport(vid, 0, 4096, 2)}, + }) + + byNode := collectNewestGenerationShardsInfo(topo, needle.VolumeId(vid)) + if len(byNode) != 2 { + t.Fatalf("unstamped shards form one generation, got %d holder(s)", len(byNode)) + } + if err := requireUniformShardSizes(needle.VolumeId(vid), byNode); err == nil { + t.Error("a truncated legacy shard must still hold the deletion back") + } +}