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): read format json backward compatible, check crc with old version

Browse Source 
with: #1000464074

Signed-off-by: mawei029 <mawei2@oppo.com>
This commit is contained in:
mawei029 2026-01-04 09:21:35 +08:00 committed by 梁曟風
parent e122cb60f3
commit 2294b13288
10 changed files with 286 additions and 150 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
blobstore/blobnode/core/chunk/compact_test.go
View File

@ -41,6 +41,7 @@ import (
type diskMock struct {
diskID proto.DiskID
nodeID proto.NodeID
conf *core.Config
dataPath string
metaPath string
@ -53,6 +54,10 @@ func (mock *diskMock) ID() proto.DiskID {
return mock.diskID
}
func (mock *diskMock) NodeID() proto.NodeID {
return mock.nodeID
}
func (mock *diskMock) Status() (status proto.DiskStatus) {
return mock.status
}

64
blobstore/blobnode/core/disk/disk.go
View File

@ -46,15 +46,15 @@ const (
RandomIntervalS = 30
)
var StateTransitionRules = map[clustermgr.ChunkStatus][]clustermgr.ChunkStatus{
clustermgr.ChunkStatusDefault: {clustermgr.ChunkStatusNormal},
clustermgr.ChunkStatusNormal: {clustermgr.ChunkStatusNormal, clustermgr.ChunkStatusReadOnly},
clustermgr.ChunkStatusReadOnly: {clustermgr.ChunkStatusNormal, clustermgr.ChunkStatusReadOnly, clustermgr.ChunkStatusRelease},
}
var (
_chunkVer = []byte{0x1}
_diskVer = []byte{0x1}
StateTransitionRules = map[clustermgr.ChunkStatus][]clustermgr.ChunkStatus{
clustermgr.ChunkStatusDefault: {clustermgr.ChunkStatusNormal},
clustermgr.ChunkStatusNormal: {clustermgr.ChunkStatusNormal, clustermgr.ChunkStatusReadOnly},
clustermgr.ChunkStatusReadOnly: {clustermgr.ChunkStatusNormal, clustermgr.ChunkStatusReadOnly, clustermgr.ChunkStatusRelease},
}
)
type DiskStorageWrapper struct {
@ -238,6 +238,7 @@ func (dsw *DiskStorageWrapper) RestoreChunkStorage(ctx context.Context) (err err
type DiskStorage struct {
DiskID proto.DiskID
nodeID proto.NodeID
Lock sync.RWMutex
SuperBlock *SuperBlock
@ -393,6 +394,10 @@ func (ds *DiskStorage) ID() (id proto.DiskID) {
return ds.DiskID
}
func (ds *DiskStorage) NodeID() proto.NodeID {
return ds.nodeID
}
func (ds *DiskStorage) SetStatus(status proto.DiskStatus) {
ds.Lock.Lock()
ds.status = status
@ -1059,10 +1064,8 @@ func newDiskStorage(ctx context.Context, conf core.Config) (ds *DiskStorage, err
}
// TODO: support NodeID: old verstion -> v1.5.2: we will remove these code in the next version
if formatInfo.NodeID == 0 || dm.NodeID == 0 {
formatInfo.NodeID = conf.NodeID
formatInfo.NodeCtime = time.Now().UnixNano()
dm, err = updateDiskMeta(ctx, sb, formatInfo, conf.Path)
if dm.NodeID == 0 {
dm, err = updateDiskMeta(ctx, sb, *formatInfo, conf.Path)
if err != nil {
return nil, err
}
@ -1105,6 +1108,7 @@ func newDiskStorage(ctx context.Context, conf core.Config) (ds *DiskStorage, err
ds = &DiskStorage{
DiskID: dm.DiskID,
nodeID: dm.NodeID, // from formatInfo
SuperBlock: sb,
DataPath: diskDataPath,
MetaPath: diskMetaPath,
@ -1158,22 +1162,20 @@ func registerDisk(ctx context.Context, sb *SuperBlock, conf *core.Config) (dm co
now := time.Now().UnixNano()
format := &core.FormatInfo{
FormatInfoProtectedField: core.FormatInfoProtectedField{
DiskID: diskID,
NodeID: conf.NodeID,
Version: _diskVer[0],
Format: core.FormatMetaTypeV1,
Ctime: now,
NodeCtime: now,
},
format := &core.FormatInfo{}
format.NodeID = conf.NodeID
format.NodeCtime = now
format.FormatInfoProtectedField = core.FormatInfoProtectedField{
DiskID: diskID,
Version: _diskVer[0],
Format: core.FormatMetaTypeV1,
Ctime: now,
}
checkSum, err := format.CalCheckSum()
if err != nil {
if err = format.CalCheckSum(); err != nil {
span.Errorf("cal format info crc failed: %v", err)
return
}
format.CheckSum = checkSum
// dm.Host =
dm = core.DiskMeta{
@ -1200,20 +1202,13 @@ func registerDisk(ctx context.Context, sb *SuperBlock, conf *core.Config) (dm co
return
}
func updateDiskMeta(ctx context.Context, sb *SuperBlock, format *core.FormatInfo, path string) (dm core.DiskMeta, err error) {
func updateDiskMeta(ctx context.Context, sb *SuperBlock, format core.FormatInfo, path string) (dm core.DiskMeta, err error) {
span := trace.SpanFromContextSafe(ctx)
span.Infof("update disk[%d:%s] meta", format.DiskID, path)
checkSum, err := format.CalCheckSum()
if err != nil {
span.Errorf("cal format info crc failed: %v", err)
return
}
format.CheckSum = checkSum
// update disk meta
dm = core.DiskMeta{
FormatInfo: *format,
FormatInfo: format,
Mtime: format.NodeCtime,
Registered: true,
Status: proto.DiskStatusNormal,
@ -1225,12 +1220,7 @@ func updateDiskMeta(ctx context.Context, sb *SuperBlock, format *core.FormatInfo
return
}
if err = core.SaveDiskFormatInfo(ctx, path, format); err != nil {
span.Errorf("Failed save disk[%s] format info, err:%v", path, err)
return
}
span.Infof("update disk[%d:%s] meta, success", format.DiskID, path)
span.Infof("update disk[%d:%s] meta, success", dm.DiskID, path)
return
}

22
blobstore/blobnode/core/disk/disk_test.go
View File

@ -895,29 +895,17 @@ func TestDiskUpdateDiskMeta(t *testing.T) {
NotifyCompacting: setChunkCompactFn,
HandleIOError: handleIOErrorFn,
HostInfo: core.HostInfo{
NodeID: proto.NodeID(0), // Old version: NodeID is 0
NodeID: proto.NodeID(1), // Old version: NodeID is 1
},
}
ds, err := NewDiskStorage(ctx, diskConfig)
require.NoError(t, err)
require.NotNil(t, ds)
// Manually set format info NodeID to 0 to simulate old version
// Verify format info has NodeID = 1
formatInfo, err := core.ReadFormatInfo(ctx, diskpath)
require.NoError(t, err)
formatInfo.NodeID = proto.NodeID(0)
formatInfo.NodeCtime = 0
checkSum, err := formatInfo.CalCheckSum()
require.NoError(t, err)
formatInfo.CheckSum = checkSum
err = core.SaveDiskFormatInfo(ctx, diskpath, formatInfo)
require.NoError(t, err)
// Verify format info has NodeID = 0
formatInfo, err = core.ReadFormatInfo(ctx, diskpath)
require.NoError(t, err)
require.Equal(t, proto.NodeID(0), formatInfo.NodeID)
require.Equal(t, proto.NodeID(1), formatInfo.NodeID)
// Close disk storage properly
done := make(chan struct{})
@ -942,13 +930,13 @@ func TestDiskUpdateDiskMeta(t *testing.T) {
// Step 3: Verify NodeID is updated
formatInfo, err = core.ReadFormatInfo(ctx, diskpath)
require.NoError(t, err)
require.Equal(t, expectedNodeID, formatInfo.NodeID)
require.Equal(t, proto.NodeID(1), formatInfo.NodeID)
require.NotEqual(t, int64(0), formatInfo.NodeCtime) // NodeCtime should be updated
// Verify disk meta also has correct NodeID
dm, err := ds.SuperBlock.LoadDiskInfo(ctx)
require.NoError(t, err)
require.Equal(t, expectedNodeID, dm.NodeID)
require.Equal(t, proto.NodeID(1), dm.NodeID)
}
func TestRegisterDiskWithNodeID(t *testing.T) {

13
blobstore/blobnode/core/disk/superblock_test.go
View File

@ -108,13 +108,12 @@ func TestSuperBlock_RegisterDisk(t *testing.T) {
diskid := proto.DiskID(1024)
now := int64(time.Now().UnixNano())
format := &core.FormatInfo{
FormatInfoProtectedField: core.FormatInfoProtectedField{
DiskID: diskid,
Version: 0x1,
Format: core.FormatMetaTypeV1,
Ctime: now,
},
format := &core.FormatInfo{}
format.FormatInfoProtectedField = core.FormatInfoProtectedField{
DiskID: diskid,
Version: 1,
Format: core.FormatMetaTypeV1,
Ctime: now,
}
dm := core.DiskMeta{

116
blobstore/blobnode/core/format.go
View File

@ -39,16 +39,12 @@ import (
const (
formatConfigFile = ".format.json"
formatConfigFileTmp = ".format.json.tmp"
)
const (
_systemMeta = ".sys"
_trashPrefix = ".trash"
_dataSpacePrefix = "data"
_metaSpacePrefix = "meta"
)
const (
FormatMetaTypeV1 = "fs"
formatInfoCheckSumPoly = uint32(0xebf0ace5)
)
@ -56,20 +52,32 @@ const (
var (
ErrFormatInfoCheckSum = errors.New("format info check sum error")
ErrInvalidPathPrefix = errors.New("invalid path prefix")
ErrFormatV2CrcIsEmpty = errors.New("format info version 2 crc is empty")
)
// FormatInfoProtectedFieldInitVersion (not contains NodeID and NodeCtime)
type FormatInfoProtectedField struct {
Version uint8 `json:"version"`
DiskID proto.DiskID `json:"diskid"`
DiskID proto.DiskID `json:"diskid"`
Version uint8 `json:"version"`
Ctime int64 `json:"ctime"`
Format string `json:"format"`
}
type FormatInfoV1 struct {
FormatInfoProtectedField
CheckSum uint32 `json:"check_sum"`
}
// version 2: with v1 version and NodeID, NodeCtime
type FormatInfoProtectedFieldV2 struct {
FormatInfoV1
NodeID proto.NodeID `json:"nodeid"`
Ctime int64 `json:"ctime"`
NodeCtime int64 `json:"node_ctime"`
Format string `json:"format"`
}
type FormatInfo struct {
FormatInfoProtectedField
CheckSum uint32 `json:"check_sum"`
FormatInfoProtectedFieldV2
CheckSumV2 uint32 `json:"check_sum_v2"`
}
func sysRootPath(diskRoot string) (path string) {
@ -159,7 +167,7 @@ func SaveDiskFormatInfo(ctx context.Context, diskPath string, formatInfo *Format
}
file.Close()
// rename
// rename, replace save with new file
err = os.Rename(configFileTemp, configFile)
if err != nil {
span.Errorf("Failed rename, err:%v", err)
@ -167,13 +175,10 @@ func SaveDiskFormatInfo(ctx context.Context, diskPath string, formatInfo *Format
}
span.Infof("save format info success")
return nil
}
func ReadFormatInfo(ctx context.Context, diskRootPath string) (
formatInfo *FormatInfo, err error,
) {
func ReadFormatInfo(ctx context.Context, diskRootPath string) (info *FormatInfo, err error) {
span := trace.SpanFromContextSafe(ctx)
configFile := filepath.Join(sysRootPath(diskRootPath), formatConfigFile)
@ -183,18 +188,22 @@ func ReadFormatInfo(ctx context.Context, diskRootPath string) (
return nil, err
}
formatInfo = &FormatInfo{}
if err = json.Unmarshal(buf, formatInfo); err != nil {
info = &FormatInfo{}
if err = json.Unmarshal(buf, info); err != nil {
span.Errorf("Failed unmarshal, err:%v", err)
return nil, err
}
err = formatInfo.Verify()
if err != nil {
if info.CheckSumV2 == 0 {
return info, ErrFormatV2CrcIsEmpty
}
if err = info.Verify(); err != nil {
span.Errorf("Failed check format info crc, err:%v", err)
return nil, err
}
return formatInfo, nil
return info, nil
}
func IsFormatConfigExist(diskRootPath string) (bool, error) {
@ -202,7 +211,61 @@ func IsFormatConfigExist(diskRootPath string) (bool, error) {
return base.IsFileExists(configFile)
}
func (fi *FormatInfo) CalCheckSum() (uint32, error) {
func (fi *FormatInfo) Verify() error {
if fi.CheckSumV2 == 0 {
return ErrFormatV2CrcIsEmpty
}
checkSum, err := fi.calCheckSumV2()
if err != nil {
return err
}
if checkSum != fi.CheckSumV2 {
return ErrFormatInfoCheckSum
}
return nil
}
func (fi *FormatInfo) VerifyV1() error {
checkSum, err := fi.calCheckSumV1()
if err != nil {
return err
}
if checkSum != fi.CheckSum {
return ErrFormatInfoCheckSum
}
return nil
}
func (fi *FormatInfo) CalCheckSum() error {
checkSum, err := fi.calCheckSumV2()
if err != nil {
return err
}
fi.CheckSumV2 = checkSum
return nil
}
func (fi *FormatInfo) calCheckSumV2() (uint32, error) {
crc := crc32.New(crc32.MakeTable(formatInfoCheckSumPoly))
b, err := json.Marshal(fi.FormatInfoProtectedFieldV2)
if err != nil {
return proto.InvalidCrc32, err
}
_, err = crc.Write(b)
if err != nil {
return proto.InvalidCrc32, err
}
return crc.Sum32(), nil
}
func (fi *FormatInfo) calCheckSumV1() (uint32, error) {
crc := crc32.New(crc32.MakeTable(formatInfoCheckSumPoly))
b, err := json.Marshal(fi.FormatInfoProtectedField)
@ -217,14 +280,3 @@ func (fi *FormatInfo) CalCheckSum() (uint32, error) {
return crc.Sum32(), nil
}
func (fi *FormatInfo) Verify() error {
checkSum, err := fi.CalCheckSum()
if err != nil {
return err
}
if checkSum != fi.CheckSum {
return ErrFormatInfoCheckSum
}
return nil
}

133
blobstore/blobnode/core/format_test.go
View File

@ -16,8 +16,10 @@ package core
import (
"context"
"encoding/json"
"fmt"
"os"
"reflect"
"path/filepath"
"testing"
"time"
@ -27,55 +29,50 @@ import (
)
func TestFormatInfo(t *testing.T) {
formatInfo := &FormatInfo{
FormatInfoProtectedField: FormatInfoProtectedField{
DiskID: proto.DiskID(101),
Version: 1,
Format: FormatMetaTypeV1,
Ctime: time.Now().UnixNano(),
},
formatInfo := &FormatInfo{}
formatInfo.FormatInfoProtectedField = FormatInfoProtectedField{
DiskID: proto.DiskID(101),
Version: 1,
Format: FormatMetaTypeV1,
Ctime: time.Now().UnixNano(),
}
checkSum, err := formatInfo.CalCheckSum()
checkSum, err := formatInfo.calCheckSumV2()
require.NoError(t, err)
formatInfo.CheckSum = checkSum
ctx := context.Background()
diskPath, err := os.MkdirTemp(os.TempDir(), "BlobNodeTestFormatInfo")
require.NoError(t, err)
defer os.RemoveAll(diskPath)
sysPath := filepath.Join(diskPath, ".sys")
err = os.MkdirAll(sysPath, 0o755)
require.NoError(t, err)
oldFormatJSON := fmt.Sprintf(`{"version":1,"diskid":101,"ctime":%d,"format":"fs","check_sum":%d}`,
formatInfo.Ctime, checkSum)
formatFile := filepath.Join(sysPath, ".format.json")
err = os.WriteFile(formatFile, []byte(oldFormatJSON), 0o644)
require.NoError(t, err)
_, err = ReadFormatInfo(ctx, diskPath)
require.NotNil(t, err)
require.ErrorIs(t, err, ErrFormatV2CrcIsEmpty)
// update formatInfo, with nodeID
formatInfo.NodeID = proto.NodeID(2)
formatInfo.NodeCtime = time.Now().UnixNano()
// formatInfo.Version = DiskFormatVersionNode
err = formatInfo.CalCheckSum()
require.NoError(t, err)
err = SaveDiskFormatInfo(ctx, diskPath, formatInfo)
require.NoError(t, err)
info, err := ReadFormatInfo(ctx, diskPath)
require.NoError(t, err)
require.Equal(t, true, reflect.DeepEqual(*info, *formatInfo))
formatInfo.Ctime = time.Now().UnixNano()
err = SaveDiskFormatInfo(ctx, diskPath, formatInfo)
require.NoError(t, err)
_, err = ReadFormatInfo(ctx, diskPath)
require.Error(t, err)
// update formatInfo, nodeID
formatInfo.NodeID = proto.NodeID(2)
formatInfo.NodeCtime = time.Now().UnixNano()
checkSum, err = formatInfo.CalCheckSum()
require.NoError(t, err)
formatInfo.CheckSum = checkSum
err = SaveDiskFormatInfo(ctx, diskPath, formatInfo)
require.NoError(t, err)
info, err = ReadFormatInfo(ctx, diskPath)
require.NoError(t, err)
require.Equal(t, true, reflect.DeepEqual(*info, *formatInfo))
require.Equal(t, info.NodeID, formatInfo.NodeID)
require.NotEqual(t, proto.NodeID(0), info.NodeID)
}
func TestEnsureDiskArea(t *testing.T) {
@ -83,3 +80,69 @@ func TestEnsureDiskArea(t *testing.T) {
err := EnsureDiskArea(diskPath, "")
require.Error(t, err)
}
func TestOldVersionFomatInfo(t *testing.T) {
ctx := context.Background()
diskPath, err := os.MkdirTemp(os.TempDir(), "BlobNodeTestOldVersionFormatInfo")
require.NoError(t, err)
defer os.RemoveAll(diskPath)
sysPath := filepath.Join(diskPath, ".sys")
err = os.MkdirAll(sysPath, 0o755)
require.NoError(t, err)
oldFormatContent := `{"diskid":4,"version":1,"ctime":1766558856695434352,"format":"fs","check_sum":1501699157}`
formatFile := filepath.Join(sysPath, ".format.json")
err = os.WriteFile(formatFile, []byte(oldFormatContent), 0o644)
require.NoError(t, err)
// read old version
info, err := ReadFormatInfo(ctx, diskPath)
require.ErrorIs(t, err, ErrFormatV2CrcIsEmpty)
require.NotNil(t, info)
require.Equal(t, proto.DiskID(4), info.DiskID)
require.Equal(t, uint8(1), info.Version)
require.Equal(t, int64(1766558856695434352), info.Ctime)
require.Equal(t, FormatMetaTypeV1, info.Format)
require.Equal(t, uint32(1501699157), info.CheckSum)
require.Equal(t, proto.NodeID(0), info.NodeID)
require.Equal(t, int64(0), info.NodeCtime)
// update formatInfo, nodeID
info.NodeID = proto.NodeID(2)
info.NodeCtime = time.Now().UnixNano()
// info.Version = DiskFormatVersionNode
err = info.CalCheckSum()
require.NoError(t, err)
err = SaveDiskFormatInfo(ctx, diskPath, info)
require.NoError(t, err)
infoNew, err := ReadFormatInfo(ctx, diskPath)
require.NoError(t, err)
require.Equal(t, *info, *infoNew)
// rollback, startup read old version
info.Version = 1
info, err = ReadFormatInfo(ctx, diskPath)
require.NoError(t, err)
err = info.Verify()
require.NoError(t, err)
newFormatContent := `{"diskid":4,"version":1,"ctime":1766558856695434352,"format":"fs",
"nodeid":2,"node_ctime":1767166122300324697,"check_sum":1501699157,"check_sum_v2":1111,"xxx":0}`
err = os.WriteFile(formatFile, []byte(newFormatContent), 0o644)
require.NoError(t, err)
configFile := filepath.Join(sysRootPath(diskPath), formatConfigFile)
buf, err := os.ReadFile(configFile)
require.NoError(t, err)
infoV1 := &FormatInfoV1{}
err = json.Unmarshal(buf, infoV1)
require.NoError(t, err)
require.Equal(t, uint32(1501699157), infoV1.CheckSum)
}

1
blobstore/blobnode/core/proto.go
View File

@ -160,6 +160,7 @@ type ChunkAPI interface {
type DiskAPI interface {
ID() proto.DiskID
NodeID() proto.NodeID
Status() (status proto.DiskStatus)
DiskInfo() (info clustermgr.BlobNodeDiskInfo)
Stats() (stat DiskStats)

14
blobstore/blobnode/disk_mock_test.go
View File

@ -265,6 +265,20 @@ func (mr *MockDiskAPIMockRecorder) LoadDiskInfo(arg0 interface{}) *gomock.Call {
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "LoadDiskInfo", reflect.TypeOf((*MockDiskAPI)(nil).LoadDiskInfo), arg0)
}
// NodeID mocks base method.
func (m *MockDiskAPI) NodeID() proto.NodeID {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "NodeID")
ret0, _ := ret[0].(proto.NodeID)
return ret0
}
// NodeID indicates an expected call of NodeID.
func (mr *MockDiskAPIMockRecorder) NodeID() *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "NodeID", reflect.TypeOf((*MockDiskAPI)(nil).NodeID))
}
// PrepareClose mocks base method.
func (m *MockDiskAPI) PrepareClose(arg0 context.Context) {
m.ctrl.T.Helper()

44
blobstore/blobnode/startup.go
View File

@ -352,21 +352,44 @@ func setDefaultIOStat(dryRun bool) error {
return nil
}
func readFormatInfo(ctx context.Context, diskRootPath string) (
formatInfo *core.FormatInfo, err error,
) {
func readFormatInfo(ctx context.Context, diskRootPath string, nodeID proto.NodeID) (formatInfo *core.FormatInfo, err error) {
span := trace.SpanFromContextSafe(ctx)
_, err = os.ReadDir(diskRootPath)
if err != nil {
span.Errorf("read disk root path error:%s", diskRootPath)
return nil, err
}
formatInfo, err = core.ReadFormatInfo(ctx, diskRootPath)
if err != nil {
if os.IsNotExist(err) {
span.Warnf("format file not exist. must be first register")
span.Warnf("disk[%s] format file not exist. must be first register", diskRootPath)
return new(core.FormatInfo), nil
}
// case: format.info v1 upgrade to v2
if err == core.ErrFormatV2CrcIsEmpty {
if err = formatInfo.VerifyV1(); err != nil {
span.Errorf("Failed verify disk[%s] v1 format info, err:%+v", diskRootPath, err)
return nil, err
}
// upgrade to v2, format info v1 -> v2, write to disk
formatInfo.NodeID = nodeID
formatInfo.NodeCtime = time.Now().UnixNano()
if err = formatInfo.CalCheckSum(); err != nil {
span.Errorf("Failed cal disk[%s] format info crc: %v", diskRootPath, err)
return nil, err
}
if err = core.SaveDiskFormatInfo(ctx, diskRootPath, formatInfo); err != nil {
span.Errorf("Failed save disk[%s] format info, err:%v", diskRootPath, err)
return nil, err
}
return formatInfo, nil
}
span.Errorf("Failed read disk[%s] format info, err:%+v", diskRootPath, err)
return nil, err
}
@ -503,13 +526,19 @@ func startBlobnodeService(ctx context.Context, svr *Service, conf Config) (err e
}
// read disk meta. get DiskID
format, err := readFormatInfo(ctx, diskConf.Path)
format, err := readFormatInfo(ctx, diskConf.Path, conf.NodeID)
if err != nil {
svr.handleStartDiskError(ctx, foundDiskPathInCluster, diskConf.Path, 0, err)
return // skip
}
span.Debugf("local disk meta: %v", format)
// check node id after NewDiskStorage. only when it is not new disk
if format.DiskID != 0 && (format.NodeID != conf.NodeID || format.NodeID == proto.NodeID(0)) {
span.Fatalf("disk[%d:%s] nodeID not match, registered cm[%d], local[%d]",
format.DiskID, diskConf.Path, conf.NodeID, format.NodeID)
}
// found diskInfo store in cluster mgr
var ds core.DiskAPI
if format.DiskID == 0 {
@ -538,11 +567,6 @@ func startBlobnodeService(ctx context.Context, svr *Service, conf Config) (err e
svr.handleStartDiskError(ctx, foundDiskPathInCluster, diskConf.Path, format.DiskID, err)
return
}
// check node id after NewDiskStorage. only when it is not new disk
if format.DiskID != 0 && (format.NodeID != conf.NodeID || format.NodeID == proto.NodeID(0)) {
span.Fatalf("disk[%d:%s] nodeID not match, registered cm[%d], local[%d]",
format.DiskID, diskConf.Path, conf.NodeID, format.NodeID)
}
// new disk, register to cm: not found in cm, or format.diskID==0 and old disk is repaired
diskInfo, foundIDInCluster := foundDiskIDInCluster[format.DiskID]

24
blobstore/blobnode/svr_test.go
View File

@ -1210,11 +1210,11 @@ func TestService_OnlyBlobnode_OpenFailedEIO(t *testing.T) {
cmCli.EXPECT().ListHostDisk(A, A).Return([]*cmapi.BlobNodeDiskInfo{diskInfo1, diskInfo2}, nil)
cmCli.EXPECT().SetDisk(A, A, A).Return(nil)
patches := gomonkey.ApplyFunc(readFormatInfo, func(ctx context.Context, path string) (*core.FormatInfo, error) {
patches := gomonkey.ApplyFunc(readFormatInfo, func(ctx context.Context, path string, nodeID proto.NodeID) (*core.FormatInfo, error) {
if path == path1 || path == path2 {
return nil, syscall.EIO
}
return &core.FormatInfo{CheckSum: 1}, nil
return &core.FormatInfo{CheckSumV2: 1}, nil
})
defer patches.Reset()
patches2 := gomonkey.ApplyFunc(disk.NewDiskStorage, func(ctx context.Context, diskConf core.Config) (*disk.DiskStorageWrapper, error) {
@ -1432,7 +1432,7 @@ func TestService_OnlyBlobnode_Fatal(t *testing.T) {
cmCli.EXPECT().ListHostDisk(A, A).Return([]*cmapi.BlobNodeDiskInfo{diskInfo1, diskInfo2}, nil)
// cmCli.EXPECT().AllocDiskID(A).Return(proto.DiskID(102), nil)
patches := gomonkey.ApplyFunc(readFormatInfo, func(ctx context.Context, path string) (*core.FormatInfo, error) {
patches := gomonkey.ApplyFunc(readFormatInfo, func(ctx context.Context, path string, nodeID proto.NodeID) (*core.FormatInfo, error) {
if path == path1 {
return nil, errMock
}
@ -1494,17 +1494,17 @@ func TestService_OnlyBlobnode_OpenOldDisk(t *testing.T) {
cmCli.EXPECT().AddNode(A, A).Return(proto.NodeID(1), nil).Times(1)
cmCli.EXPECT().ListHostDisk(A, A).Return([]*cmapi.BlobNodeDiskInfo{diskInfo1}, nil)
format := &core.FormatInfo{
FormatInfoProtectedField: core.FormatInfoProtectedField{
DiskID: proto.DiskID(1),
NodeID: proto.NodeID(1),
Version: 1,
Format: core.FormatMetaTypeV1,
},
format := &core.FormatInfo{}
format.FormatInfoProtectedField = core.FormatInfoProtectedField{
DiskID: proto.DiskID(1),
Version: 1,
Format: core.FormatMetaTypeV1,
Ctime: time.Now().UnixNano(),
}
checkSum, err := format.CalCheckSum()
format.NodeID = 1
err = format.CalCheckSum()
require.NoError(t, err)
format.CheckSum = checkSum
err = core.SaveDiskFormatInfo(ctx, path1, format)
require.NoError(t, err)