gian/cubefs
1
0
Fork 0
mirror of https://github.com/cubefs/cubefs.git synced 2026-08-02 02:00:56 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

fix(blobnode): fix data inspect concurrent write or read map

Browse Source 
with: #1000287722

Signed-off-by: mawei029 <mawei2@oppo.com>
This commit is contained in:
mawei029 2025-08-15 14:54:28 +08:00 committed by slasher
parent 3ae408e632
commit ed90619ed6
3 changed files with 133 additions and 129 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

85
blobstore/blobnode/datainspect.go
View File

@ -63,10 +63,9 @@ type DataInspectMgr struct {
svr *Service
taskSwitch *taskswitch.TaskSwitch
round uint64 // round of data inspect
recorder recordlog.Encoder // local record log
inspected map[proto.DiskID]*sync.Map // inspected bad blobs
progress map[proto.DiskID]int // progress of data inspect
round uint64 // round of data inspect
progress sync.Map // progress of data inspect: diskID -> progress[0, 100]
recorder recordlog.Encoder // local record log
}
func NewDataInspectMgr(svr *Service, conf DataInspectConf, switchMgr *taskswitch.SwitchMgr) (*DataInspectMgr, error) {
@ -91,8 +90,6 @@ func NewDataInspectMgr(svr *Service, conf DataInspectConf, switchMgr *taskswitch
svr: svr,
taskSwitch: taskSwitch,
recorder: recorder,
inspected: make(map[proto.DiskID]*sync.Map),
progress: make(map[proto.DiskID]int),
}
return mgr, nil
}
@ -123,13 +120,13 @@ func (mgr *DataInspectMgr) inspectAllDisks(ctx context.Context) {
disks := mgr.svr.copyDiskStorages(ctx)
mgr.setLimiters(disks)
mgr.prepareDiskInspectionState(disks)
mgr.recordInspectStartPoint(ctx)
defer func() { mgr.round++ }()
defer mgr.roundIncrease()
var wg sync.WaitGroup
for _, ds := range disks {
mgr.progress[ds.ID()] = 0 // set progress to 0%, will start work inspect
if !ds.IsWritable() { // not normal disk, skip it.
if !ds.IsWritable() { // not normal disk, skip it.
continue
}
wg.Add(1)
@ -142,9 +139,11 @@ func (mgr *DataInspectMgr) inspectAllDisks(ctx context.Context) {
return
}
for id := range mgr.progress {
mgr.progress[id] = 100 // set progress to 100%, work inspect done
}
// set progress to 100%, work inspect done
mgr.progress.Range(func(key, value interface{}) bool {
mgr.progress.Store(key.(proto.DiskID), 100)
return true
})
span.Warn("finish to inspect all disks.")
mgr.waitNextRoundInspect()
}
@ -169,7 +168,7 @@ func (mgr *DataInspectMgr) inspectDisk(ds core.DiskAPI, wg *sync.WaitGroup) {
if (step != 0 && (i+1)%step == 0) || i == len(chunks)-1 {
span.Warnf("chunk inspcet progress: %d / %d", i+1, len(chunks))
}
mgr.progress[ds.ID()] = 100 * (i + 1) / len(chunks)
mgr.progress.Store(ds.ID(), 100*(i+1)/len(chunks))
if chunk.Status == clustermgr.ChunkStatusRelease {
continue
@ -198,36 +197,32 @@ func (mgr *DataInspectMgr) inspectChunk(pCtx context.Context, cs core.ChunkAPI)
span := trace.SpanFromContextSafe(pCtx)
ctx, cancel := context.WithCancel(context.Background())
span, ctx = trace.StartSpanFromContextWithTraceID(ctx, "", span.TraceID())
span.Debugf("start to inspect chunk vuid:%d, chunkid:%s.", cs.Vuid(), cs.ID())
span.Debugf("start to inspect chunk vuid:%d, chunk:%s.", cs.Vuid(), cs.ID())
ctx = bnapi.SetIoType(ctx, bnapi.BackgroundIO)
ds := cs.Disk()
total := 0
badShards := make([]bnapi.BadShard, 0)
inspected := mgr.getInspectedBlobs(ds.ID())
scanFn := func(batchShards []*bnapi.ShardInfo) (err error) {
total += len(batchShards)
for _, si := range batchShards {
_, exist := inspected.Load(si.Bid)
if si.Size <= 0 || exist {
if si.Size <= 0 {
continue
}
select {
case <-pCtx.Done():
span.Warnf("inspect chunk stop, upper level has context canceled. vuid:%d,chunkid:%s.", cs.Vuid(), cs.ID())
span.Warnf("inspect chunk stop, upper context canceled. vuid:%d, chunk:%s.", cs.Vuid(), cs.ID())
return pCtx.Err()
case <-mgr.svr.closeCh:
cancel()
span.Warnf("inspect chunk stop, service is closed. vuid:%d, chunkid:%s.", cs.Vuid(), cs.ID())
span.Warnf("inspect chunk stop, service is closed. vuid:%d, chunk:%s.", cs.Vuid(), cs.ID())
return errServiceClosed
default:
}
if err = mgr.inspectShard(ctx, cs, si, mgr.getLimiter(ds)); err != nil {
// add bad bids of chunk, to this disk. prevent next round will report these bids, it repeated
inspected.Store(si.Bid, struct{}{})
badShards = append(badShards, bnapi.BadShard{DiskID: ds.ID(), Vuid: si.Vuid, Bid: si.Bid, Err: err})
if base.IsEIO(err) {
return err
@ -239,7 +234,7 @@ func (mgr *DataInspectMgr) inspectChunk(pCtx context.Context, cs core.ChunkAPI)
err := mgr.scanShards(ctx, cs, scanFn)
mgr.reportBatchBadShards(ctx, cs, badShards)
span.Infof("finish to inspect chunk, vuid:%d, chunkid:%s, total:%d, wrong:%d, err:%+v",
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
}
@ -297,15 +292,6 @@ func (mgr *DataInspectMgr) waitNextRoundInspect() {
}
}
func (mgr *DataInspectMgr) getInspectedBlobs(diskID proto.DiskID) *sync.Map {
if inspected, exists := mgr.inspected[diskID]; exists {
return inspected
}
inspected := &sync.Map{}
mgr.inspected[diskID] = inspected
return inspected
}
func (mgr *DataInspectMgr) cleanDiskInspectMetric(ds core.DiskAPI, diskID proto.DiskID) {
dataInspectMetric.WithLabelValues(
ds.DiskInfo().ClusterID.ToString(),
@ -315,15 +301,6 @@ func (mgr *DataInspectMgr) cleanDiskInspectMetric(ds core.DiskAPI, diskID proto.
// It was reported only once. When the upper-level user at get/put, an error was found
func (mgr *DataInspectMgr) reportBadShard(ctx context.Context, cs core.ChunkAPI, blobID proto.BlobID, err error) {
diskInfo := cs.Disk().DiskInfo()
// prevent repeated reporting
inspected := mgr.getInspectedBlobs(diskInfo.DiskID)
if _, exist := inspected.Load(blobID); exist {
return
}
inspected.Store(blobID, struct{}{})
// don't report this error
if isInspectReportIgnoredError(err) {
return
@ -332,6 +309,7 @@ func (mgr *DataInspectMgr) reportBadShard(ctx context.Context, cs core.ChunkAPI,
// report one bad shard, when the upper-level user at get/put, an error was found
// It's possible that this disk has inspected this bid error before, or it might not.
// 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())
dataInspectMetric.WithLabelValues(
diskInfo.ClusterID.ToString(),
@ -423,6 +401,13 @@ func (mgr *DataInspectMgr) recordInspectStartPoint(ctx context.Context) {
}
}
func (mgr *DataInspectMgr) prepareDiskInspectionState(disks []core.DiskAPI) {
for _, ds := range disks {
// init or reset progress to 0%, at the beginning of each new round of inspection
mgr.progress.Store(ds.ID(), 0)
}
}
func (mgr *DataInspectMgr) setLimiters(disks []core.DiskAPI) {
for _, ds := range disks {
if _, ok := mgr.limits[ds.ID()]; !ok {
@ -439,6 +424,10 @@ func (mgr *DataInspectMgr) getSwitch() bool {
return mgr.taskSwitch.Enabled()
}
func (mgr *DataInspectMgr) roundIncrease() {
mgr.round++
}
func (mgr *DataInspectMgr) setAllDiskRateForce(newLimit int) {
for _, lmt := range mgr.limits {
lmt.SetLimit(rate.Limit(newLimit))
@ -454,27 +443,31 @@ func (s *Service) SetInspectRate(c *rpc.Context) {
return
}
ctx := c.Request.Context()
span := trace.SpanFromContextSafe(ctx)
if args.Rate < minRateLimit {
c.RespondError(errors.New("rate value is too small"))
return
}
span := trace.SpanFromContextSafe(c.Request.Context())
span.Infof("set data inspect rate args: %+v", args)
s.inspectMgr.setAllDiskRateForce(args.Rate)
c.Respond()
}
// GetInspectStat get data inspection state: switch open, rate, progress, interval, etc.
func (s *Service) GetInspectStat(c *rpc.Context) {
ctx := c.Request.Context()
span := trace.SpanFromContextSafe(ctx)
span := trace.SpanFromContextSafe(c.Request.Context())
progress := make(map[proto.DiskID]int)
s.inspectMgr.progress.Range(func(k, v interface{}) bool {
progress[k.(proto.DiskID)] = v.(int)
return true
})
stat := DataInspectStat{
DataInspectConf: s.inspectMgr.conf,
Open: s.inspectMgr.getSwitch(),
Progress: s.inspectMgr.progress,
Progress: progress,
}
span.Infof("data inspect args: %+v", stat)
c.RespondJSON(&stat)

163
blobstore/blobnode/datainspect_test.go
View File

@ -19,6 +19,7 @@ import (
"github.com/cubefs/cubefs/blobstore/blobnode/core"
bloberr "github.com/cubefs/cubefs/blobstore/common/errors"
"github.com/cubefs/cubefs/blobstore/common/proto"
"github.com/cubefs/cubefs/blobstore/common/recordlog"
"github.com/cubefs/cubefs/blobstore/common/rpc"
"github.com/cubefs/cubefs/blobstore/common/taskswitch"
"github.com/cubefs/cubefs/blobstore/testing/mocks"
@ -139,9 +140,8 @@ func TestDataInspect(t *testing.T) {
mgr.limits[proto.DiskID(11)].SetBurst(6)
bads, err = mgr.inspectChunk(ctx, cs)
require.NotNil(t, err)
require.Equal(t, "context canceled", err.Error())
require.Equal(t, 0, len(bads))
require.ErrorIs(t, err, errServiceClosed)
require.Equal(t, 0, len(bads))
}
{
@ -173,26 +173,10 @@ func TestDataInspect(t *testing.T) {
bads, err = mgr.inspectChunk(ctx, cs)
require.NoError(t, err)
require.Equal(t, 1, len(bads))
require.Equal(t, 1, len(mgr.inspected))
// prevent repeated bid
cs.EXPECT().Disk().Return(ds1).Times(1)
cs.EXPECT().ListShards(any, any, any, any).Return([]*bnapi.ShardInfo{{Bid: 123456, Size: 8}}, proto.BlobID(123456+1), nil)
bads, err = mgr.inspectChunk(ctx, cs)
require.NoError(t, err)
require.Equal(t, 0, len(bads))
require.Equal(t, 1, len(mgr.inspected))
cnt, inspected := 0, mgr.getInspectedBlobs(11)
inspected.Range(func(key, value interface{}) bool {
cnt++
return true
})
require.Equal(t, 1, cnt)
}
{
// inspect already delete shard, file does not exist
mgr.inspected = make(map[proto.DiskID]*sync.Map)
cs := NewMockChunkAPI(ctr)
cs.EXPECT().Vuid().Return(proto.Vuid(1001)).AnyTimes()
cs.EXPECT().ID().Return(clustermgr.ChunkID{}).AnyTimes()
@ -204,10 +188,8 @@ func TestDataInspect(t *testing.T) {
bads, err = mgr.inspectChunk(ctx, cs)
require.NoError(t, err)
require.Equal(t, 1, len(bads))
require.Equal(t, 1, len(mgr.inspected))
// no such bid
mgr.inspected = make(map[proto.DiskID]*sync.Map)
cs.EXPECT().Vuid().Return(proto.Vuid(1001)).AnyTimes()
cs.EXPECT().ID().Return(clustermgr.ChunkID{}).AnyTimes()
cs.EXPECT().Disk().Return(ds1).Times(1)
@ -218,12 +200,10 @@ func TestDataInspect(t *testing.T) {
bads, err = mgr.inspectChunk(ctx, cs)
require.NoError(t, err)
require.Equal(t, 1, len(bads))
require.Equal(t, 1, len(mgr.inspected))
}
{
// scanShards EIO
mgr.inspected = make(map[proto.DiskID]*sync.Map)
cs := NewMockChunkAPI(ctr)
cs.EXPECT().Vuid().Return(proto.Vuid(1001)).AnyTimes()
cs.EXPECT().ID().Return(clustermgr.ChunkID{}).AnyTimes()
@ -239,7 +219,6 @@ func TestDataInspect(t *testing.T) {
bads, err = mgr.inspectChunk(ctx, cs)
require.ErrorIs(t, syscall.EIO, err)
require.Equal(t, 1, len(bads))
require.Equal(t, 1, len(mgr.inspected))
}
close(svr.closeCh)
@ -248,6 +227,53 @@ func TestDataInspect(t *testing.T) {
mgr.loopDataInspect()
}
func TestInspectChunk_NoGoroutineLeak(t *testing.T) {
ctr := gomock.NewController(t)
ctx := context.Background()
// build service and manager
ds := NewMockDiskAPI(ctr)
svr := &Service{
Disks: map[proto.DiskID]core.DiskAPI{11: ds},
ctx: context.Background(),
closeCh: make(chan struct{}),
}
getter := mocks.NewMockAccessor(ctr)
getter.EXPECT().GetConfig(any, any).AnyTimes().Return("", nil)
getter.EXPECT().SetConfig(any, any, any).AnyTimes().Return(nil)
switchMgr := taskswitch.NewSwitchMgr(getter)
mgr, err := NewDataInspectMgr(svr, DataInspectConf{IntervalSec: 1, RateLimit: 1024 * 1024}, switchMgr)
require.NoError(t, err)
mgr.svr = svr
svr.inspectMgr = mgr
// limiter entry (avoid nil access if shards present)
ds.EXPECT().ID().AnyTimes().Return(proto.DiskID(11))
mgr.setLimiters([]core.DiskAPI{ds})
// chunk mock: empty shard list so inspectChunk returns quickly
cs := NewMockChunkAPI(ctr)
cs.EXPECT().Vuid().AnyTimes().Return(proto.Vuid(1001))
cs.EXPECT().ID().AnyTimes().Return(clustermgr.ChunkID{})
cs.EXPECT().Disk().AnyTimes().Return(ds)
cs.EXPECT().ListShards(any, any, any, any).AnyTimes().Return([]*bnapi.ShardInfo{}, proto.InValidBlobID, nil)
before := runtime.NumGoroutine()
for i := 0; i < 50; i++ {
_, err = mgr.inspectChunk(ctx, cs)
require.NoError(t, err)
}
// allow scheduler to settle
// (if a leak existed via a background goroutine, goroutine count would keep growing)
// small sleep to stabilize
// not too long to avoid slowing CI
// 50 iterations are enough to detect growth
after := runtime.NumGoroutine()
// tolerate a small delta for unrelated goroutines
require.LessOrEqual(t, after, before+5)
}
func TestDataInspectMetric(t *testing.T) {
ctx := context.Background()
ctr := gomock.NewController(t)
@ -280,7 +306,6 @@ func TestDataInspectMetric(t *testing.T) {
require.Equal(t, 0, badBidCnt)
// some bad blob
mgr.inspected = make(map[proto.DiskID]*sync.Map)
cs = NewMockChunkAPI(ctr)
bads = make([]bnapi.BadShard, total)
err1 := errors.New("fake mock error 111")
@ -323,7 +348,6 @@ func TestDataInspectMetric(t *testing.T) {
require.Equal(t, expectCnt, badBidCnt)
// one shard bad
require.Equal(t, 0, len(mgr.inspected))
badBid := proto.BlobID(1234)
ds1.EXPECT().DiskInfo().Return(clustermgr.BlobNodeDiskInfo{
DiskHeartBeatInfo: clustermgr.DiskHeartBeatInfo{DiskID: 11},
@ -333,14 +357,6 @@ func TestDataInspectMetric(t *testing.T) {
mgr.recorder.(*mocks.MockRecordLogEncoder).EXPECT().Encode(any)
mgr.reportBadShard(ctx, cs, badBid, errMock)
require.Equal(t, 1, len(mgr.inspected))
cnt, inspected := 0, mgr.getInspectedBlobs(11)
inspected.Range(func(key, value interface{}) bool {
cnt++
return true
})
require.Equal(t, 1, cnt)
}
func TestDataInspectRecord(t *testing.T) {
@ -439,49 +455,50 @@ func TestDataInspectRecord(t *testing.T) {
}
}
func TestInspectChunk_NoGoroutineLeak(t *testing.T) {
ctr := gomock.NewController(t)
ctx := context.Background()
func TestDataInspectMgr_ConcurrentAccess(t *testing.T) {
const diskCnt = 60
const concurrentCnt = 1000
mgr := &DataInspectMgr{}
// build service and manager
ds := NewMockDiskAPI(ctr)
svr := &Service{
Disks: map[proto.DiskID]core.DiskAPI{11: ds},
ctx: context.Background(),
closeCh: make(chan struct{}),
diskIDs := make([]proto.DiskID, diskCnt)
for i := 0; i < diskCnt; i++ {
diskIDs[i] = proto.DiskID(i + 1)
// mgr.progress[diskIDs[i]] = 0
mgr.progress.Store(diskIDs[i], 0)
}
getter := mocks.NewMockAccessor(ctr)
getter.EXPECT().GetConfig(any, any).AnyTimes().Return("", nil)
getter.EXPECT().SetConfig(any, any, any).AnyTimes().Return(nil)
switchMgr := taskswitch.NewSwitchMgr(getter)
mgr, err := NewDataInspectMgr(svr, DataInspectConf{IntervalSec: 1, RateLimit: 1024 * 1024}, switchMgr)
require.NoError(t, err)
mgr.svr = svr
svr.inspectMgr = mgr
// limiter entry (avoid nil access if shards present)
ds.EXPECT().ID().AnyTimes().Return(proto.DiskID(11))
mgr.setLimiters([]core.DiskAPI{ds})
// chunk mock: empty shard list so inspectChunk returns quickly
cs := NewMockChunkAPI(ctr)
cs.EXPECT().Vuid().AnyTimes().Return(proto.Vuid(1001))
cs.EXPECT().ID().AnyTimes().Return(clustermgr.ChunkID{})
cs.EXPECT().Disk().AnyTimes().Return(ds)
cs.EXPECT().ListShards(any, any, any, any).AnyTimes().Return([]*bnapi.ShardInfo{}, proto.InValidBlobID, nil)
before := runtime.NumGoroutine()
for i := 0; i < 50; i++ {
_, err = mgr.inspectChunk(ctx, cs)
require.NoError(t, err)
var writeWg sync.WaitGroup
for _, diskID := range diskIDs {
writeWg.Add(1)
go func(did proto.DiskID) {
defer writeWg.Done()
for i := 0; i < concurrentCnt; i++ {
// mgr.progress[did] = i % 100
mgr.progress.Store(did, i%100)
}
}(diskID)
}
// allow scheduler to settle
// (if a leak existed via a background goroutine, goroutine count would keep growing)
// small sleep to stabilize
// not too long to avoid slowing CI
// 50 iterations are enough to detect growth
after := runtime.NumGoroutine()
// tolerate a small delta for unrelated goroutines
require.LessOrEqual(t, after, before+5)
var readWg sync.WaitGroup
for i := 0; i < 2; i++ {
readWg.Add(1)
go func() {
defer readWg.Done()
for j := 0; j < concurrentCnt; j++ {
progressCopy := make(map[proto.DiskID]int)
// for k, v := range mgr.progress {
// progressCopy[k] = v
// }
mgr.progress.Range(func(k, v interface{}) bool {
progressCopy[k.(proto.DiskID)] = v.(int)
return true
})
_ = len(progressCopy)
}
}()
}
// wait all done, and no panic(concurrent write map, concurrent iteration map)
writeWg.Wait()
readWg.Wait()
}

14
blobstore/blobnode/svr_test.go
View File

@ -47,7 +47,6 @@ import (
bloberr "github.com/cubefs/cubefs/blobstore/common/errors"
"github.com/cubefs/cubefs/blobstore/common/iostat"
"github.com/cubefs/cubefs/blobstore/common/proto"
"github.com/cubefs/cubefs/blobstore/common/recordlog"
"github.com/cubefs/cubefs/blobstore/common/rpc"
"github.com/cubefs/cubefs/blobstore/common/taskswitch"
"github.com/cubefs/cubefs/blobstore/common/trace"
@ -338,7 +337,6 @@ func newTestBlobNodeService(t *testing.T, path string) (*Service, *mockClusterMg
DiskConfig: core.RuntimeConfig{DiskReservedSpaceB: 1, CompactReservedSpaceB: 1},
Clustermgr: cc,
HeartbeatIntervalSec: 600,
InspectConf: DataInspectConf{Record: recordlog.Config{Dir: filepath.Join(workDir, "inspect")}},
}
if path == "iopslimit" {
ioFlowStat, _ := flow.NewIOFlowStat("default", false)
@ -429,7 +427,6 @@ func TestService_CmdpChunk(t *testing.T) {
DiskConfig: core.RuntimeConfig{DiskReservedSpaceB: 1, CompactReservedSpaceB: 1},
Clustermgr: cc,
HeartbeatIntervalSec: 600,
InspectConf: DataInspectConf{Record: recordlog.Config{Dir: filepath.Join(workDir, "inspect")}},
}
conf.DiskConfig.MustMountPoint = true
@ -1101,7 +1098,6 @@ func TestService_OnlyBlobnode(t *testing.T) {
DiskConfig: core.RuntimeConfig{DiskReservedSpaceB: 1, CompactReservedSpaceB: 1},
Clustermgr: cc,
HeartbeatIntervalSec: 600,
InspectConf: DataInspectConf{Record: recordlog.Config{Dir: filepath.Join(workDir, "inspect")}},
StartMode: proto.ServiceNameBlobNode,
}
@ -1141,7 +1137,6 @@ func TestService_OnlyBlobnode_OpenFailedEIO(t *testing.T) {
},
DiskConfig: core.RuntimeConfig{DiskReservedSpaceB: 1, CompactReservedSpaceB: 1},
HeartbeatIntervalSec: 600,
InspectConf: DataInspectConf{Record: recordlog.Config{Dir: filepath.Join(workDir, "inspect")}},
}
// open readFormat eio, report broken disk
@ -1229,8 +1224,7 @@ func TestService_OnlyBlobnode_OpenDiskNormal(t *testing.T) {
require.NoError(t, err)
conf := Config{
Disks: []core.Config{{BaseConfig: core.BaseConfig{Path: path1, AutoFormat: true, MaxChunks: 700}, MetaConfig: db.MetaConfig{}}},
InspectConf: DataInspectConf{Record: recordlog.Config{Dir: filepath.Join(workDir, "inspect")}},
Disks: []core.Config{{BaseConfig: core.BaseConfig{Path: path1, AutoFormat: true, MaxChunks: 700}, MetaConfig: db.MetaConfig{}}},
}
// open disk success
@ -1285,7 +1279,6 @@ func TestService_OnlyBlobnode_Fatal(t *testing.T) {
// {BaseConfig: core.BaseConfig{Path: path2, AutoFormat: true}, MetaConfig: db.MetaConfig{}},
// {BaseConfig: core.BaseConfig{Path: "wrongPath", AutoFormat: true}, MetaConfig: db.MetaConfig{}},
},
InspectConf: DataInspectConf{Record: recordlog.Config{Dir: filepath.Join(workDir, "inspect")}},
}
// new disk, read meta fake error
@ -1357,7 +1350,6 @@ func TestService_OnlyBlobnode_OpenOldDisk(t *testing.T) {
Disks: []core.Config{
{BaseConfig: core.BaseConfig{Path: path1, AutoFormat: true, MaxChunks: 700}, MetaConfig: db.MetaConfig{}},
},
InspectConf: DataInspectConf{Record: recordlog.Config{Dir: filepath.Join(workDir, "inspect")}},
}
// old disk, repairing, skip
@ -1438,10 +1430,12 @@ func TestService_DataInspect(t *testing.T) {
ts, err := taskswitch.NewSwitchMgr(getter).AddSwitch(proto.TaskSwitchDataInspect.String())
require.NoError(t, err)
svr.inspectMgr = &DataInspectMgr{
progress: map[proto.DiskID]int{101: 85, 202: 95},
// progress: map[proto.DiskID]int{101: 85, 202: 95},
taskSwitch: ts,
conf: DataInspectConf{RateLimit: 4096},
}
svr.inspectMgr.progress.Store(proto.DiskID(101), 88)
svr.inspectMgr.progress.Store(proto.DiskID(202), 99)
totalUrl := testServer.URL + "/inspect/stat"
resp, err := HTTPRequest(http.MethodGet, totalUrl)