From 81643df8cb34954bfdce610bb01cc8cb56c5a7d9 Mon Sep 17 00:00:00 2001 From: chihe Date: Wed, 24 Dec 2025 14:26:21 +0800 Subject: [PATCH] fix(master): add triggerCondition for dataNode balance @formatter:off Signed-off-by: chihe --- master/cluster_balance.go | 133 +----------------- master/data_partition.go | 2 +- master/flash_group.go | 3 +- remotecache/flashgroupmanager/config.go | 11 +- remotecache/flashgroupmanager/flash_group.go | 6 +- remotecache/flashgroupmanager/flash_node.go | 3 +- .../flashgroupmanager/flash_node_topology.go | 70 +++++---- remotecache/flashgroupmanager/http_server.go | 13 +- remotecache/flashgroupmanager/server.go | 17 ++- remotecache/flashnode/flashnode.go | 4 +- sdk/data/stream/extent_client.go | 3 +- 11 files changed, 97 insertions(+), 168 deletions(-) diff --git a/master/cluster_balance.go b/master/cluster_balance.go index 92c0aacc3..d1a1d9e17 100644 --- a/master/cluster_balance.go +++ b/master/cluster_balance.go @@ -1831,18 +1831,6 @@ func (c *Cluster) isOverloadDataNode(BalanceType uint32, dataNode *DataNode) boo } } -func (c *Cluster) getOverLoadDataNodesByDiskUsage() []*DataNode { - overloadNodes := make([]*DataNode, 0) - c.dataNodes.Range(func(addr, node interface{}) bool { - dataNode := node.(*DataNode) - if dataNode.isActive && dataNode.UsageRatio >= c.cfg.DataNodeBalanceByDiskUsageHigh { - overloadNodes = append(overloadNodes, dataNode) - } - return true - }) - return overloadNodes -} - // for inter-nodeset balancing: // given a datapartition, get a node set different than the excludedNodeSet // the selection critieria is @@ -1894,18 +1882,6 @@ func (c *Cluster) getUnderLoadNodesInNodeSetByDiskUsage(dp *DataPartition, nodes return underloadDataNodes, nil } -func (c *Cluster) getOverLoadDataNodesByDPCount() []*DataNode { - overloadNodes := make([]*DataNode, 0) - c.dataNodes.Range(func(addr, node interface{}) bool { - dataNode := node.(*DataNode) - if dataNode.isActive && dataNode.DataPartitionCount >= c.cfg.DataNodeBalanceByDPCountHigh { - overloadNodes = append(overloadNodes, dataNode) - } - return true - }) - return overloadNodes -} - func (c *Cluster) getUnderLoadNodeSetByDPCount(dp *DataPartition, zone string, excludedNodeSet uint64, numCopies int) (nodeset *nodeSet, err error) { var z *Zone @@ -1966,7 +1942,8 @@ func (c *Cluster) postBalanceDecommissionSucccess(BalanceType uint32, dp *DataPa if len(hosts) != 0 { log.LogInfof("action[postBalanceDecommissionSucccess], clusterID[%v], node[%v], dp[%v] still has replica in the nodeset [%v]", c.Name, datanode.ID, dp.PartitionID, datanode.NodeSetID) - if err := c.markDecommissionDataPartition(dp, hosts[0], dstNodeSet, false, BalanceByDiskUsage, lowPriorityDecommissionWeight); err != nil { + triggerCondition := fmt.Sprintf(" postBalanceDecommission_dp(%v)", dp.PartitionID) + if err := c.markDecommissionDataPartition(dp, hosts[0], dstNodeSet, false, BalanceByDiskUsage, lowPriorityDecommissionWeight, triggerCondition); err != nil { log.LogWarnf("action[postBalanceDecommissionSucccess], clusterID[%v], node[%v] failed to mark decommission data partition[%v], error[%v]", c.Name, datanode.ID, dp.PartitionID, err) } @@ -1985,109 +1962,3 @@ func (c *Cluster) getReplicaHostsInNodeSet(dp *DataPartition, nodeset uint64) [] return hosts } - -func (c *Cluster) handleDataNodeBalanceByDiskUsage(d *DataNode) (err error) { - var ( - sizeToBalance uint64 - sizeMarkedDecomm uint64 - numDPMarkedDecomm uint64 - dp *DataPartition - ) - - if !d.isActive || d.UsageRatio <= c.cfg.DataNodeBalanceByDiskUsageHigh { - return fmt.Errorf("action[handleDataNodeBalanceByDiskUsage], clusterID[%v], node[%v] failed, the node is no longer active or overload by disk usage", c.Name, d.ID) - } - - sizeToBalance = uint64(float64(d.Total) * (d.UsageRatio - c.cfg.DataNodeBalanceByDiskUsageHigh)) - - copiedDataPartitionReports := make([]*proto.DataPartitionReport, len(d.DataPartitionReports)) - copy(copiedDataPartitionReports, d.DataPartitionReports) - sort.Slice(copiedDataPartitionReports, func(i, j int) bool { return copiedDataPartitionReports[i].Used > copiedDataPartitionReports[j].Used }) - - for i := 0; i < len(copiedDataPartitionReports) && sizeMarkedDecomm < sizeToBalance && numDPMarkedDecomm < uint64(d.DataPartitionCount); i++ { - dpReport := copiedDataPartitionReports[i] - if dp, err = c.getDataPartitionByID(dpReport.PartitionID); err != nil { - log.LogErrorf("action[handleDataNodeBalanceByDiskUsage], clusterID[%v], node[%v] failed to find data partition[%v], error[%v]", c.Name, d.ID, dp.PartitionID, err) - continue - } - - if err = c.markDecommissionDataPartition(dp, d, 0, false, BalanceByDiskUsage, lowPriorityDecommissionWeight); err != nil { - log.LogErrorf("action[handleDataNodeBalanceByDiskUsage], clusterID[%v], node[%v] failed to mark decommission data partition[%v], error[%v]", c.Name, d.ID, dp.PartitionID, err) - continue - } - sizeMarkedDecomm += dp.used - numDPMarkedDecomm += 1 - } - - log.LogInfof("action[handleDataNodeBalanceByDiskUsage], clusterID[%v], node[%v] sizeMarkedDecomm[%v], numDPMarkedDecomm[%v]", c.Name, d.ID, sizeMarkedDecomm, numDPMarkedDecomm) - return nil -} - -func (c *Cluster) handleDataNodeBalanceByDPCount(d *DataNode) (err error) { - var ( - numDPsToBalance uint32 - sizeMarkedDecomm uint64 - numDPMarkedDecomm uint32 - dp *DataPartition - ) - - if !d.isActive || d.DataPartitionCount <= c.cfg.DataNodeBalanceByDPCountHigh { - return fmt.Errorf("action[handleDataNodeBalanceByDPCount], clusterID[%v], node[%v] failed, the node is no longer active or overload by dp count", c.Name, d.ID) - } - - numDPsToBalance = d.DataPartitionCount - c.cfg.DataNodeBalanceByDPCountHigh - - copiedDataPartitionReports := make([]*proto.DataPartitionReport, len(d.DataPartitionReports)) - copy(copiedDataPartitionReports, d.DataPartitionReports) - sort.Slice(copiedDataPartitionReports, func(i, j int) bool { return copiedDataPartitionReports[i].Used < copiedDataPartitionReports[j].Used }) - - for i := 0; i < len(copiedDataPartitionReports) && numDPMarkedDecomm < numDPsToBalance && numDPMarkedDecomm < d.DataPartitionCount; i++ { - dpReport := copiedDataPartitionReports[i] - if dp, err = c.getDataPartitionByID(dpReport.PartitionID); err != nil { - log.LogErrorf("action[handleDataNodeBalanceByDPCount], clusterID[%v], node[%v] failed to find data partition[%v], error[%v]", c.Name, d.ID, dp.PartitionID, err) - continue - } - - if err = c.markDecommissionDataPartition(dp, d, 0, false, BalanceByDPCount, lowPriorityDecommissionWeight); err != nil { - log.LogErrorf("action[handleDataNodeBalanceByDPCount], clusterID[%v], node[%v] failed to mark decommission data partition[%v], error[%v]", c.Name, d.ID, dp.PartitionID, err) - continue - } - sizeMarkedDecomm += dp.used - numDPMarkedDecomm += 1 - } - - log.LogInfof("action[handleDataNodeBalanceByDPCount], clusterID[%v], node[%v] sizeMarkedDecomm[%v], numDPMarkedDecomm[%v]", c.Name, d.ID, sizeMarkedDecomm, numDPMarkedDecomm) - return nil -} - -func (c *Cluster) doDataNodeBalanceByDiskUsage() { - overLoadDataNodes := c.getOverLoadDataNodesByDiskUsage() - for _, node := range overLoadDataNodes { - if err := c.handleDataNodeBalanceByDiskUsage(node); err != nil { - log.LogErrorf(err.Error()) - } - } -} - -func (c *Cluster) doDataNodeBalanceByDPCount() { - overLoadDataNodes := c.getOverLoadDataNodesByDPCount() - for _, node := range overLoadDataNodes { - if err := c.handleDataNodeBalanceByDPCount(node); err != nil { - log.LogErrorf(err.Error()) - } - } -} - -func (c *Cluster) scheduleToBalanceDataNode() { - c.runTask(&cTask{ - tickTime: time.Second * time.Duration(c.cfg.DataNodeBalanceInterval), - name: "scheduleToBalanceDataNode", - function: func() (fin bool) { - if c.partition != nil && c.partition.IsRaftLeader() { - c.doDataNodeBalanceByDiskUsage() - c.doDataNodeBalanceByDPCount() - } - return - }, - }) -} diff --git a/master/data_partition.go b/master/data_partition.go index a4c433cd0..82e1c18ed 100644 --- a/master/data_partition.go +++ b/master/data_partition.go @@ -2290,7 +2290,7 @@ func (partition *DataPartition) TryAcquireDecommissionToken(c *Cluster) bool { if partition.DecommissionDstNodeSet == 0 { // mark success, remove in advance if !c.isOverloadDataNode(partition.DecommissionType, datanode) { - partition.SetDecommissionStatus(DecommissionSuccess) + partition.SetDecommissionStatus(DecommissionSuccess, "isOverloadDataNodeFalse", err.Error()) log.LogInfof("action[TryAcquireDecommissionToken]dp %v canceled because src node is no longer overload, src node %v, err %v", partition.PartitionID, partition.DecommissionSrcAddr, err.Error()) return true diff --git a/master/flash_group.go b/master/flash_group.go index aaa0ad7f6..b5ada48f5 100644 --- a/master/flash_group.go +++ b/master/flash_group.go @@ -116,7 +116,8 @@ func (m *Server) createFlashGroup(w http.ResponseWriter, r *http.Request) { } func (c *Cluster) createFlashGroup(setSlots []uint32, setWeight uint32, gradualFlag bool, - step uint32) (fg *flashgroupmanager.FlashGroup, err error) { + step uint32, +) (fg *flashgroupmanager.FlashGroup, err error) { defer func() { if err != nil { log.LogErrorf("action[addFlashGroup],clusterID[%v] err:%v ", c.Name, err.Error()) diff --git a/remotecache/flashgroupmanager/config.go b/remotecache/flashgroupmanager/config.go index 52a62980a..92a4cdd3a 100644 --- a/remotecache/flashgroupmanager/config.go +++ b/remotecache/flashgroupmanager/config.go @@ -91,12 +91,21 @@ func parsePeerAddr(peerAddr string) (id uint64, ip string, port uint64, err erro func (cfg *clusterConfig) parsePeers(peerStr string) error { peerArr := strings.Split(peerStr, commaSplit) cfg.peerAddrs = peerArr + hp := cfg.heartbeatPort + rp := cfg.replicaPort + + if hp < 0 || hp > 65535 { + return fmt.Errorf("invalid heartbeatPort: %d", hp) + } + if rp < 0 || rp > 65535 { + return fmt.Errorf("invalid replicaPort: %d", rp) + } for _, peerAddr := range peerArr { id, ip, port, err := parsePeerAddr(peerAddr) if err != nil { return err } - cfg.peers = append(cfg.peers, raftstore.PeerAddress{Peer: proto.Peer{ID: id}, Address: ip, HeartbeatPort: int(cfg.heartbeatPort), ReplicaPort: int(cfg.replicaPort)}) + cfg.peers = append(cfg.peers, raftstore.PeerAddress{Peer: proto.Peer{ID: id}, Address: ip, HeartbeatPort: int(hp), ReplicaPort: int(rp)}) address := fmt.Sprintf("%v:%v", ip, port) syslog.Println(address) AddrDatabase[id] = address diff --git a/remotecache/flashgroupmanager/flash_group.go b/remotecache/flashgroupmanager/flash_group.go index 275fc3595..15f3f5502 100644 --- a/remotecache/flashgroupmanager/flash_group.go +++ b/remotecache/flashgroupmanager/flash_group.go @@ -306,7 +306,8 @@ func (fg *FlashGroup) GetPendingSlotsCount() (count int) { } func (fg *FlashGroup) UpdateStatus(status proto.FlashGroupStatus, - syncUpdateFlashGroupFunc SyncUpdateFlashGroupFunc, flashNodeTopo *FlashNodeTopology) (err error) { + syncUpdateFlashGroupFunc SyncUpdateFlashGroupFunc, flashNodeTopo *FlashNodeTopology, +) (err error) { fg.lock.Lock() defer fg.lock.Unlock() oldStatus := fg.Status @@ -322,7 +323,8 @@ func (fg *FlashGroup) UpdateStatus(status proto.FlashGroupStatus, } func (fg *FlashGroup) UpdateSlots(topology *FlashNodeTopology, needDeleteFgFlag bool, syncDeleteFlashGroupFunc SyncDeleteFlashGroupFunc, - syncUpdateFlashGroupFunc SyncUpdateFlashGroupFunc) (err error) { + syncUpdateFlashGroupFunc SyncUpdateFlashGroupFunc, +) (err error) { fg.lock.Lock() var updatedSlotsNum uint32 var newSlotStatus proto.SlotStatus diff --git a/remotecache/flashgroupmanager/flash_node.go b/remotecache/flashgroupmanager/flash_node.go index 3b966de70..112c515b3 100644 --- a/remotecache/flashgroupmanager/flash_node.go +++ b/remotecache/flashgroupmanager/flash_node.go @@ -129,7 +129,8 @@ func (flashNode *FlashNode) checkLiveliness() { func (flashNode *FlashNode) createHeartbeatTask(masterAddr string, flashNodeHandleReadTimeout int, flashNodeReadDataNodeTimeout int, flashHotKeyMissCount int, - flashReadFlowLimit int64, flashWriteFlowLimit int64, flashKeyFlowLimit int64) (task *proto.AdminTask) { + flashReadFlowLimit int64, flashWriteFlowLimit int64, flashKeyFlowLimit int64, +) (task *proto.AdminTask) { request := &proto.HeartBeatRequest{ CurrTime: time.Now().Unix(), MasterAddr: masterAddr, diff --git a/remotecache/flashgroupmanager/flash_node_topology.go b/remotecache/flashgroupmanager/flash_node_topology.go index f958f653b..a3e9d2264 100644 --- a/remotecache/flashgroupmanager/flash_node_topology.go +++ b/remotecache/flashgroupmanager/flash_node_topology.go @@ -102,7 +102,8 @@ func NewFlashNodeTopology() (t *FlashNodeTopology) { } func (t *FlashNodeTopology) gradualCreateFlashGroup(fgID uint64, syncUpdateFlashGroupFunc SyncUpdateFlashGroupFunc, - setSlots []uint32, setWeight uint32, step uint32) (flashGroup *FlashGroup, err error) { + setSlots []uint32, setWeight uint32, step uint32, +) (flashGroup *FlashGroup, err error) { t.createFlashGroupLock.Lock() defer t.createFlashGroupLock.Unlock() @@ -130,7 +131,8 @@ func (t *FlashNodeTopology) gradualCreateFlashGroup(fgID uint64, syncUpdateFlash } func (t *FlashNodeTopology) createFlashGroup(fgID uint64, syncAddFlashGroupFunc SyncAddFlashGroupFunc, - setSlots []uint32, setWeight uint32) (flashGroup *FlashGroup, err error) { + setSlots []uint32, setWeight uint32, +) (flashGroup *FlashGroup, err error) { t.createFlashGroupLock.Lock() defer t.createFlashGroupLock.Unlock() @@ -291,8 +293,8 @@ func (t *FlashNodeTopology) allocateNewSlotsForCreateFlashGroup(fgID uint64, set if len(slots) > 0 { return } - - for len(slots) < int(weight)*defaultFlashGroupSlotsCount { + target := uint32(defaultFlashGroupSlotsCount) * weight + for uint32(len(slots)) < target { slot := allocateNewSlot() if _, ok := t.slotsMap[slot]; ok { continue @@ -315,7 +317,8 @@ func (t *FlashNodeTopology) GetFlashGroupsAdminView(fgStatus proto.FlashGroupSta } func (t *FlashNodeTopology) gradualRemoveFlashGroup(flashGroup *FlashGroup, - syncUpdateFlashGroupFunc SyncUpdateFlashGroupFunc, step uint32) (err error) { + syncUpdateFlashGroupFunc SyncUpdateFlashGroupFunc, step uint32, +) (err error) { t.createFlashGroupLock.Lock() defer t.createFlashGroupLock.Unlock() @@ -324,7 +327,8 @@ func (t *FlashNodeTopology) gradualRemoveFlashGroup(flashGroup *FlashGroup, func (t *FlashNodeTopology) gradualExpandOrShrinkFlashGroupSlots(flashGroup *FlashGroup, syncUpdateFlashGroupFunc SyncUpdateFlashGroupFunc, newSlotStatus proto.SlotStatus, - pendingSlots []uint32, step uint32) (err error) { + pendingSlots []uint32, step uint32, +) (err error) { flashGroup.lock.Lock() oldSlotStatus := flashGroup.SlotStatus oldStep := flashGroup.Step @@ -416,7 +420,8 @@ func (t *FlashNodeTopology) Clear() { } func (t *FlashNodeTopology) AddFlashNode(clusterName, nodeAddr, zoneName, version string, - allocateCommonIDFunc AllocateCommonIDFunc, syncAddFlashNodeFunc SyncAddFlashNodeFunc) (id uint64, err error) { + allocateCommonIDFunc AllocateCommonIDFunc, syncAddFlashNodeFunc SyncAddFlashNodeFunc, +) (id uint64, err error) { t.mu.Lock() defer func() { t.mu.Unlock() @@ -474,7 +479,8 @@ func (t *FlashNodeTopology) ListFlashNodes(showAll, active bool) map[string][]*p } func (t *FlashNodeTopology) UpdateFlashNode(flashNode *FlashNode, enable bool, - syncUpdateFlashNodeFunc SyncUpdateFlashNodeFunc) (err error) { + syncUpdateFlashNodeFunc SyncUpdateFlashNodeFunc, +) (err error) { flashNode.Lock() defer flashNode.Unlock() if flashNode.IsEnable != enable { @@ -492,7 +498,8 @@ func (t *FlashNodeTopology) UpdateFlashNode(flashNode *FlashNode, enable bool, } func (t *FlashNodeTopology) RemoveFlashNode(clusterName string, flashNode *FlashNode, - syncDeleteFlashNodeFunc SyncDeleteFlashNodeFunc) (err error) { + syncDeleteFlashNodeFunc SyncDeleteFlashNodeFunc, +) (err error) { log.LogWarnf("action[removeFlashNode], ZoneName[%s] Node[%s] offline", flashNode.ZoneName, flashNode.Addr) var flashGroupID uint64 if flashGroupID, err = t.deleteFlashNode(clusterName, flashNode, syncDeleteFlashNodeFunc); err != nil { @@ -523,7 +530,8 @@ func (t *FlashNodeTopology) RemoveFlashNode(clusterName string, flashNode *Flash } func (t *FlashNodeTopology) deleteFlashNode(clusterName string, flashNode *FlashNode, - syncDeleteFlashNodeFunc SyncDeleteFlashNodeFunc) (oldFlashGroupID uint64, err error) { + syncDeleteFlashNodeFunc SyncDeleteFlashNodeFunc, +) (oldFlashGroupID uint64, err error) { flashNode.Lock() defer flashNode.Unlock() oldFlashGroupID = flashNode.FlashGroupID @@ -578,7 +586,8 @@ func (t *FlashNodeTopology) TurnFlashGroup(enabled bool) { func (t *FlashNodeTopology) CreateFlashGroup(id uint64, syncUpdateFlashGroupFunc SyncUpdateFlashGroupFunc, syncAddFlashGroupFunc SyncAddFlashGroupFunc, setSlots []uint32, setWeight uint32, gradualFlag bool, - step uint32) (fg *FlashGroup, err error) { + step uint32, +) (fg *FlashGroup, err error) { if gradualFlag { if fg, err = t.gradualCreateFlashGroup(id, syncUpdateFlashGroupFunc, setSlots, setWeight, step); err != nil { return @@ -595,7 +604,8 @@ func (t *FlashNodeTopology) CreateFlashGroup(id uint64, syncUpdateFlashGroupFunc func (t *FlashNodeTopology) RemoveFlashGroup(id uint64, gradualFlag bool, step uint32, syncUpdateFlashGroupFunc SyncUpdateFlashGroupFunc, syncUpdateFlashNodeFunc SyncUpdateFlashNodeFunc, - syncDeleteFlashGroupFunc SyncDeleteFlashGroupFunc) (flashGroup *FlashGroup, err error) { + syncDeleteFlashGroupFunc SyncDeleteFlashGroupFunc, +) (flashGroup *FlashGroup, err error) { if flashGroup, err = t.GetFlashGroup(id); err != nil { return } @@ -622,7 +632,8 @@ func (t *FlashNodeTopology) RemoveFlashGroup(id uint64, gradualFlag bool, } func (t *FlashNodeTopology) removeAllFlashNodeFromFlashGroup(flashGroup *FlashGroup, - syncUpdateFlashNodeFunc SyncUpdateFlashNodeFunc) (err error) { + syncUpdateFlashNodeFunc SyncUpdateFlashNodeFunc, +) (err error) { flashNodeHosts := flashGroup.GetFlashNodeHosts(false) successHost := make([]string, 0) for _, flashNodeHost := range flashNodeHosts { @@ -638,7 +649,8 @@ func (t *FlashNodeTopology) removeAllFlashNodeFromFlashGroup(flashGroup *FlashGr } func (t *FlashNodeTopology) removeFlashNodeFromFlashGroup(addr string, flashGroup *FlashGroup, - syncUpdateFlashNodeFunc SyncUpdateFlashNodeFunc) (err error) { + syncUpdateFlashNodeFunc SyncUpdateFlashNodeFunc, +) (err error) { var flashNode *FlashNode if flashNode, err = t.setFlashNodeToUnused(addr, flashGroup.ID, syncUpdateFlashNodeFunc); err != nil { return @@ -682,7 +694,8 @@ func (t *FlashNodeTopology) setFlashNodeToUnused(addr string, flashGroupID uint6 } func (t *FlashNodeTopology) FlashGroupAddFlashNode(flashGroupID uint64, addr string, zoneName string, - count int, syncUpdateFlashNodeFunc SyncUpdateFlashNodeFunc) (flashGroup *FlashGroup, err error) { + count int, syncUpdateFlashNodeFunc SyncUpdateFlashNodeFunc, +) (flashGroup *FlashGroup, err error) { if flashGroup, err = t.GetFlashGroup(flashGroupID); err != nil { return } @@ -699,7 +712,8 @@ func (t *FlashNodeTopology) FlashGroupAddFlashNode(flashGroupID uint64, addr str } func (t *FlashNodeTopology) addFlashNodeToFlashGroup(addr string, flashGroup *FlashGroup, - syncUpdateFlashNodeFunc SyncUpdateFlashNodeFunc) (err error) { + syncUpdateFlashNodeFunc SyncUpdateFlashNodeFunc, +) (err error) { var flashNode *FlashNode if flashNode, err = t.setFlashNodeToFlashGroup(addr, flashGroup.ID, syncUpdateFlashNodeFunc); err != nil { return @@ -709,7 +723,8 @@ func (t *FlashNodeTopology) addFlashNodeToFlashGroup(addr string, flashGroup *Fl } func (t *FlashNodeTopology) setFlashNodeToFlashGroup(addr string, flashGroupID uint64, - syncUpdateFlashNodeFunc SyncUpdateFlashNodeFunc) (flashNode *FlashNode, err error) { + syncUpdateFlashNodeFunc SyncUpdateFlashNodeFunc, +) (flashNode *FlashNode, err error) { if flashNode, err = t.PeekFlashNode(addr); err != nil { return } @@ -735,7 +750,8 @@ func (t *FlashNodeTopology) setFlashNodeToFlashGroup(addr string, flashGroupID u } func (t *FlashNodeTopology) selectFlashNodesFromZoneAddToFlashGroup(zoneName string, count int, excludeHosts []string, - flashGroup *FlashGroup, syncUpdateFlashNodeFunc SyncUpdateFlashNodeFunc) (err error) { + flashGroup *FlashGroup, syncUpdateFlashNodeFunc SyncUpdateFlashNodeFunc, +) (err error) { flashNodeZone, err := t.GetZone(zoneName) if err != nil { return @@ -758,7 +774,8 @@ func (t *FlashNodeTopology) selectFlashNodesFromZoneAddToFlashGroup(zoneName str } func (t *FlashNodeTopology) FlashGroupRemoveFlashNode(flashGroupID uint64, addr string, zoneName string, - count int, syncUpdateFlashNodeFunc SyncUpdateFlashNodeFunc) (flashGroup *FlashGroup, err error) { + count int, syncUpdateFlashNodeFunc SyncUpdateFlashNodeFunc, +) (flashGroup *FlashGroup, err error) { if flashGroup, err = t.GetFlashGroup(flashGroupID); err != nil { return } @@ -772,7 +789,8 @@ func (t *FlashNodeTopology) FlashGroupRemoveFlashNode(flashGroupID uint64, addr } func (t *FlashNodeTopology) removeFlashNodesFromTargetZone(zoneName string, count int, - flashGroup *FlashGroup, syncUpdateFlashNodeFunc SyncUpdateFlashNodeFunc) (err error) { + flashGroup *FlashGroup, syncUpdateFlashNodeFunc SyncUpdateFlashNodeFunc, +) (err error) { flashNodeHosts := flashGroup.getTargetZoneFlashNodeHosts(zoneName) if len(flashNodeHosts) < count { return fmt.Errorf("flashNodeHostsCount:%v less than expectCount:%v,flashNodeHosts:%v", len(flashNodeHosts), count, flashNodeHosts) @@ -822,7 +840,8 @@ func (t *FlashNodeTopology) SaveFlashGroup(group *FlashGroup) { } func (t *FlashNodeTopology) CreateFlashNodeHeartBeatTasks(leader string, handleReadTimeout, readDataNodeTimeout, - hotKeyMissCount int, flashReadFlowLimit int64, flashWriteFlowLimit int64, flashKeyFlowLimit int64) []*proto.AdminTask { + hotKeyMissCount int, flashReadFlowLimit int64, flashWriteFlowLimit int64, flashKeyFlowLimit int64, +) []*proto.AdminTask { tasks := make([]*proto.AdminTask, 0) t.flashNodeMap.Range(func(addr, flashNode interface{}) bool { node := flashNode.(*FlashNode) @@ -895,7 +914,8 @@ func (t *FlashNodeTopology) Load() (err error) { } func (t *FlashNodeTopology) UpdateFlashGroupSlots(syncDeleteFlashGroupFunc SyncDeleteFlashGroupFunc, - syncUpdateFlashGroupFunc SyncUpdateFlashGroupFunc, syncUpdateFlashNodeFunc SyncUpdateFlashNodeFunc) { + syncUpdateFlashGroupFunc SyncUpdateFlashGroupFunc, syncUpdateFlashNodeFunc SyncUpdateFlashNodeFunc, +) { isNotUpdated := true t.flashGroupMap.Range(func(key, value interface{}) bool { flashGroup := value.(*FlashGroup) @@ -923,7 +943,8 @@ func (t *FlashNodeTopology) UpdateFlashGroupSlots(syncDeleteFlashGroupFunc SyncD } func (t *FlashNodeTopology) updateFlashGroupSlots(flashGroup *FlashGroup, syncDeleteFlashGroupFunc SyncDeleteFlashGroupFunc, - syncUpdateFlashGroupFunc SyncUpdateFlashGroupFunc, syncUpdateFlashNodeFunc SyncUpdateFlashNodeFunc) (err error) { + syncUpdateFlashGroupFunc SyncUpdateFlashGroupFunc, syncUpdateFlashNodeFunc SyncUpdateFlashNodeFunc, +) (err error) { var needDeleteFgFlag bool if flashGroup.GetSlotStatus() == proto.SlotStatus_Deleting { @@ -935,7 +956,8 @@ func (t *FlashNodeTopology) updateFlashGroupSlots(flashGroup *FlashGroup, syncDe } func (t *FlashNodeTopology) checkShrinkOrDeleteFlashGroup(flashGroup *FlashGroup, - syncUpdateFlashNodeFunc SyncUpdateFlashNodeFunc) (needDeleteFgFlag bool, err error) { + syncUpdateFlashNodeFunc SyncUpdateFlashNodeFunc, +) (needDeleteFgFlag bool, err error) { leftPendingSlotsNum := uint32(flashGroup.GetPendingSlotsCount()) - flashGroup.Step if (leftPendingSlotsNum <= 0) && (flashGroup.GetPendingSlotsCount() == flashGroup.GetSlotsCount()) { needDeleteFgFlag = true diff --git a/remotecache/flashgroupmanager/http_server.go b/remotecache/flashgroupmanager/http_server.go index 867e97f63..0f23d9c09 100644 --- a/remotecache/flashgroupmanager/http_server.go +++ b/remotecache/flashgroupmanager/http_server.go @@ -2,6 +2,7 @@ package flashgroupmanager import ( "fmt" + "math" "net/http" "net/http/httputil" "time" @@ -162,8 +163,14 @@ func (m *FlashGroupManager) registerAPIMiddleware(route *mux.Router) { func (m *FlashGroupManager) newReverseProxy() *httputil.ReverseProxy { tr := &http.Transport{} if m.config != nil { + ps := m.config.httpProxyPoolSize // uint64 + if ps > uint64(math.MaxInt) { + log.LogErrorf("newReverseProxy httpProxyPoolSize[%d] invalid", ps) + return nil + } + poolSize := int(ps) tr = proto.GetHttpTransporter(&proto.HttpCfg{ - PoolSize: int(m.config.httpProxyPoolSize), + PoolSize: poolSize, }) } @@ -177,5 +184,7 @@ func (m *FlashGroupManager) newReverseProxy() *httputil.ReverseProxy { } func (m *FlashGroupManager) proxy(w http.ResponseWriter, r *http.Request) { - m.reverseProxy.ServeHTTP(w, r) + if m.reverseProxy != nil { + m.reverseProxy.ServeHTTP(w, r) + } } diff --git a/remotecache/flashgroupmanager/server.go b/remotecache/flashgroupmanager/server.go index c80e44145..f35a402aa 100644 --- a/remotecache/flashgroupmanager/server.go +++ b/remotecache/flashgroupmanager/server.go @@ -4,6 +4,7 @@ import ( "context" "fmt" syslog "log" + "math" "net/http" "net/http/httputil" "strconv" @@ -230,13 +231,25 @@ func (m *FlashGroupManager) initCluster() { } func (m *FlashGroupManager) createRaftServer(cfg *config.Config) (err error) { + hp := m.config.heartbeatPort + rp := m.config.replicaPort + + if hp < 0 || hp > 65535 { + return fmt.Errorf("invalid heartbeatPort: %d", hp) + } + if rp < 0 || rp > 65535 { + return fmt.Errorf("invalid replicaPort: %d", rp) + } + if hp > int64(math.MaxInt) || rp > int64(math.MaxInt) { + return fmt.Errorf("port overflows int") + } raftCfg := &raftstore.Config{ NodeID: m.id, RaftPath: m.walDir, IPAddr: cfg.GetString(IP), NumOfLogsToRetain: m.retainLogs, - HeartbeatPort: int(m.config.heartbeatPort), - ReplicaPort: int(m.config.replicaPort), + HeartbeatPort: int(hp), + ReplicaPort: int(rp), TickInterval: m.tickInterval, ElectionTick: m.electionTick, RecvBufSize: m.raftRecvBufSize, diff --git a/remotecache/flashnode/flashnode.go b/remotecache/flashnode/flashnode.go index 0764080d3..6c2a1c032 100644 --- a/remotecache/flashnode/flashnode.go +++ b/remotecache/flashnode/flashnode.go @@ -795,8 +795,8 @@ func unescapeMountField(s string) string { var b strings.Builder for i := 0; i < len(s); i++ { if s[i] == '\\' && i+3 < len(s) { - if o, err := strconv.ParseInt(s[i+1:i+4], 8, 0); err == nil { - b.WriteByte(byte(o)) + if u, err := strconv.ParseUint(s[i+1:i+4], 8, 8); err == nil { + b.WriteByte(byte(u)) i += 3 continue } diff --git a/sdk/data/stream/extent_client.go b/sdk/data/stream/extent_client.go index 6c980062a..0ec503b58 100644 --- a/sdk/data/stream/extent_client.go +++ b/sdk/data/stream/extent_client.go @@ -710,7 +710,8 @@ func (client *ExtentClient) SetFileSize(inode uint64, size int, sync bool) { // Write writes the data. func (client *ExtentClient) Write(inode uint64, offset int, data []byte, flags int, checkFunc func() error, - storageClass uint32, isMigration, waitForFlush bool) (write int, err error) { + storageClass uint32, isMigration, waitForFlush bool, +) (write int, err error) { prefix := fmt.Sprintf("Write{ino(%v)offset(%v)size(%v)}", inode, offset, len(data)) s := client.GetStreamer(inode) if s == nil {