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(dataemeta): cluster switch on snapshot take effect on module handle protocal process

Browse Source 
Signed-off-by: leonrayang <chl696@sina.com>
This commit is contained in:
leonrayang 2024-10-13 21:50:27 +08:00 committed by AmazingChi
parent d4d64fb931
commit 111f9e9903
12 changed files with 74 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
datanode/partition.go
View File

@ -270,16 +270,17 @@ func LoadDataPartition(partitionDir string, disk *Disk) (dp *DataPartition, err
}
dpCfg := &dataPartitionCfg{
VolName: meta.VolumeID,
PartitionSize: meta.PartitionSize,
PartitionType: meta.PartitionType,
PartitionID: meta.PartitionID,
ReplicaNum: meta.ReplicaNum,
Peers: meta.Peers,
Hosts: meta.Hosts,
RaftStore: disk.space.GetRaftStore(),
NodeID: disk.space.GetNodeID(),
ClusterID: disk.space.GetClusterID(),
VolName: meta.VolumeID,
PartitionSize: meta.PartitionSize,
PartitionType: meta.PartitionType,
PartitionID: meta.PartitionID,
ReplicaNum: meta.ReplicaNum,
Peers: meta.Peers,
Hosts: meta.Hosts,
RaftStore: disk.space.GetRaftStore(),
NodeID: disk.space.GetNodeID(),
ClusterID: disk.space.GetClusterID(),
IsEnableSnapshot: disk.space.dataNode.clusterEnableSnapshot,
}
if dp, err = newDataPartition(dpCfg, disk, false); err != nil {
return
@ -409,7 +410,7 @@ func newDataPartition(dpCfg *dataPartitionCfg, disk *Disk, isCreate bool) (dp *D
dp = partition
go partition.statusUpdateScheduler()
go partition.startEvict()
if isCreate {
if isCreate && dpCfg.IsEnableSnapshot {
if err = dp.getVerListFromMaster(); err != nil {
log.LogErrorf("action[newDataPartition] vol %v dp %v loadFromMaster verList failed err %v", dp.volumeID, dp.partitionID, err)
return

3
datanode/partition_op_by_raft.go
View File

@ -340,6 +340,9 @@ func (dp *DataPartition) Submit(val []byte) (retCode uint8, err error) {
}
func (dp *DataPartition) CheckWriteVer(p *repl.Packet) (err error) {
if !dp.config.IsEnableSnapshot {
return
}
log.LogDebugf("action[CheckWriteVer] packet %v dpseq %v ", p, dp.verSeq)
if atomic.LoadUint64(&dp.verSeq) == p.VerSeq {
return

1
datanode/partition_raft.go
View File

@ -52,6 +52,7 @@ type dataPartitionCfg struct {
CreateType int
Forbidden bool
DpRepairBlockSize uint64
IsEnableSnapshot bool
}
func (dp *DataPartition) raftPort() (heartbeat, replica int, err error) {

10
datanode/server.go
View File

@ -191,10 +191,11 @@ type DataNode struct {
dpMaxRepairErrCnt uint64
clusterUuid string
clusterUuidEnable bool
serviceIDKey string
cpuUtil atomicutil.Float64
cpuSamplerDone chan struct{}
// dpRepairTimeOut uint64
clusterEnableSnapshot bool
serviceIDKey string
cpuUtil atomicutil.Float64
cpuSamplerDone chan struct{}
enableGcTimer bool
gcTimer *util.RecycleTimer
@ -634,6 +635,7 @@ func (s *DataNode) register(cfg *config.Config) {
masterAddr := MasterClient.Leader()
s.clusterUuid = ci.ClusterUuid
s.clusterUuidEnable = ci.ClusterUuidEnable
s.clusterEnableSnapshot = ci.ClusterEnableSnapshot
s.clusterID = ci.Cluster
if LocalIP == "" {
LocalIP = string(ci.Ip)

27
datanode/space_manager.go
View File

@ -491,19 +491,20 @@ func (manager *SpaceManager) DetachDataPartition(partitionID uint64) {
func (manager *SpaceManager) CreatePartition(request *proto.CreateDataPartitionRequest) (dp *DataPartition, err error) {
dpCfg := &dataPartitionCfg{
PartitionID: request.PartitionId,
VolName: request.VolumeId,
Peers: request.Members,
Hosts: request.Hosts,
RaftStore: manager.raftStore,
NodeID: manager.nodeID,
ClusterID: manager.clusterID,
PartitionSize: request.PartitionSize,
PartitionType: int(request.PartitionTyp),
ReplicaNum: request.ReplicaNum,
VerSeq: request.VerSeq,
CreateType: request.CreateType,
Forbidden: false,
PartitionID: request.PartitionId,
VolName: request.VolumeId,
Peers: request.Members,
Hosts: request.Hosts,
RaftStore: manager.raftStore,
NodeID: manager.nodeID,
ClusterID: manager.clusterID,
PartitionSize: request.PartitionSize,
PartitionType: int(request.PartitionTyp),
ReplicaNum: request.ReplicaNum,
VerSeq: request.VerSeq,
CreateType: request.CreateType,
Forbidden: false,
IsEnableSnapshot: manager.dataNode.clusterEnableSnapshot,
}
log.LogInfof("action[CreatePartition] dp %v dpCfg.Peers %v request.Members %v",
dpCfg.PartitionID, dpCfg.Peers, request.Members)

5
datanode/wrap_operator.go
View File

@ -419,7 +419,10 @@ func (s *DataNode) handleUpdateVerPacket(p *repl.Packet) {
p.PacketOkReply()
}
}()
if !s.clusterEnableSnapshot {
err = fmt.Errorf("cluster not enable snapshot!")
return
}
task := &proto.AdminTask{}
err = json.Unmarshal(p.Data, task)
if err != nil {

11
master/api_service.go
View File

@ -953,11 +953,12 @@ func (m *Server) getIPAddr(w http.ResponseWriter, r *http.Request) {
DpMaxRepairErrCnt: dpMaxRepairErrCnt,
DirChildrenNumLimit: dirChildrenNumLimit,
// Ip: strings.Split(r.RemoteAddr, ":")[0],
Ip: iputil.RealIP(r),
EbsAddr: m.bStoreAddr,
ServicePath: m.servicePath,
ClusterUuid: m.cluster.clusterUuid,
ClusterUuidEnable: m.cluster.clusterUuidEnable,
Ip: iputil.RealIP(r),
EbsAddr: m.bStoreAddr,
ServicePath: m.servicePath,
ClusterUuid: m.cluster.clusterUuid,
ClusterUuidEnable: m.cluster.clusterUuidEnable,
ClusterEnableSnapshot: m.cluster.cfg.EnableSnapshot,
}
sendOkReply(w, r, newSuccessHTTPReply(cInfo))

22
metanode/manager_op.go
View File

@ -2598,6 +2598,9 @@ func (m *metadataManager) commitCreateVersion(VolumeID string, VerSeq uint64, Op
}
func (m *metadataManager) updatePackRspSeq(mp MetaPartition, p *Packet) {
if !m.metaNode.clusterEnableSnapshot {
return
}
if mp.GetVerSeq() > p.VerSeq {
log.LogDebugf("action[checkmultiSnap.multiVersionstatus] mp ver [%v], packet ver [%v]", mp.GetVerSeq(), p.VerSeq)
p.VerSeq = mp.GetVerSeq() // used to response to client and try update verSeq of client
@ -2608,6 +2611,9 @@ func (m *metadataManager) updatePackRspSeq(mp MetaPartition, p *Packet) {
}
func (m *metadataManager) checkMultiVersionStatus(mp MetaPartition, p *Packet) (err error) {
if !m.metaNode.clusterEnableSnapshot {
return
}
if (p.ExtentType&proto.MultiVersionFlag == 0) && mp.GetVerSeq() > 0 {
log.LogWarnf("action[checkmultiSnap.multiVersionstatus] volname [%v] mp ver [%v], client use old ver before snapshot", mp.GetVolName(), mp.GetVerSeq())
return fmt.Errorf("client use old ver before snapshot")
@ -2697,21 +2703,26 @@ func (m *metadataManager) opMultiVersionOp(conn net.Conn, p *Packet,
remoteAddr string,
) (err error) {
// For ack to master
data := p.Data
m.responseAckOKToMaster(conn, p)
var (
opAgain bool
start = time.Now()
data = p.Data
req = &proto.MultiVersionOpRequest{}
resp = &proto.MultiVersionOpResponse{}
adminTask = &proto.AdminTask{
Request: req,
}
opAgain bool
decode = json.NewDecoder(bytes.NewBuffer(data))
)
log.LogDebugf("action[opMultiVersionOp] volume %v op [%v]", req.VolumeID, req.Op)
start := time.Now()
decode := json.NewDecoder(bytes.NewBuffer(data))
if !m.metaNode.clusterEnableSnapshot {
err = fmt.Errorf("cluster not EnableSnapshot")
log.LogErrorf("opMultiVersionOp volume %v", err)
goto end
}
decode.UseNumber()
if err = decode.Decode(adminTask); err != nil {
resp.Status = proto.TaskFailed
@ -2719,6 +2730,7 @@ func (m *metadataManager) opMultiVersionOp(conn net.Conn, p *Packet,
log.LogErrorf("action[opMultiVersionOp] %v mp err %v do Decoder", req.VolumeID, err.Error())
goto end
}
log.LogDebugf("action[opMultiVersionOp] volume %v op [%v]", req.VolumeID, req.Op)
resp.Status = proto.TaskSucceeds
resp.VolumeID = req.VolumeID

2
metanode/metanode.go
View File

@ -72,6 +72,7 @@ type MetaNode struct {
connectionCnt int64
clusterUuid string
clusterUuidEnable bool
clusterEnableSnapshot bool
serviceIDKey string
control common.Control
@ -461,6 +462,7 @@ func (m *MetaNode) register() (err error) {
}
m.clusterUuid = clusterInfo.ClusterUuid
m.clusterUuidEnable = clusterInfo.ClusterUuidEnable
m.clusterEnableSnapshot = clusterInfo.ClusterEnableSnapshot
m.clusterId = clusterInfo.Cluster
nodeAddress = m.localAddr + ":" + m.listen
step++

4
metanode/nodeinfo.go
View File

@ -57,7 +57,9 @@ func (m *MetaNode) startUpdateNodeInfo() {
return
case <-ticker.C:
m.updateNodeInfo()
m.metadataManager.checkVolVerList()
if m.clusterEnableSnapshot {
m.metadataManager.checkVolVerList()
}
}
}
}

5
metanode/partition_fsm.go
View File

@ -491,6 +491,11 @@ func (mp *metaPartition) runVersionOp() {
}
func (mp *metaPartition) fsmVersionOp(reqData []byte) (err error) {
if mp.manager != nil && mp.manager.metaNode != nil && !mp.manager.metaNode.clusterEnableSnapshot {
err = fmt.Errorf("clusterEnableSnapshot not enabled")
log.LogErrorf("action[fsmVersionOp] mp[%v] err %v", mp.config.PartitionId, err)
return
}
mp.multiVersionList.RWLock.Lock()
defer mp.multiVersionList.RWLock.Unlock()

1
proto/admin_proto.go
View File

@ -560,6 +560,7 @@ type ClusterInfo struct {
ServicePath string
ClusterUuid string
ClusterUuidEnable bool
ClusterEnableSnapshot bool
}
// CreateDataPartitionRequest defines the request to create a data partition.