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 auto decommission disk and config

Browse Source 
Signed-off-by: NaturalSelect <huangzhibin1@oppo.com>
This commit is contained in:
NaturalSelect 2024-03-26 14:21:36 +08:00 committed by longerfly
parent 08f46f88e8
commit 610e6bd457
23 changed files with 439 additions and 112 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

50
cli/cmd/cluster.go
View File

@ -20,6 +20,7 @@ import (
"github.com/cubefs/cubefs/proto"
"github.com/cubefs/cubefs/sdk/master"
"github.com/cubefs/cubefs/util/strutil"
"github.com/spf13/cobra"
)
@ -41,6 +42,7 @@ func newClusterCmd(client *master.MasterClient) *cobra.Command {
newClusterSetParasCmd(client),
newClusterDisableMpDecommissionCmd(client),
newClusterSetVolDeletionDelayTimeCmd(client),
newClusterEnableAutoDecommissionDisk(client),
)
return clusterCmd
}
@ -58,6 +60,7 @@ const (
nodeAutoRepairRateKey = "autoRepairRate"
nodeMaxDpCntLimit = "maxDpCntLimit"
cmdForbidMpDecommission = "forbid meta partition decommission"
cmdEnableAutoDecommissionDiskShort = "enable auto decommission disk"
)
func newClusterInfoCmd(client *master.MasterClient) *cobra.Command {
@ -129,11 +132,11 @@ func newClusterFreezeCmd(client *master.MasterClient) *cobra.Command {
ValidArgs: []string{"true", "false"},
Short: cmdClusterFreezeShort,
Args: cobra.MinimumNArgs(1),
Long: `Turn on or off the automatic allocation of the data partitions.
Long: `Turn on or off the automatic allocation of the data partitions.
If 'freeze=false', CubeFS WILL automatically allocate new data partitions for the volume when:
1. the used space is below the max capacity,
2. and the number of r&w data partition is less than 20.
If 'freeze=true', CubeFS WILL NOT automatically allocate new data partitions `,
Run: func(cmd *cobra.Command, args []string) {
var (
@ -235,6 +238,7 @@ func newClusterSetParasCmd(client *master.MasterClient) *cobra.Command {
metaNodesetSelector := ""
dataNodeSelector := ""
metaNodeSelector := ""
markBrokenDiskThreshold := ""
cmd := &cobra.Command{
Use: CliOpSetCluster,
Short: cmdClusterSetClusterInfoShort,
@ -244,10 +248,17 @@ func newClusterSetParasCmd(client *master.MasterClient) *cobra.Command {
errout(err)
}()
if markBrokenDiskThreshold != "" {
val, err := strutil.ParsePercent(markBrokenDiskThreshold)
if err != nil {
return
}
markBrokenDiskThreshold = fmt.Sprintf("%v", val)
}
if err = client.AdminAPI().SetClusterParas(optDelBatchCount, optMarkDeleteRate, optDelWorkerSleepMs,
optAutoRepairRate, optLoadFactor, opMaxDpCntLimit, clientIDKey,
dataNodesetSelector, metaNodesetSelector,
dataNodeSelector, metaNodeSelector); err != nil {
dataNodeSelector, metaNodeSelector, markBrokenDiskThreshold); err != nil {
return
}
stdout("Cluster parameters has been set successfully. \n")
@ -264,6 +275,7 @@ func newClusterSetParasCmd(client *master.MasterClient) *cobra.Command {
cmd.Flags().StringVar(&metaNodesetSelector, CliFlagMetaNodesetSelector, "", "Set the nodeset select policy(metanode) for cluster")
cmd.Flags().StringVar(&dataNodeSelector, CliFlagDataNodeSelector, "", "Set the node select policy(datanode) for cluster")
cmd.Flags().StringVar(&metaNodeSelector, CliFlagMetaNodeSelector, "", "Set the node select policy(metanode) for cluster")
cmd.Flags().StringVar(&markBrokenDiskThreshold, CliFlagMarkDiskBrokenThreshold, "", "Threshold to mark disk as broken")
return cmd
}
@ -273,7 +285,7 @@ func newClusterDisableMpDecommissionCmd(client *master.MasterClient) *cobra.Comm
ValidArgs: []string{"true", "false"},
Short: cmdForbidMpDecommission,
Args: cobra.MinimumNArgs(1),
Long: `Forbid or allow MetaPartition decommission in the cluster.
Long: `Forbid or allow MetaPartition decommission in the cluster.
the forbid flag is false by default when cluster created
If 'forbid=false', MetaPartition decommission/migrate and MetaNode decommission is allowed.
If 'forbid=true', MetaPartition decommission/migrate and MetaNode decommission is forbidden.`,
@ -301,3 +313,33 @@ If 'forbid=true', MetaPartition decommission/migrate and MetaNode decommission i
}
return cmd
}
func newClusterEnableAutoDecommissionDisk(client *master.MasterClient) *cobra.Command {
cmd := &cobra.Command{
Use: CliOpEnableAutoDecommission + " [STATUS]",
ValidArgs: []string{"true", "false"},
Short: cmdEnableAutoDecommissionDiskShort,
Args: cobra.MinimumNArgs(1),
Run: func(cmd *cobra.Command, args []string) {
var (
err error
enable bool
)
defer func() {
errout(err)
}()
if enable, err = strconv.ParseBool(args[0]); err != nil {
return
}
if err = client.AdminAPI().SetAutoDecommissionDisk(enable); err != nil {
return
}
if enable {
stdout("Enable auto decommission successful!\n")
} else {
stdout("Disable auto decommission successful!\n")
}
},
}
return cmd
}

