mirror of
https://github.com/cubefs/cubefs.git
synced 2026-08-02 02:00:56 +00:00
feat(blobnode): repair crc-corrupt shards in place during inspect
with: #1001092565 Signed-off-by: xiejian <xiejian3@oppo.com>
This commit is contained in:
xiejian
沈杰
parent
7dde9a7d3c
commit
e899894ce1
@ -17,8 +17,10 @@ import (
|
||||
|
||||
bnapi "github.com/cubefs/cubefs/blobstore/api/blobnode"
|
||||
"github.com/cubefs/cubefs/blobstore/api/clustermgr"
|
||||
"github.com/cubefs/cubefs/blobstore/api/proxy"
|
||||
"github.com/cubefs/cubefs/blobstore/blobnode/base"
|
||||
"github.com/cubefs/cubefs/blobstore/blobnode/core"
|
||||
"github.com/cubefs/cubefs/blobstore/common/crc32block"
|
||||
bloberr "github.com/cubefs/cubefs/blobstore/common/errors"
|
||||
"github.com/cubefs/cubefs/blobstore/common/proto"
|
||||
"github.com/cubefs/cubefs/blobstore/common/recordlog"
|
||||
@ -52,6 +54,8 @@ type DataInspectConf struct {
|
||||
NextRoundSec int `json:"next_round_sec"` // wait next round inspect interval
|
||||
BatchReadSize int64 `json:"batch_read_size"` // max data bytes per BatchRead call; default 16MB
|
||||
|
||||
Proxy proxy.LbConfig `json:"proxy"` // used to send crc repair messages.
|
||||
|
||||
Record recordlog.Config `json:"record"`
|
||||
}
|
||||
|
||||
@ -61,6 +65,23 @@ type DataInspectStat struct {
|
||||
Progress map[proto.DiskID]int `json:"progress"`
|
||||
}
|
||||
|
||||
// lazyRepairSender builds the proxy repair client on first use and caches it.
|
||||
type lazyRepairSender struct {
|
||||
once sync.Once
|
||||
sender proxy.LbMsgSender
|
||||
build func() proxy.LbMsgSender
|
||||
}
|
||||
|
||||
// get returns the cached sender, building it once on first call.
|
||||
func (l *lazyRepairSender) get() proxy.LbMsgSender {
|
||||
l.once.Do(func() {
|
||||
if l.build != nil {
|
||||
l.sender = l.build()
|
||||
}
|
||||
})
|
||||
return l.sender
|
||||
}
|
||||
|
||||
type DataInspectMgr struct {
|
||||
conf DataInspectConf
|
||||
limits map[proto.DiskID]*rate.Limiter
|
||||
@ -71,6 +92,8 @@ type DataInspectMgr struct {
|
||||
round uint64 // round of data inspect
|
||||
progress sync.Map // progress of data inspect: diskID -> progress[0, 100]
|
||||
recorder recordlog.Encoder // local record log
|
||||
|
||||
repairSender lazyRepairSender // proxy client to send crc repair msg, lazily built on first send
|
||||
}
|
||||
|
||||
func NewDataInspectMgr(svr *Service, conf DataInspectConf, switchMgr *taskswitch.SwitchMgr) (*DataInspectMgr, error) {
|
||||
@ -100,6 +123,9 @@ func NewDataInspectMgr(svr *Service, conf DataInspectConf, switchMgr *taskswitch
|
||||
taskSwitch: taskSwitch,
|
||||
recorder: recorder,
|
||||
}
|
||||
mgr.repairSender.build = func() proxy.LbMsgSender {
|
||||
return proxy.NewMQLbClient(&mgr.conf.Proxy, svr.ClusterMgrClient, svr.Conf.HostInfo.ClusterID)
|
||||
}
|
||||
return mgr, nil
|
||||
}
|
||||
|
||||
@ -441,11 +467,49 @@ func (mgr *DataInspectMgr) inspectChunk(pCtx context.Context, cs core.ChunkAPI)
|
||||
|
||||
err := mgr.scanShards(ctx, cs, scanFn)
|
||||
mgr.reportBatchBadShards(ctx, cs, badShards)
|
||||
mgr.trySendCrcRepair(ctx, cs, badShards)
|
||||
span.Infof("finish to inspect chunk, vuid:%d, chunk:%s, total:%d, wrong:%d, err:%+v",
|
||||
cs.Vuid(), cs.ID(), total, len(badShards), err)
|
||||
return badShards, err
|
||||
}
|
||||
|
||||
// trySendCrcRepair sends an in-place crc repair message to proxy for each
|
||||
// confirmed CRC-corrupted shard.
|
||||
func (mgr *DataInspectMgr) trySendCrcRepair(ctx context.Context, cs core.ChunkAPI, badShards []bnapi.BadShard) {
|
||||
if len(badShards) == 0 {
|
||||
return
|
||||
}
|
||||
sender := mgr.repairSender.get()
|
||||
if sender == nil {
|
||||
return
|
||||
}
|
||||
span := trace.SpanFromContextSafe(ctx)
|
||||
clusterID := cs.Disk().DiskInfo().ClusterID
|
||||
|
||||
for _, bad := range badShards {
|
||||
// only data-corruption (crc) shards are repaired in place; other error
|
||||
// types are reported only and must not trigger an in-place rebuild.
|
||||
if !errors.Is(bad.Err, crc32block.ErrMismatchedCrc) {
|
||||
continue
|
||||
}
|
||||
idx := bad.Vuid.Index()
|
||||
// the inspect-crc reason tells the repairer to rebuild this idx in place
|
||||
// even though its meta still reports Normal.
|
||||
args := &proxy.ShardRepairArgs{
|
||||
ClusterID: clusterID,
|
||||
Bid: bad.Bid,
|
||||
Vid: bad.Vuid.Vid(),
|
||||
BadIdxes: []uint8{idx},
|
||||
Reason: proto.ShardRepairReasonInspectCrc,
|
||||
}
|
||||
if err := sender.SendShardRepairMsg(ctx, args); err != nil {
|
||||
span.Errorf("send crc repair msg failed, vuid:%d, bid:%d, err:%+v", bad.Vuid, bad.Bid, err)
|
||||
continue
|
||||
}
|
||||
span.Infof("send crc repair msg, vuid:%d, bid:%d, idx:%d", bad.Vuid, bad.Bid, idx)
|
||||
}
|
||||
}
|
||||
|
||||
// inspectShard checks shard integrity and metadata double-check.
|
||||
// Returns error if shard is corrupted, nil if healthy or deleted.
|
||||
func (mgr *DataInspectMgr) inspectShard(ctx context.Context, cs core.ChunkAPI, si *bnapi.ShardInfo) (err error) {
|
||||
@ -524,6 +588,12 @@ func (mgr *DataInspectMgr) reportBadShard(ctx context.Context, cs core.ChunkAPI,
|
||||
// Report with "add" and combine it with "record" for analysis and processing
|
||||
diskInfo := cs.Disk().DiskInfo()
|
||||
mgr.recordBadBids(ctx, cs, []string{blobID.ToString()}, err.Error())
|
||||
|
||||
// crc-mismatch shard is handled by in-place repair, so it is not reported to the metric
|
||||
if errors.Is(err, crc32block.ErrMismatchedCrc) {
|
||||
return
|
||||
}
|
||||
|
||||
dataInspectMetric.WithLabelValues(
|
||||
diskInfo.ClusterID.ToString(),
|
||||
diskInfo.DiskID.ToString(),
|
||||
@ -543,6 +613,7 @@ func (mgr *DataInspectMgr) reportBatchBadShards(ctx context.Context, cs core.Chu
|
||||
// "err 22": ["bid66", "77", "88"],
|
||||
// }
|
||||
uniqueErr := map[string][]string{}
|
||||
metricBadBid := 0 // count of bad bids reported to metric, excluding crc (repaired in place)
|
||||
for _, item := range items {
|
||||
if isInspectReportIgnoredError(item.Err) {
|
||||
continue
|
||||
@ -550,26 +621,34 @@ func (mgr *DataInspectMgr) reportBatchBadShards(ctx context.Context, cs core.Chu
|
||||
|
||||
uniqueErr[item.Err.Error()] = append(uniqueErr[item.Err.Error()], item.Bid.ToString())
|
||||
span.Errorf("inspect blob error, bad shard:%v", item)
|
||||
// crc-mismatch shards are handled by in-place repair, so they are still
|
||||
// recorded to the local log but not reported to the cumulative metric
|
||||
// (which cannot be cleaned after repair).
|
||||
if !errors.Is(item.Err, crc32block.ErrMismatchedCrc) {
|
||||
metricBadBid++
|
||||
}
|
||||
}
|
||||
|
||||
if len(uniqueErr) == 0 {
|
||||
return 0
|
||||
}
|
||||
|
||||
// record local log
|
||||
// record local log (includes crc-mismatch shards)
|
||||
totalBadBid, diskInfo := 0, cs.Disk().DiskInfo()
|
||||
for errStr, bids := range uniqueErr {
|
||||
totalBadBid += len(bids)
|
||||
mgr.recordBadBids(ctx, cs, bids, errStr)
|
||||
}
|
||||
|
||||
// report metric
|
||||
dataInspectMetric.WithLabelValues(
|
||||
diskInfo.ClusterID.ToString(),
|
||||
diskInfo.DiskID.ToString(),
|
||||
).Add(float64(totalBadBid))
|
||||
// report metric, excluding crc-mismatch shards
|
||||
if metricBadBid > 0 {
|
||||
dataInspectMetric.WithLabelValues(
|
||||
diskInfo.ClusterID.ToString(),
|
||||
diskInfo.DiskID.ToString(),
|
||||
).Add(float64(metricBadBid))
|
||||
}
|
||||
|
||||
span.Errorf("inspect blob error, total bad count:%d", totalBadBid)
|
||||
span.Errorf("inspect blob error, total bad count:%d, metric count:%d", totalBadBid, metricBadBid)
|
||||
return totalBadBid
|
||||
}
|
||||
|
||||
|
||||
@ -13,12 +13,16 @@ import (
|
||||
"testing"
|
||||
|
||||
"github.com/golang/mock/gomock"
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
dto "github.com/prometheus/client_model/go"
|
||||
"github.com/stretchr/testify/require"
|
||||
"golang.org/x/time/rate"
|
||||
|
||||
bnapi "github.com/cubefs/cubefs/blobstore/api/blobnode"
|
||||
"github.com/cubefs/cubefs/blobstore/api/clustermgr"
|
||||
"github.com/cubefs/cubefs/blobstore/api/proxy"
|
||||
"github.com/cubefs/cubefs/blobstore/blobnode/core"
|
||||
"github.com/cubefs/cubefs/blobstore/common/crc32block"
|
||||
bloberr "github.com/cubefs/cubefs/blobstore/common/errors"
|
||||
"github.com/cubefs/cubefs/blobstore/common/proto"
|
||||
"github.com/cubefs/cubefs/blobstore/common/recordlog"
|
||||
@ -44,9 +48,65 @@ func newDataInspectMgr(t *testing.T, conf DataInspectConf, svr *Service) *DataIn
|
||||
recorder := mocks.NewMockRecordLogEncoder(ctr)
|
||||
mgr.recorder = recorder
|
||||
|
||||
// unit tests have no real cluster manager/proxy; disable the lazy builder so
|
||||
// trySendCrcRepair gets a nil sender and returns early (svr.Conf may be nil).
|
||||
mgr.repairSender.build = nil
|
||||
|
||||
return mgr
|
||||
}
|
||||
|
||||
func TestTrySendCrcRepair(t *testing.T) {
|
||||
ctr := gomock.NewController(t)
|
||||
ctx := context.Background()
|
||||
|
||||
vuid0, _ := proto.NewVuid(100, 0, 1)
|
||||
vuid3, _ := proto.NewVuid(100, 3, 1)
|
||||
|
||||
// sender nil: no-op, no chunk methods touched.
|
||||
mgr := &DataInspectMgr{}
|
||||
mgr.trySendCrcRepair(ctx, NewMockChunkAPI(ctr), []bnapi.BadShard{{Vuid: vuid0, Bid: 1, Err: errMock}})
|
||||
|
||||
// sender set: only a crc-bad shard is sent; deleted and other errors skipped.
|
||||
var sent []*proxy.ShardRepairArgs
|
||||
sender := mocks.NewMockProxyLbRpcClient(ctr)
|
||||
sender.EXPECT().SendShardRepairMsg(any, any).Times(1).DoAndReturn(
|
||||
func(_ context.Context, args *proxy.ShardRepairArgs) error {
|
||||
sent = append(sent, args)
|
||||
return nil
|
||||
})
|
||||
mgr.repairSender.sender = sender
|
||||
|
||||
ds := NewMockDiskAPI(ctr)
|
||||
ds.EXPECT().DiskInfo().Return(clustermgr.BlobNodeDiskInfo{
|
||||
DiskInfo: clustermgr.DiskInfo{ClusterID: 7},
|
||||
}).Times(1)
|
||||
cs := NewMockChunkAPI(ctr)
|
||||
cs.EXPECT().Disk().Return(ds).Times(1)
|
||||
|
||||
mgr.trySendCrcRepair(ctx, cs, []bnapi.BadShard{
|
||||
{Vuid: vuid0, Bid: 11, Err: crc32block.ErrMismatchedCrc}, // crc bad -> sent
|
||||
{Vuid: vuid3, Bid: 22, Err: bloberr.ErrNoSuchBid}, // deleted -> skipped
|
||||
{Vuid: vuid3, Bid: 23, Err: errMock}, // other error -> only reported, skipped
|
||||
})
|
||||
|
||||
require.Equal(t, 1, len(sent))
|
||||
got := sent[0]
|
||||
require.Equal(t, proto.ClusterID(7), got.ClusterID)
|
||||
require.Equal(t, proto.BlobID(11), got.Bid)
|
||||
require.Equal(t, proto.Vid(100), got.Vid)
|
||||
require.Equal(t, []uint8{0}, got.BadIdxes)
|
||||
require.Equal(t, proto.ShardRepairReasonInspectCrc, got.Reason)
|
||||
|
||||
// sender error does not panic and does not crash inspection.
|
||||
sender2 := mocks.NewMockProxyLbRpcClient(ctr)
|
||||
sender2.EXPECT().SendShardRepairMsg(any, any).Return(errMock).Times(1)
|
||||
mgr.repairSender.sender = sender2
|
||||
ds.EXPECT().DiskInfo().Return(clustermgr.BlobNodeDiskInfo{}).Times(1)
|
||||
cs2 := NewMockChunkAPI(ctr)
|
||||
cs2.EXPECT().Disk().Return(ds).Times(1)
|
||||
mgr.trySendCrcRepair(ctx, cs2, []bnapi.BadShard{{Vuid: vuid0, Bid: 33, Err: crc32block.ErrMismatchedCrc}})
|
||||
}
|
||||
|
||||
func TestDataInspect(t *testing.T) {
|
||||
ctr := gomock.NewController(t)
|
||||
ctx := context.Background()
|
||||
@ -307,6 +367,56 @@ func TestInspectChunk_NoGoroutineLeak(t *testing.T) {
|
||||
require.LessOrEqual(t, after, before+tolerance)
|
||||
}
|
||||
|
||||
func TestReportBatchBadShardsCrcMetric(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
ctr := gomock.NewController(t)
|
||||
ds := NewMockDiskAPI(ctr)
|
||||
svr := &Service{
|
||||
Disks: map[proto.DiskID]core.DiskAPI{11: ds},
|
||||
ctx: context.Background(),
|
||||
closeCh: make(chan struct{}),
|
||||
}
|
||||
mgr := newDataInspectMgr(t, DataInspectConf{IntervalSec: 100, RateLimit: 2}, svr)
|
||||
defer close(svr.closeCh)
|
||||
|
||||
const cid, did = 9001, 9002
|
||||
ds.EXPECT().DiskInfo().Return(clustermgr.BlobNodeDiskInfo{
|
||||
DiskInfo: clustermgr.DiskInfo{ClusterID: cid},
|
||||
DiskHeartBeatInfo: clustermgr.DiskHeartBeatInfo{DiskID: did},
|
||||
}).AnyTimes()
|
||||
cs := NewMockChunkAPI(ctr)
|
||||
cs.EXPECT().Disk().Return(ds).AnyTimes()
|
||||
cs.EXPECT().Vuid().Return(proto.Vuid(1001)).AnyTimes()
|
||||
mgr.recorder.(*mocks.MockRecordLogEncoder).EXPECT().Encode(any).AnyTimes()
|
||||
|
||||
gauge := dataInspectMetric.WithLabelValues(proto.ClusterID(cid).ToString(), proto.DiskID(did).ToString())
|
||||
gauge.Set(0)
|
||||
|
||||
// crc-only batch: recorded to local log but NOT counted into the metric.
|
||||
crcBads := []bnapi.BadShard{
|
||||
{DiskID: did, Vuid: 1001, Bid: 1, Err: crc32block.ErrMismatchedCrc},
|
||||
{DiskID: did, Vuid: 1001, Bid: 2, Err: crc32block.ErrMismatchedCrc},
|
||||
}
|
||||
require.Equal(t, 2, mgr.reportBatchBadShards(ctx, cs, crcBads))
|
||||
require.Equal(t, float64(0), gaugeValue(gauge))
|
||||
|
||||
// mixed batch: only the non-crc shard is reported to the metric.
|
||||
mixedBads := []bnapi.BadShard{
|
||||
{DiskID: did, Vuid: 1001, Bid: 3, Err: crc32block.ErrMismatchedCrc},
|
||||
{DiskID: did, Vuid: 1001, Bid: 4, Err: errMock},
|
||||
}
|
||||
require.Equal(t, 2, mgr.reportBatchBadShards(ctx, cs, mixedBads))
|
||||
require.Equal(t, float64(1), gaugeValue(gauge))
|
||||
}
|
||||
|
||||
func gaugeValue(g prometheus.Gauge) float64 {
|
||||
m := &dto.Metric{}
|
||||
if err := g.Write(m); err != nil {
|
||||
return -1
|
||||
}
|
||||
return m.GetGauge().GetValue()
|
||||
}
|
||||
|
||||
func TestDataInspectMetric(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
ctr := gomock.NewController(t)
|
||||
|
||||
@ -234,6 +234,14 @@ func (getter *MockGetter) Delete(ctx context.Context, vuid proto.Vuid, bid proto
|
||||
getter.vunits[vuid].delete(bid)
|
||||
}
|
||||
|
||||
// Corrupt garbles the shard data in place while keeping the meta Normal,
|
||||
// simulating a silent CRC corruption detected by data inspect.
|
||||
func (getter *MockGetter) Corrupt(ctx context.Context, vuid proto.Vuid, bid proto.BlobID) {
|
||||
getter.mu.Lock()
|
||||
defer getter.mu.Unlock()
|
||||
getter.vunits[vuid].corrupt(bid)
|
||||
}
|
||||
|
||||
func (getter *MockGetter) StatShard(ctx context.Context, location proto.VunitLocation, bid proto.BlobID, ioType api.IOType) (si *client.ShardInfo, err error) {
|
||||
getter.mu.Lock()
|
||||
defer getter.mu.Unlock()
|
||||
@ -404,6 +412,23 @@ func (m *mockVunit) recover(bid proto.BlobID) {
|
||||
m.bidInfos[bid].Flag = api.ShardStatusNormal
|
||||
}
|
||||
|
||||
// corrupt simulates silent data corruption: the on-disk data is garbled while
|
||||
// the shard meta (Flag/Crc) stays Normal, just like a real CRC rot found by
|
||||
// data inspect. statShard still reports Normal; only reading the data reveals it.
|
||||
func (m *mockVunit) corrupt(bid proto.BlobID) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
if _, ok := m.shards[bid]; !ok {
|
||||
return
|
||||
}
|
||||
bad := make([]byte, len(m.shards[bid]))
|
||||
for i := range bad {
|
||||
bad[i] = m.shards[bid][i] ^ 0xff
|
||||
}
|
||||
m.shards[bid] = bad
|
||||
m.crc32[bid] = crc32.ChecksumIEEE(bad)
|
||||
}
|
||||
|
||||
type mockShardGetter struct {
|
||||
body io.ReadCloser
|
||||
bids []api.BidInfo
|
||||
|
||||
@ -89,9 +89,15 @@ func (repairer *ShardRepairer) RepairShard(ctx context.Context, task *proto.Shar
|
||||
}
|
||||
|
||||
shardInfos := repairer.listShardsInfo(ctx, task.Sources, task.Bid)
|
||||
// crc-repair tasks must rebuild every BadIdx in place even if meta reports
|
||||
// Normal; such tasks are identified by the inspect-crc reason.
|
||||
var forceSet map[int]struct{}
|
||||
if task.Reason == proto.ShardRepairReasonInspectCrc {
|
||||
forceSet = uint8SliceToIntSet(task.BadIdxs)
|
||||
}
|
||||
// check missed shard has repaired
|
||||
badIdxs := sliceUint8ToInt(task.BadIdxs)
|
||||
repaired, err := hasRepaired(shardInfos, badIdxs)
|
||||
repaired, err := hasRepaired(shardInfos, badIdxs, forceSet)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@ -99,7 +105,7 @@ func (repairer *ShardRepairer) RepairShard(ctx context.Context, task *proto.Shar
|
||||
return nil
|
||||
}
|
||||
// get repair shards:
|
||||
shouldRepairIdxs, shardSize, err := getRepairShards(ctx, shardInfos, task.CodeMode, badIdxs)
|
||||
shouldRepairIdxs, shardSize, err := getRepairShards(ctx, shardInfos, task.CodeMode, badIdxs, forceSet)
|
||||
if err != nil {
|
||||
span.Errorf("get repair shards failed: task[%+v], err[%+v]", err, task)
|
||||
return err
|
||||
@ -156,7 +162,7 @@ func (repairer *ShardRepairer) RepairShard(ctx context.Context, task *proto.Shar
|
||||
return err
|
||||
}
|
||||
|
||||
func hasRepaired(shardInfos []*ShardInfoEx, repairIdxs []int) (bool, error) {
|
||||
func hasRepaired(shardInfos []*ShardInfoEx, repairIdxs []int, forceSet map[int]struct{}) (bool, error) {
|
||||
var repairCnt int
|
||||
for idx, shard := range shardInfos {
|
||||
if !contains(idx, repairIdxs) {
|
||||
@ -166,6 +172,10 @@ func hasRepaired(shardInfos []*ShardInfoEx, repairIdxs []int) (bool, error) {
|
||||
if shard.IsBad() {
|
||||
return false, errcode.ErrDestReplicaBad
|
||||
}
|
||||
// crc-bad idx: Normal in meta but corrupt, never count as repaired.
|
||||
if _, force := forceSet[idx]; force {
|
||||
continue
|
||||
}
|
||||
if shard.Normal() {
|
||||
repairCnt++
|
||||
}
|
||||
@ -177,7 +187,7 @@ func hasRepaired(shardInfos []*ShardInfoEx, repairIdxs []int) (bool, error) {
|
||||
return false, nil
|
||||
}
|
||||
|
||||
func getRepairShards(ctx context.Context, shardInfos []*ShardInfoEx, mode codemode.CodeMode, repaireIdxs []int,
|
||||
func getRepairShards(ctx context.Context, shardInfos []*ShardInfoEx, mode codemode.CodeMode, repaireIdxs []int, forceSet map[int]struct{},
|
||||
) (repairIdx []int, shardSize int64, err error) {
|
||||
span := trace.SpanFromContextSafe(ctx)
|
||||
// step1:
|
||||
@ -204,8 +214,12 @@ func getRepairShards(ctx context.Context, shardInfos []*ShardInfoEx, mode codemo
|
||||
}
|
||||
|
||||
// step2:check miss too many shards to repair
|
||||
// exclude crc-bad idxs from the tally, else CanRecover is over-counted.
|
||||
existStatus := workutils.NewBidExistStatus(mode)
|
||||
for _, shard := range shardInfos {
|
||||
for idx, shard := range shardInfos {
|
||||
if _, force := forceSet[idx]; force {
|
||||
continue
|
||||
}
|
||||
if shard.Normal() {
|
||||
shardSize = shard.ShardSize()
|
||||
existStatus.Exist(shard.info.Vuid.Index())
|
||||
@ -223,7 +237,8 @@ func getRepairShards(ctx context.Context, shardInfos []*ShardInfoEx, mode codemo
|
||||
if !contains(idx, repaireIdxs) {
|
||||
continue
|
||||
}
|
||||
if shard.NotExist() {
|
||||
// missing shard, or crc-bad shard to rebuild in place.
|
||||
if _, force := forceSet[idx]; shard.NotExist() || force {
|
||||
shouldRepairIdx = append(shouldRepairIdx, idx)
|
||||
}
|
||||
}
|
||||
@ -302,3 +317,12 @@ func sliceUint8ToInt(s []uint8) []int {
|
||||
}
|
||||
return ret
|
||||
}
|
||||
|
||||
// uint8SliceToIntSet builds an int set from a uint8 slice.
|
||||
func uint8SliceToIntSet(s []uint8) map[int]struct{} {
|
||||
set := make(map[int]struct{}, len(s))
|
||||
for _, e := range s {
|
||||
set[int(e)] = struct{}{}
|
||||
}
|
||||
return set
|
||||
}
|
||||
|
||||
@ -74,7 +74,7 @@ func getRepairShardsTest(ctx context.Context,
|
||||
|
||||
}
|
||||
// get repair shards:
|
||||
return getRepairShards(ctx, shardInfos, mode, badIdxs)
|
||||
return getRepairShards(ctx, shardInfos, mode, badIdxs, nil)
|
||||
}
|
||||
|
||||
func testGetRepairShards(t *testing.T, mode codemode.CodeMode) {
|
||||
@ -154,6 +154,112 @@ func testGetRepairShards(t *testing.T, mode codemode.CodeMode) {
|
||||
require.Equal(t, 0, len(shouldRepair))
|
||||
}
|
||||
|
||||
func TestUint8SliceToIntSet(t *testing.T) {
|
||||
require.Equal(t, 0, len(uint8SliceToIntSet(nil)))
|
||||
set := uint8SliceToIntSet([]uint8{0, 2, 2})
|
||||
require.Equal(t, 2, len(set))
|
||||
_, ok := set[0]
|
||||
require.True(t, ok)
|
||||
_, ok = set[2]
|
||||
require.True(t, ok)
|
||||
_, ok = set[1]
|
||||
require.False(t, ok)
|
||||
}
|
||||
|
||||
func TestHasRepairedWithForceSet(t *testing.T) {
|
||||
mode := codemode.EC6P6
|
||||
replicas := genMockVol(1, mode)
|
||||
bids := []proto.BlobID{1}
|
||||
sizes := []int64{1025}
|
||||
getter := NewMockGetterWithBids(replicas, mode, bids, sizes)
|
||||
repairer := NewShardRepairer(getter)
|
||||
|
||||
shardInfos := repairer.listShardsInfo(context.Background(), replicas, 1)
|
||||
|
||||
// idx 0 is Normal: without force it is treated as already repaired.
|
||||
repaired, err := hasRepaired(shardInfos, []int{0}, nil)
|
||||
require.NoError(t, err)
|
||||
require.True(t, repaired)
|
||||
|
||||
// with force set, a Normal-but-corrupt idx must NOT be treated as repaired.
|
||||
repaired, err = hasRepaired(shardInfos, []int{0}, map[int]struct{}{0: {}})
|
||||
require.NoError(t, err)
|
||||
require.False(t, repaired)
|
||||
}
|
||||
|
||||
func TestGetRepairShardsForceCrc(t *testing.T) {
|
||||
mode := codemode.EC6P6
|
||||
replicas := genMockVol(1, mode)
|
||||
bids := []proto.BlobID{1}
|
||||
sizes := []int64{1025}
|
||||
getter := NewMockGetterWithBids(replicas, mode, bids, sizes)
|
||||
repairer := NewShardRepairer(getter)
|
||||
|
||||
// idx 0 still Normal (data corrupt) but flagged as crc-bad -> must be repaired.
|
||||
shardInfos := repairer.listShardsInfo(context.Background(), replicas, 1)
|
||||
forceSet := map[int]struct{}{0: {}}
|
||||
shouldRepair, shardSize, err := getRepairShards(context.Background(), shardInfos, mode, []int{0}, forceSet)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, int64(1025), shardSize)
|
||||
require.Equal(t, true, repairIdxsEqual(shouldRepair, []int{0}))
|
||||
|
||||
// CanRecover must exclude crc-bad idxs: with N-data shards corrupt+forced and
|
||||
// the rest deleted, the blob is unrecoverable and an error is expected.
|
||||
codeInfo := mode.Tactic()
|
||||
getter2 := NewMockGetterWithBids(replicas, mode, bids, sizes)
|
||||
repairer2 := NewShardRepairer(getter2)
|
||||
var forced []int
|
||||
forceSet2 := map[int]struct{}{}
|
||||
for i := 0; i < codeInfo.M+1; i++ {
|
||||
forced = append(forced, i)
|
||||
forceSet2[i] = struct{}{}
|
||||
}
|
||||
shardInfos2 := repairer2.listShardsInfo(context.Background(), replicas, 1)
|
||||
_, _, err = getRepairShards(context.Background(), shardInfos2, mode, forced, forceSet2)
|
||||
require.Error(t, err)
|
||||
require.EqualError(t, errcode.ErrShardMayBeLost, err.Error())
|
||||
}
|
||||
|
||||
func TestRepairShardCrcInPlace(t *testing.T) {
|
||||
testWithAllMode(t, testRepairShardCrcInPlace)
|
||||
}
|
||||
|
||||
func testRepairShardCrcInPlace(t *testing.T, mode codemode.CodeMode) {
|
||||
replicas := genMockVol(1, mode)
|
||||
bids := []proto.BlobID{1}
|
||||
sizes := []int64{1025}
|
||||
|
||||
workutils.TaskBufPool = workutils.NewBufPool(&workutils.BufConfig{
|
||||
MigrateBufSize: 4 * 1024,
|
||||
MigrateBufCapacity: 100,
|
||||
RepairBufSize: 2 * 1024,
|
||||
RepairBufCapacity: 100,
|
||||
})
|
||||
|
||||
// origin crc of idx 0
|
||||
originGetter := NewMockGetterWithBids(replicas, mode, bids, sizes)
|
||||
originCrc := originGetter.getShardCrc32(replicas[0].Vuid, 1)
|
||||
|
||||
// case 1: crc-bad idx 0 is Normal but corrupt; WITHOUT the inspect-crc reason
|
||||
// it is a no-op (legacy behaviour preserved), corruption stays.
|
||||
getter := NewMockGetterWithBids(replicas, mode, bids, sizes)
|
||||
getter.Corrupt(context.Background(), replicas[0].Vuid, 1)
|
||||
corruptCrc := getter.getShardCrc32(replicas[0].Vuid, 1)
|
||||
require.NotEqual(t, originCrc, corruptCrc)
|
||||
repairer := NewShardRepairer(getter)
|
||||
legacyTask := &proto.ShardRepairTask{Bid: 1, CodeMode: mode, Sources: replicas, BadIdxs: []uint8{0}}
|
||||
require.NoError(t, repairer.RepairShard(context.Background(), legacyTask))
|
||||
require.Equal(t, corruptCrc, getter.getShardCrc32(replicas[0].Vuid, 1)) // unchanged
|
||||
|
||||
// case 2: same corruption but WITH the inspect-crc reason -> rebuilt in place.
|
||||
crcTask := &proto.ShardRepairTask{
|
||||
Bid: 1, CodeMode: mode, Sources: replicas,
|
||||
BadIdxs: []uint8{0}, Reason: proto.ShardRepairReasonInspectCrc,
|
||||
}
|
||||
require.NoError(t, repairer.RepairShard(context.Background(), crcTask))
|
||||
require.Equal(t, originCrc, getter.getShardCrc32(replicas[0].Vuid, 1)) // restored
|
||||
}
|
||||
|
||||
func repairIdxsEqual(idxs1, idxs2 []int) bool {
|
||||
if len(idxs1) != len(idxs2) {
|
||||
return false
|
||||
|
||||
@ -80,6 +80,11 @@ func (msg *DeleteMsg) SetDeleteStage(stage BlobDeleteStage) {
|
||||
}
|
||||
}
|
||||
|
||||
// ShardRepairReasonInspectCrc marks a shard repair triggered by data-inspect
|
||||
// crc corruption. The consumer must rebuild every BadIdx in place even if their
|
||||
// meta still reports Normal.
|
||||
const ShardRepairReasonInspectCrc = "inspect-crc"
|
||||
|
||||
type ShardRepairMsg struct {
|
||||
ClusterID ClusterID `json:"cluster_id"`
|
||||
Bid BlobID `json:"bid"`
|
||||
|
||||
@ -26,10 +26,14 @@ func TestShardRepairMsg_IsValid(t *testing.T) {
|
||||
msg ShardRepairMsg
|
||||
ok bool
|
||||
}{
|
||||
{ShardRepairMsg{1, 1, 1, []uint8{1}, 0, "access", ""}, true},
|
||||
{ShardRepairMsg{1, 0, 1, []uint8{1}, 0, "access", ""}, false},
|
||||
{ShardRepairMsg{1, 1, 0, []uint8{1}, 0, "access", ""}, false},
|
||||
{ShardRepairMsg{1, 1, 1, []uint8{}, 0, "access", ""}, false},
|
||||
{ShardRepairMsg{ClusterID: 1, Bid: 1, Vid: 1, BadIdx: []uint8{1}, Reason: "access"}, true},
|
||||
{ShardRepairMsg{ClusterID: 1, Bid: 0, Vid: 1, BadIdx: []uint8{1}, Reason: "access"}, false},
|
||||
{ShardRepairMsg{ClusterID: 1, Bid: 1, Vid: 0, BadIdx: []uint8{1}, Reason: "access"}, false},
|
||||
{ShardRepairMsg{ClusterID: 1, Bid: 1, Vid: 1, BadIdx: []uint8{}, Reason: "access"}, false},
|
||||
// crc repair: identified by reason, validity still requires a BadIdx.
|
||||
{ShardRepairMsg{ClusterID: 1, Bid: 1, Vid: 1, BadIdx: []uint8{1}, Reason: ShardRepairReasonInspectCrc}, true},
|
||||
// crc repair without any BadIdx is invalid.
|
||||
{ShardRepairMsg{ClusterID: 1, Bid: 1, Vid: 1, Reason: ShardRepairReasonInspectCrc}, false},
|
||||
} {
|
||||
require.Equal(t, cs.ok, cs.msg.IsValid())
|
||||
}
|
||||
|
||||
Loading…
Reference in New Issue
Block a user
