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): auto delete backup directories

Browse Source 
Signed-off-by: chihe <chihe@oppo.com>
This commit is contained in:
chihe 2024-07-26 10:50:53 +08:00 committed by AmazingChi
parent 7ae6457821
commit 2fb56e7282
19 changed files with 183 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
cli/cmd/cluster.go
View File

@ -260,6 +260,7 @@ func newClusterSetParasCmd(client *master.MasterClient) *cobra.Command {
opMaxMpCntLimit := ""
dpRepairTimeout := ""
dpTimeout := ""
dpBackupTimeout := ""
cmd := &cobra.Command{
Use: CliOpSetCluster,
Short: cmdClusterSetClusterInfoShort,
@ -333,13 +334,26 @@ func newClusterSetParasCmd(client *master.MasterClient) *cobra.Command {
dpTimeout = strconv.FormatInt(int64(heartbeatTimeout.Seconds()), 10)
}
if dpBackupTimeout != "" {
var backupTimeout time.Duration
backupTimeout, err = time.ParseDuration(dpBackupTimeout)
if err != nil {
return
}
if backupTimeout < time.Second {
err = fmt.Errorf("dp backup timeout %v smaller than 1s", backupTimeout)
return
}
dpBackupTimeout = strconv.FormatInt(int64(backupTimeout), 10)
}
if err = client.AdminAPI().SetClusterParas(optDelBatchCount, optMarkDeleteRate, optDelWorkerSleepMs,
optAutoRepairRate, optLoadFactor, opMaxDpCntLimit, opMaxMpCntLimit, clientIDKey,
dataNodesetSelector, metaNodesetSelector,
dataNodeSelector, metaNodeSelector, markBrokenDiskThreshold,
autoDecommissionDisk, autoDecommissionDiskInterval,
autoDpMetaRepair, autoDpMetaRepairParallelCnt,
dpRepairTimeout, dpTimeout); err != nil {
dpRepairTimeout, dpTimeout, dpBackupTimeout); err != nil {
return
}
stdout("Cluster parameters has been set successfully. \n")
@ -364,6 +378,7 @@ func newClusterSetParasCmd(client *master.MasterClient) *cobra.Command {
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)")
cmd.Flags().StringVar(&dpBackupTimeout, CliFlagDpBackupTimeout, "", "Data partition backup directory timeout(example: 1h)")
return cmd
}

1
cli/cmd/const.go
View File

@ -140,6 +140,7 @@ const (
CliFlagDpTimeout = "dpTimeout"
CliFlagAutoDecommissionDisk = "autoDecommissionDisk"
CliFlagAutoDecommissionDiskInterval = "autoDecommissionDiskInterval"
CliFlagDpBackupTimeout = "dpBackupTimeout"
// CliFlagSetDataPartitionCount = "count" use dp-count instead

4
cli/cmd/datapartition.go
View File

@ -352,6 +352,7 @@ func newDataPartitionReplicateCmd(client *master.MasterClient) *cobra.Command {
func newDataPartitionDeleteReplicaCmd(client *master.MasterClient) *cobra.Command {
var clientIDKey string
var raftForceDel bool
cmd := &cobra.Command{
Use: CliOpDelReplica + " [ADDRESS] [DATA PARTITION ID]",
Short: cmdDataPartitionDeleteReplicaShort,
@ -368,7 +369,7 @@ func newDataPartitionDeleteReplicaCmd(client *master.MasterClient) *cobra.Comman
if partitionID, err = strconv.ParseUint(args[1], 10, 64); err != nil {
return
}
if err = client.AdminAPI().DeleteDataReplica(partitionID, address, clientIDKey); err != nil {
if err = client.AdminAPI().DeleteDataReplica(partitionID, address, clientIDKey, raftForceDel); err != nil {
return
}
stdoutln("Delete replication successfully")
@ -381,6 +382,7 @@ func newDataPartitionDeleteReplicaCmd(client *master.MasterClient) *cobra.Comman
},
}
cmd.Flags().StringVar(&clientIDKey, CliFlagClientIDKey, client.ClientIDKey(), CliUsageClientIDKey)
cmd.Flags().BoolVarP(&raftForceDel, "raftForceDel", "r", false, "true for raftForceDel")
return cmd
}

1
cli/cmd/fmt.go
View File

@ -78,6 +78,7 @@ func formatClusterView(cv *proto.ClusterView, cn *proto.ClusterNodeInfo, cp *pro
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))
sb.WriteString(fmt.Sprintf(" DpBackupTimeout : %v\n", cv.DpBackupTimeout))
return sb.String()
}

71
datanode/disk.go
View File

@ -98,6 +98,7 @@ type Disk struct {
extentRepairReadDp uint64
BackupDataPartitions sync.Map
recoverStatus uint32
BackupReplicaLk sync.RWMutex
}
const (
@ -141,6 +142,7 @@ func NewDisk(path string, reservedSpace, diskRdonlySpace uint64, maxErrCnt int,
err = nil
}
d.startScheduleToUpdateSpaceInfo()
d.startScheduleToDeleteBackupReplicaDirectories()
d.limitFactor = make(map[uint32]*rate.Limiter)
d.limitFactor[proto.FlowReadType] = rate.NewLimiter(rate.Limit(proto.QosDefaultDiskMaxFLowLimit), proto.QosDefaultBurst)
@ -700,8 +702,8 @@ func (d *Disk) RestorePartition(visitor PartitionVisitor) (err error) {
toDeleteExpiredPartitionNames = append(toDeleteExpiredPartitionNames, name)
log.LogInfof("action[RestorePartition] find expired partition on path(%s)", name)
}
if d.isBackupPartitionDir(filename) {
if partitionID, err = unmarshalBackupPartitionDirName(filename); err != nil {
if d.isBackupPartitionDirToDelete(filename) {
if partitionID, _, err = unmarshalBackupPartitionDirNameAndTimestamp(filename); err != nil {
log.LogErrorf("action[RestorePartition] unmarshal partitionName(%v) from disk(%v) err(%v) ",
filename, d.Path, err.Error())
} else {
@ -871,11 +873,6 @@ func (d *Disk) QueryExtentRepairReadLimitStatus() (bool, uint64) {
return d.enableExtentRepairReadLimit, d.extentRepairReadDp
}
func (d *Disk) isBackupPartitionDir(filename string) (isBackupPartitionDir bool) {
isBackupPartitionDir = RegexpBackupDataPartitionDir.MatchString(filename)
return
}
func (d *Disk) AddBackupPartitionDir(id uint64) {
d.BackupDataPartitions.Store(id, proto.BackupDataPartitionInfo{Addr: d.dataNode.localServerAddr, Disk: d.Path, PartitionID: id})
}
@ -889,7 +886,7 @@ func (d *Disk) GetBackupPartitionDirList() (backupInfos []proto.BackupDataPartit
return backupInfos
}
func unmarshalBackupPartitionDirName(name string) (partitionID uint64, err error) {
func unmarshalBackupPartitionDirNameAndTimestamp(name string) (partitionID uint64, timestamp int64, err error) {
arr := strings.Split(name, "_")
if len(arr) != 4 {
err = fmt.Errorf("error backupDataPartition name(%v)", name)
@ -898,6 +895,19 @@ func unmarshalBackupPartitionDirName(name string) (partitionID uint64, err error
if partitionID, err = strconv.ParseUint(arr[2], 10, 64); err != nil {
return
}
arr = strings.Split(name, "-")
if len(arr) != 2 {
err = fmt.Errorf("error backupDataPartition name(%v)", name)
return
}
timestampStr := arr[1]
t, err := time.Parse("20060102150405", timestampStr)
if err != nil {
err = fmt.Errorf("error backupDataPartition timestamp(%v)", timestampStr)
return
}
timestamp = t.Unix()
return
}
@ -917,3 +927,48 @@ func (d *Disk) isBackupPartitionDirToDelete(filename string) (isBackupPartitionD
isBackupPartitionDir = RegexpBackupDataPartitionDirToDelete.MatchString(filename)
return
}
func (d *Disk) startScheduleToDeleteBackupReplicaDirectories() {
go func() {
ticker := time.NewTicker(time.Minute * 5)
defer func() {
ticker.Stop()
}()
for {
select {
case <-ticker.C:
d.BackupReplicaLk.Lock()
log.LogDebugf("action[startScheduleToDeleteBackupReplicaDirectories] begin.")
fileInfoList, err := os.ReadDir(d.Path)
if err != nil {
log.LogErrorf("action[startScheduleToDeleteBackupReplicaDirectories] read dir(%v) err(%v).", d.Path, err)
d.BackupReplicaLk.Unlock()
continue
}
var ts int64
for _, fileInfo := range fileInfoList {
filename := fileInfo.Name()
if !d.isBackupPartitionDirToDelete(filename) {
continue
}
if _, _, err = unmarshalBackupPartitionDirNameAndTimestamp(filename); err != nil {
log.LogErrorf("action[startScheduleToDeleteBackupReplicaDirectories] unmarshal partitionName(%v) from disk(%v) err(%v) ",
filename, d.Path, err.Error())
continue
}
if time.Now().Sub(time.Unix(ts, 0)) > d.dataNode.dpBackupTimeout {
err = os.RemoveAll(path.Join(d.Path, filename))
if err != nil {
log.LogWarnf("action[startScheduleToDeleteBackupReplicaDirectories] failed to remove %v err(%v) ",
path.Join(d.Path, filename), err.Error())
}
}
}
d.BackupReplicaLk.Unlock()
}
}
}()
}

20
datanode/partition.go
View File

@ -694,16 +694,16 @@ func (dp *DataPartition) RemoveAll(force bool) (err error) {
parent := path.Dir(originalPath)
fileName := path.Base(originalPath)
newFilename := BackupPartitionPrefix + fileName
newPath := path.Join(parent, newFilename)
_, err = os.Stat(newPath)
if err == nil {
newPathWithTimestamp := fmt.Sprintf("%v-%v", newPath, time.Now().Format("20060102150405"))
err = os.Rename(newPath, newPathWithTimestamp)
if err != nil {
log.LogWarnf("action[Stop]:dp(%v) rename dir from %v to %v,err %v", dp.info(), newPath, newPathWithTimestamp, err)
return err
}
}
newPath := fmt.Sprintf("%v-%v", path.Join(parent, newFilename), time.Now().Format("20060102150405"))
//_, err = os.Stat(newPath)
//if err == nil {
// newPathWithTimestamp := fmt.Sprintf("%v-%v", newPath, time.Now().Format("20060102150405"))
// err = os.Rename(newPath, newPathWithTimestamp)
// if err != nil {
// log.LogWarnf("action[Stop]:dp(%v) rename dir from %v to %v,err %v", dp.info(), newPath, newPathWithTimestamp, err)
// return err
// }
//}
err = os.Rename(originalPath, newPath)
if err == nil {
dp.path = newPath

3
datanode/server.go
View File

@ -200,6 +200,7 @@ type DataNode struct {
diskUnavailablePartitionErrorCount uint64 // disk status becomes unavailable when disk error partition count reaches this value
started int32
dpBackupTimeout time.Duration
}
type verOp2Phase struct {
@ -748,7 +749,7 @@ func (s *DataNode) registerHandler() {
http.HandleFunc("/reloadDataPartition", s.reloadDataPartition)
http.HandleFunc("/setDiskExtentReadLimitStatus", s.setDiskExtentReadLimitStatus)
http.HandleFunc("/queryDiskExtentReadLimitStatus", s.queryDiskExtentReadLimitStatus)
// http.HandleFunc("/detachDataPartition", s.detachDataPartition)
http.HandleFunc("/detachDataPartition", s.detachDataPartition)
http.HandleFunc("/loadDataPartition", s.loadDataPartition)
http.HandleFunc("/releaseDiskExtentReadLimitToken", s.releaseDiskExtentReadLimitToken)
http.HandleFunc("/markDataPartitionBroken", s.markDataPartitionBroken)

20
datanode/space_manager.go
View File

@ -746,16 +746,16 @@ func (manager *SpaceManager) deleteDataPartitionNotLoaded(id uint64, decommissio
if partitionID == id {
rootPath := path.Join(d.Path, filename)
if force {
newPath := path.Join(d.Path, BackupPartitionPrefix+filename)
_, err := os.Stat(newPath)
if err == nil {
newPathWithTimestamp := fmt.Sprintf("%v-%v", newPath, time.Now().Format("20060102150405"))
err = os.Rename(newPath, newPathWithTimestamp)
if err != nil {
log.LogWarnf("action[deleteDataPartitionNotLoaded]: rename dir from %v to %v,err %v", newPath, newPathWithTimestamp, err)
return err
}
}
newPath := fmt.Sprintf("%v-%v", path.Join(d.Path, BackupPartitionPrefix+filename), time.Now().Format("20060102150405"))
//_, err := os.Stat(newPath)
//if err == nil {
// newPathWithTimestamp := fmt.Sprintf("%v-%v", newPath, time.Now().Format("20060102150405"))
// err = os.Rename(newPath, newPathWithTimestamp)
// if err != nil {
// log.LogWarnf("action[deleteDataPartitionNotLoaded]: rename dir from %v to %v,err %v", newPath, newPathWithTimestamp, err)
// return err
// }
//}
err = os.Rename(rootPath, newPath)
if err == nil {
d.AddBackupPartitionDir(id)

26
datanode/wrap_operator.go
View File

@ -20,6 +20,7 @@ import (
"encoding/json"
"fmt"
"hash/crc32"
"math"
"net"
"os"
"path"
@ -530,7 +531,7 @@ func (s *DataNode) handleHeartbeatPacket(p *repl.Packet) {
request.EnableDiskQos,
s.diskQosEnable)
}
s.dpBackupTimeout, err = time.ParseDuration(request.DpBackupTimeout)
log.LogDebugf("handleHeartbeatPacket receive req(%v)", task.RequestID)
// NOTE: set decommission disks
s.checkDecommissionDisks(request.DecommissionDisks)
@ -1809,45 +1810,56 @@ func (s *DataNode) handlePacketToRecoverBackupDataReplica(p *repl.Packet) {
log.LogErrorf("action[handlePacketToRecoverBackupDataReplica] disk(%v) is not found err(%v).", request.Disk, err)
return
}
disk.BackupReplicaLk.Lock()
fileInfoList, err := os.ReadDir(disk.Path)
if err != nil {
log.LogErrorf("action[handlePacketToRecoverBackupDataReplica] read dir(%v) err(%v).", disk.Path, err)
disk.BackupReplicaLk.Unlock()
return
}
rootDir := ""
minDiff := math.MaxInt64
currentTime := time.Now().Unix()
for _, fileInfo := range fileInfoList {
filename := fileInfo.Name()
if !disk.isBackupPartitionDir(filename) {
if !disk.isBackupPartitionDirToDelete(filename) {
continue
}
if id, err := unmarshalBackupPartitionDirName(filename); err != nil {
if id, ts, err := unmarshalBackupPartitionDirNameAndTimestamp(filename); err != nil {
log.LogErrorf("action[handlePacketToRecoverBackupDataReplica] unmarshal partitionName(%v) from disk(%v) err(%v) ",
filename, disk.Path, err.Error())
continue
} else {
if id == request.PartitionId {
rootDir = filename
diff := int(math.Abs(float64(ts - currentTime)))
if diff < minDiff {
minDiff = diff
rootDir = filename
}
}
}
}
if rootDir == "" {
err = errors.NewErrorf("dp(%v) root not found in dir(%v)", request.PartitionId, disk.Path)
log.LogErrorf("action[handlePacketToRecoverBackupDataReplica] err %v", err.Error())
disk.BackupReplicaLk.Unlock()
return
}
log.LogDebugf("action[handlePacketToRecoverBackupDataReplica] ready to recover %v", rootDir)
// rename root dir back to normal
newPath := strings.Replace(rootDir, BackupPartitionPrefix, "", 1)
parts := strings.Split(newPath, "-")
newPath = parts[0]
err = os.Rename(path.Join(disk.Path, rootDir), path.Join(disk.Path, newPath))
if err != nil {
log.LogErrorf("action[handlePacketToRecoverBackupDataReplica] rename disk %v rootDir %v to %v failed %v.",
disk.Path, rootDir, newPath, err)
disk.BackupReplicaLk.Unlock()
return
}
disk.BackupReplicaLk.Unlock()
_, err = LoadDataPartition(path.Join(request.Disk, newPath), disk)
if err != nil {
log.LogErrorf("action[handlePacketToRecoverBackupDataReplica] load disk %v rootDir %v failed err %v.",

11
master/api_args_parse.go
View File

@ -1281,7 +1281,16 @@ func parseAndExtractSetNodeInfoParams(r *http.Request) (params map[string]interf
}
params[nodeDpRepairTimeOutKey] = val
}
if value = r.FormValue(nodeDpBackupKey); value != "" {
noParams = false
val := uint64(0)
val, err = strconv.ParseUint(value, 10, 64)
if err != nil {
err = unmatchedKey(nodeDpBackupKey)
return
}
params[nodeDpBackupKey] = val
}
if value = r.FormValue(nodeDpMaxRepairErrCntKey); value != "" {
noParams = false
val := uint64(0)

10
master/api_service.go
View File

@ -795,6 +795,7 @@ func (m *Server) getCluster(w http.ResponseWriter, r *http.Request) {
MaxMetaPartitionID: m.cluster.idAlloc.metaPartitionID,
VolDeletionDelayTimeHour: m.cluster.cfg.volDelayDeleteTimeHour,
DpRepairTimeout: m.cluster.GetDecommissionDataPartitionRecoverTimeOut().String(),
DpBackupTimeout: m.cluster.GetDecommissionDataPartitionBackupTimeOut().String(),
MarkDiskBrokenThreshold: m.cluster.getMarkDiskBrokenThreshold(),
EnableAutoDpMetaRepair: m.cluster.getEnableAutoDpMetaRepair(),
AutoDpMetaRepairParallelCnt: m.cluster.GetAutoDpMetaRepairParallelCnt(),
@ -3103,6 +3104,15 @@ func (m *Server) setNodeInfoHandler(w http.ResponseWriter, r *http.Request) {
}
}
if val, ok := params[nodeDpBackupKey]; ok {
if v, ok := val.(uint64); ok {
if err = m.cluster.setDataPartitionBackupTimeOut(v); err != nil {
sendErrReply(w, r, newErrHTTPReply(err))
return
}
}
}
if val, ok := params[nodeDpMaxRepairErrCntKey]; ok {
if v, ok := val.(uint64); ok {
if err = m.cluster.setDataPartitionMaxRepairErrCnt(v); err != nil {

20
master/cluster.go
View File

@ -776,7 +776,7 @@ func (c *Cluster) checkDataNodeHeartbeat() {
node := dataNode.(*DataNode)
node.checkLiveness()
log.LogDebugf("checkDataNodeHeartbeat checkLiveness for data node %v %v", node.Addr, id.String())
task := node.createHeartbeatTask(c.masterAddr(), c.diskQosEnable)
task := node.createHeartbeatTask(c.masterAddr(), c.diskQosEnable, c.GetDecommissionDataPartitionBackupTimeOut().String())
log.LogDebugf("checkDataNodeHeartbeat createHeartbeatTask for data node %v task %v %v", node.Addr,
task.RequestID, id.String())
hbReq := task.Request.(*proto.HeartBeatRequest)
@ -3755,6 +3755,17 @@ func (c *Cluster) setDataPartitionRepairTimeOut(val uint64) (err error) {
return
}
func (c *Cluster) setDataPartitionBackupTimeOut(val uint64) (err error) {
oldVal := atomic.LoadUint64(&c.cfg.DpBackupTimeOut)
atomic.StoreUint64(&c.cfg.DpBackupTimeOut, val)
if err = c.syncPutCluster(); err != nil {
log.LogErrorf("action[setDataPartitionBackupTimeOut] err[%v]", err)
atomic.StoreUint64(&c.cfg.DpBackupTimeOut, oldVal)
err = proto.ErrPersistenceByRaft
return
}
return
}
func (c *Cluster) setDataNodeAutoRepairLimitRate(val uint64) (err error) {
oldVal := atomic.LoadUint64(&c.cfg.DataNodeAutoRepairLimitRate)
atomic.StoreUint64(&c.cfg.DataNodeAutoRepairLimitRate, val)
@ -5009,6 +5020,13 @@ func (c *Cluster) GetDecommissionDataPartitionRecoverTimeOut() time.Duration {
return time.Duration(c.cfg.DpRepairTimeOut)
}
func (c *Cluster) GetDecommissionDataPartitionBackupTimeOut() time.Duration {
if c.cfg.DpBackupTimeOut == 0 {
return time.Hour * 24 * 7
}
return time.Duration(c.cfg.DpBackupTimeOut)
}
func (c *Cluster) GetDecommissionDiskLimit() (limit uint32) {
limit = atomic.LoadUint32(&c.DecommissionDiskLimit)
if limit == 0 {

1
master/config.go
View File

@ -129,6 +129,7 @@ type clusterConfig struct {
DataNodeAutoRepairLimitRate uint64 // datanode autorepair limit rate
DpMaxRepairErrCnt uint64
DpRepairTimeOut uint64
DpBackupTimeOut uint64
peers []raftstore.PeerAddress
peerAddrs []string
heartbeatPort int64

1
master/const.go
View File

@ -79,6 +79,7 @@ const (
nodeDeleteWorkerSleepMs = "deleteWorkerSleepMs"
nodeAutoRepairRateKey = "autoRepairRate"
nodeDpRepairTimeOutKey = "dpRepairTimeOut"
nodeDpBackupKey = "dpBackupTimeout"
nodeDpMaxRepairErrCntKey = "dpMaxRepairErrCnt"
clusterLoadFactorKey = "loadFactor"
maxDpCntLimitKey = "maxDpCntLimit"

3
master/data_node.go
View File

@ -366,7 +366,7 @@ func (dataNode *DataNode) clean() {
dataNode.TaskManager.exitCh <- struct{}{}
}
func (dataNode *DataNode) createHeartbeatTask(masterAddr string, enableDiskQos bool) (task *proto.AdminTask) {
func (dataNode *DataNode) createHeartbeatTask(masterAddr string, enableDiskQos bool, dpBackupTimeout string) (task *proto.AdminTask) {
request := &proto.HeartBeatRequest{
CurrTime: time.Now().Unix(),
MasterAddr: masterAddr,
@ -378,6 +378,7 @@ func (dataNode *DataNode) createHeartbeatTask(masterAddr string, enableDiskQos b
request.QosFlowReadLimit = dataNode.QosFlowRLimit
request.QosFlowWriteLimit = dataNode.QosFlowWLimit
request.DecommissionDisks = dataNode.getDecommissionedDisks()
request.DpBackupTimeout = dpBackupTimeout
task = proto.NewAdminTask(proto.OpDataNodeHeartbeat, dataNode.Addr, request)
return

7
master/metadata_fsm_op.go
View File

@ -60,6 +60,7 @@ type clusterValue struct {
MaxConcurrentLcNodes uint64
DpMaxRepairErrCnt uint64
DpRepairTimeOut uint64
DpBackupTimeOut uint64
EnableAutoDecommissionDisk bool
AutoDecommissionDiskInterval int64
DecommissionDiskLimit uint32
@ -97,6 +98,7 @@ func newClusterValue(c *Cluster) (cv *clusterValue) {
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(),
@ -1028,6 +1030,10 @@ func (c *Cluster) updateDataPartitionRepairTimeOut(val uint64) {
atomic.StoreUint64(&c.cfg.DpRepairTimeOut, val)
}
func (c *Cluster) updateDataPartitionBackupTimeOut(val uint64) {
atomic.StoreUint64(&c.cfg.DpBackupTimeOut, val)
}
func (c *Cluster) updateDataPartitionTimeoutSec(val int64) {
atomic.StoreInt64(&c.cfg.DataPartitionTimeOutSec, val)
}
@ -1215,6 +1221,7 @@ func (c *Cluster) loadClusterValue() (err error) {
c.updateDataNodeAutoRepairLimit(cv.DataNodeAutoRepairLimitRate)
c.updateDataPartitionMaxRepairErrCnt(cv.DpMaxRepairErrCnt)
c.updateDataPartitionRepairTimeOut(cv.DpRepairTimeOut)
c.updateDataPartitionBackupTimeOut(cv.DpBackupTimeOut)
c.updateMaxDpCntLimit(cv.MaxDpCntLimit)
c.updateMaxMpCntLimit(cv.MaxMpCntLimit)
if cv.MetaPartitionInodeIdStep == 0 {

1
proto/admin_proto.go
View File

@ -718,6 +718,7 @@ type HeartBeatRequest struct {
DisableAuditVols []string
DecommissionDisks []string // NOTE: for datanode
VolDpRepairBlockSize map[string]uint64
DpBackupTimeout string
}
// DataPartitionReport defines the partition report.

1
proto/model.go
View File

@ -135,6 +135,7 @@ type ClusterView struct {
AutoDecommissionDiskInterval string
DecommissionDiskLimit uint32
DpRepairTimeout string
DpBackupTimeout string
DpTimeout string
DataNodeStatInfo *NodeStatInfo
MetaNodeStatInfo *NodeStatInfo

8
sdk/master/api_admin.go
View File

@ -190,11 +190,12 @@ func (api *AdminAPI) DecommissionMetaPartition(metaPartitionID uint64, nodeAddr,
return
}
func (api *AdminAPI) DeleteDataReplica(dataPartitionID uint64, nodeAddr, clientIDKey string) (err error) {
func (api *AdminAPI) DeleteDataReplica(dataPartitionID uint64, nodeAddr, clientIDKey string, raftForce bool) (err error) {
request := newRequest(get, proto.AdminDeleteDataReplica).Header(api.h)
request.addParam("id", strconv.FormatUint(dataPartitionID, 10))
request.addParam("addr", nodeAddr)
request.addParam("clientIDKey", clientIDKey)
request.addParam("raftForceDel", strconv.FormatBool(raftForce))
_, err = api.mc.serveRequest(request)
return
}
@ -519,7 +520,7 @@ func (api *AdminAPI) SetClusterParas(batchCount, markDeleteRate, deleteWorkerSle
dataNodesetSelector, metaNodesetSelector, dataNodeSelector, metaNodeSelector string, markDiskBrokenThreshold string,
enableAutoDecommissionDisk string, autoDecommissionDiskInterval string,
enableAutoDpMetaRepair string, autoDpMetaRepairParallelCnt string,
dpRepairTimeout string, dpTimeout string,
dpRepairTimeout string, dpTimeout string, dpBackupTimeout string,
) (err error) {
request := newRequest(get, proto.AdminSetNodeInfo).Header(api.h)
request.addParam("batchCount", batchCount)
@ -556,6 +557,9 @@ func (api *AdminAPI) SetClusterParas(batchCount, markDeleteRate, deleteWorkerSle
if dpTimeout != "" {
request.addParam("dpTimeout", dpTimeout)
}
if dpBackupTimeout != "" {
request.addParam("dpBackupTimeout", dpBackupTimeout)
}
_, err = api.mc.serveRequest(request)
return
}