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 getting clean task for empty meta partition.#22834621

Browse Source 
Signed-off-by: Wu Huocheng <wuhuocheng@oppo.com>
This commit is contained in:
Wu Huocheng 2024-12-24 17:39:37 +08:00 committed by zhumingze1108
parent 799173432e
commit d8bdc8d44f
5 changed files with 205 additions and 86 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

55
master/api_service.go
View File

@ -8582,12 +8582,16 @@ func (m *Server) freezeEmptyMetaPartition(w http.ResponseWriter, r *http.Request
break
}
}
go m.cluster.FreezeEmptyMetaPartitionJob(name, freezeList)
err = m.cluster.FreezeEmptyMetaPartitionJob(name, freezeList)
rstMsg := fmt.Sprintf("Freeze empty volume(%s) meta partitions(%d)", name, cleans)
auditlog.LogMasterOp("freezeEmptyMetaPartition", rstMsg, nil)
auditlog.LogMasterOp("freezeEmptyMetaPartition", rstMsg, err)
if err != nil {
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeInternalError, Msg: err.Error()})
return
}
sendOkReply(w, r, newSuccessHTTPReply(fmt.Sprintf("Master will freeze empty meta partition of volume (%s) after 10 minutes", name)))
sendOkReply(w, r, newSuccessHTTPReply(fmt.Sprintf("Master will freeze empty meta partition of volume (%s) after 10 minutes. Task id: %s", name, name)))
}
func (m *Server) cleanEmptyMetaPartition(w http.ResponseWriter, r *http.Request) {
@ -8622,12 +8626,16 @@ func (m *Server) cleanEmptyMetaPartition(w http.ResponseWriter, r *http.Request)
return
}
go m.cluster.StartCleanEmptyMetaPartition(name)
err = m.cluster.StartCleanEmptyMetaPartition(name)
rstMsg := fmt.Sprintf("Clean volume(%s) empty meta partitions", name)
auditlog.LogMasterOp("cleanEmptyMetaPartition", rstMsg, nil)
auditlog.LogMasterOp("cleanEmptyMetaPartition", rstMsg, err)
if err != nil {
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeInternalError, Msg: err.Error()})
return
}
sendOkReply(w, r, newSuccessHTTPReply(fmt.Sprintf("Clean frozen meta partition for volume (%s) in the background. It may takes several hours.", name)))
sendOkReply(w, r, newSuccessHTTPReply(fmt.Sprintf("Clean frozen meta partition for volume (%s) in the background. It may takes several hours. task id: %s", name, name)))
}
func (m *Server) removeBackupMetaPartition(w http.ResponseWriter, r *http.Request) {
@ -8653,3 +8661,38 @@ func (m *Server) removeBackupMetaPartition(w http.ResponseWriter, r *http.Reques
sendOkReply(w, r, newSuccessHTTPReply("Remove all backup meta partitions successfully."))
}
func (m *Server) getCleanMetaPartitionTask(w http.ResponseWriter, r *http.Request) {
metric := exporter.NewTPCnt(apiToMetricsName(proto.AdminMetaPartitionGetCleanTask))
defer func() {
doStatAndMetric(proto.AdminMetaPartitionGetCleanTask, metric, nil, nil)
}()
var (
name string
err error
)
if err = r.ParseForm(); err != nil {
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: err.Error()})
return
}
name = r.FormValue(nameKey)
m.cluster.mu.Lock()
defer m.cluster.mu.Unlock()
if name == "" {
sendOkReply(w, r, newSuccessHTTPReply(m.cluster.cleanTask))
} else {
task, ok := m.cluster.cleanTask[name]
if !ok {
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: fmt.Sprintf("Can't find task for volume(%s)", name)})
return
}
sendOkReply(w, r, newSuccessHTTPReply(task))
}
return
}

224
master/cluster.go
View File

