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): allow to create part of migrate plan.#1000169625

Browse Source 
Signed-off-by: Wu Huocheng <wuhuocheng@oppo.com>
This commit is contained in:
Wu Huocheng 2025-06-10 19:19:13 +08:00 committed by zhumingze1108
parent f04b0218c4
commit 055123bce7
2 changed files with 29 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
master/api_service_balance.go
View File

@ -549,7 +549,8 @@ func (m *Server) offlineMetaNode(w http.ResponseWriter, r *http.Request) {
}
if plan, err = m.cluster.CreateOfflineMetaNodePlan(offLineAddr); err != nil {
log.LogErrorf("Failed to create kick out plan for metanode(%s) err: %s", offLineAddr, err.Error())
err = fmt.Errorf("Create plan metanode(%s) err: %s, msg: %s", offLineAddr, err.Error(), plan.Msg)
log.LogErrorf(err.Error())
sendErrReply(w, r, newErrHTTPReply(err))
return
}

39
master/cluster_balance.go
View File

@ -34,6 +34,8 @@ type GetMigrateAddrParam struct {
LeastSize uint64
}
var NotEnoughResource = fmt.Errorf("not enough resource")
func (c *Cluster) FreezeEmptyMetaPartitionJob(name string, freezeList []*MetaPartition) error {
c.mu.Lock()
task, ok := c.cleanTask[name]
@ -540,20 +542,27 @@ func UpdateMetaReplicaPlanCount(mpPlan *proto.MetaBalancePlan, overLoadNodes []*
return nil
}
func FindMigrateDestination(migratePlan *proto.ClusterPlan) error {
for _, mp := range migratePlan.Plan {
func FindMigrateDestination(migratePlan *proto.ClusterPlan) (err error) {
for i, mp := range migratePlan.Plan {
if mp.CrossZone {
err := FindMigrateDestRetainZone(migratePlan, mp)
if err != nil {
log.LogErrorf("FindMigrateDestRetainZone error: %s", err.Error())
return err
}
err = FindMigrateDestRetainZone(migratePlan, mp)
} else {
err := FindMigrateDestInOneNodeSet(migratePlan, mp)
if err != nil {
log.LogErrorf("FindMigrateDestInOneNodeSet error: %s", err.Error())
return err
err = FindMigrateDestInOneNodeSet(migratePlan, mp)
}
if err == NotEnoughResource {
if i <= 0 {
migratePlan.Msg = fmt.Sprintf("require to migrate (%d) mp, but not create plan", len(migratePlan.Plan))
log.LogErrorf(migratePlan.Msg)
return
}
migratePlan.Msg = fmt.Sprintf("require to migrate (%d) mp, only create (%d) plan", len(migratePlan.Plan), i)
migratePlan.Plan = migratePlan.Plan[:i]
log.LogWarnf(migratePlan.Msg)
return nil
} else if err != nil {
log.LogErrorf("Fail to find reasonable metanode to create plan: %s", err.Error())
return err
}
}
@ -820,7 +829,7 @@ func FindMigrateDestInOneNodeSet(migratePlan *proto.ClusterPlan, mpPlan *proto.M
}
if !find {
log.LogWarnf("getParam: %+v. mpPlan: %+v. Resource: %+v", getParam, mpPlan, convertStructToJson(migratePlan.Low))
return fmt.Errorf("can't find the request low pressure nodes (%d)", getParam.RequestNum)
return NotEnoughResource
}
err := MigratePlanOriginalToDest(migratePlan, mpPlan, dests)
@ -1076,6 +1085,7 @@ func (c *Cluster) DoMetaPartitionBalanceTask(plan *proto.ClusterPlan) {
if !c.PlanRun {
plan.Status = PlanTaskStop
c.PlanRun = false
plan.Msg = "migrate plan is stopped"
err = c.syncUpdateBalanceTask(plan)
if err != nil {
log.LogErrorf("syncUpdateBalanceTask error: %s", err.Error())
@ -1085,6 +1095,7 @@ func (c *Cluster) DoMetaPartitionBalanceTask(plan *proto.ClusterPlan) {
}
if c.partition == nil || !c.partition.IsRaftLeader() {
c.PlanRun = false
plan.Msg = "master leader is changed"
return
}
// switch raft leader if the source is leader. And waiting for the leader to be elected.
@ -1143,6 +1154,10 @@ func (c *Cluster) DoMetaPartitionBalanceTask(plan *proto.ClusterPlan) {
if err != nil {
log.LogErrorf("offlineMetaNode err: %s", err.Error())
plan.Msg = err.Error()
err = c.syncUpdateBalanceTask(plan)
if err != nil {
log.LogErrorf("syncUpdateBalanceTask err: %s", err.Error())
}
}
}
}