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(metanode): add set meta partition freeze raft command.#22834621

Browse Source 
Signed-off-by: Wu Huocheng <wuhuocheng@oppo.com>
This commit is contained in:
Wu Huocheng 2024-12-23 15:22:45 +08:00 committed by zhumingze1108
parent 1565adf97d
commit f00702a0d0
7 changed files with 101 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
master/cluster.go
View File

@ -6410,18 +6410,21 @@ func (c *Cluster) DoCleanEmptyMetaPartition(name string) error {
}
func (c *Cluster) FreezeEmptyMetaPartition(mp *MetaPartition, freeze bool) error {
for _, replica := range mp.Replicas {
task := replica.createTaskToFreezeReplica(mp.PartitionID, freeze)
metaNode, err := c.metaNode(task.OperatorAddr)
if err != nil {
log.LogErrorf("failed to get metanode(%s), error: %s", task.OperatorAddr, err.Error())
return err
}
_, err = metaNode.Sender.syncSendAdminTask(task)
if err != nil {
log.LogErrorf("action[FreezeEmptyMetaPartition] meta partition(%d), err: %s", mp.PartitionID, err.Error())
return err
}
mr, err := mp.getMetaReplicaLeader()
if err != nil {
log.LogErrorf("get meta replica leader error: %s", err.Error())
return err
}
task := mr.createTaskToFreezeReplica(mp.PartitionID, freeze)
metaNode, err := c.metaNode(task.OperatorAddr)
if err != nil {
log.LogErrorf("failed to get metanode(%s), error: %s", task.OperatorAddr, err.Error())
return err
}
_, err = metaNode.Sender.syncSendAdminTask(task)
if err != nil {
log.LogErrorf("action[FreezeEmptyMetaPartition] meta partition(%d), err: %s", mp.PartitionID, err.Error())
return err
}
return nil

5
metanode/const.go
View File

@ -105,6 +105,8 @@ type (
DeleteMigrationExtentKeyRequest = proto.DeleteMigrationExtentKeyRequest
// Client -> MetaNode
UpdateInodeMetaRequest = proto.UpdateInodeMetaRequest
// Master -> MetaNode
SetFreezeReq = proto.FreezeMetaPartitionRequest
)
// op code should be fixed, order change will cause raft fsm log apply fail
@ -208,6 +210,9 @@ const (
opFSMInternalBatchFreeInodeMigrationExtentKey = 89
opFSMSetInodeCreateTime = 90 // for debug
opFSMSetMigrationExtentKeyDeleteImmediately = 91
// freeze meta partition
opFSMSetFreeze = 92
)
// new inode opCode

11
metanode/manager_op.go
View File

@ -3071,7 +3071,16 @@ func (m *metadataManager) opFreezeEmptyMetaPartition(conn net.Conn, p *Packet,
return
}
mp.SetForbidden(req.Freeze)
if !m.serveProxy(conn, mp, p) {
return
}
err = mp.SetFreeze(req)
if err != nil {
p.PacketErrorWithBody(proto.OpErr, ([]byte)(err.Error()))
m.respondToClientWithVer(conn, p)
return
}
p.PacketOkReply()
m.respondToClientWithVer(conn, p)

23
metanode/partition.go
View File

@ -102,6 +102,7 @@ type MetaPartitionConfig struct {
ConnPool *util.ConnectPool `json:"-"`
Forbidden bool `json:"-"`
ForbidWriteOpOfProtoVer0 bool `json:"ForbidWriteOpOfProtoVer0"`
Freeze bool `json:"freeze"`
}
func (c *MetaPartitionConfig) checkMeta() (err error) {
@ -294,6 +295,7 @@ type MetaPartition interface {
UpdateVolumeView(dataView *proto.DataPartitionsView, volumeView *proto.SimpleVolView)
GetStatByStorageClass() []*proto.StatOfStorageClass
GetMigrateStatByStorageClass() []*proto.StatOfStorageClass
SetFreeze(req *proto.FreezeMetaPartitionRequest) (err error)
}
type UidManager struct {
@ -1936,3 +1938,24 @@ func (mp *metaPartition) CloseAndBackupRaft() (err error) {
err = mp.raftPartition.CloseAndBackup()
return
}
func (mp *metaPartition) SetFreeze(req *proto.FreezeMetaPartitionRequest) (err error) {
mp.config.Freeze = req.Freeze
reqData, err := json.Marshal(*req)
if err != nil {
return
}
r, err := mp.submit(opFSMSetFreeze, reqData)
if err != nil {
return
}
if status := r.(uint8); status != proto.OpOk {
p := &Packet{}
p.ResultCode = status
err = errors.NewErrorf("[SetFreeze]: %s", p.GetResultMsg())
return
}
return nil
}

6
metanode/partition_fsm.go
View File

@ -511,6 +511,12 @@ func (mp *metaPartition) Apply(command []byte, index uint64) (resp interface{},
return
}
err = mp.fsmUpdateInodeMeta(req)
case opFSMSetFreeze:
req := &SetFreezeReq{}
if err = json.Unmarshal(msg.V, req); err != nil {
return
}
resp, err = mp.fsmSetFreeze(req.Freeze)
default:
// do nothing
case opFSMSyncInodeAccessTime:

14
metanode/partition_fsmop.go
View File

@ -269,3 +269,17 @@ func (mp *metaPartition) IsEquareCreateMetaPartitionRequst(request *proto.Create
return
}
func (mp *metaPartition) fsmSetFreeze(freeze bool) (status uint8,
err error,
) {
status = proto.OpOk
oldVal := mp.config.Freeze
mp.config.Freeze = freeze
if err = mp.PersistMetadata(); err != nil {
status = proto.OpDiskErr
mp.config.Freeze = oldVal
}
return
}

38
util/mocktest/raftstore/partition.go
View File

@ -18,10 +18,6 @@ type MockPartition struct {
recorder *MockPartitionMockRecorder
}
func (m *MockPartition) IsRestoring() bool {
return true
}
// MockPartitionMockRecorder is the mock recorder for MockPartition.
type MockPartitionMockRecorder struct {
mock *MockPartition
@ -68,6 +64,20 @@ func (mr *MockPartitionMockRecorder) ChangeMember(changeType, peer, context inte
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ChangeMember", reflect.TypeOf((*MockPartition)(nil).ChangeMember), changeType, peer, context)
}
// CloseAndBackup mocks base method.
func (m *MockPartition) CloseAndBackup() error {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "CloseAndBackup")
ret0, _ := ret[0].(error)
return ret0
}
// CloseAndBackup indicates an expected call of CloseAndBackup.
func (mr *MockPartitionMockRecorder) CloseAndBackup() *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CloseAndBackup", reflect.TypeOf((*MockPartition)(nil).CloseAndBackup))
}
// CommittedIndex mocks base method.
func (m *MockPartition) CommittedIndex() uint64 {
m.ctrl.T.Helper()
@ -82,12 +92,6 @@ func (mr *MockPartitionMockRecorder) CommittedIndex() *gomock.Call {
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CommittedIndex", reflect.TypeOf((*MockPartition)(nil).CommittedIndex))
}
// IsRestoring indicates an expected call of IsRestoring.
func (mr *MockPartitionMockRecorder) IsRestoring() *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "IsRestoring", reflect.TypeOf((*MockPartition)(nil).IsRestoring))
}
// Delete mocks base method.
func (m *MockPartition) Delete() error {
m.ctrl.T.Helper()
@ -130,6 +134,20 @@ func (mr *MockPartitionMockRecorder) IsRaftLeader() *gomock.Call {
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "IsRaftLeader", reflect.TypeOf((*MockPartition)(nil).IsRaftLeader))
}
// IsRestoring mocks base method.
func (m *MockPartition) IsRestoring() bool {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "IsRestoring")
ret0, _ := ret[0].(bool)
return ret0
}
// IsRestoring indicates an expected call of IsRestoring.
func (mr *MockPartitionMockRecorder) IsRestoring() *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "IsRestoring", reflect.TypeOf((*MockPartition)(nil).IsRestoring))
}
// LeaderTerm mocks base method.
func (m *MockPartition) LeaderTerm() (uint64, uint64) {
m.ctrl.T.Helper()