From cb6818c4d09083aaf1c3cdbafc3c2b8ebc5ccfc0 Mon Sep 17 00:00:00 2001 From: mawei029 Date: Mon, 9 Jun 2025 15:35:00 +0800 Subject: [PATCH] refactor(blobnode): data inspect record local incorrect bids log with: #1000165799 Signed-off-by: mawei029 --- blobstore/blobnode/chunkreport_test.go | 2 + blobstore/blobnode/config.go | 12 ++-- blobstore/blobnode/datainspect.go | 87 +++++++++++++++++++++----- blobstore/blobnode/datainspect_test.go | 70 ++++++++++++++------- blobstore/blobnode/heartbeat_test.go | 3 + blobstore/blobnode/shard.go | 2 +- blobstore/blobnode/svr_test.go | 3 + 7 files changed, 134 insertions(+), 45 deletions(-) diff --git a/blobstore/blobnode/chunkreport_test.go b/blobstore/blobnode/chunkreport_test.go index 19cfa848f..c9282fa29 100644 --- a/blobstore/blobnode/chunkreport_test.go +++ b/blobstore/blobnode/chunkreport_test.go @@ -35,6 +35,7 @@ import ( "github.com/cubefs/cubefs/blobstore/blobnode/core" "github.com/cubefs/cubefs/blobstore/blobnode/db" "github.com/cubefs/cubefs/blobstore/common/proto" + "github.com/cubefs/cubefs/blobstore/common/recordlog" ) func TestChunkReport(t *testing.T) { @@ -158,6 +159,7 @@ func TestChunkReport2(t *testing.T) { Clustermgr: cc, HeartbeatIntervalSec: 600, ChunkReportIntervalSec: 1, + InspectConf: DataInspectConf{Record: recordlog.Config{Dir: filepath.Join(workDir, "inspect")}}, } service, err := NewService(conf) require.NoError(t, err) diff --git a/blobstore/blobnode/config.go b/blobstore/blobnode/config.go index a8a9f0503..358f42593 100644 --- a/blobstore/blobnode/config.go +++ b/blobstore/blobnode/config.go @@ -39,12 +39,14 @@ const ( DefaultChunkReportIntervalSec = 60 // 1 min DefaultCleanExpiredStatIntervalSec = 60 * 60 // 60 min DefaultChunkGcIntervalSec = 30 * 60 // 30 min - DefaultChunkInspectIntervalSec = 24 * 60 * 60 // 24 hour DefaultChunkProtectionPeriodSec = 48 * 60 * 60 // 48 hour DefaultDiskStatusCheckIntervalSec = 2 * 60 // 2 min DefaultDeleteQpsLimitPerDisk = 128 - DefaultInspectRate = 4 * 1024 * 1024 // rate limit 4MB per second + + defaultInspectIntervalSec = 24 * 60 * 60 // 24 hour + defaultInspectRate = 4 * 1024 * 1024 // rate limit 4MB per second + defaultInspectLogChunkSize = uint(29) ) var ( @@ -109,8 +111,10 @@ func configInit(config *Config) { if config.DeleteQpsLimitPerDisk <= 0 { config.DeleteQpsLimitPerDisk = DefaultDeleteQpsLimitPerDisk } - defaulter.LessOrEqual(&config.InspectConf.IntervalSec, DefaultChunkInspectIntervalSec) - defaulter.LessOrEqual(&config.InspectConf.RateLimit, DefaultInspectRate) + defaulter.LessOrEqual(&config.InspectConf.IntervalSec, defaultInspectIntervalSec) + defaulter.LessOrEqual(&config.InspectConf.RateLimit, defaultInspectRate) + defaulter.LessOrEqual(&config.InspectConf.Record.ChunkBits, defaultInspectLogChunkSize) + defaulter.LessOrEqual(&config.HostInfo.DiskType, proto.DiskTypeHDD) } diff --git a/blobstore/blobnode/datainspect.go b/blobstore/blobnode/datainspect.go index 4e7287405..4a004da23 100644 --- a/blobstore/blobnode/datainspect.go +++ b/blobstore/blobnode/datainspect.go @@ -18,6 +18,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/common/trace" @@ -38,7 +39,7 @@ var ( Name: "data_inspect", Help: "blobnode data inspect", }, - []string{"cluster_id", "idc", "rack", "host", "disk_id", "vuid", "bid", "error"}, + []string{"cluster_id", "idc", "rack", "host", "disk_id"}, ) errServiceClosed = errors.New("service is closed") ) @@ -46,6 +47,8 @@ var ( type DataInspectConf struct { IntervalSec int `json:"interval_sec"` // next round inspect interval RateLimit int `json:"rate_limit"` // max rate limit per second + + Record recordlog.Config `json:"record"` } type DataInspectStat struct { @@ -59,6 +62,8 @@ type DataInspectMgr struct { svr *Service taskSwitch *taskswitch.TaskSwitch + + recorder recordlog.Encoder } func NewDataInspectMgr(svr *Service, conf DataInspectConf, switchMgr *taskswitch.SwitchMgr) (*DataInspectMgr, error) { @@ -67,11 +72,18 @@ func NewDataInspectMgr(svr *Service, conf DataInspectConf, switchMgr *taskswitch return nil, err } + // init data inspect record + recorder, err := recordlog.NewEncoder(&conf.Record) + if err != nil { + return nil, err + } + mgr := &DataInspectMgr{ conf: conf, limits: make(map[proto.DiskID]*rate.Limiter), svr: svr, taskSwitch: taskSwitch, + recorder: recorder, } return mgr, nil } @@ -90,6 +102,7 @@ func (mgr *DataInspectMgr) loopDataInspect() { mgr.inspectAllDisks(ctx) case <-mgr.svr.closeCh: + mgr.recorder.Close() span.Warnf("loop inspect data closed.") return } @@ -125,6 +138,9 @@ func (mgr *DataInspectMgr) inspectDisk(ctx context.Context, ds core.DiskAPI, wg defer wg.Done() span := trace.SpanFromContextSafe(ctx) + // clean metric + mgr.cleanDiskInspectMetric(ds) + chunks, err := ds.ListChunks(ctx) if err != nil { span.Errorf("ListChunks error:%v", err) @@ -220,7 +236,7 @@ func (mgr *DataInspectMgr) inspectChunk(pCtx context.Context, cs core.ChunkAPI) } err := mgr.ScanShard(ctx, cs, startBid, scanFn) - mgr.reportBatchBadShards(cs, badShards) + mgr.reportBatchBadShards(ctx, cs, badShards) if err != nil { return nil, err } @@ -261,8 +277,18 @@ func (mgr *DataInspectMgr) waitNextRoundInspect() { } } +func (mgr *DataInspectMgr) cleanDiskInspectMetric(ds core.DiskAPI) { + diskInfo := ds.DiskInfo() + dataInspectMetric.WithLabelValues(diskInfo.ClusterID.ToString(), + diskInfo.Idc, + diskInfo.Rack, + diskInfo.Host, + diskInfo.DiskID.ToString(), + ).Set(0) +} + // It was reported only once. When the upper-level user at get/put, an error was found -func (mgr *DataInspectMgr) reportBadShard(cs core.ChunkAPI, blobID proto.BlobID, err error) { +func (mgr *DataInspectMgr) reportBadShard(ctx context.Context, cs core.ChunkAPI, blobID proto.BlobID, err error) { if isInspectReportIgnoredError(err) { return } @@ -274,13 +300,13 @@ func (mgr *DataInspectMgr) reportBadShard(cs core.ChunkAPI, blobID proto.BlobID, diskInfo.Rack, diskInfo.Host, diskInfo.DiskID.ToString(), - cs.Vuid().ToString(), - bid, - err.Error()).Set(1) + ).Set(1) + + mgr.recordBadBids(ctx, cs, []string{bid}, err.Error()) } // Aggregate a batch of errors and report them all at once(the same chunk), Because the repair of data is often at the granularity of chunks -func (mgr *DataInspectMgr) reportBatchBadShards(cs core.ChunkAPI, items []bnapi.BadShard) int { +func (mgr *DataInspectMgr) reportBatchBadShards(ctx context.Context, cs core.ChunkAPI, items []bnapi.BadShard) int { if len(items) == 0 { return 0 } @@ -303,23 +329,52 @@ func (mgr *DataInspectMgr) reportBatchBadShards(cs core.ChunkAPI, items []bnapi. return 0 } + // record local log totalBadBid := 0 - diskInfo := cs.Disk().DiskInfo() for errStr, bids := range uniqueErr { - dataInspectMetric.WithLabelValues(diskInfo.ClusterID.ToString(), - diskInfo.Idc, - diskInfo.Rack, - diskInfo.Host, - diskInfo.DiskID.ToString(), - cs.Vuid().ToString(), - strings.Join(bids, ","), - errStr).Set(float64(len(bids))) + mgr.recordBadBids(ctx, cs, bids, errStr) totalBadBid += len(bids) } + // report metric + diskInfo := cs.Disk().DiskInfo() + dataInspectMetric.WithLabelValues(diskInfo.ClusterID.ToString(), + diskInfo.Idc, + diskInfo.Rack, + diskInfo.Host, + diskInfo.DiskID.ToString(), + ).Set(float64(totalBadBid)) + return totalBadBid } +type badBidRecord struct { + ClusterID proto.ClusterID `json:"cluster_id"` + DiskID proto.DiskID `json:"disk_id"` + Vuid proto.Vuid `json:"vuid"` + Timestamp int64 `json:"ts"` + Bids string `json:"bids"` + Reason string `json:"reason"` +} + +func (mgr *DataInspectMgr) recordBadBids(ctx context.Context, cs core.ChunkAPI, bids []string, errStr string) { + span := trace.SpanFromContextSafe(ctx) + + // record local log + diskInfo := cs.Disk().DiskInfo() + record := badBidRecord{ + ClusterID: diskInfo.ClusterID, + DiskID: diskInfo.DiskID, + Vuid: cs.Vuid(), + Bids: strings.Join(bids, ","), + Timestamp: time.Now().Unix(), + Reason: errStr, + } + if err := mgr.recorder.Encode(record); err != nil { + span.Warnf("fail to write bad data inspect record: record[%v], err[%+v]", record, err) + } +} + func (mgr *DataInspectMgr) setLimiters(disks []core.DiskAPI) { for _, ds := range disks { if _, ok := mgr.limits[ds.ID()]; !ok { diff --git a/blobstore/blobnode/datainspect_test.go b/blobstore/blobnode/datainspect_test.go index 259bf9f38..2838972a0 100644 --- a/blobstore/blobnode/datainspect_test.go +++ b/blobstore/blobnode/datainspect_test.go @@ -9,9 +9,9 @@ import ( "sync" "testing" - "github.com/cubefs/cubefs/util/errors" "github.com/golang/mock/gomock" "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" @@ -21,8 +21,32 @@ import ( "github.com/cubefs/cubefs/blobstore/common/rpc" "github.com/cubefs/cubefs/blobstore/common/taskswitch" "github.com/cubefs/cubefs/blobstore/testing/mocks" + "github.com/cubefs/cubefs/util/errors" ) +func newDataInspectMgr(t *testing.T, conf DataInspectConf, svr *Service) *DataInspectMgr { + ctr := gomock.NewController(t) + + 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) + taskSwitch, err := switchMgr.AddSwitch(proto.TaskSwitchDataInspect.String()) + require.NoError(t, err) + + // init data inspect record + recorder := mocks.NewMockRecordLogEncoder(ctr) + + mgr := &DataInspectMgr{ + conf: conf, + limits: make(map[proto.DiskID]*rate.Limiter), + svr: svr, + taskSwitch: taskSwitch, + recorder: recorder, + } + return mgr +} + func TestDataInspect(t *testing.T) { ctr := gomock.NewController(t) ctx := context.Background() @@ -33,15 +57,11 @@ func TestDataInspect(t *testing.T) { ctx: context.Background(), closeCh: make(chan struct{}), } - cfg := DataInspectConf{IntervalSec: 100, RateLimit: 2} - 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, cfg, switchMgr) + var err error + cfg := DataInspectConf{IntervalSec: 100, RateLimit: 2} + mgr := newDataInspectMgr(t, cfg, svr) svr.inspectMgr = mgr - require.NoError(t, err) require.Equal(t, cfg.IntervalSec, mgr.conf.IntervalSec) ds1.EXPECT().IsWritable().AnyTimes().Return(true) @@ -50,8 +70,10 @@ func TestDataInspect(t *testing.T) { { ds1.EXPECT().ID().Times(2).Return(proto.DiskID(11)) ds1.EXPECT().ListChunks(any).Return(nil, errMock) + ds1.EXPECT().DiskInfo().Times(1) ds2.EXPECT().ID().Times(2).Return(proto.DiskID(22)) ds2.EXPECT().ListChunks(any).Return(nil, errMock) + ds2.EXPECT().DiskInfo().Times(1) // close(svr.closeCh) mgr.inspectAllDisks(ctx) @@ -72,6 +94,7 @@ func TestDataInspect(t *testing.T) { ds1.EXPECT().ID().Times(1).Return(proto.DiskID(11)) ds1.EXPECT().ListChunks(any).Return([]core.VuidMeta{{Vuid: proto.Vuid(1001)}}, nil) ds1.EXPECT().GetChunkStorage(any).Return(cs, true) + ds1.EXPECT().DiskInfo().Return(clustermgr.BlobNodeDiskInfo{}).Times(1) mgr.inspectDisk(ctx, ds1, &wg) } @@ -150,10 +173,11 @@ func TestDataInspect(t *testing.T) { ds1.EXPECT().ID().Return(proto.DiskID(11)).Times(1) // bad bid report metric - cs.EXPECT().Disk().Return(ds1).Times(1) + cs.EXPECT().Disk().Return(ds1).Times(1 + 1) // cs.EXPECT().Vuid().Return(proto.Vuid(1001)) - ds1.EXPECT().DiskInfo().Return(clustermgr.BlobNodeDiskInfo{}).Times(1) - ds1.EXPECT().ID().Return(proto.DiskID(11)).Times(2) + ds1.EXPECT().DiskInfo().Return(clustermgr.BlobNodeDiskInfo{}).Times(1 + 1) + ds1.EXPECT().ID().Return(proto.DiskID(11)).Times(1) + mgr.recorder.(*mocks.MockRecordLogEncoder).EXPECT().Encode(any).Times(1) mgr.limits[proto.DiskID(11)].SetLimit(4) mgr.limits[proto.DiskID(11)].SetBurst(8) @@ -183,7 +207,7 @@ func TestDataInspect(t *testing.T) { cs.EXPECT().Disk().Return(ds1) cs.EXPECT().Read(any, any).Return(int64(0), bloberr.ErrNoSuchBid) cs.EXPECT().ListShards(any, any, any, any).Return([]*bnapi.ShardInfo{{Bid: 123456, Size: 8}}, proto.BlobID(123456+1), nil) - ds1.EXPECT().ID().Return(proto.DiskID(11)).Times(1) + ds1.EXPECT().ID().Return(proto.DiskID(11)).Times(2) mgr.limits[proto.DiskID(11)].SetLimit(4) mgr.limits[proto.DiskID(11)].SetBurst(8) @@ -193,9 +217,12 @@ func TestDataInspect(t *testing.T) { } close(svr.closeCh) + mgr.recorder.(*mocks.MockRecordLogEncoder).EXPECT().Close().Times(1) + mgr.loopDataInspect() } func TestDataInspectMetric(t *testing.T) { + ctx := context.Background() ctr := gomock.NewController(t) ds1 := NewMockDiskAPI(ctr) svr := &Service{ @@ -204,15 +231,9 @@ func TestDataInspectMetric(t *testing.T) { 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) cfg := DataInspectConf{IntervalSec: 100, RateLimit: 2} - - mgr, err := NewDataInspectMgr(svr, cfg, switchMgr) + mgr := newDataInspectMgr(t, cfg, svr) svr.inspectMgr = mgr - require.NoError(t, err) defer close(svr.closeCh) // no bad blob @@ -227,7 +248,7 @@ func TestDataInspectMetric(t *testing.T) { } } - badBidCnt := mgr.reportBatchBadShards(cs, bads) + badBidCnt := mgr.reportBatchBadShards(ctx, cs, bads) require.Equal(t, 0, badBidCnt) // some bad blob @@ -255,8 +276,6 @@ func TestDataInspectMetric(t *testing.T) { } } - cs.EXPECT().Disk().Return(ds1) - cs.EXPECT().Vuid().Return(proto.Vuid(1001)).Times(2) ds1.EXPECT().DiskInfo().Return(clustermgr.BlobNodeDiskInfo{ DiskInfo: clustermgr.DiskInfo{ ClusterID: 1, @@ -266,9 +285,12 @@ func TestDataInspectMetric(t *testing.T) { Path: "", }, DiskHeartBeatInfo: clustermgr.DiskHeartBeatInfo{DiskID: 11}, - }) + }).Times(1 + 2) + cs.EXPECT().Disk().Return(ds1).Times(1 + 2) + cs.EXPECT().Vuid().Return(proto.Vuid(1001)).Times(2) + mgr.recorder.(*mocks.MockRecordLogEncoder).EXPECT().Encode(any).Times(2) - badBidCnt = mgr.reportBatchBadShards(cs, bads) + badBidCnt = mgr.reportBatchBadShards(ctx, cs, bads) require.Equal(t, expectCnt, badBidCnt) } diff --git a/blobstore/blobnode/heartbeat_test.go b/blobstore/blobnode/heartbeat_test.go index b20698f47..2b2c75da1 100644 --- a/blobstore/blobnode/heartbeat_test.go +++ b/blobstore/blobnode/heartbeat_test.go @@ -33,6 +33,7 @@ import ( "github.com/cubefs/cubefs/blobstore/blobnode/core" "github.com/cubefs/cubefs/blobstore/blobnode/db" "github.com/cubefs/cubefs/blobstore/common/proto" + "github.com/cubefs/cubefs/blobstore/common/recordlog" ) func TestHeartbeat(t *testing.T) { @@ -100,6 +101,7 @@ func TestHeartbeat2(t *testing.T) { }, Clustermgr: cc, HeartbeatIntervalSec: 1, + InspectConf: DataInspectConf{Record: recordlog.Config{Dir: filepath.Join(workDir, "inspect")}}, } service, err := NewService(conf) require.NoError(t, err) @@ -173,6 +175,7 @@ func TestHeartbeat3(t *testing.T) { }, Clustermgr: cc, HeartbeatIntervalSec: 600, + InspectConf: DataInspectConf{Record: recordlog.Config{Dir: filepath.Join(workDir, "inspect")}}, } service, err := NewService(conf) require.NoError(t, err) diff --git a/blobstore/blobnode/shard.go b/blobstore/blobnode/shard.go index fdfd5f238..7b52945ea 100644 --- a/blobstore/blobnode/shard.go +++ b/blobstore/blobnode/shard.go @@ -154,7 +154,7 @@ func (s *Service) ShardGet(c *rpc.Context) { if err != nil { span.Errorf("Failed read. args:%v err:%v, written:%v", args, err, written) if isShardErr(err) { - s.inspectMgr.reportBadShard(cs, args.Bid, err) + s.inspectMgr.reportBadShard(ctx, cs, args.Bid, err) } if !wroteHeader { err = handlerBidNotFoundErr(err) diff --git a/blobstore/blobnode/svr_test.go b/blobstore/blobnode/svr_test.go index 55a39b659..0f502cdcd 100644 --- a/blobstore/blobnode/svr_test.go +++ b/blobstore/blobnode/svr_test.go @@ -41,6 +41,7 @@ import ( "github.com/cubefs/cubefs/blobstore/blobnode/db" 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/trace" "github.com/cubefs/cubefs/blobstore/util/errors" @@ -329,6 +330,7 @@ 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) @@ -407,6 +409,7 @@ 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