@ -104,6 +104,16 @@ type ClusterDecommission struct {
server *Server
}
type CleanTask struct {
Name string `json:"name"`
Status string `json:"status"`
TaskCnt int `json:"taskCount"`
FreezeCnt int `json:"freezeCount"`
CleanCnt int `json:"cleanCount"`
ResetCnt int `json:"resetCount"`
Timeout time.Time `json:"-"`
}
// Cluster stores all the cluster-level information.
type Cluster struct {
Name string
@ -146,7 +156,8 @@ type Cluster struct {
flashNodeTopo *flashNodeTopology
cleanTask []string
cleanTask map[string]*CleanTask
Cleaning bool
mu sync.Mutex
}
@ -440,7 +451,7 @@ func newCluster(name string, leaderInfo *LeaderInfo, fsm *MetadataFsm, partition
c.AutoDecommissionInterval.Store(int64(defaultAutoDecommissionDiskInterval))
c.server = server
c.flashNodeTopo = newFlashNodeTopology()
c.cleanTask = make([]string, 0)
c.cleanTask = make(map[string]*CleanTask)
return
}
@ -6320,40 +6331,119 @@ func (c *Cluster) checkMultipleReplicasOnSameMachine(hosts []string) (err error)
return nil
}
func (c *Cluster) StartCleanEmptyMetaPartition(name string) (err error) {
// skip the same volume's waiting task.
func (c *Cluster) FreezeEmptyMetaPartitionJob(name string, freezeList []*MetaPartition) error {
c.mu.Lock()
for _, task := range c.cleanTask {
if task == name {
task, ok := c.cleanTask[name]
if ok {
if task.Status == CleanTaskFreezing || task.Status == CleanTaskBackuping {
c.mu.Unlock()
return nil
return fmt.Errorf("The clean task for volume(%s) is %s", name, task.Status)
}
task.Status = CleanTaskFreezing
} else {
task = &CleanTask{
Name: name,
Status: CleanTaskFreezing,
TaskCnt: len(freezeList),
}
c.cleanTask[name] = task
}
c.cleanTask = append(c.cleanTask, name)
c.mu.Unlock()
err = c.DoCleanEmptyMetaPartition(name)
go func() {
// waiting for client to update meta partition 10 minutes.
time.Sleep(WaitForClientUpdateTimeMin * time.Minute)
for _, mp := range freezeList {
// freeze meta partition.
err := c.FreezeEmptyMetaPartition(mp, true)
if err != nil {
log.LogErrorf("Failed to freeze volume(%s) meta partition(%d), error: %s", name, mp.PartitionID, err.Error())
continue
}
task.FreezeCnt += 1
}
task.Status = CleanTaskFreezed
task.Timeout = time.Now().Add(WaitForTaskDeleteByHour * time.Hour)
c.mu.Lock()
if !c.Cleaning {
go c.DeleteCleanTasks()
c.Cleaning = true
}
c.mu.Unlock()
}()
return nil
}
func (c *Cluster) FreezeEmptyMetaPartition(mp *MetaPartition, freeze bool) error {
mr, err := mp.getMetaReplicaLeader()
if err != nil {
log.LogErrorf("Failed to clean volume(%s) empty meta partition, error: %s", name, err.Error())
log.LogErrorf("get meta replica leader error: %s", err.Error())
return err
}
task := mr.createTaskToFreezeReplica(mp.PartitionID, freeze)
metaNode, err := c.metaNode(task.OperatorAddr)
if err != nil {
log.LogErrorf("failed to get metanode(%s), error: %s", task.OperatorAddr, err.Error())
return err
}
_, err = metaNode.Sender.syncSendAdminTask(task)
if err != nil {
log.LogErrorf("action[FreezeEmptyMetaPartition] meta partition(%d), err: %s", mp.PartitionID, err.Error())
return err
}
return nil
}
func (c *Cluster) StartCleanEmptyMetaPartition(name string) error {
c.mu.Lock()
index := -1
for i, task := range c.cleanTask {
if task == name {
index = i
break
task, ok := c.cleanTask[name]
if ok {
if task.Status == CleanTaskFreezing || task.Status == CleanTaskBackuping {
c.mu.Unlock()
return fmt.Errorf("The clean task for volume(%s) is %s", name, task.Status)
}
}
if index >= 0 {
c.cleanTask = append(c.cleanTask[:index], c.cleanTask[index+1:]...)
task.Status = CleanTaskBackuping
} else {
task = &CleanTask{
Name: name,
Status: CleanTaskBackuping,
}
c.cleanTask[name] = task
}
c.mu.Unlock()
return err
go func() {
err := c.DoCleanEmptyMetaPartition(name)
if err != nil {
log.LogErrorf("Failed to clean volume(%s) empty meta partition, error: %s", name, err.Error())
}
task.Status = CleanTaskBackuped
task.Timeout = time.Now().Add(WaitForTaskDeleteByHour * time.Hour)
c.mu.Lock()
if !c.Cleaning {
go c.DeleteCleanTasks()
c.Cleaning = true
}
c.mu.Unlock()
}()
return nil
}
func (c *Cluster) DoCleanEmptyMetaPartition(name string) error {
c.mu.Lock()
task, ok := c.cleanTask[name]
if !ok {
log.LogErrorf("Can't find clean task for volume(%s)", name)
return fmt.Errorf("Can't find clean task for volume(%s)", name)
}
c.mu.Unlock()
vol, err := c.getVol(name)
if err != nil {
log.LogErrorf("DoCleanEmptyMetaPartition get volume(%s) error: %s", name, err.Error())
@ -6388,6 +6478,7 @@ func (c *Cluster) DoCleanEmptyMetaPartition(name string) error {
log.LogErrorf("volume(%s) meta partition(%d) update failed: %s", name, mp.PartitionID, err.Error())
continue
}
task.ResetCnt += 1
} else {
err = c.CleanEmptyMetaPartition(mp)
if err != nil {
@ -6396,6 +6487,7 @@ func (c *Cluster) DoCleanEmptyMetaPartition(name string) error {
}
deleteMaps[key] = mp
task.CleanCnt += 1
}
}
@ -6409,67 +6501,6 @@ func (c *Cluster) DoCleanEmptyMetaPartition(name string) error {
return nil
}
func (c *Cluster) FreezeEmptyMetaPartitionJob(name string, freezeList []*MetaPartition) (err error) {
// skip the same volume's waiting task.
c.mu.Lock()
for _, task := range c.cleanTask {
if task == name {
c.mu.Unlock()
return nil
}
}
c.cleanTask = append(c.cleanTask, name)
c.mu.Unlock()
// waiting for client to update meta partition 10 minutes.
time.Sleep(WaitForClientUpdateTimeMin * time.Minute)
for _, mp := range freezeList {
// freeze meta partition.
err = c.FreezeEmptyMetaPartition(mp, true)
if err != nil {
log.LogErrorf("Failed to freeze volume(%s) meta partition(%d), error: %s", name, mp.PartitionID, err.Error())
continue
}
}
c.mu.Lock()
index := -1
for i, task := range c.cleanTask {
if task == name {
index = i
break
}
}
if index >= 0 {
c.cleanTask = append(c.cleanTask[:index], c.cleanTask[index+1:]...)
}
c.mu.Unlock()
return err
}
func (c *Cluster) FreezeEmptyMetaPartition(mp *MetaPartition, freeze bool) error {
mr, err := mp.getMetaReplicaLeader()
if err != nil {
log.LogErrorf("get meta replica leader error: %s", err.Error())
return err
}
task := mr.createTaskToFreezeReplica(mp.PartitionID, freeze)
metaNode, err := c.metaNode(task.OperatorAddr)
if err != nil {
log.LogErrorf("failed to get metanode(%s), error: %s", task.OperatorAddr, err.Error())
return err
}
_, err = metaNode.Sender.syncSendAdminTask(task)
if err != nil {
log.LogErrorf("action[FreezeEmptyMetaPartition] meta partition(%d), err: %s", mp.PartitionID, err.Error())
return err
}
return nil
}
func (c *Cluster) CleanEmptyMetaPartition(mp *MetaPartition) error {
for _, replica := range mp.Replicas {
task := replica.createTaskToBackupReplica(mp.PartitionID)
@ -6487,3 +6518,36 @@ func (c *Cluster) CleanEmptyMetaPartition(mp *MetaPartition) error {
return nil
}
func (c *Cluster) DeleteCleanTasks() {
ticker := time.NewTicker(time.Hour)
defer ticker.Stop()
for {
select {
case <-ticker.C:
c.mu.Lock()
now := time.Now()
keysToDelete := make([]string, 0, len(c.cleanTask))
for key, task := range c.cleanTask {
if task.Status == CleanTaskFreezing || task.Status == CleanTaskBackuping {
continue
}
if task.Timeout.Before(now) {
keysToDelete = append(keysToDelete, key)
}
}
for _, key := range keysToDelete {
delete(c.cleanTask, key)
}
if len(c.cleanTask) == 0 {
c.Cleaning = false
c.mu.Unlock()
return
}
c.mu.Unlock()
case <-c.stopc:
return
}
}
}

8
master/const.go
View File

@ -251,6 +251,7 @@ const (
defaultFlashGroupSlotsCount = 32
RsvEmptyMetaPartitionCnt = 2
WaitForClientUpdateTimeMin = 10
WaitForTaskDeleteByHour = 24
)
const (
@ -490,3 +491,10 @@ func NodeTypeString(nodeType NodeType) string {
return fmt.Sprintf("unKnownNodeType(%v)", nodeType)
}
}
const (
CleanTaskFreezing = "freezing"
CleanTaskFreezed = "freezeDone"
CleanTaskBackuping = "backuping"
CleanTaskBackuped = "backupDone"
)

3
master/http_server.go
View File

@ -563,6 +563,9 @@ func (m *Server) registerAPIRoutes(router *mux.Router) {
router.NewRoute().Methods(http.MethodGet, http.MethodPost).
Path(proto.AdminMetaPartitionRemoveBackup).
HandlerFunc(m.removeBackupMetaPartition)
router.NewRoute().Methods(http.MethodGet).
Path(proto.AdminMetaPartitionGetCleanTask).
HandlerFunc(m.getCleanMetaPartitionTask)
// data partition management APIs
router.NewRoute().Methods(http.MethodGet).

1
proto/admin_proto.go
View File

@ -207,6 +207,7 @@ const (
AdminMetaPartitionFreezeEmpty = "/metaPartition/freezeEmpty"
AdminMetaPartitionCleanEmpty = "/metaPartition/cleanEmpty"
AdminMetaPartitionRemoveBackup = "/metaPartition/removeBackup"
AdminMetaPartitionGetCleanTask = "/metaPartition/getCleanTask"
AdminAddMetaReplica = "/metaReplica/add"
AdminDeleteMetaReplica = "/metaReplica/delete"
AdminPutDataPartitions = "/dataPartitions/set"