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(master): add triggerCondition for dataNode balance

Browse Source 
@formatter:off

Signed-off-by: chihe <chihe@oppo.com>
This commit is contained in:
chihe 2025-12-24 14:26:21 +08:00
parent aa70981991
commit 81643df8cb
11 changed files with 97 additions and 168 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

133
master/cluster_balance.go
View File

@ -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
},
})
}

2
master/data_partition.go
View File

@ -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

3
master/flash_group.go
View File

@ -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())

11
remotecache/flashgroupmanager/config.go
View File

@ -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

6
remotecache/flashgroupmanager/flash_group.go
View File

@ -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

3
remotecache/flashgroupmanager/flash_node.go
View File

@ -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,

70
remotecache/flashgroupmanager/flash_node_topology.go
View File

@ -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

13
remotecache/flashgroupmanager/http_server.go
View File

@ -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)
}
}

17
remotecache/flashgroupmanager/server.go
View File

@ -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,

4
remotecache/flashnode/flashnode.go
View File

@ -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
}

3
sdk/data/stream/extent_client.go
View File

@ -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 {