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

fix(datanode): if dpBackupTimeout is not set, then do not delete back up directories.

Browse Source 
Signed-off-by: chihe <chihe@oppo.com>
This commit is contained in:
chihe 2024-10-10 15:11:18 +08:00 committed by AmazingChi
parent 111f9e9903
commit 852dafafd2
5 changed files with 30 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
cli/cmd/cluster.go
View File

@ -338,8 +338,8 @@ func newClusterSetParasCmd(client *master.MasterClient) *cobra.Command {
if err != nil {
return
}
if backupTimeout < time.Second {
err = fmt.Errorf("dp backup timeout %v smaller than 1s", backupTimeout)
if backupTimeout < proto.DefaultDataPartitionBackupTimeOut {
err = fmt.Errorf("dp backup timeout %v smaller than %v", backupTimeout, proto.DefaultDataPartitionBackupTimeOut)
return
}

20
datanode/disk.go
View File

@ -30,6 +30,7 @@ import (
"github.com/cubefs/cubefs/depends/tiglabs/raft"
"github.com/cubefs/cubefs/proto"
"github.com/cubefs/cubefs/util/auditlog"
"github.com/cubefs/cubefs/util/errors"
"github.com/cubefs/cubefs/util/exporter"
"github.com/cubefs/cubefs/util/loadutil"
@ -676,7 +677,9 @@ func (d *Disk) RestorePartition(visitor PartitionVisitor) (err error) {
}
}()
if !checkDiskPathWithMaster(d.Path, partitionID, replicaDiskInfos) {
// if master is updated after dataNode, replicaDiskInfos should be empty
// so do not rename dp with BackupPartitionPrefix
if len(replicaDiskInfos) != 0 && !checkDiskPathWithMaster(d.Path, partitionID, replicaDiskInfos) {
msg := fmt.Sprintf("action[RestorePartition] replica for partition(%v) on disk (%v) is not "+
"matched with master",
partitionID, d.Path)
@ -798,7 +801,9 @@ func (d *Disk) deleteExpiredPartitions(toDeleteExpiredPartitionNames []string) (
log.LogErrorf("action[deleteExpiredPartitions] delete expiredPartition %v automatically fail, err(%v)", partitionName, err)
continue
}
log.LogInfof("action[deleteExpiredPartitions] delete expiredPartition %v automatically", partitionName)
msg := fmt.Sprintf("action[deleteExpiredPartitions] delete expiredPartition %v automatically", partitionName)
log.LogInfof("%v", msg)
auditlog.LogDataNodeOp("deleteExpiredPartitions", msg, nil)
time.Sleep(time.Second)
} else {
notDeletedExpiredPartitionNames = append(notDeletedExpiredPartitionNames, partitionName)
@ -958,6 +963,10 @@ func (d *Disk) startScheduleToDeleteBackupReplicaDirectories() {
for {
select {
case <-ticker.C:
if d.dataNode.dpBackupTimeout <= 0 {
log.LogDebugf("action[startScheduleToDeleteBackupReplicaDirectories] skip.")
continue
}
d.BackupReplicaLk.Lock()
log.LogDebugf("action[startScheduleToDeleteBackupReplicaDirectories] begin.")
fileInfoList, err := os.ReadDir(d.Path)
@ -985,6 +994,9 @@ func (d *Disk) startScheduleToDeleteBackupReplicaDirectories() {
if err != nil {
log.LogWarnf("action[startScheduleToDeleteBackupReplicaDirectories] failed to remove %v err(%v) ",
path.Join(d.Path, filename), err.Error())
} else {
msg := fmt.Sprintf("action[startScheduleToDeleteBackupReplicaDirectories] remove %v", path.Join(d.Path, filename))
auditlog.LogDataNodeOp("DeleteBackupReplicaDirectories", msg, nil)
}
}
}
@ -995,8 +1007,8 @@ func (d *Disk) startScheduleToDeleteBackupReplicaDirectories() {
}
func checkDiskPathWithMaster(diskPath string, id uint64, infos map[uint64]string) bool {
if diskPath == infos[id] {
return true
if value, ok := infos[id]; ok {
return value == diskPath
}
return false
}

8
datanode/wrap_operator.go
View File

@ -535,8 +535,12 @@ 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)
dpBackupTimeout, _ := time.ParseDuration(request.DpBackupTimeout)
if dpBackupTimeout <= proto.DefaultDataPartitionBackupTimeOut {
dpBackupTimeout = proto.DefaultDataPartitionBackupTimeOut
}
s.dpBackupTimeout = dpBackupTimeout
log.LogDebugf("handleHeartbeatPacket receive req(%v) dpBackupTimeout(%v)", task.RequestID, dpBackupTimeout)
// NOTE: set decommission disks
s.checkDecommissionDisks(request.DecommissionDisks)
log.LogDebugf("handleHeartbeatPacket checkDecommissionDisks req(%v) cost %v",

5
master/cluster.go
View File

@ -3884,6 +3884,9 @@ func (c *Cluster) setDataPartitionRepairTimeOut(val uint64) (err error) {
func (c *Cluster) setDataPartitionBackupTimeOut(val uint64) (err error) {
oldVal := atomic.LoadUint64(&c.cfg.DpBackupTimeOut)
if val < uint64(proto.DefaultDataPartitionBackupTimeOut/time.Second) {
val = uint64(proto.DefaultDataPartitionBackupTimeOut / time.Second)
}
atomic.StoreUint64(&c.cfg.DpBackupTimeOut, val)
if err = c.syncPutCluster(); err != nil {
log.LogErrorf("action[setDataPartitionBackupTimeOut] err[%v]", err)
@ -5200,7 +5203,7 @@ func (c *Cluster) GetDecommissionDataPartitionRecoverTimeOut() time.Duration {
func (c *Cluster) GetDecommissionDataPartitionBackupTimeOut() time.Duration {
if c.cfg.DpBackupTimeOut == 0 {
return time.Hour * 24 * 7
return proto.DefaultDataPartitionBackupTimeOut
}
return time.Duration(c.cfg.DpBackupTimeOut)
}

2
proto/admin_proto.go
View File

@ -1077,6 +1077,8 @@ const (
QosDefaultDiskMaxIoLimit int = 100000
)
const DefaultDataPartitionBackupTimeOut = time.Hour * 24 * 7
func QosTypeString(factorType uint32) string {
switch factorType {
case IopsReadType: