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(datanode): check disk path for replica with master when executing attachPartition

Browse Source 
Signed-off-by: chihe <chihe@oppo.com>
This commit is contained in:
chihe 2024-06-20 10:03:38 +08:00 committed by longerfly
parent 9dd4c9cf1b
commit 2911125677
19 changed files with 119 additions and 364 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
client/fuse.go
View File

@ -83,11 +83,11 @@ const (
)
const (
LoggerDir = "client"
LoggerPrefix = "client"
LoggerOutput = "output.log"
ModuleName = "fuseclient"
ConfigKeyExporterPort = "exporterKey"
// LoggerDir = "client"
LoggerPrefix = "client"
LoggerOutput = "output.log"
ModuleName = "fuseclient"
// ConfigKeyExporterPort = "exporterKey"
ControlCommandSetRate = "/rate/set"
ControlCommandGetRate = "/rate/get"

9
datanode/partition.go
View File

@ -687,10 +687,10 @@ func (dp *DataPartition) ForceLoadHeader() {
dp.loadExtentHeaderStatus = FinishLoadDataPartitionExtentHeader
}
func (dp *DataPartition) RemoveAll(decommissionType uint32, force, isSpecialReplica bool) (err error) {
func (dp *DataPartition) RemoveAll(decommissionType uint32, force bool) (err error) {
dp.persistMetaMutex.Lock()
defer dp.persistMetaMutex.Unlock()
if force && isSpecialReplica && decommissionType == proto.AutoDecommission {
if force && decommissionType == proto.AutoDecommission {
originalPath := dp.Path()
parent := path.Dir(originalPath)
fileName := path.Base(originalPath)
@ -1497,12 +1497,11 @@ func (dp *DataPartition) reload(s *SpaceManager) error {
dp.Stop()
dp.Disk().DetachDataPartition(dp)
log.LogDebugf("data partition %v is detached", dp.partitionID)
dp2, err := LoadDataPartition(rootDir, disk)
_, err := LoadDataPartition(rootDir, disk)
if err != nil {
return err
}
s.AttachPartition(dp2)
return err
return nil
}
func (dp *DataPartition) resetDiskErrCnt() {

147
datanode/server_handler.go
View File

@ -557,148 +557,6 @@ func (s *DataNode) reloadDataPartition(w http.ResponseWriter, r *http.Request) {
}
}
func (s *DataNode) setDiskExtentReadLimitStatus(w http.ResponseWriter, r *http.Request) {
var status common.Bool
if err := parseArgs(r, status.Status()); err != nil {
s.buildFailureResp(w, http.StatusBadRequest, err.Error())
return
}
for _, disk := range s.space.disks {
disk.SetExtentRepairReadLimitStatus(status.V)
}
s.buildSuccessResp(w, "success")
}
type DiskExtentReadLimitInfo struct {
DiskPath string `json:"diskPath"`
ExtentReadLimitStatus bool `json:"extentReadLimitStatus"`
Dp uint64 `json:"dp"`
}
type DiskExtentReadLimitStatusResponse struct {
Infos []DiskExtentReadLimitInfo `json:"infos"`
}
func (s *DataNode) queryDiskExtentReadLimitStatus(w http.ResponseWriter, r *http.Request) {
resp := &DiskExtentReadLimitStatusResponse{}
for _, disk := range s.space.disks {
status, dp := disk.QueryExtentRepairReadLimitStatus()
resp.Infos = append(resp.Infos, DiskExtentReadLimitInfo{DiskPath: disk.Path, ExtentReadLimitStatus: status, Dp: dp})
}
s.buildSuccessResp(w, resp)
}
func (s *DataNode) detachDataPartition(w http.ResponseWriter, r *http.Request) {
var pid common.Uint
if err := parseArgs(r, pid.ID()); err != nil {
err = fmt.Errorf("parse form fail: %v", err)
s.buildFailureResp(w, http.StatusBadRequest, err.Error())
return
}
partitionID := pid.V
partition := s.space.Partition(partitionID)
if partition == nil {
s.buildFailureResp(w, http.StatusBadRequest, "partition not exist")
return
}
// store disk path and root of dp
disk := partition.disk
rootDir := partition.path
log.LogDebugf("data partition disk %v rootDir %v", disk, rootDir)
s.space.partitionMutex.Lock()
delete(s.space.partitions, partitionID)
s.space.partitionMutex.Unlock()
partition.Stop()
partition.Disk().DetachDataPartition(partition)
log.LogDebugf("data partition %v is detached", partitionID)
s.buildSuccessResp(w, "success")
}
func (s *DataNode) releaseDiskExtentReadLimitToken(w http.ResponseWriter, r *http.Request) {
var diskParam common.String
if err := parseArgs(r, diskParam.Disk()); err != nil {
err = fmt.Errorf("parse form fail: %v", err)
s.buildFailureResp(w, http.StatusBadRequest, err.Error())
return
}
diskPath := diskParam.V
// store disk path and root of dp
disk, err := s.space.GetDisk(diskPath)
if err != nil {
log.LogErrorf("action[loadDataPartition] disk(%v) is not found err(%v).", diskPath, err)
s.buildFailureResp(w, http.StatusBadRequest, fmt.Sprintf("disk %v is not found", diskPath))
return
}
disk.ReleaseReadExtentToken()
s.buildSuccessResp(w, "success")
}
func (s *DataNode) loadDataPartition(w http.ResponseWriter, r *http.Request) {
var pid common.Uint
var diskParam common.String
if err := parseArgs(r, pid.ID(), diskParam.Disk()); err != nil {
err = fmt.Errorf("parse form fail: %v", err)
s.buildFailureResp(w, http.StatusBadRequest, err.Error())
return
}
partitionID := pid.V
diskPath := diskParam.V
partition := s.space.Partition(partitionID)
if partition != nil {
s.buildFailureResp(w, http.StatusBadRequest, "partition is already loaded")
return
}
// store disk path and root of dp
disk, err := s.space.GetDisk(diskPath)
if err != nil {
log.LogErrorf("action[loadDataPartition] disk(%v) is not found err(%v).", diskPath, err)
s.buildFailureResp(w, http.StatusBadRequest, fmt.Sprintf("disk %v is not found", diskPath))
return
}
fileInfoList, err := os.ReadDir(disk.Path)
if err != nil {
log.LogErrorf("action[loadDataPartition] read dir(%v) err(%v).", disk.Path, err)
s.buildFailureResp(w, http.StatusBadRequest, fmt.Sprintf(" read dir(%v) err(%v)", disk.Path, err))
return
}
rootDir := ""
for _, fileInfo := range fileInfoList {
filename := fileInfo.Name()
if !disk.isPartitionDir(filename) {
continue
}
if id, _, err := unmarshalPartitionName(filename); err != nil {
log.LogErrorf("action[RestorePartition] unmarshal partitionName(%v) from disk(%v) err(%v) ",
filename, disk.Path, err.Error())
continue
} else {
if id == partitionID {
rootDir = filename
}
}
}
if rootDir == "" {
log.LogErrorf("action[loadDataPartition] dp root not found in dir(%v) .", disk.Path)
s.buildFailureResp(w, http.StatusBadRequest, fmt.Sprintf("dp root not found in dir(%v)", disk.Path))
return
}
log.LogDebugf("data partition disk %v rootDir %v", disk, rootDir)
_, err = LoadDataPartition(path.Join(diskPath, rootDir), disk)
if err != nil {
s.buildFailureResp(w, http.StatusBadRequest, err.Error())
} else {
s.buildSuccessResp(w, "success")
}
}
// NOTE: use for test
func (s *DataNode) markDataPartitionBroken(w http.ResponseWriter, r *http.Request) {
const (
@ -904,8 +762,6 @@ func (s *DataNode) loadDataPartition(w http.ResponseWriter, r *http.Request) {
for _, fileInfo := range fileInfoList {
filename := fileInfo.Name()
if !disk.isPartitionDir(filename) {
if disk.isExpiredPartitionDir(filename) {
}
continue
}
@ -927,11 +783,10 @@ func (s *DataNode) loadDataPartition(w http.ResponseWriter, r *http.Request) {
log.LogDebugf("data partition disk %v rootDir %v", disk, rootDir)
dp, err := LoadDataPartition(path.Join(diskPath, rootDir), disk)
_, err = LoadDataPartition(path.Join(diskPath, rootDir), disk)
if err != nil {
s.buildFailureResp(w, http.StatusBadRequest, err.Error())
} else {
s.space.AttachPartition(dp)
s.buildSuccessResp(w, "success")
}
}

178
datanode/space_manager.go
View File

@ -40,27 +40,25 @@ const DefaultStopDpLimit = 4
// SpaceManager manages the disk space.
type SpaceManager struct {
clusterID string
disks map[string]*Disk
partitions map[uint64]*DataPartition
raftStore raftstore.RaftStore
nodeID uint64
diskMutex sync.RWMutex
partitionMutex sync.RWMutex
stats *Stats
stopC chan bool
selectedIndex int // TODO what is selected index
diskList []string
dataNode *DataNode
createPartitionMutex sync.RWMutex
rand *rand.Rand
currentLoadDpCount int
currentStopDpCount int
diskUtils map[string]*atomicutil.Float64
samplerDone chan struct{}
allDisksLoaded bool
dataNodeIDs map[string]uint64
dataNodeIDsMutex sync.RWMutex
clusterID string
disks map[string]*Disk
partitions map[uint64]*DataPartition
raftStore raftstore.RaftStore
nodeID uint64
diskMutex sync.RWMutex
partitionMutex sync.RWMutex
stats *Stats
stopC chan bool
diskList []string
dataNode *DataNode
rand *rand.Rand
currentLoadDpCount int
currentStopDpCount int
diskUtils map[string]*atomicutil.Float64
samplerDone chan struct{}
allDisksLoaded bool
dataNodeIDs map[string]uint64
dataNodeIDsMutex sync.RWMutex
}
const diskSampleDuration = 1 * time.Second
@ -293,70 +291,7 @@ func (manager *SpaceManager) LoadDisk(path string, reservedSpace, diskRdonlySpac
log.LogDebugf("action[LoadDisk] load disk from path(%v).", path)
visitor = func(dp *DataPartition) {
manager.partitionMutex.Lock()
if loadedDp, has := manager.partitions[dp.partitionID]; !has {
manager.partitions[dp.partitionID] = dp
manager.partitionMutex.Unlock()
log.LogDebugf("action[LoadDisk] put partition(%v) to manager.", dp.partitionID)
} else {
manager.partitionMutex.Unlock()
if loadedDp.disk.Path == dp.disk.Path {
log.LogWarnf("[LoadDisk] dp(%v) is loaded, to load(%v), but disk path is the same",
loadedDp.info(), dp.info())
return
}
log.LogWarnf("action[LoadDisk] dp(%v) is loaded, to load(%v).", loadedDp.info(), dp.info())
_, _, infos, err := dp.fetchReplicasFromMaster()
if err != nil {
manager.DetachDataPartition(loadedDp.partitionID)
loadedDp.Stop()
loadedDp.Disk().DetachDataPartition(loadedDp)
log.LogErrorf("action[LoadDisk] dp(%v) is detached,due to get dp info failed(%v).",
loadedDp.info(), err)
} else {
var correctReplic ReplicaInfo
for _, replica := range infos {
if replica.Addr == manager.dataNode.localServerAddr {
correctReplic = replica
break
}
}
if correctReplic.Disk == "" && correctReplic.Addr == "" {
loadedDp.Stop()
loadedDp.Disk().DetachDataPartition(loadedDp)
log.LogErrorf("action[LoadDisk] dp(%v) is detached,due to data node not contains in replicas"+
"from master.", loadedDp.info())
} else {
if loadedDp.disk.Path == correctReplic.Disk {
dp.Stop()
dp.Disk().DetachDataPartition(dp)
if err := dp.RemoveAll(proto.InitialDecommission, false, false); err != nil {
log.LogErrorf("action[LoadDisk]failed to remove dp(%v) dir(%v), err(%v)",
dp.partitionID, dp.Path(), err)
}
} else {
// detach loaded dp
loadedDp.Stop()
loadedDp.Disk().DetachDataPartition(loadedDp)
if err := loadedDp.RemoveAll(proto.InitialDecommission, false, false); err != nil {
log.LogErrorf("action[LoadDisk]failed to remove dp(%v) dir(%v), err(%v)",
loadedDp.partitionID, loadedDp.Path(), err)
}
dp.Stop()
dp.Disk().DetachDataPartition(dp)
dp2, err := LoadDataPartition(dp.path, dp.disk)
if err != nil {
log.LogErrorf("action[LoadDisk]failed to load dp %v (%v_%v), err(%v)",
dp.partitionID, dp.path, dp.disk, err)
} else {
manager.AttachPartition(dp2)
}
}
}
}
}
// do noting here, dp is attached to space manager in RestorePartition
}
if _, err = manager.GetDisk(path); err != nil {
@ -503,8 +438,67 @@ func (manager *SpaceManager) AttachPartition(dp *DataPartition) {
log.LogInfof("[AttachPartition] load dp(%v) attach using time(%v)", dp.partitionID, time.Since(begin))
}()
manager.partitionMutex.Lock()
defer manager.partitionMutex.Unlock()
manager.partitions[dp.partitionID] = dp
if loadedDp, has := manager.partitions[dp.partitionID]; !has {
manager.partitions[dp.partitionID] = dp
manager.partitionMutex.Unlock()
log.LogDebugf("action[AttachPartition] put partition(%v) to manager.", dp.partitionID)
} else {
manager.partitionMutex.Unlock()
if loadedDp.disk.Path == dp.disk.Path {
log.LogWarnf("[AttachPartition] dp(%v) is loaded, to load(%v), but disk path is the same",
loadedDp.info(), dp.info())
return
}
log.LogWarnf("action[AttachPartition] dp(%v) is loaded, to load(%v).", loadedDp.info(), dp.info())
_, _, infos, err := dp.fetchReplicasFromMaster()
if err != nil {
manager.DetachDataPartition(loadedDp.partitionID)
loadedDp.Stop()
loadedDp.Disk().DetachDataPartition(loadedDp)
log.LogErrorf("action[LoadDisk] dp(%v) is detached,due to get dp info failed(%v).",
loadedDp.info(), err)
} else {
var correctReplica ReplicaInfo
for _, replica := range infos {
if replica.Addr == manager.dataNode.localServerAddr {
correctReplica = replica
break
}
}
if correctReplica.Disk == "" && correctReplica.Addr == "" {
loadedDp.Stop()
loadedDp.Disk().DetachDataPartition(loadedDp)
log.LogErrorf("action[LoadDisk] dp(%v) is detached,due to data node not contains in replicas"+
"from master.", loadedDp.info())
} else {
if loadedDp.disk.Path == correctReplica.Disk {
dp.Stop()
dp.Disk().DetachDataPartition(dp)
if err := dp.RemoveAll(proto.InitialDecommission, true); err != nil {
log.LogErrorf("action[LoadDisk]failed to remove dp(%v) dir(%v), err(%v)",
dp.partitionID, dp.Path(), err)
}
} else {
// detach loaded dp
loadedDp.Stop()
loadedDp.Disk().DetachDataPartition(loadedDp)
if err := loadedDp.RemoveAll(proto.InitialDecommission, true); err != nil {
log.LogErrorf("action[LoadDisk]failed to remove dp(%v) dir(%v), err(%v)",
loadedDp.partitionID, loadedDp.Path(), err)
}
dp.Stop()
dp.Disk().DetachDataPartition(dp)
_, err := LoadDataPartition(dp.path, dp.disk)
if err != nil {
log.LogErrorf("action[LoadDisk]failed to load dp %v (%v_%v), err(%v)",
dp.partitionID, dp.path, dp.disk, err)
}
}
}
}
}
}
// DetachDataPartition removes a data partition from the partition map.
@ -554,14 +548,14 @@ func (manager *SpaceManager) CreatePartition(request *proto.CreateDataPartitionR
}
// DeletePartition deletes a partition based on the partition id.
func (manager *SpaceManager) DeletePartition(dpID uint64, decommissionType uint32, force, isSpecialReplica bool) (err error) {
func (manager *SpaceManager) DeletePartition(dpID uint64, decommissionType uint32, force bool) (err error) {
manager.partitionMutex.Lock()
dp := manager.partitions[dpID]
if dp == nil {
manager.partitionMutex.Unlock()
// maybe dp not loaded when triggered disk error, need to remove disk root dir
err = manager.deleteDataPartitionNotLoaded(dpID, decommissionType, force, isSpecialReplica)
err = manager.deleteDataPartitionNotLoaded(dpID, decommissionType, force)
return err
}
@ -569,7 +563,7 @@ func (manager *SpaceManager) DeletePartition(dpID uint64, decommissionType uint3
manager.partitionMutex.Unlock()
dp.Stop()
dp.Disk().DetachDataPartition(dp)
if err := dp.RemoveAll(decommissionType, force, isSpecialReplica); err != nil {
if err := dp.RemoveAll(decommissionType, force); err != nil {
return err
}
return nil
@ -637,7 +631,7 @@ func (s *DataNode) buildHeartBeatResponse(response *proto.DataNodeHeartbeatRespo
}
}
func (manager *SpaceManager) deleteDataPartitionNotLoaded(id uint64, decommissionType uint32, force, isSpecialReplica bool) error {
func (manager *SpaceManager) deleteDataPartitionNotLoaded(id uint64, decommissionType uint32, force bool) error {
disks := manager.GetDisks()
for _, d := range disks {
if d.HasDiskErrPartition(id) {
@ -665,7 +659,7 @@ func (manager *SpaceManager) deleteDataPartitionNotLoaded(id uint64, decommissio
} else {
if partitionID == id {
rootPath := path.Join(d.Path, filename)
if decommissionType == proto.AutoDecommission && isSpecialReplica && force {
if decommissionType == proto.AutoDecommission && force {
newPath := path.Join(d.Path, BackupPartitionPrefix+filename)
err = os.Rename(rootPath, newPath)
if err == nil {

5
datanode/wrap_operator.go
View File

@ -623,7 +623,7 @@ func (s *DataNode) handlePacketToDeleteDataPartition(p *repl.Packet) {
if err != nil {
return
} else {
err = s.space.DeletePartition(request.PartitionId, request.DecommissionType, request.Force, request.IsSpecialReplica)
err = s.space.DeletePartition(request.PartitionId, request.DecommissionType, request.Force)
}
} else {
err = fmt.Errorf("illegal opcode ")
@ -1845,12 +1845,11 @@ func (s *DataNode) handlePacketToRecoverBackupDataReplica(p *repl.Packet) {
return
}
dp, err := LoadDataPartition(path.Join(request.Disk, newPath), disk)
_, err = LoadDataPartition(path.Join(request.Disk, newPath), disk)
if err != nil {
log.LogErrorf("action[handlePacketToRecoverBackupDataReplica] load disk %v rootDir %v failed err %v.",
disk.Path, rootDir, err)
} else {
s.space.AttachPartition(dp)
log.LogInfof("action[handlePacketToRecoverBackupDataReplica] load disk %v rootDir %v success .",
disk.Path, rootDir)
}

10
master/api_service.go
View File

@ -18,7 +18,6 @@ import (
"context"
"encoding/json"
"fmt"
"github.com/cubefs/cubefs/util/auditlog"
"io"
"math"
"net/http"
@ -32,6 +31,7 @@ import (
"github.com/cubefs/cubefs/proto"
"github.com/cubefs/cubefs/util"
"github.com/cubefs/cubefs/util/auditlog"
"github.com/cubefs/cubefs/util/compressor"
"github.com/cubefs/cubefs/util/cryptoutil"
"github.com/cubefs/cubefs/util/errors"
@ -1639,6 +1639,7 @@ func (m *Server) addDataReplica(w http.ResponseWriter, r *http.Request) {
return
}
time.Sleep(1 * time.Second)
continue
}
break
}
@ -5448,10 +5449,10 @@ func (m *Server) GetAllVersionInfo(w http.ResponseWriter, r *http.Request) {
sendErrReply(w, r, newErrHTTPReply(proto.ErrVolNotExists))
return
}
//if !proto.IsHot(vol.VolType) {
// if !proto.IsHot(vol.VolType) {
// sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeVersionOpError, Msg: "vol need be hot one"})
// return
//}
// }
verList = vol.VersionMgr.getVersionList()
@ -7075,7 +7076,6 @@ func (m *Server) cancelDisableDisk(w http.ResponseWriter, r *http.Request) {
dp.setRestoreReplicaStop()
auditlog.LogMasterOp("CancelDataPartitionDecommission", msg, nil)
}
//m.cluster.DecommissionDisks.Delete(key)
disk.SetDecommissionStatus(DecommissionCancel)
rstMsg := fmt.Sprintf("cancel decommission disk[%s] successfully ", key)
sendOkReply(w, r, newSuccessHTTPReply(rstMsg))
@ -7170,12 +7170,12 @@ func (m *Server) recoverDiskErrorReplica(w http.ResponseWriter, r *http.Request)
if !dp.setRestoreReplicaForbidden() {
retry++
if retry > defaultDecommissionRetryLimit {
} else {
err = errors.NewErrorf("set RestoreReplicaMetaForbidden failed")
sendErrReply(w, r, newErrHTTPReply(err))
return
}
time.Sleep(1 * time.Second)
continue
}
break
}

4
master/cluster.go
View File

@ -1773,7 +1773,6 @@ func (c *Cluster) getHostFromNormalZone(nodeType uint32, excludeZones []string,
zoneNumNeed int, specifiedZoneName string) (hosts []string, peers []proto.Peer, err error,
) {
var zonesQualified []*Zone
zonesQualified = make([]*Zone, 0)
if replicaNum <= zoneNumNeed {
zoneNumNeed = replicaNum
}
@ -5037,8 +5036,7 @@ func (c *Cluster) syncRecoverBackupDataPartitionReplica(host, disk string, dp *D
if err != nil {
return
}
var task *proto.AdminTask
task = dp.createTaskToRecoverBackupDataPartitionReplica(host, disk)
task := dp.createTaskToRecoverBackupDataPartitionReplica(host, disk)
if _, err = dataNode.TaskManager.syncSendAdminTask(task); err != nil {
return
}

11
master/data_node.go
View File

@ -260,13 +260,6 @@ func (dataNode *DataNode) isWriteAbleWithSizeNoLock(size uint64) (ok bool) {
return
}
func (dataNode *DataNode) isWriteAbleWithSize(size uint64) (ok bool) {
dataNode.RLock()
defer dataNode.RUnlock()
return dataNode.isWriteAbleWithSizeNoLock(size)
}
func (dataNode *DataNode) GetUsed() uint64 {
dataNode.RLock()
defer dataNode.RUnlock()
@ -591,7 +584,7 @@ func (dataNode *DataNode) delDecommissionDiskFromCache(c *Cluster) {
func (dataNode *DataNode) getBackupDataPartitionIDs() (ids []uint64) {
dataNode.RLock()
dataNode.RUnlock()
defer dataNode.RUnlock()
ids = make([]uint64, 0)
for _, info := range dataNode.BackupDataPartitions {
ids = append(ids, info.PartitionID)
@ -601,7 +594,7 @@ func (dataNode *DataNode) getBackupDataPartitionIDs() (ids []uint64) {
func (dataNode *DataNode) getBackupDataPartitionInfo(id uint64) (proto.BackupDataPartitionInfo, error) {
dataNode.RLock()
dataNode.RUnlock()
defer dataNode.RUnlock()
for _, info := range dataNode.BackupDataPartitions {
if info.PartitionID == id {
return info, nil

8
master/data_partition.go
View File

@ -295,7 +295,7 @@ func (partition *DataPartition) createTaskToCreateDataPartition(addr string, dat
func (partition *DataPartition) createTaskToDeleteDataPartition(addr string, raftForceDel bool) (task *proto.AdminTask) {
task = proto.NewAdminTask(proto.OpDeleteDataPartition, addr,
newDeleteDataPartitionRequest(partition.PartitionID, partition.DecommissionType, raftForceDel, partition.isSpecialReplicaCnt()))
newDeleteDataPartitionRequest(partition.PartitionID, partition.DecommissionType, raftForceDel))
partition.resetTaskID(task)
return
}
@ -1143,6 +1143,7 @@ directly:
}
// wait for checkReplicaMeta ended
time.Sleep(3 * time.Second)
continue
}
break
}
@ -1163,6 +1164,7 @@ directly:
}
// wait for checkReplicaMeta ended
time.Sleep(3 * time.Second)
continue
}
break
}
@ -2026,7 +2028,7 @@ func (partition *DataPartition) checkReplicaMeta(c *Cluster) (err error) {
}
}
//if len(partition.Peers) == int(partition.ReplicaNum) && len(partition.Peers) > len(partition.Replicas) {
// if len(partition.Peers) == int(partition.ReplicaNum) && len(partition.Peers) > len(partition.Replicas) {
// for _, peer := range partition.Peers {
// found := false
// for _, replica := range partition.Replicas {
@ -2048,7 +2050,7 @@ func (partition *DataPartition) checkReplicaMeta(c *Cluster) (err error) {
// return
// }
// }
//}
// }
// find redundant peers from replica meta
force := false
for _, replica := range partition.Replicas {

45
master/data_partition_check.go
View File

@ -152,51 +152,6 @@ func (partition *DataPartition) checkLeader(c *Cluster, clusterID string, timeOu
WarnMetrics.WarnDpNoLeader(clusterID, partition.PartitionID, partition.ReplicaNum, report)
}
partition.Unlock()
// if report && partition.ReplicaNum == 1 && partition.GetDecommissionStatus() == DecommissionInitial &&
// len(partition.Hosts) == 1 && len(partition.Peers) == 1 {
// replica := partition.Replicas[0]
// // heart beat is not arrived
// if len(replica.LocalPeers) == 0 {
// return
// }
// redundantPeers := findRedundantPeers(partition.Hosts, replica.LocalPeers)
// log.LogInfof("action[checkLeader] partition %v try to recover leader local peers(%v) redundantPeers(%v)",
// partition.PartitionID, replica.LocalPeers, redundantPeers)
// for _, peer := range redundantPeers {
// log.LogInfof("action[checkLeader] partition %v remove peer (%v)",
// partition.PartitionID, peer)
// dataNode, err := c.dataNode(peer)
// if err != nil {
// log.LogInfof("action[checkLeader] partition %v find data node for peer (%v) failed %v",
// partition.PartitionID, peer, err)
// continue
// }
// removePeer := proto.Peer{ID: dataNode.ID, Addr: peer}
// if err := c.removeDataPartitionRaftMember(partition, removePeer, true); err != nil {
// log.LogInfof("action[checkLeader] partition %v remove peer (%v) failed %v",
// partition.PartitionID, peer, err)
// continue
// }
// }
// }
return
}
func findRedundantPeers(a []string, b []proto.Peer) []string {
var redundantPeers []string
for _, item := range b {
found := false
for _, value := range a {
if item.Addr == value {
found = true
break
}
}
if !found {
redundantPeers = append(redundantPeers, item.Addr)
}
}
return redundantPeers
}
// Check if there is any missing replica for a data partition.

3
master/operate_util.go
View File

@ -49,12 +49,11 @@ func newCreateDataPartitionRequest(volName string, ID uint64, replicaNum int, me
return
}
func newDeleteDataPartitionRequest(ID uint64, decommissionType uint32, raftForceDel, isSpecialReplica bool) (req *proto.DeleteDataPartitionRequest) {
func newDeleteDataPartitionRequest(ID uint64, decommissionType uint32, raftForceDel bool) (req *proto.DeleteDataPartitionRequest) {
req = &proto.DeleteDataPartitionRequest{
PartitionId: ID,
DecommissionType: decommissionType,
Force: raftForceDel,
IsSpecialReplica: isSpecialReplica,
}
return
}

40
master/topology.go
View File

@ -30,10 +30,10 @@ import (
)
type rsManager struct {
nodeType NodeType
// nodeType NodeType
nodes *sync.Map
zoneIndexForNode int
zones []*Zone
// zones []*Zone
}
func (rsm *rsManager) clear() {
@ -58,12 +58,6 @@ func newTopology() (t *topology) {
return
}
func (t *topology) getZoneLen() int {
t.zoneLock.RLock()
defer t.zoneLock.RUnlock()
return len(t.zones)
}
func (t *topology) zoneLen() int {
t.zoneLock.RLock()
defer t.zoneLock.RUnlock()
@ -149,15 +143,6 @@ func (t *topology) deleteDataNode(dataNode *DataNode) {
t.dataTopology.nodes.Delete(dataNode.Addr)
}
func (t *topology) getZoneByDataNode(dataNode *DataNode) (zone *Zone, err error) {
_, ok := t.dataTopology.nodes.Load(dataNode.Addr)
if !ok {
return nil, errors.Trace(dataNodeNotFound(dataNode.Addr), "%v not found", dataNode.Addr)
}
return t.getZone(dataNode.ZoneName)
}
func (t *topology) putMetaNode(metaNode *MetaNode) (err error) {
if _, ok := t.metaTopology.nodes.Load(metaNode.Addr); ok {
return
@ -1296,7 +1281,7 @@ func (t *topology) allocZonesForNode(rsMgr *rsManager, zoneNumNeed, replicaNum i
if len(t.domainExcludeZones) > 0 {
zones = t.getDomainExcludeZones()
log.LogInfof("action[allocZonesForMetaNode] getDomainExcludeZones zones [%v]", t.domainExcludeZones)
} else if specialZones != nil && len(specialZones) > 0 {
} else if len(specialZones) > 0 {
zones = specialZones
zoneNumNeed = len(specialZones)
} else {
@ -1336,15 +1321,6 @@ func (t *topology) allocZonesForNode(rsMgr *rsManager, zoneNumNeed, replicaNum i
return
}
func (ns *nodeSet) dataNodeCount() int {
var count int
ns.dataNodes.Range(func(key, value interface{}) bool {
count++
return true
})
return count
}
// Zone stores all the zone related information
type Zone struct {
name string
@ -1699,7 +1675,7 @@ func (zone *Zone) isUsedRatio(ratio float64) (can bool) {
zone.metaNodes.Range(func(addr, value interface{}) bool {
metaNode := value.(*MetaNode)
if metaNode.IsActive && metaNode.isWritable() {
if metaNode.IsActive && metaNode.IsWriteAble() {
metaNodeUsed += metaNode.Used
} else {
metaNodeUsed += metaNode.Total
@ -1839,14 +1815,6 @@ func (zone *Zone) loadDataNodeQosLimit() {
})
}
func (zone *Zone) dataNodeCount() (len int) {
zone.dataNodes.Range(func(key, value interface{}) bool {
len++
return true
})
return
}
func (zone *Zone) updateDecommissionLimit(limit int32, c *Cluster) (err error) {
nodeSets := zone.getAllNodeSet()

2
master/topology_test.go
View File

@ -136,7 +136,7 @@ func TestAllocZones(t *testing.T) {
t.Logf("ChooseTargetDataHosts in single zone,hosts[%v]", hosts)
// cross zone
hosts, _, err = cluster.getHostFromNormalZone(TypeDataPartition, nil, nil, nil, replicaNum, 2, "")
_, _, err = cluster.getHostFromNormalZone(TypeDataPartition, nil, nil, nil, replicaNum, 2, "")
require.NoError(t, err)
// specific zone

2
metanode/partition.go
View File

@ -500,7 +500,6 @@ type metaPartition struct {
vol *Vol
manager *metadataManager
isLoadingMetaPartition bool
summaryLock sync.Mutex
ebsClient *blobstore.BlobStoreClient
volType int
isFollowerRead bool
@ -512,7 +511,6 @@ type metaPartition struct {
uniqChecker *uniqChecker
verSeq uint64
multiVersionList *proto.VolVersionInfoList
versionLock sync.Mutex
verUpdateChan chan []byte
enableAuditLog bool
recycleInodeDelFileFlag atomicutil.Flag

1
proto/admin_proto.go
View File

@ -553,7 +553,6 @@ type DeleteDataPartitionRequest struct {
PartitionSize int
Force bool
DecommissionType uint32
IsSpecialReplica bool
}
// DeleteDataPartitionResponse defines the response to the request of deleting a data partition.

1
sdk/data/blobstore/reader.go
View File

@ -21,6 +21,7 @@ import (
"os"
"sync"
"syscall"
"time"
"github.com/cubefs/cubefs/blockcache/bcache"
"github.com/cubefs/cubefs/proto"

2
sdk/data/wrapper/default_random_selector.go
View File

@ -140,8 +140,6 @@ func (s *DefaultRandomSelector) RemoveDP(partitionID uint64) {
s.localLeaderPartitions = newLocalLeaderPartitions
s.Unlock()
log.LogDebugf("RemoveDP: finish, partitionID[%v], len(s.localLeaderPartitions)=%v\n", partitionID, len(s.localLeaderPartitions))
return
}
func (s *DefaultRandomSelector) Count() int {

1
util/atomicutil/bool.go
View File

@ -31,7 +31,6 @@ func (b *Bool) Store(v bool) {
val = 1
}
atomic.StoreUint32(&b.val, val)
return
}
func (b *Bool) CompareAndSwap(old bool, newVal bool) (swaped bool) {

4
util/log/log.go
View File

@ -349,9 +349,7 @@ func InitLog(dir, module string, level Level, rotate *LogRotate, logLeftSpaceLim
}
if rotate.headRoom == 0 {
var minLogLeftSpaceLimit float64
minLogLeftSpaceLimit = float64(fs.Blocks*uint64(fs.Bsize)) * logLeftSpaceLimitRatio / 1024 / 1024
minLogLeftSpaceLimit := float64(fs.Blocks*uint64(fs.Bsize)) * logLeftSpaceLimitRatio / 1024 / 1024
rotate.SetHeadRoomMb(int64(math.Min(minLogLeftSpaceLimit, DefaultHeadRoom)))
}