mirror of
https://github.com/cubefs/cubefs.git
synced 2026-08-02 02:00:56 +00:00
chore(master): rerange the codes.#22834621
Signed-off-by: Wu Huocheng <wuhuocheng@oppo.com>
This commit is contained in:
Wu Huocheng
zhumingze1108
parent
f13cc3a1b6
commit
e27923fa1c
161
cli/cmd/mpbalance.go
Normal file
161
cli/cmd/mpbalance.go
Normal file
@ -0,0 +1,161 @@
|
||||
// Copyright 2025 The CubeFS Authors.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
|
||||
// implied. See the License for the specific language governing
|
||||
// permissions and limitations under the License.
|
||||
|
||||
package cmd
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
|
||||
"github.com/cubefs/cubefs/proto"
|
||||
"github.com/cubefs/cubefs/sdk/master"
|
||||
"github.com/spf13/cobra"
|
||||
)
|
||||
|
||||
const (
|
||||
cmdBalanceUse = "balance-task [COMMAND]"
|
||||
cmdBalanceShort = "balance memory usage ratio in meta node"
|
||||
)
|
||||
|
||||
func newBalanceCmd(client *master.MasterClient) *cobra.Command {
|
||||
cmd := &cobra.Command{
|
||||
Use: cmdBalanceUse,
|
||||
Short: cmdBalanceShort,
|
||||
}
|
||||
cmd.AddCommand(
|
||||
newCreateBalanceTaskCmd(client),
|
||||
newDisplayBalanceTaskCmd(client),
|
||||
newRunBalanceTaskCmd(client),
|
||||
newStopBalanceTaskCmd(client),
|
||||
newDeleteBalanceTaskCmd(client),
|
||||
)
|
||||
return cmd
|
||||
}
|
||||
|
||||
const (
|
||||
cmdCreateBalanceTask = "create"
|
||||
cmdCreateBalanceTaskShort = "Create meta partition balance task"
|
||||
cmdDisplayBalanceTask = "display"
|
||||
cmdDisplayBalanceTaskShort = "Display meta partition balance task"
|
||||
cmdRunBalanceTask = "run"
|
||||
cmdRunBalanceTaskShort = "Run meta partition balance task"
|
||||
cmdStopBalanceTask = "stop"
|
||||
cmdStopBalanceTaskShort = "Stop meta partition balance task"
|
||||
cmdDeleteBalanceTask = "delete"
|
||||
cmdDeleteBalanceTaskShort = "Delete meta partition balance task"
|
||||
)
|
||||
|
||||
func newCreateBalanceTaskCmd(client *master.MasterClient) *cobra.Command {
|
||||
cmd := &cobra.Command{
|
||||
Use: cmdCreateBalanceTask,
|
||||
Short: cmdCreateBalanceTaskShort,
|
||||
Run: func(cmd *cobra.Command, args []string) {
|
||||
var err error
|
||||
defer func() {
|
||||
errout(err)
|
||||
}()
|
||||
var task *proto.ClusterPlan
|
||||
if task, err = client.AdminAPI().CreateBalanceTask(); err != nil {
|
||||
return
|
||||
}
|
||||
out, err := json.MarshalIndent(task, "", " ")
|
||||
if err != nil {
|
||||
stdout("marshal task failed: %s", err.Error())
|
||||
return
|
||||
}
|
||||
stdout("%s", string(out))
|
||||
},
|
||||
}
|
||||
return cmd
|
||||
}
|
||||
|
||||
func newDisplayBalanceTaskCmd(client *master.MasterClient) *cobra.Command {
|
||||
cmd := &cobra.Command{
|
||||
Use: cmdDisplayBalanceTask,
|
||||
Short: cmdDisplayBalanceTaskShort,
|
||||
Run: func(cmd *cobra.Command, args []string) {
|
||||
var err error
|
||||
defer func() {
|
||||
errout(err)
|
||||
}()
|
||||
var task *proto.ClusterPlan
|
||||
if task, err = client.AdminAPI().GetBalanceTask(); err != nil {
|
||||
return
|
||||
}
|
||||
out, err := json.MarshalIndent(task, "", " ")
|
||||
if err != nil {
|
||||
stdout("marshal task failed: %s", err.Error())
|
||||
return
|
||||
}
|
||||
stdout("%s", string(out))
|
||||
},
|
||||
}
|
||||
return cmd
|
||||
}
|
||||
|
||||
func newRunBalanceTaskCmd(client *master.MasterClient) *cobra.Command {
|
||||
cmd := &cobra.Command{
|
||||
Use: cmdRunBalanceTask,
|
||||
Short: cmdRunBalanceTaskShort,
|
||||
Run: func(cmd *cobra.Command, args []string) {
|
||||
var err error
|
||||
defer func() {
|
||||
errout(err)
|
||||
}()
|
||||
var result string
|
||||
if result, err = client.AdminAPI().RunBalanceTask(); err != nil {
|
||||
return
|
||||
}
|
||||
stdout("%s", result)
|
||||
},
|
||||
}
|
||||
return cmd
|
||||
}
|
||||
|
||||
func newStopBalanceTaskCmd(client *master.MasterClient) *cobra.Command {
|
||||
cmd := &cobra.Command{
|
||||
Use: cmdStopBalanceTask,
|
||||
Short: cmdStopBalanceTaskShort,
|
||||
Run: func(cmd *cobra.Command, args []string) {
|
||||
var err error
|
||||
defer func() {
|
||||
errout(err)
|
||||
}()
|
||||
var result string
|
||||
if result, err = client.AdminAPI().StopBalanceTask(); err != nil {
|
||||
return
|
||||
}
|
||||
stdout("%s", result)
|
||||
},
|
||||
}
|
||||
return cmd
|
||||
}
|
||||
|
||||
func newDeleteBalanceTaskCmd(client *master.MasterClient) *cobra.Command {
|
||||
cmd := &cobra.Command{
|
||||
Use: cmdDeleteBalanceTask,
|
||||
Short: cmdDeleteBalanceTaskShort,
|
||||
Run: func(cmd *cobra.Command, args []string) {
|
||||
var err error
|
||||
defer func() {
|
||||
errout(err)
|
||||
}()
|
||||
var result string
|
||||
if result, err = client.AdminAPI().DeleteBalanceTask(); err != nil {
|
||||
return
|
||||
}
|
||||
stdout("%s", result)
|
||||
},
|
||||
}
|
||||
return cmd
|
||||
}
|
||||
@ -82,6 +82,7 @@ func NewRootCmd(client *master.MasterClient) *CubeFSCmd {
|
||||
newVersionCmd(client),
|
||||
newFlashNodeCmd(client),
|
||||
newFlashGroupCmd(client),
|
||||
newBalanceCmd(client),
|
||||
)
|
||||
return cmd
|
||||
}
|
||||
|
||||
@ -8473,243 +8473,6 @@ func (m *Server) getUpgradeCompatibleSettings(w http.ResponseWriter, r *http.Req
|
||||
sendOkReply(w, r, newSuccessHTTPReply(cInfo))
|
||||
}
|
||||
|
||||
func (m *Server) getMetaPartitionEmptyStatus(w http.ResponseWriter, r *http.Request) {
|
||||
metric := exporter.NewTPCnt(apiToMetricsName(proto.AdminMetaPartitionEmptyStatus))
|
||||
defer func() {
|
||||
doStatAndMetric(proto.AdminMetaPartitionEmptyStatus, metric, nil, nil)
|
||||
}()
|
||||
|
||||
mpsStatus := make([]proto.VolEmptyMpStats, 0, len(m.cluster.vols))
|
||||
for _, name := range m.cluster.allVolNames() {
|
||||
vol, err := m.cluster.getVol(name)
|
||||
if err != nil {
|
||||
log.LogErrorf("[getMetaPartitionEmptyStatus] getVol(%s) failed: %s", name, err.Error())
|
||||
continue
|
||||
}
|
||||
// skip the deleted volume.
|
||||
if vol.Status == proto.VolStatusMarkDelete {
|
||||
continue
|
||||
}
|
||||
volStatus := proto.VolEmptyMpStats{
|
||||
Name: name,
|
||||
}
|
||||
volStatus.MetaPartitions = make([]*proto.MetaPartitionView, 0, len(vol.MetaPartitions))
|
||||
volStatus.Total = len(vol.MetaPartitions)
|
||||
mps := vol.getSortMetaPartitions()
|
||||
for _, mp := range mps {
|
||||
if mp.IsFreeze || mp.IsEmptyToBeClean() {
|
||||
volStatus.EmptyCount++
|
||||
volStatus.MetaPartitions = append(volStatus.MetaPartitions, getMetaPartitionView(mp))
|
||||
}
|
||||
}
|
||||
if volStatus.EmptyCount > RsvEmptyMetaPartitionCnt {
|
||||
mpsStatus = append(mpsStatus, volStatus)
|
||||
}
|
||||
}
|
||||
|
||||
sendOkReply(w, r, newSuccessHTTPReply(mpsStatus))
|
||||
}
|
||||
|
||||
func (m *Server) freezeEmptyMetaPartition(w http.ResponseWriter, r *http.Request) {
|
||||
metric := exporter.NewTPCnt(apiToMetricsName(proto.AdminMetaPartitionFreezeEmpty))
|
||||
defer func() {
|
||||
doStatAndMetric(proto.AdminMetaPartitionFreezeEmpty, metric, nil, nil)
|
||||
}()
|
||||
|
||||
var (
|
||||
name string
|
||||
count int
|
||||
err error
|
||||
)
|
||||
|
||||
if err = r.ParseForm(); err != nil {
|
||||
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: err.Error()})
|
||||
return
|
||||
}
|
||||
|
||||
if name, err = extractName(r); err != nil {
|
||||
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: err.Error()})
|
||||
return
|
||||
}
|
||||
|
||||
if count, err = extractUint(r, countKey); err != nil {
|
||||
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: err.Error()})
|
||||
return
|
||||
}
|
||||
if count < RsvEmptyMetaPartitionCnt {
|
||||
// reserve 2 empty mp at least, not include the last one.
|
||||
count = RsvEmptyMetaPartitionCnt
|
||||
}
|
||||
|
||||
vol, err := m.cluster.getVol(name)
|
||||
if err != nil {
|
||||
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: err.Error()})
|
||||
return
|
||||
}
|
||||
|
||||
if vol.Status == proto.VolStatusMarkDelete {
|
||||
sendOkReply(w, r, newSuccessHTTPReply(fmt.Sprintf("volume (%s) is deleted already.", name)))
|
||||
return
|
||||
}
|
||||
|
||||
mps := vol.getSortMetaPartitions()
|
||||
if len(mps) <= RsvEmptyMetaPartitionCnt {
|
||||
err = fmt.Errorf("the all meta partition number is less than %d", RsvEmptyMetaPartitionCnt)
|
||||
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: err.Error()})
|
||||
return
|
||||
}
|
||||
|
||||
total := 0
|
||||
for _, mp := range mps {
|
||||
if mp.IsEmptyToBeClean() {
|
||||
total++
|
||||
}
|
||||
}
|
||||
|
||||
cleans := total - count
|
||||
if cleans <= 0 {
|
||||
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: "reserve mp number is larger than or equal empty number"})
|
||||
return
|
||||
}
|
||||
|
||||
freezeList := make([]*MetaPartition, 0, cleans)
|
||||
i := 0
|
||||
for j := len(mps) - 1; j >= 0; j -= 1 {
|
||||
mp := mps[j]
|
||||
if !mp.IsEmptyToBeClean() {
|
||||
continue
|
||||
}
|
||||
|
||||
mp.IsFreeze = true
|
||||
if mp.Status == proto.ReadWrite {
|
||||
mp.Status = proto.ReadOnly
|
||||
}
|
||||
// store the meta partition status.
|
||||
err = m.cluster.syncUpdateMetaPartition(mp)
|
||||
if err != nil {
|
||||
log.LogErrorf("volume(%s) meta partition(%d) update failed: %s", name, mp.PartitionID, err.Error())
|
||||
continue
|
||||
}
|
||||
freezeList = append(freezeList, mp)
|
||||
|
||||
i++
|
||||
if i >= cleans {
|
||||
break
|
||||
}
|
||||
}
|
||||
err = m.cluster.FreezeEmptyMetaPartitionJob(name, freezeList)
|
||||
|
||||
rstMsg := fmt.Sprintf("Freeze empty volume(%s) meta partitions(%d)", name, cleans)
|
||||
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. Task id: %s", name, name)))
|
||||
}
|
||||
|
||||
func (m *Server) cleanEmptyMetaPartition(w http.ResponseWriter, r *http.Request) {
|
||||
metric := exporter.NewTPCnt(apiToMetricsName(proto.AdminMetaPartitionCleanEmpty))
|
||||
defer func() {
|
||||
doStatAndMetric(proto.AdminMetaPartitionCleanEmpty, 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
|
||||
}
|
||||
|
||||
if name, err = extractName(r); err != nil {
|
||||
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: err.Error()})
|
||||
return
|
||||
}
|
||||
|
||||
vol, err := m.cluster.getVol(name)
|
||||
if err != nil {
|
||||
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: err.Error()})
|
||||
return
|
||||
}
|
||||
|
||||
if vol.Status == proto.VolStatusMarkDelete {
|
||||
sendOkReply(w, r, newSuccessHTTPReply(fmt.Sprintf("volume (%s) is deleted already.", name)))
|
||||
return
|
||||
}
|
||||
|
||||
err = m.cluster.StartCleanEmptyMetaPartition(name)
|
||||
|
||||
rstMsg := fmt.Sprintf("Clean volume(%s) empty meta partitions", name)
|
||||
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. task id: %s", name, name)))
|
||||
}
|
||||
|
||||
func (m *Server) removeBackupMetaPartition(w http.ResponseWriter, r *http.Request) {
|
||||
metric := exporter.NewTPCnt(apiToMetricsName(proto.AdminMetaPartitionRemoveBackup))
|
||||
defer func() {
|
||||
doStatAndMetric(proto.AdminMetaPartitionRemoveBackup, metric, nil, nil)
|
||||
}()
|
||||
|
||||
m.cluster.metaNodes.Range(func(key, value interface{}) bool {
|
||||
metanode, ok := value.(*MetaNode)
|
||||
if !ok {
|
||||
return true
|
||||
}
|
||||
task := proto.NewAdminTask(proto.OpRemoveBackupMetaPartition, metanode.Addr, nil)
|
||||
_, err := metanode.Sender.syncSendAdminTask(task)
|
||||
if err != nil {
|
||||
log.LogErrorf("failed to remove empty meta partition")
|
||||
}
|
||||
return true
|
||||
})
|
||||
|
||||
auditlog.LogMasterOp("removeBackupMetaPartition", "clean all backup meta partitions", nil)
|
||||
|
||||
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))
|
||||
}
|
||||
}
|
||||
|
||||
func removeDuplicatePaths(paths []string) []string {
|
||||
uniquePaths := make(map[string]bool)
|
||||
var result []string
|
||||
@ -8752,112 +8515,3 @@ func deduplicateAndRemoveContained(pathStr string) string {
|
||||
finalPaths := removeSubPaths(uniquePaths)
|
||||
return strings.Join(finalPaths, ",")
|
||||
}
|
||||
|
||||
func (m *Server) createBalancePlan(w http.ResponseWriter, r *http.Request) {
|
||||
var err error
|
||||
metric := exporter.NewTPCnt(apiToMetricsName(proto.CreateBalanceTask))
|
||||
defer func() {
|
||||
doStatAndMetric(proto.CreateBalanceTask, metric, err, nil)
|
||||
}()
|
||||
|
||||
var plan *ClusterPlan
|
||||
// search the raft storage. Only store one plan
|
||||
plan, err = m.cluster.loadBalanceTask()
|
||||
if err == nil && plan != nil {
|
||||
err = fmt.Errorf("There is a meta partition task plan already. Please remove it before create new one.")
|
||||
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeInternalError, Msg: err.Error(), Data: plan})
|
||||
return
|
||||
}
|
||||
|
||||
plan, err = m.cluster.GetMetaNodePressureView()
|
||||
if err != nil {
|
||||
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeInternalError, Msg: err.Error(), Data: plan})
|
||||
return
|
||||
}
|
||||
|
||||
// Save into raft storage.
|
||||
err = m.cluster.syncAddBalanceTask(plan)
|
||||
if err != nil {
|
||||
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeInternalError, Msg: err.Error(), Data: plan})
|
||||
return
|
||||
}
|
||||
|
||||
sendOkReply(w, r, newSuccessHTTPReply(plan))
|
||||
}
|
||||
|
||||
func (m *Server) getBalancePlan(w http.ResponseWriter, r *http.Request) {
|
||||
var err error
|
||||
metric := exporter.NewTPCnt(apiToMetricsName(proto.GetBalanceTask))
|
||||
defer func() {
|
||||
doStatAndMetric(proto.GetBalanceTask, metric, err, nil)
|
||||
}()
|
||||
|
||||
var plan *ClusterPlan
|
||||
// search the raft storage. Only store one plan
|
||||
plan, err = m.cluster.loadBalanceTask()
|
||||
if err != nil {
|
||||
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeInternalError, Msg: err.Error(), Data: plan})
|
||||
return
|
||||
}
|
||||
|
||||
sendOkReply(w, r, newSuccessHTTPReply(plan))
|
||||
}
|
||||
|
||||
func (m *Server) runBalancePlan(w http.ResponseWriter, r *http.Request) {
|
||||
var err error
|
||||
metric := exporter.NewTPCnt(apiToMetricsName(proto.RunBalanceTask))
|
||||
defer func() {
|
||||
doStatAndMetric(proto.RunBalanceTask, metric, err, nil)
|
||||
}()
|
||||
|
||||
// search the raft storage. Only store one plan
|
||||
err = m.cluster.RunMetaPartitionBalanceTask()
|
||||
if err != nil {
|
||||
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeInternalError, Msg: err.Error()})
|
||||
return
|
||||
}
|
||||
|
||||
sendOkReply(w, r, newSuccessHTTPReply("Start running balance task successfully."))
|
||||
}
|
||||
|
||||
func (m *Server) stopBalancePlan(w http.ResponseWriter, r *http.Request) {
|
||||
var err error
|
||||
metric := exporter.NewTPCnt(apiToMetricsName(proto.StopBalanceTask))
|
||||
defer func() {
|
||||
doStatAndMetric(proto.StopBalanceTask, metric, err, nil)
|
||||
}()
|
||||
|
||||
// search the raft storage. Only store one plan
|
||||
err = m.cluster.StopMetaPartitionBalanceTask()
|
||||
if err != nil {
|
||||
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeInternalError, Msg: err.Error()})
|
||||
return
|
||||
}
|
||||
|
||||
sendOkReply(w, r, newSuccessHTTPReply("Stop balance task successfully."))
|
||||
}
|
||||
|
||||
func (m *Server) deleteBalancePlan(w http.ResponseWriter, r *http.Request) {
|
||||
var err error
|
||||
metric := exporter.NewTPCnt(apiToMetricsName(proto.DeleteBalanceTask))
|
||||
defer func() {
|
||||
doStatAndMetric(proto.DeleteBalanceTask, metric, err, nil)
|
||||
}()
|
||||
|
||||
m.cluster.mu.Lock()
|
||||
defer m.cluster.mu.Unlock()
|
||||
|
||||
if m.cluster.PlanRun {
|
||||
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: "Please stop the running task before deleting it."})
|
||||
return
|
||||
}
|
||||
|
||||
// search the raft storage. Only store one plan
|
||||
_, err = m.cluster.syncDeleteBalanceTask()
|
||||
if err != nil {
|
||||
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeInternalError, Msg: err.Error()})
|
||||
return
|
||||
}
|
||||
|
||||
sendOkReply(w, r, newSuccessHTTPReply("Delete balance plan task successfully."))
|
||||
}
|
||||
|
||||
572
master/api_service_balance.go
Normal file
572
master/api_service_balance.go
Normal file
@ -0,0 +1,572 @@
|
||||
// Copyright 2025 The CubeFS Authors.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
|
||||
// implied. See the License for the specific language governing
|
||||
// permissions and limitations under the License.
|
||||
|
||||
package master
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"net/http"
|
||||
"strconv"
|
||||
|
||||
"github.com/cubefs/cubefs/proto"
|
||||
"github.com/cubefs/cubefs/util"
|
||||
"github.com/cubefs/cubefs/util/auditlog"
|
||||
"github.com/cubefs/cubefs/util/exporter"
|
||||
"github.com/cubefs/cubefs/util/log"
|
||||
)
|
||||
|
||||
type MetaReplicaInfo struct {
|
||||
MaxInodeID uint64 `json:"MaxInodeID"`
|
||||
InodeCount uint64 `json:"InodeCount"`
|
||||
DentryCount uint64 `json:"DentryCount"`
|
||||
FreeListLen uint64 `json:"FreeListLen"`
|
||||
TxCnt uint64 `json:"TxCnt"`
|
||||
TxRbInoCnt uint64 `json:"TxRbInoCnt"`
|
||||
TxRbDenCnt uint64 `json:"TxRbDenCnt"`
|
||||
}
|
||||
|
||||
type MigrateResult struct {
|
||||
Mp MetaReplicaInfo `json:"mp"`
|
||||
Target MetaReplicaInfo `json:"target"`
|
||||
}
|
||||
|
||||
func (m *Server) getMetaPartitionEmptyStatus(w http.ResponseWriter, r *http.Request) {
|
||||
metric := exporter.NewTPCnt(apiToMetricsName(proto.AdminMetaPartitionEmptyStatus))
|
||||
defer func() {
|
||||
doStatAndMetric(proto.AdminMetaPartitionEmptyStatus, metric, nil, nil)
|
||||
}()
|
||||
|
||||
mpsStatus := make([]proto.VolEmptyMpStats, 0, len(m.cluster.vols))
|
||||
for _, name := range m.cluster.allVolNames() {
|
||||
vol, err := m.cluster.getVol(name)
|
||||
if err != nil {
|
||||
log.LogErrorf("[getMetaPartitionEmptyStatus] getVol(%s) failed: %s", name, err.Error())
|
||||
continue
|
||||
}
|
||||
// skip the deleted volume.
|
||||
if vol.Status == proto.VolStatusMarkDelete {
|
||||
continue
|
||||
}
|
||||
volStatus := proto.VolEmptyMpStats{
|
||||
Name: name,
|
||||
}
|
||||
volStatus.MetaPartitions = make([]*proto.MetaPartitionView, 0, len(vol.MetaPartitions))
|
||||
volStatus.Total = len(vol.MetaPartitions)
|
||||
mps := vol.getSortMetaPartitions()
|
||||
for _, mp := range mps {
|
||||
if mp.IsFreeze || mp.IsEmptyToBeClean() {
|
||||
volStatus.EmptyCount++
|
||||
volStatus.MetaPartitions = append(volStatus.MetaPartitions, getMetaPartitionView(mp))
|
||||
}
|
||||
}
|
||||
if volStatus.EmptyCount > RsvEmptyMetaPartitionCnt {
|
||||
mpsStatus = append(mpsStatus, volStatus)
|
||||
}
|
||||
}
|
||||
|
||||
sendOkReply(w, r, newSuccessHTTPReply(mpsStatus))
|
||||
}
|
||||
|
||||
func (m *Server) freezeEmptyMetaPartition(w http.ResponseWriter, r *http.Request) {
|
||||
metric := exporter.NewTPCnt(apiToMetricsName(proto.AdminMetaPartitionFreezeEmpty))
|
||||
defer func() {
|
||||
doStatAndMetric(proto.AdminMetaPartitionFreezeEmpty, metric, nil, nil)
|
||||
}()
|
||||
|
||||
var (
|
||||
name string
|
||||
count int
|
||||
err error
|
||||
)
|
||||
|
||||
if err = r.ParseForm(); err != nil {
|
||||
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: err.Error()})
|
||||
return
|
||||
}
|
||||
|
||||
if name, err = extractName(r); err != nil {
|
||||
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: err.Error()})
|
||||
return
|
||||
}
|
||||
|
||||
if count, err = extractUint(r, countKey); err != nil {
|
||||
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: err.Error()})
|
||||
return
|
||||
}
|
||||
if count < RsvEmptyMetaPartitionCnt {
|
||||
// reserve 2 empty mp at least, not include the last one.
|
||||
count = RsvEmptyMetaPartitionCnt
|
||||
}
|
||||
|
||||
vol, err := m.cluster.getVol(name)
|
||||
if err != nil {
|
||||
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: err.Error()})
|
||||
return
|
||||
}
|
||||
|
||||
if vol.Status == proto.VolStatusMarkDelete {
|
||||
sendOkReply(w, r, newSuccessHTTPReply(fmt.Sprintf("volume (%s) is deleted already.", name)))
|
||||
return
|
||||
}
|
||||
|
||||
mps := vol.getSortMetaPartitions()
|
||||
if len(mps) <= RsvEmptyMetaPartitionCnt {
|
||||
err = fmt.Errorf("the all meta partition number is less than %d", RsvEmptyMetaPartitionCnt)
|
||||
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: err.Error()})
|
||||
return
|
||||
}
|
||||
|
||||
total := 0
|
||||
for _, mp := range mps {
|
||||
if mp.IsEmptyToBeClean() {
|
||||
total++
|
||||
}
|
||||
}
|
||||
|
||||
cleans := total - count
|
||||
if cleans <= 0 {
|
||||
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: "reserve mp number is larger than or equal empty number"})
|
||||
return
|
||||
}
|
||||
|
||||
freezeList := make([]*MetaPartition, 0, cleans)
|
||||
i := 0
|
||||
for j := len(mps) - 1; j >= 0; j -= 1 {
|
||||
mp := mps[j]
|
||||
if !mp.IsEmptyToBeClean() {
|
||||
continue
|
||||
}
|
||||
|
||||
mp.IsFreeze = true
|
||||
if mp.Status == proto.ReadWrite {
|
||||
mp.Status = proto.ReadOnly
|
||||
}
|
||||
// store the meta partition status.
|
||||
err = m.cluster.syncUpdateMetaPartition(mp)
|
||||
if err != nil {
|
||||
log.LogErrorf("volume(%s) meta partition(%d) update failed: %s", name, mp.PartitionID, err.Error())
|
||||
continue
|
||||
}
|
||||
freezeList = append(freezeList, mp)
|
||||
|
||||
i++
|
||||
if i >= cleans {
|
||||
break
|
||||
}
|
||||
}
|
||||
err = m.cluster.FreezeEmptyMetaPartitionJob(name, freezeList)
|
||||
|
||||
rstMsg := fmt.Sprintf("Freeze empty volume(%s) meta partitions(%d)", name, cleans)
|
||||
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. Task id: %s", name, name)))
|
||||
}
|
||||
|
||||
func (m *Server) cleanEmptyMetaPartition(w http.ResponseWriter, r *http.Request) {
|
||||
metric := exporter.NewTPCnt(apiToMetricsName(proto.AdminMetaPartitionCleanEmpty))
|
||||
defer func() {
|
||||
doStatAndMetric(proto.AdminMetaPartitionCleanEmpty, 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
|
||||
}
|
||||
|
||||
if name, err = extractName(r); err != nil {
|
||||
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: err.Error()})
|
||||
return
|
||||
}
|
||||
|
||||
vol, err := m.cluster.getVol(name)
|
||||
if err != nil {
|
||||
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: err.Error()})
|
||||
return
|
||||
}
|
||||
|
||||
if vol.Status == proto.VolStatusMarkDelete {
|
||||
sendOkReply(w, r, newSuccessHTTPReply(fmt.Sprintf("volume (%s) is deleted already.", name)))
|
||||
return
|
||||
}
|
||||
|
||||
err = m.cluster.StartCleanEmptyMetaPartition(name)
|
||||
|
||||
rstMsg := fmt.Sprintf("Clean volume(%s) empty meta partitions", name)
|
||||
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. task id: %s", name, name)))
|
||||
}
|
||||
|
||||
func (m *Server) removeBackupMetaPartition(w http.ResponseWriter, r *http.Request) {
|
||||
metric := exporter.NewTPCnt(apiToMetricsName(proto.AdminMetaPartitionRemoveBackup))
|
||||
defer func() {
|
||||
doStatAndMetric(proto.AdminMetaPartitionRemoveBackup, metric, nil, nil)
|
||||
}()
|
||||
|
||||
m.cluster.metaNodes.Range(func(key, value interface{}) bool {
|
||||
metanode, ok := value.(*MetaNode)
|
||||
if !ok {
|
||||
return true
|
||||
}
|
||||
task := proto.NewAdminTask(proto.OpRemoveBackupMetaPartition, metanode.Addr, nil)
|
||||
_, err := metanode.Sender.syncSendAdminTask(task)
|
||||
if err != nil {
|
||||
log.LogErrorf("failed to remove empty meta partition")
|
||||
}
|
||||
return true
|
||||
})
|
||||
|
||||
auditlog.LogMasterOp("removeBackupMetaPartition", "clean all backup meta partitions", nil)
|
||||
|
||||
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
|
||||
}
|
||||
|
||||
func parseMigratePartitionParam(r *http.Request) (srcAddr, targetAddr string, id uint64, err error) {
|
||||
if err = r.ParseForm(); err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
srcAddr = r.FormValue(srcAddrKey)
|
||||
if srcAddr == "" {
|
||||
err = fmt.Errorf("parseMigratePartitionParam source address is empty")
|
||||
return
|
||||
}
|
||||
if ipAddr, ok := util.ParseAddrToIpAddr(srcAddr); ok {
|
||||
srcAddr = ipAddr
|
||||
}
|
||||
|
||||
targetAddr = r.FormValue(targetAddrKey)
|
||||
if targetAddr == "" {
|
||||
err = fmt.Errorf("parseMigratePartitionParam target address is empty")
|
||||
return
|
||||
}
|
||||
if ipAddr, ok := util.ParseAddrToIpAddr(targetAddr); ok {
|
||||
targetAddr = ipAddr
|
||||
}
|
||||
|
||||
if srcAddr == targetAddr {
|
||||
err = fmt.Errorf("parseMigratePartitionParam srcAddr %s can't be equal to targetAddr %s", srcAddr, targetAddr)
|
||||
return
|
||||
}
|
||||
|
||||
value := r.FormValue(idKey)
|
||||
if value == "" {
|
||||
err = fmt.Errorf("parseMigratePartitionParam meta partition id is needed")
|
||||
return
|
||||
}
|
||||
|
||||
if id, err = strconv.ParseUint(value, 10, 64); err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
func (m *Server) migrateMetaPartitionHandler(w http.ResponseWriter, r *http.Request) {
|
||||
var (
|
||||
srcAddr string
|
||||
targetAddr string
|
||||
mpid uint64
|
||||
err error
|
||||
mp *MetaPartition
|
||||
)
|
||||
metric := exporter.NewTPCnt(apiToMetricsName(proto.MigrateMetaPartition))
|
||||
defer func() {
|
||||
doStatAndMetric(proto.MigrateMetaPartition, metric, err, nil)
|
||||
}()
|
||||
|
||||
srcAddr, targetAddr, mpid, err = parseMigratePartitionParam(r)
|
||||
if err != nil {
|
||||
sendErrReply(w, r, newErrHTTPReply(err))
|
||||
return
|
||||
}
|
||||
|
||||
targetNode, err := m.cluster.metaNode(targetAddr)
|
||||
if err != nil {
|
||||
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeMetaNodeNotExists, Msg: err.Error()})
|
||||
return
|
||||
}
|
||||
|
||||
if !targetNode.IsWriteAble() || !targetNode.PartitionCntLimited() {
|
||||
err = fmt.Errorf("[%s] is not writable, can't used as target addr for migrate", targetAddr)
|
||||
sendErrReply(w, r, newErrHTTPReply(err))
|
||||
return
|
||||
}
|
||||
|
||||
mp, err = m.cluster.getMetaPartitionByID(mpid)
|
||||
if err != nil {
|
||||
err = fmt.Errorf("Failed to get meta partition (%d)", mpid)
|
||||
sendErrReply(w, r, newErrHTTPReply(err))
|
||||
return
|
||||
}
|
||||
|
||||
if err = m.cluster.migrateMetaPartition(srcAddr, targetAddr, mp); err != nil {
|
||||
sendErrReply(w, r, newErrHTTPReply(err))
|
||||
return
|
||||
}
|
||||
|
||||
rstMsg := fmt.Sprintf("migrateMetaPartitionHandler id(%d) from src [%s] to target[%s] has migrate successfully", mpid, srcAddr, targetAddr)
|
||||
auditlog.LogMasterOp("MigrateMetaPartition", rstMsg, nil)
|
||||
sendOkReply(w, r, newSuccessHTTPReply(rstMsg))
|
||||
}
|
||||
|
||||
func parseMigrateResultParam(r *http.Request) (targetAddr string, id uint64, err error) {
|
||||
if err = r.ParseForm(); err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
targetAddr = r.FormValue(targetAddrKey)
|
||||
if targetAddr == "" {
|
||||
err = fmt.Errorf("parseMigrateResultParam targetAddrKey is null")
|
||||
return
|
||||
}
|
||||
if ipAddr, ok := util.ParseAddrToIpAddr(targetAddr); ok {
|
||||
targetAddr = ipAddr
|
||||
}
|
||||
|
||||
value := r.FormValue(idKey)
|
||||
if value == "" {
|
||||
err = fmt.Errorf("parseMigrateResultParam meta partition id is needed")
|
||||
return
|
||||
}
|
||||
|
||||
if id, err = strconv.ParseUint(value, 10, 64); err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
func (m *Server) migrateResultHandler(w http.ResponseWriter, r *http.Request) {
|
||||
var (
|
||||
targetAddr string
|
||||
mpid uint64
|
||||
err error
|
||||
mp *MetaPartition
|
||||
)
|
||||
metric := exporter.NewTPCnt(apiToMetricsName(proto.MigrateResult))
|
||||
defer func() {
|
||||
doStatAndMetric(proto.MigrateResult, metric, err, nil)
|
||||
}()
|
||||
|
||||
targetAddr, mpid, err = parseMigrateResultParam(r)
|
||||
if err != nil {
|
||||
sendErrReply(w, r, newErrHTTPReply(err))
|
||||
return
|
||||
}
|
||||
|
||||
mp, err = m.cluster.getMetaPartitionByID(mpid)
|
||||
if err != nil {
|
||||
err = fmt.Errorf("Failed to get meta partition (%d)", mpid)
|
||||
sendErrReply(w, r, newErrHTTPReply(err))
|
||||
return
|
||||
}
|
||||
|
||||
ret := MigrateResult{
|
||||
Mp: MetaReplicaInfo{
|
||||
MaxInodeID: mp.MaxInodeID,
|
||||
InodeCount: mp.InodeCount,
|
||||
DentryCount: mp.DentryCount,
|
||||
FreeListLen: mp.FreeListLen,
|
||||
TxCnt: mp.TxCnt,
|
||||
TxRbInoCnt: mp.TxRbInoCnt,
|
||||
TxRbDenCnt: mp.TxRbDenCnt,
|
||||
},
|
||||
}
|
||||
|
||||
metaNode, err := m.cluster.metaNode(targetAddr)
|
||||
if err != nil {
|
||||
err = fmt.Errorf("Failed to get meta partition (%d)", mpid)
|
||||
sendErrReply(w, r, newErrHTTPReply(err))
|
||||
return
|
||||
}
|
||||
|
||||
for _, mpv := range metaNode.metaPartitionInfos {
|
||||
if mpv.PartitionID != mpid {
|
||||
continue
|
||||
}
|
||||
ret.Target.MaxInodeID = mpv.MaxInodeID
|
||||
ret.Target.InodeCount = mpv.InodeCnt
|
||||
ret.Target.DentryCount = mpv.DentryCnt
|
||||
ret.Target.FreeListLen = mpv.FreeListLen
|
||||
ret.Target.TxCnt = mpv.TxCnt
|
||||
ret.Target.TxRbInoCnt = mpv.TxRbInoCnt
|
||||
ret.Target.TxRbDenCnt = mpv.TxRbDenCnt
|
||||
}
|
||||
|
||||
sendOkReply(w, r, newSuccessHTTPReply(ret))
|
||||
}
|
||||
|
||||
func (m *Server) createBalancePlan(w http.ResponseWriter, r *http.Request) {
|
||||
var err error
|
||||
metric := exporter.NewTPCnt(apiToMetricsName(proto.CreateBalanceTask))
|
||||
defer func() {
|
||||
doStatAndMetric(proto.CreateBalanceTask, metric, err, nil)
|
||||
}()
|
||||
|
||||
var plan *proto.ClusterPlan
|
||||
// search the raft storage. Only store one plan
|
||||
plan, err = m.cluster.loadBalanceTask()
|
||||
if err == nil && plan != nil {
|
||||
err = fmt.Errorf("There is a meta partition task plan already. Please remove it before create a new one.")
|
||||
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeInternalError, Msg: err.Error(), Data: plan})
|
||||
return
|
||||
}
|
||||
|
||||
plan, err = m.cluster.GetMetaNodePressureView()
|
||||
if err != nil {
|
||||
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeInternalError, Msg: err.Error(), Data: plan})
|
||||
return
|
||||
}
|
||||
|
||||
if plan.Total <= 0 {
|
||||
sendOkReply(w, r, newSuccessHTTPReply("Not find meta node that needs partition rebalance."))
|
||||
return
|
||||
}
|
||||
|
||||
// Save into raft storage.
|
||||
err = m.cluster.syncAddBalanceTask(plan)
|
||||
if err != nil {
|
||||
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeInternalError, Msg: err.Error(), Data: plan})
|
||||
return
|
||||
}
|
||||
|
||||
auditlog.LogMasterOp("createBalancePlan", "create meta partition balance task", nil)
|
||||
|
||||
sendOkReply(w, r, newSuccessHTTPReply(plan))
|
||||
}
|
||||
|
||||
func (m *Server) getBalancePlan(w http.ResponseWriter, r *http.Request) {
|
||||
var err error
|
||||
metric := exporter.NewTPCnt(apiToMetricsName(proto.GetBalanceTask))
|
||||
defer func() {
|
||||
doStatAndMetric(proto.GetBalanceTask, metric, err, nil)
|
||||
}()
|
||||
|
||||
var plan *proto.ClusterPlan
|
||||
plan, err = m.cluster.loadBalanceTask()
|
||||
if err != nil {
|
||||
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeInternalError, Msg: err.Error(), Data: plan})
|
||||
return
|
||||
}
|
||||
|
||||
if plan == nil || plan.Total <= 0 {
|
||||
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeInternalError, Msg: "Meta partition migrate plan doesn't existed."})
|
||||
return
|
||||
}
|
||||
|
||||
sendOkReply(w, r, newSuccessHTTPReply(plan))
|
||||
}
|
||||
|
||||
func (m *Server) runBalancePlan(w http.ResponseWriter, r *http.Request) {
|
||||
var err error
|
||||
metric := exporter.NewTPCnt(apiToMetricsName(proto.RunBalanceTask))
|
||||
defer func() {
|
||||
doStatAndMetric(proto.RunBalanceTask, metric, err, nil)
|
||||
}()
|
||||
|
||||
err = m.cluster.RunMetaPartitionBalanceTask()
|
||||
if err != nil {
|
||||
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeInternalError, Msg: err.Error()})
|
||||
return
|
||||
}
|
||||
|
||||
auditlog.LogMasterOp("runBalancePlan", "start to run meta partition balance task", nil)
|
||||
|
||||
sendOkReply(w, r, newSuccessHTTPReply("Start running balance task successfully."))
|
||||
}
|
||||
|
||||
func (m *Server) stopBalancePlan(w http.ResponseWriter, r *http.Request) {
|
||||
var err error
|
||||
metric := exporter.NewTPCnt(apiToMetricsName(proto.StopBalanceTask))
|
||||
defer func() {
|
||||
doStatAndMetric(proto.StopBalanceTask, metric, err, nil)
|
||||
}()
|
||||
|
||||
err = m.cluster.StopMetaPartitionBalanceTask()
|
||||
if err != nil {
|
||||
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeInternalError, Msg: err.Error()})
|
||||
return
|
||||
}
|
||||
|
||||
auditlog.LogMasterOp("stopBalancePlan", "stop meta partition balance task", nil)
|
||||
|
||||
sendOkReply(w, r, newSuccessHTTPReply("Stop balance task successfully."))
|
||||
}
|
||||
|
||||
func (m *Server) deleteBalancePlan(w http.ResponseWriter, r *http.Request) {
|
||||
var err error
|
||||
metric := exporter.NewTPCnt(apiToMetricsName(proto.DeleteBalanceTask))
|
||||
defer func() {
|
||||
doStatAndMetric(proto.DeleteBalanceTask, metric, err, nil)
|
||||
}()
|
||||
|
||||
err = m.cluster.DeleteMetaPartitionBalanceTask()
|
||||
if err != nil {
|
||||
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeInternalError, Msg: err.Error()})
|
||||
return
|
||||
}
|
||||
|
||||
auditlog.LogMasterOp("deleteBalancePlan", "Remove meta partition balance task", nil)
|
||||
|
||||
sendOkReply(w, r, newSuccessHTTPReply("Delete balance plan task successfully."))
|
||||
}
|
||||
@ -1845,19 +1845,34 @@ func TestUpdateVolAutoDpMetaRepair(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestGetMetaPartitionEmptyStatus(t *testing.T) {
|
||||
name := "emptyMpVol"
|
||||
createVol(map[string]interface{}{nameKey: name}, t)
|
||||
vol, err := server.cluster.getVol(name)
|
||||
if err != nil {
|
||||
t.Errorf("failed to get vol %v, err %v", name, err)
|
||||
return
|
||||
}
|
||||
|
||||
if err = vol.addMetaPartitions(server.cluster, 3); err != nil {
|
||||
t.Errorf("failed to get vol %v, err %v", name, err)
|
||||
return
|
||||
}
|
||||
|
||||
reqUrl := fmt.Sprintf("%v%v", hostAddr, proto.AdminMetaPartitionEmptyStatus)
|
||||
|
||||
process(reqUrl, t)
|
||||
}
|
||||
|
||||
func TestMetaPartitionFreezeEmpty(t *testing.T) {
|
||||
reqUrl := fmt.Sprintf("%v%v?name=setDiscardVol&count=3", hostAddr, proto.AdminMetaPartitionFreezeEmpty)
|
||||
reqUrl := fmt.Sprintf("%v%v?name=emptyMpVol&count=3", hostAddr, proto.AdminMetaPartitionFreezeEmpty)
|
||||
|
||||
process(reqUrl, t)
|
||||
}
|
||||
|
||||
func TestCleanEmptyMetaPartition(t *testing.T) {
|
||||
reqUrl := fmt.Sprintf("%v%v?name=setDiscardVol", hostAddr, proto.AdminMetaPartitionCleanEmpty)
|
||||
name := "emptyMpVol2"
|
||||
createVol(map[string]interface{}{nameKey: name}, t)
|
||||
reqUrl := fmt.Sprintf("%v%v?name=emptyMpVol2", hostAddr, proto.AdminMetaPartitionCleanEmpty)
|
||||
|
||||
process(reqUrl, t)
|
||||
}
|
||||
|
||||
@ -6339,224 +6339,3 @@ func (c *Cluster) checkMultipleReplicasOnSameMachine(hosts []string) (err error)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Cluster) FreezeEmptyMetaPartitionJob(name string, freezeList []*MetaPartition) error {
|
||||
c.mu.Lock()
|
||||
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)
|
||||
}
|
||||
task.Status = CleanTaskFreezing
|
||||
} else {
|
||||
task = &CleanTask{
|
||||
Name: name,
|
||||
Status: CleanTaskFreezing,
|
||||
TaskCnt: len(freezeList),
|
||||
}
|
||||
c.cleanTask[name] = task
|
||||
}
|
||||
c.mu.Unlock()
|
||||
|
||||
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("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()
|
||||
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)
|
||||
}
|
||||
task.Status = CleanTaskBackuping
|
||||
} else {
|
||||
task = &CleanTask{
|
||||
Name: name,
|
||||
Status: CleanTaskBackuping,
|
||||
}
|
||||
c.cleanTask[name] = task
|
||||
}
|
||||
c.mu.Unlock()
|
||||
|
||||
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())
|
||||
return err
|
||||
}
|
||||
|
||||
if vol.Status == proto.VolStatusMarkDelete {
|
||||
log.LogInfof("volume(%s) is deleted before cleaned empty meta partitions.", name)
|
||||
return nil
|
||||
}
|
||||
|
||||
deleteMaps := make(map[uint64]*MetaPartition)
|
||||
mps := vol.cloneMetaPartitionMap()
|
||||
for key, mp := range mps {
|
||||
if !mp.IsFreeze {
|
||||
continue
|
||||
}
|
||||
|
||||
// restore back the mp status if it is written.
|
||||
if mp.InodeCount != 0 || mp.DentryCount != 0 {
|
||||
// freeze meta partition.
|
||||
err = c.FreezeEmptyMetaPartition(mp, false)
|
||||
if err != nil {
|
||||
log.LogErrorf("Failed to unfreeze volume(%s) meta partition(%d), error: %s", name, mp.PartitionID, err.Error())
|
||||
continue
|
||||
}
|
||||
|
||||
mp.IsFreeze = false
|
||||
// store the meta partition status.
|
||||
err = c.syncUpdateMetaPartition(mp)
|
||||
if err != nil {
|
||||
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 {
|
||||
log.LogErrorf("action[DoCleanEmptyMetaPartition] clean meta partition(%d) error: %s", mp.PartitionID, err.Error())
|
||||
continue
|
||||
}
|
||||
|
||||
deleteMaps[key] = mp
|
||||
task.CleanCnt += 1
|
||||
}
|
||||
}
|
||||
|
||||
vol.mpsLock.Lock()
|
||||
for key, val := range deleteMaps {
|
||||
c.syncDeleteMetaPartition(val)
|
||||
delete(vol.MetaPartitions, key)
|
||||
}
|
||||
vol.mpsLock.UnLock()
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Cluster) CleanEmptyMetaPartition(mp *MetaPartition) error {
|
||||
for _, replica := range mp.Replicas {
|
||||
task := replica.createTaskToBackupReplica(mp.PartitionID)
|
||||
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) 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
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@ -485,7 +485,7 @@ const (
|
||||
flashNodePrefix = keySeparator + "fn" + keySeparator
|
||||
flashGroupPrefix = keySeparator + "fg" + keySeparator
|
||||
|
||||
balanceTaskPrefix = keySeparator + "balanceTask" + keySeparator
|
||||
balanceTaskKey = keySeparator + "balanceTask"
|
||||
)
|
||||
|
||||
// selector enum
|
||||
|
||||
@ -2192,61 +2192,58 @@ func (c *Cluster) loadLcResults() (err error) {
|
||||
return
|
||||
}
|
||||
|
||||
// key=#balanceTask#,value=json.Marshal(ClusterPlan)
|
||||
func (c *Cluster) syncAddBalanceTask(task *ClusterPlan) (err error) {
|
||||
// key=#balanceTask,value=json.Marshal(ClusterPlan)
|
||||
func (c *Cluster) syncAddBalanceTask(task *proto.ClusterPlan) (err error) {
|
||||
return c.putBalanceTaskInfo(opSyncAddBalanceTask, task)
|
||||
}
|
||||
|
||||
func (c *Cluster) syncUpdateBalanceTask(task *ClusterPlan) (err error) {
|
||||
func (c *Cluster) syncUpdateBalanceTask(task *proto.ClusterPlan) (err error) {
|
||||
return c.putBalanceTaskInfo(opSyncUpdateBalanceTask, task)
|
||||
}
|
||||
|
||||
func (c *Cluster) putBalanceTaskInfo(opType uint32, task *ClusterPlan) (err error) {
|
||||
func (c *Cluster) putBalanceTaskInfo(opType uint32, task *proto.ClusterPlan) error {
|
||||
balanceTask, err := c.buildBalanceTaskRaftCmd(opType, task)
|
||||
if err != nil {
|
||||
return
|
||||
return err
|
||||
}
|
||||
return c.submit(balanceTask)
|
||||
}
|
||||
|
||||
func (c *Cluster) buildBalanceTaskRaftCmd(opType uint32, task *ClusterPlan) (balanceTask *RaftCmd, err error) {
|
||||
balanceTask = new(RaftCmd)
|
||||
func (c *Cluster) buildBalanceTaskRaftCmd(opType uint32, task *proto.ClusterPlan) (*RaftCmd, error) {
|
||||
balanceTask := new(RaftCmd)
|
||||
balanceTask.Op = opType
|
||||
balanceTask.K = balanceTaskPrefix
|
||||
balanceTask.K = balanceTaskKey
|
||||
var err error
|
||||
if balanceTask.V, err = json.Marshal(task); err != nil {
|
||||
return balanceTask, errors.New(err.Error())
|
||||
return nil, fmt.Errorf("balance task op(%d) encode err: %s", opType, err.Error())
|
||||
}
|
||||
return
|
||||
return balanceTask, nil
|
||||
}
|
||||
|
||||
func (c *Cluster) loadBalanceTask() (task *ClusterPlan, err error) {
|
||||
result, err := c.fsm.store.Get([]byte(balanceTaskPrefix))
|
||||
func (c *Cluster) loadBalanceTask() (*proto.ClusterPlan, error) {
|
||||
result, err := c.fsm.store.GetByKey([]byte(balanceTaskKey))
|
||||
if err != nil {
|
||||
err = fmt.Errorf("balanceTaskPrefix get failed, err [%s]", err.Error())
|
||||
return nil, err
|
||||
return nil, fmt.Errorf("loadBalanceTask GetByKey err: %s", err.Error())
|
||||
}
|
||||
|
||||
err = json.Unmarshal(result.([]byte), task)
|
||||
if err != nil {
|
||||
err = fmt.Errorf("loadBalanceTask get failed, err [%s]", err.Error())
|
||||
return nil, err
|
||||
if len(result) == 0 {
|
||||
return nil, fmt.Errorf("Not find balance task.")
|
||||
}
|
||||
|
||||
return
|
||||
task := new(proto.ClusterPlan)
|
||||
err = json.Unmarshal(result, task)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("loadBalanceTask decode json err: %s", err.Error())
|
||||
}
|
||||
|
||||
return task, nil
|
||||
}
|
||||
|
||||
func (c *Cluster) syncDeleteBalanceTask() (task *ClusterPlan, err error) {
|
||||
result, err := c.fsm.store.Del(balanceTaskPrefix, true)
|
||||
func (c *Cluster) syncDeleteBalanceTask() error {
|
||||
err := c.fsm.store.DelByKey([]byte(balanceTaskKey), true)
|
||||
if err != nil {
|
||||
err = fmt.Errorf("delete balanceTaskPrefix err [%s]", err.Error())
|
||||
return nil, err
|
||||
log.LogErrorf("DelByKey err: %s", err.Error())
|
||||
}
|
||||
|
||||
err = json.Unmarshal(result.([]byte), task)
|
||||
if err != nil {
|
||||
err = fmt.Errorf("loadBalanceTask get failed, err [%s]", err.Error())
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return
|
||||
return err
|
||||
}
|
||||
|
||||
60
proto/cluster_balance.go
Normal file
60
proto/cluster_balance.go
Normal file
@ -0,0 +1,60 @@
|
||||
package proto
|
||||
|
||||
type MetaReplicaRec struct {
|
||||
Source string `json:"source" bson:"source"`
|
||||
SrcMemSize uint64 `json:"srcMemSize" bson:"srcmemsize"`
|
||||
SrcNodeSetId uint64 `json:"srcNodeSetId" bson:"srcnodesetid"`
|
||||
SrcZoneName string `json:"srcZoneName" bson:"srczonename"`
|
||||
Destination string `json:"destination" bson:"destination"`
|
||||
DstId uint64 `json:"dstID" bson:"dstid"`
|
||||
DstNodeSetId uint64 `json:"dstNodeSetId" bson:"dstnodesetid"`
|
||||
DstZoneName string `json:"dstZoneName" bson:"dstzonename"`
|
||||
Status string `json:"status" bson:"status"`
|
||||
}
|
||||
|
||||
type MetaPartitionPlan struct {
|
||||
ID uint64 `json:"id" bson:"id"`
|
||||
CrossZone bool `json:"crossZone" bson:"crosszone"`
|
||||
Original []*MetaReplicaRec `json:"original" bson:"original"`
|
||||
OverLoad []*MetaReplicaRec `json:"overLoad" bson:"overload"`
|
||||
Plan []*MetaReplicaRec `json:"plan" bson:"plan"`
|
||||
PlanNum int `json:"planNum" bson:"plannum"`
|
||||
InodeCount uint64 `json:"inodeCount" bson:"inodecount"`
|
||||
}
|
||||
|
||||
type MetaNodeRec struct {
|
||||
ID uint64 `json:"id"`
|
||||
Addr string `json:"address"`
|
||||
DomainAddr string `json:"domainAddress"`
|
||||
ZoneName string `json:"zone"`
|
||||
NodeSetID uint64 `json:"nodeSetId"`
|
||||
Total uint64 `json:"totalMem"`
|
||||
Used uint64 `json:"usedMem"`
|
||||
Free uint64 `json:"freeMem"`
|
||||
Ratio float64 `json:"ratio"`
|
||||
MpCount int `json:"mpCount"`
|
||||
MetaPartitions []uint64 `json:"metaPartition"`
|
||||
InodeCount uint64 `json:"inodeCount"`
|
||||
Estimate int `json:"estimate"`
|
||||
PlanCnt int `json:"planCount"`
|
||||
}
|
||||
|
||||
type NodeSetPressureView struct {
|
||||
NodeSetID uint64 `json:"nodeSetId"`
|
||||
Number int `json:"number"`
|
||||
MetaNodes map[uint64]*MetaNodeRec `json:"metaNodes"`
|
||||
}
|
||||
|
||||
type ZonePressureView struct {
|
||||
ZoneName string `json:"zone"`
|
||||
Status string `json:"status"`
|
||||
NodeSet map[uint64]*NodeSetPressureView `json:"nodeSet"`
|
||||
}
|
||||
|
||||
type ClusterPlan struct {
|
||||
Low map[string]*ZonePressureView `json:"-" bson:"-"`
|
||||
Plan []*MetaPartitionPlan `json:"plan" bson:"plan"`
|
||||
DoneNum int `json:"doneCount" bson:"donenum"`
|
||||
Total int `json:"total" bson:"total"`
|
||||
Status string `json:"status" bson:"status"`
|
||||
}
|
||||
@ -300,3 +300,23 @@ func (rs *RocksDBStore) Clear() (err error) {
|
||||
err = rs.db.Write(wo, wb)
|
||||
return
|
||||
}
|
||||
|
||||
// Get returns the value based on the given key.
|
||||
func (rs *RocksDBStore) GetByKey(key []byte) ([]byte, error) {
|
||||
ro := gorocksdb.NewDefaultReadOptions()
|
||||
ro.SetFillCache(false)
|
||||
defer ro.Destroy()
|
||||
return rs.db.GetBytes(ro, key)
|
||||
}
|
||||
|
||||
// Del deletes a key-value pair.
|
||||
func (rs *RocksDBStore) DelByKey(key []byte, isSync bool) (err error) {
|
||||
wo := gorocksdb.NewDefaultWriteOptions()
|
||||
wo.SetSync(isSync)
|
||||
defer func() {
|
||||
wo.Destroy()
|
||||
}()
|
||||
|
||||
err = rs.db.Delete(wo, key)
|
||||
return
|
||||
}
|
||||
|
||||
@ -1002,3 +1002,36 @@ func (api *AdminAPI) ClientFlashGroups() (fgView proto.FlashGroupView, err error
|
||||
err = api.mc.requestWith(&fgView, newRequest(get, proto.ClientFlashGroups).Header(api.h))
|
||||
return
|
||||
}
|
||||
|
||||
func (api *AdminAPI) CreateBalanceTask() (task *proto.ClusterPlan, err error) {
|
||||
task = &proto.ClusterPlan{
|
||||
Low: make(map[string]*proto.ZonePressureView, 0),
|
||||
Plan: make([]*proto.MetaPartitionPlan, 0),
|
||||
}
|
||||
err = api.mc.requestWith(task, newRequest(get, proto.CreateBalanceTask).Header(api.h))
|
||||
return
|
||||
}
|
||||
|
||||
func (api *AdminAPI) GetBalanceTask() (task *proto.ClusterPlan, err error) {
|
||||
task = &proto.ClusterPlan{
|
||||
Low: make(map[string]*proto.ZonePressureView, 0),
|
||||
Plan: make([]*proto.MetaPartitionPlan, 0),
|
||||
}
|
||||
err = api.mc.requestWith(task, newRequest(get, proto.GetBalanceTask).Header(api.h))
|
||||
return
|
||||
}
|
||||
|
||||
func (api *AdminAPI) RunBalanceTask() (result string, err error) {
|
||||
err = api.mc.requestWith(&result, newRequest(get, proto.RunBalanceTask).Header(api.h))
|
||||
return
|
||||
}
|
||||
|
||||
func (api *AdminAPI) StopBalanceTask() (result string, err error) {
|
||||
err = api.mc.requestWith(&result, newRequest(get, proto.StopBalanceTask).Header(api.h))
|
||||
return
|
||||
}
|
||||
|
||||
func (api *AdminAPI) DeleteBalanceTask() (result string, err error) {
|
||||
err = api.mc.requestWith(&result, newRequest(get, proto.DeleteBalanceTask).Header(api.h))
|
||||
return
|
||||
}
|
||||
|
||||
Loading…
Reference in New Issue
Block a user
