style(api): improve code comments for clarity and consistency. #3942

- Add spaces after comment slashes for better readability
- Standardize comment formatting across multiple files
- Enhance clarity of comments in data partition handling

Signed-off-by: Victor1319 <zengxuewei@oppo.com>
This commit is contained in:
Victor1319 2025-11-11 19:37:05 +08:00 committed by Victor1319
parent 655dfbf4d9
commit 070b82370d
5 changed files with 140 additions and 150 deletions

View File

@ -604,126 +604,126 @@ func (m *Server) offlineMetaNode(w http.ResponseWriter, r *http.Request) {
sendOkReply(w, r, newSuccessHTTPReply(rstMsg))
}
func parseNodeBalanceInfoParams(r *http.Request) (params map[string]interface{}, err error) {
if err = r.ParseForm(); err != nil {
return
}
var value string
noParams := true
// func parseNodeBalanceInfoParams(r *http.Request) (params map[string]interface{}, err error) {
// if err = r.ParseForm(); err != nil {
// return
// }
// var value string
// noParams := true
if value = r.FormValue(dataNodeBalanceByDiskUsageHighKey); value != "" {
noParams = false
var val float64
val, err = strconv.ParseFloat(value, 64)
if err != nil {
err = unmatchedKey(dataNodeBalanceByDiskUsageHighKey)
return
}
params[dataNodeBalanceByDiskUsageHighKey] = val
}
// if value = r.FormValue(dataNodeBalanceByDiskUsageHighKey); value != "" {
// noParams = false
// var val float64
// val, err = strconv.ParseFloat(value, 64)
// if err != nil {
// err = unmatchedKey(dataNodeBalanceByDiskUsageHighKey)
// return
// }
// params[dataNodeBalanceByDiskUsageHighKey] = val
// }
if value = r.FormValue(dataNodeBalanceByDiskUsageLowKey); value != "" {
noParams = false
var val float64
val, err = strconv.ParseFloat(value, 64)
if err != nil {
err = unmatchedKey(dataNodeBalanceByDiskUsageLowKey)
return
}
params[dataNodeBalanceByDiskUsageLowKey] = val
}
// if value = r.FormValue(dataNodeBalanceByDiskUsageLowKey); value != "" {
// noParams = false
// var val float64
// val, err = strconv.ParseFloat(value, 64)
// if err != nil {
// err = unmatchedKey(dataNodeBalanceByDiskUsageLowKey)
// return
// }
// params[dataNodeBalanceByDiskUsageLowKey] = val
// }
if value = r.FormValue(dataNodeBalanceByDPCountHighKey); value != "" {
noParams = false
var val int64
val, err = strconv.ParseInt(value, 10, 32)
if err != nil {
err = unmatchedKey(dataNodeBalanceByDPCountHighKey)
return
}
params[dataNodeBalanceByDPCountHighKey] = val
}
// if value = r.FormValue(dataNodeBalanceByDPCountHighKey); value != "" {
// noParams = false
// var val int64
// val, err = strconv.ParseInt(value, 10, 32)
// if err != nil {
// err = unmatchedKey(dataNodeBalanceByDPCountHighKey)
// return
// }
// params[dataNodeBalanceByDPCountHighKey] = val
// }
if value = r.FormValue(dataNodeBalanceByDPCountLowKey); value != "" {
noParams = false
var val int64
val, err = strconv.ParseInt(value, 10, 32)
if err != nil {
err = unmatchedKey(dataNodeBalanceByDPCountLowKey)
return
}
params[dataNodeBalanceByDPCountLowKey] = val
}
// if value = r.FormValue(dataNodeBalanceByDPCountLowKey); value != "" {
// noParams = false
// var val int64
// val, err = strconv.ParseInt(value, 10, 32)
// if err != nil {
// err = unmatchedKey(dataNodeBalanceByDPCountLowKey)
// return
// }
// params[dataNodeBalanceByDPCountLowKey] = val
// }
if noParams {
err = fmt.Errorf("no key assigned")
return
}
// if noParams {
// err = fmt.Errorf("no key assigned")
// return
// }
return
}
// return
// }
func (m *Server) setDataNodeBalanceInfoHandler(w http.ResponseWriter, r *http.Request) {
var (
params map[string]interface{}
err error
val interface{}
keyExists bool
newDataNodeBalanceByDiskUsageLow float64
newDataNodeBalanceByDiskUsageHigh float64
newDataNodeBalanceByDPCountLow uint32
newDataNodeBalanceByDPCountHigh uint32
)
metric := exporter.NewTPCnt(apiToMetricsName(proto.SetDataNodeBalanceInfo))
defer func() {
doStatAndMetric(proto.SetDataNodeBalanceInfo, metric, err, nil)
AuditLog(r, proto.SetDataNodeBalanceInfo, fmt.Sprintf("params: %v", params), err)
}()
// func (m *Server) setDataNodeBalanceInfoHandler(w http.ResponseWriter, r *http.Request) {
// var (
// params map[string]interface{}
// err error
// val interface{}
// keyExists bool
// newDataNodeBalanceByDiskUsageLow float64
// newDataNodeBalanceByDiskUsageHigh float64
// newDataNodeBalanceByDPCountLow uint32
// newDataNodeBalanceByDPCountHigh uint32
// )
// metric := exporter.NewTPCnt(apiToMetricsName(proto.SetDataNodeBalanceInfo))
// defer func() {
// doStatAndMetric(proto.SetDataNodeBalanceInfo, metric, err, nil)
// AuditLog(r, proto.SetDataNodeBalanceInfo, fmt.Sprintf("params: %v", params), err)
// }()
if params, err = parseNodeBalanceInfoParams(r); err != nil {
return
}
// if params, err = parseNodeBalanceInfoParams(r); err != nil {
// return
// }
if val, keyExists = params[dataNodeBalanceByDiskUsageLowKey]; keyExists {
if v, keyExists := val.(float64); keyExists {
newDataNodeBalanceByDiskUsageLow = v
}
}
// if val, keyExists = params[dataNodeBalanceByDiskUsageLowKey]; keyExists {
// if v, keyExists := val.(float64); keyExists {
// newDataNodeBalanceByDiskUsageLow = v
// }
// }
if val, keyExists = params[dataNodeBalanceByDiskUsageHighKey]; keyExists {
if v, keyExists := val.(float64); keyExists {
newDataNodeBalanceByDiskUsageHigh = v
}
}
// if val, keyExists = params[dataNodeBalanceByDiskUsageHighKey]; keyExists {
// if v, keyExists := val.(float64); keyExists {
// newDataNodeBalanceByDiskUsageHigh = v
// }
// }
if val, keyExists = params[dataNodeBalanceByDPCountLowKey]; keyExists {
if v, keyExists := val.(uint32); keyExists {
newDataNodeBalanceByDPCountLow = v
}
}
// if val, keyExists = params[dataNodeBalanceByDPCountLowKey]; keyExists {
// if v, keyExists := val.(uint32); keyExists {
// newDataNodeBalanceByDPCountLow = v
// }
// }
if val, keyExists = params[dataNodeBalanceByDPCountHighKey]; keyExists {
if v, keyExists := val.(uint32); keyExists {
newDataNodeBalanceByDPCountHigh = v
}
}
// if val, keyExists = params[dataNodeBalanceByDPCountHighKey]; keyExists {
// if v, keyExists := val.(uint32); keyExists {
// newDataNodeBalanceByDPCountHigh = v
// }
// }
//check parameters
if newDataNodeBalanceByDiskUsageHigh <= newDataNodeBalanceByDiskUsageLow {
err = fmt.Errorf("DataNodeBalanceByDiskUsageHigh is lower than DataNodeBalanceByDiskUsageHigh")
sendErrReply(w, r, newErrHTTPReply(err))
return
}
// // check parameters
// if newDataNodeBalanceByDiskUsageHigh <= newDataNodeBalanceByDiskUsageLow {
// err = fmt.Errorf("DataNodeBalanceByDiskUsageHigh is lower than DataNodeBalanceByDiskUsageHigh")
// sendErrReply(w, r, newErrHTTPReply(err))
// return
// }
if newDataNodeBalanceByDPCountHigh <= newDataNodeBalanceByDPCountLow {
err = fmt.Errorf("DataNodeBalanceByDPCountHigh is lower than DataNodeBalanceByDiskUsageLow")
sendErrReply(w, r, newErrHTTPReply(err))
return
}
// if newDataNodeBalanceByDPCountHigh <= newDataNodeBalanceByDPCountLow {
// err = fmt.Errorf("DataNodeBalanceByDPCountHigh is lower than DataNodeBalanceByDiskUsageLow")
// sendErrReply(w, r, newErrHTTPReply(err))
// return
// }
//set the parameters
m.cluster.cfg.DataNodeBalanceByDiskUsageLow = newDataNodeBalanceByDiskUsageLow
m.cluster.cfg.DataNodeBalanceByDiskUsageHigh = newDataNodeBalanceByDiskUsageHigh
m.cluster.cfg.DataNodeBalanceByDPCountLow = newDataNodeBalanceByDPCountLow
m.cluster.cfg.DataNodeBalanceByDPCountHigh = newDataNodeBalanceByDPCountHigh
}
// // set the parameters
// m.cluster.cfg.DataNodeBalanceByDiskUsageLow = newDataNodeBalanceByDiskUsageLow
// m.cluster.cfg.DataNodeBalanceByDiskUsageHigh = newDataNodeBalanceByDiskUsageHigh
// m.cluster.cfg.DataNodeBalanceByDPCountLow = newDataNodeBalanceByDPCountLow
// m.cluster.cfg.DataNodeBalanceByDPCountHigh = newDataNodeBalanceByDPCountHigh
// }