127
cli/cmd/const.go
View File

@ -44,6 +44,8 @@ const (
CliOpGetDiscard = "get-discard"
CliOpSetDiscard = "set-discard"
CliOpForbidMpDecommission = "forbid-mp-decommission"
CliOpQueryDecommission = "query-decommission"
CliOpEnableAutoDecommission = "enable-auto-decommission"
// Shorthand format of operation name
CliOpDecommissionShortHand = "dec"
@ -59,68 +61,69 @@ 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"
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"
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"
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"
// CliFlagSetDataPartitionCount = "count" use dp-count instead

29
cli/cmd/datanode.go
View File

@ -38,14 +38,16 @@ func newDataNodeCmd(client *master.MasterClient) *cobra.Command {
newDataNodeInfoCmd(client),
newDataNodeDecommissionCmd(client),
newDataNodeMigrateCmd(client),
newDataNodeQueryDecommissionedDisk(client),
)
return cmd
}
const (
cmdDataNodeListShort = "List information of data nodes"
cmdDataNodeInfoShort = "Show information of a data node"
cmdDataNodeDecommissionInfoShort = "decommission partitions in a data node to others"
cmdDataNodeListShort = "List information of data nodes"
cmdDataNodeInfoShort = "Show information of a data node"
cmdDataNodeDecommissionInfoShort = "decommission partitions in a data node to others"
cmdDataNodeQueryDecommissionedDisksShort = "query datanode decommissioned disks"
)
func newDataNodeListCmd(client *master.MasterClient) *cobra.Command {
@ -171,3 +173,24 @@ func newDataNodeMigrateCmd(client *master.MasterClient) *cobra.Command {
cmd.Flags().StringVar(&clientIDKey, CliFlagClientIDKey, client.ClientIDKey(), CliUsageClientIDKey)
return cmd
}
func newDataNodeQueryDecommissionedDisk(client *master.MasterClient) *cobra.Command {
cmd := &cobra.Command{
Use: CliOpQueryDecommission + " [{HOST}:{PORT}]",
Short: cmdDataNodeQueryDecommissionedDisksShort,
Args: cobra.MinimumNArgs(1),
RunE: func(cmd *cobra.Command, args []string) error {
disks, err := client.NodeAPI().QueryDecommissionedDisks(args[0])
if err != nil {
stdout("%v", err)
return err
}
stdoutln("[Decommissioned disks]")
for _, disk := range disks.Disks {
stdout("%v", disk)
}
return nil
},
}
return cmd
}

2
cli/cmd/fmt.go
View File

@ -70,6 +70,8 @@ func formatClusterView(cv *proto.ClusterView, cn *proto.ClusterNodeInfo, cp *pro
sb.WriteString(fmt.Sprintf(" EbsAddr : %v\n", cp.EbsAddr))
sb.WriteString(fmt.Sprintf(" LoadFactor : %v\n", cn.LoadFactor))
sb.WriteString(fmt.Sprintf(" volDeletionDelayTime : %v h\n", cv.VolDeletionDelayTimeHour))
sb.WriteString(fmt.Sprintf(" EnableAutoDecommission: %v\n", cv.EnableAutoDecommission))
sb.WriteString(fmt.Sprintf(" MarkDiskBrokenThreshold : %v%%\n", cv.MarkDiskBrokenThreshold*100))
return sb.String()
}

2
datanode/disk.go
View File

@ -448,7 +448,6 @@ func (d *Disk) triggerDiskError(rwFlag uint8, dpId uint64) {
}
d.AddDiskErrPartition(dpId)
diskErrCnt := d.getTotalErrCnt()
diskErrPartitionCnt := d.GetDiskErrPartitionCount()
if diskErrPartitionCnt >= d.dataNode.diskUnavailablePartitionErrorCount {
@ -725,6 +724,7 @@ func (d *Disk) getSelectWeight() float64 {
func (d *Disk) AddDiskErrPartition(dpId uint64) {
if _, ok := d.DiskErrPartitionSet[dpId]; !ok {
log.LogWarnf("[AddDiskErrPartition] mark dp(%v) as broken", dpId)
d.DiskErrPartitionSet[dpId] = struct{}{}
}
}

1
datanode/partition.go
View File

@ -742,6 +742,7 @@ func (dp *DataPartition) checkIsDiskError(err error, rwFlag uint8) {
return
}
log.LogWarnf("[checkIsDiskError] disk(%v) dp(%v) meet io error", dp.Path(), dp.partitionID)
dp.stopRaft()
dp.incDiskErrCnt()
dp.disk.triggerDiskError(rwFlag, dp.partitionID)

2
datanode/server.go
View File

@ -655,6 +655,8 @@ func (s *DataNode) registerHandler() {
// http.HandleFunc("/detachDataPartition", s.detachDataPartition)
// http.HandleFunc("/loadDataPartition", s.loadDataPartition)
http.HandleFunc("/releaseDiskExtentReadLimitToken", s.releaseDiskExtentReadLimitToken)
http.HandleFunc("/markDataPartitionBroken", s.markDataPartitionBroken)
http.HandleFunc("/markDiskBroken", s.markDiskBroken)
}
func (s *DataNode) startTCPService() (err error) {

56
datanode/server_handler.go
View File

@ -22,6 +22,7 @@ import (
"path"
"strconv"
"sync/atomic"
"syscall"
"github.com/cubefs/cubefs/cmd/common"
"github.com/cubefs/cubefs/depends/tiglabs/raft"
@ -656,3 +657,58 @@ func (s *DataNode) loadDataPartition(w http.ResponseWriter, r *http.Request) {
s.buildSuccessResp(w, "success")
}
}
// NOTE: use for test
func (s *DataNode) markDataPartitionBroken(w http.ResponseWriter, r *http.Request) {
const (
paramID = "id"
)
if err := r.ParseForm(); err != nil {
err = fmt.Errorf("parse form fail: %v", err)
s.buildFailureResp(w, http.StatusBadRequest, err.Error())
return
}
partitionID, err := strconv.ParseUint(r.FormValue(paramID), 10, 64)
if err != nil {
err = fmt.Errorf("parse param %v fail: %v", paramID, err)
s.buildFailureResp(w, http.StatusBadRequest, err.Error())
return
}
partition := s.space.Partition(partitionID)
if partition == nil {
s.buildFailureResp(w, http.StatusBadRequest, "partition not found")
return
}
partition.checkIsDiskError(syscall.EIO, WriteFlag|ReadFlag)
s.buildSuccessResp(w, "success")
}
func (s *DataNode) markDiskBroken(w http.ResponseWriter, r *http.Request) {
const (
paramDisk = "disk"
)
if err := r.ParseForm(); err != nil {
err = fmt.Errorf("parse form fail: %v", err)
s.buildFailureResp(w, http.StatusBadRequest, err.Error())
return
}
diskPath := r.FormValue(paramDisk)
// store disk path and root of dp
disk, err := s.space.GetDisk(diskPath)
if err != nil {
log.LogErrorf("[markDiskBroken] disk(%v) is not found err(%v).", diskPath, err)
s.buildFailureResp(w, http.StatusBadRequest, fmt.Sprintf("disk %v is not found", diskPath))
return
}
dps := disk.DataPartitionList()
for _, dpId := range dps {
partition := s.space.Partition(dpId)
if partition == nil {
log.LogErrorf("[markDiskBroken] dp(%v) not found", dpId)
s.buildFailureResp(w, http.StatusBadRequest, "partition not found")
return
}
partition.checkIsDiskError(syscall.EIO, WriteFlag|ReadFlag)
}
s.buildSuccessResp(w, "success")
}

13
datanode/space_manager.go
View File

@ -484,8 +484,19 @@ func (s *DataNode) buildHeartBeatResponse(response *proto.DataNodeHeartbeatRespo
disks := space.GetDisks()
for _, d := range disks {
if d.Status == proto.Unavailable {
brokenDpsCnt := d.GetDiskErrPartitionCount()
brokenDps := d.GetDiskErrPartitionList()
log.LogInfof("[buildHeartBeatResponse] disk(%v) status(%v) broken dp len(%v)", d.Path, d.Status, brokenDpsCnt)
if d.Status == proto.Unavailable || brokenDpsCnt != 0 {
response.BadDisks = append(response.BadDisks, d.Path)
bds := proto.BadDiskStat{
DiskPath: d.Path,
TotalPartitionCnt: d.PartitionCount(),
DiskErrPartitionList: brokenDps,
}
response.BadDiskStats = append(response.BadDiskStats, bds)
log.LogErrorf("[buildHeartBeatResponse] disk(%v) total(%v) broken dp len(%v) %v",
d.Path, bds.TotalPartitionCnt, brokenDpsCnt, brokenDps)
}
bds := proto.DiskStat{

48
master/api_service.go
View File

@ -788,6 +788,8 @@ func (m *Server) getCluster(w http.ResponseWriter, r *http.Request) {
MaxMetaNodeID: m.cluster.idAlloc.commonID,
MaxMetaPartitionID: m.cluster.idAlloc.metaPartitionID,
VolDeletionDelayTimeHour: m.cluster.cfg.volDelayDeleteTimeHour,
MarkDiskBrokenThreshold: m.cluster.getMarkDiskBrokenThreshold(),
EnableAutoDecommission: m.cluster.AutoDecommissionDiskIsEnabled(),
MasterNodes: make([]proto.NodeView, 0),
MetaNodes: make([]proto.NodeView, 0),
DataNodes: make([]proto.NodeView, 0),
@ -1875,13 +1877,6 @@ func (m *Server) decommissionDataPartition(w http.ResponseWriter, r *http.Reques
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: err.Error()})
return
}
replica, err := dp.getReplica(addr)
if err != nil {
rstMsg = fmt.Sprintf(" dataPartitionID :%v not find replica for addr %v",
partitionID, addr)
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: rstMsg})
return
}
node, err := c.dataNode(addr)
if err != nil {
rstMsg = fmt.Sprintf(" dataPartitionID :%v not find datanode for addr %v",
@ -1889,28 +1884,12 @@ func (m *Server) decommissionDataPartition(w http.ResponseWriter, r *http.Reques
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: rstMsg})
return
}
zone, err := c.t.getZone(node.ZoneName)
err = m.cluster.markDecommissionDataPartition(dp, node, raftForce)
if err != nil {
rstMsg = fmt.Sprintf(" dataPartitionID :%v not find zone for addr %v",
partitionID, addr)
rstMsg = err.Error()
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: rstMsg})
return
}
ns, err := zone.getNodeSet(node.NodeSetID)
if err != nil {
rstMsg = fmt.Sprintf(" dataPartitionID :%v not find nodeset for addr %v",
partitionID, addr)
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: rstMsg})
return
}
if !dp.MarkDecommissionStatus(addr, "", replica.DiskPath, raftForce, 0, c) {
rstMsg = fmt.Sprintf(" dataPartitionID :%v mark decommission failed",
partitionID)
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: rstMsg})
return
}
c.syncUpdateDataPartition(dp)
ns.AddToDecommissionDataPartitionList(dp, c)
rstMsg = fmt.Sprintf(proto.AdminDecommissionDataPartition+" dataPartitionID :%v on node:%v successfully", partitionID, addr)
sendOkReply(w, r, newSuccessHTTPReply(rstMsg))
}
@ -3073,6 +3052,15 @@ func (m *Server) setNodeInfoHandler(w http.ResponseWriter, r *http.Request) {
}
}
if val, ok := params[markDiskBrokenThresholdKey]; ok {
if markDiskBrokenThreshold, ok := val.(float64); ok {
if err = m.cluster.setMarkDiskBrokenThreshold(markDiskBrokenThreshold); err != nil {
sendErrReply(w, r, newErrHTTPReply(err))
return
}
}
}
dataNodesetSelector := extractDataNodesetSelector(r)
metaNodesetSelector := extractMetaNodesetSelector(r)
dataNodeSelector := extractDataNodeSelector(r)
@ -5644,9 +5632,9 @@ func (m *Server) queryDisableDisk(w http.ResponseWriter, r *http.Request) {
nodeAddr string
err error
)
metric := exporter.NewTPCnt(apiToMetricsName(proto.RecommissionDisk))
metric := exporter.NewTPCnt(apiToMetricsName(proto.QueryDisableDisk))
defer func() {
doStatAndMetric(proto.RecommissionDisk, metric, err, nil)
doStatAndMetric(proto.QueryDisableDisk, metric, err, nil)
}()
if nodeAddr, err = parseAndExtractNodeAddr(r); err != nil {
@ -5661,11 +5649,15 @@ func (m *Server) queryDisableDisk(w http.ResponseWriter, r *http.Request) {
disks := node.getDecommissionedDisks()
disksInfo := &proto.DecommissionedDisks{
Node: nodeAddr,
Disks: disks,
}
rstMsg = fmt.Sprintf("datanode[%v] disable disk[%v]",
nodeAddr, disks)
Warn(m.clusterName, rstMsg)
sendOkReply(w, r, newSuccessHTTPReply(rstMsg))
sendOkReply(w, r, newSuccessHTTPReply(disksInfo))
}
func parseReqToDecoDataNodeProgress(r *http.Request) (nodeAddr string, err error) {

12
master/api_service_test.go
View File

@ -1521,3 +1521,15 @@ func TestSetVolumeDpRepairBlockSize(t *testing.T) {
require.EqualValues(t, proto.DefaultDpRepairBlockSize, vol.dpRepairBlockSize)
require.True(t, checkVolDpRepairBlockSize(name, proto.DefaultDpRepairBlockSize))
}
func TestSetMarkDiskBrokenThreshold(t *testing.T) {
reqUrl := fmt.Sprintf("%v%v", hostAddr, proto.AdminSetClusterInfo)
setVal := 0.5
oldVal := server.cluster.getMarkDiskBrokenThreshold()
setUrl := fmt.Sprintf("%v?%v=%v&dirSizeLimit=0", reqUrl, markDiskBrokenThresholdKey, setVal)
unsetUrl := fmt.Sprintf("%v?%v=%v&dirSizeLimit=0", reqUrl, markDiskBrokenThresholdKey, oldVal)
process(setUrl, t)
require.EqualValues(t, setVal, server.cluster.getMarkDiskBrokenThreshold())
process(unsetUrl, t)
require.EqualValues(t, oldVal, server.cluster.getMarkDiskBrokenThreshold())
}

133
master/cluster.go
View File

@ -34,6 +34,7 @@ import (
authSDK "github.com/cubefs/cubefs/sdk/auth"
masterSDK "github.com/cubefs/cubefs/sdk/master"
"github.com/cubefs/cubefs/util"
"github.com/cubefs/cubefs/util/atomicutil"
"github.com/cubefs/cubefs/util/compressor"
"github.com/cubefs/cubefs/util/config"
"github.com/cubefs/cubefs/util/errors"
@ -102,6 +103,7 @@ type Cluster struct {
snapshotMgr *snapshotDelManager
DecommissionDiskFactor float64
S3ApiQosQuota *sync.Map // (api,uid,limtType) -> limitQuota
MarkDiskBrokenThreshold atomicutil.Float64
}
type delayDeleteVolInfo struct {
@ -361,6 +363,7 @@ func newCluster(name string, leaderInfo *LeaderInfo, fsm *MetadataFsm, partition
c.snapshotMgr = newSnapshotManager()
c.snapshotMgr.cluster = c
c.S3ApiQosQuota = new(sync.Map)
c.MarkDiskBrokenThreshold.Store(defaultMarkDiskBrokenThreshold)
return
}
@ -3830,6 +3833,29 @@ func (c *Cluster) setMaxDpCntLimit(val uint64) (err error) {
return
}
func (c *Cluster) setMarkDiskBrokenThreshold(val float64) (err error) {
if val <= 0 || val > 1 {
val = defaultMarkDiskBrokenThreshold
}
oldVal := c.MarkDiskBrokenThreshold.Load()
c.MarkDiskBrokenThreshold.Store(val)
if err = c.syncPutCluster(); err != nil {
log.LogErrorf("[setMarkDiskBrokenThreshold] failed to set mark disk broken threshold, err(%v)", err)
c.MarkDiskBrokenThreshold.Store(oldVal)
err = proto.ErrPersistenceByRaft
return
}
return
}
func (c *Cluster) getMarkDiskBrokenThreshold() (v float64) {
v = c.MarkDiskBrokenThreshold.Load()
if v <= 0 || v > 1 {
v = defaultMarkDiskBrokenThreshold
}
return
}
func (c *Cluster) setClusterCreateTime(createTime int64) (err error) {
oldVal := c.CreateTime
c.CreateTime = createTime
@ -4235,7 +4261,8 @@ func (c *Cluster) checkDecommissionDisk() {
func (c *Cluster) scheduleToBadDisk() {
go func() {
for {
if c.partition.IsRaftLeader() {
if c.partition.IsRaftLeader() && c.AutoDecommissionDiskIsEnabled() {
log.LogDebugf("[scheduleToBadDisk] check bad disk")
c.checkBadDisk()
}
time.Sleep(10 * time.Second)
@ -4243,17 +4270,75 @@ func (c *Cluster) scheduleToBadDisk() {
}()
}
func (c *Cluster) canAutoDecommissionDisk(addr string, diskPath string) (yes bool, status uint32) {
key := fmt.Sprintf("%s_%s", addr, diskPath)
if value, ok := c.DecommissionDisks.Load(key); ok {
d := value.(*DecommissionDisk)
status = d.GetDecommissionStatus()
yes = status != markDecommission && status != DecommissionRunning && status != DecommissionPause
return
}
yes = true
return
}
func (c *Cluster) handleDataNodeBadDisk(dataNode *DataNode) {
dataNode.RLock()
defer dataNode.RUnlock()
log.LogDebugf("[handleDataNodeBadDisk] datanode(%v) bad disk cnt(%v)", dataNode.Addr, len(dataNode.BadDiskStats))
for _, disk := range dataNode.BadDiskStats {
log.LogDebugf("[handleDataNodeBadDisk] handle data node(%v) bad disk(%v), bad dp(%v) / total dp(%v)", dataNode.Addr, disk.DiskPath, len(disk.DiskErrPartitionList), disk.TotalPartitionCnt)
ratio := float64(len(disk.DiskErrPartitionList)) / float64(disk.TotalPartitionCnt)
if ratio <= c.getMarkDiskBrokenThreshold() {
// NOTE: decommission broken dps
for _, dpId := range disk.DiskErrPartitionList {
log.LogDebugf("[handleDataNodeBadDisk] try to decommission dp(%v)", dpId)
dp, err := c.getDataPartitionByID(dpId)
if err != nil {
log.LogErrorf("[handleDataNodeBadDisk] failed to get data node(%v) dp(%v), err(%v)", dataNode.Addr, dpId, err)
continue
}
// NOTE: replica not found, maybe decommissioned
if dp.findReplica(dataNode.Addr) == -1 {
log.LogInfof("[handleDataNodeBadDisk] data node(%v) not found in dp(%v) maybe decommissioned?", dataNode.Addr, dpId)
continue
}
// NOTE: check decommission status
if dp.canMarkDecommission() {
err = c.markDecommissionDataPartition(dp, dataNode, false)
if err != nil {
log.LogErrorf("[handleDataNodeBadDisk] failed to decommssion dp(%v) on data node(%v) disk(%v), err(%v)", dataNode.Addr, disk.DiskPath, dp.PartitionID, err)
continue
}
} else {
log.LogInfof("[handleDataNodeBadDisk] dp(%v) on data node(%v) disk(%v) cannot decommission now, skip, status(%v)", dp.PartitionID, dataNode.Addr, disk.DiskPath, dp.GetDecommissionStatus())
}
}
} else if len(disk.DiskErrPartitionList) != 0 {
log.LogDebugf("[handleDataNodeBadDisk] try to decommission disk(%v)", disk.DiskPath)
// NOTE: decommission all dps and disable disk
ok, status := c.canAutoDecommissionDisk(dataNode.Addr, disk.DiskPath)
if !ok {
log.LogInfof("[handleDataNodeBadDisk] cannnot auto decommission dp on data node(%v) disk(%v) status(%v), skip", dataNode.Addr, disk.DiskPath, status)
continue
}
log.LogDebugf("[handleDataNodeBadDisk] decommission disk(%v)", disk.DiskPath)
err := c.migrateDisk(dataNode.Addr, disk.DiskPath, "", false, 0, true, AutoDecommission)
if err != nil {
log.LogErrorf("[handleDataNodeBadDisk] failed to decommission broken disk(%v) on data node(%v), err(%v)", disk.DiskPath, dataNode.Addr, err)
}
}
}
}
func (c *Cluster) checkBadDisk() {
c.dataNodes.Range(func(addr, node interface{}) bool {
//TODO add to auto decommission disk
//dataNode, ok := node.(*DataNode)
//if !ok {
// return true
//}
//for _, badDisk := range dataNode.BadDisks {
//
//}
dataNode, ok := node.(*DataNode)
if !ok {
return true
}
c.handleDataNodeBadDisk(dataNode)
return true
})
}
@ -4291,6 +4376,7 @@ func (c *Cluster) TryDecommissionDisk(disk *DecommissionDisk) {
node.Addr, disk.DiskPath, lastBadPartitionIds)
badPartitions = mergeDataPartitionArr(badPartitions, lastBadPartitions)
log.LogDebugf("[TryDecommissionDisk] data node(%v) disk(%v) bad dps(%v)", node.Addr, disk.DiskPath, len(badPartitions))
if len(badPartitions) == 0 {
log.LogInfof("action[TryDecommissionDisk] receive decommissionDisk node[%v] "+
"no any partitions on disk[%v],offline successfully",
@ -4799,3 +4885,30 @@ func (c *Cluster) GetDecommissionDataPartitionRecoverTimeOut() time.Duration {
return time.Second * time.Duration(c.cfg.DpRepairTimeOut)
}
}
func (c *Cluster) markDecommissionDataPartition(dp *DataPartition, src *DataNode, raftForce bool) (err error) {
addr := src.Addr
replica, err := dp.getReplica(addr)
if err != nil {
err = errors.NewErrorf(" dataPartitionID :%v not find replica for addr %v", dp.PartitionID, addr)
return
}
zone, err := c.t.getZone(src.ZoneName)
if err != nil {
err = errors.NewErrorf(" dataPartitionID :%v not find zone for addr %v", dp.PartitionID, addr)
return
}
ns, err := zone.getNodeSet(src.NodeSetID)
if err != nil {
err = errors.NewErrorf(" dataPartitionID :%v not find nodeset for addr %v", dp.PartitionID, addr)
return
}
if !dp.MarkDecommissionStatus(addr, "", replica.DiskPath, raftForce, 0, c) {
err = errors.NewErrorf(" dataPartitionID :%v mark decommission failed", dp.PartitionID)
return
}
// TODO: handle error
c.syncUpdateDataPartition(dp)
ns.AddToDecommissionDataPartitionList(dp, c)
return
}

1
master/cluster_task.go
View File

@ -1015,6 +1015,7 @@ func (c *Cluster) handleDataNodeHeartbeatResp(nodeAddr string, resp *proto.DataN
dataNode.SetIoUtils(resp.IoUtils)
dataNode.updateNodeMetric(resp)
if err = c.t.putDataNode(dataNode); err != nil {
log.LogErrorf("action[handleDataNodeHeartbeatResp] dataNode[%v],zone[%v],node set[%v], err[%v]", dataNode.Addr, dataNode.ZoneName, dataNode.NodeSetID, err)
}

2
master/const.go
View File

@ -135,6 +135,7 @@ const (
DecommissionType = "decommissionType"
decommissionDiskFactor = "decommissionDiskFactor"
dpRepairBlockSizeKey = "dpRepairBlockSize"
markDiskBrokenThresholdKey = "markDiskBrokenThreshold"
)
const (
@ -206,6 +207,7 @@ const (
defaultClientReqPeriodSeconds = 1
defaultMaxQuotaNumPerVol = 100
defaultVolDelayDeleteTimeHour = 48
defaultMarkDiskBrokenThreshold = 0.5
)
const (

5
master/data_node.go
View File

@ -48,8 +48,9 @@ type DataNode struct {
TotalPartitionSize uint64
NodeSetID uint64
PersistenceDataPartitions []uint64
BadDisks []string // Keep this old field for compatibility
DiskStats []proto.DiskStat // key: disk path
BadDisks []string // Keep this old field for compatibility
BadDiskStats []proto.BadDiskStat // key: disk path
DiskStats []proto.DiskStat // key: disk path
DecommissionedDisks sync.Map
ToBeOffline bool
RdOnly bool

1
master/data_partition.go
View File

@ -1072,6 +1072,7 @@ func (partition *DataPartition) MarkDecommissionStatus(srcAddr, dstAddr, srcDisk
}
func (partition *DataPartition) SetDecommissionStatus(status uint32) {
log.LogDebugf("[SetDecommissionStatus] set dp(%v) decommission status to status(%v)", partition.PartitionID, status)
atomic.StoreUint32(&partition.DecommissionStatus, status)
}

6
master/disk_manager.go
View File

@ -48,6 +48,7 @@ func (c *Cluster) checkDiskRecoveryProgress() {
c.badPartitionMutex.Lock()
defer c.badPartitionMutex.Unlock()
log.LogDebugf("[checkDiskRecoveryProgress] check disk recovery progress")
c.BadDataPartitionIds.Range(func(key, value interface{}) bool {
badDataPartitionIds := value.([]uint64)
newBadDpIds := make([]uint64, 0)
@ -59,6 +60,7 @@ func (c *Cluster) checkDiskRecoveryProgress() {
}
// do not update status if paused
if partition.IsDecommissionPaused() {
log.LogInfof("[checkDiskRecoveryProgress] dp(%v) decommission pause", partitionID)
continue
}
_, err = c.getVol(partition.VolName)
@ -67,6 +69,7 @@ func (c *Cluster) checkDiskRecoveryProgress() {
c.Name, partitionID, partition.VolName))
continue
}
log.LogDebugf("[checkDiskRecoveryProgress] dp(%v) decommission status(%v)", partitionID, partition.GetDecommissionStatus())
log.LogInfof("action[checkDiskRecoveryProgress] dp %v isSpec %v replicas %v conf replicas num %v",
partition.PartitionID, partition.isSpecialReplicaCnt(), len(partition.Replicas), int(partition.ReplicaNum))
if len(partition.Replicas) == 0 {
@ -99,6 +102,7 @@ func (c *Cluster) checkDiskRecoveryProgress() {
continue
}
if newReplica.isRepairing() {
log.LogInfof("[checkDiskRecoveryProgress] dp(%v) new replica(%v) report time(%v) is repairing", partition.PartitionID, newReplica.Addr, time.Unix(newReplica.ReportTime, 0))
if !partition.isSpecialReplicaCnt() {
masterNode, _ := partition.getReplica(partition.Hosts[0])
duration := time.Unix(masterNode.ReportTime, 0).Sub(time.Unix(newReplica.ReportTime, 0))
@ -136,6 +140,7 @@ func (c *Cluster) checkDiskRecoveryProgress() {
newBadDpIds = append(newBadDpIds, partitionID)
} else {
if partition.isSpecialReplicaCnt() {
log.LogWarnf("[checkDiskRecoveryProgress] dp(%v) special replica cnt, skip", partition.PartitionID)
continue // change dp decommission status in decommission function
}
// do not add to BadDataPartitionIds
@ -307,6 +312,7 @@ func (dd *DecommissionDisk) updateDecommissionStatus(c *Cluster, debug bool) (ui
stopNum++
stopPartitionIds = append(stopPartitionIds, dp.PartitionID)
}
log.LogDebugf("[updateDecommissionStatus] dp(%v) decommission status(%v)", dp.PartitionID, dp.GetDecommissionStatus())
partitionIds = append(partitionIds, dp.PartitionID)
}
progress = float64(totalNum-len(partitions)) / float64(totalNum)

10
master/metadata_fsm_op.go
View File

@ -62,6 +62,7 @@ type clusterValue struct {
EnableAutoDecommissionDisk bool
DecommissionDiskFactor float64
VolDeletionDelayTimeHour int64
MarkDiskBrokenThreshold float64
}
func newClusterValue(c *Cluster) (cv *clusterValue) {
@ -93,6 +94,7 @@ func newClusterValue(c *Cluster) (cv *clusterValue) {
EnableAutoDecommissionDisk: c.EnableAutoDecommissionDisk,
DecommissionDiskFactor: c.DecommissionDiskFactor,
VolDeletionDelayTimeHour: c.cfg.volDelayDeleteTimeHour,
MarkDiskBrokenThreshold: c.getMarkDiskBrokenThreshold(),
}
return cv
}
@ -1010,6 +1012,13 @@ func (c *Cluster) updateInodeIdStep(val uint64) {
atomic.StoreUint64(&c.cfg.MetaPartitionInodeIdStep, val)
}
func (c *Cluster) updateMarkDiskBrokenThreshold(val float64) {
if val <= 0 || val > 1 {
val = defaultMarkDiskBrokenThreshold
}
c.MarkDiskBrokenThreshold.Store(val)
}
func (c *Cluster) loadZoneValue() (err error) {
var ok bool
result, err := c.fsm.store.SeekForPrefix([]byte(zonePrefix))
@ -1151,6 +1160,7 @@ func (c *Cluster) loadClusterValue() (err error) {
log.LogInfof("action[loadClusterValue], metaNodeThreshold[%v]", cv.Threshold)
c.checkDataReplicasEnable = cv.CheckDataReplicasEnable
c.updateMarkDiskBrokenThreshold(cv.MarkDiskBrokenThreshold)
}
return
}

7
proto/admin_proto.go
View File

@ -718,6 +718,12 @@ type DataNodeQosResponse struct {
Result string
}
type BadDiskStat struct {
DiskPath string
TotalPartitionCnt int
DiskErrPartitionList []uint64
}
type DiskStat struct {
Status int
DiskPath string
@ -747,6 +753,7 @@ type DataNodeHeartbeatResponse struct {
Status uint8
Result string
BadDisks []string // Keep this old field for compatibility
BadDiskStats []BadDiskStat // key: disk path
DiskStats []DiskStat // key: disk path
CpuUtil float64 `json:"cpuUtil"`
IoUtils map[string]float64 `json:"ioUtil"`

7
proto/model.go
View File

@ -122,6 +122,8 @@ type ClusterView struct {
MaxMetaNodeID uint64
MaxMetaPartitionID uint64
VolDeletionDelayTimeHour int64
MarkDiskBrokenThreshold float64
EnableAutoDecommission bool
DataNodeStatInfo *NodeStatInfo
MetaNodeStatInfo *NodeStatInfo
VolStatInfo []*VolStatInfo
@ -466,3 +468,8 @@ type DecommissionDataPartitionInfo struct {
ErrorMessage string
NeedRollbackTimes uint32
}
type DecommissionedDisks struct {
Node string
Disks []string
}

12
sdk/master/api_admin.go
View File

@ -498,7 +498,7 @@ func (api *AdminAPI) SetMasterVolDeletionDelayTime(volDeletionDelayTimeHour int)
}
func (api *AdminAPI) SetClusterParas(batchCount, markDeleteRate, deleteWorkerSleepMs, autoRepairRate, loadFactor, maxDpCntLimit, clientIDKey string,
dataNodesetSelector, metaNodesetSelector, dataNodeSelector, metaNodeSelector string,
dataNodesetSelector, metaNodesetSelector, dataNodeSelector, metaNodeSelector string, markDiskBrokenThreshold string,
) (err error) {
request := newRequest(get, proto.AdminSetNodeInfo).Header(api.h)
request.addParam("batchCount", batchCount)
@ -513,6 +513,9 @@ func (api *AdminAPI) SetClusterParas(batchCount, markDeleteRate, deleteWorkerSle
request.addParam("metaNodesetSelector", metaNodesetSelector)
request.addParam("dataNodeSelector", dataNodeSelector)
request.addParam("metaNodeSelector", metaNodeSelector)
if markDiskBrokenThreshold != "" {
request.addParam("markDiskBrokenThreshold", markDiskBrokenThreshold)
}
_, err = api.mc.serveRequest(request)
return
}
@ -727,3 +730,10 @@ func (api *AdminAPI) DelBucketLifecycle(volume string) (err error) {
func (api *AdminAPI) GetS3QoSInfo() (data []byte, err error) {
return api.mc.serveRequest(newRequest(get, proto.S3QoSGet).Header(api.h))
}
func (api *AdminAPI) SetAutoDecommissionDisk(enable bool) (err error) {
request := newRequest(post, proto.AdminEnableAutoDecommissionDisk)
request.addParam("enable", strconv.FormatBool(enable))
_, err = api.mc.serveRequest(request)
return
}

6
sdk/master/api_node.go
View File

@ -166,3 +166,9 @@ func (api *NodeAPI) AddLcNode(serverAddr string) (id uint64, err error) {
func (api *NodeAPI) ResponseLcNodeTask(task *proto.AdminTask) (err error) {
return api.mc.request(newRequest(post, proto.GetLcNodeTaskResponse).Header(api.h).Body(task))
}
func (api *NodeAPI) QueryDecommissionedDisks(addr string) (disks *proto.DecommissionedDisks, err error) {
disks = &proto.DecommissionedDisks{}
err = api.mc.requestWith(disks, newRequest(get, proto.QueryDisableDisk).Header(api.h).addParam("addr", addr))
return
}

19
util/strutil/fmt.go
View File

@ -93,3 +93,22 @@ func FormatSize(size uint64) (sizeStr string) {
sizeStr = fmt.Sprintf("%v%v", size, pair.Suffix)
return
}
func ParsePercent(valStr string) (val float64, err error) {
base := float64(1)
if strings.HasPrefix(valStr, "%") {
base = 1
valStr = strings.TrimSuffix(valStr, "%")
}
val, err = strconv.ParseFloat(valStr, 64)
if err != nil {
return
}
val *= base
return
}
func FormatPercent(val float64) (valStr string) {
valStr = fmt.Sprintf("%v%%", val*100)
return
}