From 3cebf9fdbfb2bb9aa2e4164820e9e79cc55012f8 Mon Sep 17 00:00:00 2001 From: tangdeyi Date: Wed, 7 May 2025 15:47:29 +0800 Subject: [PATCH] feat(clustermgr): lock unlock volume support epoch with #1000102800 Signed-off-by: tangdeyi --- blobstore/api/clustermgr/volume.go | 5 +- .../persistence/volumedb/volumetbl.go | 1 + blobstore/clustermgr/volume.go | 4 +- blobstore/clustermgr/volume_test.go | 3 +- blobstore/clustermgr/volumemgr/applier.go | 10 +- blobstore/clustermgr/volumemgr/proto.go | 9 ++ blobstore/clustermgr/volumemgr/volume_task.go | 38 +++-- .../clustermgr/volumemgr/volume_task_test.go | 48 +++++- blobstore/clustermgr/volumemgr/volumemgr.go | 95 ++++++++---- .../clustermgr/volumemgr/volumemgr_test.go | 141 ++++++++++++++++-- blobstore/common/errors/clustermgr.go | 2 + blobstore/common/errors/errors.go | 1 + 12 files changed, 287 insertions(+), 70 deletions(-) diff --git a/blobstore/api/clustermgr/volume.go b/blobstore/api/clustermgr/volume.go index 08ff167f3..358b256b5 100644 --- a/blobstore/api/clustermgr/volume.go +++ b/blobstore/api/clustermgr/volume.go @@ -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"` } diff --git a/blobstore/clustermgr/persistence/volumedb/volumetbl.go b/blobstore/clustermgr/persistence/volumedb/volumetbl.go index 17761d0db..152c37620 100644 --- a/blobstore/clustermgr/persistence/volumedb/volumetbl.go +++ b/blobstore/clustermgr/persistence/volumedb/volumetbl.go @@ -52,6 +52,7 @@ type VolumeRecord struct { Free uint64 Used uint64 CreateByNodeID uint64 + Epoch uint32 } type VolumeTaskRecord struct { diff --git a/blobstore/clustermgr/volume.go b/blobstore/clustermgr/volume.go index 540ca9d0b..e8803742c 100644 --- a/blobstore/clustermgr/volume.go +++ b/blobstore/clustermgr/volume.go @@ -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) { diff --git a/blobstore/clustermgr/volume_test.go b/blobstore/clustermgr/volume_test.go index 89fd37933..39fe21b7f 100644 --- a/blobstore/clustermgr/volume_test.go +++ b/blobstore/clustermgr/volume_test.go @@ -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) diff --git a/blobstore/clustermgr/volumemgr/applier.go b/blobstore/clustermgr/volumemgr/applier.go index 52fbee431..9581cca50 100644 --- a/blobstore/clustermgr/volumemgr/applier.go +++ b/blobstore/clustermgr/volumemgr/applier.go @@ -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() diff --git a/blobstore/clustermgr/volumemgr/proto.go b/blobstore/clustermgr/volumemgr/proto.go index d804047cb..ca4beaae0 100644 --- a/blobstore/clustermgr/volumemgr/proto.go +++ b/blobstore/clustermgr/volumemgr/proto.go @@ -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 } diff --git a/blobstore/clustermgr/volumemgr/volume_task.go b/blobstore/clustermgr/volumemgr/volume_task.go index f3ab281b7..32e563216 100644 --- a/blobstore/clustermgr/volumemgr/volume_task.go +++ b/blobstore/clustermgr/volumemgr/volume_task.go @@ -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 } diff --git a/blobstore/clustermgr/volumemgr/volume_task_test.go b/blobstore/clustermgr/volumemgr/volume_task_test.go index 4de5e06c9..b93a93e0a 100644 --- a/blobstore/clustermgr/volumemgr/volume_task_test.go +++ b/blobstore/clustermgr/volumemgr/volume_task_test.go @@ -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) diff --git a/blobstore/clustermgr/volumemgr/volumemgr.go b/blobstore/clustermgr/volumemgr/volumemgr.go index 2028de920..6b9ce2ab3 100644 --- a/blobstore/clustermgr/volumemgr/volumemgr.go +++ b/blobstore/clustermgr/volumemgr/volumemgr.go @@ -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 } diff --git a/blobstore/clustermgr/volumemgr/volumemgr_test.go b/blobstore/clustermgr/volumemgr/volumemgr_test.go index 48fb0bafc..1a73a99b1 100644 --- a/blobstore/clustermgr/volumemgr/volumemgr_test.go +++ b/blobstore/clustermgr/volumemgr/volumemgr_test.go @@ -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) { diff --git a/blobstore/common/errors/clustermgr.go b/blobstore/common/errors/clustermgr.go index b041cf487..41dfa8bc1 100644 --- a/blobstore/common/errors/clustermgr.go +++ b/blobstore/common/errors/clustermgr.go @@ -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) ) diff --git a/blobstore/common/errors/errors.go b/blobstore/common/errors/errors.go index e5c6dc5f5..5eab0e910 100644 --- a/blobstore/common/errors/errors.go +++ b/blobstore/common/errors/errors.go @@ -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",