diff --git a/datanode/data_partition_repair.go b/datanode/data_partition_repair.go index f4bd911e0..abdbb6a69 100644 --- a/datanode/data_partition_repair.go +++ b/datanode/data_partition_repair.go @@ -17,20 +17,18 @@ package datanode import ( "encoding/binary" "encoding/json" + "fmt" + "hash/crc32" "math" "net" "sync" "sync/atomic" "time" - "github.com/cubefs/cubefs/util" - - "fmt" - "hash/crc32" - "github.com/cubefs/cubefs/proto" "github.com/cubefs/cubefs/repl" "github.com/cubefs/cubefs/storage" + "github.com/cubefs/cubefs/util" "github.com/cubefs/cubefs/util/errors" "github.com/cubefs/cubefs/util/log" ) @@ -58,7 +56,6 @@ func NewDataPartitionRepairTask(extentFiles []*storage.ExtentInfo, tinyDeleteRec extentFile.Source = source task.extents[extentFile.FileID] = extentFile } - return } @@ -90,7 +87,7 @@ func (dp *DataPartition) repair(extentType uint8) { } } - //fix dp replica index panic , using replica copy + // fix dp replica index panic , using replica copy replica := dp.getReplicaCopy() repairTasks := make([]*DataPartitionRepairTask, len(replica)) err := dp.buildDataPartitionRepairTask(repairTasks, extentType, tinyExtents, replica) @@ -407,8 +404,8 @@ func (dp *DataPartition) buildExtentRepairTasks(repairTasks []*DataPartitionRepa func (dp *DataPartition) notifyFollower(wg *sync.WaitGroup, index int, members []*DataPartitionRepairTask) (err error) { p := repl.NewPacketToNotifyExtentRepair(dp.partitionID) // notify all the followers to repair var conn *net.TCPConn - //target := dp.getReplicaAddr(index) - //fix repair case panic,may be dp's replicas is change + // target := dp.getReplicaAddr(index) + // fix repair case panic,may be dp's replicas is change target := members[index].addr p.Data, _ = json.Marshal(members[index]) @@ -456,16 +453,13 @@ func (dp *DataPartition) NotifyExtentRepair(members []*DataPartitionRepairTask) return } -//const MaxRepairErrCnt = 1000 - // DoStreamExtentFixRepair executes the repair on the followers. func (dp *DataPartition) doStreamExtentFixRepair(wg *sync.WaitGroup, remoteExtentInfo *storage.ExtentInfo) { defer wg.Done() err := dp.streamRepairExtent(remoteExtentInfo) - if err != nil { - //only decommission repair need to check err cnt + // only decommission repair need to check err cnt if dp.isDecommissionRecovering() { atomic.AddUint64(&dp.recoverErrCnt, 1) if atomic.LoadUint64(&dp.recoverErrCnt) >= dp.dataNode.GetDpMaxRepairErrCnt() { @@ -539,9 +533,7 @@ func (dp *DataPartition) streamRepairExtent(remoteExtentInfo *storage.ExtentInfo return } currFixOffset := localExtentInfo.Size - var ( - hasRecoverySize uint64 - ) + var hasRecoverySize uint64 var loopTimes uint64 for currFixOffset < remoteExtentInfo.Size { if dp.stopRecover && dp.isDecommissionRecovering() { @@ -604,7 +596,7 @@ func (dp *DataPartition) streamRepairExtent(remoteExtentInfo *storage.ExtentInfo if reply.ArgLen == TinyExtentRepairReadResponseArgLen { remoteAvaliSize = binary.BigEndian.Uint64(reply.Arg[9:TinyExtentRepairReadResponseArgLen]) } - if reply.Arg != nil { //compact v1.2.0 recovery + if reply.Arg != nil { // compact v1.2.0 recovery isEmptyResponse = reply.Arg[0] == EmptyResponse } if isEmptyResponse { @@ -644,7 +636,6 @@ func (dp *DataPartition) streamRepairExtent(remoteExtentInfo *storage.ExtentInfo } return - } func intMin(a, b int) int { diff --git a/datanode/disk.go b/datanode/disk.go index c92aa6460..c62d95d8a 100644 --- a/datanode/disk.go +++ b/datanode/disk.go @@ -165,7 +165,6 @@ func (d *Disk) PartitionCount() int { } func (d *Disk) CanWrite() bool { - if d.Status == proto.ReadWrite || !d.RejectWrite { return true } @@ -303,7 +302,7 @@ const ( func (d *Disk) checkDiskStatus() { path := path.Join(d.Path, DiskStatusFile) - fp, err := os.OpenFile(path, os.O_CREATE|os.O_TRUNC|os.O_RDWR, 0755) + fp, err := os.OpenFile(path, os.O_CREATE|os.O_TRUNC|os.O_RDWR, 0o755) if err != nil { d.triggerDiskError(err) return @@ -355,7 +354,6 @@ func (d *Disk) updateSpaceInfo() (err error) { } else if d.Available <= 0 { d.Status = proto.ReadOnly - } else { d.Status = proto.ReadWrite } @@ -441,7 +439,7 @@ func (d *Disk) isExpiredPartitionDir(filename string) (isExpiredPartitionDir boo // RestorePartition reads the files stored on the local disk and restores the data partitions. func (d *Disk) RestorePartition(visitor PartitionVisitor) (err error) { - var convert = func(node *proto.DataNodeInfo) *DataNodeInfo { + convert := func(node *proto.DataNodeInfo) *DataNodeInfo { result := &DataNodeInfo{} result.Addr = node.Addr result.PersistenceDataPartitions = node.PersistenceDataPartitions @@ -525,7 +523,6 @@ func (d *Disk) RestorePartition(visitor PartitionVisitor) (err error) { if visitor != nil { visitor(dp) } - }(partitionID, filename) } diff --git a/datanode/metric.go b/datanode/metric.go index be95af15b..cec43463e 100644 --- a/datanode/metric.go +++ b/datanode/metric.go @@ -16,9 +16,10 @@ package datanode import ( "fmt" + "time" + "github.com/cubefs/cubefs/util/exporter" "github.com/cubefs/cubefs/util/log" - "time" ) const ( diff --git a/datanode/nodeinfo.go b/datanode/nodeinfo.go index 951d297ff..4e2712fc2 100644 --- a/datanode/nodeinfo.go +++ b/datanode/nodeinfo.go @@ -18,9 +18,7 @@ const ( MaxRepairErrCnt = 1000 ) -var ( - nodeInfoStopC = make(chan struct{}, 0) -) +var nodeInfoStopC = make(chan struct{}, 0) func (m *DataNode) startUpdateNodeInfo() { ticker := time.NewTicker(UpdateNodeInfoTicket) diff --git a/datanode/partition.go b/datanode/partition.go index 2b6fc9d51..d44423076 100644 --- a/datanode/partition.go +++ b/datanode/partition.go @@ -17,28 +17,26 @@ package datanode import ( "encoding/json" "fmt" + "hash/crc32" "io/ioutil" "math" + "net" "os" "path" + "sort" "strconv" "strings" "sync" "sync/atomic" - "time" - - "github.com/cubefs/cubefs/util" - - "hash/crc32" - "net" - "sort" "syscall" + "time" raftProto "github.com/cubefs/cubefs/depends/tiglabs/raft/proto" "github.com/cubefs/cubefs/proto" "github.com/cubefs/cubefs/raftstore" "github.com/cubefs/cubefs/repl" "github.com/cubefs/cubefs/storage" + "github.com/cubefs/cubefs/util" "github.com/cubefs/cubefs/util/errors" "github.com/cubefs/cubefs/util/exporter" "github.com/cubefs/cubefs/util/log" @@ -150,9 +148,9 @@ type DataPartition struct { verSeqPrepare uint64 verSeqCommitStatus int8 multiVersionList []*MetaMultiSnapshotInfo - decommissionRepairProgress float64 //record repair progress for decommission datapartition + decommissionRepairProgress float64 // record repair progress for decommission datapartition stopRecover bool - recoverErrCnt uint64 //donot reset, if reach max err cnt, delete this dp + recoverErrCnt uint64 // donot reset, if reach max err cnt, delete this dp } func (dp *DataPartition) IsForbidden() bool { @@ -164,7 +162,6 @@ func (dp *DataPartition) SetForbidden(status bool) { } func CreateDataPartition(dpCfg *dataPartitionCfg, disk *Disk, request *proto.CreateDataPartitionRequest) (dp *DataPartition, err error) { - if dp, err = newDataPartition(dpCfg, disk, true); err != nil { return } @@ -174,7 +171,7 @@ func CreateDataPartition(dpCfg *dataPartitionCfg, disk *Disk, request *proto.Cre } else { // init leaderSize to partitionSize disk.updateDisk(uint64(request.LeaderSize)) - //ensure heartbeat report Recovering + // ensure heartbeat report Recovering dp.partitionStatus = proto.Recovering go dp.StartRaftAfterRepair(false) } @@ -238,9 +235,7 @@ func (dp *DataPartition) ForceSetRaftRunning() { // It reads the partition metadata file stored under the specified directory // and creates the partition instance. func LoadDataPartition(partitionDir string, disk *Disk) (dp *DataPartition, err error) { - var ( - metaFileData []byte - ) + var metaFileData []byte if metaFileData, err = ioutil.ReadFile(path.Join(partitionDir, DataPartitionMetadataFileName)); err != nil { return } @@ -548,7 +543,7 @@ func (dp *DataPartition) PersistMetadata() (err error) { metaData []byte ) fileName := path.Join(dp.Path(), TempMetadataFileName) - if metadataFile, err = os.OpenFile(fileName, os.O_CREATE|os.O_RDWR, 0666); err != nil { + if metadataFile, err = os.OpenFile(fileName, os.O_CREATE|os.O_RDWR, 0o666); err != nil { return } defer func() { @@ -557,8 +552,8 @@ func (dp *DataPartition) PersistMetadata() (err error) { os.Remove(fileName) }() - //sp := sortedPeers(dp.config.Peers) - //sort.Sort(sp) + // sp := sortedPeers(dp.config.Peers) + // sort.Sort(sp) md := &DataPartitionMetadata{ VolumeID: dp.config.VolName, @@ -583,6 +578,7 @@ func (dp *DataPartition) PersistMetadata() (err error) { err = os.Rename(fileName, path.Join(dp.Path(), DataPartitionMetadataFileName)) return } + func (dp *DataPartition) statusUpdateScheduler() { ticker := time.NewTicker(time.Minute) snapshotTicker := time.NewTicker(time.Minute * 5) @@ -628,7 +624,7 @@ func (dp *DataPartition) statusUpdate() { status = proto.ReadOnly } if dp.isNormalType() && dp.raftStatus == RaftStatusStopped { - //dp is still recovering + // dp is still recovering if dp.DataPartitionCreateType == proto.DecommissionedCreateDataPartition { status = proto.Recovering } else { @@ -645,9 +641,7 @@ func parseFileName(filename string) (extentID uint64, isExtent bool) { if isExtent = storage.RegexpExtentFile.MatchString(filename); !isExtent { return } - var ( - err error - ) + var err error if extentID, err = strconv.ParseUint(filename, 10, 64); err != nil { isExtent = false return @@ -778,9 +772,8 @@ func (dp *DataPartition) compareReplicas(v1, v2 []string) (equals bool) { // Fetch the replica information from the master. func (dp *DataPartition) fetchReplicasFromMaster() (isLeader bool, replicas []string, err error) { - var partition *proto.DataPartitionInfo - var retry = 0 + retry := 0 for { if partition, err = MasterClient.AdminAPI().GetDataPartition(dp.volumeID, dp.partitionID); err != nil { retry++ @@ -983,7 +976,7 @@ func (dp *DataPartition) doStreamFixTinyDeleteRecord(repairTask *DataPartitionRe } DeleteLimiterWait() dp.disk.allocCheckLimit(proto.IopsWriteType, 1) - //log.LogInfof("doStreamFixTinyDeleteRecord Delete PartitionID(%v)_Extent(%v)_Offset(%v)_Size(%v)", dp.partitionID, extentID, offset, size) + // log.LogInfof("doStreamFixTinyDeleteRecord Delete PartitionID(%v)_Extent(%v)_Offset(%v)_Size(%v)", dp.partitionID, extentID, offset, size) store.MarkDelete(extentID, int64(offset), int64(size)) } } @@ -997,7 +990,7 @@ func (dp *DataPartition) ChangeRaftMember(changeType raftProto.ConfChangeType, p func (dp *DataPartition) canRemoveSelf() (canRemove bool, err error) { var partition *proto.DataPartitionInfo - var retry = 0 + retry := 0 for { if partition, err = MasterClient.AdminAPI().GetDataPartition(dp.volumeID, dp.partitionID); err != nil { log.LogErrorf("action[canRemoveSelf] err[%v]", err) @@ -1167,7 +1160,6 @@ func (dp *DataPartition) doExtentEvict(vv *proto.SimpleVolView) { break } } - } func (dp *DataPartition) startEvict() { @@ -1248,27 +1240,27 @@ func getWithDefault(base, def int) int { } func (dp *DataPartition) StopDecommissionRecover(stop bool) { - //only work for decommission repair + // only work for decommission repair if !dp.isDecommissionRecovering() { log.LogWarnf("[StopDecommissionRecover] dp(%d) is not in recovering status: type %d status %d", dp.partitionID, dp.partitionType, dp.Status()) return } - //for check timeout + // for check timeout dp.stopRecover = stop dp.PersistMetadata() } func (dp *DataPartition) isDecommissionRecovering() bool { - //decommission recover failed or success will set to normal + // decommission recover failed or success will set to normal return dp.DataPartitionCreateType == proto.DecommissionedCreateDataPartition - } + func (dp *DataPartition) handleDecommissionRecoverFailed() { if !dp.isDecommissionRecovering() { return } - //prevent status changing from Unavailable to Recovering again in statusUpdate() + // prevent status changing from Unavailable to Recovering again in statusUpdate() dp.partitionType = proto.NormalCreateDataPartition dp.partitionStatus = proto.Unavailable log.LogWarnf("[handleDecommissionRecoverFailed] dp(%d) recover failed reach max limit", dp.partitionID) diff --git a/datanode/partition_op_by_raft.go b/datanode/partition_op_by_raft.go index 3ad3229f6..15c80ea37 100644 --- a/datanode/partition_op_by_raft.go +++ b/datanode/partition_op_by_raft.go @@ -19,14 +19,15 @@ import ( "encoding/binary" "encoding/json" "fmt" + "net" + "strings" + "github.com/cubefs/cubefs/depends/tiglabs/raft" "github.com/cubefs/cubefs/proto" "github.com/cubefs/cubefs/repl" "github.com/cubefs/cubefs/storage" "github.com/cubefs/cubefs/util/exporter" "github.com/cubefs/cubefs/util/log" - "net" - "strings" ) type RaftCmdItem struct { @@ -187,7 +188,6 @@ type ItemIterator struct { // NewItemIterator creates a new item iterator. func NewItemIterator(applyID uint64) *ItemIterator { - si := new(ItemIterator) si.applyID = applyID return si @@ -299,9 +299,7 @@ func (dp *DataPartition) RandomWriteSubmit(pkg *repl.Packet) (err error) { log.LogErrorf("action[RandomWriteSubmit] [%v] marshal error %v", dp.partitionID, err) return } - var ( - resp interface{} - ) + var resp interface{} resp, err = dp.Put(nil, val) pkg.ResultCode, _ = resp.(uint8) if err != nil { diff --git a/datanode/partition_raft.go b/datanode/partition_raft.go index 2999bd290..63859ef63 100644 --- a/datanode/partition_raft.go +++ b/datanode/partition_raft.go @@ -77,7 +77,6 @@ func (dp *DataPartition) raftPort() (heartbeat, replica int, err error) { // StartRaft start raft instance when data partition start or restore. func (dp *DataPartition) StartRaft(isLoad bool) (err error) { - // cache or preload partition not support raft and repair. if !dp.isNormalType() { return nil @@ -173,7 +172,7 @@ func (dp *DataPartition) CanRemoveRaftMember(peer proto.Peer, force bool) error if nodeID.NodeID == peer.ID { continue } - //check nodeID is valid + // check nodeID is valid hasDownReplicasExcludePeer = append(hasDownReplicasExcludePeer, nodeID.NodeID) } @@ -201,7 +200,6 @@ func (dp *DataPartition) CanRemoveRaftMember(peer proto.Peer, force bool) error // 2. collect the applied ids from raft members. // 3. based on the minimum applied id to cutoff and delete the saved raft log in order to free the disk space. func (dp *DataPartition) StartRaftLoggingSchedule() { - // cache or preload partition not support raft and repair. if !dp.isNormalType() { return @@ -360,7 +358,6 @@ func (dp *DataPartition) StartRaftAfterRepair(isLoad bool) { // Add a raft node. func (dp *DataPartition) addRaftNode(req *proto.AddDataPartitionRaftMemberRequest, index uint64) (isUpdated bool, err error) { - // cache or preload partition not support raft and repair. if !dp.isNormalType() { return false, fmt.Errorf("addRaftNode (%v) not support", dp) @@ -401,7 +398,6 @@ func (dp *DataPartition) addRaftNode(req *proto.AddDataPartitionRaftMemberReques // Delete a raft node. func (dp *DataPartition) removeRaftNode(req *proto.RemoveDataPartitionRaftMemberRequest, index uint64) (isUpdated bool, err error) { - // cache or preload partition not support raft and repair. if !dp.isNormalType() { return false, fmt.Errorf("removeRaftNode (%v) not support", dp) @@ -459,7 +455,7 @@ func (dp *DataPartition) removeRaftNode(req *proto.RemoveDataPartitionRaftMember func (dp *DataPartition) storeAppliedID(applyIndex uint64) (err error) { filename := path.Join(dp.Path(), TempApplyIndexFile) - fp, err := os.OpenFile(filename, os.O_RDWR|os.O_APPEND|os.O_TRUNC|os.O_CREATE, 0755) + fp, err := os.OpenFile(filename, os.O_RDWR|os.O_APPEND|os.O_TRUNC|os.O_CREATE, 0o755) if err != nil { return } @@ -538,14 +534,14 @@ func (s *DataNode) startRaftServer(cfg *config.Config) (err error) { RaftHeartbetPort: s.raftHeartbeat, RaftReplicaPort: s.raftReplica, } - var ok = false + ok := false if ok, err = config.CheckOrStoreConstCfg(s.raftDir, config.DefaultConstConfigFile, &constCfg); !ok { log.LogErrorf("constCfg check failed %v %v %v %v", s.raftDir, config.DefaultConstConfigFile, constCfg, err) return fmt.Errorf("constCfg check failed %v %v %v %v", s.raftDir, config.DefaultConstConfigFile, constCfg, err) } if _, err = os.Stat(s.raftDir); err != nil { - if err = os.MkdirAll(s.raftDir, 0755); err != nil { + if err = os.MkdirAll(s.raftDir, 0o755); err != nil { err = errors.NewErrorf("create raft server dir: %s", err.Error()) log.LogErrorf("action[startRaftServer] cannot start raft server err(%v)", err) return @@ -655,14 +651,12 @@ func (dp *DataPartition) findMaxAppliedID(allAppliedIDs []uint64) (maxAppliedID // Get the partition size from the leader. func (dp *DataPartition) getLeaderPartitionSize(maxExtentID uint64) (size uint64, err error) { - var ( - conn *net.TCPConn - ) + var conn *net.TCPConn p := NewPacketToGetPartitionSize(dp.partitionID) p.ExtentID = maxExtentID target := dp.getReplicaAddr(0) - conn, err = gConnPool.GetConnect(target) //get remote connect + conn, err = gConnPool.GetConnect(target) // get remote connect if err != nil { err = errors.Trace(err, " partition(%v) get host(%v) connect", dp.partitionID, target) return @@ -692,12 +686,10 @@ func (dp *DataPartition) getLeaderPartitionSize(maxExtentID uint64) (size uint64 } func (dp *DataPartition) getMaxExtentIDAndPartitionSize(target string) (maxExtentID, PartitionSize uint64, err error) { - var ( - conn *net.TCPConn - ) + var conn *net.TCPConn p := NewPacketToGetMaxExtentIDAndPartitionSIze(dp.partitionID) - conn, err = gConnPool.GetConnect(target) //get remote connect + conn, err = gConnPool.GetConnect(target) // get remote connect if err != nil { err = errors.Trace(err, " partition(%v) get host(%v) connect", dp.partitionID, target) return diff --git a/datanode/server.go b/datanode/server.go index 6431c06ea..0f6d08e02 100644 --- a/datanode/server.go +++ b/datanode/server.go @@ -16,9 +16,11 @@ package datanode import ( "bytes" + "errors" "fmt" "net" "net/http" + "os" "os/exec" "regexp" "runtime" @@ -26,11 +28,8 @@ import ( "strings" "sync" "sync/atomic" - "time" - - "errors" - "os" "syscall" + "time" "github.com/cubefs/cubefs/cmd/common" "github.com/cubefs/cubefs/proto" @@ -55,7 +54,7 @@ var ( LocalIP, serverPort string gConnPool = util.NewConnectPool() - //MasterClient = masterSDK.NewMasterClient(nil, false) + // MasterClient = masterSDK.NewMasterClient(nil, false) MasterClient *masterSDK.MasterCLientWithResolver ) @@ -99,15 +98,15 @@ const ( CfgDiskRdonlySpace = "diskRdonlySpace" // int // smux Config - ConfigKeyEnableSmuxClient = "enableSmuxConnPool" //bool - ConfigKeySmuxPortShift = "smuxPortShift" //int - ConfigKeySmuxMaxConn = "smuxMaxConn" //int - ConfigKeySmuxStreamPerConn = "smuxStreamPerConn" //int - ConfigKeySmuxMaxBuffer = "smuxMaxBuffer" //int - ConfigKeySmuxTotalStream = "sumxTotalStream" //int + ConfigKeyEnableSmuxClient = "enableSmuxConnPool" // bool + ConfigKeySmuxPortShift = "smuxPortShift" // int + ConfigKeySmuxMaxConn = "smuxMaxConn" // int + ConfigKeySmuxStreamPerConn = "smuxStreamPerConn" // int + ConfigKeySmuxMaxBuffer = "smuxMaxBuffer" // int + ConfigKeySmuxTotalStream = "sumxTotalStream" // int - //rate limit control enable - ConfigDiskQosEnable = "diskQosEnable" //bool + // rate limit control enable + ConfigDiskQosEnable = "diskQosEnable" // bool ConfigServiceIDKey = "serviceIDKey" ) @@ -148,7 +147,7 @@ type DataNode struct { metrics *DataNodeMetrics metricsDegrade int64 metricsCnt uint64 - volUpdating sync.Map //map[string]*verOp2Phase + volUpdating sync.Map // map[string]*verOp2Phase control common.Control @@ -213,11 +212,11 @@ func doStart(server common.Server, cfg *config.Config) (err error) { s.registerMetrics() s.register(cfg) - //parse the smux config + // parse the smux config if err = s.parseSmuxConfig(cfg); err != nil { return } - //connection pool must be created before initSpaceManager + // connection pool must be created before initSpaceManager s.initConnPool() // init limit @@ -245,7 +244,7 @@ func doStart(server common.Server, cfg *config.Config) (err error) { return } - //smux listening & smux connection pool + // smux listening & smux connection pool if err = s.startSmuxService(cfg); err != nil { return } @@ -349,6 +348,7 @@ func (s *DataNode) updateQosLimit() { disk.updateQosLimiter() } } + func (s *DataNode) startSpaceManager(cfg *config.Config) (err error) { s.startTime = time.Now().Unix() s.space = NewSpaceManager(s) @@ -464,9 +464,7 @@ func parseDiskPath(pathStr string) (disks []string, err error) { // registers the data node on the master to report the information such as IsIPV4 address. // The startup of a data node will be blocked until the registration succeeds. func (s *DataNode) register(cfg *config.Config) { - var ( - err error - ) + var err error timer := time.NewTimer(0) @@ -522,7 +520,7 @@ type DataNodeInfo struct { } func (s *DataNode) checkLocalPartitionMatchWithMaster() (lackPartitions []uint64, err error) { - var convert = func(node *proto.DataNodeInfo) *DataNodeInfo { + convert := func(node *proto.DataNodeInfo) *DataNodeInfo { result := &DataNodeInfo{} result.Addr = node.Addr result.PersistenceDataPartitions = node.PersistenceDataPartitions @@ -770,7 +768,7 @@ func (s *DataNode) parseSmuxConfig(cfg *config.Config) error { } } - //smux conn pool config + // smux conn pool config if s.enableSmuxConnPool { s.smuxConnPoolConfig = util.DefaultSmuxConnPoolConfig() if maxBuffer > 0 { diff --git a/datanode/server_handler.go b/datanode/server_handler.go index 262fe20a1..764aba4af 100644 --- a/datanode/server_handler.go +++ b/datanode/server_handler.go @@ -17,7 +17,6 @@ package datanode import ( "encoding/json" "fmt" - "github.com/cubefs/cubefs/util/config" "net/http" "os" "path" @@ -27,11 +26,10 @@ import ( "github.com/cubefs/cubefs/depends/tiglabs/raft" "github.com/cubefs/cubefs/proto" "github.com/cubefs/cubefs/storage" + "github.com/cubefs/cubefs/util/config" ) -var ( - AutoRepairStatus = true -) +var AutoRepairStatus = true func (s *DataNode) getDiskAPI(w http.ResponseWriter, r *http.Request) { disks := make([]interface{}, 0) diff --git a/datanode/space_manager.go b/datanode/space_manager.go index 9ae90ffb1..af45f12a8 100644 --- a/datanode/space_manager.go +++ b/datanode/space_manager.go @@ -16,11 +16,10 @@ package datanode import ( "fmt" - "sync" - "time" - "math" "os" + "sync" + "time" "github.com/cubefs/cubefs/proto" "github.com/cubefs/cubefs/raftstore" @@ -75,7 +74,7 @@ func (manager *SpaceManager) Stop() { close(manager.samplerDone) // Parallel stop data partitions. const maxParallelism = 128 - var parallelism = int(math.Min(float64(maxParallelism), float64(len(manager.partitions)))) + parallelism := int(math.Min(float64(maxParallelism), float64(len(manager.partitions)))) wg := sync.WaitGroup{} partitionC := make(chan *DataPartition, parallelism) wg.Add(1) @@ -182,6 +181,7 @@ func (manager *SpaceManager) GetClusterID() (clusterID string) { func (manager *SpaceManager) SetRaftStore(raftStore raftstore.RaftStore) { manager.raftStore = raftStore } + func (manager *SpaceManager) GetRaftStore() (raftStore raftstore.RaftStore) { return manager.raftStore } @@ -345,6 +345,7 @@ func (manager *SpaceManager) minPartitionCnt(decommissionedDisks []string) (d *D d = minWeightDisk return d } + func (manager *SpaceManager) statUpdateScheduler() { go func() { ticker := time.NewTicker(10 * time.Second) @@ -420,7 +421,6 @@ func (manager *SpaceManager) CreatePartition(request *proto.CreateDataPartitionR // DeletePartition deletes a partition based on the partition id. func (manager *SpaceManager) DeletePartition(dpID uint64) { - manager.partitionMutex.Lock() dp := manager.partitions[dpID] diff --git a/datanode/wrap_operator.go b/datanode/wrap_operator.go index 6cdef6371..ac601f208 100644 --- a/datanode/wrap_operator.go +++ b/datanode/wrap_operator.go @@ -19,14 +19,13 @@ import ( "encoding/binary" "encoding/json" "fmt" + "hash/crc32" "net" "strconv" + "strings" "sync/atomic" "time" - "hash/crc32" - "strings" - "github.com/cubefs/cubefs/depends/tiglabs/raft" raftProto "github.com/cubefs/cubefs/depends/tiglabs/raft/proto" "github.com/cubefs/cubefs/proto" @@ -38,9 +37,7 @@ import ( "github.com/cubefs/cubefs/util/log" ) -var ( - ErrForbiddenDataPartition = errors.New("the data partition is forbidden") -) +var ErrForbiddenDataPartition = errors.New("the data partition is forbidden") func (s *DataNode) getPacketTpLabels(p *repl.Packet) map[string]string { labels := make(map[string]string) @@ -268,7 +265,6 @@ func (s *DataNode) handlePacketToCreateDataPartition(p *repl.Packet) { } func (s *DataNode) commitCreateVersion(volumeID string, verSeq uint64) (err error) { - for _, dp := range s.space.partitions { if dp.volumeID != volumeID { continue @@ -305,9 +301,7 @@ func (s *DataNode) commitCreateVersion(volumeID string, verSeq uint64) (err erro } func (s *DataNode) prepareCreateVersion(req *proto.MultiVersionOpRequest) (err error, opAagin bool) { - var ( - ver2Phase *verOp2Phase - ) + var ver2Phase *verOp2Phase if value, ok := s.volUpdating.Load(req.VolumeID); ok { ver2Phase = value.(*verOp2Phase) if req.VerSeq < ver2Phase.verSeq { @@ -451,7 +445,6 @@ func (s *DataNode) handleUpdateVerPacket(p *repl.Packet) { return } }() - } func (s *DataNode) checkVolumeForbidden(volNames []string) { @@ -545,7 +538,6 @@ func (s *DataNode) handleHeartbeatPacket(p *repl.Packet) { return } }() - } // Handle OpDeleteDataPartition packet. @@ -580,15 +572,12 @@ func (s *DataNode) handlePacketToDeleteDataPartition(p *repl.Packet) { log.LogErrorf("action[handlePacketToDeleteDataPartition] err(%v).", err) } log.LogInfof(fmt.Sprintf("action[handlePacketToDeleteDataPartition] %v error(%v)", request.PartitionId, err)) - } // Handle OpLoadDataPartition packet. func (s *DataNode) handlePacketToLoadDataPartition(p *repl.Packet) { task := &proto.AdminTask{} - var ( - err error - ) + var err error defer func() { if err != nil { p.PackErrorBody(ActionLoadDataPartition, err.Error()) @@ -602,9 +591,7 @@ func (s *DataNode) handlePacketToLoadDataPartition(p *repl.Packet) { } func (s *DataNode) asyncLoadDataPartition(task *proto.AdminTask) { - var ( - err error - ) + var err error request := &proto.LoadDataPartitionRequest{} response := &proto.LoadDataPartitionResponse{} if task.OpCode == proto.OpLoadDataPartition { @@ -636,9 +623,7 @@ func (s *DataNode) asyncLoadDataPartition(task *proto.AdminTask) { // Handle OpMarkDelete packet. func (s *DataNode) handleMarkDeletePacket(p *repl.Packet, c net.Conn) { - var ( - err error - ) + var err error defer func() { if err != nil { p.PackErrorBody(ActionBatchMarkDelete, err.Error()) @@ -673,9 +658,7 @@ func (s *DataNode) handleMarkDeletePacket(p *repl.Packet, c net.Conn) { // Handle OpMarkDelete packet. func (s *DataNode) handleBatchMarkDeletePacket(p *repl.Packet, c net.Conn) { - var ( - err error - ) + var err error defer func() { if err != nil { log.LogErrorf(fmt.Sprintf("(%v) error(%v).", p.GetUniqueLogId(), err)) @@ -898,9 +881,7 @@ func (s *DataNode) handleRandomWritePacket(p *repl.Packet) { } func (s *DataNode) handleStreamReadPacket(p *repl.Packet, connect net.Conn, isRepairRead bool) { - var ( - err error - ) + var err error defer func() { if err != nil { p.PackErrorBody(ActionStreamRead, err.Error()) @@ -924,9 +905,7 @@ func (s *DataNode) handleStreamReadPacket(p *repl.Packet, connect net.Conn, isRe } func (s *DataNode) handleExtentRepairReadPacket(p *repl.Packet, connect net.Conn, isRepairRead bool) { - var ( - err error - ) + var err error log.LogDebugf("handleExtentRepairReadPacket %v", p) defer func() { if err != nil { @@ -1107,9 +1086,7 @@ func (s *DataNode) tinyExtentRepairRead(request *repl.Packet, connect net.Conn) } avaliReplySize := uint64(needReplySize) - var ( - newOffset, newEnd int64 - ) + var newOffset, newEnd int64 for { if needReplySize <= 0 { break @@ -1123,9 +1100,7 @@ func (s *DataNode) tinyExtentRepairRead(request *repl.Packet, connect net.Conn) return } if newOffset > offset { - var ( - replySize int64 - ) + var replySize int64 if replySize, err = s.writeEmptyPacketOnTinyExtentRepairRead(reply, newOffset, offset, connect); err != nil { return } @@ -1166,9 +1141,7 @@ func (s *DataNode) tinyExtentRepairRead(request *repl.Packet, connect net.Conn) } func (s *DataNode) handlePacketToReadTinyDeleteRecordFile(p *repl.Packet, connect net.Conn) { - var ( - err error - ) + var err error defer func() { if err != nil { p.PackErrorBody(ActionStreamReadTinyDeleteRecord, err.Error()) @@ -1214,9 +1187,7 @@ func (s *DataNode) handlePacketToReadTinyDeleteRecordFile(p *repl.Packet, connec // Handle OpNotifyReplicasToRepair packet. func (s *DataNode) handlePacketToNotifyExtentRepair(p *repl.Packet) { - var ( - err error - ) + var err error partition := p.Object.(*DataPartition) mf := new(DataPartitionRepairTask) err = json.Unmarshal(p.Data, mf) @@ -1485,9 +1456,7 @@ func (s *DataNode) handlePacketToRemoveDataPartitionRaftMember(p *repl.Packet) { } func (s *DataNode) handlePacketToDataPartitionTryToLeader(p *repl.Packet) { - var ( - err error - ) + var err error defer func() { if err != nil { diff --git a/datanode/wrap_prepare.go b/datanode/wrap_prepare.go index 0c040e393..7d38e5443 100644 --- a/datanode/wrap_prepare.go +++ b/datanode/wrap_prepare.go @@ -17,13 +17,13 @@ package datanode import ( "encoding/json" "fmt" - "github.com/cubefs/cubefs/util/log" "hash/crc32" "sync/atomic" "github.com/cubefs/cubefs/proto" "github.com/cubefs/cubefs/repl" "github.com/cubefs/cubefs/storage" + "github.com/cubefs/cubefs/util/log" ) func (s *DataNode) Prepare(p *repl.Packet) (err error) { @@ -86,7 +86,7 @@ func (s *DataNode) checkCrc(p *repl.Packet) (err error) { func (s *DataNode) checkPartition(p *repl.Packet) (err error) { dp := s.space.Partition(p.PartitionID) if dp == nil { - //err = proto.ErrDataPartitionNotExists + // err = proto.ErrDataPartitionNotExists err = fmt.Errorf("data partition not exists %v", p.PartitionID) return }