diff --git a/client/fuse.go b/client/fuse.go index 4a70c7014..f57e20e70 100644 --- a/client/fuse.go +++ b/client/fuse.go @@ -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" diff --git a/datanode/partition.go b/datanode/partition.go index f74336cf7..067c97cc2 100644 --- a/datanode/partition.go +++ b/datanode/partition.go @@ -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() { diff --git a/datanode/server_handler.go b/datanode/server_handler.go index 91f630f8a..21cbee883 100644 --- a/datanode/server_handler.go +++ b/datanode/server_handler.go @@ -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") } } diff --git a/datanode/space_manager.go b/datanode/space_manager.go index db4b3f19e..a7370c5a6 100644 --- a/datanode/space_manager.go +++ b/datanode/space_manager.go @@ -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 { diff --git a/datanode/wrap_operator.go b/datanode/wrap_operator.go index cde534620..4b796041e 100644 --- a/datanode/wrap_operator.go +++ b/datanode/wrap_operator.go @@ -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) } diff --git a/master/api_service.go b/master/api_service.go index 96be9bec5..8c90451d7 100644 --- a/master/api_service.go +++ b/master/api_service.go @@ -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 } diff --git a/master/cluster.go b/master/cluster.go index fd32a5d4f..641559071 100644 --- a/master/cluster.go +++ b/master/cluster.go @@ -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 } diff --git a/master/data_node.go b/master/data_node.go index 1acd33cd1..dbdcd9de4 100644 --- a/master/data_node.go +++ b/master/data_node.go @@ -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 diff --git a/master/data_partition.go b/master/data_partition.go index 3730dbab3..e873ee5f8 100644 --- a/master/data_partition.go +++ b/master/data_partition.go @@ -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 { diff --git a/master/data_partition_check.go b/master/data_partition_check.go index b72354c03..d5e7f56b9 100644 --- a/master/data_partition_check.go +++ b/master/data_partition_check.go @@ -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. diff --git a/master/operate_util.go b/master/operate_util.go index 45a09e6de..3af42b2c1 100644 --- a/master/operate_util.go +++ b/master/operate_util.go @@ -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 } diff --git a/master/topology.go b/master/topology.go index e63c8181f..81d5fd125 100644 --- a/master/topology.go +++ b/master/topology.go @@ -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() diff --git a/master/topology_test.go b/master/topology_test.go index 7fd95036d..263366f85 100644 --- a/master/topology_test.go +++ b/master/topology_test.go @@ -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 diff --git a/metanode/partition.go b/metanode/partition.go index b02378776..422c2e395 100644 --- a/metanode/partition.go +++ b/metanode/partition.go @@ -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 diff --git a/proto/admin_proto.go b/proto/admin_proto.go index ba4519f9f..bd405dbbe 100644 --- a/proto/admin_proto.go +++ b/proto/admin_proto.go @@ -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. diff --git a/sdk/data/blobstore/reader.go b/sdk/data/blobstore/reader.go index 4d994cd53..df76a8ea1 100644 --- a/sdk/data/blobstore/reader.go +++ b/sdk/data/blobstore/reader.go @@ -21,6 +21,7 @@ import ( "os" "sync" "syscall" + "time" "github.com/cubefs/cubefs/blockcache/bcache" "github.com/cubefs/cubefs/proto" diff --git a/sdk/data/wrapper/default_random_selector.go b/sdk/data/wrapper/default_random_selector.go index 5e995545b..cb275a0ee 100644 --- a/sdk/data/wrapper/default_random_selector.go +++ b/sdk/data/wrapper/default_random_selector.go @@ -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 { diff --git a/util/atomicutil/bool.go b/util/atomicutil/bool.go index 3f774b54e..7b280f8ce 100644 --- a/util/atomicutil/bool.go +++ b/util/atomicutil/bool.go @@ -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) { diff --git a/util/log/log.go b/util/log/log.go index 7dcc2cf0b..ea3c70106 100644 --- a/util/log/log.go +++ b/util/log/log.go @@ -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))) }