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): support config decommission settings

Browse Source 
Signed-off-by: NaturalSelect <huangzhibin1@oppo.com>
This commit is contained in:
NaturalSelect 2024-06-20 19:01:02 +08:00 committed by AmazingChi
parent 8386685205
commit 5365949f7c
26 changed files with 931 additions and 289 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
cli/cmd/cluster.go
View File

@ -253,7 +253,10 @@ func newClusterSetParasCmd(client *master.MasterClient) *cobra.Command {
dataNodeSelector := ""
metaNodeSelector := ""
markBrokenDiskThreshold := ""
autoDecommissionDisk := ""
autoDecommissionDiskInterval := ""
autoDpMetaRepair := ""
autoDpMetaRepairParallelCnt := ""
opMaxMpCntLimit := ""
dpRepairTimeout := ""
dpTimeout := ""
@ -273,11 +276,36 @@ func newClusterSetParasCmd(client *master.MasterClient) *cobra.Command {
}
markBrokenDiskThreshold = fmt.Sprintf("%v", val)
}
if autoDecommissionDisk != "" {
if _, err = strconv.ParseBool(autoDecommissionDisk); err != nil {
return
}
}
if autoDecommissionDiskInterval != "" {
var interval time.Duration
interval, err = time.ParseDuration(autoDecommissionDiskInterval)
if err != nil {
return
}
if interval < time.Second {
err = fmt.Errorf("auto decommission disk interval %v smaller than 1s", interval)
return
}
autoDecommissionDiskInterval = strconv.FormatInt(int64(interval), 10)
}
if autoDpMetaRepair != "" {
if _, err = strconv.ParseBool(autoDpMetaRepair); err != nil {
return
}
}
if autoDpMetaRepairParallelCnt != "" {
if _, err = strconv.ParseInt(autoDpMetaRepairParallelCnt, 10, 64); err != nil {
return
}
}
if dpRepairTimeout != "" {
var repairTimeout time.Duration
@ -308,7 +336,10 @@ func newClusterSetParasCmd(client *master.MasterClient) *cobra.Command {
if err = client.AdminAPI().SetClusterParas(optDelBatchCount, optMarkDeleteRate, optDelWorkerSleepMs,
optAutoRepairRate, optLoadFactor, opMaxDpCntLimit, opMaxMpCntLimit, clientIDKey,
dataNodesetSelector, metaNodesetSelector,
dataNodeSelector, metaNodeSelector, markBrokenDiskThreshold, autoDpMetaRepair, dpRepairTimeout, dpTimeout); err != nil {
dataNodeSelector, metaNodeSelector, markBrokenDiskThreshold,
autoDecommissionDisk, autoDecommissionDiskInterval,
autoDpMetaRepair, autoDpMetaRepairParallelCnt,
dpRepairTimeout, dpTimeout); err != nil {
return
}
stdout("Cluster parameters has been set successfully. \n")
@ -328,8 +359,11 @@ func newClusterSetParasCmd(client *master.MasterClient) *cobra.Command {
cmd.Flags().StringVar(&metaNodeSelector, CliFlagMetaNodeSelector, "", "Set the node select policy(metanode) for cluster")
cmd.Flags().StringVar(&markBrokenDiskThreshold, CliFlagMarkDiskBrokenThreshold, "", "Threshold to mark disk as broken")
cmd.Flags().StringVar(&autoDpMetaRepair, CliFlagAutoDpMetaRepair, "", "Enable or disable auto data partition meta repair")
cmd.Flags().StringVar(&autoDpMetaRepairParallelCnt, CliFlagAutoDpMetaRepairParallelCnt, "", "Parallel count of auto data partition meta repair")
cmd.Flags().StringVar(&dpRepairTimeout, CliFlagDpRepairTimeout, "", "Data partition repair timeout(example: 1h)")
cmd.Flags().StringVar(&dpTimeout, CliFlagDpTimeout, "", "Data partition heartbeat timeout(example: 10s)")
cmd.Flags().StringVar(&autoDecommissionDisk, CliFlagAutoDecommissionDisk, "", "Enable od disable auto decommission disk")
cmd.Flags().StringVar(&autoDecommissionDiskInterval, CliFlagAutoDecommissionDiskInterval, "", "Interval of auto decommission disk(example: 10s)")
return cmd
}

141
cli/cmd/const.go
View File

@ -68,75 +68,78 @@ const (
CliResourceConfig = "config"
// Flags
CliFlagName = "name"
CliFlagOnwer = "user"
CliFlagDataPartitionSize = "dp-size"
CliFlagDataPartitionCount = "dp-count"
CliFlagMetaPartitionCount = "mp-count"
CliFlagReplicas = "replicas"
CliFlagEnable = "enable"
CliFlagEnableFollowerRead = "follower-read"
CliFlagAuthenticate = "authenticate"
CliFlagCapacity = "capacity"
CliFlagBusiness = "description"
CliFlagMPCount = "mp-count"
CliFlagDPCount = "dp-count"
CliFlagReplicaNum = "replica-num"
CliFlagSize = "size"
CliFlagVolType = "vol-type"
CliFlagFollowerRead = "follower-read"
CliFlagCacheRuleKey = "cache-rule-key"
CliFlagEbsBlkSize = "ebs-blk-size"
CliFlagCacheCapacity = "cache-capacity"
CliFlagCacheAction = "cache-action"
CliFlagCacheThreshold = "cache-threshold"
CliFlagCacheTTL = "cache-ttl"
CliFlagCacheHighWater = "cache-high-water"
CliFlagCacheLowWater = "cache-low-water"
CliFlagCacheLRUInterval = "cache-lru-interval"
CliFlagCacheRule = "cache-rule"
CliFlagThreshold = "threshold"
CliFlagAddress = "addr"
CliFlagDiskPath = "path"
CliFlagAuthKey = "authkey"
CliFlagINodeStartID = "inode-start"
CliFlagId = "id"
CliFlagZoneName = "zone-name"
CliFlagDescription = "description"
CliFlagAutoRepairRate = "autoRepairRate"
CliFlagDelBatchCount = "batchCount"
CliFlagDelWorkerSleepMs = "deleteWorkerSleepMs"
CliFlagLoadFactor = "loadFactor"
CliFlagMarkDelRate = "markDeleteRate"
CliFlagMaxDpCntLimit = "maxDpCntLimit"
CliFlagMaxMpCntLimit = "maxMpCntLimit"
CliFlagDataNodeSelector = "dataNodeSelector"
CliFlagMetaNodeSelector = "metaNodeSelector"
CliFlagDataNodesetSelector = "dataNodesetSelector"
CliFlagMetaNodesetSelector = "metaNodesetSelector"
CliFlagCrossZone = "crossZone"
CliNormalZonesFirst = "normalZonesFirst"
CliFlagCount = "count"
CliDpReadOnlyWhenVolFull = "readonly-when-full"
CliTxMask = "transaction-mask"
CliTxTimeout = "transaction-timeout"
CliTxOpLimit = "transaction-limit"
CliTxConflictRetryNum = "tx-conflict-retry-num"
CliTxConflictRetryInterval = "tx-conflict-retry-Interval"
CliTxForceReset = "transaction-force-reset"
CliFlagMaxFiles = "maxFiles"
CliFlagMaxBytes = "maxBytes"
CliFlagMaxConcurrencyInode = "maxConcurrencyInode"
CliFlagForceInode = "forceInode"
CliFlagEnableQuota = "enableQuota"
CliFlagDeleteLockTime = "delete-lock-time"
CliFlagClientIDKey = "clientIDKey"
CliFlagMarkDiskBrokenThreshold = "markBrokenDiskThreshold"
CliFlagForce = "force"
CliFlagEnableCrossZone = "cross-zone"
CliFlagAutoDpMetaRepair = "autoDpMetaRepair"
CliFlagDpRepairTimeout = "dpRepairTimeout"
CliFlagDpTimeout = "dpTimeout"
CliFlagName = "name"
CliFlagOnwer = "user"
CliFlagDataPartitionSize = "dp-size"
CliFlagDataPartitionCount = "dp-count"
CliFlagMetaPartitionCount = "mp-count"
CliFlagReplicas = "replicas"
CliFlagEnable = "enable"
CliFlagEnableFollowerRead = "follower-read"
CliFlagAuthenticate = "authenticate"
CliFlagCapacity = "capacity"
CliFlagBusiness = "description"
CliFlagMPCount = "mp-count"
CliFlagDPCount = "dp-count"
CliFlagReplicaNum = "replica-num"
CliFlagSize = "size"
CliFlagVolType = "vol-type"
CliFlagFollowerRead = "follower-read"
CliFlagCacheRuleKey = "cache-rule-key"
CliFlagEbsBlkSize = "ebs-blk-size"
CliFlagCacheCapacity = "cache-capacity"
CliFlagCacheAction = "cache-action"
CliFlagCacheThreshold = "cache-threshold"
CliFlagCacheTTL = "cache-ttl"
CliFlagCacheHighWater = "cache-high-water"
CliFlagCacheLowWater = "cache-low-water"
CliFlagCacheLRUInterval = "cache-lru-interval"
CliFlagCacheRule = "cache-rule"
CliFlagThreshold = "threshold"
CliFlagAddress = "addr"
CliFlagDiskPath = "path"
CliFlagAuthKey = "authkey"
CliFlagINodeStartID = "inode-start"
CliFlagId = "id"
CliFlagZoneName = "zone-name"
CliFlagDescription = "description"
CliFlagAutoRepairRate = "autoRepairRate"
CliFlagDelBatchCount = "batchCount"
CliFlagDelWorkerSleepMs = "deleteWorkerSleepMs"
CliFlagLoadFactor = "loadFactor"
CliFlagMarkDelRate = "markDeleteRate"
CliFlagMaxDpCntLimit = "maxDpCntLimit"
CliFlagMaxMpCntLimit = "maxMpCntLimit"
CliFlagDataNodeSelector = "dataNodeSelector"
CliFlagMetaNodeSelector = "metaNodeSelector"
CliFlagDataNodesetSelector = "dataNodesetSelector"
CliFlagMetaNodesetSelector = "metaNodesetSelector"
CliFlagCrossZone = "crossZone"
CliNormalZonesFirst = "normalZonesFirst"
CliFlagCount = "count"
CliDpReadOnlyWhenVolFull = "readonly-when-full"
CliTxMask = "transaction-mask"
CliTxTimeout = "transaction-timeout"
CliTxOpLimit = "transaction-limit"
CliTxConflictRetryNum = "tx-conflict-retry-num"
CliTxConflictRetryInterval = "tx-conflict-retry-Interval"
CliTxForceReset = "transaction-force-reset"
CliFlagMaxFiles = "maxFiles"
CliFlagMaxBytes = "maxBytes"
CliFlagMaxConcurrencyInode = "maxConcurrencyInode"
CliFlagForceInode = "forceInode"
CliFlagEnableQuota = "enableQuota"
CliFlagDeleteLockTime = "delete-lock-time"
CliFlagClientIDKey = "clientIDKey"
CliFlagMarkDiskBrokenThreshold = "markBrokenDiskThreshold"
CliFlagForce = "force"
CliFlagEnableCrossZone = "cross-zone"
CliFlagAutoDpMetaRepair = "autoDpMetaRepair"
CliFlagAutoDpMetaRepairParallelCnt = "autoDpMetaRepairParallelCnt"
CliFlagDpRepairTimeout = "dpRepairTimeout"
CliFlagDpTimeout = "dpTimeout"
CliFlagAutoDecommissionDisk = "autoDecommissionDisk"
CliFlagAutoDecommissionDiskInterval = "autoDecommissionDiskInterval"
// CliFlagSetDataPartitionCount = "count" use dp-count instead

24
cli/cmd/fmt.go
View File

@ -65,17 +65,19 @@ func formatClusterView(cv *proto.ClusterView, cn *proto.ClusterNodeInfo, cp *pro
sb.WriteString(fmt.Sprintf(" DataNode available : %v GB\n", cv.DataNodeStatInfo.AvailGB))
sb.WriteString(fmt.Sprintf(" DataNode total : %v GB\n", cv.DataNodeStatInfo.TotalGB))
sb.WriteString(fmt.Sprintf(" Volume count : %v\n", len(cv.VolStatInfo)))
sb.WriteString(fmt.Sprintf(" Allow Mp Decomm : %v\n", formatEnabledDisabled(!cv.ForbidMpDecommission)))
sb.WriteString(fmt.Sprintf(" EbsAddr : %v\n", cp.EbsAddr))
sb.WriteString(fmt.Sprintf(" LoadFactor : %v\n", cn.LoadFactor))
sb.WriteString(fmt.Sprintf(" DpRepairTimeout : %v\n", cv.DpRepairTimeout))
sb.WriteString(fmt.Sprintf(" DataPartitionTimeout : %v\n", cv.DpTimeout))
sb.WriteString(fmt.Sprintf(" volDeletionDelayTime : %v h\n", cv.VolDeletionDelayTimeHour))
sb.WriteString(fmt.Sprintf(" EnableAutoDecommission: %v\n", cv.EnableAutoDecommission))
sb.WriteString(fmt.Sprintf(" EnableAutoDpMetaRepair: %v\n", cv.EnableAutoDpMetaRepair))
sb.WriteString(fmt.Sprintf(" MarkDiskBrokenThreshold : %v\n", strutil.FormatPercent(cv.MarkDiskBrokenThreshold)))
sb.WriteString(fmt.Sprintf(" DecommissionDiskLimit: %v\n", cv.DecommissionDiskLimit))
sb.WriteString(fmt.Sprintf(" Volume count : %v\n", len(cv.VolStatInfo)))
sb.WriteString(fmt.Sprintf(" Allow Mp Decomm : %v\n", formatEnabledDisabled(!cv.ForbidMpDecommission)))
sb.WriteString(fmt.Sprintf(" EbsAddr : %v\n", cp.EbsAddr))
sb.WriteString(fmt.Sprintf(" LoadFactor : %v\n", cn.LoadFactor))
sb.WriteString(fmt.Sprintf(" DpRepairTimeout : %v\n", cv.DpRepairTimeout))
sb.WriteString(fmt.Sprintf(" DataPartitionTimeout : %v\n", cv.DpTimeout))
sb.WriteString(fmt.Sprintf(" volDeletionDelayTime : %v h\n", cv.VolDeletionDelayTimeHour))
sb.WriteString(fmt.Sprintf(" EnableAutoDecommission : %v\n", cv.EnableAutoDecommission))
sb.WriteString(fmt.Sprintf(" AutoDecommissionDiskIntervak : %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(" MarkDiskBrokenThreshold : %v\n", strutil.FormatPercent(cv.MarkDiskBrokenThreshold)))
sb.WriteString(fmt.Sprintf(" DecommissionDiskLimit : %v\n", cv.DecommissionDiskLimit))
return sb.String()
}

33
master/api_args_parse.go
View File

@ -1329,6 +1329,28 @@ func parseAndExtractSetNodeInfoParams(r *http.Request) (params map[string]interf
params[markDiskBrokenThresholdKey] = val
}
if value = r.FormValue(autoDecommissionDiskKey); value != "" {
noParams = false
val := false
val, err = strconv.ParseBool(value)
if err != nil {
err = unmatchedKey(autoDecommissionDiskKey)
return
}
params[autoDecommissionDiskKey] = val
}
if value = r.FormValue(autoDecommissionDiskIntervalKey); value != "" {
noParams = false
val := int64(0)
val, err = strconv.ParseInt(value, 10, 64)
if err != nil {
err = unmatchedKey(autoDecommissionDiskIntervalKey)
return
}
params[autoDecommissionDiskIntervalKey] = time.Duration(val)
}
if value = r.FormValue(autoDpMetaRepairKey); value != "" {
noParams = false
val := false
@ -1340,6 +1362,17 @@ func parseAndExtractSetNodeInfoParams(r *http.Request) (params map[string]interf
params[autoDpMetaRepairKey] = val
}
if value = r.FormValue(autoDpMetaRepairParallelCntKey); value != "" {
noParams = false
val := int64(0)
val, err = strconv.ParseInt(value, 10, 32)
if err != nil {
err = unmatchedKey(autoDpMetaRepairParallelCntKey)
return
}
params[autoDpMetaRepairParallelCntKey] = int(val)
}
if value = r.FormValue(dpTimeoutKey); value != "" {
noParams = false
val := int64(0)

78
master/api_service.go
View File

@ -783,29 +783,31 @@ func (m *Server) getCluster(w http.ResponseWriter, r *http.Request) {
}()
cv := &proto.ClusterView{
Name: m.cluster.Name,
CreateTime: time.Unix(m.cluster.CreateTime, 0).Format(proto.TimeFormat),
LeaderAddr: m.leaderInfo.addr,
DisableAutoAlloc: m.cluster.DisableAutoAllocate,
ForbidMpDecommission: m.cluster.ForbidMpDecommission,
MetaNodeThreshold: m.cluster.cfg.MetaNodeThreshold,
Applied: m.fsm.applied,
MaxDataPartitionID: m.cluster.idAlloc.dataPartitionID,
MaxMetaNodeID: m.cluster.idAlloc.commonID,
MaxMetaPartitionID: m.cluster.idAlloc.metaPartitionID,
VolDeletionDelayTimeHour: m.cluster.cfg.volDelayDeleteTimeHour,
DpRepairTimeout: m.cluster.GetDecommissionDataPartitionRecoverTimeOut(),
MarkDiskBrokenThreshold: m.cluster.getMarkDiskBrokenThreshold(),
EnableAutoDpMetaRepair: m.cluster.getEnableAutoDpMetaRepair(),
EnableAutoDecommission: m.cluster.AutoDecommissionDiskIsEnabled(),
DecommissionDiskLimit: m.cluster.GetDecommissionDiskLimit(),
DpTimeout: time.Duration(m.cluster.getDataPartitionTimeoutSec()) * time.Second,
MasterNodes: make([]proto.NodeView, 0),
MetaNodes: make([]proto.NodeView, 0),
DataNodes: make([]proto.NodeView, 0),
VolStatInfo: make([]*proto.VolStatInfo, 0),
BadPartitionIDs: make([]proto.BadPartitionView, 0),
BadMetaPartitionIDs: make([]proto.BadPartitionView, 0),
Name: m.cluster.Name,
CreateTime: time.Unix(m.cluster.CreateTime, 0).Format(proto.TimeFormat),
LeaderAddr: m.leaderInfo.addr,
DisableAutoAlloc: m.cluster.DisableAutoAllocate,
ForbidMpDecommission: m.cluster.ForbidMpDecommission,
MetaNodeThreshold: m.cluster.cfg.MetaNodeThreshold,
Applied: m.fsm.applied,
MaxDataPartitionID: m.cluster.idAlloc.dataPartitionID,
MaxMetaNodeID: m.cluster.idAlloc.commonID,
MaxMetaPartitionID: m.cluster.idAlloc.metaPartitionID,
VolDeletionDelayTimeHour: m.cluster.cfg.volDelayDeleteTimeHour,
DpRepairTimeout: m.cluster.GetDecommissionDataPartitionRecoverTimeOut(),
MarkDiskBrokenThreshold: m.cluster.getMarkDiskBrokenThreshold(),
EnableAutoDpMetaRepair: m.cluster.getEnableAutoDpMetaRepair(),
AutoDpMetaRepairParallelCnt: m.cluster.GetAutoDpMetaRepairParallelCnt(),
EnableAutoDecommission: m.cluster.AutoDecommissionDiskIsEnabled(),
AutoDecommissionDiskInterval: m.cluster.GetAutoDecommissionDiskInterval(),
DecommissionDiskLimit: m.cluster.GetDecommissionDiskLimit(),
DpTimeout: time.Duration(m.cluster.getDataPartitionTimeoutSec()) * time.Second,
MasterNodes: make([]proto.NodeView, 0),
MetaNodes: make([]proto.NodeView, 0),
DataNodes: make([]proto.NodeView, 0),
VolStatInfo: make([]*proto.VolStatInfo, 0),
BadPartitionIDs: make([]proto.BadPartitionView, 0),
BadMetaPartitionIDs: make([]proto.BadPartitionView, 0),
}
vols := m.cluster.allVolNames()
@ -3160,6 +3162,27 @@ func (m *Server) setNodeInfoHandler(w http.ResponseWriter, r *http.Request) {
}
}
if val, ok := params[autoDecommissionDiskKey]; ok {
if autoDecomm, ok := val.(bool); ok {
old := m.cluster.EnableAutoDecommissionDisk.Load()
m.cluster.EnableAutoDecommissionDisk.Store(autoDecomm)
if err = m.cluster.syncPutCluster(); err != nil {
m.cluster.EnableAutoDecommissionDisk.Store(old)
sendErrReply(w, r, newErrHTTPReply(err))
return
}
}
}
if val, ok := params[autoDecommissionDiskIntervalKey]; ok {
if interval, ok := val.(time.Duration); ok {
if err = m.cluster.setAutoDecommissionDiskInterval(interval); err != nil {
sendErrReply(w, r, newErrHTTPReply(err))
return
}
}
}
if val, ok := params[autoDpMetaRepairKey]; ok {
if autoRepair, ok := val.(bool); ok {
if err = m.cluster.setEnableAutoDpMetaRepair(autoRepair); err != nil {
@ -3169,6 +3192,15 @@ func (m *Server) setNodeInfoHandler(w http.ResponseWriter, r *http.Request) {
}
}
if val, ok := params[autoDpMetaRepairParallelCntKey]; ok {
if cnt, ok := val.(int); ok {
if err = m.cluster.setAutoDpMetaRepairParallelCnt(cnt); err != nil {
sendErrReply(w, r, newErrHTTPReply(err))
return
}
}
}
if val, ok := params[dpTimeoutKey]; ok {
if dpTimeout, ok := val.(int64); ok {
if err = m.cluster.setDataPartitionTimeout(dpTimeout); err != nil {

36
master/api_service_test.go
View File

@ -1645,6 +1645,30 @@ func TestSetMarkDiskBrokenThreshold(t *testing.T) {
require.EqualValues(t, oldVal, server.cluster.getMarkDiskBrokenThreshold())
}
func TestSetEnableAutoDecommissionDisk(t *testing.T) {
reqUrl := fmt.Sprintf("%v%v", hostAddr, proto.AdminSetNodeInfo)
oldVal := server.cluster.EnableAutoDecommissionDisk.Load()
setVal := !oldVal
setUrl := fmt.Sprintf("%v?%v=%v&dirSizeLimit=0", reqUrl, autoDecommissionDiskKey, setVal)
unsetUrl := fmt.Sprintf("%v?%v=%v&dirSizeLimit=0", reqUrl, autoDecommissionDiskKey, oldVal)
process(setUrl, t)
require.EqualValues(t, setVal, server.cluster.EnableAutoDecommissionDisk.Load())
process(unsetUrl, t)
require.EqualValues(t, oldVal, server.cluster.EnableAutoDecommissionDisk.Load())
}
func TestSetDecommissionDiskInterval(t *testing.T) {
reqUrl := fmt.Sprintf("%v%v", hostAddr, proto.AdminSetNodeInfo)
oldVal := server.cluster.GetAutoDecommissionDiskInterval()
setVal := oldVal + 10*time.Second
setUrl := fmt.Sprintf("%v?%v=%v&dirSizeLimit=0", reqUrl, autoDecommissionDiskIntervalKey, int64(setVal))
unsetUrl := fmt.Sprintf("%v?%v=%v&dirSizeLimit=0", reqUrl, autoDecommissionDiskIntervalKey, int64(oldVal))
process(setUrl, t)
require.EqualValues(t, setVal, server.cluster.GetAutoDecommissionDiskInterval())
process(unsetUrl, t)
require.EqualValues(t, oldVal, server.cluster.GetAutoDecommissionDiskInterval())
}
func TestSetEnableAutoDpMetaRepair(t *testing.T) {
reqUrl := fmt.Sprintf("%v%v", hostAddr, proto.AdminSetNodeInfo)
oldVal := server.cluster.getEnableAutoDpMetaRepair()
@ -1657,6 +1681,18 @@ func TestSetEnableAutoDpMetaRepair(t *testing.T) {
require.EqualValues(t, oldVal, server.cluster.getEnableAutoDpMetaRepair())
}
func TestSetAutoDpMetaRepairParallelCnt(t *testing.T) {
reqUrl := fmt.Sprintf("%v%v", hostAddr, proto.AdminSetNodeInfo)
oldVal := server.cluster.GetAutoDpMetaRepairParallelCnt()
setVal := 200
setUrl := fmt.Sprintf("%v?%v=%v&dirSizeLimit=0", reqUrl, autoDpMetaRepairParallelCntKey, setVal)
unsetUrl := fmt.Sprintf("%v?%v=%v&dirSizeLimit=0", reqUrl, autoDpMetaRepairParallelCntKey, oldVal)
process(setUrl, t)
require.EqualValues(t, setVal, server.cluster.GetAutoDpMetaRepairParallelCnt())
process(unsetUrl, t)
require.EqualValues(t, oldVal, server.cluster.GetAutoDpMetaRepairParallelCnt())
}
func TestSetDpTimeout(t *testing.T) {
reqUrl := fmt.Sprintf("%v%v", hostAddr, proto.AdminSetNodeInfo)
oldVal := server.cluster.getDataPartitionTimeoutSec()

48
master/cluster.go
View File

@ -86,7 +86,8 @@ type Cluster struct {
apiLimiter *ApiLimiter
DecommissionDisks sync.Map
DecommissionLimit uint64
EnableAutoDecommissionDisk bool
EnableAutoDecommissionDisk atomicutil.Bool
AutoDecommissionInterval atomicutil.Int64
AutoDecommissionDiskMux sync.Mutex
checkAutoCreateDataPartition bool
masterClient *masterSDK.MasterClient
@ -106,6 +107,7 @@ type Cluster struct {
S3ApiQosQuota *sync.Map // (api,uid,limtType) -> limitQuota
MarkDiskBrokenThreshold atomicutil.Float64
EnableAutoDpMetaRepair atomicutil.Bool
AutoDpMetaRepairParallelCnt atomicutil.Uint32
}
type delayDeleteVolInfo struct {
@ -371,6 +373,7 @@ func newCluster(name string, leaderInfo *LeaderInfo, fsm *MetadataFsm, partition
c.S3ApiQosQuota = new(sync.Map)
c.MarkDiskBrokenThreshold.Store(defaultMarkDiskBrokenThreshold)
c.EnableAutoDpMetaRepair.Store(defaultEnableDpMetaRepair)
c.AutoDecommissionInterval.Store(int64(defaultAutoDecommissionDiskInterval))
return
}
@ -4257,7 +4260,7 @@ func (c *Cluster) scheduleToBadDisk() {
if c.partition.IsRaftLeader() && c.AutoDecommissionDiskIsEnabled() && c.metaReady {
c.checkBadDisk()
}
time.Sleep(10 * time.Second)
time.Sleep(c.GetAutoDecommissionDiskInterval())
}
}()
}
@ -4916,13 +4919,50 @@ func (c *Cluster) addDecommissionDiskToNodeset(dd *DecommissionDisk) (err error)
func (c *Cluster) AutoDecommissionDiskIsEnabled() bool {
c.AutoDecommissionDiskMux.Lock()
defer c.AutoDecommissionDiskMux.Unlock()
return c.EnableAutoDecommissionDisk
return c.EnableAutoDecommissionDisk.Load()
}
func (c *Cluster) SetAutoDecommissionDisk(flag bool) {
c.AutoDecommissionDiskMux.Lock()
defer c.AutoDecommissionDiskMux.Unlock()
c.EnableAutoDecommissionDisk = flag
c.EnableAutoDecommissionDisk.Store(flag)
}
func (c *Cluster) GetAutoDecommissionDiskInterval() (interval time.Duration) {
tmp := c.AutoDecommissionInterval.Load()
if tmp == 0 {
tmp = int64(defaultAutoDecommissionDiskInterval)
}
interval = time.Duration(tmp)
return
}
func (c *Cluster) setAutoDecommissionDiskInterval(interval time.Duration) (err error) {
old := c.AutoDecommissionInterval.Load()
c.AutoDecommissionInterval.Store(int64(interval))
if err = c.syncPutCluster(); err != nil {
c.AutoDecommissionInterval.Store(old)
return
}
return
}
func (c *Cluster) GetAutoDpMetaRepairParallelCnt() (cnt int) {
cnt = int(c.AutoDpMetaRepairParallelCnt.Load())
if cnt == 0 {
cnt = defaultAutoDpMetaRepairPallarelCnt
}
return
}
func (c *Cluster) setAutoDpMetaRepairParallelCnt(cnt int) (err error) {
old := c.AutoDpMetaRepairParallelCnt.Load()
c.AutoDpMetaRepairParallelCnt.Store(uint32(cnt))
if err = c.syncPutCluster(); err != nil {
c.AutoDpMetaRepairParallelCnt.Store(old)
return
}
return
}
func (c *Cluster) GetDecommissionDataPartitionRecoverTimeOut() time.Duration {

187
master/const.go
View File

@ -50,97 +50,100 @@ const (
forbiddenKey = "forbidden"
deleteVolKey = "delete"
forceDelVolKey = "forceDelVol"
ebsBlkSizeKey = "ebsBlkSize"
cacheCapacity = "cacheCap"
cacheActionKey = "cacheAction"
cacheThresholdKey = "cacheThreshold"
cacheTTLKey = "cacheTTL"
cacheHighWaterKey = "cacheHighWater"
cacheLowWaterKey = "cacheLowWater"
cacheLRUIntervalKey = "cacheLRUInterval"
clientVersion = "version"
domainIdKey = "domainId"
volOwnerKey = "owner"
volAuthKey = "authKey"
replicaNumKey = "replicaNum"
followerReadKey = "followerRead"
authenticateKey = "authenticate"
akKey = "ak"
keywordsKey = "keywords"
zoneNameKey = "zoneName"
nodesetIdKey = "nodesetId"
crossZoneKey = "crossZone"
normalZonesFirstKey = "normalZonesFirst"
userKey = "user"
nodeHostsKey = "hosts"
nodeDeleteBatchCountKey = "batchCount"
nodeMarkDeleteRateKey = "markDeleteRate"
nodeDeleteWorkerSleepMs = "deleteWorkerSleepMs"
nodeAutoRepairRateKey = "autoRepairRate"
nodeDpRepairTimeOutKey = "dpRepairTimeOut"
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"
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"
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"
fullPathKey = "fullPath"
inodeKey = "inode"
quotaKey = "quotaId"
enableQuota = "enableQuota"
dpDiscardKey = "dpDiscard"
ignoreDiscardKey = "ignoreDiscard"
TrashIntervalKey = "trashInterval"
ClientIDKey = "clientIDKey"
verSeqKey = "verSeq"
Periodic = "periodic"
DecommissionType = "decommissionType"
decommissionDiskLimit = "decommissionDiskLimit"
dpRepairBlockSizeKey = "dpRepairBlockSize"
markDiskBrokenThresholdKey = "markDiskBrokenThreshold"
decommissionTypeKey = "decommissionType"
autoDpMetaRepairKey = "autoDpMetaRepair"
dpTimeoutKey = "dpTimeout"
forceDelVolKey = "forceDelVol"
ebsBlkSizeKey = "ebsBlkSize"
cacheCapacity = "cacheCap"
cacheActionKey = "cacheAction"
cacheThresholdKey = "cacheThreshold"
cacheTTLKey = "cacheTTL"
cacheHighWaterKey = "cacheHighWater"
cacheLowWaterKey = "cacheLowWater"
cacheLRUIntervalKey = "cacheLRUInterval"
clientVersion = "version"
domainIdKey = "domainId"
volOwnerKey = "owner"
volAuthKey = "authKey"
replicaNumKey = "replicaNum"
followerReadKey = "followerRead"
authenticateKey = "authenticate"
akKey = "ak"
keywordsKey = "keywords"
zoneNameKey = "zoneName"
nodesetIdKey = "nodesetId"
crossZoneKey = "crossZone"
normalZonesFirstKey = "normalZonesFirst"
userKey = "user"
nodeHostsKey = "hosts"
nodeDeleteBatchCountKey = "batchCount"
nodeMarkDeleteRateKey = "markDeleteRate"
nodeDeleteWorkerSleepMs = "deleteWorkerSleepMs"
nodeAutoRepairRateKey = "autoRepairRate"
nodeDpRepairTimeOutKey = "dpRepairTimeOut"
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"
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"
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"
fullPathKey = "fullPath"
inodeKey = "inode"
quotaKey = "quotaId"
enableQuota = "enableQuota"
dpDiscardKey = "dpDiscard"
ignoreDiscardKey = "ignoreDiscard"
TrashIntervalKey = "trashInterval"
ClientIDKey = "clientIDKey"
verSeqKey = "verSeq"
Periodic = "periodic"
DecommissionType = "decommissionType"
decommissionDiskLimit = "decommissionDiskLimit"
dpRepairBlockSizeKey = "dpRepairBlockSize"
markDiskBrokenThresholdKey = "markDiskBrokenThreshold"
decommissionTypeKey = "decommissionType"
autoDecommissionDiskKey = "autoDecommissionDisk"
autoDecommissionDiskIntervalKey = "autoDecommissionDiskInterval"
autoDpMetaRepairKey = "autoDpMetaRepair"
autoDpMetaRepairParallelCntKey = "autoDpMetaRepairParallelCnt"
dpTimeoutKey = "dpTimeout"
)
const (
@ -214,6 +217,8 @@ const (
defaultVolDelayDeleteTimeHour = 48
defaultMarkDiskBrokenThreshold = 0 // decommission all dp from disk
defaultEnableDpMetaRepair = false
defaultAutoDpMetaRepairPallarelCnt = 100
defaultAutoDecommissionDiskInterval = 10 * time.Second
maxMpCreationCount = 10
)

140
master/metadata_fsm_op.go
View File

@ -35,72 +35,76 @@ 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
CheckDataReplicasEnable bool
FileStatsEnable bool
ClusterUuid string
ClusterUuidEnable bool
MetaPartitionInodeIdStep uint64
MaxConcurrentLcNodes uint64
DpMaxRepairErrCnt uint64
DpRepairTimeOut uint64
EnableAutoDecommissionDisk bool
DecommissionDiskLimit uint32
VolDeletionDelayTimeHour int64
MarkDiskBrokenThreshold float64
EnableAutoDpMetaRepair bool
DataPartitionTimeoutSec int64
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
CheckDataReplicasEnable bool
FileStatsEnable bool
ClusterUuid string
ClusterUuidEnable bool
MetaPartitionInodeIdStep uint64
MaxConcurrentLcNodes uint64
DpMaxRepairErrCnt uint64
DpRepairTimeOut uint64
EnableAutoDecommissionDisk bool
AutoDecommissionDiskInterval int64
DecommissionDiskLimit uint32
VolDeletionDelayTimeHour int64
MarkDiskBrokenThreshold float64
EnableAutoDpMetaRepair bool
AutoDpMetaRepairParallelCnt uint32
DataPartitionTimeoutSec int64
}
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.cfg.MaxDpCntLimit,
MaxMpCntLimit: c.cfg.MaxMpCntLimit,
FaultDomain: c.FaultDomain,
DiskQosEnable: c.diskQosEnable,
QosLimitUpload: uint64(c.QosAcceptLimit.Limit()),
DirChildrenNumLimit: c.cfg.DirChildrenNumLimit,
DecommissionLimit: c.DecommissionLimit,
CheckDataReplicasEnable: c.checkDataReplicasEnable,
FileStatsEnable: c.fileStatsEnable,
ClusterUuid: c.clusterUuid,
ClusterUuidEnable: c.clusterUuidEnable,
MetaPartitionInodeIdStep: c.cfg.MetaPartitionInodeIdStep,
MaxConcurrentLcNodes: c.cfg.MaxConcurrentLcNodes,
DpMaxRepairErrCnt: c.cfg.DpMaxRepairErrCnt,
DpRepairTimeOut: c.cfg.DpRepairTimeOut,
EnableAutoDecommissionDisk: c.EnableAutoDecommissionDisk,
DecommissionDiskLimit: c.GetDecommissionDiskLimit(),
VolDeletionDelayTimeHour: c.cfg.volDelayDeleteTimeHour,
MarkDiskBrokenThreshold: c.getMarkDiskBrokenThreshold(),
EnableAutoDpMetaRepair: c.getEnableAutoDpMetaRepair(),
DataPartitionTimeoutSec: c.getDataPartitionTimeoutSec(),
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.cfg.MaxDpCntLimit,
MaxMpCntLimit: c.cfg.MaxMpCntLimit,
FaultDomain: c.FaultDomain,
DiskQosEnable: c.diskQosEnable,
QosLimitUpload: uint64(c.QosAcceptLimit.Limit()),
DirChildrenNumLimit: c.cfg.DirChildrenNumLimit,
DecommissionLimit: c.DecommissionLimit,
CheckDataReplicasEnable: c.checkDataReplicasEnable,
FileStatsEnable: c.fileStatsEnable,
ClusterUuid: c.clusterUuid,
ClusterUuidEnable: c.clusterUuidEnable,
MetaPartitionInodeIdStep: c.cfg.MetaPartitionInodeIdStep,
MaxConcurrentLcNodes: c.cfg.MaxConcurrentLcNodes,
DpMaxRepairErrCnt: c.cfg.DpMaxRepairErrCnt,
DpRepairTimeOut: c.cfg.DpRepairTimeOut,
EnableAutoDecommissionDisk: c.EnableAutoDecommissionDisk.Load(),
AutoDecommissionDiskInterval: c.AutoDecommissionInterval.Load(),
DecommissionDiskLimit: c.GetDecommissionDiskLimit(),
VolDeletionDelayTimeHour: c.cfg.volDelayDeleteTimeHour,
MarkDiskBrokenThreshold: c.getMarkDiskBrokenThreshold(),
EnableAutoDpMetaRepair: c.getEnableAutoDpMetaRepair(),
AutoDpMetaRepairParallelCnt: c.AutoDpMetaRepairParallelCnt.Load(),
DataPartitionTimeoutSec: c.getDataPartitionTimeoutSec(),
}
return cv
}
@ -1057,6 +1061,14 @@ func (c *Cluster) updateEnableAutoDpMetaRepair(val bool) {
c.EnableAutoDpMetaRepair.Store(val)
}
func (c *Cluster) updateAutoDecommissionDiskInterval(val int64) {
c.AutoDecommissionInterval.Store(val)
}
func (c *Cluster) updateAutoDpMetaRepairParallelCnt(cnt uint32) {
c.AutoDpMetaRepairParallelCnt.Store(cnt)
}
func (c *Cluster) updateDecommissionDiskLimit(val uint32) {
if val < 1 {
val = 1
@ -1179,7 +1191,8 @@ func (c *Cluster) loadClusterValue() (err error) {
c.clusterUuid = cv.ClusterUuid
c.clusterUuidEnable = cv.ClusterUuidEnable
c.DecommissionLimit = cv.DecommissionLimit
c.EnableAutoDecommissionDisk = cv.EnableAutoDecommissionDisk
c.EnableAutoDecommissionDisk.Store(cv.EnableAutoDecommissionDisk)
c.updateAutoDecommissionDiskInterval(cv.AutoDecommissionDiskInterval)
c.DecommissionLimit = cv.DecommissionLimit
c.cfg.volDelayDeleteTimeHour = cv.VolDeletionDelayTimeHour
@ -1213,6 +1226,7 @@ func (c *Cluster) loadClusterValue() (err error) {
c.checkDataReplicasEnable = cv.CheckDataReplicasEnable
c.updateMarkDiskBrokenThreshold(cv.MarkDiskBrokenThreshold)
c.updateEnableAutoDpMetaRepair(cv.EnableAutoDpMetaRepair)
c.updateAutoDpMetaRepairParallelCnt(cv.AutoDpMetaRepairParallelCnt)
c.updateDataPartitionTimeoutSec(cv.DataPartitionTimeoutSec)
}
return

2
master/vol.go
View File

@ -607,7 +607,7 @@ func (vol *Vol) checkDataPartitions(c *Cluster) (cnt int) {
partitions := vol.dataPartitions.clonePartitions()
checkMetaDp := make(map[uint64]*DataPartition)
checkMetaPool := routinepool.NewRoutinePool(10)
checkMetaPool := routinepool.NewRoutinePool(c.GetAutoDpMetaRepairParallelCnt())
defer checkMetaPool.WaitAndClose()
var checkMetaDpWg sync.WaitGroup

52
proto/model.go
View File

@ -117,31 +117,33 @@ type MetaReplicaInfo struct {
// ClusterView provides the view of a cluster.
type ClusterView struct {
Name string
CreateTime string
LeaderAddr string
DisableAutoAlloc bool
ForbidMpDecommission bool
MetaNodeThreshold float32
Applied uint64
MaxDataPartitionID uint64
MaxMetaNodeID uint64
MaxMetaPartitionID uint64
VolDeletionDelayTimeHour int64
MarkDiskBrokenThreshold float64
EnableAutoDpMetaRepair bool
EnableAutoDecommission bool
DecommissionDiskLimit uint32
DpRepairTimeout time.Duration
DpTimeout time.Duration
DataNodeStatInfo *NodeStatInfo
MetaNodeStatInfo *NodeStatInfo
VolStatInfo []*VolStatInfo
BadPartitionIDs []BadPartitionView
BadMetaPartitionIDs []BadPartitionView
MasterNodes []NodeView
MetaNodes []NodeView
DataNodes []NodeView
Name string
CreateTime string
LeaderAddr string
DisableAutoAlloc bool
ForbidMpDecommission bool
MetaNodeThreshold float32
Applied uint64
MaxDataPartitionID uint64
MaxMetaNodeID uint64
MaxMetaPartitionID uint64
VolDeletionDelayTimeHour int64
MarkDiskBrokenThreshold float64
EnableAutoDpMetaRepair bool
AutoDpMetaRepairParallelCnt int
EnableAutoDecommission bool
AutoDecommissionDiskInterval time.Duration
DecommissionDiskLimit uint32
DpRepairTimeout time.Duration
DpTimeout time.Duration
DataNodeStatInfo *NodeStatInfo
MetaNodeStatInfo *NodeStatInfo
VolStatInfo []*VolStatInfo
BadPartitionIDs []BadPartitionView
BadMetaPartitionIDs []BadPartitionView
MasterNodes []NodeView
MetaNodes []NodeView
DataNodes []NodeView
}
// ClusterNode defines the structure of a cluster node

13
sdk/master/api_admin.go
View File

@ -517,7 +517,9 @@ func (api *AdminAPI) SetMasterVolDeletionDelayTime(volDeletionDelayTimeHour int)
func (api *AdminAPI) SetClusterParas(batchCount, markDeleteRate, deleteWorkerSleepMs, autoRepairRate, loadFactor, maxDpCntLimit, maxMpCntLimit, clientIDKey string,
dataNodesetSelector, metaNodesetSelector, dataNodeSelector, metaNodeSelector string, markDiskBrokenThreshold string,
enableAutoDpMetaRepair string, dpRepairTimeout string, dpTimeout string,
enableAutoDecommissionDisk string, autoDecommissionDiskInterval string,
enableAutoDpMetaRepair string, autoDpMetaRepairParallelCnt string,
dpRepairTimeout string, dpTimeout string,
) (err error) {
request := newRequest(get, proto.AdminSetNodeInfo).Header(api.h)
request.addParam("batchCount", batchCount)
@ -536,9 +538,18 @@ func (api *AdminAPI) SetClusterParas(batchCount, markDeleteRate, deleteWorkerSle
if markDiskBrokenThreshold != "" {
request.addParam("markDiskBrokenThreshold", markDiskBrokenThreshold)
}
if enableAutoDecommissionDisk != "" {
request.addParam("autoDecommissionDisk", enableAutoDecommissionDisk)
}
if autoDecommissionDiskInterval != "" {
request.addParam("autoDecommissionDiskInterval", autoDecommissionDiskInterval)
}
if enableAutoDpMetaRepair != "" {
request.addParam("autoDpMetaRepair", enableAutoDpMetaRepair)
}
if autoDpMetaRepairParallelCnt != "" {
request.addParam("autoDpMetaRepairParallelCnt", autoDpMetaRepairParallelCnt)
}
if dpRepairTimeout != "" {
request.addParam("dpRepairTimeOut", dpRepairTimeout)
}

9
util/atomicutil/bool.go
View File

@ -45,3 +45,12 @@ func (b *Bool) CompareAndSwap(old bool, newVal bool) (swaped bool) {
swaped = atomic.CompareAndSwapUint32(&b.val, oldVal, val)
return
}
func (b *Bool) Swap(new bool) (old bool) {
tmp := uint32(0)
if new {
tmp = 1
}
old = atomic.SwapUint32(&b.val, tmp) == 1
return
}

2
util/atomicutil/bool_test.go
View File

@ -27,4 +27,6 @@ func TestAtomicBool(t *testing.T) {
require.True(t, b.Load())
require.True(t, b.CompareAndSwap(true, false))
require.False(t, b.Load())
require.False(t, b.Swap(true))
require.True(t, b.Load())
}

42
util/atomicutil/float32.go Normal file
View File

@ -0,0 +1,42 @@
// Copyright 2023 The CubeFS Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
// implied. See the License for the specific language governing
// permissions and limitations under the License.
package atomicutil
import (
"math"
"sync/atomic"
)
type Float32 struct {
val uint32
}
func (f *Float32) Load() float32 {
return math.Float32frombits(atomic.LoadUint32(&f.val))
}
func (f *Float32) Store(val float32) {
atomic.StoreUint32(&f.val, math.Float32bits(val))
}
func (f *Float32) CompareAndSwap(old float32, new float32) (swaped bool) {
swaped = atomic.CompareAndSwapUint32(&f.val, math.Float32bits(old), math.Float32bits(new))
return
}
func (f *Float32) Swap(new float32) (old float32) {
old = math.Float32frombits(atomic.SwapUint32(&f.val, math.Float32bits(new)))
return
}

32
util/atomicutil/float32_test.go Normal file
View File

@ -0,0 +1,32 @@
// Copyright 2023 The CubeFS Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
// implied. See the License for the specific language governing
// permissions and limitations under the License.
package atomicutil_test
import (
"testing"
"github.com/cubefs/cubefs/util/atomicutil"
"github.com/stretchr/testify/require"
)
func TestAtomicFloat32(t *testing.T) {
f := atomicutil.Float32{}
testVal := float32(1.0)
f.Store(testVal)
require.Equal(t, testVal, f.Load())
require.True(t, f.CompareAndSwap(testVal, 0))
require.EqualValues(t, 0, f.Swap(testVal))
require.EqualValues(t, testVal, f.Load())
}

10
util/atomicutil/float64.go
View File

@ -30,3 +30,13 @@ func (f *Float64) Load() float64 {
func (f *Float64) Store(val float64) {
atomic.StoreUint64(&f.val, math.Float64bits(val))
}
func (f *Float64) CompareAndSwap(old float64, new float64) (swaped bool) {
swaped = atomic.CompareAndSwapUint64(&f.val, math.Float64bits(old), math.Float64bits(new))
return
}
func (f *Float64) Swap(new float64) (old float64) {
old = math.Float64frombits(atomic.SwapUint64(&f.val, math.Float64bits(new)))
return
}

3
util/atomicutil/float64_test.go
View File

@ -26,4 +26,7 @@ func TestAtomicFloat64(t *testing.T) {
testVal := float64(1.0)
f.Store(testVal)
require.Equal(t, testVal, f.Load())
require.True(t, f.CompareAndSwap(testVal, 0))
require.EqualValues(t, 0, f.Swap(testVal))
require.EqualValues(t, testVal, f.Load())
}

50
util/atomicutil/int32.go Normal file
View File

@ -0,0 +1,50 @@
// Copyright 2024 The CubeFS Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
// implied. See the License for the specific language governing
// permissions and limitations under the License.
package atomicutil
import "sync/atomic"
type Int32 struct {
v int32
}
func (i *Int32) Load() (v int32) {
v = atomic.LoadInt32(&i.v)
return
}
func (i *Int32) Store(v int32) {
atomic.StoreInt32(&i.v, v)
}
func (i *Int32) CompareAndSwap(old int32, new int32) (swaped bool) {
swaped = atomic.CompareAndSwapInt32(&i.v, old, new)
return
}
func (i *Int32) Add(v int32) (new int32) {
new = atomic.AddInt32(&i.v, v)
return
}
func (i *Int32) Sub(v int32) (new int32) {
new = atomic.AddInt32(&i.v, -v)
return
}
func (i *Int32) Swap(v int32) (old int32) {
old = atomic.SwapInt32(&i.v, v)
return
}

33
util/atomicutil/int32_test.go Normal file
View File

@ -0,0 +1,33 @@
// Copyright 2024 The CubeFS Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
// implied. See the License for the specific language governing
// permissions and limitations under the License.
package atomicutil_test
import (
"testing"
"github.com/cubefs/cubefs/util/atomicutil"
"github.com/stretchr/testify/require"
)
func TestInt32(t *testing.T) {
i := atomicutil.Int32{}
i.Store(1)
require.EqualValues(t, 1, i.Load())
require.EqualValues(t, 2, i.Add(1))
require.EqualValues(t, 2, i.Swap(1))
require.True(t, i.CompareAndSwap(1, 2))
require.EqualValues(t, 2, i.Load())
require.EqualValues(t, 1, i.Sub(1))
}

50
util/atomicutil/int64.go Normal file
View File

@ -0,0 +1,50 @@
// Copyright 2024 The CubeFS Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
// implied. See the License for the specific language governing
// permissions and limitations under the License.
package atomicutil
import "sync/atomic"
type Int64 struct {
v int64
}
func (i *Int64) Load() (v int64) {
v = atomic.LoadInt64(&i.v)
return
}
func (i *Int64) Store(v int64) {
atomic.StoreInt64(&i.v, v)
}
func (i *Int64) CompareAndSwap(old int64, new int64) (swaped bool) {
swaped = atomic.CompareAndSwapInt64(&i.v, old, new)
return
}
func (i *Int64) Add(v int64) (new int64) {
new = atomic.AddInt64(&i.v, v)
return
}
func (i *Int64) Sub(v int64) (new int64) {
new = atomic.AddInt64(&i.v, -v)
return
}
func (i *Int64) Swap(v int64) (old int64) {
old = atomic.SwapInt64(&i.v, v)
return
}

33
util/atomicutil/int64_test.go Normal file
View File

@ -0,0 +1,33 @@
// Copyright 2024 The CubeFS Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
// implied. See the License for the specific language governing
// permissions and limitations under the License.
package atomicutil_test
import (
"testing"
"github.com/cubefs/cubefs/util/atomicutil"
"github.com/stretchr/testify/require"
)
func TestInt64(t *testing.T) {
i := atomicutil.Int64{}
i.Store(1)
require.EqualValues(t, 1, i.Load())
require.EqualValues(t, 2, i.Add(1))
require.EqualValues(t, 2, i.Swap(1))
require.True(t, i.CompareAndSwap(1, 2))
require.EqualValues(t, 2, i.Load())
require.EqualValues(t, 1, i.Sub(1))
}

50
util/atomicutil/uint32.go Normal file
View File

@ -0,0 +1,50 @@
// Copyright 2024 The CubeFS Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
// implied. See the License for the specific language governing
// permissions and limitations under the License.
package atomicutil
import "sync/atomic"
type Uint32 struct {
v uint32
}
func (i *Uint32) Load() (v uint32) {
v = atomic.LoadUint32(&i.v)
return
}
func (i *Uint32) Store(v uint32) {
atomic.StoreUint32(&i.v, v)
}
func (i *Uint32) CompareAndSwap(old uint32, new uint32) (swaped bool) {
swaped = atomic.CompareAndSwapUint32(&i.v, old, new)
return
}
func (i *Uint32) Add(v uint32) (new uint32) {
new = atomic.AddUint32(&i.v, v)
return
}
func (i *Uint32) Sub(v int32) (new uint32) {
new = atomic.AddUint32(&i.v, ^uint32(v-1))
return
}
func (i *Uint32) Swap(v uint32) (old uint32) {
old = atomic.SwapUint32(&i.v, v)
return
}

33
util/atomicutil/uint32_test.go Normal file
View File

@ -0,0 +1,33 @@
// Copyright 2024 The CubeFS Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
// implied. See the License for the specific language governing
// permissions and limitations under the License.
package atomicutil_test
import (
"testing"
"github.com/cubefs/cubefs/util/atomicutil"
"github.com/stretchr/testify/require"
)
func TestUint64(t *testing.T) {
i := atomicutil.Uint32{}
i.Store(1)
require.EqualValues(t, 1, i.Load())
require.EqualValues(t, 2, i.Add(1))
require.EqualValues(t, 2, i.Swap(1))
require.True(t, i.CompareAndSwap(1, 2))
require.EqualValues(t, 2, i.Load())
require.EqualValues(t, 1, i.Sub(1))
}

50
util/atomicutil/uint64.go Normal file
View File

@ -0,0 +1,50 @@
// Copyright 2024 The CubeFS Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
// implied. See the License for the specific language governing
// permissions and limitations under the License.
package atomicutil
import "sync/atomic"
type Uint64 struct {
v uint64
}
func (i *Uint64) Load() (v uint64) {
v = atomic.LoadUint64(&i.v)
return
}
func (i *Uint64) Store(v uint64) {
atomic.StoreUint64(&i.v, v)
}
func (i *Uint64) CompareAndSwap(old uint64, new uint64) (swaped bool) {
swaped = atomic.CompareAndSwapUint64(&i.v, old, new)
return
}
func (i *Uint64) Add(v uint64) (new uint64) {
new = atomic.AddUint64(&i.v, v)
return
}
func (i *Uint64) Sub(v int64) (new uint64) {
new = atomic.AddUint64(&i.v, ^uint64(v-1))
return
}
func (i *Uint64) Swap(v uint64) (old uint64) {
old = atomic.SwapUint64(&i.v, v)
return
}

33
util/atomicutil/uint64_test.go Normal file
View File

@ -0,0 +1,33 @@
// Copyright 2024 The CubeFS Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
// implied. See the License for the specific language governing
// permissions and limitations under the License.
package atomicutil_test
import (
"testing"
"github.com/cubefs/cubefs/util/atomicutil"
"github.com/stretchr/testify/require"
)
func TestUint32(t *testing.T) {
i := atomicutil.Uint32{}
i.Store(1)
require.EqualValues(t, 1, i.Load())
require.EqualValues(t, 2, i.Add(1))
require.EqualValues(t, 2, i.Swap(1))
require.True(t, i.CompareAndSwap(1, 2))
require.EqualValues(t, 2, i.Load())
require.EqualValues(t, 1, i.Sub(1))
}