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

fix(master): loopback checking mp status and add error message.#1000160843

Browse Source 
1. Skip the no leader status and continue the loop back checking mp status.
2. Add plan msg to display error information.

Signed-off-by: Wu Huocheng <wuhuocheng@oppo.com>
This commit is contained in:
Wu Huocheng 2025-06-05 16:29:44 +08:00 committed by zhumingze1108
parent e3a4a71afa
commit f5be071244
2 changed files with 49 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

62
master/cluster_balance.go
View File

@ -1040,6 +1040,7 @@ func (c *Cluster) DoMetaPartitionBalanceTask(plan *proto.ClusterPlan) {
if err != nil {
log.LogErrorf("VerifyAllDestinationsIsLowLoad err: %s", err.Error())
plan.Status = PlanTaskError
plan.Msg = err.Error()
c.PlanRun = false
err1 := c.syncUpdateBalanceTask(plan)
if err1 != nil {
@ -1051,16 +1052,14 @@ func (c *Cluster) DoMetaPartitionBalanceTask(plan *proto.ClusterPlan) {
mp, err = c.getMetaPartitionByID(mpPlan.ID)
if err != nil {
log.LogErrorf("skip rebalance meta partition(%d) error: %s", mpPlan.ID, err.Error())
plan.Msg = err.Error()
continue
}
if mp.IsRecover {
log.LogWarnf("skip rebalance meta partition(%d), because it is recovering.", mpPlan.ID)
continue
}
if !mp.isLeaderExist() {
log.LogWarnf("skip rebalance meta partition(%d), because no leader exist.", mpPlan.ID)
err = c.waitForMetaPartitionReady(mp)
if err != nil {
log.LogErrorf("waitForMetaPartitionReady err: %s", err.Error())
plan.Msg = err.Error()
continue
}
@ -1070,6 +1069,7 @@ func (c *Cluster) DoMetaPartitionBalanceTask(plan *proto.ClusterPlan) {
err = c.syncUpdateBalanceTask(plan)
if err != nil {
log.LogErrorf("syncUpdateBalanceTask error: %s", err.Error())
plan.Msg = err.Error()
return
}
@ -1079,6 +1079,7 @@ func (c *Cluster) DoMetaPartitionBalanceTask(plan *proto.ClusterPlan) {
err = c.syncUpdateBalanceTask(plan)
if err != nil {
log.LogErrorf("syncUpdateBalanceTask error: %s", err.Error())
plan.Msg = err.Error()
}
return
}
@ -1090,7 +1091,7 @@ func (c *Cluster) DoMetaPartitionBalanceTask(plan *proto.ClusterPlan) {
err = c.changeAndCheckMetaPartitionLeader(mrPlan, mpPlan, mp)
if err != nil {
log.LogErrorf("changeAndCheckMetaPartitionLeader error: %s", err.Error())
c.SetMetaReplicaPlanStatusError(plan, mrPlan)
c.SetMetaReplicaPlanStatusError(plan, mrPlan, err.Error())
return
}
@ -1098,7 +1099,7 @@ func (c *Cluster) DoMetaPartitionBalanceTask(plan *proto.ClusterPlan) {
err = c.migrateMetaPartition(mrPlan.Source, mrPlan.Destination, mp)
if err != nil {
log.LogErrorf("migrateMetaPartition(%d) from %s to %s error: %s", mpPlan.ID, mrPlan.Source, mrPlan.Destination, err.Error())
c.SetMetaReplicaPlanStatusError(plan, mrPlan)
c.SetMetaReplicaPlanStatusError(plan, mrPlan, err.Error())
return
}
@ -1109,7 +1110,7 @@ func (c *Cluster) DoMetaPartitionBalanceTask(plan *proto.ClusterPlan) {
err = c.WaitForMetaPartitionMigrateDone(mp, mrPlan.Destination)
if err != nil {
log.LogErrorf("WaitForMetaPartitionMigrateDone mpid(%d) meta replica(%s) error: %s", mpPlan.ID, mrPlan.Destination, err.Error())
c.SetMetaReplicaPlanStatusError(plan, mrPlan)
c.SetMetaReplicaPlanStatusError(plan, mrPlan, err.Error())
return
}
@ -1118,6 +1119,7 @@ func (c *Cluster) DoMetaPartitionBalanceTask(plan *proto.ClusterPlan) {
err = c.syncUpdateBalanceTask(plan)
if err != nil {
log.LogErrorf("syncUpdateBalanceTask error: %s", err.Error())
plan.Msg = err.Error()
return
}
log.LogDebugf("Migrate meta partition(%d) from %s to %s done", mpPlan.ID, mrPlan.Source, mrPlan.Destination)
@ -1133,19 +1135,22 @@ func (c *Cluster) DoMetaPartitionBalanceTask(plan *proto.ClusterPlan) {
err = c.syncUpdateBalanceTask(plan)
if err != nil {
log.LogErrorf("syncUpdateBalanceTask err: %s", err.Error())
plan.Msg = err.Error()
}
if plan.Type == OfflinePlan {
err = c.offlineMetaNode(plan)
if err != nil {
log.LogErrorf("offlineMetaNode err: %s", err.Error())
plan.Msg = err.Error()
}
}
}
func (c *Cluster) SetMetaReplicaPlanStatusError(plan *proto.ClusterPlan, mrPlan *proto.MrBalanceInfo) {
func (c *Cluster) SetMetaReplicaPlanStatusError(plan *proto.ClusterPlan, mrPlan *proto.MrBalanceInfo, msg string) {
c.PlanRun = false
plan.Status = PlanTaskError
plan.Msg = msg
mrPlan.Status = PlanTaskError
err := c.syncUpdateBalanceTask(plan)
if err != nil {
@ -1157,6 +1162,8 @@ func (c *Cluster) WaitForMetaPartitionMigrateDone(mp *MetaPartition, addr string
ticker := time.NewTicker(1 * time.Second)
defer ticker.Stop()
var err error
var ready bool
for i := 0; i < 600; i++ {
select {
case <-ticker.C:
@ -1166,10 +1173,10 @@ func (c *Cluster) WaitForMetaPartitionMigrateDone(mp *MetaPartition, addr string
if !mp.isLeaderExist() {
continue
}
ready, err := CheckRaftStatus(mp, addr)
ready, err = CheckRaftStatus(mp, addr)
if err != nil {
log.LogErrorf("CheckRaftStatus err: %s", err.Error())
return err
log.LogWarnf("CheckRaftStatus err: %s", err.Error())
continue
}
if ready {
return nil
@ -1179,6 +1186,9 @@ func (c *Cluster) WaitForMetaPartitionMigrateDone(mp *MetaPartition, addr string
return fmt.Errorf("cluster is stopping")
}
}
if err != nil {
return err
}
return fmt.Errorf("Waiting for meta partition(%d) destination(%s) timeout", mp.PartitionID, addr)
}
@ -1623,3 +1633,27 @@ func selectOneLeaderAddr(mrPlan *proto.MrBalanceInfo, mpPlan *proto.MetaBalanceP
return ""
}
func (c *Cluster) waitForMetaPartitionReady(mp *MetaPartition) error {
for i := 0; i < CheckMetaLeaderRetry; i++ {
if !mp.IsRecover && mp.isLeaderExist() {
return nil
}
time.Sleep(CheckMetaLeaderInterval * time.Second)
}
if mp.IsRecover {
err := fmt.Errorf("skip rebalance meta partition(%d), because it is recovering.", mp.PartitionID)
log.LogWarn(err.Error())
return err
}
if !mp.isLeaderExist() {
err := fmt.Errorf("skip rebalance meta partition(%d), because no leader exist.", mp.PartitionID)
log.LogWarn(err.Error())
return err
}
return nil
}

1
proto/cluster_balance.go
View File

@ -65,4 +65,5 @@ type ClusterPlan struct {
Status string `json:"status" bson:"status"`
Expire time.Time `json:"expire" bson:"expire"`
Type string `json:"type" bson:"type"`
Msg string `json:"msg" bson:"msg"`
}