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): Add a configuration item to determine if freezing before volume deletion.

Browse Source 
Signed-off-by: shuqiang-zheng <zhengshuqiang@oppo.com>
This commit is contained in:
shuqiang-zheng 2024-01-08 18:34:00 +08:00 committed by shuqiang-zheng
parent ae3b170fe6
commit dea6c736c5
7 changed files with 165 additions and 70 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

106
master/api_service.go
View File

@ -2052,6 +2052,21 @@ func (m *Server) markDeleteVol(w http.ResponseWriter, r *http.Request) {
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: err.Error()})
return
}
if enableDirectDeleteVol {
if err = m.cluster.markDeleteVol(name, authKey, false, true); err != nil {
sendErrReply(w, r, newErrHTTPReply(err))
return
}
if err = m.user.deleteVolPolicy(name); err != nil {
sendErrReply(w, r, newErrHTTPReply(err))
return
}
msg = fmt.Sprintf("delete vol[%v] successfully,from[%v]", name, r.RemoteAddr)
log.LogWarn(msg)
sendOkReply(w, r, newSuccessHTTPReply(msg))
return
}
vol, err := m.cluster.getVol(name)
if err != nil {
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeVolNotExists, Msg: err.Error()})
@ -2060,51 +2075,78 @@ func (m *Server) markDeleteVol(w http.ResponseWriter, r *http.Request) {
if status {
oldForbiden := vol.Forbidden
vol.Forbidden = true
vol.authKey = authKey
vol.deleteExecTime = time.Now().Add(time.Duration(m.config.volDelayDeleteTime) * time.Hour)
vol.user = m.user
defer func() {
if err != nil {
vol.Forbidden = oldForbiden
vol.authKey = ""
vol.deleteExecTime = time.Time{}
vol.user = nil
}
}()
if err = m.cluster.syncUpdateVol(vol); err != nil {
if err = m.cluster.markDeleteVol(name, authKey, false, true); err != nil {
sendErrReply(w, r, newErrHTTPReply(err))
return
}
vol.setDpForbid()
vol.setMpForbid()
m.cluster.delayDeleteVolsInfo = append(m.cluster.delayDeleteVolsInfo, &delayDeleteVolInfo{volName: name, authKey: authKey, execTime: time.Now().Add(time.Duration(m.config.volDelayDeleteTime) * time.Hour), user: m.user})
log.LogDebugf("delete vol[%v], slice[%v]", name, m.cluster.delayDeleteVolsInfo)
m.cluster.deleteVolMutex.Lock()
m.cluster.delayDeleteVolsInfo = append(m.cluster.delayDeleteVolsInfo, &delayDeleteVolInfo{volName: name, authKey: authKey, execTime: vol.deleteExecTime, user: m.user})
m.cluster.deleteVolMutex.Unlock()
msg = fmt.Sprintf("delete vol: forbid vol[%v] successfully,from[%v]", name, r.RemoteAddr)
log.LogWarn(msg)
} else {
var index int
var value *delayDeleteVolInfo
for index, value = range m.cluster.delayDeleteVolsInfo {
if value.volName == name {
break
}
}
if index == len(m.cluster.delayDeleteVolsInfo)-1 && value.volName != name {
msg := fmt.Sprintf("vol[%v] was not previously deleted", name)
err = errors.New(msg)
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeVolNotDelete, Msg: err.Error()})
return
}
m.cluster.delayDeleteVolsInfo = append(m.cluster.delayDeleteVolsInfo[:index], m.cluster.delayDeleteVolsInfo[index+1:]...)
log.LogDebugf("undelete vol[%v], slice[%v]", name, m.cluster.delayDeleteVolsInfo)
oldForbiden := vol.Forbidden
vol.Forbidden = false
defer func() {
if err != nil {
vol.Forbidden = oldForbiden
}
}()
if err = m.cluster.syncUpdateVol(vol); err != nil {
sendErrReply(w, r, newErrHTTPReply(err))
return
}
msg = fmt.Sprintf("undelete vol: unforbid vol[%v] successfully,from[%v]", name, r.RemoteAddr)
log.LogWarn(msg)
sendOkReply(w, r, newSuccessHTTPReply(msg))
return
}
var index int
var value *delayDeleteVolInfo
if len(m.cluster.delayDeleteVolsInfo) == 0 {
msg := fmt.Sprintf("vol[%v] was not previously deleted or already deleted", name)
err = errors.New(msg)
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeVolNotDelete, Msg: err.Error()})
return
}
m.cluster.deleteVolMutex.RLock()
for index, value = range m.cluster.delayDeleteVolsInfo {
if value.volName == name {
break
}
}
m.cluster.deleteVolMutex.RUnlock()
if index == len(m.cluster.delayDeleteVolsInfo)-1 && value.volName != name {
msg := fmt.Sprintf("vol[%v] was not previously deleted or already deleted", name)
err = errors.New(msg)
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeVolNotDelete, Msg: err.Error()})
return
}
oldForbiden := vol.Forbidden
oldAuthKey := vol.authKey
oldDeleteExecTime := vol.deleteExecTime
oldUser := vol.user
vol.Forbidden = false
vol.authKey = ""
vol.deleteExecTime = time.Time{}
vol.user = nil
defer func() {
if err != nil {
vol.Forbidden = oldForbiden
vol.authKey = oldAuthKey
vol.deleteExecTime = oldDeleteExecTime
vol.user = oldUser
}
}()
if err = m.cluster.markDeleteVol(name, authKey, false, false); err != nil {
sendErrReply(w, r, newErrHTTPReply(err))
return
}
m.cluster.deleteVolMutex.Lock()
m.cluster.delayDeleteVolsInfo = append(m.cluster.delayDeleteVolsInfo[:index], m.cluster.delayDeleteVolsInfo[index+1:]...)
m.cluster.deleteVolMutex.Unlock()
msg = fmt.Sprintf("undelete vol: unforbid vol[%v] successfully,from[%v]", name, r.RemoteAddr)
log.LogWarn(msg)
sendOkReply(w, r, newSuccessHTTPReply(msg))
}

