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

enhance: support decommission broken replica for 2-replica no leader dp

Browse Source 
Signed-off-by: Victor1319 <834863182@qq.com>
This commit is contained in:
Victor1319 2022-10-10 15:06:35 +08:00 committed by leonrayang
parent fc3d883629
commit b1dd7fa937
7 changed files with 141 additions and 113 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

156
master/api_service.go
View File

@ -18,13 +18,14 @@ import (
"context"
"encoding/json"
"fmt"
"golang.org/x/time/rate"
"net/http"
"regexp"
"strconv"
"sync/atomic"
"time"
"golang.org/x/time/rate"
"strings"
"github.com/cubefs/cubefs/proto"
@ -858,18 +859,20 @@ func (m *Server) deleteDataReplica(w http.ResponseWriter, r *http.Request) {
}
// force only be used in scenario that dp of two replicas volume no leader caused by one replica crash
var value string
if value = r.FormValue(raftForceDelKey); value != "" {
raftForce, _ = strconv.ParseBool(value)
if raftForce && dp.ReplicaNum != 2 {
msg = fmt.Sprintf("failed! replicaNum [%v] and force should be used in two replcias datapartition", dp.ReplicaNum)
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: msg})
return
}
raftForce, err = parseRaftForce(r)
if err != nil {
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: err.Error()})
return
}
if value = r.FormValue(forceKey); value != "" {
force, _ = strconv.ParseBool(value)
if raftForce && dp.ReplicaNum != 2 {
msg = fmt.Sprintf("failed! replicaNum [%v] and force should be used in two replcias datapartition", dp.ReplicaNum)
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: msg})
return
}
force, err = pareseBoolWithDefault(r, forceKey, false)
if err != nil {
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: err.Error()})
return
}
if err = m.cluster.removeDataReplica(dp, addr, !force, raftForce); err != nil {
@ -950,6 +953,7 @@ func (m *Server) decommissionDataPartition(w http.ResponseWriter, r *http.Reques
dp *DataPartition
addr string
partitionID uint64
raftForce bool
err error
)
@ -961,14 +965,20 @@ func (m *Server) decommissionDataPartition(w http.ResponseWriter, r *http.Reques
sendErrReply(w, r, newErrHTTPReply(proto.ErrDataPartitionNotExists))
return
}
raftForce, err = parseRaftForce(r)
if err != nil {
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: err.Error()})
return
}
if dp.isSpecialReplicaCnt() {
rstMsg = fmt.Sprintf(proto.AdminDecommissionDataPartition+" dataPartitionID :%v is special replica cnt %v on node:%v async running,need check later",
partitionID, dp.ReplicaNum, addr)
go m.cluster.decommissionDataPartition(addr, dp, handleDataPartitionOfflineErr)
go m.cluster.decommissionDataPartition(addr, dp, raftForce, handleDataPartitionOfflineErr)
sendOkReply(w, r, newSuccessHTTPReply(rstMsg))
return
}
if err = m.cluster.decommissionDataPartition(addr, dp, handleDataPartitionOfflineErr); err != nil {
if err = m.cluster.decommissionDataPartition(addr, dp, raftForce, handleDataPartitionOfflineErr); err != nil {
sendErrReply(w, r, newErrHTTPReply(err))
return
}
@ -1181,22 +1191,6 @@ func (m *Server) volShrink(w http.ResponseWriter, r *http.Request) {
sendOkReply(w, r, newSuccessHTTPReply(msg))
}
// func (c *Cluster) checkZoneArgs(zoneName string, crossZone bool) error {
// var zones []string
// if zoneName == "" {
// zoneName = DefaultZoneName
// }
// zones = strings.Split(zoneName, ",")
// for _, name := range zones {
// if _, err := c.t.getZone(name); err != nil {
// return err
// }
// }
// return nil
// }
func (m *Server) checkCreateReq(req *createVolReq) (err error) {
if !proto.IsHot(req.volType) && !proto.IsCold(req.volType) {
@ -1539,6 +1533,7 @@ func (m *Server) decommissionDataNode(w http.ResponseWriter, r *http.Request) {
rstMsg string
offLineAddr string
limit int
raftForce bool
err error
)
@ -1546,13 +1541,18 @@ func (m *Server) decommissionDataNode(w http.ResponseWriter, r *http.Request) {
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: err.Error()})
return
}
raftForce, err = parseRaftForce(r)
if err != nil {
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: err.Error()})
return
}
if _, err = m.cluster.dataNode(offLineAddr); err != nil {
sendErrReply(w, r, newErrHTTPReply(proto.ErrDataNodeNotExists))
return
}
if err = m.cluster.migrateDataNode(offLineAddr, "", limit); err != nil {
if err = m.cluster.migrateDataNode(offLineAddr, "", limit, raftForce); err != nil {
sendErrReply(w, r, newErrHTTPReply(err))
return
}
@ -1567,6 +1567,11 @@ func (m *Server) migrateDataNodeHandler(w http.ResponseWriter, r *http.Request)
sendErrReply(w, r, newErrHTTPReply(err))
return
}
raftForce, err := parseRaftForce(r)
if err != nil {
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: err.Error()})
return
}
if limit > defaultMigrateDpCnt {
sendErrReply(w, r, newErrHTTPReply(fmt.Errorf("limit %d can't be bigger than %d", limit, defaultMigrateDpCnt)))
@ -1597,7 +1602,7 @@ func (m *Server) migrateDataNodeHandler(w http.ResponseWriter, r *http.Request)
return
}
if err = m.cluster.migrateDataNode(srcAddr, targetAddr, limit); err != nil {
if err = m.cluster.migrateDataNode(srcAddr, targetAddr, limit, raftForce); err != nil {
sendErrReply(w, r, newErrHTTPReply(err))
return
}
@ -2353,6 +2358,7 @@ func (m *Server) decommissionDisk(w http.ResponseWriter, r *http.Request) {
offLineAddr, diskPath string
err error
limit int
raftForce bool
badPartitionIds []uint64
badPartitions []*DataPartition
)
@ -2361,6 +2367,11 @@ func (m *Server) decommissionDisk(w http.ResponseWriter, r *http.Request) {
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: err.Error()})
return
}
raftForce, err = parseRaftForce(r)
if err != nil {
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: err.Error()})
return
}
if node, err = m.cluster.dataNode(offLineAddr); err != nil {
sendErrReply(w, r, newErrHTTPReply(proto.ErrDataNodeNotExists))
@ -2384,7 +2395,7 @@ func (m *Server) decommissionDisk(w http.ResponseWriter, r *http.Request) {
rstMsg = fmt.Sprintf("receive decommissionDisk node[%v] disk[%v] limit [%d], badPartitionIds[%v] has offline successfully",
node.Addr, diskPath, limit, badPartitionIds)
if err = m.cluster.decommissionDisk(node, diskPath, badPartitions); err != nil {
if err = m.cluster.decommissionDisk(node, raftForce, diskPath, badPartitions); err != nil {
sendErrReply(w, r, newErrHTTPReply(err))
return
}
@ -2605,36 +2616,6 @@ func (m *Server) loadMetaPartition(w http.ResponseWriter, r *http.Request) {
sendOkReply(w, r, newSuccessHTTPReply(msg))
}
func parseReqToDecoDisk(r *http.Request) (nodeAddr, diskPath string, limit int, err error) {
if err = r.ParseForm(); err != nil {
return
}
nodeAddr, err = extractNodeAddr(r)
if err != nil {
return
}
diskPath, err = extractDiskPath(r)
if err != nil {
return
}
limit, err = parseUintParam(r, countKey)
return
}
func extractPosixAcl(r *http.Request) (enablePosix bool, err error) {
var value string
if value = r.FormValue(enablePosixAclKey); value == "" {
return
}
status, err := strconv.ParseBool(value)
if err != nil {
return false, fmt.Errorf("parse %s failed, val %s", enablePosixAclKey, value)
}
return status, nil
}
func (m *Server) migrateMetaNodeHandler(w http.ResponseWriter, r *http.Request) {
srcAddr, targetAddr, limit, err := parseMigrateNodeParam(r)
if err != nil {
@ -2758,12 +2739,61 @@ func (m *Server) getRaftStatus(w http.ResponseWriter, r *http.Request) {
sendOkReply(w, r, newSuccessHTTPReply(data))
}
func parseReqToDecoDisk(r *http.Request) (nodeAddr, diskPath string, limit int, err error) {
if err = r.ParseForm(); err != nil {
return
}
nodeAddr, err = extractNodeAddr(r)
if err != nil {
return
}
diskPath, err = extractDiskPath(r)
if err != nil {
return
}
limit, err = parseUintParam(r, countKey)
return
}
type getVolParameter struct {
name string
authKey string
skipOwnerValidation bool
}
func pareseBoolWithDefault(r *http.Request, key string, old bool) (bool, error) {
val := r.FormValue(key)
if val == "" {
return old, nil
}
newVal, err := strconv.ParseBool(val)
if err != nil {
return false, fmt.Errorf("parse %s bool val err, err %s", key, err.Error())
}
return newVal, nil
}
func parseRaftForce(r *http.Request) (bool, error) {
return pareseBoolWithDefault(r, raftForceDelKey, false)
}
func extractPosixAcl(r *http.Request) (enablePosix bool, err error) {
var value string
if value = r.FormValue(enablePosixAclKey); value == "" {
return
}
status, err := strconv.ParseBool(value)
if err != nil {
return false, fmt.Errorf("parse %s failed, val %s", enablePosixAclKey, value)
}
return status, nil
}
func (m *Server) getMetaPartitions(w http.ResponseWriter, r *http.Request) {
var (
name string

34
master/cluster.go
View File

@ -16,12 +16,13 @@ package master
import (
"fmt"
"golang.org/x/time/rate"
"strings"
"sync"
"sync/atomic"
"time"
"golang.org/x/time/rate"
"github.com/cubefs/cubefs/datanode"
"github.com/cubefs/cubefs/proto"
"github.com/cubefs/cubefs/raftstore"
@ -1354,10 +1355,10 @@ func (c *Cluster) getAllMetaPartitionsByMetaNode(addr string) (partitions []*Met
}
func (c *Cluster) decommissionCancel(dataNode *DataNode) (err error) {
if dataNode.ToBeOffline == false {
log.LogInfof("action[decommissionCancel] dataNode is not on offline %v", dataNode.Addr)
return
}
// if dataNode.ToBeOffline {
// log.LogInfof("action[decommissionCancel] dataNode is not on offline %v", dataNode.Addr)
// return
// }
partitions := c.getAllDataPartitionByDataNode(dataNode.Addr)
for _, dp := range partitions {
if dp.isSpecialReplicaCnt() && dp.SingleDecommissionStatus > 0 {
@ -1371,7 +1372,7 @@ func (c *Cluster) decommissionCancel(dataNode *DataNode) (err error) {
return
}
func (c *Cluster) migrateDataNode(srcAddr, targetAddr string, limit int) (err error) {
func (c *Cluster) migrateDataNode(srcAddr, targetAddr string, limit int, raftForce bool) (err error) {
var toBeOffLinePartitions []*DataPartition
msg := fmt.Sprintf("action[migrateDataNode], src(%s) migrate to target(%s) cnt(%d)", srcAddr, targetAddr, limit)
@ -1385,7 +1386,7 @@ func (c *Cluster) migrateDataNode(srcAddr, targetAddr string, limit int) (err er
srcNode.MigrateLock.Lock()
defer srcNode.MigrateLock.Unlock()
if srcNode.ToBeOffline == true {
if srcNode.ToBeOffline {
err = fmt.Errorf("migrate src(%v) is still on working, please wait,check or cancel if abnormal", srcAddr)
log.LogWarnf("action[migrateDataNode] %v", err)
return
@ -1430,7 +1431,7 @@ func (c *Cluster) migrateDataNode(srcAddr, targetAddr string, limit int) (err er
wg.Add(1)
go func(dp *DataPartition) {
defer wg.Done()
if err1 := c.migrateDataPartition(srcNode.Addr, targetAddr, dp, dataNodeOfflineErr); err1 != nil {
if err1 := c.migrateDataPartition(srcNode.Addr, targetAddr, dp, raftForce, dataNodeOfflineErr); err1 != nil {
errChannel <- err1
}
}(toBeOffLinePartitions[i])
@ -1479,8 +1480,8 @@ func (c *Cluster) migrateDataNode(srcAddr, targetAddr string, limit int) (err er
return
}
func (c *Cluster) decommissionDataNode(dataNode *DataNode) (err error) {
return c.migrateDataNode(dataNode.Addr, "", 0)
func (c *Cluster) decommissionDataNode(dataNode *DataNode, force bool) (err error) {
return c.migrateDataNode(dataNode.Addr, "", 0, false)
}
func (c *Cluster) delDataNodeFromCache(dataNode *DataNode) {
@ -1597,8 +1598,7 @@ ERR:
// 4. synchronized create a new data partition
// 5. Set the data partition as readOnly.
// 6. persistent the new host list
func (c *Cluster) migrateDataPartition(srcAddr, targetAddr string, dp *DataPartition, errMsg string) (err error) {
func (c *Cluster) migrateDataPartition(srcAddr, targetAddr string, dp *DataPartition, raftForce bool, errMsg string) (err error) {
var (
targetHosts []string
newAddr string
@ -1700,7 +1700,7 @@ func (c *Cluster) migrateDataPartition(srcAddr, targetAddr string, dp *DataParti
}()
// if single replica wait for
if dp.isSpecialReplicaCnt() {
if dp.ReplicaNum == 1 || dp.ReplicaNum == 2 && !raftForce {
dp.Status = proto.ReadOnly
dp.isRecover = true
c.putBadDataPartitionIDs(replica, srcAddr, dp.PartitionID)
@ -1709,7 +1709,7 @@ func (c *Cluster) migrateDataPartition(srcAddr, targetAddr string, dp *DataParti
goto errHandler
}
} else {
if err = c.removeDataReplica(dp, srcAddr, false, false); err != nil {
if err = c.removeDataReplica(dp, srcAddr, false, raftForce); err != nil {
goto errHandler
}
if err = c.addDataReplica(dp, newAddr); err != nil {
@ -1760,8 +1760,8 @@ errHandler:
// 4. synchronized create a new data partition
// 5. Set the data partition as readOnly.
// 6. persistent the new host list
func (c *Cluster) decommissionDataPartition(offlineAddr string, dp *DataPartition, errMsg string) (err error) {
return c.migrateDataPartition(offlineAddr, "", dp, errMsg)
func (c *Cluster) decommissionDataPartition(offlineAddr string, dp *DataPartition, raftForce bool, errMsg string) (err error) {
return c.migrateDataPartition(offlineAddr, "", dp, raftForce, errMsg)
}
func (c *Cluster) validateDecommissionDataPartition(dp *DataPartition, offlineAddr string, raftForceDel bool) (err error) {
@ -1779,7 +1779,7 @@ func (c *Cluster) validateDecommissionDataPartition(dp *DataPartition, offlineAd
}
// if the partition can be offline or not
if err = dp.canBeOffLine(offlineAddr, raftForceDel); err != nil {
if err = dp.canBeOffLine(offlineAddr); err != nil {
log.LogInfof("action[validateDecommissionDataPartition] dp vol %v dp %v err %v", dp.VolName, dp.PartitionID, err)
return
}

16
master/data_node.go
View File

@ -145,6 +145,22 @@ func (dataNode *DataNode) isWriteAble() (ok bool) {
return
}
func (dataNode *DataNode) canAllocDp() bool {
if !dataNode.isWriteAble() {
return false
}
if dataNode.ToBeOffline {
return false
}
if !dataNode.dpCntInLimit() {
return false
}
return true
}
func (dataNode *DataNode) dpCntInLimit() bool {
return dataNode.DataPartitionCount <= dpCntOneNodeLimit()
}

27
master/data_partition.go
View File

@ -245,19 +245,10 @@ func (partition *DataPartition) hasMissingOneReplica(addr string, replicaNum int
return
}
func (partition *DataPartition) canBeOffLine(offlineAddr string, raftForceDel bool) (err error) {
func (partition *DataPartition) canBeOffLine(offlineAddr string) (err error) {
msg := fmt.Sprintf("action[canOffLine],partitionID:%v RocksDBHost:%v offLine:%v ",
partition.PartitionID, partition.Hosts, offlineAddr)
liveReplicas := partition.liveReplicas(defaultDataPartitionTimeOutSec)
if partition.isSpecialReplicaCnt() {
if len(liveReplicas) != 1 {
msg = fmt.Sprintf(msg+" err:%v liveReplicas:%v ", proto.ErrCannotBeOffLine, len(liveReplicas))
log.LogError(msg)
err = fmt.Errorf(msg)
return
}
return
}
otherLiveReplicas := make([]*DataReplica, 0)
for i := 0; i < len(liveReplicas); i++ {
replica := liveReplicas[i]
@ -270,26 +261,16 @@ func (partition *DataPartition) canBeOffLine(offlineAddr string, raftForceDel bo
msg = fmt.Sprintf(msg+" err:%v liveReplicas:%v ", proto.ErrCannotBeOffLine, len(liveReplicas))
log.LogError(msg)
err = fmt.Errorf(msg)
return
}
if partition.ReplicaNum == 1 && len(liveReplicas) == 0 {
if len(liveReplicas) == 0 {
msg = fmt.Sprintf(msg+" err:%v replicaNum:%v liveReplicas:%v ", proto.ErrCannotBeOffLine, partition.ReplicaNum, len(liveReplicas))
log.LogError(msg)
err = fmt.Errorf(msg)
return
}
if partition.ReplicaNum == 2 {
// raft quorum not take effect then do raftForceDel delete replica
if raftForceDel {
log.LogWarnf("action[canBeOffLine] raftForceDel delete addr %v, liveReplicas:%v ", offlineAddr, len(liveReplicas))
return
}
if len(otherLiveReplicas) == 0 {
msg = fmt.Sprintf(msg+" err:%v replicaNum:%v liveReplicas:%v ", proto.ErrCannotBeOffLine, partition.ReplicaNum, len(liveReplicas))
log.LogError(msg)
err = fmt.Errorf(msg)
}
}
return
}

6
master/disk_manager.go
View File

@ -97,13 +97,13 @@ func (c *Cluster) checkDiskRecoveryProgress() {
})
}
func (c *Cluster) decommissionDisk(dataNode *DataNode, badDiskPath string, badPartitions []*DataPartition) (err error) {
msg := fmt.Sprintf("action[decommissionDisk], node[%v] OffLine,disk[%v]", dataNode.Addr, badDiskPath)
func (c *Cluster) decommissionDisk(dataNode *DataNode, raftForce bool, badDiskPath string, badPartitions []*DataPartition) (err error) {
msg := fmt.Sprintf("action[decommissionDisk], Node[%v] OffLine,disk[%v]", dataNode.Addr, badDiskPath)
log.LogWarn(msg)
for _, dp := range badPartitions {
go func(dp *DataPartition) {
if err = c.decommissionDataPartition(dataNode.Addr, dp, diskOfflineErr); err != nil {
if err = c.decommissionDataPartition(dataNode.Addr, dp, raftForce, diskOfflineErr); err != nil {
return
}
}(dp)

13
master/gapi_cluster.go
View File

@ -5,10 +5,6 @@ import (
"context"
"encoding/json"
"fmt"
"github.com/cubefs/cubefs/proto"
"github.com/cubefs/cubefs/util/log"
"github.com/samsarahq/thunder/graphql"
"github.com/samsarahq/thunder/graphql/schemabuilder"
"io/ioutil"
"os"
"path/filepath"
@ -16,6 +12,11 @@ import (
"strings"
"sync"
"time"
"github.com/cubefs/cubefs/proto"
"github.com/cubefs/cubefs/util/log"
"github.com/samsarahq/thunder/graphql"
"github.com/samsarahq/thunder/graphql/schemabuilder"
)
type ClusterService struct {
@ -200,7 +201,7 @@ func (m *ClusterService) decommissionDisk(ctx context.Context, args struct {
}
rstMsg := fmt.Sprintf("receive decommissionDisk node[%v] disk[%v], badPartitionIds[%v] has offline successfully",
node.Addr, args.DiskPath, badPartitionIds)
if err = m.cluster.decommissionDisk(node, args.DiskPath, badPartitions); err != nil {
if err = m.cluster.decommissionDisk(node, false, args.DiskPath, badPartitions); err != nil {
return nil, err
}
Warn(m.cluster.Name, rstMsg)
@ -218,7 +219,7 @@ func (m *ClusterService) decommissionDataNode(ctx context.Context, args struct {
if err != nil {
return nil, err
}
if err := m.cluster.decommissionDataNode(node); err != nil {
if err := m.cluster.decommissionDataNode(node, false); err != nil {
return nil, err
}
rstMsg := fmt.Sprintf("decommission data node [%v] submited,please check laster!", args.OffLineAddr)

2
master/node_selector.go
View File

@ -162,7 +162,7 @@ func getAvailCarryDataNodeTab(maxTotal uint64, excludeHosts []string, dataNodes
log.LogDebugf("contains return")
return true
}
if !dataNode.isWriteAble() || dataNode.ToBeOffline || !dataNode.dpCntInLimit() {
if !dataNode.canAllocDp() {
log.LogInfof("dataNode [%v] is not writeable, offline %v, dpCnt %d", dataNode.Addr, dataNode.ToBeOffline, dataNode.DataPartitionCount)
log.LogDebugf("isWritable return")
return true