mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-08-16 04:06:44 +00:00
ec.encode: require the shards to agree on size before deleting the volume (#10769)
* ec.encode: require the shards to agree on size before deleting the volume Before an encode deletes the volume it just encoded, it asks whether enough shards exist and whether they are spread across nodes. Both are questions about presence: nothing asks whether those shards are whole. Every shard takes one piece of each block row, so they are all written to the same length. One that disagrees was truncated, half copied, or landed on a disk that filled up -- and counting cannot see it, so the source volume is deleted on the strength of a set that cannot rebuild it. Compare the sizes the cluster already reports (shard_sizes travels in the heartbeat) and hold the deletion back when they disagree, naming the odd shard and its holder. Sizes reported as zero are skipped rather than read as a disagreement: a volume server that predates shard-size reporting, or one that has not heartbeated them yet, must not strand every encode in the volume-plus-shards state this check exists to avoid. * ec.encode: judge shard sizes on the newest encode generation only The size check collected every shard the master reports for the volume, while the recoverability check beside it counts only the newest encode generation. A re-encode can change the ratio, so an orphaned older generation -- one the pre-encode sweep could not reach, but the master still hears about -- has shards of a different length by nature. Merging those into the comparison makes a healthy current set look inconsistent, and because the orphan keeps being reported, every retry fails and the encode is left holding the volume and its shards for good. Collect sizes the way CollectEcShardBitsByNode collects bits: fenced to the newest EncodeTsNs, with unstamped entries forming the one legacy generation.
This commit is contained in:
Chris Lu
@@ -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))
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user