12
master/api_service_test.go
View File

@ -459,8 +459,14 @@ func TestMarkDeleteVol(t *testing.T) {
reqURL := fmt.Sprintf("%v%v?name=%v&authKey=%v", hostAddr, proto.AdminDeleteVol, name, buildAuthKey(testOwner))
process(reqURL, t)
if len(server.cluster.delayDeleteVolsInfo) != 1 || server.cluster.delayDeleteVolsInfo[0].volName != name {
t.Errorf("expect vol %v in delayDeleteVolsInfo, but is not exist", name)
userInfo, err := server.user.getUserInfo("cfs")
if err != nil {
t.Error(err)
return
}
if contains(userInfo.Policy.OwnVols, name) {
t.Errorf("expect no vol %v in own vols, but is exist", name)
return
}
}
@ -596,7 +602,7 @@ func delVol(name string, t *testing.T) {
vol, err := server.cluster.getVol(name)
assert.True(t, err == nil)
assert.True(t, vol.Forbidden == true)
assert.True(t, vol.Status == markDelete)
}
func setVolCapacity(capacity uint64, url string, t *testing.T) {

27
master/cluster.go
View File

@ -51,6 +51,7 @@ type Cluster struct {
metaNodes sync.Map
volMutex sync.RWMutex // volume mutex
createVolMutex sync.RWMutex // create volume mutex
deleteVolMutex sync.RWMutex //delete volume mutex
mnMutex sync.RWMutex // meta node mutex
dnMutex sync.RWMutex // data node mutex
nsMutex sync.RWMutex // nodeset mutex
@ -479,16 +480,13 @@ func (c *Cluster) scheduleToCheckDelayDeleteVols() {
if len(c.delayDeleteVolsInfo) == 0 {
continue
}
c.deleteVolMutex.Lock()
currentDeleteVol := c.delayDeleteVolsInfo[0]
if currentDeleteVol.execTime.Sub(time.Now()) > 0 {
c.deleteVolMutex.Unlock()
continue
}
go func() {
if err := c.markDeleteVol(currentDeleteVol.volName, currentDeleteVol.authKey, false); err != nil {
msg := fmt.Sprintf("delete vol[%v] failed: err:[%v]", currentDeleteVol.volName, err)
log.LogError(msg)
return
}
if err := currentDeleteVol.user.deleteVolPolicy(currentDeleteVol.volName); err != nil {
msg := fmt.Sprintf("delete vol[%v] failed: err:[%v]", currentDeleteVol.volName, err)
log.LogError(msg)
@ -499,9 +497,11 @@ func (c *Cluster) scheduleToCheckDelayDeleteVols() {
}()
if len(c.delayDeleteVolsInfo) == 1 {
c.delayDeleteVolsInfo = make([]*delayDeleteVolInfo, 0)
c.deleteVolMutex.Unlock()
continue
}
c.delayDeleteVolsInfo = c.delayDeleteVolsInfo[1:]
c.deleteVolMutex.Unlock()
case <-c.stopc:
ticker.Stop()
return
@ -1371,7 +1371,7 @@ func (c *Cluster) deleteVol(name string) {
return
}
func (c *Cluster) markDeleteVol(name, authKey string, force bool) (err error) {
func (c *Cluster) markDeleteVol(name, authKey string, force bool, isNotCancel bool) (err error) {
var (
vol *Vol
serverAuthKey string
@ -1382,6 +1382,20 @@ func (c *Cluster) markDeleteVol(name, authKey string, force bool) (err error) {
return proto.ErrVolNotExists
}
if !isNotCancel {
serverAuthKey = vol.Owner
if !matchKey(serverAuthKey, authKey) {
return proto.ErrVolAuthKeyNotMatch
}
vol.Status = proto.VolStatusNormal
if err = c.syncUpdateVol(vol); err != nil {
vol.Status = proto.VolStatusMarkDelete
return proto.ErrPersistenceByRaft
}
return
}
if !c.cfg.volForceDeletion {
volDentryCount := uint64(0)
mpsCopy := vol.cloneMetaPartitionMap()
@ -1417,7 +1431,6 @@ func (c *Cluster) markDeleteVol(name, authKey string, force bool) (err error) {
vol.Status = proto.VolStatusNormal
return proto.ErrPersistenceByRaft
}
return
}

2
master/gapi_volume.go
View File

@ -266,7 +266,7 @@ func (s *VolumeService) markDeleteVol(ctx context.Context, args struct {
return nil, err
}
if err = s.cluster.markDeleteVol(args.Name, args.AuthKey, false); err != nil {
if err = s.cluster.markDeleteVol(args.Name, args.AuthKey, false, true); err != nil {
return nil, err
}

11
master/metadata_fsm_op.go
View File

@ -250,6 +250,10 @@ type volValue struct {
Authenticate bool
DpReadOnlyWhenVolFull bool
AuthKey string
DeleteExecTime time.Time
User *User
CrossZone bool
DomainOn bool
ZoneName string
@ -353,6 +357,9 @@ func newVolValue(vol *Vol) (vv *volValue) {
DpReadOnlyWhenVolFull: vol.DpReadOnlyWhenVolFull,
Forbidden: vol.Forbidden,
EnableAuditLog: vol.EnableAuditLog,
AuthKey: vol.authKey,
DeleteExecTime: vol.deleteExecTime,
User: vol.user,
}
return
@ -1478,6 +1485,10 @@ func (c *Cluster) loadVols() (err error) {
c.putVol(vol)
log.LogInfof("action[loadVols],vol[%v]", vol.Name)
if vol.Forbidden && vol.Status == bsProto.VolStatusMarkDelete {
c.delayDeleteVolsInfo = append(c.delayDeleteVolsInfo, &delayDeleteVolInfo{volName: vol.Name, authKey: vol.authKey, execTime: vol.deleteExecTime, user: vol.user})
log.LogInfof("action[loadDelayDeleteVols],vol[%v]", vol.Name)
}
}
return
}

57
master/server.go
View File

@ -38,31 +38,32 @@ import (
// configuration keys
const (
ClusterName = "clusterName"
ID = "id"
IP = "ip"
Port = "port"
LogLevel = "logLevel"
LogDir = "logDir"
WalDir = "walDir"
StoreDir = "storeDir"
EbsAddrKey = "ebsAddr"
BStoreAddrKey = "bStoreAddr"
EbsServicePathKey = "ebsServicePath"
BStoreServicePathKey = "bStoreServicePath"
GroupID = 1
ModuleName = "master"
CfgRetainLogs = "retainLogs"
DefaultRetainLogs = 20000
cfgTickInterval = "tickInterval"
cfgRaftRecvBufSize = "raftRecvBufSize"
cfgElectionTick = "electionTick"
SecretKey = "masterServiceKey"
Stat = "stat"
Authenticate = "authenticate"
AuthNodeHost = "authNodeHost"
AuthNodeEnableHTTPS = "authNodeEnableHTTPS"
AuthNodeCertFile = "authNodeCertFile"
ClusterName = "clusterName"
ID = "id"
IP = "ip"
Port = "port"
LogLevel = "logLevel"
LogDir = "logDir"
WalDir = "walDir"
StoreDir = "storeDir"
EbsAddrKey = "ebsAddr"
BStoreAddrKey = "bStoreAddr"
EbsServicePathKey = "ebsServicePath"
BStoreServicePathKey = "bStoreServicePath"
GroupID = 1
ModuleName = "master"
CfgRetainLogs = "retainLogs"
DefaultRetainLogs = 20000
cfgTickInterval = "tickInterval"
cfgRaftRecvBufSize = "raftRecvBufSize"
cfgElectionTick = "electionTick"
SecretKey = "masterServiceKey"
cfgEnableDirectDeleteVol = "enableDirectDeleteVol"
Stat = "stat"
Authenticate = "authenticate"
AuthNodeHost = "authNodeHost"
AuthNodeEnableHTTPS = "authNodeEnableHTTPS"
AuthNodeCertFile = "authNodeCertFile"
)
var (
@ -70,8 +71,9 @@ var (
volNameRegexp = regexp.MustCompile("^[a-zA-Z0-9][a-zA-Z0-9_.-]{1,61}[a-zA-Z0-9]$")
ownerRegexp = regexp.MustCompile("^[A-Za-z][A-Za-z0-9_]{0,20}$")
useConnPool = true // for test
gConfig *clusterConfig
useConnPool = true // for test
enableDirectDeleteVol = true
gConfig *clusterConfig
)
var overSoldFactor = defaultOverSoldFactor
@ -373,6 +375,7 @@ func (m *Server) checkConfig(cfg *config.Config) (err error) {
m.config.volDeletionDentryThreshold = uint64(threshold)
m.config.volDelayDeleteTime = cfg.GetInt64WithDefault(cfgVolDeletionDelayTime, 48)
enableDirectDeleteVol = cfg.GetBoolWithDefault(cfgEnableDirectDeleteVol, true)
return
}

20
master/vol.go
View File

@ -115,6 +115,9 @@ type Vol struct {
mpsLock *mpsLockManager
EnableAuditLog bool
preloadCapacity uint64
authKey string
deleteExecTime time.Time
user *User
}
func newVol(vv volValue) (vol *Vol) {
@ -206,6 +209,9 @@ func newVolFromVolValue(vv *volValue) (vol *Vol) {
}
vol.Forbidden = vv.Forbidden
vol.EnableAuditLog = vv.EnableAuditLog
vol.authKey = vv.AuthKey
vol.deleteExecTime = vv.DeleteExecTime
vol.user = vv.User
return vol
}
@ -1194,6 +1200,20 @@ func (vol *Vol) checkStatus(c *Cluster) {
if vol.Status != proto.VolStatusMarkDelete {
return
}
if vol.Forbidden && len(c.delayDeleteVolsInfo) != 0 {
var value *delayDeleteVolInfo
c.deleteVolMutex.RLock()
for _, value = range c.delayDeleteVolsInfo {
if value.volName == vol.Name {
break
}
}
c.deleteVolMutex.RUnlock()
if value.volName == vol.Name {
return
}
}
log.LogInfof("action[volCheckStatus] vol[%v],status[%v]", vol.Name, vol.Status)
metaTasks := vol.getTasksToDeleteMetaPartitions()
dataTasks := vol.getTasksToDeleteDataPartitions()