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(cli): support raftForce for datanode decommission by cfs-cli

Browse Source 
Signed-off-by: chihe <chihe@oppo.com>
This commit is contained in:
chihe 2024-08-16 16:45:33 +08:00 committed by AmazingChi
parent 46157d2c06
commit da33bc2068
11 changed files with 84 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
cli/cmd/const.go
View File

@ -146,6 +146,7 @@ const (
CliFlagTrashInterval = "trashInterval"
CliFlagAccessTimeValidInterval = "accessTimeValidInterval"
CliFlagEnablePersistAccessTime = "enablePersistAccessTime"
CliFlagDecommissionRaftForce = "raftForceDel"
// CliFlagSetDataPartitionCount = "count" use dp-count instead
// Shorthand format of resource name

8
cli/cmd/datanode.go
View File

@ -112,8 +112,9 @@ func newDataNodeInfoCmd(client *master.MasterClient) *cobra.Command {
func newDataNodeDecommissionCmd(client *master.MasterClient) *cobra.Command {
var (
optCount int
clientIDKey string
optCount int
clientIDKey string
raftForceDel bool
)
cmd := &cobra.Command{
Use: CliOpDecommission + " [{HOST}:{PORT}]",
@ -124,7 +125,7 @@ func newDataNodeDecommissionCmd(client *master.MasterClient) *cobra.Command {
stdoutln("Migrate dp count should >= 0")
return nil
}
if err := client.NodeAPI().DataNodeDecommission(args[0], optCount, clientIDKey); err != nil {
if err := client.NodeAPI().DataNodeDecommission(args[0], optCount, clientIDKey, raftForceDel); err != nil {
return err
}
stdoutln("Decommission data node successfully")
@ -139,6 +140,7 @@ func newDataNodeDecommissionCmd(client *master.MasterClient) *cobra.Command {
}
cmd.Flags().IntVar(&optCount, CliFlagCount, 0, "DataNode delete mp count")
cmd.Flags().StringVar(&clientIDKey, CliFlagClientIDKey, client.ClientIDKey(), CliUsageClientIDKey)
cmd.Flags().BoolVarP(&raftForceDel, CliFlagDecommissionRaftForce, "r", false, "true for raftForceDel")
return cmd
}

2
cli/cmd/datapartition.go
View File

@ -310,7 +310,7 @@ func newDataPartitionDecommissionCmd(client *master.MasterClient) *cobra.Command
return validDataNodes(client, toComplete), cobra.ShellCompDirectiveNoFileComp
},
}
cmd.Flags().BoolVarP(&raftForceDel, "raftForceDel", "r", false, "true for raftForceDel")
cmd.Flags().BoolVarP(&raftForceDel, CliFlagDecommissionRaftForce, "r", false, "true for raftForceDel")
cmd.Flags().StringVar(&decommissionType, "decommissionType", "1", "decommission type")
cmd.Flags().StringVar(&clientIDKey, CliFlagClientIDKey, client.ClientIDKey(), CliUsageClientIDKey)
return cmd

2
datanode/partition.go
View File

@ -17,6 +17,7 @@ package datanode
import (
"encoding/json"
"fmt"
"github.com/cubefs/cubefs/util/auditlog"
"hash/crc32"
"io/ioutil"
syslog "log"
@ -637,6 +638,7 @@ func (dp *DataPartition) Stop() {
}
msg := fmt.Sprintf("[Stop] stop disk(%v) dp(%v) using time(%v), slow(%v)", diskPath, dp.partitionID, time.Since(begin), time.Since(begin) > 100*time.Millisecond)
log.LogInfo(msg)
auditlog.LogDataNodeOp("DataPartitionStop", msg, nil)
}()
dp.stopOnce.Do(func() {
log.LogInfof("action[Stop]:dp(%v) stop once", dp.info())

5
datanode/partition_raft.go
View File

@ -18,6 +18,7 @@ import (
"encoding/binary"
"encoding/json"
"fmt"
"github.com/cubefs/cubefs/util/auditlog"
syslog "log"
"net"
"os"
@ -153,7 +154,9 @@ func (dp *DataPartition) stopRaft() {
if !dp.isNormalType() {
return
}
log.LogErrorf("[FATAL] stop raft partition(%v)", dp.info())
msg := fmt.Sprintf("stop raft partition(%v)", dp.info())
log.LogErrorf("[FATAL] %v", msg)
auditlog.LogDataNodeOp("DataPartitionStopRaft", msg, nil)
dp.raftPartition.Stop()
}
}

10
datanode/partition_raftfsm.go
View File

@ -19,6 +19,7 @@ import (
"encoding/binary"
"encoding/json"
"fmt"
"github.com/cubefs/cubefs/util/auditlog"
"sync"
"sync/atomic"
@ -86,6 +87,7 @@ func (dp *DataPartition) ApplyMemberChange(confChange *raftproto.ConfChange, ind
// Change memory the status
var (
isUpdated bool
msg string
)
switch confChange.Type {
case raftproto.ConfAddNode:
@ -93,7 +95,8 @@ func (dp *DataPartition) ApplyMemberChange(confChange *raftproto.ConfChange, ind
if err = json.Unmarshal(confChange.Context, req); err != nil {
return
}
log.LogInfof("action[ApplyMemberChange] ConfAddNode [%v], partitionId [%v] index(%v)", req.AddPeer, req.PartitionId, index)
msg = fmt.Sprintf("ConfAddNode [%v], partitionId [%v] index(%v)", req.AddPeer, req.PartitionId, index)
log.LogInfof("action[ApplyMemberChange] %v", msg)
isUpdated, err = dp.addRaftNode(req, index)
if isUpdated && err == nil {
// Perform the update replicas operation asynchronously after the execution of the member change applying
@ -114,13 +117,16 @@ func (dp *DataPartition) ApplyMemberChange(confChange *raftproto.ConfChange, ind
}()
updateWG.Wait()
}
auditlog.LogDataNodeOp("DataPartitionMemberChange", msg, err)
case raftproto.ConfRemoveNode:
req := &proto.RemoveDataPartitionRaftMemberRequest{}
if err = json.Unmarshal(confChange.Context, req); err != nil {
return
}
log.LogInfof("action[ApplyMemberChange] ConfRemoveNode [%v], partitionId [%v] index(%v)", req.RemovePeer, req.PartitionId, index)
msg = fmt.Sprintf(" ConfRemoveNode [%v], partitionId [%v] index(%v)", req.RemovePeer, req.PartitionId, index)
log.LogInfof("action[ApplyMemberChange] %v", msg)
isUpdated, err = dp.removeRaftNode(req, index)
auditlog.LogDataNodeOp("DataPartitionMemberChange", msg, err)
case raftproto.ConfUpdateNode:
log.LogDebugf("[updateRaftNode]: not support.")
default:

7
datanode/space_manager.go
View File

@ -16,6 +16,7 @@ package datanode
import (
"fmt"
"github.com/cubefs/cubefs/util/auditlog"
"math"
"math/rand"
"os"
@ -532,6 +533,12 @@ func (manager *SpaceManager) CreatePartition(request *proto.CreateDataPartitionR
log.LogErrorf("[CreatePartition] dp(%v) failed to select disk", dpCfg.PartitionID)
return nil, ErrNoSpaceToCreatePartition
}
defer func() {
msg := fmt.Sprintf("dp %v request.Members %v on disk %v",
dpCfg.PartitionID, request.Members, disk.Path)
auditlog.LogDataNodeOp("DataPartitionCreate", msg, err)
}()
if dp, err = CreateDataPartition(dpCfg, disk, request); err != nil {
return
}

2
datanode/wrap_operator.go
View File

@ -19,6 +19,7 @@ import (
"encoding/binary"
"encoding/json"
"fmt"
"github.com/cubefs/cubefs/util/auditlog"
"hash/crc32"
"math"
"net"
@ -623,6 +624,7 @@ func (s *DataNode) handlePacketToDeleteDataPartition(p *repl.Packet) {
} else {
log.LogInfof(fmt.Sprintf("action[handlePacketToDeleteDataPartition] %v success", request.PartitionId))
}
auditlog.LogDataNodeOp("DeleteDataPartition", fmt.Sprintf("%v is deleted", request.PartitionId), err)
}
// Handle OpLoadDataPartition packet.

18
master/cluster.go
View File

@ -4183,6 +4183,11 @@ func (c *Cluster) TryDecommissionDataNode(dataNode *DataNode) {
if left-dpCnt >= 0 {
err = c.migrateDisk(dataNode, disk, dataNode.DecommissionDstAddr, dataNode.DecommissionRaftForce, dpCnt, true, ManualDecommission)
if err != nil {
if strings.Contains(err.Error(), "still on working") {
decommissionDiskList = append(decommissionDiskList, disk)
log.LogWarnf("action[TryDecommissionDataNode] disk(%v_%v) is decommissioning,add it to "+
"decommissionDiskList", dataNode.Addr, disk)
}
msg := fmt.Sprintf("disk(%v_%v)failed to mark decommission", dataNode.Addr, disk)
log.LogWarnf("action[TryDecommissionDataNode] %v failed %v", msg, err)
auditlog.LogMasterOp("DiskDecommission", msg, err)
@ -4193,6 +4198,11 @@ func (c *Cluster) TryDecommissionDataNode(dataNode *DataNode) {
} else {
err = c.migrateDisk(dataNode, disk, dataNode.DecommissionDstAddr, dataNode.DecommissionRaftForce, left, true, ManualDecommission)
if err != nil {
if strings.Contains(err.Error(), "still on working") {
decommissionDiskList = append(decommissionDiskList, disk)
log.LogWarnf("action[TryDecommissionDataNode] disk(%v_%v) is decommissioning,add it to "+
"decommissionDiskList", dataNode.Addr, disk)
}
msg := fmt.Sprintf("disk(%v_%v)failed to mark decommission", dataNode.Addr, disk)
log.LogWarnf("action[TryDecommissionDataNode] %v failed %v", msg, err)
auditlog.LogMasterOp("DiskDecommission", msg, err)
@ -4217,10 +4227,10 @@ func (c *Cluster) TryDecommissionDataNode(dataNode *DataNode) {
dataNode.ToBeOffline = true
dataNode.DecommissionDiskList = decommissionDiskList
dataNode.DecommissionDpTotal = decommissionDpTotal
log.LogInfof("action[TryDecommissionDataNode] try decommission disk[%v] from dataNode[%s] "+
"raftForce [%v] to dst [%v] DecommissionDpTotal[%v]",
decommissionDiskList, dataNode.Addr, dataNode.DecommissionRaftForce,
dataNode.DecommissionDstAddr, dataNode.DecommissionDpTotal)
msg := fmt.Sprintf(" try decommission disk[%v] from dataNode[%s] raftForce [%v] to dst [%v] DecommissionDpTotal[%v]",
decommissionDiskList, dataNode.Addr, dataNode.DecommissionRaftForce, dataNode.DecommissionDstAddr, dataNode.DecommissionDpTotal)
log.LogInfof("action[TryDecommissionDataNode] %v", msg)
auditlog.LogMasterOp("DataNodeDecommission", msg, nil)
}
func (c *Cluster) migrateDisk(dataNode *DataNode, diskPath, dstPath string, raftForce bool, limit int, diskDisable bool, migrateType uint32) (err error) {

3
sdk/master/api_node.go
View File

@ -107,11 +107,12 @@ func (api *NodeAPI) ResponseDataNodeTask(task *proto.AdminTask) (err error) {
return api.mc.request(newRequest(post, proto.GetDataNodeTaskResponse).Header(api.h).Body(task))
}
func (api *NodeAPI) DataNodeDecommission(nodeAddr string, count int, clientIDKey string) (err error) {
func (api *NodeAPI) DataNodeDecommission(nodeAddr string, count int, clientIDKey string, raftForce bool) (err error) {
request := newRequest(get, proto.DecommissionDataNode).Header(api.h).NoTimeout()
request.addParam("addr", nodeAddr)
request.addParam("count", strconv.Itoa(count))
request.addParam("clientIDKey", clientIDKey)
request.addParam("raftForceDel", strconv.FormatBool(raftForce))
if _, err = api.mc.serveRequest(request); err != nil {
return
}

47
util/auditlog/auditlog.go
View File

@ -460,15 +460,6 @@ func LogTxOp(clientAddr, volume, op, txId string, err error, latency int64) {
gAdt.LogTxOp(clientAddr, volume, op, txId, err, latency)
}
func LogMasterOp(op, msg string, err error) {
gAdtMutex.RLock()
defer gAdtMutex.RUnlock()
if gAdt == nil {
return
}
gAdt.formatMasterLog(op, msg, err)
}
func ResetWriterBufferSize(size int) {
gAdtMutex.Lock()
defer gAdtMutex.Unlock()
@ -692,3 +683,41 @@ func isPathSafe(filePath string) bool {
match, _ := regexp.MatchString(safePattern, filePath)
return match
}
func LogMasterOp(op, msg string, err error) {
gAdtMutex.RLock()
defer gAdtMutex.RUnlock()
if gAdt == nil {
return
}
gAdt.formatMasterLog(op, msg, err)
}
func LogDataNodeOp(op, msg string, err error) {
gAdtMutex.RLock()
defer gAdtMutex.RUnlock()
if gAdt == nil {
return
}
gAdt.formatDataNodeLog(op, msg, err)
}
func (a *Audit) formatDataNodeLog(op, msg string, err error) {
if entry := a.formatDataNodeAudit(op, msg, err); entry != "" {
if a.prefix != nil {
entry = fmt.Sprintf("%s%s", a.prefix.String(), entry)
}
a.AddLog(entry)
}
}
func (a *Audit) formatDataNodeAudit(op, msg string, err error) (str string) {
var errStr string
if err != nil {
errStr = err.Error()
} else {
errStr = "nil"
}
str = fmt.Sprintf("%v, %v, %v, ERR: %v", a.formatCommonHeader(), op, msg, errStr)
return
}