gian/cubefs
1
0
Fork 0
mirror of https://github.com/cubefs/cubefs.git synced 2026-08-02 10:06:14 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

feat(metasdkmaster): support read quoram when mp no leader. #22782472

Browse Source 
Signed-off-by: Victor1319 <zengxuewei@oppo.com>
This commit is contained in:
Victor1319 2024-11-14 11:42:25 +08:00 committed by AmazingChi
parent 447fcc1d61
commit 644508479e
20 changed files with 421 additions and 99 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
cli/cmd/fmt.go
View File

@ -254,6 +254,7 @@ func formatSimpleVolView(svv *proto.SimpleVolView) string {
sb.WriteString(fmt.Sprintf(" EnableAutoDpMetaRepair : %v\n", svv.EnableAutoDpMetaRepair))
sb.WriteString(fmt.Sprintf(" Quota : %v\n", formatEnabledDisabled(svv.EnableQuota)))
sb.WriteString(fmt.Sprintf(" AccessTimeValidInterval : %v\n", time.Duration(svv.AccessTimeInterval)*time.Second))
sb.WriteString(fmt.Sprintf(" MetaLeaderRetryTimeout : %v\n", time.Duration(svv.LeaderRetryTimeOut)*time.Second))
sb.WriteString(fmt.Sprintf(" EnablePersistAccessTime : %v\n", svv.EnablePersistAccessTime))
sb.WriteString(fmt.Sprintf(" ForbidWriteOpOfProtoVer0 : %v\n", svv.ForbidWriteOpOfProtoVer0))
if svv.Forbidden && svv.Status == 1 {

14
cli/cmd/vol.go
View File

@ -323,6 +323,7 @@ func newVolUpdateCmd(client *master.MasterClient) *cobra.Command {
var optTxOpLimitVal int
var optReplicaNum string
var optDeleteLockTime int64
var optLeaderRetryTime int64
var optEnableQuota string
var optEnableDpAutoMetaRepair string
var optTrashInterval int64
@ -478,6 +479,18 @@ func newVolUpdateCmd(client *master.MasterClient) *cobra.Command {
confirmString.WriteString(fmt.Sprintf(" DeleteLockTime : %v h\n", vv.DeleteLockTime))
}
if optLeaderRetryTime >= 0 {
if optLeaderRetryTime != vv.LeaderRetryTimeOut {
isChange = true
confirmString.WriteString(fmt.Sprintf(" LeaderRetryTimeout : %v s -> %v s\n", vv.LeaderRetryTimeOut, optLeaderRetryTime))
vv.LeaderRetryTimeOut = optLeaderRetryTime
} else {
confirmString.WriteString(fmt.Sprintf(" LeaderRetryTimeout : %v s\n", vv.LeaderRetryTimeOut))
}
} else {
confirmString.WriteString(fmt.Sprintf(" LeaderRetryTimeout : %v s\n", vv.LeaderRetryTimeOut))
}
// var maskStr string
if optTxMask != "" {
var oldMask, newMask proto.TxOpMask
@ -823,6 +836,7 @@ func newVolUpdateCmd(client *master.MasterClient) *cobra.Command {
cmd.Flags().StringVar(&optReplicaNum, CliFlagReplicaNum, "", "Specify data partition replicas number(default 3 for normal volume,1 for low volume)")
cmd.Flags().StringVar(&optEnableQuota, CliFlagEnableQuota, "", "Enable quota")
cmd.Flags().Int64Var(&optDeleteLockTime, CliFlagDeleteLockTime, -1, "Specify delete lock time[Unit: hour] for volume")
cmd.Flags().Int64Var(&optLeaderRetryTime, "leader-retry-timeout", -1, "Specify leader retry timeout for mp read [Unit: second] for volume, default 0")
cmd.Flags().StringVar(&clientIDKey, CliFlagClientIDKey, client.ClientIDKey(), CliUsageClientIDKey)
cmd.Flags().StringVar(&optEnableDpAutoMetaRepair, CliFlagAutoDpMetaRepair, "", "Enable or disable dp auto meta repair")
cmd.Flags().IntVar(&optVolStorageClass, CliFlagVolStorageClass, 0, "specify volStorageClass")

5
master/api_args_parse.go
View File

@ -400,6 +400,7 @@ type updateVolReq struct {
deleteLockTime int64
followerRead bool
metaFollowerRead bool
leaderRetryTimeout int64
authenticate bool
enablePosixAcl bool
enableTransaction proto.TxOpMask
@ -525,6 +526,10 @@ func parseVolUpdateReq(r *http.Request, vol *Vol, req *updateVolReq) (err error)
return
}
if req.leaderRetryTimeout, err = extractInt64WithDefault(r, proto.LeaderRetryTimeoutKey, vol.LeaderRetryTimeout); err != nil {
return
}
if req.enablePosixAcl, err = extractBoolWithDefault(r, enablePosixAclKey, vol.enablePosixAcl); err != nil {
return
}

37
master/api_service.go
View File

@ -2496,12 +2496,13 @@ func (m *Server) updateVol(w http.ResponseWriter, r *http.Request) {
newArgs.capacity = req.capacity
newArgs.deleteLockTime = req.deleteLockTime
newArgs.followerRead = req.followerRead
newArgs.metaFollowerRead = req.followerRead
newArgs.metaFollowerRead = req.metaFollowerRead
newArgs.authenticate = req.authenticate
newArgs.dpSelectorName = req.dpSelectorName
newArgs.dpSelectorParm = req.dpSelectorParm
newArgs.enablePosixAcl = req.enablePosixAcl
newArgs.enableTransaction = req.enableTransaction
newArgs.leaderRetryTimeout = req.leaderRetryTimeout
newArgs.txTimeout = req.txTimeout
newArgs.txConflictRetryNum = req.txConflictRetryNum
newArgs.txConflictRetryInterval = req.txConflictRetryInterval
@ -2520,7 +2521,8 @@ func (m *Server) updateVol(w http.ResponseWriter, r *http.Request) {
newArgs.volStorageClass = req.volStorageClass
newArgs.forbidWriteOpOfProtoVer0 = req.forbidWriteOpOfProtoVer0
log.LogWarnf("[updateVolOut] name [%s], z1 [%s], z2[%s] replicaNum[%v]", req.name, req.zoneName, vol.Name, req.replicaNum)
log.LogWarnf("[updateVolOut] name [%s], z1 [%s], z2[%s] replicaNum[%v], FR[%v], metaFR[%v]",
req.name, req.zoneName, vol.Name, req.replicaNum, req.followerRead, req.metaFollowerRead)
if err = m.cluster.updateVol(req.name, req.authKey, newArgs); err != nil {
sendErrReply(w, r, newErrHTTPReply(err))
return
@ -2950,6 +2952,7 @@ func (m *Server) createVol(w http.ResponseWriter, r *http.Request) {
}
if vol, err = m.cluster.createVol(req); err != nil {
sendErrReply(w, r, newErrHTTPReply(err))
return
}
@ -3062,19 +3065,21 @@ func newSimpleView(vol *Vol) (view *proto.SimpleVolView) {
}
view = &proto.SimpleVolView{
ID: vol.ID,
Name: vol.Name,
Owner: vol.Owner,
ZoneName: vol.zoneName,
DpReplicaNum: vol.dpReplicaNum,
MpReplicaNum: vol.mpReplicaNum,
InodeCount: volInodeCount,
DentryCount: volDentryCount,
MaxMetaPartitionID: maxPartitionID,
Status: vol.Status,
Capacity: vol.Capacity,
FollowerRead: vol.FollowerRead,
MetaFollowerRead: vol.MetaFollowerRead,
ID: vol.ID,
Name: vol.Name,
Owner: vol.Owner,
ZoneName: vol.zoneName,
DpReplicaNum: vol.dpReplicaNum,
MpReplicaNum: vol.mpReplicaNum,
InodeCount: volInodeCount,
DentryCount: volDentryCount,
MaxMetaPartitionID: maxPartitionID,
Status: vol.Status,
Capacity: vol.Capacity,
FollowerRead: vol.FollowerRead,
MetaFollowerRead: vol.MetaFollowerRead,
LeaderRetryTimeOut: vol.LeaderRetryTimeout,
EnablePosixAcl: vol.enablePosixAcl,
EnableQuota: vol.enableQuota,
EnableTransactionV1: proto.GetMaskString(vol.enableTransaction),
@ -5656,6 +5661,8 @@ func volStat(vol *Vol, countByMeta bool) (stat *proto.VolStatInfo) {
stat.StatByStorageClass = vol.StatByStorageClass
stat.StatMigrateStorageClass = vol.StatMigrateStorageClass
stat.StatByDpMediaType = vol.StatByDpMediaType
stat.MetaFollowerRead = vol.MetaFollowerRead
stat.LeaderRetryTimeOut = int(vol.LeaderRetryTimeout)
log.LogDebugf("[volStat] vol[%v] total[%v],usedSize[%v] TrashInterval[%v] DefaultStorageClass[%v]",
vol.Name, stat.TotalSize, stat.UsedSize, stat.TrashInterval, stat.DefaultStorageClass)

2
master/cluster.go
View File

@ -876,7 +876,7 @@ func (c *Cluster) checkMetaNodeHeartbeat() {
defer c.volMutex.RUnlock()
for _, vol := range c.vols {
if vol.MetaFollowerRead {
if vol.FollowerRead {
hbReq.FLReadVols = append(hbReq.FLReadVols, vol.Name)
}
if vol.DisableAuditLog {

28
master/metadata_fsm_op.go
View File

@ -292,19 +292,20 @@ type volValue struct {
DeleteExecTime time.Time
User *User
CrossZone bool
DomainOn bool
ZoneName string
OSSAccessKey string
OSSSecretKey string
CreateTime int64
DeleteLockTime int64
Description string
DpSelectorName string
DpSelectorParm string
DefaultPriority bool
DomainId uint64
VolType int
CrossZone bool
DomainOn bool
ZoneName string
OSSAccessKey string
OSSSecretKey string
CreateTime int64
DeleteLockTime int64
LeaderRetryTimeOut int64
Description string
DpSelectorName string
DpSelectorParm string
DefaultPriority bool
DomainId uint64
VolType int
EbsBlkSize int
CacheCapacity uint64
@ -368,6 +369,7 @@ func newVolValue(vol *Vol) (vv *volValue) {
Owner: vol.Owner,
FollowerRead: vol.FollowerRead,
MetaFollowerRead: vol.MetaFollowerRead,
LeaderRetryTimeOut: vol.LeaderRetryTimeout,
Authenticate: vol.authenticate,
CrossZone: vol.crossZone,
DomainOn: vol.domainOn,

5
master/vol.go
View File

@ -59,6 +59,7 @@ type VolVarargs struct {
enableAutoDpMetaRepair bool
accessTimeValidInterval int64
enablePersistAccessTime bool
leaderRetryTimeout int64
volStorageClass uint32
allowedStorageClass []uint32
forbidWriteOpOfProtoVer0 bool
@ -147,6 +148,7 @@ type Vol struct {
AccessTimeInterval int64
EnablePersistAccessTime bool
AccessTimeValidInterval int64
LeaderRetryTimeout int64 // s
EnableAutoMetaRepair atomicutil.Bool
ForbidWriteOpOfProtoVer0 atomicutil.Bool
@ -188,6 +190,7 @@ func newVol(vv volValue) (vol *Vol) {
vol.Capacity = vv.Capacity
vol.FollowerRead = vv.FollowerRead
vol.MetaFollowerRead = vv.MetaFollowerRead
vol.LeaderRetryTimeout = vv.LeaderRetryTimeOut
vol.authenticate = vv.Authenticate
vol.crossZone = vv.CrossZone
vol.zoneName = vv.ZoneName
@ -1819,6 +1822,7 @@ func setVolFromArgs(args *VolVarargs, vol *Vol) {
vol.txOpLimit = args.txOpLimit
vol.dpReplicaNum = args.dpReplicaNum
vol.crossZone = args.crossZone
vol.LeaderRetryTimeout = args.leaderRetryTimeout
if proto.IsVolSupportStorageClass(args.allowedStorageClass, proto.StorageClass_BlobStore) {
vol.EbsBlkSize = args.coldArgs.objBlockSize
@ -1886,6 +1890,7 @@ func getVolVarargs(vol *Vol) *VolVarargs {
deleteLockTime: vol.DeleteLockTime,
followerRead: vol.FollowerRead,
metaFollowerRead: vol.MetaFollowerRead,
leaderRetryTimeout: vol.LeaderRetryTimeout,
authenticate: vol.authenticate,
dpSelectorName: vol.dpSelectorName,
dpSelectorParm: vol.dpSelectorParm,

2
metanode/manager.go
View File

@ -376,6 +376,8 @@ func (m *metadataManager) HandleMetadataOperation(conn net.Conn, p *Packet, remo
err = m.opQuotaCreateDentry(conn, p, remoteAddr)
case proto.OpMetaGetUniqID:
err = m.opMetaGetUniqID(conn, p, remoteAddr)
case proto.OpMetaGetAppliedID:
err = m.opMetaGetAppliedID(conn, p, remoteAddr)
case proto.OpMetaInodeAccessTimeGet:
err = m.opMetaInodeAccessTimeGet(conn, p, remoteAddr)
// multi version

61
metanode/manager_op.go
View File

@ -16,6 +16,7 @@ package metanode
import (
"bytes"
"encoding/binary"
"encoding/json"
"fmt"
"net"
@ -979,7 +980,7 @@ func (m *metadataManager) opReadDirOnly(conn net.Conn, p *Packet,
err = errors.NewErrorf("[%v],req[%v],err[%v]", p.GetOpMsgWithReqAndResult(), req, string(p.Data))
return
}
if !mp.IsFollowerRead() && !m.serveProxy(conn, mp, p) {
if !m.serveProxy(conn, mp, p) {
return
}
err = mp.ReadDirOnly(req, p)
@ -1007,7 +1008,7 @@ func (m *metadataManager) opReadDir(conn net.Conn, p *Packet,
err = errors.NewErrorf("[%v],req[%v],err[%v]", p.GetOpMsgWithReqAndResult(), req, string(p.Data))
return
}
if !mp.IsFollowerRead() && !m.serveProxy(conn, mp, p) {
if !m.serveProxy(conn, mp, p) {
return
}
err = mp.ReadDir(req, p)
@ -1035,7 +1036,7 @@ func (m *metadataManager) opReadDirLimit(conn net.Conn, p *Packet,
err = errors.NewErrorf("[%v],req[%v],err[%v]", p.GetOpMsgWithReqAndResult(), req, string(p.Data))
return
}
if !mp.IsFollowerRead() && !m.serveProxy(conn, mp, p) {
if !m.serveProxy(conn, mp, p) {
return
}
err = mp.ReadDirLimit(req, p)
@ -1061,7 +1062,7 @@ func (m *metadataManager) opMetaInodeGet(conn net.Conn, p *Packet, remoteAddr st
err = errors.NewErrorf("getPartition [%v],req[%v],err[%v]", p.GetOpMsgWithReqAndResult(), req, string(p.Data))
return
}
if !mp.IsFollowerRead() && !m.serveProxy(conn, mp, p) {
if !m.serveProxy(conn, mp, p) {
return
}
if err = mp.InodeGet(req, p); err != nil {
@ -1222,7 +1223,7 @@ func (m *metadataManager) opMetaLookup(conn net.Conn, p *Packet,
return
}
if !mp.IsFollowerRead() && !m.serveProxy(conn, mp, p) {
if !m.serveProxy(conn, mp, p) {
return
}
err = mp.Lookup(req, p)
@ -1327,7 +1328,7 @@ func (m *metadataManager) opMetaExtentsList(conn net.Conn, p *Packet,
err = errors.NewErrorf("[%v] req: %v, resp: %v", p.GetOpMsgWithReqAndResult(), req, err.Error())
return
}
if !mp.IsFollowerRead() && !m.serveProxy(conn, mp, p) {
if !m.serveProxy(conn, mp, p) {
return
}
@ -1357,7 +1358,7 @@ func (m *metadataManager) opMetaObjExtentsList(conn net.Conn, p *Packet,
err = errors.NewErrorf("[%v] req: %v, resp: %v", p.GetOpMsgWithReqAndResult(), req, err.Error())
return
}
if !mp.IsFollowerRead() && !m.serveProxy(conn, mp, p) {
if !m.serveProxy(conn, mp, p) {
return
}
@ -1808,7 +1809,7 @@ func (m *metadataManager) opMetaBatchInodeGet(conn net.Conn, p *Packet,
err = errors.NewErrorf("[%v] req: %v, resp: %v", p.GetOpMsgWithReqAndResult(), req, err.Error())
return
}
if !mp.IsFollowerRead() && !m.serveProxy(conn, mp, p) {
if !m.serveProxy(conn, mp, p) {
return
}
err = mp.InodeGetBatch(req, p)
@ -2002,7 +2003,7 @@ func (m *metadataManager) opMetaGetXAttr(conn net.Conn, p *Packet, remoteAddr st
err = errors.NewErrorf("[%v] req: %v, resp: %v", p.GetOpMsgWithReqAndResult(), req, err.Error())
return
}
if !mp.IsFollowerRead() && !m.serveProxy(conn, mp, p) {
if !m.serveProxy(conn, mp, p) {
return
}
err = mp.GetXAttr(req, p)
@ -2027,7 +2028,7 @@ func (m *metadataManager) opMetaLockDir(conn net.Conn, p *Packet, remoteAddr str
err = errors.NewErrorf("[%v] req: %v, resp: %v", p.GetOpMsgWithReqAndResult(), req, err.Error())
return
}
if !mp.IsFollowerRead() && !m.serveProxy(conn, mp, p) {
if !m.serveProxy(conn, mp, p) {
return
}
err = mp.LockDir(req, p)
@ -2077,7 +2078,7 @@ func (m *metadataManager) opMetaBatchGetXAttr(conn net.Conn, p *Packet, remoteAd
err = errors.NewErrorf("[%v] req: %v, resp: %v", p.GetOpMsgWithReqAndResult(), req, err.Error())
return
}
if !mp.IsFollowerRead() && !m.serveProxy(conn, mp, p) {
if !m.serveProxy(conn, mp, p) {
return
}
err = mp.BatchGetXAttr(req, p)
@ -2133,7 +2134,7 @@ func (m *metadataManager) opMetaListXAttr(conn net.Conn, p *Packet, remoteAddr s
err = errors.NewErrorf("[%v] req: %v, resp: %v", p.GetOpMsgWithReqAndResult(), req, err.Error())
return
}
if !mp.IsFollowerRead() && !m.serveProxy(conn, mp, p) {
if !m.serveProxy(conn, mp, p) {
return
}
err = mp.ListXAttr(req, p)
@ -2286,7 +2287,7 @@ func (m *metadataManager) opGetMultipart(conn net.Conn, p *Packet, remote string
err = errors.NewErrorf("[opGetMultipart] req: %v, resp: %v", req, err.Error())
return
}
if !mp.IsFollowerRead() && !m.serveProxy(conn, mp, p) {
if !m.serveProxy(conn, mp, p) {
return
}
err = mp.GetMultipart(req, p)
@ -2307,7 +2308,7 @@ func (m *metadataManager) opAppendMultipart(conn net.Conn, p *Packet, remote str
m.respondToClientWithVer(conn, p)
return
}
if !mp.IsFollowerRead() && !m.serveProxy(conn, mp, p) {
if !m.serveProxy(conn, mp, p) {
return
}
err = mp.AppendMultipart(req, p)
@ -2330,9 +2331,10 @@ func (m *metadataManager) opListMultipart(conn net.Conn, p *Packet, remoteAddr s
err = errors.NewErrorf("[opListMultipart] req: %v, resp: %v", req, err.Error())
return
}
if !mp.IsFollowerRead() && !m.serveProxy(conn, mp, p) {
if !m.serveProxy(conn, mp, p) {
return
}
err = mp.ListMultipart(req, p)
_ = m.respondToClient(conn, p)
return
@ -2477,6 +2479,33 @@ func (m *metadataManager) opMetaGetInodeQuota(conn net.Conn, p *Packet, remote s
return
}
func (m *metadataManager) opMetaGetAppliedID(conn net.Conn, p *Packet, remote string) (err error) {
req := &proto.GetAppliedIDRequest{}
if err = json.Unmarshal(p.Data, req); err != nil {
p.PacketErrorWithBody(proto.OpErr, ([]byte)(err.Error()))
m.respondToClient(conn, p)
err = errors.NewErrorf("[opMetaGetAppliedID] req: %v, resp: %v", req, err.Error())
return
}
mp, err := m.getPartition(req.PartitionId)
if err != nil {
p.PacketErrorWithBody(proto.OpErr, ([]byte)(err.Error()))
m.respondToClient(conn, p)
err = errors.NewErrorf("[opMetaGetAppliedID] req: %v, resp: %v", req, err.Error())
return
}
appliedID := mp.GetAppliedID()
buf := make([]byte, 8)
binary.BigEndian.PutUint64(buf, appliedID)
p.PacketOkWithBody(buf)
m.respondToClient(conn, p)
log.LogDebugf("%s [opMetaGetAppliedID] req: %d - %v, req args: %v, resp: %v, body: %v",
remote, p.GetReqID(), req, string(p.Arg), p.GetResultMsg(), appliedID)
return
}
func (m *metadataManager) opMetaGetUniqID(conn net.Conn, p *Packet,
remoteAddr string,
) (err error) {
@ -2851,7 +2880,7 @@ func (m *metadataManager) opMetaInodeAccessTimeGet(conn net.Conn, p *Packet,
err = errors.NewErrorf("[%v],req[%v],err[%v]", p.GetOpMsgWithReqAndResult(), req, string(p.Data))
return
}
if !mp.IsFollowerRead() && !m.serveProxy(conn, mp, p) {
if !m.serveProxy(conn, mp, p) {
return
}
if err = mp.InodeGetAccessTime(req, p); err != nil {

24
metanode/manager_proxy.go
View File

@ -107,10 +107,28 @@ func (m *metadataManager) serveProxy(conn net.Conn, mp MetaPartition,
return false
}
followerRead := func() bool {
if !p.IsReadMetaPkt() {
return false
}
if p.ProtoVersion == proto.PacketProtoVersion0 && mp.IsFollowerRead() {
return true
}
return p.IsFollowerReadMetaPkt()
}
if leaderAddr, ok = mp.IsLeader(); ok {
return
}
if leaderAddr == "" {
if followerRead() {
log.LogDebugf("read from follower: p(%v), arg(%v)", p, mp.GetBaseConfig().PartitionId)
return true
}
err = fmt.Errorf("mpId(%v) %v", mp.GetBaseConfig().PartitionId, ErrNoLeader)
p.PacketErrorWithBody(proto.OpAgain, []byte(err.Error()))
goto end
@ -141,7 +159,13 @@ func (m *metadataManager) serveProxy(conn net.Conn, mp MetaPartition,
reqID, reqOp, p.ReqID, p.Opcode)
}
m.connPool.PutConnect(mConn, NoClosedConnect)
end:
if p.Opcode != proto.OpOk && followerRead() {
log.LogWarnf("read from follower after try leader failed: p(%v)", p)
return true
}
m.respondToClient(conn, p)
if err != nil {
log.LogErrorf("[serveProxy]: req: %d - %v, %v, packet(%v)", p.GetReqID(),

60
metanode/partition.go
View File

@ -253,10 +253,11 @@ type OpPartition interface {
GetVolName() (volName string)
IsLeader() (leaderAddr string, isLeader bool)
LeaderTerm() (leaderID, term uint64)
GetCursor() uint64
GetAppliedID() uint64
GetUniqId() uint64
IsFollowerRead() bool
SetFollowerRead(bool)
GetCursor() uint64
GetUniqId() uint64
GetBaseConfig() MetaPartitionConfig
ResponseLoadMetaPartition(p *Packet) (err error)
PersistMetadata() (err error)
@ -594,6 +595,31 @@ type metaPartition struct {
syncAtimeCh chan uint64
}
// IsLeader returns the raft leader address and if the current meta partition is the leader.
func (mp *metaPartition) SetFollowerRead(fRead bool) {
if mp.raftPartition == nil {
return
}
mp.isFollowerRead = fRead
}
// IsLeader returns the raft leader address and if the current meta partition is the leader.
func (mp *metaPartition) IsFollowerRead() (ok bool) {
if mp.raftPartition == nil {
return false
}
if !mp.isFollowerRead {
return false
}
if mp.raftPartition.IsRestoring() {
return false
}
return true
}
func (mp *metaPartition) IsForbidden() bool {
return mp.config.Forbidden
}
@ -1048,31 +1074,6 @@ func (mp *metaPartition) GetVerSeq() uint64 {
return atomic.LoadUint64(&mp.verSeq)
}
// IsLeader returns the raft leader address and if the current meta partition is the leader.
func (mp *metaPartition) SetFollowerRead(fRead bool) {
if mp.raftPartition == nil {
return
}
mp.isFollowerRead = fRead
}
// IsLeader returns the raft leader address and if the current meta partition is the leader.
func (mp *metaPartition) IsFollowerRead() (ok bool) {
if mp.raftPartition == nil {
return false
}
if !mp.isFollowerRead {
return false
}
if mp.raftPartition.IsRestoring() {
return false
}
return true
}
// IsLeader returns the raft leader address and if the current meta partition is the leader.
func (mp *metaPartition) IsLeader() (leaderAddr string, ok bool) {
if mp.raftPartition == nil {
@ -1115,6 +1116,11 @@ func (mp *metaPartition) GetCursor() uint64 {
return atomic.LoadUint64(&mp.config.Cursor)
}
// GetAppliedID returns applied ID of raft
func (mp *metaPartition) GetAppliedID() uint64 {
return atomic.LoadUint64(&mp.applyID)
}
// GetUniqId returns the uniqid stored in the config.
func (mp *metaPartition) GetUniqId() uint64 {
return atomic.LoadUint64(&mp.config.UniqId)

5
proto/admin_proto.go
View File

@ -383,7 +383,8 @@ var GApiInfo map[string]string = map[string]string{
}
const (
MetaFollowerReadKey = "metaFollowerRead"
MetaFollowerReadKey = "metaFollowerRead"
LeaderRetryTimeoutKey = "leaderRetryTimeout"
)
// const TimeFormat = "2006-01-02 15:04:05"
@ -1187,6 +1188,7 @@ type SimpleVolView struct {
DomainOn bool
CreateTime string
DeleteLockTime int64
LeaderRetryTimeOut int64
EnableToken bool
EnablePosixAcl bool
EnableQuota bool
@ -1201,6 +1203,7 @@ type SimpleVolView struct {
DpSelectorParm string
DefaultZonePrior bool
DpReadOnlyWhenVolFull bool
LeaderRetryTimeout int64
VolType int
ObjBlockSize int

5
proto/fs_proto.go
View File

@ -1048,6 +1048,11 @@ type GetUniqIDResponse struct {
Start uint64 `json:"start"`
}
type GetAppliedIDRequest struct {
VolName string `json:"vol"`
PartitionId uint64 `json:"pid"`
}
type LockDirRequest struct {
VolName string `json:"vol"`
PartitionId uint64 `json:"pid"`

2
proto/model.go
View File

@ -292,6 +292,8 @@ type VolStatInfo struct {
TrashInterval int64 `json:"TrashIntervalV2"`
DefaultStorageClass uint32
CacheDpStorageClass uint32
MetaFollowerRead bool
LeaderRetryTimeOut int
StatByStorageClass []*StatOfStorageClass
StatMigrateStorageClass []*StatOfStorageClass
StatByDpMediaType []*StatOfStorageClass

29
proto/packet.go
View File

@ -43,7 +43,8 @@ func GenerateRequestID() int64 {
}
const (
AddrSplit = "/"
AddrSplit = "/"
FollowerReadFlag = 'F'
)
// Operations
@ -186,7 +187,8 @@ const (
OpMetaTxGet uint8 = 0xAB
// Operations: Client -> MetaNode.
OpMetaGetUniqID uint8 = 0xAC
OpMetaGetUniqID uint8 = 0xAC
OpMetaGetAppliedID uint8 = 0xAD
// Multi version snapshot
OpRandomWriteAppend uint8 = 0xB1
@ -414,8 +416,15 @@ func (p *Packet) GetCopy() *Packet {
}
func (p *Packet) String() string {
return fmt.Sprintf("ReqID(%v)Op(%v)PartitionID(%v)ResultCode(%v)ExID(%v)ExtOffset(%v)KernelOff(%v)Type(%v)VerSeq(%v)ProtoVer(%v)Size(%v)",
p.ReqID, p.GetOpMsg(), p.PartitionID, p.GetResultMsg(), p.ExtentID, p.ExtentOffset, p.KernelOffset, p.ExtentType, p.VerSeq, p.ProtoVersion, p.Size)
return fmt.Sprintf("ReqID(%v)Op(%v)PartitionID(%v)ResultCode(%v)ExID(%v)ExtOffset(%v)KernelOff(%v)Type(%v)VerSeq(%v)ProtoVer(%v)Size(%v)FollowerRead(%v)",
p.ReqID, p.GetOpMsg(), p.PartitionID, p.GetResultMsg(), p.ExtentID, p.ExtentOffset, p.KernelOffset, p.ExtentType, p.VerSeq, p.ProtoVersion, p.Size, p.IsFollowerReadMetaPkt())
}
func (p *Packet) IsFollowerReadMetaPkt() bool {
if p.ArgLen == 1 && p.Arg[0] == FollowerReadFlag {
return true
}
return false
}
// GetStoreType returns the store type.
@ -646,6 +655,8 @@ func (p *Packet) GetOpMsg() (m string) {
m = "OpMetaTxLinkInode"
case OpMetaTxGet:
m = "OpMetaTxGet"
case OpMetaGetAppliedID:
m = "OpMetaGetAppliedId"
case OpMetaBatchSetInodeQuota:
m = "OpMetaBatchSetInodeQuota"
case OpMetaBatchDeleteInodeQuota:
@ -1016,6 +1027,16 @@ func (p *Packet) IsReadOperation() bool {
p.Opcode == OpSnapshotExtentRepairRead
}
func (p *Packet) IsReadMetaPkt() bool {
if p.Opcode == OpMetaLookup || p.Opcode == OpMetaInodeGet || p.Opcode == OpMetaBatchInodeGet ||
p.Opcode == OpMetaReadDir || p.Opcode == OpMetaExtentsList || p.Opcode == OpGetMultipart ||
p.Opcode == OpMetaGetXAttr || p.Opcode == OpMetaListXAttr || p.Opcode == OpListMultiparts ||
p.Opcode == OpMetaBatchGetXAttr || p.Opcode == OpMetaObjExtentsList || p.Opcode == OpMetaReadDirLimit || p.Opcode == OpMetaGetInodeQuota {
return true
}
return false
}
// ReadFromConn reads the data from the given connection.
// Recognize the version bit and parse out version,
// to avoid version field rsp back , the rsp of random write from datanode with replace OpRandomWriteVer to OpRandomWriteVerRsp

1
sdk/master/api_admin.go
View File

@ -317,6 +317,7 @@ func (api *AdminAPI) UpdateVolume(
request.addParam("enablePersistAccessTime", strconv.FormatBool(vv.EnablePersistAccessTime))
request.addParam("volStorageClass", strconv.FormatUint(uint64(vv.VolStorageClass), 10))
request.addParam("forbidWriteOpOfProtoVersion0", strconv.FormatBool(vv.ForbidWriteOpOfProtoVer0))
request.addParam(proto.LeaderRetryTimeoutKey, strconv.FormatUint(uint64(vv.LeaderRetryTimeOut), 10))
if txMask != "" {
request.addParam("enableTxMask", txMask)

223
sdk/meta/conn.go
View File

@ -15,20 +15,25 @@
package meta
import (
"encoding/binary"
"fmt"
"net"
"sync"
"syscall"
"time"
"github.com/cubefs/cubefs/proto"
"github.com/cubefs/cubefs/util/errors"
"github.com/cubefs/cubefs/util/exporter"
"github.com/cubefs/cubefs/util/log"
"github.com/cubefs/cubefs/util/stat"
)
const (
SendRetryLimit = 200 // times
SendRetryInterval = 100 // ms
MaxRetryTime = 10 * 60
MinRetryTime = 20 // s
)
type MetaConn struct {
@ -57,7 +62,7 @@ func (mw *MetaWrapper) putConn(mc *MetaConn, err error) {
mw.conns.PutConnectEx(mc.conn, err)
}
func (mw *MetaWrapper) sendToMetaPartition(mp *MetaPartition, req *proto.Packet) (*proto.Packet, error) {
func (mw *MetaWrapper) sendToMetaPartitionLeader(mp *MetaPartition, req *proto.Packet, sendTimeLimit int) (*proto.Packet, error) {
var (
resp *proto.Packet
err error
@ -66,16 +71,11 @@ func (mw *MetaWrapper) sendToMetaPartition(mp *MetaPartition, req *proto.Packet)
start time.Time
lastSeq uint64
)
var sendTimeLimit int
if mw.metaSendTimeout < 20 {
sendTimeLimit = 20 * 1000 // ms
} else if mw.metaSendTimeout >= MaxRetryTime {
sendTimeLimit = MaxRetryTime * 1000
} else {
sendTimeLimit = int(mw.metaSendTimeout) * 1000 // ms
}
delta := (sendTimeLimit*2/SendRetryLimit - SendRetryInterval*2) / SendRetryLimit // ms
delta := SendRetryInterval
if sendTimeLimit > MinRetryTime*1000 {
delta = (sendTimeLimit*2/SendRetryLimit - SendRetryInterval) / SendRetryLimit // ms
}
log.LogDebugf("mw.metaSendTimeout: %v s, sendTimeLimit: %v ms, delta: %v ms, req %v", mw.metaSendTimeout, sendTimeLimit, delta, req)
req.ExtentType |= proto.PacketProtocolVersionFlag
@ -85,12 +85,12 @@ func (mw *MetaWrapper) sendToMetaPartition(mp *MetaPartition, req *proto.Packet)
addr = mp.LeaderAddr
if addr == "" {
err = errors.New(fmt.Sprintf("sendToMetaPartition: failed due to empty leader addr and goto retry, req(%v) mp(%v)", req, mp))
err = errors.New(fmt.Sprintf("sendToMetaPartitionLeader: failed due to empty leader addr and goto retry, req(%v) mp(%v)", req, mp))
goto retry
}
mc, err = mw.getConn(mp.PartitionID, addr)
if err != nil {
log.LogWarnf("sendToMetaPartition: getConn failed and goto retry, req(%v) mp(%v) addr(%v) err(%v)", req, mp, addr, err)
log.LogWarnf("sendToMetaPartitionLeader: getConn failed and goto retry, req(%v) mp(%v) addr(%v) err(%v)", req, mp, addr, err)
goto retry
}
if mw.Client != nil { // compatible lcNode not init Client
@ -117,10 +117,10 @@ sendWithList:
req.ExtentType |= proto.VersionListFlag
req.VerList = make([]*proto.VolVersionInfo, len(mw.Client.GetVerMgr().VerList))
copy(req.VerList, mw.Client.GetVerMgr().VerList)
log.LogWarnf("sendToMetaPartition: leader failed and goto retry, req(%v) mp(%v) mc(%v) err(%v) resp(%v)", req, mp, mc, err, resp)
log.LogWarnf("sendToMetaPartitionLeader: leader failed and goto retry, req(%v) mp(%v) mc(%v) err(%v) resp(%v)", req, mp, mc, err, resp)
goto sendWithList
}
log.LogWarnf("sendToMetaPartition: leader failed and goto retry, req(%v) mp(%v) mc(%v) err(%v), addr(%s)",
log.LogWarnf("sendToMetaPartitionLeader: leader failed and goto retry, req(%v) mp(%v) mc(%v) err(%v), addr(%s)",
req, mp, mc, err, addr)
mw.putConn(mc, err)
retry:
@ -130,7 +130,7 @@ retry:
mc, err = mw.getConn(mp.PartitionID, addr)
errs[j] = err
if err != nil {
log.LogWarnf("sendToMetaPartition: getConn failed and continue to retry, req(%v) mp(%v) addr(%v) err(%v)", req, mp, addr, err)
log.LogWarnf("sendToMetaPartitionLeader: getConn failed and continue to retry, req(%v) mp(%v) addr(%v) err(%v)", req, mp, addr, err)
continue
}
resp, err = mc.send(req)
@ -143,14 +143,14 @@ retry:
} else {
errs[j] = err
}
log.LogWarnf("sendToMetaPartition: retry failed req(%v) mp(%v) mc(%v) errs(%v) resp(%v)", req, mp, mc, errs, resp)
log.LogWarnf("sendToMetaPartitionLeader: retry failed req(%v) mp(%v) mc(%v) errs(%v) resp(%v)", req, mp, mc, errs, resp)
}
if time.Since(start) > time.Duration(sendTimeLimit)*time.Millisecond {
log.LogWarnf("sendToMetaPartition: retry timeout req(%v) mp(%v) time(%v)", req, mp, time.Since(start))
log.LogWarnf("sendToMetaPartitionLeader: retry timeout req(%v) mp(%v) time(%v)", req, mp, time.Since(start))
break
}
sendRetryInterval := time.Duration(SendRetryInterval+i*delta) * time.Millisecond
log.LogWarnf("sendToMetaPartition: req(%v) mp(%v) retry in (%v), retry_iteration (%v), retry_totalTime (%v)", req, mp,
log.LogWarnf("sendToMetaPartitionLeader: req(%v) mp(%v) retry in (%v), retry_iteration (%v), retry_totalTime (%v)", req, mp,
sendRetryInterval, i+1, time.Since(start))
time.Sleep(sendRetryInterval)
}
@ -166,6 +166,193 @@ out:
return resp, nil
}
func (mw *MetaWrapper) sendToMetaPartition(mp *MetaPartition, req *proto.Packet) (*proto.Packet, error) {
if req.IsReadMetaPkt() {
return mw.sendReadToMP(mp, req)
}
var sendTimeLimit int
if mw.metaSendTimeout < MinRetryTime {
sendTimeLimit = MinRetryTime * 1000
} else if mw.metaSendTimeout >= MaxRetryTime {
sendTimeLimit = MaxRetryTime * 1000
} else {
sendTimeLimit = int(mw.metaSendTimeout) * 1000 // ms
}
return mw.sendToMetaPartitionLeader(mp, req, sendTimeLimit)
}
func (mw *MetaWrapper) sendReadToMP(mp *MetaPartition, req *proto.Packet) (resp *proto.Packet, err error) {
leaderRetryTimeOut := mw.leaderRetryTimeout * 1000
if leaderRetryTimeOut <= 0 {
leaderRetryTimeOut = MinRetryTime * 1000
}
resp, err = mw.sendToMetaPartitionLeader(mp, req, int(leaderRetryTimeOut))
if err == nil && !resp.ShouldRetry() {
return
}
log.LogWarnf("sendReadToMP: send to leader failed, try to read quorum, req (%v), mp(%v), err(%v), resp(%v)", req, mp, err, resp)
return mw.readQuoramFromHosts(mp, req)
}
func (mw *MetaWrapper) readQuoramFromHosts(mp *MetaPartition, req *proto.Packet) (resp *proto.Packet, err error) {
var sendTimeLimit int
var mc *MetaConn
if mw.metaSendTimeout < MinRetryTime {
sendTimeLimit = MinRetryTime * 1000
} else if mw.metaSendTimeout >= MaxRetryTime {
sendTimeLimit = MaxRetryTime * 1000
} else {
sendTimeLimit = int(mw.metaSendTimeout) * 1000 // ms
}
delta := (sendTimeLimit*2/SendRetryLimit - SendRetryInterval*2) / SendRetryLimit // ms
start := time.Now()
for i := 0; i < SendRetryLimit; i++ {
activeHosts, quoramHosts := mw.getMpHosts(mp)
if len(activeHosts) < mp.Quoram() {
if !mw.FollowerRead {
log.LogWarnf("readQuoramFromHosts: activeHosts(%d) quoramHosts(%v) < quoram(%v), req(%v), mp(%v)",
len(activeHosts), len(quoramHosts), mp.Quoram(), req, mp.PartitionID)
goto wait
}
log.LogWarnf("readQuoramFromHosts: follower read is open, use follower to read, active(%v), quoram(%v) req(%v), mp (%d)",
activeHosts, quoramHosts, *req, mp.PartitionID)
}
req.ArgLen = 1
req.Arg = make([]byte, req.ArgLen)
req.Arg[0] = proto.FollowerReadFlag
for _, addr := range quoramHosts {
mc, err = mw.getConn(mp.PartitionID, addr)
if err != nil {
log.LogWarnf("readQuoramFromHosts: getConn failed and continue to retry, req(%v) mp(%v) addr(%v) err(%v)", req, mp, addr, err)
continue
}
resp, err = mc.send(req)
mw.putConn(mc, err)
if err == nil && !resp.ShouldRetry() {
goto out
}
log.LogWarnf("readQuoramFromHosts: retry failed req(%v) mp(%v) mc(%v) errs(%v) resp(%v)", req, mp, mc, err, resp)
}
wait:
if time.Since(start) > time.Duration(sendTimeLimit)*time.Millisecond {
log.LogWarnf("readQuoramFromHosts: retry timeout req(%v) mp(%v) time(%v)", req, mp, time.Since(start))
break
}
sendRetryInterval := time.Duration(SendRetryInterval+i*delta) * time.Millisecond
log.LogWarnf("readQuoramFromHosts: req(%v) mp(%v) retry in (%v), retry_iteration (%v), retry_totalTime (%v)", req, mp,
sendRetryInterval, i+1, time.Since(start))
time.Sleep(sendRetryInterval)
}
out:
log.LogDebugf("readQuoramFromHosts: succeed! req(%v) mc(%v) resp(%v)", req, mc, resp)
if err != nil || resp == nil {
return nil, errors.New(fmt.Sprintf("readQuoramFromHosts failed: req(%v) mp(%v) err(%v) resp(%v)", req, mp, err, resp))
}
return
}
func (mw *MetaWrapper) getMpHosts(mp *MetaPartition) (activeHosts, quoramHosts []string) {
ids := make(map[string]uint64, len(mp.Members))
var (
wg sync.WaitGroup
lock sync.Mutex
maxAppliedID uint64
)
for _, addr := range mp.Members {
wg.Add(1)
go func(curAddr string) {
defer wg.Done()
appliedID, err := mw.getAppliedID(mp, curAddr)
if err != nil {
return
}
lock.Lock()
ids[curAddr] = appliedID
lock.Unlock()
log.LogDebugf("getMpHosts: get apply id[%v] from host[%v], pid[%v]", appliedID, curAddr, mp.PartitionID)
}(addr)
}
wg.Wait()
maxID := uint64(0)
targetHosts := make([]string, 0)
for _, id := range ids {
if id >= maxID {
maxID = id
}
}
for addr, id := range ids {
if id == maxID {
targetHosts = append(targetHosts, addr)
}
activeHosts = append(activeHosts, addr)
}
log.LogDebugf("getTargetHosts: get max apply id[%v] from hosts[%v], pid[%v]", maxAppliedID, targetHosts, mp.PartitionID)
return activeHosts, targetHosts
}
func (mw *MetaWrapper) getAppliedID(mp *MetaPartition, addr string) (id uint64, err error) {
req := &proto.GetAppliedIDRequest{
PartitionId: mp.PartitionID,
}
packet := proto.NewPacketReqID()
packet.Opcode = proto.OpMetaGetAppliedID
err = packet.MarshalData(req)
if err != nil {
log.LogWarnf("getAppliedID err: (%v), req(%v)", err, *req)
return
}
bgTime := stat.BeginStat()
defer func() {
stat.EndStat("getAppliedID", err, bgTime, 1)
}()
metric := exporter.NewTPCnt(packet.GetOpMsg())
defer func() {
metric.SetWithLabels(err, map[string]string{exporter.Vol: mw.volname})
}()
mc, err := mw.getConn(mp.PartitionID, addr)
if err != nil {
log.LogWarnf("getAppliedID: getConn failed and continue to retry, req(%v) mp(%v) addr(%v) err(%v)", req, mp, addr, err)
return 0, err
}
resp, err := mc.send(packet)
mw.putConn(mc, err)
if err != nil || parseStatus(resp.ResultCode) != statusOK {
log.LogWarnf("getAppliedID: packet(%v) mp(%v) addr(%v) req(%v) result(%v), err(%v)", packet, mp, addr, *req, resp, err)
err = errors.New("getAppliedID error")
return
}
id = binary.BigEndian.Uint64(resp.Data)
log.LogDebugf("getAppliedID: get applied addr %s, mp %d, id %d, req %v", addr, mp.PartitionID, id, packet.ReqID)
return
}
func (mc *MetaConn) send(req *proto.Packet) (resp *proto.Packet, err error) {
req.ExtentType |= proto.PacketProtocolVersionFlag

2
sdk/meta/meta.go
View File

@ -158,6 +158,7 @@ type MetaWrapper struct {
singleflight singleflight.Group
EnableSummary bool
metaSendTimeout int64
leaderRetryTimeout int64 // s
DirChildrenNumLimit uint32
EnableTransaction proto.TxOpMask
TxTimeout int64
@ -190,6 +191,7 @@ type MetaWrapper struct {
DefaultStorageClass uint32
CacheDpStorageClass uint32
InnerReq bool
FollowerRead bool
}
type uniqidRange struct {

8
sdk/meta/partition.go
View File

@ -29,9 +29,13 @@ type MetaPartition struct {
Status int8
}
func (this *MetaPartition) Less(than btree.Item) bool {
func (mp *MetaPartition) Less(than btree.Item) bool {
that := than.(*MetaPartition)
return this.Start < that.Start
return mp.Start < that.Start
}
func (mp *MetaPartition) Quoram() int {
return len(mp.Members)/2 + 1
}
func (mp *MetaPartition) Copy() btree.Item {

6
sdk/meta/view.go
View File

@ -175,8 +175,10 @@ func (mw *MetaWrapper) updateVolStatInfo() (err error) {
atomic.StoreUint64(&mw.usedSize, info.UsedSize)
atomic.StoreUint64(&mw.inodeCount, info.InodeCount)
atomic.StoreUint32(&mw.DefaultStorageClass, info.DefaultStorageClass)
log.LogInfof("[updateVolStatInfo]: info(%+v), defaultStorageClass(%v)",
info, proto.StorageClassString(info.DefaultStorageClass))
mw.FollowerRead = info.MetaFollowerRead
mw.leaderRetryTimeout = int64(info.LeaderRetryTimeOut)
log.LogInfof("[updateVolStatInfo]: info(%+v), defaultStorageClass(%v), followerRead(%v), timout(%v)",
info, proto.StorageClassString(info.DefaultStorageClass), mw.FollowerRead, mw.leaderRetryTimeout)
// 0 means disable trash
if mw.disableTrashByClient {
log.LogDebugf("updateVolStatInfo: trash for %v is disabled by client", mw.volname)