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

feat(clustermgr): lock unlock volume support epoch

Browse Source 
with #1000102800

Signed-off-by: tangdeyi <tangdeyi@oppo.com>
This commit is contained in:
tangdeyi 2025-05-07 15:47:29 +08:00 committed by slasher
parent f2bd5dc417
commit 3cebf9fdbf
12 changed files with 287 additions and 70 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
blobstore/api/clustermgr/volume.go
View File

@ -70,6 +70,7 @@ type VolumeInfoBase struct {
Free uint64 `json:"free"`
Used uint64 `json:"used"`
CreateByNodeID uint64 `json:"create_by_node_id"`
Epoch uint32 `json:"epoch"`
}
type AllocVolumeInfo struct {
@ -156,7 +157,8 @@ func (c *Client) RetainVolume(ctx context.Context, args *RetainVolumeArgs) (ret
}
type LockVolumeArgs struct {
Vid proto.Vid `json:"vid"`
Vid proto.Vid `json:"vid"`
Epoch uint32 `json:"epoch"`
}
func (c *Client) LockVolume(ctx context.Context, args *LockVolumeArgs) (err error) {
@ -166,6 +168,7 @@ func (c *Client) LockVolume(ctx context.Context, args *LockVolumeArgs) (err erro
type UnlockVolumeArgs struct {
Vid proto.Vid `json:"vid"`
Epoch uint32 `json:"epoch"`
Force bool `json:"force"`
}

1
blobstore/clustermgr/persistence/volumedb/volumetbl.go
View File

@ -52,6 +52,7 @@ type VolumeRecord struct {
Free uint64
Used uint64
CreateByNodeID uint64
Epoch uint32
}
type VolumeTaskRecord struct {

4
blobstore/clustermgr/volume.go
View File

@ -240,7 +240,7 @@ func (s *Service) VolumeLock(c *rpc.Context) {
}
span.Debugf("accept VolumeLock request, args: %v", args)
c.RespondError(s.VolumeMgr.LockVolume(ctx, args.Vid))
c.RespondError(s.VolumeMgr.LockVolume(ctx, args))
}
func (s *Service) VolumeUnlock(c *rpc.Context) {
@ -253,7 +253,7 @@ func (s *Service) VolumeUnlock(c *rpc.Context) {
}
span.Debugf("accept VolumeUnlock request, args: %v", args)
c.RespondError(s.VolumeMgr.UnlockVolume(ctx, args.Vid, args.Force))
c.RespondError(s.VolumeMgr.UnlockVolume(ctx, args))
}
func (s *Service) VolumeUnitAlloc(c *rpc.Context) {

3
blobstore/clustermgr/volume_test.go
View File

@ -322,7 +322,8 @@ func TestService_VolumeLock(t *testing.T) {
// unlock volume
{
args := &clustermgr.UnlockVolumeArgs{
Vid: proto.Vid(1),
Vid: proto.Vid(1),
Epoch: 1,
}
err := cmClient.UnlockVolume(ctx, args)
require.NoError(t, err)

10
blobstore/clustermgr/volumemgr/applier.go
View File

@ -90,9 +90,11 @@ type AllocVolumeCtx struct {
}
type ChangeVolStatusCtx struct {
Vid proto.Vid `json:"vid"`
TaskID string `json:"task_id"`
TaskType base.VolumeTaskType `json:"type"`
Vid proto.Vid `json:"vid"`
TaskID string `json:"task_id"`
TaskType base.VolumeTaskType `json:"type"`
Epoch uint32 `json:"epoch"`
PendingErrKey interface{} `json:"pending_err_key"`
}
type allocVolumeUnitCtx struct {
@ -197,7 +199,7 @@ func (v *VolumeMgr) Apply(ctx context.Context, operTypes []int32, datas [][]byte
continue
}
v.applyTaskPool.Run(v.getTaskIdx(args.Vid), func() {
if err = v.applyVolumeTask(taskCtx, args.Vid, args.TaskID, args.TaskType); err != nil {
if err = v.applyVolumeTask(taskCtx, args); err != nil {
errs[idx] = errors.Info(err, "apply change volume status failed, args: ", args).Detail(err)
}
wg.Done()

9
blobstore/clustermgr/volumemgr/proto.go
View File

@ -68,6 +68,7 @@ func (vol *volume) ToRecord() *volumedb.VolumeRecord {
Free: vol.volInfoBase.Free,
Used: vol.volInfoBase.Used,
CreateByNodeID: vol.volInfoBase.CreateByNodeID,
Epoch: vol.volInfoBase.Epoch,
}
}
@ -86,6 +87,14 @@ func (vol *volume) ToVolumeInfo() cm.VolumeInfo {
}
}
func (vol *volume) getEpoch() uint32 {
return vol.volInfoBase.Epoch
}
func (vol *volume) increaseEpoch() {
vol.volInfoBase.Epoch += 1
}
func (vol *volume) getStatus() proto.VolumeStatus {
return vol.volInfoBase.Status
}

38
blobstore/clustermgr/volumemgr/volume_task.go
View File

@ -144,23 +144,38 @@ func (m *VolumeMgr) setVolumeStatus(task *volTask) error {
return nil
}
func (m *VolumeMgr) applyVolumeTask(ctx context.Context, vid proto.Vid, taskID string, t base.VolumeTaskType) error {
func (m *VolumeMgr) applyVolumeTask(ctx context.Context, args *ChangeVolStatusCtx) error {
// get volume from cache
span := trace.SpanFromContextSafe(ctx)
vol := m.all.getVol(vid)
task := newVolTask(vid, t, taskID, m.setVolumeStatus)
vol := m.all.getVol(args.Vid)
task := newVolTask(args.Vid, args.TaskType, args.TaskID, m.setVolumeStatus)
span.Infof("create task %s", task.String())
// set volume status=lock if t is base.VolumeTaskTypeLock
var (
err error
addNewTask bool
taskRecord = &volumedb.VolumeTaskRecord{
Vid: vid,
Vid: args.Vid,
TaskType: task.taskType,
TaskId: task.taskId,
}
)
switch t {
// double check volume epoch when apply (api concurrent request or raft log replay)
err = vol.withRLocked(func() error {
if vol.getEpoch() != args.Epoch {
return apierrs.ErrVolumeEpochNotMatch
}
return nil
})
if err != nil {
// return err by pendingEntries in commit case
if _, ok := m.pendingEntries.Load(args.PendingErrKey); ok {
m.pendingEntries.Store(args.PendingErrKey, err)
}
return nil
}
switch args.TaskType {
case base.VolumeTaskTypeLock:
err = vol.withLocked(func() error {
if !vol.canLock() {
@ -171,6 +186,7 @@ func (m *VolumeMgr) applyVolumeTask(ctx context.Context, vid proto.Vid, taskID s
addNewTask = true
// set volume status into lock, it'll call change volume status function
vol.setStatus(ctx, proto.VolumeStatusLock)
vol.increaseEpoch()
rec := vol.ToRecord()
// store task to db
return m.volumeTbl.PutVolumeAndTask(rec, taskRecord)
@ -184,6 +200,7 @@ func (m *VolumeMgr) applyVolumeTask(ctx context.Context, vid proto.Vid, taskID s
addNewTask = true
vol.setStatus(ctx, proto.VolumeStatusUnlocking)
vol.increaseEpoch()
rec := vol.ToRecord()
// store task to db
return m.volumeTbl.PutVolumeAndTask(rec, taskRecord)
@ -195,14 +212,14 @@ func (m *VolumeMgr) applyVolumeTask(ctx context.Context, vid proto.Vid, taskID s
return nil
}
oldTask := m.taskMgr.GetTask(vid)
oldTask := m.taskMgr.GetTask(args.Vid)
if oldTask != nil {
if oldTask.taskType == t {
if oldTask.taskType == args.TaskType {
span.Warnf("volume already in unlocking force")
return nil
}
// other task type exist, remove old task firstly
if err = m.volumeTbl.DeleteTaskRecord(vid); err != nil {
if err = m.volumeTbl.DeleteTaskRecord(args.Vid); err != nil {
span.Errorf("remove old task type task %s error: %v", task.String(), err)
return err
}
@ -210,12 +227,13 @@ func (m *VolumeMgr) applyVolumeTask(ctx context.Context, vid proto.Vid, taskID s
addNewTask = true
vol.setStatus(ctx, proto.VolumeStatusUnlocking)
vol.increaseEpoch()
rec := vol.ToRecord()
// store task to db
return m.volumeTbl.PutVolumeAndTask(rec, taskRecord)
})
default:
span.Panicf("Unknown task type(%d)", t)
span.Panicf("Unknown task type(%d)", args.TaskType)
}
if err != nil {
span.Errorf("persist task %s error: %v", task.String(), err)
@ -226,7 +244,7 @@ func (m *VolumeMgr) applyVolumeTask(ctx context.Context, vid proto.Vid, taskID s
}
// add task into taskManager
m.lastTaskIdMap.Store(vid, task.taskId)
m.lastTaskIdMap.Store(args.Vid, task.taskId)
m.taskMgr.AddTask(task)
return nil
}

48
blobstore/clustermgr/volumemgr/volume_task_test.go
View File

@ -122,7 +122,13 @@ func TestTaskProc(t *testing.T) {
volMgr.all.putVol(vol)
_, ctx := trace.StartSpanFromContext(context.Background(), "")
// volume lock
volMgr.applyVolumeTask(ctx, vol.vid, uuid.New().String(), base.VolumeTaskTypeLock)
args := &ChangeVolStatusCtx{
Vid: vol.vid,
TaskID: uuid.New().String(),
TaskType: base.VolumeTaskTypeLock,
Epoch: vol.getEpoch(),
}
volMgr.applyVolumeTask(ctx, args)
require.Equal(t, proto.VolumeStatusLock, vol.volInfoBase.Status)
taskid, hit := volMgr.lastTaskIdMap.Load(vol.vid)
require.True(t, hit)
@ -132,7 +138,13 @@ func TestTaskProc(t *testing.T) {
require.False(t, hit)
// volume unlock
volMgr.applyVolumeTask(ctx, vol.vid, uuid.New().String(), base.VolumeTaskTypeUnlock)
args = &ChangeVolStatusCtx{
Vid: vol.vid,
TaskID: uuid.New().String(),
TaskType: base.VolumeTaskTypeUnlock,
Epoch: vol.getEpoch(),
}
volMgr.applyVolumeTask(ctx, args)
taskid, hit = volMgr.lastTaskIdMap.Load(vol.vid)
require.True(t, hit)
time.Sleep(100 * time.Millisecond) // wait task finish
@ -141,7 +153,13 @@ func TestTaskProc(t *testing.T) {
// test unlock volume force
{
volMgr.applyVolumeTask(ctx, vol.vid, uuid.New().String(), base.VolumeTaskTypeLock)
args = &ChangeVolStatusCtx{
Vid: vol.vid,
TaskID: uuid.New().String(),
TaskType: base.VolumeTaskTypeLock,
Epoch: vol.getEpoch(),
}
volMgr.applyVolumeTask(ctx, args)
require.Equal(t, proto.VolumeStatusLock, vol.volInfoBase.Status)
taskid, hit = volMgr.lastTaskIdMap.Load(vol.vid)
require.True(t, hit)
@ -150,16 +168,34 @@ func TestTaskProc(t *testing.T) {
_, hit = volMgr.lastTaskIdMap.Load(vol.vid)
require.False(t, hit)
volMgr.applyVolumeTask(ctx, vol.vid, uuid.New().String(), base.VolumeTaskTypeUnlock)
args = &ChangeVolStatusCtx{
Vid: vol.vid,
TaskID: uuid.New().String(),
TaskType: base.VolumeTaskTypeUnlock,
Epoch: vol.getEpoch(),
}
volMgr.applyVolumeTask(ctx, args)
_, hit = volMgr.lastTaskIdMap.Load(vol.vid)
require.True(t, hit)
// unlock volume force, replace the old unlock volume task
err = volMgr.applyVolumeTask(ctx, vol.vid, uuid.New().String(), base.VolumeTaskTypeUnlockForce)
args = &ChangeVolStatusCtx{
Vid: vol.vid,
TaskID: uuid.New().String(),
TaskType: base.VolumeTaskTypeUnlockForce,
Epoch: vol.getEpoch(),
}
err = volMgr.applyVolumeTask(ctx, args)
require.NoError(t, err)
taskid, hit = volMgr.lastTaskIdMap.Load(vol.vid)
require.True(t, hit)
// unlock volume force repeatedly
err = volMgr.applyVolumeTask(ctx, vol.vid, uuid.New().String(), base.VolumeTaskTypeUnlockForce)
args = &ChangeVolStatusCtx{
Vid: vol.vid,
TaskID: uuid.New().String(),
TaskType: base.VolumeTaskTypeUnlockForce,
Epoch: vol.getEpoch(),
}
err = volMgr.applyVolumeTask(ctx, args)
require.NoError(t, err)
err = volMgr.applyRemoveVolumeTask(ctx, vol.vid, taskid.(string), base.VolumeTaskTypeUnlockForce)
require.NoError(t, err)

95
blobstore/clustermgr/volumemgr/volumemgr.go
View File

@ -398,36 +398,55 @@ func (v *VolumeMgr) DiskWritableChange(ctx context.Context, diskID proto.DiskID)
return
}
func (v *VolumeMgr) LockVolume(ctx context.Context, vid proto.Vid) error {
func (v *VolumeMgr) LockVolume(ctx context.Context, args *cm.LockVolumeArgs) error {
span := trace.SpanFromContextSafe(ctx)
vol := v.all.getVol(vid)
vol := v.all.getVol(args.Vid)
if vol == nil {
span.Errorf("volume not found, vid: %d", vid)
span.Errorf("volume not found, vid: %d", args.Vid)
return apierrors.ErrVolumeNotExist
}
vol.lock.RLock()
status := vol.getStatus()
// volume already locked
if status == proto.VolumeStatusLock {
vol.lock.RUnlock()
alreadyLocked := false
err := vol.withRLocked(func() error {
status := vol.getStatus()
// volume already locked
if status == proto.VolumeStatusLock {
alreadyLocked = true
return nil
}
if !vol.canLock() {
span.Warnf("can't lock volume, volume(%d), current status(%d)", args.Vid, status)
return apierrors.ErrLockNotAllow
}
epoch := vol.getEpoch()
if epoch != args.Epoch {
span.Warnf("volume epoch(%d) not match with args(%d), volume(%d)", epoch, args.Epoch, args.Vid)
return apierrors.ErrVolumeEpochNotMatch
}
return nil
})
if alreadyLocked {
return nil
}
if !vol.canLock() {
vol.lock.RUnlock()
span.Warnf("can't lock volume, volume %d, current status(%d)", vid, status)
return apierrors.ErrLockNotAllow
if err != nil {
return err
}
vol.lock.RUnlock()
pendingErrKey := uuid.New().String()
v.pendingEntries.Store(pendingErrKey, nil)
// clear pending entry key
defer v.pendingEntries.Delete(pendingErrKey)
param := ChangeVolStatusCtx{
Vid: vid,
TaskID: uuid.New().String(),
TaskType: base.VolumeTaskTypeLock,
Vid: args.Vid,
TaskID: uuid.New().String(),
TaskType: base.VolumeTaskTypeLock,
Epoch: args.Epoch,
PendingErrKey: pendingErrKey,
}
data, err := json.Marshal(param)
if err != nil {
span.Errorf("json marshal failed, vid: %d, error: %v", vid, err)
span.Errorf("json marshal failed, vid: %d, error: %v", args.Vid, err)
return apierrors.ErrCMUnexpect
}
@ -437,38 +456,45 @@ func (v *VolumeMgr) LockVolume(ctx context.Context, vid proto.Vid) error {
span.Errorf("raft propose error: %v", err)
return apierrors.ErrRaftPropose
}
if value, _ := v.pendingEntries.Load(pendingErrKey); value != nil {
return value.(error)
}
vol.lock.RLock()
status = vol.getStatus()
status := vol.getStatus()
vol.lock.RUnlock()
if status != proto.VolumeStatusLock {
span.Errorf("volume %d status(%d) is not locked", vid, status)
span.Errorf("volume %d status(%d) is not locked", args.Vid, status)
return apierrors.ErrCMUnexpect
}
return nil
}
func (v *VolumeMgr) UnlockVolume(ctx context.Context, vid proto.Vid, force bool) error {
func (v *VolumeMgr) UnlockVolume(ctx context.Context, args *cm.UnlockVolumeArgs) error {
span := trace.SpanFromContextSafe(ctx)
vol := v.all.getVol(vid)
vol := v.all.getVol(args.Vid)
if vol == nil {
span.Errorf("volume not found, vid: %d", vid)
span.Errorf("volume not found, vid: %d", args.Vid)
return apierrors.ErrVolumeNotExist
}
propose := false
err := vol.withRLocked(func() error {
status := vol.getStatus()
if status == proto.VolumeStatusIdle || (!force && status == proto.VolumeStatusUnlocking) {
if status == proto.VolumeStatusIdle || (!args.Force && status == proto.VolumeStatusUnlocking) {
return nil
}
if status == proto.VolumeStatusActive {
span.Warnf("can't unlock volume, volume %d, current status(%d)", vid, vol.getStatus())
span.Warnf("can't unlock volume, volume %d, current status(%d)", args.Vid, vol.getStatus())
return apierrors.ErrUnlockNotAllow
}
epoch := vol.getEpoch()
if epoch != args.Epoch {
span.Warnf("volume epoch(%d) not match with args(%d), volume %d", epoch, args.Epoch, args.Vid)
return apierrors.ErrVolumeEpochNotMatch
}
propose = true
return nil
@ -477,18 +503,24 @@ func (v *VolumeMgr) UnlockVolume(ctx context.Context, vid proto.Vid, force bool)
return err
}
pendingErrKey := uuid.New().String()
v.pendingEntries.Store(pendingErrKey, nil)
// clear pending entry key
defer v.pendingEntries.Delete(pendingErrKey)
taskType := base.VolumeTaskTypeUnlock
if force {
if args.Force {
taskType = base.VolumeTaskTypeUnlockForce
}
param := ChangeVolStatusCtx{
Vid: vid,
TaskID: uuid.New().String(),
TaskType: taskType,
Vid: args.Vid,
TaskID: uuid.New().String(),
TaskType: taskType,
Epoch: args.Epoch,
PendingErrKey: pendingErrKey,
}
data, err := json.Marshal(param)
if err != nil {
span.Errorf("json marshal failed, vid: %d, error: %v", vid, err)
span.Errorf("json marshal failed, vid: %d, error: %v", args.Vid, err)
return apierrors.ErrCMUnexpect
}
@ -498,6 +530,9 @@ func (v *VolumeMgr) UnlockVolume(ctx context.Context, vid proto.Vid, force bool)
span.Errorf("raft propose error:%v", err)
return apierrors.ErrRaftPropose
}
if value, _ := v.pendingEntries.Load(pendingErrKey); value != nil {
return value.(error)
}
return nil
}

141
blobstore/clustermgr/volumemgr/volumemgr_test.go
View File

@ -25,6 +25,7 @@ import (
"testing"
"time"
apierrors "github.com/cubefs/cubefs/blobstore/common/errors"
"github.com/golang/mock/gomock"
"github.com/google/uuid"
"github.com/stretchr/testify/require"
@ -339,7 +340,12 @@ func Test_NewVolumeMgr(t *testing.T) {
vol1.lock.Unlock()
// test exec task
err = mockVolumeMgr.applyVolumeTask(context.Background(), 2, uuid.New().String(), base.VolumeTaskTypeLock)
args := &ChangeVolStatusCtx{
Vid: 2,
TaskID: uuid.New().String(),
TaskType: base.VolumeTaskTypeLock,
}
err = mockVolumeMgr.applyVolumeTask(context.Background(), args)
require.NoError(t, err)
vol2 := mockVolumeMgr.all.getVol(2)
require.Equal(t, proto.VolumeStatusLock, vol2.volInfoBase.Status)
@ -826,31 +832,63 @@ func TestVolumeMgr_LockVolume(t *testing.T) {
defer clean()
// not allow lock active volume
err := mockVolumeMgr.LockVolume(context.Background(), 1)
args := &clustermgr.LockVolumeArgs{
Vid: 1,
Epoch: 0,
}
err := mockVolumeMgr.LockVolume(context.Background(), args)
require.Error(t, err)
// vid not exist
err = mockVolumeMgr.LockVolume(context.Background(), 55)
args = &clustermgr.LockVolumeArgs{
Vid: 55,
Epoch: 0,
}
err = mockVolumeMgr.LockVolume(context.Background(), args)
require.Error(t, err)
mockRaftServer := mocks.NewMockRaftServer(gomock.NewController(t))
mockVolumeMgr.raftServer = mockRaftServer
mockRaftServer.EXPECT().Propose(gomock.Any(), gomock.Any()).Return(nil)
// not apply ,
// lock locked volume
vol2 := mockVolumeMgr.all.getVol(2)
require.Equal(t, proto.VolumeStatusIdle, vol2.volInfoBase.Status)
err = mockVolumeMgr.LockVolume(context.Background(), 2)
args = &clustermgr.LockVolumeArgs{
Vid: 2,
Epoch: 0,
}
err = mockVolumeMgr.LockVolume(context.Background(), args)
require.Error(t, err)
require.Equal(t, proto.VolumeStatusIdle, vol2.volInfoBase.Status)
err = mockVolumeMgr.applyVolumeTask(context.Background(), 2, uuid.New().String(), base.VolumeTaskTypeLock)
ctxArgs := &ChangeVolStatusCtx{
Vid: 2,
TaskID: uuid.New().String(),
TaskType: base.VolumeTaskTypeLock,
}
err = mockVolumeMgr.applyVolumeTask(context.Background(), ctxArgs)
require.NoError(t, err)
vol2 = mockVolumeMgr.all.getVol(2)
require.Equal(t, proto.VolumeStatusLock, vol2.volInfoBase.Status)
err = mockVolumeMgr.LockVolume(context.Background(), 2)
args = &clustermgr.LockVolumeArgs{
Vid: 2,
Epoch: 0,
}
err = mockVolumeMgr.LockVolume(context.Background(), args)
require.NoError(t, err)
// volume epoch
volumeInfo, err := mockVolumeMgr.GetVolumeInfo(context.Background(), 2)
require.NoError(t, err)
require.Equal(t, uint32(1), volumeInfo.Epoch)
// volume epoch not match
args = &clustermgr.LockVolumeArgs{
Vid: 4,
Epoch: 2,
}
err = mockVolumeMgr.LockVolume(context.Background(), args)
require.Error(t, err)
}
func TestVolumeMgr_UnlockVolume(t *testing.T) {
@ -861,49 +899,120 @@ func TestVolumeMgr_UnlockVolume(t *testing.T) {
mockVolumeMgr.raftServer = mockRaftServer
mockRaftServer.EXPECT().Propose(gomock.Any(), gomock.Any()).Return(nil).Times(2)
// failed case: lock status can unlock
// success case: idle status can unlock
vol2 := mockVolumeMgr.all.getVol(2)
require.Equal(t, proto.VolumeStatusIdle, vol2.volInfoBase.Status)
err := mockVolumeMgr.UnlockVolume(context.Background(), 2, false)
args := &clustermgr.UnlockVolumeArgs{
Vid: 2,
Epoch: 0,
Force: false,
}
err := mockVolumeMgr.UnlockVolume(context.Background(), args)
require.NoError(t, err)
// failed case: active status cannot unlock
args = &clustermgr.UnlockVolumeArgs{
Vid: 3,
Epoch: 0,
Force: false,
}
err = mockVolumeMgr.UnlockVolume(context.Background(), args)
require.Error(t, err)
// failed case: vid not exist
err = mockVolumeMgr.UnlockVolume(context.Background(), 55, false)
args = &clustermgr.UnlockVolumeArgs{
Vid: 55,
Epoch: 0,
Force: false,
}
err = mockVolumeMgr.UnlockVolume(context.Background(), args)
require.Error(t, err)
vol2.lock.Lock()
vol2.volInfoBase.Status = proto.VolumeStatusLock
vol2.lock.Unlock()
err = mockVolumeMgr.UnlockVolume(context.Background(), 2, false)
args = &clustermgr.UnlockVolumeArgs{
Vid: 2,
Epoch: 0,
Force: false,
}
err = mockVolumeMgr.UnlockVolume(context.Background(), args)
require.NoError(t, err)
err = mockVolumeMgr.UnlockVolume(context.Background(), 2, true)
args = &clustermgr.UnlockVolumeArgs{
Vid: 2,
Epoch: 0,
Force: true,
}
err = mockVolumeMgr.UnlockVolume(context.Background(), args)
require.NoError(t, err)
ret, err := mockVolumeMgr.GetVolumeInfo(context.Background(), 2)
require.NoError(t, err)
require.Equal(t, proto.VolumeStatusLock, ret.Status)
err = mockVolumeMgr.applyVolumeTask(context.Background(), 2, uuid.New().String(), base.VolumeTaskTypeUnlock)
ctxArgs := &ChangeVolStatusCtx{
Vid: 2,
TaskID: uuid.New().String(),
TaskType: base.VolumeTaskTypeUnlock,
Epoch: 0,
}
err = mockVolumeMgr.applyVolumeTask(context.Background(), ctxArgs)
require.NoError(t, err)
ret, err = mockVolumeMgr.GetVolumeInfo(context.Background(), 2)
require.NoError(t, err)
require.Equal(t, proto.VolumeStatusUnlocking, ret.Status)
require.Equal(t, uint32(1), ret.Epoch)
// epoch not match
args = &clustermgr.UnlockVolumeArgs{
Vid: 2,
Epoch: 0,
Force: true,
}
err = mockVolumeMgr.UnlockVolume(context.Background(), args)
require.Error(t, err)
// volume status id idle , cannot apply volume unlock task, direct return but error is nil
err = mockVolumeMgr.applyVolumeTask(context.Background(), 2, uuid.NewString(), base.VolumeTaskTypeUnlock)
ctxArgs = &ChangeVolStatusCtx{
Vid: 2,
TaskID: uuid.New().String(),
TaskType: base.VolumeTaskTypeUnlock,
Epoch: 1,
}
err = mockVolumeMgr.applyVolumeTask(context.Background(), ctxArgs)
require.NoError(t, err)
vol2.lock.Lock()
vol2.volInfoBase.Status = proto.VolumeStatusLock
vol2.lock.Unlock()
err = mockVolumeMgr.applyVolumeTask(context.Background(), 2, uuid.New().String(), base.VolumeTaskTypeUnlockForce)
ctxArgs = &ChangeVolStatusCtx{
Vid: 2,
TaskID: uuid.New().String(),
TaskType: base.VolumeTaskTypeUnlockForce,
Epoch: 1,
}
err = mockVolumeMgr.applyVolumeTask(context.Background(), ctxArgs)
require.NoError(t, err)
ret, err = mockVolumeMgr.GetVolumeInfo(context.Background(), 2)
require.NoError(t, err)
require.Equal(t, proto.VolumeStatusUnlocking, ret.Status)
// volume epoch not match
pendingKey := uuid.New().String()
mockVolumeMgr.pendingEntries.Store(pendingKey, nil)
ctxArgs = &ChangeVolStatusCtx{
Vid: 2,
TaskID: uuid.New().String(),
TaskType: base.VolumeTaskTypeUnlockForce,
Epoch: 100,
PendingErrKey: pendingKey,
}
err = mockVolumeMgr.applyVolumeTask(context.Background(), ctxArgs)
require.NoError(t, err)
v, _ := mockVolumeMgr.pendingEntries.Load(pendingKey)
require.Equal(t, apierrors.ErrVolumeEpochNotMatch, v)
}
func TestVolumeMgr_Report(t *testing.T) {

2
blobstore/common/errors/clustermgr.go
View File

@ -57,6 +57,7 @@ const (
CodeOldIsLeanerNotMatch = 943
CodeConcurrentAllocShardUnit = 944
CodeShardInitNotDone = 945
CodeVolumeEpochNotMatch = 946
)
var (
@ -102,4 +103,5 @@ var (
ErrOldIsLeanerNotMatch = Error(CodeOldIsLeanerNotMatch)
ErrConcurrentAllocShardUnit = Error(CodeConcurrentAllocShardUnit)
ErrShardInitNotDone = Error(CodeShardInitNotDone)
ErrVolumeEpochNotMatch = Error(CodeVolumeEpochNotMatch)
)

1
blobstore/common/errors/errors.go
View File

@ -102,6 +102,7 @@ var errCodeMap = map[int]string{
CodeOldIsLeanerNotMatch: "old leaner not match",
CodeConcurrentAllocShardUnit: "concurrent alloc shard unit",
CodeShardInitNotDone: "shard init not done",
CodeVolumeEpochNotMatch: "volume epoch not match",
// scheduler
CodeNotingTodo: "nothing to do",