View File

@ -1848,9 +1848,7 @@ func (c *Cluster) getOverLoadDataNodesByDiskUsage() []*DataNode {
// the selection critieria is
// the target nodeset has at least numCopies underload dataNodes, and these underload datanodes do not hold the replica of dp
func (c *Cluster) getUnderLoadNodeSetByDiskUsage(dp *DataPartition, zone string, excludedNodeSet uint64, numCopies int) (nodeset *nodeSet, err error) {
var (
z *Zone
)
var z *Zone
z, err = c.t.getZone(zone)
if err != nil {
@ -1879,9 +1877,7 @@ func (c *Cluster) getUnderLoadNodeSetByDiskUsage(dp *DataPartition, zone string,
// get underloadDataNodes within given nodeset that do not hold replica of dp
func (c *Cluster) getUnderLoadNodesInNodeSetByDiskUsage(dp *DataPartition, nodeset uint64) (underloadDataNodes []*DataNode, err error) {
var (
ns *nodeSet
)
var ns *nodeSet
if ns, err = c.t.getNodeSetByNodeSetId(nodeset); err != nil {
return nil, fmt.Errorf("getUnderLoadNodesInNodeSetByDiskUsage: failed to find nodeset %v", nodeset)
@ -1911,9 +1907,7 @@ func (c *Cluster) getOverLoadDataNodesByDPCount() []*DataNode {
}
func (c *Cluster) getUnderLoadNodeSetByDPCount(dp *DataPartition, zone string, excludedNodeSet uint64, numCopies int) (nodeset *nodeSet, err error) {
var (
z *Zone
)
var z *Zone
z, err = c.t.getZone(zone)
if err != nil {
@ -1941,9 +1935,7 @@ func (c *Cluster) getUnderLoadNodeSetByDPCount(dp *DataPartition, zone string, e
}
func (c *Cluster) getUnderLoadNodesInNodeSetByDPCount(dp *DataPartition, nodeset uint64) (underloadDataNodes []*DataNode, err error) {
var (
ns *nodeSet
)
var ns *nodeSet
if ns, err = c.t.getNodeSetByNodeSetId(nodeset); err != nil {
return nil, fmt.Errorf("getUnderLoadNodesInNodeSetByDPCount failed: %v", err)
@ -1983,9 +1975,7 @@ func (c *Cluster) postBalanceDecommissionSucccess(BalanceType uint32, dp *DataPa
}
func (c *Cluster) getReplicaHostsInNodeSet(dp *DataPartition, nodeset uint64) []*DataNode {
var (
hosts []*DataNode
)
var hosts []*DataNode
for _, replica := range dp.Replicas {
if replica.dataNode.NodeSetID == nodeset {
@ -1998,9 +1988,9 @@ func (c *Cluster) getReplicaHostsInNodeSet(dp *DataPartition, nodeset uint64) []
func (c *Cluster) handleDataNodeBalanceByDiskUsage(d *DataNode) (err error) {
var (
sizeToBalance uint64 = 0
sizeMarkedDecomm uint64 = 0
numDPMarkedDecomm uint64 = 0
sizeToBalance uint64
sizeMarkedDecomm uint64
numDPMarkedDecomm uint64
dp *DataPartition
)
@ -2035,9 +2025,9 @@ func (c *Cluster) handleDataNodeBalanceByDiskUsage(d *DataNode) (err error) {
func (c *Cluster) handleDataNodeBalanceByDPCount(d *DataNode) (err error) {
var (
numDPsToBalance uint32 = 0
sizeMarkedDecomm uint64 = 0
numDPMarkedDecomm uint32 = 0
numDPsToBalance uint32
sizeMarkedDecomm uint64
numDPMarkedDecomm uint32
dp *DataPartition
)

View File

@ -110,11 +110,11 @@ const (
// DefaultMetaPartitionMissSec = 3600
//
defaultDataNodeBalanceOn = false
defaultDataNodeBalanceInterval = 10 * 60 //interval to perform datanode balance task
defaultDataNodeBalanceByDiskUsageHigh = 0.8 //high usage by disk usage
defaultDataNodeBalanceByDiskUsageLow = 0.1 //low usage by disk usage
defaultDataNodeBalanceByDPCountHigh = 3000 //high usage by DP count
defaultDataNodeBalanceByDPCountLow = 1000 //low usage by DP count
defaultDataNodeBalanceInterval = 10 * 60 // interval to perform datanode balance task
defaultDataNodeBalanceByDiskUsageHigh = 0.8 // high usage by disk usage
defaultDataNodeBalanceByDiskUsageLow = 0.1 // low usage by disk usage
defaultDataNodeBalanceByDPCountHigh = 3000 // high usage by DP count
defaultDataNodeBalanceByDPCountLow = 1000 // low usage by DP count
defaultIntervalToAlarmMissingMetaPartition = 10 * 60 // interval of checking if a replica is missing
defaultMetaPartitionMemUsageThreshold float32 = 0.75 // memory usage threshold on a meta partition

View File

@ -2222,29 +2222,29 @@ func (partition *DataPartition) TryAcquireDecommissionToken(c *Cluster) bool {
partition.decommissionInfo(), partition.DecommissionDstAddr, time.Since(begin).String(), err, result)
}()
//find target for balancing
// find target for balancing
if partition.DecommissionType == BalanceByDPCount || partition.DecommissionType == BalanceByDiskUsage {
if datanode, err = c.dataNode(partition.DecommissionSrcAddr); err != nil {
log.LogWarnf("action[TryAcquireDecommissionToken]dp %v find given datanode %v failed:%v",
partition.PartitionID, partition.DecommissionSrcAddr, err.Error())
goto errHandler
}
//first replica, determine which nodeset to transfer to
// first replica, determine which nodeset to transfer to
if partition.DecommissionDstNodeSet == 0 {
//mark success, remove in advance
// mark success, remove in advance
if !c.isOverloadDataNode(partition.DecommissionType, datanode) {
partition.SetDecommissionStatus(DecommissionSuccess)
log.LogInfof("action[TryAcquireDecommissionToken]dp %v canceled because src node is no longer overload",
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
}
//first find underload node in same nodeset
// first find underload node in same nodeset
if underloadDataNodes, err = c.getUnderLoadNodesInNodeSet(partition.DecommissionType, partition, datanode.NodeSetID); err != nil {
log.LogWarnf("action[TryAcquireDecommissionToken]dp %v failed: %v", err)
log.LogWarnf("action[TryAcquireDecommissionToken]dp %v failed: %v", partition.PartitionID, err)
goto errHandler
}
if len(underloadDataNodes) != 0 {
//find nodeset
// find nodeset
ns, err = c.t.getNodeSetByNodeSetId(datanode.NodeSetID)
if err != nil {
log.LogWarnf("action[TryAcquireDecommissionToken]dp %v find given dst nodeset %v failed:%v",
@ -2252,15 +2252,15 @@ func (partition *DataPartition) TryAcquireDecommissionToken(c *Cluster) bool {
goto errHandler
}
} else {
//if no underload nodes in src nodeset, try find another nodeset
// if no underload nodes in src nodeset, try find another nodeset
hostsInNodeSet := c.getReplicaHostsInNodeSet(partition, datanode.NodeSetID)
if ns, err = c.getUnderLoadNodeSet(partition.DecommissionType, partition, datanode.ZoneName, datanode.NodeSetID, len(hostsInNodeSet)); err != nil {
log.LogWarnf("action[TryAcquireDecommissionToken]dp %v find underload nodeset failed:%v",
partition.PartitionID, partition.DecommissionDstNodeSet, err.Error())
log.LogWarnf("action[TryAcquireDecommissionToken]dp %v find underload nodeset failed:%v, src node %v, nodeset %v",
partition.PartitionID, partition.DecommissionDstNodeSet, err.Error(), datanode.NodeSetID)
goto errHandler
}
if underloadDataNodes, err = c.getUnderLoadNodesInNodeSet(partition.DecommissionType, partition, ns.ID); err != nil {
log.LogWarnf("action[TryAcquireDecommissionToken]dp %v failed: %v", err)
log.LogWarnf("action[TryAcquireDecommissionToken]dp %v failed: %v", partition.PartitionID, err)
goto errHandler
}
}
@ -2272,16 +2272,16 @@ func (partition *DataPartition) TryAcquireDecommissionToken(c *Cluster) bool {
goto errHandler
}
if underloadDataNodes, err = c.getUnderLoadNodesInNodeSet(partition.DecommissionType, partition, partition.DecommissionDstNodeSet); err != nil {
log.LogWarnf("action[TryAcquireDecommissionToken]dp %v failed: %v", err)
log.LogWarnf("action[TryAcquireDecommissionToken]dp %v failed: %v", partition.PartitionID, err)
goto errHandler
}
}
//can't find any underloadnodes
// can't find any underloadnodes
if len(underloadDataNodes) == 0 {
log.LogWarnf("action[TryAcquireDecommissionToken]dp %v failed: can't find underload nodes in nodeset %v", partition.PartitionID, ns.ID)
goto errHandler
}
//to this point, ns and its underload nodes are decided
// to this point, ns and its underload nodes are decided
if ns.AcquireDecommissionToken(partition.PartitionID) {
partition.DecommissionDstAddr = underloadDataNodes[0].Addr
partition.DecommissionDstNodeSet = underloadDataNodes[0].NodeSetID

View File

@ -2345,8 +2345,8 @@ func (l *DecommissionDataPartitionList) handleDpTraverseToReleaseToken(dp *DataP
if datanode, err := c.dataNode(dp.DecommissionSrcAddr); err != nil {
log.LogWarnf("action[DecommissionListTraverse]ns %v(%p) dp[%v] failed to find datanode %v err %v", l.nsId, l, dp.decommissionInfo(), dp.DecommissionSrcAddr, err)
} else {
//when dp.DecommissionDstNodeSet == 0
//the balance of this dp is canceled and will be cleaned afterwards
// when dp.DecommissionDstNodeSet == 0
// the balance of this dp is canceled and will be cleaned afterwards
balanceType := dp.DecommissionType
dstNodeSet := dp.DecommissionDstNodeSet
defer c.postBalanceDecommissionSucccess(balanceType, dp, datanode, dstNodeSet)