diff --git a/datanode/disk.go b/datanode/disk.go index 7768b6f23..b260af32f 100644 --- a/datanode/disk.go +++ b/datanode/disk.go @@ -298,12 +298,15 @@ func (d *Disk) updateQosLimiter() { } if d.dataNode.diskAsyncReadIops > 0 { d.limitFactor[proto.IopsAsyncReadType].SetLimit(rate.Limit(d.dataNode.diskAsyncReadIops)) + d.limitFactor[proto.IopsAsyncReadType].SetBurst(d.dataNode.diskAsyncReadIops / 2) } if d.dataNode.diskAsyncWriteIops > 0 { d.limitFactor[proto.IopsAsyncWriteType].SetLimit(rate.Limit(d.dataNode.diskAsyncWriteIops)) + d.limitFactor[proto.IopsAsyncWriteType].SetBurst(d.dataNode.diskAsyncWriteIops / 2) } if d.dataNode.diskDeleteIops > 0 { d.limitFactor[proto.IopsDeleteType].SetLimit(rate.Limit(d.dataNode.diskDeleteIops)) + d.limitFactor[proto.IopsDeleteType].SetBurst(d.dataNode.diskDeleteIops / 2) } for i := proto.IopsReadType; i < proto.FlowDeleteType; i++ { log.LogInfof("action[updateQosLimiter] type %v limit %v", proto.QosTypeString(i), d.limitFactor[i].Limit()) @@ -338,6 +341,9 @@ func (d *Disk) allocCheckAsyncLimit(factorType uint32, used uint32) error { if !d.dataNode.diskAsyncQosEnable { return nil } + if factorType == proto.FlowAsyncReadType || factorType == proto.FlowAsyncWriteType { + return nil + } ctx := context.Background() d.limitFactor[factorType].WaitN(ctx, int(used)) diff --git a/metanode/api_handler.go b/metanode/api_handler.go index 58f5b6f9d..e598d5f3c 100644 --- a/metanode/api_handler.go +++ b/metanode/api_handler.go @@ -95,6 +95,9 @@ func (m *MetaNode) registerAPIHandler() (err error) { http.HandleFunc("/setGOGC", m.setGOGCHandler) http.HandleFunc("/getGOGC", m.getGOGCHandler) http.HandleFunc("/reloadMp", m.reloadMpHandler) + http.HandleFunc("/setQosEnable", m.setQosEnableHandler) + http.HandleFunc("/setMetaQos", m.setMetaQosHandler) + http.HandleFunc("/getMetaQos", m.getMetaQosHandler) return } @@ -1270,7 +1273,7 @@ func (m *MetaNode) getGOGCHandler(w http.ResponseWriter, r *http.Request) { } data, _ := resp.Marshal() if _, err := w.Write(data); err != nil { - log.LogErrorf("[setGOGCHandler] response %s", err) + log.LogErrorf("[getGOGCHandler] response %s", err) } }() if m.metadataManager == nil { @@ -1310,3 +1313,113 @@ func (m *MetaNode) reloadMpHandler(w http.ResponseWriter, r *http.Request) { } err = m.metadataManager.ReloadPartition(id) } + +func (m *MetaNode) setQosEnableHandler(w http.ResponseWriter, r *http.Request) { + const ( + paramEnable = "enable" + ) + var ( + enable bool + err error + ) + resp := NewAPIResponse(http.StatusOK, http.StatusText(http.StatusOK)) + defer func() { + if err != nil { + resp.Msg = err.Error() + resp.Code = http.StatusBadRequest + } + data, _ := resp.Marshal() + if _, err := w.Write(data); err != nil { + log.LogErrorf("[setQosEnalbeHandler] response %s", err) + } + }() + if err = r.ParseForm(); err != nil { + return + } + enable, err = strconv.ParseBool(r.FormValue(paramEnable)) + if err != nil { + err = fmt.Errorf("parse param %v fail: %v", enable, err) + return + } + m.qosEnable = enable + log.LogWarnf("[setQosEnable] change qosEnable to %v success", m.qosEnable) +} + +func (m *MetaNode) setMetaQosHandler(w http.ResponseWriter, r *http.Request) { + const ( + paramReadDirIops = "readDirIops" + ) + var err error + + resp := NewAPIResponse(http.StatusOK, http.StatusText(http.StatusOK)) + defer func() { + if err != nil { + resp.Msg = err.Error() + resp.Code = http.StatusBadRequest + } + data, _ := resp.Marshal() + if _, err := w.Write(data); err != nil { + log.LogErrorf("[setMetaQosHandler] response %s", err) + } + }() + if err = r.ParseForm(); err != nil { + return + } + + parser := func(key string) (val int, err error, has bool) { + valStr := r.FormValue(key) + if valStr == "" { + return 0, nil, false + } + has = true + val, err = strconv.Atoi(valStr) + return + } + + updated := false + for key, pVal := range map[string]*int{ + paramReadDirIops: &m.readDirIops, + } { + val, err, has := parser(key) + if err != nil { + return + } + if has { + updated = true + *pVal = val + } + } + + if updated { + if m.metadataManager == nil { + err = fmt.Errorf("metadataManager is nil") + return + } + m.metadataManager.UpdateQosLimit() + } +} + +func (m *MetaNode) getMetaQosHandler(w http.ResponseWriter, r *http.Request) { + var err error + resp := NewAPIResponse(http.StatusOK, http.StatusText(http.StatusOK)) + defer func() { + if err != nil { + resp.Msg = err.Error() + resp.Code = http.StatusBadRequest + } + data, _ := resp.Marshal() + if _, err := w.Write(data); err != nil { + log.LogErrorf("[getMetaQosHandler] response %s", err) + } + }() + + metaQos := &struct { + QosEnable bool `json:"qosEnable"` + ReadDirIops int `json:"readDirIops"` + }{ + QosEnable: m.qosEnable, + ReadDirIops: m.readDirIops, + } + + resp.Data = metaQos +} diff --git a/metanode/const.go b/metanode/const.go index 0177ab97f..95be44064 100644 --- a/metanode/const.go +++ b/metanode/const.go @@ -249,6 +249,8 @@ const ( cfgServiceIDKey = "serviceIDKey" cfgEnableGcTimer = "enableGcTimer" // bool CfgGcRecyclePercent = "gcRecyclePercent" + cfsQosEnable = "qosEnable" // bool + cfgReadDirIops = "readDirIops" // int metaNodeDeleteBatchCountKey = "batchCount" configNameResolveInterval = "nameResolveInterval" // int diff --git a/metanode/manager.go b/metanode/manager.go index 106d5b1e5..c3748ab22 100644 --- a/metanode/manager.go +++ b/metanode/manager.go @@ -38,6 +38,7 @@ import ( "github.com/cubefs/cubefs/util/loadutil" "github.com/cubefs/cubefs/util/log" "github.com/cubefs/cubefs/util/strutil" + "golang.org/x/time/rate" ) const ( @@ -52,8 +53,11 @@ const UpdateVolTicket = 2 * time.Minute const ( gcTimerDuration = 10 * time.Second defaultGcRecyclePercent = 0.90 + defaultReadDirIops = 100 ) +var TryAgainError = errors.New("try again") + // MetadataManager manages all the meta partitions. type MetadataManager interface { Start() error @@ -65,6 +69,7 @@ type MetadataManager interface { GetAllVolumes() (volumes *util.Set) checkVolVerList() (err error) ReloadPartition(id int) (err error) + UpdateQosLimit() } // MetadataManagerConfig defines the configures in the metadata manager. @@ -108,6 +113,7 @@ type metadataManager struct { gogcValue int gcRecyclePercent float64 gcTimer *util.RecycleTimer + limitFactor map[uint32]*rate.Limiter } func (m *metadataManager) GetAllVolumes() (volumes *util.Set) { @@ -881,9 +887,30 @@ func (m *metadataManager) GetLeaderPartitions() map[uint64]MetaPartition { return mps } +func (m *metadataManager) allocCheckLimit(factorType uint32) error { + if !m.metaNode.qosEnable { + return nil + } + + if !m.limitFactor[factorType].Allow() { + return TryAgainError + } + // ctx := context.Background() + // m.limitFactor[factorType].WaitN(ctx, int(used)) + return nil +} + +func (m *metadataManager) UpdateQosLimit() { + if m.metaNode.readDirIops > 0 { + m.limitFactor[readDirIops].SetLimit(rate.Limit(m.metaNode.readDirIops)) + m.limitFactor[readDirIops].SetBurst(m.metaNode.readDirIops / 2) + } + log.LogWarnf("[UpdataQosLimit] update readDirIops [%v]", m.metaNode.readDirIops) +} + // NewMetadataManager returns a new metadata manager. func NewMetadataManager(conf MetadataManagerConfig, metaNode *MetaNode) MetadataManager { - return &metadataManager{ + m := &metadataManager{ nodeId: conf.NodeID, zoneName: conf.ZoneName, rootDir: conf.RootDir, @@ -895,7 +922,11 @@ func NewMetadataManager(conf MetadataManagerConfig, metaNode *MetaNode) Metadata gogcValue: DefaultGOGCValue, enableGcTimer: conf.EnableGcTimer, gcRecyclePercent: conf.GcRecyclePercent, + limitFactor: make(map[uint32]*rate.Limiter), } + m.limitFactor[readDirIops] = rate.NewLimiter(rate.Limit(metaNode.readDirIops), metaNode.readDirIops/2) + + return m } // isExpiredPartition return whether one partition is expired diff --git a/metanode/manager_op.go b/metanode/manager_op.go index 73a0ebc1b..b8f3c5444 100644 --- a/metanode/manager_op.go +++ b/metanode/manager_op.go @@ -1017,6 +1017,13 @@ func (m *metadataManager) opReadDirOnly(conn net.Conn, p *Packet, if !m.serveProxy(conn, mp, p) { return } + err = m.allocCheckLimit(readDirIops) + if err != nil { + p.PacketErrorWithBody(proto.OpErr, ([]byte)(err.Error())) + m.respondToClient(conn, p) + log.LogWarnf("[%v],req[%v],err[%v]", p.GetOpMsgWithReqAndResult(), req, string(p.Data)) + return + } err = mp.ReadDirOnly(req, p) m.respondToClient(conn, p) log.LogDebugf("%s [%v]req: %v , resp: %v, body: %s", remoteAddr, @@ -1045,6 +1052,13 @@ func (m *metadataManager) opReadDir(conn net.Conn, p *Packet, if !m.serveProxy(conn, mp, p) { return } + err = m.allocCheckLimit(readDirIops) + if err != nil { + p.PacketErrorWithBody(proto.OpErr, ([]byte)(err.Error())) + m.respondToClient(conn, p) + log.LogWarnf("[%v],req[%v],err[%v]", p.GetOpMsgWithReqAndResult(), req, string(p.Data)) + return + } err = mp.ReadDir(req, p) m.respondToClient(conn, p) log.LogDebugf("%s [%v]req: %v , resp: %v, body: %s", remoteAddr, diff --git a/metanode/metanode.go b/metanode/metanode.go index 6cd2e5957..fb5839bcb 100644 --- a/metanode/metanode.go +++ b/metanode/metanode.go @@ -49,6 +49,10 @@ var ( legacyReplicaStorageClass uint32 // for compatibility when older version upgrade to hybrid cloud ) +const ( + readDirIops uint32 = 0x01 +) + // only used for mp check func SetLegacyType(t uint32) { legacyReplicaStorageClass = t @@ -84,6 +88,8 @@ type MetaNode struct { serviceIDKey string nodeForbidWriteOpOfProtoVer0 bool // whether forbid by node granularity, VolsForbidWriteOpOfProtoVer0 map[string]struct{} // whether forbid by volume granularity, + qosEnable bool + readDirIops int control common.Control } @@ -269,6 +275,14 @@ func (m *MetaNode) parseConfig(cfg *config.Config) (err error) { return fmt.Errorf("bad cfgRaftReplicaPort config") } + m.qosEnable = cfg.GetBoolWithDefault(cfsQosEnable, true) + m.readDirIops = cfg.GetInt(cfgReadDirIops) + if m.readDirIops <= 0 { + m.readDirIops = defaultReadDirIops + } + syslog.Printf("conf qosEnable=%v readDirIops=%v", m.qosEnable, m.readDirIops) + log.LogInfof("[parseConfig] qosEnable[%v] readDirIops[%v]", m.qosEnable, m.readDirIops) + raftRetainLogs := cfg.GetString(cfgRetainLogs) if raftRetainLogs != "" { if m.raftRetainLogs, err = strconv.ParseUint(raftRetainLogs, 10, 64); err != nil { @@ -446,7 +460,6 @@ func (m *MetaNode) newMetaManager(cfg *config.Config) (err error) { log.LogError(err.Error()) return err } - } // load metadataManager diff --git a/metanode/partition_test.go b/metanode/partition_test.go index 3aa8274c0..0ea6115f3 100644 --- a/metanode/partition_test.go +++ b/metanode/partition_test.go @@ -15,12 +15,16 @@ package metanode import ( + "encoding/json" + "math" "os" "path" + "sync" "testing" "time" "github.com/stretchr/testify/require" + "golang.org/x/time/rate" "github.com/cubefs/cubefs/proto" "github.com/cubefs/cubefs/util/fileutil" @@ -247,3 +251,101 @@ func TestDoFileStats(t *testing.T) { duration := time.Since(startTime) t.Logf("DoFileStats cost time %v", duration) } + +func TestLimitReadDir(t *testing.T) { + testPath := "/tmp/testMetaPartition/" + os.RemoveAll(testPath) + defer os.RemoveAll(testPath) + mpC := &MetaPartitionConfig{ + PartitionId: 1, + VolName: "test_vol", + Start: 0, + End: 100, + PartitionType: 1, + Peers: nil, + RootDir: testPath, + } + metaM := &metadataManager{ + nodeId: 1, + zoneName: "test", + raftStore: nil, + partitions: make(map[uint64]MetaPartition), + metaNode: &MetaNode{qosEnable: true}, + fileStatsConfig: &fileStatsConfig{}, + limitFactor: make(map[uint32]*rate.Limiter), + } + metaM.limitFactor[readDirIops] = rate.NewLimiter(rate.Limit(2), 10) + + partition := NewMetaPartition(mpC, metaM) + require.NotNil(t, partition) + mp, ok := partition.(*metaPartition) + require.True(t, ok) + t.Logf("readDirIops:%v", mp.manager.limitFactor[readDirIops].Limit()) + + req := &ReadDirLimitReq{ + PartitionID: partitionId, + VolName: mp.GetVolName(), + ParentID: 1, + Limit: math.MaxUint64, + VerSeq: 0, + } + + const totalRequests = 20 + var wg sync.WaitGroup + start := time.Now() + + for i := 0; i < totalRequests; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for { + err := mp.manager.allocCheckLimit(readDirIops) + if err == TryAgainError { + time.Sleep(time.Millisecond) + continue + } + resp := mp.readDirLimit(req) + _, err = json.Marshal(resp) + if err != nil { + t.Errorf("readDir err: %v", err) + } + return + } + }() + } + + wg.Wait() + costTime1 := time.Since(start) + + t.Logf("costTime1: %v", costTime1) + + mp.manager.limitFactor[readDirIops].SetLimit(rate.Limit(10)) + t.Logf("readDirIops:%v", mp.manager.limitFactor[readDirIops].Limit()) + // mp.manager.metaNode.qosEnable = true + start = time.Now() + + for i := 0; i < totalRequests; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for { + err := mp.manager.allocCheckLimit(readDirIops) + if err == TryAgainError { + time.Sleep(time.Millisecond) + continue + } + resp := mp.readDirLimit(req) + _, err = json.Marshal(resp) + if err != nil { + t.Errorf("readDir err: %v", err) + } + return + } + }() + } + wg.Wait() + costTime2 := time.Since(start) + t.Logf("costTime2: %v", costTime2) + + require.True(t, costTime1 > costTime2) +}