feat(meta): Add readDir iops limiter. #1000187434

Signed-off-by: zhumingze <zhumingze@oppo.com>

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))

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
+}

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

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

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,

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

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)
+}