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

feat(master): Modify some variable names and function parameters. #1000239501

Browse Source 
1. Set distributionOptimizationThreshold as a global variable
2. Add select type in selectParam to get different types of thresholds
3. Shorten some variable names

Signed-off-by: zhumingze <zhumingze@oppo.com>
This commit is contained in:
zhumingze 2025-10-24 15:07:34 +08:00 committed by 常亮
parent bb05433a85
commit 35862a45e4
21 changed files with 446 additions and 425 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
cli/cmd/cluster.go
View File

@ -294,7 +294,7 @@ func newClusterSetParasCmd(client *master.MasterClient) *cobra.Command {
readDataNodeTimeout := ""
rackAware := ""
// Distribution optimization parameters
distributionOptimizationConcurrentDpCount := ""
distributionOptimizationConDpCnt := ""
distributionOptimizationThreshold := ""
cmd := &cobra.Command{
Use: CliOpSetCluster,
@ -450,11 +450,11 @@ func newClusterSetParasCmd(client *master.MasterClient) *cobra.Command {
}
// Validate distribution optimization parameters
if distributionOptimizationConcurrentDpCount != "" {
if distributionOptimizationConDpCnt != "" {
var count int64
count, err = strconv.ParseInt(distributionOptimizationConcurrentDpCount, 10, 64)
count, err = strconv.ParseInt(distributionOptimizationConDpCnt, 10, 64)
if err != nil {
err = fmt.Errorf("param distributionOptimizationConcurrentDpCount(%v) should be int", distributionOptimizationConcurrentDpCount)
err = fmt.Errorf("param distributionOptimizationConcurrentDpCount(%v) should be int", distributionOptimizationConDpCnt)
return
}
if count <= 0 {
@ -482,7 +482,7 @@ func newClusterSetParasCmd(client *master.MasterClient) *cobra.Command {
autoDpMetaRepair, autoDpMetaRepairParallelCnt, autoDistributionOptimization,
dpRepairTimeout, dpTimeout, mpTimeout, dpBackupTimeout, decommissionDpLimit, decommissionDiskLimit,
forbidWriteOpOfProtoVersion0, dataMediaType, handleTimeout, readDataNodeTimeout, rackAware,
distributionOptimizationConcurrentDpCount, distributionOptimizationThreshold); err != nil {
distributionOptimizationConDpCnt, distributionOptimizationThreshold); err != nil {
return
}
stdout("Cluster parameters has been set successfully. \n")
@ -519,7 +519,7 @@ func newClusterSetParasCmd(client *master.MasterClient) *cobra.Command {
cmd.Flags().StringVar(&readDataNodeTimeout, "flashNodeReadDataNodeTimeout", "", "Specify flash node read data node timeout (example:3000ms)")
cmd.Flags().StringVar(&rackAware, CliFlagRackAware, "", "Set rack aware level: 0(none), 1(weak), 2(strong)")
// Distribution optimization parameters
cmd.Flags().StringVar(&distributionOptimizationConcurrentDpCount, CliFlagDistributionOptimizationConcurrentDpCount, "", "Concurrent data partition count for distribution optimization")
cmd.Flags().StringVar(&distributionOptimizationConDpCnt, CliFlagDistributionOptimizationConDpCnt, "", "Concurrent data partition count for distribution optimization")
cmd.Flags().StringVar(&distributionOptimizationThreshold, CliFlagDistributionOptimizationThreshold, "", "Threshold for distribution optimization (0.0-1.0)")
return cmd
}

14
cli/cmd/const.go
View File

@ -161,13 +161,13 @@ const (
CliFlagAllowedStorageClass = "allowedStorageClass"
CliFlagVolStorageClass = "volStorageClass"
// Distribution Optimization parameters
CliFlagDistributionOptimizationConcurrentDpCount = "distributionOptimizationConcurrentDpCount"
CliFlagDistributionOptimizationThreshold = "distributionOptimizationThreshold"
CliFlagMediaType = "mediaType"
CliForbidWriteOpOfProtoVersion0 = "forbidWriteOpOfProtoVersion0"
CliFlagVolQuotaClass = "quotaClass"
CliFlagVolQuotaOfClass = "quotaOfStorageClass"
CliFlagStoreMode = "store-mode"
CliFlagDistributionOptimizationConDpCnt = "distributionOptimizationConDpCnt"
CliFlagDistributionOptimizationThreshold = "distributionOptimizationThreshold"
CliFlagMediaType = "mediaType"
CliForbidWriteOpOfProtoVersion0 = "forbidWriteOpOfProtoVersion0"
CliFlagVolQuotaClass = "quotaClass"
CliFlagVolQuotaOfClass = "quotaOfStorageClass"
CliFlagStoreMode = "store-mode"
CliFlagRemoteCacheEnable = "remoteCacheEnable"
CliFlagRemoteCachePath = "remoteCachePath"

4
cli/cmd/fmt.go
View File

@ -92,7 +92,9 @@ func formatClusterView(cv *proto.ClusterView, cn *proto.ClusterNodeInfo, cp *pro
sb.WriteString(fmt.Sprintf(" AutoDecommissionDiskInterval : %v\n", cv.AutoDecommissionDiskInterval))
sb.WriteString(fmt.Sprintf(" EnableAutoDpMetaRepair : %v\n", cv.EnableAutoDpMetaRepair))
sb.WriteString(fmt.Sprintf(" AutoDpMetaRepairParallelCnt : %v\n", cv.AutoDpMetaRepairParallelCnt))
sb.WriteString(fmt.Sprintf(" EnableAutoDistributionOptimization : %v\n", cv.EnableAutoDistributionOptimization))
sb.WriteString(fmt.Sprintf(" EnableDistributionOptimization : %v\n", cv.EnableDistributionOptimization))
sb.WriteString(fmt.Sprintf(" DistributionOptimizationThreshold : %v\n", cv.DistributionOptimizationThreshold))
sb.WriteString(fmt.Sprintf(" DistributionOptimizationConDpCnt : %v\n", cv.DistributionOptimizationConDpCnt))
sb.WriteString(fmt.Sprintf(" MarkDiskBrokenThreshold : %v\n", strutil.FormatPercent(cv.MarkDiskBrokenThreshold)))
sb.WriteString(fmt.Sprintf(" DecommissionFirstHostDiskParallelLimit : %v\n", cv.DecommissionFirstHostDiskParallelLimit))
sb.WriteString(fmt.Sprintf(" DecommissionDpLimit : %v\n", cv.DecommissionLimit))

6
master/api_args_parse.go
View File

@ -1674,15 +1674,15 @@ func parseAndExtractSetNodeInfoParams(r *http.Request) (params map[string]interf
}
// distribution optimization configs
if value = r.FormValue(distributionOptimizationConcurrentDpCountKey); value != "" {
if value = r.FormValue(distributionOptimizationConDpCntKey); value != "" {
noParams = false
val := int64(0)
val, err = strconv.ParseInt(value, 10, 64)
if err != nil {
err = unmatchedKey(distributionOptimizationConcurrentDpCountKey)
err = unmatchedKey(distributionOptimizationConDpCntKey)
return
}
params[distributionOptimizationConcurrentDpCountKey] = val
params[distributionOptimizationConDpCntKey] = val
}
if value = r.FormValue(distributionOptimizationThresholdKey); value != "" {

12
master/api_service.go
View File

@ -1016,7 +1016,9 @@ func (m *Server) getCluster(w http.ResponseWriter, r *http.Request) {
AutoDpMetaRepairParallelCnt: m.cluster.GetAutoDpMetaRepairParallelCnt(),
EnableAutoDecommission: m.cluster.AutoDecommissionDiskIsEnabled(),
AutoDecommissionDiskInterval: m.cluster.GetAutoDecommissionDiskInterval().String(),
EnableAutoDistributionOptimization: m.cluster.getEnableAutoDistributionOptimization(),
EnableDistributionOptimization: m.cluster.getEnableDistributionOptimization(),
DistributionOptimizationThreshold: getDistributionOptimizationThreshold(),
DistributionOptimizationConDpCnt: m.cluster.DistributionOptimizationConDpCnt.Load(),
DecommissionLimit: atomic.LoadUint64(&m.cluster.DecommissionLimit),
DecommissionFirstHostDiskParallelLimit: atomic.LoadUint64(&m.cluster.DecommissionFirstHostDiskParallelLimit),
DecommissionDiskLimit: m.cluster.GetDecommissionDiskLimit(),
@ -3934,7 +3936,7 @@ func (m *Server) setNodeInfoHandler(w http.ResponseWriter, r *http.Request) {
if val, ok := params[autoDistributionOptimizationKey]; ok {
if autoDistributionOptimization, ok := val.(bool); ok {
if err = m.cluster.setEnableAutoDistributionOptimization(autoDistributionOptimization); err != nil {
if err = m.cluster.setEnableDistributionOptimization(autoDistributionOptimization); err != nil {
sendErrReply(w, r, newErrHTTPReply(err))
return
}
@ -3942,9 +3944,9 @@ func (m *Server) setNodeInfoHandler(w http.ResponseWriter, r *http.Request) {
}
// distribution optimization configs (new keys)
if val, ok := params[distributionOptimizationConcurrentDpCountKey]; ok {
if val, ok := params[distributionOptimizationConDpCntKey]; ok {
if v, ok := val.(int64); ok {
if err = m.cluster.setDistributionOptimizationConcurrentDpCount(v); err != nil {
if err = m.cluster.setDistributionOptimizationConDpCnt(v); err != nil {
sendErrReply(w, r, newErrHTTPReply(err))
return
}
@ -8684,7 +8686,7 @@ func (m *Server) setDistributionOptimizationEnable(w http.ResponseWriter, r *htt
return
}
if err = m.cluster.setEnableAutoDistributionOptimization(enable); err != nil {
if err = m.cluster.setEnableDistributionOptimization(enable); err != nil {
log.LogErrorf("action[setDistributionOptimizationEnable] setEnableDistributionOptimization failed %v", err)
sendErrReply(w, r, newErrHTTPReply(proto.ErrPersistenceByRaft))
return

88
master/cluster.go
View File

@ -52,8 +52,9 @@ const (
)
var (
clusterDpCntLimit uint64
clusterMpCntLimit uint64
clusterDpCntLimit uint64
clusterMpCntLimit uint64
distributionOptimizationThreshold atomicutil.Float64
)
// nolint: structcheck
@ -123,17 +124,16 @@ type ClusterDecommission struct {
MarkDiskBrokenThreshold atomicutil.Float64
badPartitionMutex sync.RWMutex // BadDataPartitionIds and BadMetaPartitionIds operate mutex
ForbidMpDecommission bool
EnableAutoDpMetaRepair atomicutil.Bool
EnableAutoDecommissionDisk atomicutil.Bool
AutoDecommissionInterval atomicutil.Int64
AutoDpMetaRepairParallelCnt atomicutil.Uint32
EnableAutoDistributionOptimization atomicutil.Bool
DistributionOptimizationConcurrentDpCount atomicutil.Int64
DistributionOptimizationThreshold atomicutil.Float64
NodeSetUnbalancedDPs atomicutil.Int64
RackConflictDPs atomicutil.Int64
server *Server
ForbidMpDecommission bool
EnableAutoDpMetaRepair atomicutil.Bool
EnableAutoDecommissionDisk atomicutil.Bool
AutoDecommissionInterval atomicutil.Int64
AutoDpMetaRepairParallelCnt atomicutil.Uint32
EnableDistributionOptimization atomicutil.Bool
DistributionOptimizationConDpCnt atomicutil.Int64
NodeSetUnbalancedDPs atomicutil.Int64
RackConflictDPs atomicutil.Int64
server *Server
}
type CleanTask struct {
@ -459,6 +459,9 @@ func newCluster(name string, leaderInfo *LeaderInfo, fsm *MetadataFsm, partition
if clusterMpCntLimit == 0 {
atomic.StoreUint64(&clusterMpCntLimit, defaultMaxMpCntLimit)
}
if distributionOptimizationThreshold.Load() == 0 {
distributionOptimizationThreshold.Store(defaultDistributionOptimizationThreshold)
}
c.t = newTopology()
c.BadDataPartitionIds = new(sync.Map)
c.BadMetaPartitionIds = new(sync.Map)
@ -493,9 +496,8 @@ func newCluster(name string, leaderInfo *LeaderInfo, fsm *MetadataFsm, partition
c.MarkDiskBrokenThreshold.Store(defaultMarkDiskBrokenThreshold)
c.EnableAutoDpMetaRepair.Store(defaultEnableDpMetaRepair)
c.AutoDecommissionInterval.Store(int64(defaultAutoDecommissionDiskInterval))
c.EnableAutoDistributionOptimization.Store(defaultEnableAutoDistributionOptimization)
c.DistributionOptimizationConcurrentDpCount.Store(int64(defaultDistributionOptimizationConcurrentDpCount))
c.DistributionOptimizationThreshold.Store(defaultDistributionOptimizationThreshold)
c.EnableDistributionOptimization.Store(defaultEnableDistributionOptimization)
c.DistributionOptimizationConDpCnt.Store(int64(defaultDistributionOptimizationConDpCnt))
c.NodeSetUnbalancedDPs.Store(0)
c.RackConflictDPs.Store(0)
c.server = server
@ -2367,7 +2369,7 @@ func (c *Cluster) chooseZone2Plus1(rsMgr *rsManager, zones []*Zone,
for _, zone := range zoneList {
paramCopy.replicaNum = num
selectedHosts, selectedPeers, e := zone.getAvailNodeHosts(nodeType, paramCopy, 1)
selectedHosts, selectedPeers, e := zone.getAvailNodeHosts(nodeType, paramCopy)
if e != nil {
log.LogErrorf("action[chooseZone2Plus1] getAvailNodeHosts param[%v] error: [%v]", paramCopy.String(), e)
return nil, nil, e
@ -2403,7 +2405,7 @@ func (c *Cluster) chooseZoneNormal(zones []*Zone, nodeType uint32, param *select
zone := zones[c.lastZoneIdxForNode]
c.lastZoneIdxForNode = (c.lastZoneIdxForNode + 1) % len(zones)
paramCopy.replicaNum = 1
selectedHosts, selectedPeers, err := zone.getAvailNodeHosts(nodeType, paramCopy, 1)
selectedHosts, selectedPeers, err := zone.getAvailNodeHosts(nodeType, paramCopy)
if err != nil {
// no zone available
if j == len(zones)-1 {
@ -2455,11 +2457,11 @@ func (c *Cluster) getHostFromNormalZoneForCreate(nodeType uint32, replicaNum int
replicaNum: replicaNum,
rackLevel: rackLevel,
}
return c.getHostFromNormalZone(nodeType, nil, zoneNum, specifiedZoneName, dataMediaType, param, 1)
return c.getHostFromNormalZone(nodeType, nil, zoneNum, specifiedZoneName, dataMediaType, param)
}
func (c *Cluster) getHostFromNormalZone(nodeType uint32, excludeZones []string, zoneNumNeed int,
specifiedZoneName string, dataMediaType uint32, param *selectParam, threshold float64) (hosts []string, peers []proto.Peer, err error,
specifiedZoneName string, dataMediaType uint32, param *selectParam) (hosts []string, peers []proto.Peer, err error,
) {
log.LogInfof("[getHostFromNormalZone] dataMediaType(%v) nodeType(%v) replicaNum(%v) zoneNumNeed(%v) specifiedZoneName(%v)",
proto.MediaTypeString(nodeType), nodeType, param.replicaNum, zoneNumNeed, specifiedZoneName)
@ -2492,7 +2494,7 @@ func (c *Cluster) getHostFromNormalZone(nodeType uint32, excludeZones []string,
if len(zonesQualified) == 1 {
log.LogInfof("action[getHostFromNormalZone] zones [%v]", zonesQualified[0].name)
if hosts, peers, err = zonesQualified[0].getAvailNodeHosts(nodeType, param, threshold); err != nil {
if hosts, peers, err = zonesQualified[0].getAvailNodeHosts(nodeType, param); err != nil {
log.LogWarnf("action[getHostFromNormalZone] err[%v]", err)
return
}
@ -3143,7 +3145,7 @@ func (c *Cluster) migrateDataPartition(srcAddr, targetAddr string, dp *DataParti
log.LogErrorf("[migrateDataPartition] check mediaType err: %v", err.Error())
goto errHandler
}
} else if targetHosts, _, err = ns.getAvailDataNodeHosts(param, 1); err != nil {
} else if targetHosts, _, err = ns.getAvailDataNodeHosts(param); err != nil {
if _, ok := c.vols[dp.VolName]; !ok {
log.LogWarnf("clusterID[%v] partitionID:%v on node:%v offline failed,PersistenceHosts:[%v]",
c.Name, dp.PartitionID, srcAddr, dp.Hosts)
@ -3158,10 +3160,10 @@ func (c *Cluster) migrateDataPartition(srcAddr, targetAddr string, dp *DataParti
excludeNodeSets = append(excludeNodeSets, ns.ID)
param.excludeNodeSets = excludeNodeSets
if targetHosts, _, err = zone.getAvailNodeHosts(TypeDataPartition, param, 1); err != nil {
if targetHosts, _, err = zone.getAvailNodeHosts(TypeDataPartition, param); err != nil {
// select data nodes from the other zone
zones = dp.getLiveZones(srcAddr)
if targetHosts, _, err = c.getHostFromNormalZone(TypeDataPartition, zones, 1, "", dp.MediaType, param, 1); err != nil {
if targetHosts, _, err = c.getHostFromNormalZone(TypeDataPartition, zones, 1, "", dp.MediaType, param); err != nil {
goto errHandler
}
}
@ -3342,7 +3344,7 @@ func (c *Cluster) updateDataNodeSize(addr string, dp *DataPartition) error {
dataNode.Lock()
defer dataNode.Unlock()
if !dataNode.isWriteAbleWithSizeNoLock(10 * util.GB) {
if !dataNode.isWriteAbleWithSizeNoLock(10*util.GB, 1) {
return fmt.Errorf("new datanode %s is not writable AvailableSpace(%v) isActive(%v) RdOnly(%v) Total(%v) Used(%v)",
addr, dataNode.AvailableSpace, dataNode.isActive, dataNode.RdOnly, dataNode.Total, dataNode.Used)
}
@ -3392,7 +3394,7 @@ func (c *Cluster) addDataReservedResource(addrs []string, dp *DataPartition) err
dn.Lock()
if !dn.isWriteAbleWithSizeNoLock(10 * util.GB) {
if !dn.isWriteAbleWithSizeNoLock(10*util.GB, 1) {
dn.Unlock()
return fmt.Errorf("new datanode %s is not writable AvailableSpace(%v) isActive(%v) RdOnly(%v) Total(%v) Used(%v) PreResearvedSpace(%v)",
addr, dn.AvailableSpace, dn.isActive, dn.RdOnly, dn.Total, dn.Used, dn.PreResearvedSpace)
@ -5222,12 +5224,12 @@ func (c *Cluster) getEnableAutoDpMetaRepair() (v bool) {
return
}
func (c *Cluster) setEnableAutoDistributionOptimization(val bool) (err error) {
oldVal := c.EnableAutoDistributionOptimization.Load()
c.EnableAutoDistributionOptimization.Store(val)
func (c *Cluster) setEnableDistributionOptimization(val bool) (err error) {
oldVal := c.EnableDistributionOptimization.Load()
c.EnableDistributionOptimization.Store(val)
if err = c.syncPutCluster(); err != nil {
log.LogErrorf("[setEnableAutoDistributionOptimization] failed to set enable auto distribution optimization, err(%v)", err)
c.EnableAutoDistributionOptimization.Store(oldVal)
log.LogErrorf("[setEnableDistributionOptimization] failed to set enable auto distribution optimization, err(%v)", err)
c.EnableDistributionOptimization.Store(oldVal)
err = proto.ErrPersistenceByRaft
return
}
@ -5236,12 +5238,12 @@ func (c *Cluster) setEnableAutoDistributionOptimization(val bool) (err error) {
return
}
func (c *Cluster) getEnableAutoDistributionOptimization() bool {
return c.EnableAutoDistributionOptimization.Load()
func (c *Cluster) getEnableDistributionOptimization() bool {
return c.EnableDistributionOptimization.Load()
}
func (c *Cluster) updateEnableAutoDistributionOptimization(val bool) {
c.EnableAutoDistributionOptimization.Store(val)
func (c *Cluster) updateEnableDistributionOptimization(val bool) {
c.EnableDistributionOptimization.Store(val)
}
func (c *Cluster) getDataPartitionTimeoutSec() (val int64) {
@ -7297,17 +7299,25 @@ func (c *Cluster) getMetaPartitionStoreMode(mp *MetaPartition, srcAddr string) (
return
}
func (c *Cluster) setDistributionOptimizationConcurrentDpCount(count int64) error {
c.DistributionOptimizationConcurrentDpCount.Store(count)
func (c *Cluster) setDistributionOptimizationConDpCnt(count int64) error {
c.DistributionOptimizationConDpCnt.Store(count)
if err := c.syncPutCluster(); err != nil {
log.LogWarnf("setDistributionOptimizationConcurrentDpCount: sync put cluster failed, err(%v)", err)
log.LogWarnf("setDistributionOptimizationConDpCnt: sync put cluster failed, err(%v)", err)
return err
}
return nil
}
func getDistributionOptimizationThreshold() float64 {
val := distributionOptimizationThreshold.Load()
if val > 0 && val <= 1 {
return val
}
return defaultDistributionOptimizationThreshold
}
func (c *Cluster) setDistributionOptimizationThreshold(threshold float64) error {
c.DistributionOptimizationThreshold.Store(threshold)
distributionOptimizationThreshold.Store(threshold)
if err := c.syncPutCluster(); err != nil {
log.LogWarnf("setDistributionOptimizationThreshold: sync put cluster failed, err(%v)", err)
return err

6
master/cluster_task.go
View File

@ -165,7 +165,7 @@ func (c *Cluster) migrateMetaPartition(srcAddr, targetAddr string, mp *MetaParti
newPeers = []proto.Peer{{
Addr: targetAddr,
}}
} else if _, newPeers, err = ns.getAvailMetaNodeHosts(param, dstStoreMode, 1); err != nil {
} else if _, newPeers, err = ns.getAvailMetaNodeHosts(param, dstStoreMode); err != nil {
if _, ok := c.vols[mp.volName]; !ok {
log.LogWarnf("[migrateMetaPartition] clusterID[%v] partitionID:%v on node:[%v]",
c.Name, mp.PartitionID, mp.Hosts)
@ -179,7 +179,7 @@ func (c *Cluster) migrateMetaPartition(srcAddr, targetAddr string, mp *MetaParti
// choose a meta node in other node set in the same zone
excludeNodeSets = append(excludeNodeSets, ns.ID)
param.excludeNodeSets = excludeNodeSets
if _, newPeers, err = zone.getAvailNodeHosts(nodeType, param, 1); err != nil {
if _, newPeers, err = zone.getAvailNodeHosts(nodeType, param); err != nil {
zones = mp.getLiveZones(srcAddr)
var excludeZone []string
if len(zones) == 0 {
@ -188,7 +188,7 @@ func (c *Cluster) migrateMetaPartition(srcAddr, targetAddr string, mp *MetaParti
excludeZone = append(excludeZone, zones[0])
}
// choose a meta node in other zone
if _, newPeers, err = c.getHostFromNormalZone(nodeType, excludeZone, 1, "", proto.MediaType_Unspecified, param, 1); err != nil {
if _, newPeers, err = c.getHostFromNormalZone(nodeType, excludeZone, 1, "", proto.MediaType_Unspecified, param); err != nil {
goto errHandler
}
}

334
master/const.go
View File

@ -50,115 +50,115 @@ const (
forbiddenKey = "forbidden"
deleteVolKey = "delete"
forceDelVolKey = "forceDelVol"
ebsBlkSizeKey = "ebsBlkSize"
clientVersion = "version"
domainIdKey = "domainId"
volOwnerKey = "owner"
volAuthKey = "authKey"
replicaNumKey = "replicaNum"
followerReadKey = "followerRead"
authenticateKey = "authenticate"
akKey = "ak"
keywordsKey = "keywords"
zoneNameKey = "zoneName"
rackKey = "rack" // 添加 rack 字段的 key
nodesetIdKey = "nodesetId"
crossZoneKey = "crossZone"
normalZonesFirstKey = "normalZonesFirst"
userKey = "user"
nodeDeleteBatchCountKey = "batchCount"
nodeMarkDeleteRateKey = "markDeleteRate"
nodeDeleteWorkerSleepMs = "deleteWorkerSleepMs"
nodeAutoRepairRateKey = "autoRepairRate"
nodeDpRepairTimeOutKey = "dpRepairTimeOut"
nodeDpBackupKey = "dpBackupTimeout"
nodeDpMaxRepairErrCntKey = "dpMaxRepairErrCnt"
clusterLoadFactorKey = "loadFactor"
maxDpCntLimitKey = "maxDpCntLimit"
maxMpCntLimitKey = "maxMpCntLimit"
clusterCreateTimeKey = "clusterCreateTime"
descriptionKey = "description"
dpSelectorNameKey = "dpSelectorName"
dpSelectorParmKey = "dpSelectorParm"
nodeTypeKey = "nodeType"
ratio = "ratio"
rdOnlyKey = "rdOnly"
srcAddrKey = "srcAddr"
targetAddrKey = "targetAddr"
forceKey = "force"
raftForceDelKey = "raftForceDel"
weightKey = "weight"
dstNodeSetKey = "dstNodeSet"
enablePosixAclKey = "enablePosixAcl"
enableTxMaskKey = "enableTxMask"
txTimeoutKey = "txTimeout"
txConflictRetryNumKey = "txConflictRetryNum"
txConflictRetryIntervalKey = "txConflictRetryInterval"
txOpLimitKey = "txOpLimit"
txForceResetKey = "txForceReset"
QosEnableKey = "qosEnable"
DiskEnableKey = "diskenable"
IopsWKey = "iopsWKey"
IopsRKey = "iopsRKey"
FlowWKey = "flowWKey"
FlowRKey = "flowRKey"
ClientReqPeriod = "reqPeriod"
ClientTriggerCnt = "triggerCnt"
QosMasterLimit = "qosLimit"
decommissionFirstHostDiskParallelLimit = "decommissionFirstHostDiskParallelLimit"
decommissionFirstHostParallelLimit = "decommissionFirstHostParallelLimit"
decommissionLimit = "decommissionLimit"
DiskDisableKey = "diskDisable"
Limit = "limit"
TimeOut = "timeout"
CountByMeta = "countByMeta"
dpReadOnlyWhenVolFull = "dpReadOnlyWhenVolFull"
PeriodicKey = "periodic"
IPKey = "ip"
OperateKey = "op"
UIDKey = "uid"
CapacityKey = "capacity"
configKey = "config"
MaxFilesKey = "maxFiles"
MaxBytesKey = "maxBytes"
quotaKey = "quotaId"
enableQuota = "enableQuota"
dpDiscardKey = "dpDiscard"
ignoreDiscardKey = "ignoreDiscard"
TrashIntervalKey = "trashInterval"
ClientIDKey = "clientIDKey"
verSeqKey = "verSeq"
Periodic = "periodic"
decommissionDiskLimit = "decommissionDiskLimit"
dpRepairBlockSizeKey = "dpRepairBlockSize"
markDiskBrokenThresholdKey = "markDiskBrokenThreshold"
decommissionTypeKey = "decommissionType"
autoDecommissionDiskKey = "autoDecommissionDisk"
autoDecommissionDiskIntervalKey = "autoDecommissionDiskInterval"
autoDpMetaRepairKey = "autoDpMetaRepair"
autoDpMetaRepairParallelCntKey = "autoDpMetaRepairParallelCnt"
autoDistributionOptimizationKey = "autoDistributionOptimization"
distributionOptimizationConcurrentDpCountKey = "distributionOptimizationConcurrentDpCount"
distributionOptimizationThresholdKey = "distributionOptimizationThreshold"
dpTimeoutKey = "dpTimeout"
mpTimeoutKey = "mpTimeout"
rackAwareLevelKey = "rackAware"
ShowAll = "showAll"
trashIntervalKey = "trashInterval"
accessTimeIntervalKey = "accessTimeValidInterval"
enablePersistAccessTimeKey = "enablePersistAccessTime"
mediaTypeKey = "mediaType"
allowedStorageClassKey = "allowedStorageClass"
volStorageClassKey = "volStorageClass"
opLogDimensionKey = "opLogDimension"
volNameKey = "volName"
dpIdKey = "dpId"
diskNameKey = "diskName"
forbidWriteOpOfProtoVersion0 = "forbidWriteOpOfProtoVersion0"
quotaClass = "quotaClass"
quotaOfClass = "quotaOfStorageClass"
dataMediaTypeKey = "dataMediaType"
forceDelVolKey = "forceDelVol"
ebsBlkSizeKey = "ebsBlkSize"
clientVersion = "version"
domainIdKey = "domainId"
volOwnerKey = "owner"
volAuthKey = "authKey"
replicaNumKey = "replicaNum"
followerReadKey = "followerRead"
authenticateKey = "authenticate"
akKey = "ak"
keywordsKey = "keywords"
zoneNameKey = "zoneName"
rackKey = "rack" // 添加 rack 字段的 key
nodesetIdKey = "nodesetId"
crossZoneKey = "crossZone"
normalZonesFirstKey = "normalZonesFirst"
userKey = "user"
nodeDeleteBatchCountKey = "batchCount"
nodeMarkDeleteRateKey = "markDeleteRate"
nodeDeleteWorkerSleepMs = "deleteWorkerSleepMs"
nodeAutoRepairRateKey = "autoRepairRate"
nodeDpRepairTimeOutKey = "dpRepairTimeOut"
nodeDpBackupKey = "dpBackupTimeout"
nodeDpMaxRepairErrCntKey = "dpMaxRepairErrCnt"
clusterLoadFactorKey = "loadFactor"
maxDpCntLimitKey = "maxDpCntLimit"
maxMpCntLimitKey = "maxMpCntLimit"
clusterCreateTimeKey = "clusterCreateTime"
descriptionKey = "description"
dpSelectorNameKey = "dpSelectorName"
dpSelectorParmKey = "dpSelectorParm"
nodeTypeKey = "nodeType"
ratio = "ratio"
rdOnlyKey = "rdOnly"
srcAddrKey = "srcAddr"
targetAddrKey = "targetAddr"
forceKey = "force"
raftForceDelKey = "raftForceDel"
weightKey = "weight"
dstNodeSetKey = "dstNodeSet"
enablePosixAclKey = "enablePosixAcl"
enableTxMaskKey = "enableTxMask"
txTimeoutKey = "txTimeout"
txConflictRetryNumKey = "txConflictRetryNum"
txConflictRetryIntervalKey = "txConflictRetryInterval"
txOpLimitKey = "txOpLimit"
txForceResetKey = "txForceReset"
QosEnableKey = "qosEnable"
DiskEnableKey = "diskenable"
IopsWKey = "iopsWKey"
IopsRKey = "iopsRKey"
FlowWKey = "flowWKey"
FlowRKey = "flowRKey"
ClientReqPeriod = "reqPeriod"
ClientTriggerCnt = "triggerCnt"
QosMasterLimit = "qosLimit"
decommissionFirstHostDiskParallelLimit = "decommissionFirstHostDiskParallelLimit"
decommissionFirstHostParallelLimit = "decommissionFirstHostParallelLimit"
decommissionLimit = "decommissionLimit"
DiskDisableKey = "diskDisable"
Limit = "limit"
TimeOut = "timeout"
CountByMeta = "countByMeta"
dpReadOnlyWhenVolFull = "dpReadOnlyWhenVolFull"
PeriodicKey = "periodic"
IPKey = "ip"
OperateKey = "op"
UIDKey = "uid"
CapacityKey = "capacity"
configKey = "config"
MaxFilesKey = "maxFiles"
MaxBytesKey = "maxBytes"
quotaKey = "quotaId"
enableQuota = "enableQuota"
dpDiscardKey = "dpDiscard"
ignoreDiscardKey = "ignoreDiscard"
TrashIntervalKey = "trashInterval"
ClientIDKey = "clientIDKey"
verSeqKey = "verSeq"
Periodic = "periodic"
decommissionDiskLimit = "decommissionDiskLimit"
dpRepairBlockSizeKey = "dpRepairBlockSize"
markDiskBrokenThresholdKey = "markDiskBrokenThreshold"
decommissionTypeKey = "decommissionType"
autoDecommissionDiskKey = "autoDecommissionDisk"
autoDecommissionDiskIntervalKey = "autoDecommissionDiskInterval"
autoDpMetaRepairKey = "autoDpMetaRepair"
autoDpMetaRepairParallelCntKey = "autoDpMetaRepairParallelCnt"
autoDistributionOptimizationKey = "autoDistributionOptimization"
distributionOptimizationConDpCntKey = "distributionOptimizationConDpCnt"
distributionOptimizationThresholdKey = "distributionOptimizationThreshold"
dpTimeoutKey = "dpTimeout"
mpTimeoutKey = "mpTimeout"
rackAwareLevelKey = "rackAware"
ShowAll = "showAll"
trashIntervalKey = "trashInterval"
accessTimeIntervalKey = "accessTimeValidInterval"
enablePersistAccessTimeKey = "enablePersistAccessTime"
mediaTypeKey = "mediaType"
allowedStorageClassKey = "allowedStorageClass"
volStorageClassKey = "volStorageClass"
opLogDimensionKey = "opLogDimension"
volNameKey = "volName"
dpIdKey = "dpId"
diskNameKey = "diskName"
forbidWriteOpOfProtoVersion0 = "forbidWriteOpOfProtoVersion0"
quotaClass = "quotaClass"
quotaOfClass = "quotaOfStorageClass"
dataMediaTypeKey = "dataMediaType"
remoteCacheEnable = "remoteCacheEnable"
remoteCacheAutoPrepare = "remoteCacheAutoPrepare"
@ -209,64 +209,64 @@ const (
)
const (
defaultFaultDomainZoneCnt = 3
defaultNormalCrossZoneCnt = 3
defaultInitMetaPartitionCount = 3
defaultMaxInitMetaPartitionCount = 100
defaultMaxMetaPartitionInodeID uint64 = 1<<63 - 1
defaultMetaPartitionInodeIDStep uint64 = 1 << 22
defaultMetaNodeReservedMem uint64 = 1 << 30
runtimeStackBufSize = 4096
spaceAvailableRate = 0.90
defaultNodeSetCapacity = 18
minNumOfRWDataPartitions = 10
intervalToCheckMissingReplica = 600
intervalToWarnDataPartition = 600
intervalToLoadDataPartition = 12 * 60 * 60
defaultInitDataPartitionCnt = 3
maxInitDataPartitionCnt = 200
volExpansionRatio = 0.1
maxNumberOfDataPartitionsForExpansion = 100
EmptyCrcValue uint32 = 4045511210
DefaultZoneName = proto.DefaultZoneName
retrySendSyncTaskInternal = 3 * time.Second
defaultRangeOfCountDifferencesAllowed = 50
defaultMinusOfMaxInodeID = 1000
defaultNodeSetGrpBatchCnt = 3
defaultMaxReplicaCnt = 16
defaultIopsRLimit uint64 = 1 << 35
defaultIopsWLimit uint64 = 1 << 35
defaultFlowWLimit uint64 = 1 << 35
defaultFlowRLimit uint64 = 1 << 35
defaultLimitTypeCnt = 4
defaultClientTriggerHitCnt = 1
defaultClientReqPeriodSeconds = 1
defaultMaxQuotaNumPerVol = 100
defaultVolDelayDeleteTimeHour = 48
defaultMarkDiskBrokenThreshold = 0 // decommission all dp from disk
defaultEnableDpMetaRepair = false
defaultEnableAutoDistributionOptimization = false
defaultDistributionOptimizationConcurrentDpCount = 400
defaultDistributionOptimizationIntervalSec = 2 * 60 * 60
defaultDistributionOptimizationThreshold = 0.8
defaultAutoDpMetaRepairPallarelCnt = 100
defaultAutoDecommissionDiskInterval = 10 * time.Second
maxMpCreationCount = 10
defaultVolForbidWriteOpOfProtoVersion0 = true
defaultMetaNodeMemHighPer = 0.75
defaultMetaNodeMemLowPer = 0.3
metaNodeReserveMemorySize = 3 * 1024 * 1024 * 1024
metaNodeMemoryRatio = 2
defaultPlanExpireHours = 72
defaultGOGCLowerLimit = 30
defaultGOGCUpperLimit = 100
lowPriorityDecommissionWeight = 2
mediumPriorityDecommissionWeight = 4
highPriorityDecommissionWeight = 6
highestPriorityDecommissionWeight = 8
diskDecommissionInfoStatType = 1
dataNodeDecommissionInfoStatType = 2
defaultMpMigrateThreads = 10
defaultFaultDomainZoneCnt = 3
defaultNormalCrossZoneCnt = 3
defaultInitMetaPartitionCount = 3
defaultMaxInitMetaPartitionCount = 100
defaultMaxMetaPartitionInodeID uint64 = 1<<63 - 1
defaultMetaPartitionInodeIDStep uint64 = 1 << 22
defaultMetaNodeReservedMem uint64 = 1 << 30
runtimeStackBufSize = 4096
spaceAvailableRate = 0.90
defaultNodeSetCapacity = 18
minNumOfRWDataPartitions = 10
intervalToCheckMissingReplica = 600
intervalToWarnDataPartition = 600
intervalToLoadDataPartition = 12 * 60 * 60
defaultInitDataPartitionCnt = 3
maxInitDataPartitionCnt = 200
volExpansionRatio = 0.1
maxNumberOfDataPartitionsForExpansion = 100
EmptyCrcValue uint32 = 4045511210
DefaultZoneName = proto.DefaultZoneName
retrySendSyncTaskInternal = 3 * time.Second
defaultRangeOfCountDifferencesAllowed = 50
defaultMinusOfMaxInodeID = 1000
defaultNodeSetGrpBatchCnt = 3
defaultMaxReplicaCnt = 16
defaultIopsRLimit uint64 = 1 << 35
defaultIopsWLimit uint64 = 1 << 35
defaultFlowWLimit uint64 = 1 << 35
defaultFlowRLimit uint64 = 1 << 35
defaultLimitTypeCnt = 4
defaultClientTriggerHitCnt = 1
defaultClientReqPeriodSeconds = 1
defaultMaxQuotaNumPerVol = 100
defaultVolDelayDeleteTimeHour = 48
defaultMarkDiskBrokenThreshold = 0 // decommission all dp from disk
defaultEnableDpMetaRepair = false
defaultEnableDistributionOptimization = false
defaultDistributionOptimizationConDpCnt = 400
defaultDistributionOptimizationIntervalSec = 2 * 60 * 60
defaultDistributionOptimizationThreshold = 0.8
defaultAutoDpMetaRepairPallarelCnt = 100
defaultAutoDecommissionDiskInterval = 10 * time.Second
maxMpCreationCount = 10
defaultVolForbidWriteOpOfProtoVersion0 = true
defaultMetaNodeMemHighPer = 0.75
defaultMetaNodeMemLowPer = 0.3
metaNodeReserveMemorySize = 3 * 1024 * 1024 * 1024
metaNodeMemoryRatio = 2
defaultPlanExpireHours = 72
defaultGOGCLowerLimit = 30
defaultGOGCUpperLimit = 100
lowPriorityDecommissionWeight = 2
mediumPriorityDecommissionWeight = 4
highPriorityDecommissionWeight = 6
highestPriorityDecommissionWeight = 8
diskDecommissionInfoStatType = 1
dataNodeDecommissionInfoStatType = 2
defaultMpMigrateThreads = 10
maxTrashInterval = 365 * 24 * 60
mpReplicaDelInterval = 300 // 5 minutes

30
master/data_node.go
View File

@ -343,10 +343,7 @@ func (dataNode *DataNode) canAlloc() bool {
func (dataNode *DataNode) IsWriteAble(threshold float64) (ok bool) {
dataNode.RLock()
defer dataNode.RUnlock()
if threshold == 1 {
return dataNode.isWriteAbleWithSizeNoLock(10 * util.GB)
}
return dataNode.isWriteAbleWithThresholdNoLock(threshold)
return dataNode.isWriteAbleWithSizeNoLock(10*util.GB, threshold)
}
func (dataNode *DataNode) availableDiskCount() (cnt int) {
@ -412,31 +409,18 @@ func (dataNode *DataNode) GetStorageInfo() string {
dataNode.ToBeOffline, dataNode.availableDiskCount(), dataNode.DataPartitionCount, !dataNode.canAlloc())
}
func (dataNode *DataNode) isWriteAbleWithSizeNoLock(size uint64) (ok bool) {
func (dataNode *DataNode) isWriteAbleWithSizeNoLock(size uint64, threshold float64) (ok bool) {
if dataNode.isActive && dataNode.AvailableSpace-dataNode.PreResearvedSpace > size && !dataNode.RdOnly &&
dataNode.Total > dataNode.Used && (dataNode.Total-dataNode.Used) > size {
dataNode.Total > dataNode.Used && (dataNode.Total-dataNode.Used-dataNode.PreResearvedSpace) > size &&
float64(dataNode.AvailableSpace-dataNode.PreResearvedSpace) > (1-threshold)*float64(dataNode.Total) &&
float64(dataNode.Used+dataNode.PreResearvedSpace) <= threshold*float64(dataNode.Total) {
ok = true
}
if !ok {
log.LogInfof("[isWriteAbleWithSizeNoLock] node %v, isActive %v, TotalDpCnt %v, RdOnly %v, Total %v, AvailableSpace %v, "+
"used %v, PreResearvedSpace %v, reserved size %v",
"used %v, PreResearvedSpace %v, reserved size %v, threshold %v",
dataNode.Addr, dataNode.isActive, dataNode.DataPartitionCount, dataNode.RdOnly, dataNode.Total, dataNode.AvailableSpace, dataNode.Used,
dataNode.PreResearvedSpace, size)
}
return
}
func (dataNode *DataNode) isWriteAbleWithThresholdNoLock(threshold float64) (ok bool) {
if dataNode.isActive && float64(dataNode.AvailableSpace-dataNode.PreResearvedSpace) > threshold*float64(dataNode.Total) && !dataNode.RdOnly &&
dataNode.Total > dataNode.Used && float64(dataNode.Total-dataNode.Used) > threshold*float64(dataNode.Total) {
ok = true
}
if !ok {
log.LogInfof("[isWriteAbleWithSizeNoLock] node %v, isActive %v, TotalDpCnt %v, RdOnly %v, Total %v, AvailableSpace %v, "+
"used %v, PreResearvedSpace %v, threshold %v",
dataNode.Addr, dataNode.isActive, dataNode.DataPartitionCount, dataNode.RdOnly, dataNode.Total, dataNode.AvailableSpace, dataNode.Used,
dataNode.PreResearvedSpace, threshold)
dataNode.PreResearvedSpace, size, threshold)
}
return

29
master/data_partition.go
View File

@ -1988,7 +1988,7 @@ func (partition *DataPartition) PauseDecommission(c *Cluster) bool {
return true
}
// 处理单个源地址迁移完成，返回true表示还有更多源地址需要迁移
// Generate the next source address. Return true if there are more source addresses to migrate.
func (partition *DataPartition) ProcessNextDecommissionSrcHost(c *Cluster) bool {
var (
updatedSrcHosts []string
@ -1997,14 +1997,13 @@ func (partition *DataPartition) ProcessNextDecommissionSrcHost(c *Cluster) bool
nextDstAddr string
)
if partition.DecommissionSrcAddrs == nil || len(partition.DecommissionSrcAddrs) == 0 {
if len(partition.DecommissionSrcAddrs) == 0 {
log.LogInfof("action[ProcessNextDecommissionSrcHost] dp(%v) all sources completed, current: %v",
partition.PartitionID, partition.DecommissionSrcAddr)
return false
}
if partition.DecommissionDstAddrs == nil || len(partition.DecommissionDstAddrs) == 0 ||
len(partition.DecommissionDstAddrs) < len(partition.DecommissionSrcAddrs) {
if len(partition.DecommissionDstAddrs) != len(partition.DecommissionSrcAddrs) {
log.LogInfof("action[ProcessNextDecommissionSrcHost] dp(%v) has not dstAddrs to migrate",
partition.PartitionID)
return false
@ -2014,11 +2013,9 @@ func (partition *DataPartition) ProcessNextDecommissionSrcHost(c *Cluster) bool
if len(partition.DecommissionSrcAddrs) > 1 {
nextSrcAddr = partition.DecommissionSrcAddrs[lastIndex]
nextDstAddr = partition.DecommissionDstAddrs[0]
updatedSrcHosts = make([]string, lastIndex)
copy(updatedSrcHosts, partition.DecommissionSrcAddrs[0:lastIndex])
updatedDstHosts = make([]string, lastIndex)
copy(updatedDstHosts, partition.DecommissionDstAddrs[1:])
nextDstAddr = partition.DecommissionDstAddrs[lastIndex]
updatedSrcHosts = partition.DecommissionSrcAddrs[:lastIndex]
updatedDstHosts = partition.DecommissionDstAddrs[:lastIndex]
} else {
updatedSrcHosts = nil
nextSrcAddr = partition.DecommissionSrcAddrs[0]
@ -2353,7 +2350,7 @@ func (partition *DataPartition) TryAcquireDecommissionToken(c *Cluster) bool {
rackLevel: c.getRackAwareLevel(),
excludeRacks: c.GetExRacksByHosts(TypeDataPartition, excludeHosts, partition.DecommissionSrcAddr),
}
targetHosts, _, err = ns.getAvailDataNodeHosts(param, 1)
targetHosts, _, err = ns.getAvailDataNodeHosts(param)
if err != nil {
if partition.DecommissionDstNodeSet != 0 {
log.LogWarnf("action[TryAcquireDecommissionToken] dp %v choose from given dst nodeset %v failed:%v",
@ -2379,12 +2376,12 @@ func (partition *DataPartition) TryAcquireDecommissionToken(c *Cluster) bool {
param.excludeNodeSets = excludeNodeSets
// data nodes in a zone has the same mediaType
if targetHosts, _, err = zone.getAvailNodeHosts(TypeDataPartition, param, 1); err != nil {
if targetHosts, _, err = zone.getAvailNodeHosts(TypeDataPartition, param); err != nil {
log.LogWarnf("action[TryAcquireDecommissionToken] dp %v choose from other nodeset failed:%v",
partition.PartitionID, err.Error())
// select data nodes from the other zone
zones = partition.getLiveZones(partition.DecommissionSrcAddr)
if targetHosts, _, err = c.getHostFromNormalZone(TypeDataPartition, zones, 1, "", partition.MediaType, param, 1); err != nil {
if targetHosts, _, err = c.getHostFromNormalZone(TypeDataPartition, zones, 1, "", partition.MediaType, param); err != nil {
log.LogWarnf("action[TryAcquireDecommissionToken] dp %v choose from other zone failed:%v",
partition.PartitionID, err.Error())
goto errHandler
@ -2584,8 +2581,9 @@ func selectTargetHostsInDistributionOptimization(addrs []string, replicaNum int,
excludeHosts: addrs,
rackLevel: proto.RackAwareStrong, // use the highest rack aware level for distribution optimization,
excludeNodeSets: excludedNodesets,
selectType: proto.SelectType_DistributionOptimization,
}
availableHosts, _, err := zone.getAvailNodeHosts(TypeDataPartition, param, c.DistributionOptimizationThreshold.Load())
availableHosts, _, err := zone.getAvailNodeHosts(TypeDataPartition, param)
if err == nil && len(availableHosts) == replicaNum {
ns, _, err = getTargetNodeset(availableHosts[0], c)
if err != nil {
@ -2606,7 +2604,7 @@ func selectTargetHostsInDistributionOptimization(addrs []string, replicaNum int,
// rackLevel: proto.RackAwareStrong, // use the highest rack aware level for distribution optimization,
// }
// availableHosts, _, err := c.getHostFromNormalZone(TypeDataPartition, excludedZones, 1, "", mediaType, param, c.DistributionOptimizationThreshold.Load())
// availableHosts, _, err := c.getHostFromNormalZone(TypeDataPartition, excludedZones, 1, "", mediaType, param, getDistributionOptimizationThreshold())
// if err == nil && len(availableHosts) == needDstAddrCount {
// ns, _, err = getTargetNodeset(availableHosts[0], c)
// if err != nil {
@ -2711,6 +2709,7 @@ func selectOptimalNodes(currentAddrs []string, targetNsID uint64, c *Cluster) ([
excludeHosts: currentAddrs,
excludeRacks: excludeRacks,
rackLevel: proto.RackAwareStrong, // use the highest rack aware level for distribution optimization
selectType: proto.SelectType_DistributionOptimization,
}
// Get target NodeSet
@ -2737,7 +2736,7 @@ func selectOptimalNodes(currentAddrs []string, targetNsID uint64, c *Cluster) ([
return nil, nil, fmt.Errorf("target NodeSet %d not found", targetNsID)
}
newHosts, _, err := targetNs.getAvailDataNodeHosts(param, c.DistributionOptimizationThreshold.Load())
newHosts, _, err := targetNs.getAvailDataNodeHosts(param)
if err != nil {
return nil, nil, fmt.Errorf("failed to get available hosts in NodeSet %d: %v", targetNsID, err)
}

4
master/data_partition_test.go
View File

@ -322,7 +322,7 @@ func createTestClusterForOptimalNodes() *Cluster {
}
// Set up atomic values
cluster.DistributionOptimizationThreshold.Store(0.8)
distributionOptimizationThreshold.Store(0.8)
// Create zones
zone1 := &Zone{
@ -894,7 +894,7 @@ func createTestClusterForTargetHosts() *Cluster {
}
// Set up atomic values
cluster.DistributionOptimizationThreshold.Store(0.1) // Lower threshold for testing
distributionOptimizationThreshold.Store(0.1) // Lower threshold for testing
// Create zones
zones := []string{"zone1", "zone2", "zone3"}

10
master/distribution_optimization.go
View File

@ -35,7 +35,7 @@ func (c *Cluster) scheduleToDistributionOptimization() {
return
}
// Check if distribution optimization is enabled before execution
if !c.getEnableAutoDistributionOptimization() {
if !c.getEnableDistributionOptimization() {
log.LogDebugf("action[distributionOptimizationController] distribution optimization is disabled, skip execution")
return
}
@ -54,7 +54,7 @@ func (c *Cluster) executeDistributionOptimizationMigrations() {
}()
activeTasks := c.countActiveDistributionOptimizationTasks()
limit := c.DistributionOptimizationConcurrentDpCount.Load()
limit := c.DistributionOptimizationConDpCnt.Load()
if int64(activeTasks) >= limit {
log.LogInfof("action[executeDistributionOptimizationMigrations] already have %d active tasks, skipping execution", activeTasks)
return
@ -292,10 +292,10 @@ func (c *Cluster) buildDpHostToNsAndZone() (map[string]uint64, map[string]string
func (c *Cluster) getDistributionOptimizationStatus() *proto.DistributionOptimizationStatus {
status := &proto.DistributionOptimizationStatus{
DecommissioningDPIDs: make([]uint64, 0),
ConcurrentDpCount: c.DistributionOptimizationConcurrentDpCount.Load(),
ConcurrentDpCount: c.DistributionOptimizationConDpCnt.Load(),
BalanceIntervalSec: defaultDistributionOptimizationIntervalSec,
BalanceThreshold: c.DistributionOptimizationThreshold.Load(),
EnableDistributionOptimization: c.getEnableAutoDistributionOptimization(),
BalanceThreshold: getDistributionOptimizationThreshold(),
EnableDistributionOptimization: c.getEnableDistributionOptimization(),
DomainDistribution: &proto.DomainDistributionInfo{
SingleDomainDPs: 0,
TwoDomainDPs: 0,

6
master/distribution_optimization_test.go
View File

@ -454,9 +454,9 @@ func TestGetDistributionOptimizationStatus(t *testing.T) {
}
// Set up atomic values
cluster.DistributionOptimizationConcurrentDpCount.Store(100)
cluster.DistributionOptimizationThreshold.Store(0.8)
cluster.EnableAutoDistributionOptimization.Store(true)
cluster.DistributionOptimizationConDpCnt.Store(100)
distributionOptimizationThreshold.Store(0.8)
cluster.EnableDistributionOptimization.Store(true)
// Add mock data nodes (all in zone1, except one in zone2 for cross-zone testing)
mockNodes := map[string]*DataNode{

216
master/metadata_fsm_op.go
View File

@ -39,112 +39,112 @@ import (
transferred over the network. */
type clusterValue struct {
Name string
CreateTime int64
Threshold float32
LoadFactor float32
DisableAutoAllocate bool
ForbidMpDecommission bool
DataNodeDeleteLimitRate uint64
MetaNodeDeleteBatchCount uint64
MetaNodeDeleteWorkerSleepMs uint64
DataNodeAutoRepairLimitRate uint64
MaxDpCntLimit uint64
MaxMpCntLimit uint64
FaultDomain bool
DiskQosEnable bool
QosLimitUpload uint64
DirChildrenNumLimit uint32
DecommissionLimit uint64
DecommissionFirstHostDiskParallelLimit uint64
CheckDataReplicasEnable bool
FileStatsEnable bool
FileStatsThresholds []uint64
ClusterUuid string
ClusterUuidEnable bool
MetaPartitionInodeIdStep uint64
MaxConcurrentLcNodes uint64
DpMaxRepairErrCnt uint64
DpRepairTimeOut uint64
DpBackupTimeOut uint64
EnableAutoDecommissionDisk bool
AutoDecommissionDiskInterval int64
DecommissionDiskLimit uint32
EnableAutoDistributionOptimization bool
DistributionOptimizationConcurrentDpCount int64
DistributionOptimizationThreshold float64
VolDeletionDelayTimeHour int64
MetaNodeGOGC int
DataNodeGOGC int
MarkDiskBrokenThreshold float64
EnableAutoDpMetaRepair bool
AutoDpMetaRepairParallelCnt uint32
DataPartitionTimeoutSec int64
MetaPartitionTimeoutSec int64
ForbidWriteOpOfProtoVer0 bool
LegacyDataMediaType uint32
RaftPartitionAlreadyUseDifferentPort bool
MetaNodeMemoryHighPer float64
MetaNodeMemoryLowPer float64
AutoMpMigrate bool
FlashNodeHandleReadTimeout int
FlashNodeReadDataNodeTimeout int
RackAwareLevel uint8
Name string
CreateTime int64
Threshold float32
LoadFactor float32
DisableAutoAllocate bool
ForbidMpDecommission bool
DataNodeDeleteLimitRate uint64
MetaNodeDeleteBatchCount uint64
MetaNodeDeleteWorkerSleepMs uint64
DataNodeAutoRepairLimitRate uint64
MaxDpCntLimit uint64
MaxMpCntLimit uint64
FaultDomain bool
DiskQosEnable bool
QosLimitUpload uint64
DirChildrenNumLimit uint32
DecommissionLimit uint64
DecommissionFirstHostDiskParallelLimit uint64
CheckDataReplicasEnable bool
FileStatsEnable bool
FileStatsThresholds []uint64
ClusterUuid string
ClusterUuidEnable bool
MetaPartitionInodeIdStep uint64
MaxConcurrentLcNodes uint64
DpMaxRepairErrCnt uint64
DpRepairTimeOut uint64
DpBackupTimeOut uint64
EnableAutoDecommissionDisk bool
AutoDecommissionDiskInterval int64
DecommissionDiskLimit uint32
EnableDistributionOptimization bool
DistributionOptimizationConDpCnt int64
DistributionOptimizationThreshold float64
VolDeletionDelayTimeHour int64
MetaNodeGOGC int
DataNodeGOGC int
MarkDiskBrokenThreshold float64
EnableAutoDpMetaRepair bool
AutoDpMetaRepairParallelCnt uint32
DataPartitionTimeoutSec int64
MetaPartitionTimeoutSec int64
ForbidWriteOpOfProtoVer0 bool
LegacyDataMediaType uint32
RaftPartitionAlreadyUseDifferentPort bool
MetaNodeMemoryHighPer float64
MetaNodeMemoryLowPer float64
AutoMpMigrate bool
FlashNodeHandleReadTimeout int
FlashNodeReadDataNodeTimeout int
RackAwareLevel uint8
}
func newClusterValue(c *Cluster) (cv *clusterValue) {
cv = &clusterValue{
Name: c.Name,
CreateTime: c.CreateTime,
LoadFactor: c.cfg.ClusterLoadFactor,
Threshold: c.cfg.MetaNodeThreshold,
DataNodeDeleteLimitRate: c.cfg.DataNodeDeleteLimitRate,
MetaNodeDeleteBatchCount: c.cfg.MetaNodeDeleteBatchCount,
MetaNodeDeleteWorkerSleepMs: c.cfg.MetaNodeDeleteWorkerSleepMs,
DataNodeAutoRepairLimitRate: c.cfg.DataNodeAutoRepairLimitRate,
DisableAutoAllocate: c.DisableAutoAllocate,
ForbidMpDecommission: c.ForbidMpDecommission,
MaxDpCntLimit: c.getMaxDpCntLimit(),
MaxMpCntLimit: c.getMaxMpCntLimit(),
FaultDomain: c.FaultDomain,
DiskQosEnable: c.diskQosEnable,
QosLimitUpload: uint64(c.QosAcceptLimit.Limit()),
DirChildrenNumLimit: c.cfg.DirChildrenNumLimit,
DecommissionFirstHostDiskParallelLimit: c.DecommissionFirstHostDiskParallelLimit,
DecommissionLimit: c.DecommissionLimit,
CheckDataReplicasEnable: c.checkDataReplicasEnable,
FileStatsEnable: c.fileStatsEnable,
FileStatsThresholds: c.fileStatsThresholds,
ClusterUuid: c.clusterUuid,
ClusterUuidEnable: c.clusterUuidEnable,
MetaPartitionInodeIdStep: c.cfg.MetaPartitionInodeIdStep,
MaxConcurrentLcNodes: c.cfg.MaxConcurrentLcNodes,
DpMaxRepairErrCnt: c.cfg.DpMaxRepairErrCnt,
DpRepairTimeOut: c.cfg.DpRepairTimeOut,
DpBackupTimeOut: c.cfg.DpBackupTimeOut,
EnableAutoDecommissionDisk: c.EnableAutoDecommissionDisk.Load(),
AutoDecommissionDiskInterval: c.AutoDecommissionInterval.Load(),
DecommissionDiskLimit: c.GetDecommissionDiskLimit(),
EnableAutoDistributionOptimization: c.getEnableAutoDistributionOptimization(),
DistributionOptimizationConcurrentDpCount: c.DistributionOptimizationConcurrentDpCount.Load(),
DistributionOptimizationThreshold: c.DistributionOptimizationThreshold.Load(),
VolDeletionDelayTimeHour: c.cfg.volDelayDeleteTimeHour,
MetaNodeGOGC: c.cfg.metaNodeGOGC,
DataNodeGOGC: c.cfg.dataNodeGOGC,
MarkDiskBrokenThreshold: c.getMarkDiskBrokenThreshold(),
EnableAutoDpMetaRepair: c.getEnableAutoDpMetaRepair(),
AutoDpMetaRepairParallelCnt: c.AutoDpMetaRepairParallelCnt.Load(),
DataPartitionTimeoutSec: c.getDataPartitionTimeoutSec(),
MetaPartitionTimeoutSec: c.getMetaPartitionTimeoutSec(),
ForbidWriteOpOfProtoVer0: c.cfg.forbidWriteOpOfProtoVer0,
LegacyDataMediaType: c.legacyDataMediaType,
RaftPartitionAlreadyUseDifferentPort: c.cfg.raftPartitionAlreadyUseDifferentPort.Load(),
MetaNodeMemoryHighPer: c.cfg.metaNodeMemHighPer,
MetaNodeMemoryLowPer: c.cfg.metaNodeMemLowPer,
AutoMpMigrate: c.cfg.AutoMpMigrate,
FlashNodeHandleReadTimeout: c.cfg.flashNodeHandleReadTimeout,
FlashNodeReadDataNodeTimeout: c.cfg.flashNodeReadDataNodeTimeout,
RackAwareLevel: uint8(c.cfg.RackAwareLevel),
Name: c.Name,
CreateTime: c.CreateTime,
LoadFactor: c.cfg.ClusterLoadFactor,
Threshold: c.cfg.MetaNodeThreshold,
DataNodeDeleteLimitRate: c.cfg.DataNodeDeleteLimitRate,
MetaNodeDeleteBatchCount: c.cfg.MetaNodeDeleteBatchCount,
MetaNodeDeleteWorkerSleepMs: c.cfg.MetaNodeDeleteWorkerSleepMs,
DataNodeAutoRepairLimitRate: c.cfg.DataNodeAutoRepairLimitRate,
DisableAutoAllocate: c.DisableAutoAllocate,
ForbidMpDecommission: c.ForbidMpDecommission,
MaxDpCntLimit: c.getMaxDpCntLimit(),
MaxMpCntLimit: c.getMaxMpCntLimit(),
FaultDomain: c.FaultDomain,
DiskQosEnable: c.diskQosEnable,
QosLimitUpload: uint64(c.QosAcceptLimit.Limit()),
DirChildrenNumLimit: c.cfg.DirChildrenNumLimit,
DecommissionFirstHostDiskParallelLimit: c.DecommissionFirstHostDiskParallelLimit,
DecommissionLimit: c.DecommissionLimit,
CheckDataReplicasEnable: c.checkDataReplicasEnable,
FileStatsEnable: c.fileStatsEnable,
FileStatsThresholds: c.fileStatsThresholds,
ClusterUuid: c.clusterUuid,
ClusterUuidEnable: c.clusterUuidEnable,
MetaPartitionInodeIdStep: c.cfg.MetaPartitionInodeIdStep,
MaxConcurrentLcNodes: c.cfg.MaxConcurrentLcNodes,
DpMaxRepairErrCnt: c.cfg.DpMaxRepairErrCnt,
DpRepairTimeOut: c.cfg.DpRepairTimeOut,
DpBackupTimeOut: c.cfg.DpBackupTimeOut,
EnableAutoDecommissionDisk: c.EnableAutoDecommissionDisk.Load(),
AutoDecommissionDiskInterval: c.AutoDecommissionInterval.Load(),
DecommissionDiskLimit: c.GetDecommissionDiskLimit(),
EnableDistributionOptimization: c.getEnableDistributionOptimization(),
DistributionOptimizationConDpCnt: c.DistributionOptimizationConDpCnt.Load(),
DistributionOptimizationThreshold: getDistributionOptimizationThreshold(),
VolDeletionDelayTimeHour: c.cfg.volDelayDeleteTimeHour,
MetaNodeGOGC: c.cfg.metaNodeGOGC,
DataNodeGOGC: c.cfg.dataNodeGOGC,
MarkDiskBrokenThreshold: c.getMarkDiskBrokenThreshold(),
EnableAutoDpMetaRepair: c.getEnableAutoDpMetaRepair(),
AutoDpMetaRepairParallelCnt: c.AutoDpMetaRepairParallelCnt.Load(),
DataPartitionTimeoutSec: c.getDataPartitionTimeoutSec(),
MetaPartitionTimeoutSec: c.getMetaPartitionTimeoutSec(),
ForbidWriteOpOfProtoVer0: c.cfg.forbidWriteOpOfProtoVer0,
LegacyDataMediaType: c.legacyDataMediaType,
RaftPartitionAlreadyUseDifferentPort: c.cfg.raftPartitionAlreadyUseDifferentPort.Load(),
MetaNodeMemoryHighPer: c.cfg.metaNodeMemHighPer,
MetaNodeMemoryLowPer: c.cfg.metaNodeMemLowPer,
AutoMpMigrate: c.cfg.AutoMpMigrate,
FlashNodeHandleReadTimeout: c.cfg.flashNodeHandleReadTimeout,
FlashNodeReadDataNodeTimeout: c.cfg.flashNodeReadDataNodeTimeout,
RackAwareLevel: uint8(c.cfg.RackAwareLevel),
}
return cv
}
@ -1375,15 +1375,15 @@ func (c *Cluster) loadClusterValue() (err error) {
c.EnableAutoDecommissionDisk.Store(cv.EnableAutoDecommissionDisk)
c.updateAutoDecommissionDiskInterval(cv.AutoDecommissionDiskInterval)
c.DecommissionLimit = cv.DecommissionLimit
c.updateEnableAutoDistributionOptimization(cv.EnableAutoDistributionOptimization)
if cv.DistributionOptimizationConcurrentDpCount <= 0 {
cv.DistributionOptimizationConcurrentDpCount = int64(defaultDistributionOptimizationConcurrentDpCount)
c.updateEnableDistributionOptimization(cv.EnableDistributionOptimization)
if cv.DistributionOptimizationConDpCnt <= 0 {
cv.DistributionOptimizationConDpCnt = int64(defaultDistributionOptimizationConDpCnt)
}
if cv.DistributionOptimizationThreshold < 0 || cv.DistributionOptimizationThreshold > 1 {
cv.DistributionOptimizationThreshold = 0
cv.DistributionOptimizationThreshold = defaultDistributionOptimizationThreshold
}
c.DistributionOptimizationConcurrentDpCount.Store(cv.DistributionOptimizationConcurrentDpCount)
c.DistributionOptimizationThreshold.Store(cv.DistributionOptimizationThreshold)
c.DistributionOptimizationConDpCnt.Store(cv.DistributionOptimizationConDpCnt)
distributionOptimizationThreshold.Store(cv.DistributionOptimizationThreshold)
c.cfg.volDelayDeleteTimeHour = cv.VolDeletionDelayTimeHour
c.cfg.metaNodeGOGC = cv.MetaNodeGOGC
c.cfg.dataNodeGOGC = cv.DataNodeGOGC

20
master/node_selector.go
View File

@ -633,7 +633,7 @@ func (ns *nodeSet) getRackSets() nodeSetCollection {
return rsets
}
func (ns *nodeSet) getAvailMetaNodeHosts(param *selectParam, storeMode proto.StoreMode, threshold float64) (newHosts []string, peers []proto.Peer, err error) {
func (ns *nodeSet) getAvailMetaNodeHosts(param *selectParam, storeMode proto.StoreMode) (newHosts []string, peers []proto.Peer, err error) {
ns.nodeSelectLock.Lock()
defer ns.nodeSelectLock.Unlock()
// we need a read lock to block the modification of node selector
@ -645,6 +645,14 @@ func (ns *nodeSet) getAvailMetaNodeHosts(param *selectParam, storeMode proto.Sto
nodeType = RocksdbType
}
var threshold float64
switch param.selectType {
case proto.SelectType_Normal:
threshold = 1
case proto.SelectType_DistributionOptimization:
threshold = getDistributionOptimizationThreshold()
}
// If rack isolation is not enabled, use non-rack-aware selector directly
if param.rackLevel == proto.RackAwareNone {
return ns.getNodeSelector(nodeType, storeMode).Select(ns, param.excludeHosts, param.replicaNum, threshold)
@ -703,13 +711,21 @@ func (ns *nodeSet) selectNodesWithRack(param *selectParam, nodeType NodeType, st
}
}
func (ns *nodeSet) getAvailDataNodeHosts(param *selectParam, threshold float64) (hosts []string, peers []proto.Peer, err error) {
func (ns *nodeSet) getAvailDataNodeHosts(param *selectParam) (hosts []string, peers []proto.Peer, err error) {
ns.nodeSelectLock.Lock()
defer ns.nodeSelectLock.Unlock()
// we need a read lock to block the modification of node selector
ns.dataNodeSelectorLock.Lock()
defer ns.dataNodeSelectorLock.Unlock()
var threshold float64
switch param.selectType {
case proto.SelectType_Normal:
threshold = 1
case proto.SelectType_DistributionOptimization:
threshold = getDistributionOptimizationThreshold()
}
if param.rackLevel == proto.RackAwareNone {
return ns.dataNodeSelector.Select(ns, param.excludeHosts, param.replicaNum, threshold)
}

15
master/topology.go
View File

@ -32,6 +32,7 @@ import (
type selectParam struct {
excludeHosts []string
replicaNum int
selectType uint32
rackLevel proto.RackAwareLevel
excludeRacks []string
excludeNodeSets []uint64
@ -625,7 +626,7 @@ func (nsgm *DomainManager) getHostFromNodeSetGrpSpecific(domainGrpManager *Domai
excludeRacks: nil,
}
if host, peer, err = ns.getAvailDataNodeHosts(param, 1); err != nil {
if host, peer, err = ns.getAvailDataNodeHosts(param); err != nil {
log.LogErrorf("action[getHostFromNodeSetGrpSpecific] ns[%v] zone[%v] TypeDataPartition err[%v]", ns.ID, ns.zoneName, err)
// nsg.status = dataNodesUnAvailable
continue
@ -642,7 +643,7 @@ func (nsgm *DomainManager) getHostFromNodeSetGrpSpecific(domainGrpManager *Domai
rackLevel: proto.RackAwareNone,
excludeRacks: nil,
}
if host, peer, err = ns.getAvailMetaNodeHosts(param, storeMode, 1); err != nil {
if host, peer, err = ns.getAvailMetaNodeHosts(param, storeMode); err != nil {
log.LogErrorf("action[getHostFromNodeSetGrpSpecific] ns[%v] zone[%v] type(%d) err[%v]", ns.ID, ns.zoneName, createType, err)
// nsg.status = metaNodesUnAvailable
continue
@ -749,7 +750,7 @@ func (nsgm *DomainManager) getHostFromNodeSetGrp(domainId uint64, replicaNum uin
rackLevel: proto.RackAwareNone,
excludeRacks: nil,
}
if host, peer, err = ns.getAvailDataNodeHosts(param, 1); err != nil {
if host, peer, err = ns.getAvailDataNodeHosts(param); err != nil {
log.LogWarnf("action[getHostFromNodeSetGrp] ns[%v] zone[%v] TypeDataPartition err[%v]", ns.ID, ns.zoneName, err)
// nsg.status = dataNodesUnAvailable
continue
@ -770,7 +771,7 @@ func (nsgm *DomainManager) getHostFromNodeSetGrp(domainId uint64, replicaNum uin
rackLevel: proto.RackAwareNone,
excludeRacks: nil,
}
if host, peer, err = ns.getAvailMetaNodeHosts(param, storeMode, 1); err != nil {
if host, peer, err = ns.getAvailMetaNodeHosts(param, storeMode); err != nil {
log.LogWarnf("action[getHostFromNodeSetGrp] ns[%v] zone[%v] ModeRocksDb err[%v]", ns.ID, ns.zoneName, err)
// nsg.status = metaNodesUnAvailable
continue
@ -2210,7 +2211,7 @@ func (zone *Zone) getSpaceLeft(dataType uint32) (spaceLeft uint64) {
}
}
func (zone *Zone) getAvailNodeHosts(nodeType uint32, param *selectParam, threshold float64) (newHosts []string, peers []proto.Peer, err error) {
func (zone *Zone) getAvailNodeHosts(nodeType uint32, param *selectParam) (newHosts []string, peers []proto.Peer, err error) {
if param.replicaNum == 0 {
return
}
@ -2222,7 +2223,7 @@ func (zone *Zone) getAvailNodeHosts(nodeType uint32, param *selectParam, thresho
if err != nil {
return nil, nil, errors.Trace(err, "zone[%v] alloc node set, param[%v], err %s", zone.name, param.String(), err.Error())
}
return ns.getAvailDataNodeHosts(param, threshold)
return ns.getAvailDataNodeHosts(param)
}
storeMode := proto.StoreModeMem
@ -2235,7 +2236,7 @@ func (zone *Zone) getAvailNodeHosts(nodeType uint32, param *selectParam, thresho
return nil, nil, errors.NewErrorf("zone[%v], param[%v], err[%v]", zone.name, param.String(), err.Error())
}
return ns.getAvailMetaNodeHosts(param, storeMode, threshold)
return ns.getAvailMetaNodeHosts(param, storeMode)
}
func (zone *Zone) updateNodesetSelector(cluster *Cluster, dataNodesetSelector string, metaNodesetSelector string) error {

42
master/topology_rack_test.go
View File

@ -304,7 +304,7 @@ func TestRackBoundaryConditions(t *testing.T) {
rackLevel: proto.RackAwareStrong,
}
hosts, _, err := zone.getAvailNodeHosts(TypeDataPartition, param, 1)
hosts, _, err := zone.getAvailNodeHosts(TypeDataPartition, param)
require.NoError(t, err, "Should handle replicaNum=0 gracefully")
require.Equal(t, 0, len(hosts), "Should return empty hosts for replicaNum=0")
@ -317,7 +317,7 @@ func TestRackBoundaryConditions(t *testing.T) {
})
param.replicaNum = 1
_, _, err = zone.getAvailNodeHosts(TypeDataPartition, param, 1)
_, _, err = zone.getAvailNodeHosts(TypeDataPartition, param)
require.Error(t, err, "Should return error when no writable nodes available")
// Test case 3: Test with maximum replica number for strong rack awareness
@ -330,14 +330,14 @@ func TestRackBoundaryConditions(t *testing.T) {
// In RackAwareStrong mode, can only select up to the number of racks (6)
param.replicaNum = 6 // Maximum possible with 6 racks in strong mode
hosts, _, err = zone.getAvailNodeHosts(TypeDataPartition, param, 1)
hosts, _, err = zone.getAvailNodeHosts(TypeDataPartition, param)
require.NoError(t, err, "Should handle maximum replica number for strong rack awareness")
require.Equal(t, 6, len(hosts), "Should return 6 hosts (one per rack)")
// Test case 4: Test with weak rack awareness to select more nodes
param.rackLevel = proto.RackAwareWeak
param.replicaNum = 12 // All available nodes
hosts, _, err = zone.getAvailNodeHosts(TypeDataPartition, param, 1)
hosts, _, err = zone.getAvailNodeHosts(TypeDataPartition, param)
require.NoError(t, err, "Should handle maximum replica number with weak rack awareness")
require.Equal(t, 12, len(hosts), "Should return all available hosts with weak rack awareness")
}
@ -365,7 +365,7 @@ func TestRackMetaNodeRackAwareness(t *testing.T) {
rackLevel: proto.RackAwareStrong,
}
hosts, _, err := ns.getAvailMetaNodeHosts(param, proto.StoreModeMem, 1)
hosts, _, err := ns.getAvailMetaNodeHosts(param, proto.StoreModeMem)
require.NoError(t, err, "Should successfully select meta nodes with rack awareness")
require.Equal(t, 4, len(hosts), "Should select 4 meta node hosts")
@ -1047,20 +1047,20 @@ func TestRackComprehensiveExcludeHosts(t *testing.T) {
}
// Start with no exclusions
hosts, _, err := zone.getAvailNodeHosts(TypeDataPartition, param, 1)
hosts, _, err := zone.getAvailNodeHosts(TypeDataPartition, param)
require.NoError(t, err, "Should succeed with no exclusions")
require.Equal(t, 4, len(hosts), "Should select 4 hosts")
// Exclude some hosts gradually
param.excludeHosts = []string{hosts[0]} // Exclude first selected host
hosts, _, err = zone.getAvailNodeHosts(TypeDataPartition, param, 1)
hosts, _, err = zone.getAvailNodeHosts(TypeDataPartition, param)
require.NoError(t, err, "Should succeed after excluding one host")
require.Equal(t, 4, len(hosts), "Should still select 4 hosts")
require.NotContains(t, hosts, param.excludeHosts[0], "Excluded host should not be selected")
// Exclude more hosts
param.excludeHosts = append(param.excludeHosts, hosts[0], hosts[1]) // Exclude 3 hosts total
hosts, _, err = zone.getAvailNodeHosts(TypeDataPartition, param, 1)
hosts, _, err = zone.getAvailNodeHosts(TypeDataPartition, param)
require.NoError(t, err, "Should succeed after excluding more hosts")
require.Equal(t, 4, len(hosts), "Should still select 4 hosts")
@ -1070,7 +1070,7 @@ func TestRackComprehensiveExcludeHosts(t *testing.T) {
"192.168.1.11:17310", "192.168.1.12:17310", "192.168.1.13:17310", "192.168.1.14:17310", // All hosts from rack2
}
hosts, _, err = zone.getAvailNodeHosts(TypeDataPartition, param, 1)
hosts, _, err = zone.getAvailNodeHosts(TypeDataPartition, param)
require.NoError(t, err, "Should succeed after excluding hosts from 2 racks")
require.Equal(t, 4, len(hosts), "Should select 4 hosts from remaining racks")
@ -1107,19 +1107,19 @@ func TestRackExcludeHostsWithDifferentLevels(t *testing.T) {
excludeHosts: excludeHosts,
}
hosts, _, err := zone.getAvailNodeHosts(TypeDataPartition, param, 1)
hosts, _, err := zone.getAvailNodeHosts(TypeDataPartition, param)
require.NoError(t, err, "Should succeed with RackAwareNone")
require.Equal(t, 4, len(hosts), "Should select 4 hosts")
// Test with RackAwareWeak
param.rackLevel = proto.RackAwareWeak
hosts, _, err = zone.getAvailNodeHosts(TypeDataPartition, param, 1)
hosts, _, err = zone.getAvailNodeHosts(TypeDataPartition, param)
require.NoError(t, err, "Should succeed with RackAwareWeak")
require.Equal(t, 4, len(hosts), "Should select 4 hosts")
// Test with RackAwareStrong
param.rackLevel = proto.RackAwareStrong
hosts, _, err = zone.getAvailNodeHosts(TypeDataPartition, param, 1)
hosts, _, err = zone.getAvailNodeHosts(TypeDataPartition, param)
require.NoError(t, err, "Should succeed with RackAwareStrong")
require.Equal(t, 4, len(hosts), "Should select 4 hosts")
@ -1505,7 +1505,7 @@ func TestZoneWeakRackAwarenessWithRackExclusions(t *testing.T) {
excludeRacks: []string{"rack1", "rack2"}, // Exclude 2 racks
}
hosts, _, err := zone.getAvailNodeHosts(TypeDataPartition, param, 1)
hosts, _, err := zone.getAvailNodeHosts(TypeDataPartition, param)
require.NoError(t, err, "Zone level weak rack awareness should succeed with rack exclusions")
require.Equal(t, 3, len(hosts), "Should select 3 hosts")
@ -1571,7 +1571,7 @@ func TestZoneWeakRackAwarenessInsufficientRacks(t *testing.T) {
rackLevel: proto.RackAwareWeak,
}
hosts, _, err := zone.getAvailNodeHosts(TypeDataPartition, param, 1)
hosts, _, err := zone.getAvailNodeHosts(TypeDataPartition, param)
require.NoError(t, err, "Zone level weak rack awareness should succeed with insufficient racks")
require.Equal(t, 3, len(hosts), "Should select 3 hosts")
@ -1635,7 +1635,7 @@ func TestZoneWeakRackAwarenessWithHostExclusions(t *testing.T) {
},
}
hosts, _, err := zone.getAvailNodeHosts(TypeDataPartition, param, 1)
hosts, _, err := zone.getAvailNodeHosts(TypeDataPartition, param)
require.NoError(t, err, "Zone level weak rack awareness should succeed with host exclusions")
require.Equal(t, 4, len(hosts), "Should select 4 hosts")
@ -1699,7 +1699,7 @@ func TestZoneWeakVsStrongRackAwarenessComparison(t *testing.T) {
rackLevel: proto.RackAwareStrong,
}
hosts, _, err := zone.getAvailNodeHosts(TypeDataPartition, strongParam, 1)
hosts, _, err := zone.getAvailNodeHosts(TypeDataPartition, strongParam)
require.Error(t, err, "Zone level strong rack awareness should fail when not enough racks")
require.Nil(t, hosts, "Zone level strong rack awareness should return nil hosts when failing")
@ -1709,7 +1709,7 @@ func TestZoneWeakVsStrongRackAwarenessComparison(t *testing.T) {
rackLevel: proto.RackAwareWeak,
}
hosts, _, err = zone.getAvailNodeHosts(TypeDataPartition, weakParam, 1)
hosts, _, err = zone.getAvailNodeHosts(TypeDataPartition, weakParam)
require.NoError(t, err, "Zone level weak rack awareness should succeed with fallback mechanism")
require.Equal(t, replicaNum, len(hosts), "Should select %d hosts with weak rack awareness", replicaNum)
@ -1817,7 +1817,7 @@ func TestRackGetAvailMetaNodeHostsRocksDB(t *testing.T) {
rackLevel: proto.RackAwareNone,
}
hosts, peers, err := ns.getAvailMetaNodeHosts(param, proto.StoreModeRocksDb, 1)
hosts, peers, err := ns.getAvailMetaNodeHosts(param, proto.StoreModeRocksDb)
require.NoError(t, err, "Should successfully select meta nodes for RocksDB mode")
require.Equal(t, 3, len(hosts), "Should select 3 meta node hosts")
require.Equal(t, 3, len(peers), "Should return 3 peers")
@ -1844,7 +1844,7 @@ func TestRackGetAvailMetaNodeHostsRocksDB(t *testing.T) {
rackLevel: proto.RackAwareStrong,
}
hosts, peers, err := ns.getAvailMetaNodeHosts(param, proto.StoreModeRocksDb, 1)
hosts, peers, err := ns.getAvailMetaNodeHosts(param, proto.StoreModeRocksDb)
require.NoError(t, err, "Should successfully select meta nodes with strong rack awareness")
require.Equal(t, 2, len(hosts), "Should select 2 meta node hosts")
require.Equal(t, 2, len(peers), "Should return 2 peers")
@ -1895,7 +1895,7 @@ func TestRackGetAvailMetaNodeHostsRocksDBThreshold(t *testing.T) {
rackLevel: proto.RackAwareNone,
}
hosts, peers, err := ns.getAvailMetaNodeHosts(param, proto.StoreModeRocksDb, 1)
hosts, peers, err := ns.getAvailMetaNodeHosts(param, proto.StoreModeRocksDb)
require.NoError(t, err, "Should successfully select meta nodes within RocksDB threshold")
require.Equal(t, 2, len(hosts), "Should select 2 meta node hosts")
require.Equal(t, 2, len(peers), "Should return 2 peers")
@ -1943,7 +1943,7 @@ func TestRackGetAvailMetaNodeHostsInsufficientRocksDBNodes(t *testing.T) {
rackLevel: proto.RackAwareNone,
}
_, _, err := ns.getAvailMetaNodeHosts(param, proto.StoreModeRocksDb, 1)
_, _, err := ns.getAvailMetaNodeHosts(param, proto.StoreModeRocksDb)
require.Error(t, err, "Should return error when insufficient RocksDB nodes available")
require.Contains(t, err.Error(), "no enough writable hosts", "Error should indicate insufficient writable hosts")

8
master/topology_test.go
View File

@ -62,7 +62,7 @@ func TestSingleZone(t *testing.T) {
rackLevel: proto.RackAwareNone,
excludeRacks: nil,
}
newHosts, _, err := zones[0].getAvailNodeHosts(TypeDataPartition, param, 1)
newHosts, _, err := zones[0].getAvailNodeHosts(TypeDataPartition, param)
require.NoError(t, err)
t.Log(newHosts)
topo.deleteDataNode(createDataNodeForTopo(mds1Addr, zoneName, nodeSet))
@ -146,18 +146,18 @@ func TestAllocZones(t *testing.T) {
rackLevel: proto.RackAwareNone,
}
// don't cross zone
hosts, _, err := cluster.getHostFromNormalZone(TypeDataPartition, nil, 1, "", proto.MediaType_Unspecified, param, 1)
hosts, _, err := cluster.getHostFromNormalZone(TypeDataPartition, nil, 1, "", proto.MediaType_Unspecified, param)
require.NoError(t, err)
t.Logf("ChooseTargetDataHosts in single zone,hosts[%v]", hosts)
// cross zone
_, _, err = cluster.getHostFromNormalZone(TypeDataPartition, nil, 2, "", proto.MediaType_Unspecified, param, 1)
_, _, err = cluster.getHostFromNormalZone(TypeDataPartition, nil, 2, "", proto.MediaType_Unspecified, param)
require.NoError(t, err)
param.replicaNum = 2
// specific zone
hosts, _, err = cluster.getHostFromNormalZone(TypeDataPartition, nil, 3, zoneName1+","+zoneName2, proto.MediaType_Unspecified, param, 1)
hosts, _, err = cluster.getHostFromNormalZone(TypeDataPartition, nil, 3, zoneName1+","+zoneName2, proto.MediaType_Unspecified, param)
require.NoError(t, err)
require.EqualValues(t, getZoneCntFunc(hosts), 2)

5
proto/admin_proto.go
View File

@ -1577,6 +1577,11 @@ const (
DistributionOptimization
)
const (
SelectType_Normal uint32 = iota
SelectType_DistributionOptimization
)
type BackupDataPartitionInfo struct {
Addr string
Disk string

4
proto/model.go
View File

@ -171,7 +171,9 @@ type ClusterView struct {
AutoDpMetaRepairParallelCnt int
EnableAutoDecommission bool
AutoDecommissionDiskInterval string
EnableAutoDistributionOptimization bool
EnableDistributionOptimization bool
DistributionOptimizationThreshold float64
DistributionOptimizationConDpCnt int64
DecommissionFirstHostDiskParallelLimit uint64
DecommissionLimit uint64
DecommissionDiskLimit uint32

6
sdk/master/api_admin.go
View File

@ -606,7 +606,7 @@ func (api *AdminAPI) SetClusterParas(batchCount, markDeleteRate, deleteWorkerSle
dpRepairTimeout string, dpTimeout string, mpTimeout string, dpBackupTimeout string,
decommissionDpLimit, decommissionDiskLimit, forbidWriteOpOfProtoVersion0 string, mediaType string,
handleTimeout string, readDataNodeTimeout string, rackAware string,
distributionOptimizationConcurrentDpCount, distributionOptimizationThreshold string,
distributionOptimizationConDpCnt, distributionOptimizationThreshold string,
) (err error) {
request := newRequest(get, proto.AdminSetNodeInfo).Header(api.h)
request.addParam("batchCount", batchCount)
@ -674,8 +674,8 @@ func (api *AdminAPI) SetClusterParas(batchCount, markDeleteRate, deleteWorkerSle
request.addParam("rackAware", rackAware)
}
// Distribution optimization parameters
if distributionOptimizationConcurrentDpCount != "" {
request.addParam("distributionOptimizationConcurrentDpCount", distributionOptimizationConcurrentDpCount)
if distributionOptimizationConDpCnt != "" {
request.addParam("distributionOptimizationConDpCnt", distributionOptimizationConDpCnt)
}
if distributionOptimizationThreshold != "" {
request.addParam("distributionOptimizationThreshold", distributionOptimizationThreshold)