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(flash): set read and write flow by master

Browse Source 
with: #1000285575

Signed-off-by: clinx <chenlin1@oppo.com>
This commit is contained in:
clinx 2025-08-15 11:08:22 +08:00 committed by chihe
parent 042d9a15ab
commit 763517e3c1
24 changed files with 273 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
cli/cmd/cluster.go
View File

@ -297,6 +297,8 @@ func newClusterSetParasCmd(client *master.MasterClient) *cobra.Command {
optFlashNodeTimeoutCount := ""
optRemoteCacheSameZoneTimeout := ""
optRemoteCacheSameRegionTimeout := ""
optFlashReadFlowLimit := ""
optFlashWriteFlowLimit := ""
cmd := &cobra.Command{
Use: CliOpSetCluster,
Short: cmdClusterSetClusterInfoShort,
@ -461,6 +463,29 @@ func newClusterSetParasCmd(client *master.MasterClient) *cobra.Command {
return
}
}
if optFlashReadFlowLimit != "" {
if tmp, err = strconv.ParseInt(optFlashReadFlowLimit, 10, 64); err != nil {
err = fmt.Errorf("param (%v) failed, should be int", optFlashReadFlowLimit)
return
}
if tmp < 0 {
err = fmt.Errorf("param flashReadFlowLimit(%v) must greater than or equal to 0", optFlashReadFlowLimit)
return
}
}
if optFlashWriteFlowLimit != "" {
if tmp, err = strconv.ParseInt(optFlashWriteFlowLimit, 10, 64); err != nil {
err = fmt.Errorf("param (%v) failed, should be int", optFlashWriteFlowLimit)
return
}
if tmp < 0 {
err = fmt.Errorf("param flashWriteFlowLimit(%v) must greater than or equal to 0", optFlashWriteFlowLimit)
return
}
}
if optRemoteCacheMultiRead != "" {
if _, err = strconv.ParseBool(optRemoteCacheMultiRead); err != nil {
err = fmt.Errorf("param remoteCacheMultiRead(%v) should be true or false", optRemoteCacheMultiRead)
@ -505,7 +530,7 @@ func newClusterSetParasCmd(client *master.MasterClient) *cobra.Command {
dpRepairTimeout, dpTimeout, mpTimeout, dpBackupTimeout, decommissionDpLimit, decommissionDiskLimit,
forbidWriteOpOfProtoVersion0, dataMediaType, handleTimeout, readDataNodeTimeout,
optRcTTL, optRcReadTimeout, optRemoteCacheMultiRead, optFlashNodeTimeoutCount,
optRemoteCacheSameZoneTimeout, optRemoteCacheSameRegionTimeout, optFlashHotKeyMissCount); err != nil {
optRemoteCacheSameZoneTimeout, optRemoteCacheSameRegionTimeout, optFlashHotKeyMissCount, optFlashReadFlowLimit, optFlashWriteFlowLimit); err != nil {
return
}
stdout("Cluster parameters has been set successfully. \n")
@ -546,6 +571,8 @@ func newClusterSetParasCmd(client *master.MasterClient) *cobra.Command {
cmd.Flags().StringVar(&optRemoteCacheSameZoneTimeout, CliFlagRemoteCacheSameZoneTimeout, "", "Remote cache same zone timeout microsecond(must > 0)")
cmd.Flags().StringVar(&optRemoteCacheSameRegionTimeout, CliFlagRemoteCacheSameRegionTimeout, "", "Remote cache same region timeout millisecond(must > 0)")
cmd.Flags().StringVar(&optFlashHotKeyMissCount, CliFlagFlashHotKeyMissCount, "", "Flash hot key miss count(must > 0)")
cmd.Flags().StringVar(&optFlashReadFlowLimit, CliFlagFlashReadFlowLimit, "", "Flash read flow limit(must >= 0)")
cmd.Flags().StringVar(&optFlashWriteFlowLimit, CliFlagFlashWriteFlowLimit, "", "Flash write flow limit(must >= 0)")
return cmd
}

2
cli/cmd/const.go
View File

@ -175,6 +175,8 @@ const (
CliFlagRemoteCacheSameZoneTimeout = "remoteCacheSameZoneTimeout"
CliFlagRemoteCacheSameRegionTimeout = "remoteCacheSameRegionTimeout"
CliFlagFlashHotKeyMissCount = "flashHotKeyMissCount"
CliFlagFlashReadFlowLimit = "flashReadFlowLimit"
CliFlagFlashWriteFlowLimit = "flashWriteFlowLimit"
// CliFlagSetDataPartitionCount = "count" use dp-count instead

2
cli/cmd/fmt.go
View File

@ -105,6 +105,8 @@ func formatClusterView(cv *proto.ClusterView, cn *proto.ClusterNodeInfo, cp *pro
sb.WriteString(fmt.Sprintf(" RemoteCacheSameZoneTimeout : %v microsecond\n", cv.RemoteCacheSameZoneTimeout))
sb.WriteString(fmt.Sprintf(" RemoteCacheSameRegionTimeout : %v millisecond\n", cv.RemoteCacheSameRegionTimeout))
sb.WriteString(fmt.Sprintf(" FlashHotKeyMissCount : %v\n", cv.FlashHotKeyMissCount))
sb.WriteString(fmt.Sprintf(" FlashReadFlowLimit : %v\n", cv.FlashReadFlowLimit))
sb.WriteString(fmt.Sprintf(" FlashWriteFlowLimit : %v\n", cv.FlashWriteFlowLimit))
return sb.String()
}

24
master/api_args_parse.go
View File

@ -1718,6 +1718,28 @@ func parseAndExtractSetNodeInfoParams(r *http.Request) (params map[string]interf
params[forbidWriteOpOfProtoVersion0] = val
}
if value = r.FormValue(flashReadFlowLimit); value != "" {
noParams = false
val := int64(0)
val, err = strconv.ParseInt(value, 10, 64)
if err != nil {
err = unmatchedKey(flashReadFlowLimit)
return
}
params[flashReadFlowLimit] = val
}
if value = r.FormValue(flashWriteFlowLimit); value != "" {
noParams = false
val := int64(0)
val, err = strconv.ParseInt(value, 10, 64)
if err != nil {
err = unmatchedKey(flashWriteFlowLimit)
return
}
params[flashWriteFlowLimit] = val
}
if noParams {
err = fmt.Errorf("no key assigned")
return
@ -2164,6 +2186,8 @@ func parseSetConfigParam(r *http.Request) (config map[string]string, err error)
flashNodeHandleReadTimeout,
flashNodeReadDataNodeTimeout,
flashHotKeyMissCount,
flashReadFlowLimit,
flashWriteFlowLimit,
}
for _, val := range keyList {
key := val

47
master/api_service.go
View File

@ -968,6 +968,8 @@ func (m *Server) getCluster(w http.ResponseWriter, r *http.Request) {
FlashNodeHandleReadTimeout: m.cluster.cfg.flashNodeHandleReadTimeout,
FlashNodeReadDataNodeTimeout: m.cluster.cfg.flashNodeReadDataNodeTimeout,
FlashHotKeyMissCount: m.cluster.cfg.flashHotKeyMissCount,
FlashReadFlowLimit: m.cluster.cfg.flashReadFlowLimit,
FlashWriteFlowLimit: m.cluster.cfg.flashWriteFlowLimit,
}
vols := m.cluster.allVolNames()
@ -3690,6 +3692,24 @@ func (m *Server) setNodeInfoHandler(w http.ResponseWriter, r *http.Request) {
}
}
if val, ok := params[flashReadFlowLimit]; ok {
if v, ok := val.(int64); ok {
if err = m.setConfig(flashReadFlowLimit, strconv.FormatInt(v, 10)); err != nil {
sendErrReply(w, r, newErrHTTPReply(err))
return
}
}
}
if val, ok := params[flashWriteFlowLimit]; ok {
if v, ok := val.(int64); ok {
if err = m.setConfig(flashWriteFlowLimit, strconv.FormatInt(v, 10)); err != nil {
sendErrReply(w, r, newErrHTTPReply(err))
return
}
}
}
if val, ok := params[flashNodeReadDataNodeTimeout]; ok {
if v, ok := val.(int64); ok {
if err = m.setConfig(flashNodeReadDataNodeTimeout, strconv.FormatInt(v, 10)); err != nil {
@ -7186,7 +7206,10 @@ func (m *Server) setConfig(key string, value string) (err error) {
fnHandleReadTimeout int
fnReadDataNodeTimeout int
fnHotKeyMissCount int
fnReadFlowLimit int64
fnWriteFlowLimit int64
oldIntValue int
oldInt64Value int64
)
switch key {
@ -7243,6 +7266,22 @@ func (m *Server) setConfig(key string, value string) (err error) {
oldIntValue = m.config.flashHotKeyMissCount
m.config.flashHotKeyMissCount = fnHotKeyMissCount
case flashReadFlowLimit:
fnReadFlowLimit, err = strconv.ParseInt(value, 10, 64)
if err != nil {
return err
}
oldInt64Value = m.config.flashReadFlowLimit
m.config.flashReadFlowLimit = fnReadFlowLimit
case flashWriteFlowLimit:
fnWriteFlowLimit, err = strconv.ParseInt(value, 10, 64)
if err != nil {
return err
}
oldInt64Value = m.config.flashWriteFlowLimit
m.config.flashWriteFlowLimit = fnWriteFlowLimit
default:
err = keyNotFound("config")
return err
@ -7264,6 +7303,10 @@ func (m *Server) setConfig(key string, value string) (err error) {
m.config.flashNodeReadDataNodeTimeout = oldIntValue
case flashHotKeyMissCount:
m.config.flashHotKeyMissCount = oldIntValue
case flashReadFlowLimit:
m.config.flashReadFlowLimit = oldInt64Value
case flashWriteFlowLimit:
m.config.flashWriteFlowLimit = oldInt64Value
}
log.LogErrorf("setConfig syncPutCluster fail err %v", err)
return err
@ -7293,6 +7336,10 @@ func (m *Server) getConfig(key string) (value string, err error) {
value = strconv.Itoa(m.config.flashNodeReadDataNodeTimeout)
case flashHotKeyMissCount:
value = strconv.Itoa(m.config.flashHotKeyMissCount)
case flashReadFlowLimit:
value = strconv.FormatInt(m.config.flashReadFlowLimit, 10)
case flashWriteFlowLimit:
value = strconv.FormatInt(m.config.flashWriteFlowLimit, 10)
default:
err = keyNotFound("config")
}

10
master/config.go
View File

@ -82,7 +82,9 @@ const (
cfgDataNodeBalanceByDiskUsageLow = "dataNodeBalanceByDiskUsageLow"
cfgDataNodeBalanceByDPCountHigh = "dataNodeBalanceByDPCountHigh"
cfgDataNodeBalanceByDPCountLow = "dataNodeBalanceByDPCountLow"
flashHotKeyMissCount = "flashHotKeyMissCount"
flashHotKeyMissCount = "flashHotKeyMissCount"
flashReadFlowLimit = "flashReadFlowLimit"
flashWriteFlowLimit = "flashWriteFlowLimit"
)
// default value
@ -138,6 +140,8 @@ const (
defaultFlashNodeHandleReadTimeout = 1000
defaultFlashNodeReadDataNodeTimeout = 3000
defaultFlashHotKeyMissCount = 5
defaultFlashReadFlowLimit = 2147483648
defaultFlashWriteFlowLimit = 2147483648
defaultMetaNodeGOGC = 100
defaultDataNodeGOGC = 100
@ -217,6 +221,8 @@ type clusterConfig struct {
flashNodeHandleReadTimeout int
flashNodeReadDataNodeTimeout int
flashHotKeyMissCount int
flashReadFlowLimit int64
flashWriteFlowLimit int64
metaNodeGOGC int
dataNodeGOGC int
@ -269,6 +275,8 @@ func newClusterConfig() (cfg *clusterConfig) {
cfg.flashNodeHandleReadTimeout = defaultFlashNodeHandleReadTimeout
cfg.flashNodeReadDataNodeTimeout = defaultFlashNodeReadDataNodeTimeout
cfg.flashHotKeyMissCount = defaultFlashHotKeyMissCount
cfg.flashReadFlowLimit = defaultFlashReadFlowLimit
cfg.flashWriteFlowLimit = defaultFlashWriteFlowLimit
cfg.metaNodeGOGC = defaultMetaNodeGOGC
cfg.dataNodeGOGC = defaultDataNodeGOGC
cfg.metaNodeMemHighPer = defaultMetaNodeMemHighPer

6
master/flash_node.go
View File

@ -196,7 +196,7 @@ func (c *Cluster) checkFlashNodeHeartbeat() {
c.flashNodeTopo.flashNodeMap.Range(func(addr, flashNode interface{}) bool {
node := flashNode.(*FlashNode)
node.checkLiveliness()
task := node.createHeartbeatTask(c.masterAddr(), c.cfg.flashNodeHandleReadTimeout, c.cfg.flashNodeReadDataNodeTimeout, c.cfg.flashHotKeyMissCount)
task := node.createHeartbeatTask(c.masterAddr(), c.cfg.flashNodeHandleReadTimeout, c.cfg.flashNodeReadDataNodeTimeout, c.cfg.flashHotKeyMissCount, c.cfg.flashReadFlowLimit, c.cfg.flashWriteFlowLimit)
tasks = append(tasks, task)
return true
})
@ -213,7 +213,7 @@ func (flashNode *FlashNode) checkLiveliness() {
flashNode.Unlock()
}
func (flashNode *FlashNode) createHeartbeatTask(masterAddr string, flashNodeHandleReadTimeout int, flashNodeReadDataNodeTimeout int, flashHotKeyMissCount int) (task *proto.AdminTask) {
func (flashNode *FlashNode) createHeartbeatTask(masterAddr string, flashNodeHandleReadTimeout int, flashNodeReadDataNodeTimeout int, flashHotKeyMissCount int, flashReadFlowLimit int64, flashWriteFlowLimit int64) (task *proto.AdminTask) {
request := &proto.HeartBeatRequest{
CurrTime: time.Now().Unix(),
MasterAddr: masterAddr,
@ -221,6 +221,8 @@ func (flashNode *FlashNode) createHeartbeatTask(masterAddr string, flashNodeHand
request.FlashNodeHandleReadTimeout = flashNodeHandleReadTimeout
request.FlashNodeReadDataNodeTimeout = flashNodeReadDataNodeTimeout
request.FlashHotKeyMissCount = flashHotKeyMissCount
request.FlashReadFlowLimit = flashReadFlowLimit
request.FlashWriteFlowLimit = flashWriteFlowLimit
task = proto.NewAdminTask(proto.OpFlashNodeHeartbeat, flashNode.Addr, request)
return

11
master/metadata_fsm_op.go
View File

@ -83,6 +83,8 @@ type clusterValue struct {
FlashNodeHandleReadTimeout int
FlashNodeReadDataNodeTimeout int
FlashHotKeyMissCount int
FlashReadFlowLimit int64
FlashWriteFlowLimit int64
}
func newClusterValue(c *Cluster) (cv *clusterValue) {
@ -135,6 +137,8 @@ func newClusterValue(c *Cluster) (cv *clusterValue) {
FlashNodeHandleReadTimeout: c.cfg.flashNodeHandleReadTimeout,
FlashNodeReadDataNodeTimeout: c.cfg.flashNodeReadDataNodeTimeout,
FlashHotKeyMissCount: c.cfg.flashHotKeyMissCount,
FlashReadFlowLimit: c.cfg.flashReadFlowLimit,
FlashWriteFlowLimit: c.cfg.flashWriteFlowLimit,
}
return cv
}
@ -1430,9 +1434,12 @@ func (c *Cluster) loadClusterValue() (err error) {
}
c.cfg.flashHotKeyMissCount = cv.FlashHotKeyMissCount
c.cfg.flashReadFlowLimit = cv.FlashReadFlowLimit
c.cfg.flashWriteFlowLimit = cv.FlashWriteFlowLimit
c.cfg.flashNodeReadDataNodeTimeout = cv.FlashNodeReadDataNodeTimeout
log.LogInfof("action[loadClusterValue] flashNodeHandleReadTimeout %v(ms), flashNodeReadDataNodeTimeout %v(ms), flashHotKeyMissCount %v",
cv.FlashNodeHandleReadTimeout, cv.FlashNodeReadDataNodeTimeout, cv.FlashHotKeyMissCount)
log.LogInfof("action[loadClusterValue] flashNodeHandleReadTimeout %v(ms), flashNodeReadDataNodeTimeout %v(ms), flashHotKeyMissCount %v, flashReadFlowLimit %v, flashWriteFlowLimit %v",
cv.FlashNodeHandleReadTimeout, cv.FlashNodeReadDataNodeTimeout, cv.FlashHotKeyMissCount, cv.FlashReadFlowLimit, cv.FlashWriteFlowLimit)
}
return

2
proto/admin_proto.go
View File

@ -838,6 +838,8 @@ type FlashNodeHeartBeatInfos struct {
FlashNodeHandleReadTimeout int
FlashNodeReadDataNodeTimeout int
FlashHotKeyMissCount int
FlashReadFlowLimit int64
FlashWriteFlowLimit int64
}
// HeartBeatRequest define the heartbeat request.

2
proto/distributed_cache.go
View File

@ -427,6 +427,8 @@ type RemoteCacheConfig struct {
RemoteCacheSameZoneTimeout int64
RemoteCacheSameRegionTimeout int64
FlashHotKeyMissCount int
FlashReadFlowLimit int64
FlashWriteFlowLimit int64
}
func ComputeSourcesVersion(sources []*DataSource, gen uint64) (version uint32) {

2
proto/model.go
View File

@ -182,6 +182,8 @@ type ClusterView struct {
RemoteCacheSameZoneTimeout int64
RemoteCacheSameRegionTimeout int64
FlashHotKeyMissCount int
FlashReadFlowLimit int64
FlashWriteFlowLimit int64
}
// ClusterNode defines the structure of a cluster node

22
remotecache/flashgroupmanager/api_args_parse.go
View File

@ -145,6 +145,28 @@ func parseAndExtractSetNodeInfoParams(r *http.Request) (params map[string]interf
params[cfgFlashHotKeyMissCount] = val
}
if value = r.FormValue(cfgFlashReadFlowLimit); value != "" {
noParams = false
val := int64(0)
val, err = strconv.ParseInt(value, 10, 64)
if err != nil {
err = unmatchedKey(cfgFlashReadFlowLimit)
return
}
params[cfgFlashReadFlowLimit] = val
}
if value = r.FormValue(cfgFlashWriteFlowLimit); value != "" {
noParams = false
val := int64(0)
val, err = strconv.ParseInt(value, 10, 64)
if err != nil {
err = unmatchedKey(cfgFlashWriteFlowLimit)
return
}
params[cfgFlashWriteFlowLimit] = val
}
if value = r.FormValue(cfgFlashNodeReadDataNodeTimeout); value != "" {
noParams = false
val := int64(0)

44
remotecache/flashgroupmanager/api_service.go
View File

@ -60,6 +60,8 @@ func (m *FlashGroupManager) getCluster(w http.ResponseWriter, r *http.Request) {
FlashNodeHandleReadTimeout: m.cluster.cfg.FlashNodeHandleReadTimeout,
FlashNodeReadDataNodeTimeout: m.cluster.cfg.FlashNodeReadDataNodeTimeout,
FlashHotKeyMissCount: m.cluster.cfg.FlashHotKeyMissCount,
FlashReadFlowLimit: m.cluster.cfg.FlashReadFlowLimit,
FlashWriteFlowLimit: m.cluster.cfg.FlashWriteFlowLimit,
RemoteCacheTTL: m.config.RemoteCacheTTL,
RemoteCacheReadTimeout: m.config.RemoteCacheReadTimeout,
RemoteCacheMultiRead: m.config.RemoteCacheMultiRead,
@ -114,6 +116,8 @@ func (m *FlashGroupManager) getRemoteCacheConfig(w http.ResponseWriter, r *http.
RemoteCacheSameZoneTimeout: m.config.RemoteCacheSameZoneTimeout,
RemoteCacheSameRegionTimeout: m.config.RemoteCacheSameRegionTimeout,
FlashHotKeyMissCount: m.config.FlashHotKeyMissCount,
FlashReadFlowLimit: m.config.FlashReadFlowLimit,
FlashWriteFlowLimit: m.config.FlashWriteFlowLimit,
}
log.LogInfof("getRemoteCacheConfig config(%v)", config)
sendOkReply(w, r, newSuccessHTTPReply(config))
@ -776,6 +780,24 @@ func (m *FlashGroupManager) setNodeInfoHandler(w http.ResponseWriter, r *http.Re
}
}
if val, ok := params[cfgFlashReadFlowLimit]; ok {
if v, ok := val.(int64); ok {
if err = m.setConfig(cfgFlashReadFlowLimit, strconv.FormatInt(v, 10)); err != nil {
sendErrReply(w, r, newErrHTTPReply(err))
return
}
}
}
if val, ok := params[cfgFlashWriteFlowLimit]; ok {
if v, ok := val.(int64); ok {
if err = m.setConfig(cfgFlashWriteFlowLimit, strconv.FormatInt(v, 10)); err != nil {
sendErrReply(w, r, newErrHTTPReply(err))
return
}
}
}
sendOkReply(w, r, newSuccessHTTPReply(fmt.Sprintf("set nodeinfo params %v successfully", params)))
}
@ -790,6 +812,8 @@ func (m *FlashGroupManager) setConfig(key string, value string) (err error) {
flashNodeTimeoutCount int64
remoteCacheSameZoneTimeout int64
remoteCacheSameRegionTimeout int64
flashReadFlowLimit int64
flashWriteFlowLimit int64
oldIntValue int
oldInt64Value int64
oldBoolValue bool
@ -868,6 +892,22 @@ func (m *FlashGroupManager) setConfig(key string, value string) (err error) {
oldIntValue = m.config.FlashHotKeyMissCount
m.config.FlashHotKeyMissCount = flashHotKeyMissCount
case cfgFlashReadFlowLimit:
flashReadFlowLimit, err = strconv.ParseInt(value, 10, 64)
if err != nil {
return err
}
oldInt64Value = m.config.FlashReadFlowLimit
m.config.FlashReadFlowLimit = flashReadFlowLimit
case cfgFlashWriteFlowLimit:
flashWriteFlowLimit, err = strconv.ParseInt(value, 10, 64)
if err != nil {
return err
}
oldInt64Value = m.config.FlashWriteFlowLimit
m.config.FlashWriteFlowLimit = flashWriteFlowLimit
default:
err = keyNotFound("config")
return err
@ -893,6 +933,10 @@ func (m *FlashGroupManager) setConfig(key string, value string) (err error) {
m.config.RemoteCacheSameRegionTimeout = oldInt64Value
case cfgFlashHotKeyMissCount:
m.config.FlashHotKeyMissCount = oldIntValue
case cfgFlashReadFlowLimit:
m.config.FlashReadFlowLimit = oldInt64Value
case cfgFlashWriteFlowLimit:
m.config.FlashWriteFlowLimit = oldInt64Value
}
log.LogErrorf("setConfig syncPutCluster fail err %v", err)
return err

2
remotecache/flashgroupmanager/cluster.go
View File

@ -438,7 +438,7 @@ func (c *Cluster) checkFlashNodeHeartbeat() {
c.flashNodeTopo.flashNodeMap.Range(func(addr, flashNode interface{}) bool {
node := flashNode.(*FlashNode)
node.checkLiveliness()
task := node.createHeartbeatTask(c.masterAddr(), int(c.cfg.RemoteCacheReadTimeout), c.cfg.FlashNodeReadDataNodeTimeout, c.cfg.FlashHotKeyMissCount)
task := node.createHeartbeatTask(c.masterAddr(), int(c.cfg.RemoteCacheReadTimeout), c.cfg.FlashNodeReadDataNodeTimeout, c.cfg.FlashHotKeyMissCount, c.cfg.FlashReadFlowLimit, c.cfg.FlashWriteFlowLimit)
tasks = append(tasks, task)
return true
})

6
remotecache/flashgroupmanager/config.go
View File

@ -22,6 +22,8 @@ const (
defaultHttpReversePoolSize = 1024
defaultRetainLogs = 20000
defaultFlashHotKeyMissCount = 5
defaultFlashReadFlowLimit = 2147483648
defaultFlashWriteFlowLimit = 2147483648
)
const (
@ -34,6 +36,8 @@ const (
cfgRemoteCacheSameZoneTimeout = "remoteCacheSameZoneTimeout"
cfgRemoteCacheSameRegionTimeout = "remoteCacheSameRegionTimeout"
cfgFlashHotKeyMissCount = "flashHotKeyMissCount"
cfgFlashReadFlowLimit = "flashReadFlowLimit"
cfgFlashWriteFlowLimit = "flashWriteFlowLimit"
)
var AddrDatabase = make(map[uint64]string)
@ -58,6 +62,8 @@ func newClusterConfig() (cfg *clusterConfig) {
cfg.RemoteCacheSameZoneTimeout = cfsProto.DefaultRemoteCacheSameZoneTimeout
cfg.RemoteCacheSameRegionTimeout = cfsProto.DefaultRemoteCacheSameRegionTimeout
cfg.FlashHotKeyMissCount = defaultFlashHotKeyMissCount
cfg.FlashReadFlowLimit = defaultFlashReadFlowLimit
cfg.FlashWriteFlowLimit = defaultFlashWriteFlowLimit
return
}

4
remotecache/flashgroupmanager/flash_node.go
View File

@ -117,7 +117,7 @@ func (flashNode *FlashNode) checkLiveliness() {
flashNode.Unlock()
}
func (flashNode *FlashNode) createHeartbeatTask(masterAddr string, flashNodeHandleReadTimeout int, flashNodeReadDataNodeTimeout int, flashHotKeyMissCount int) (task *proto.AdminTask) {
func (flashNode *FlashNode) createHeartbeatTask(masterAddr string, flashNodeHandleReadTimeout int, flashNodeReadDataNodeTimeout int, flashHotKeyMissCount int, flashReadFlowLimit int64, flashWriteFlowLimit int64) (task *proto.AdminTask) {
request := &proto.HeartBeatRequest{
CurrTime: time.Now().Unix(),
MasterAddr: masterAddr,
@ -125,6 +125,8 @@ func (flashNode *FlashNode) createHeartbeatTask(masterAddr string, flashNodeHand
request.FlashNodeHandleReadTimeout = flashNodeHandleReadTimeout
request.FlashNodeReadDataNodeTimeout = flashNodeReadDataNodeTimeout
request.FlashHotKeyMissCount = flashHotKeyMissCount
request.FlashReadFlowLimit = flashReadFlowLimit
request.FlashWriteFlowLimit = flashWriteFlowLimit
task = proto.NewAdminTask(proto.OpFlashNodeHeartbeat, flashNode.Addr, request)
return

11
remotecache/flashgroupmanager/metadata_fsm_op.go
View File

@ -36,6 +36,8 @@ type clusterValue struct {
RemoteCacheSameZoneTimeout int64
RemoteCacheSameRegionTimeout int64
FlashHotKeyMissCount int
FlashReadFlowLimit int64
FlashWriteFlowLimit int64
}
func newClusterValue(c *Cluster) (cv *clusterValue) {
@ -50,6 +52,8 @@ func newClusterValue(c *Cluster) (cv *clusterValue) {
RemoteCacheSameZoneTimeout: c.cfg.RemoteCacheSameZoneTimeout,
RemoteCacheSameRegionTimeout: c.cfg.RemoteCacheSameRegionTimeout,
FlashHotKeyMissCount: c.cfg.FlashHotKeyMissCount,
FlashReadFlowLimit: c.cfg.FlashReadFlowLimit,
FlashWriteFlowLimit: c.cfg.FlashWriteFlowLimit,
}
return cv
}
@ -87,9 +91,12 @@ func (c *Cluster) loadClusterValue() (err error) {
}
c.cfg.FlashHotKeyMissCount = cv.FlashHotKeyMissCount
c.cfg.FlashReadFlowLimit = cv.FlashReadFlowLimit
c.cfg.FlashWriteFlowLimit = cv.FlashWriteFlowLimit
c.cfg.FlashNodeReadDataNodeTimeout = cv.FlashNodeReadDataNodeTimeout
log.LogInfof("action[loadClusterValue] flashNodeHandleReadTimeout %v(ms), flashNodeReadDataNodeTimeout%v(ms), flashHotKeyMissCount(%v)",
cv.FlashNodeHandleReadTimeout, cv.FlashNodeReadDataNodeTimeout, cv.FlashHotKeyMissCount)
log.LogInfof("action[loadClusterValue] flashNodeHandleReadTimeout %v(ms), flashNodeReadDataNodeTimeout%v(ms), flashHotKeyMissCount(%v), flashReadFlowLimit(%v), flashWriteFlowLimit(%v)",
cv.FlashNodeHandleReadTimeout, cv.FlashNodeReadDataNodeTimeout, cv.FlashHotKeyMissCount, cv.FlashReadFlowLimit, cv.FlashWriteFlowLimit)
if cv.RemoteCacheTTL == 0 {
cv.RemoteCacheTTL = proto.DefaultRemoteCacheTTL

2
remotecache/flashnode/flashnode.go
View File

@ -154,9 +154,11 @@ type FlashNode struct {
handleReadTimeout int
diskWriteIocc int
diskWriteFlow int
localChangeWriteFlow bool
diskWriteIoFactorFlow int
diskReadIocc int
diskReadFlow int
localChangeReadFlow bool
diskReadIoFactorFlow int
limitWrite *util.IoLimiter

10
remotecache/flashnode/flashnode_op.go
View File

@ -124,6 +124,14 @@ func (f *FlashNode) opFlashNodeHeartbeat(conn net.Conn, p *proto.Packet) (err er
if req.FlashHotKeyMissCount != 0 {
f.hotKeyMissCount = int32(req.FlashHotKeyMissCount)
}
if !f.localChangeWriteFlow && req.FlashWriteFlowLimit != int64(f.diskWriteFlow) {
f.diskWriteFlow = int(req.FlashWriteFlowLimit)
f.limitWrite.ResetFlow(f.diskWriteFlow)
}
if !f.localChangeReadFlow && req.FlashReadFlowLimit != int64(f.diskReadFlow) {
f.diskReadFlow = int(req.FlashReadFlowLimit)
f.limitRead.ResetFlow(f.diskReadFlow)
}
} else {
log.LogWarnf("decode HeartBeatRequest error: %s", err.Error())
resp.Status = proto.TaskFailed
@ -966,6 +974,7 @@ func (f *FlashNode) opSetReadIOLimits(conn net.Conn, p *proto.Packet) (err error
update = true
}
if req.Flow != -1 {
f.localChangeReadFlow = true
f.diskReadFlow = req.Flow
update = true
}
@ -1000,6 +1009,7 @@ func (f *FlashNode) opSetWriteIOLimits(conn net.Conn, p *proto.Packet) (err erro
update = true
}
if req.Flow != -1 {
f.localChangeWriteFlow = true
f.diskWriteFlow = req.Flow
update = true
}

6
remotecache/flashnode/serve_http.go
View File

@ -193,6 +193,9 @@ func (f *FlashNode) handleSetWriteDiskQos(w http.ResponseWriter, r *http.Request
return
}
if has {
if paramFlow == key {
f.localChangeWriteFlow = true
}
updated = true
*pVal = val
}
@ -236,6 +239,9 @@ func (f *FlashNode) handleSetReadDiskQos(w http.ResponseWriter, r *http.Request)
return
}
if has {
if paramFlow == key {
f.localChangeReadFlow = true
}
updated = true
*pVal = val
}

7
sdk/master/api_admin.go
View File

@ -608,6 +608,7 @@ func (api *AdminAPI) SetClusterParas(batchCount, markDeleteRate, deleteWorkerSle
remoteCacheTTL string, remoteCacheReadTimeout string,
remoteCacheMultiRead string, flashNodeTimeoutCount string,
remoteCacheSameZoneTimeout string, remoteCacheSameRegionTimeout string, flashHotKeyMissCount string,
flashReadFlowLimit string, flashWriteFlowLimit string,
) (err error) {
request := newRequest(get, proto.AdminSetNodeInfo).Header(api.h)
request.addParam("batchCount", batchCount)
@ -690,6 +691,12 @@ func (api *AdminAPI) SetClusterParas(batchCount, markDeleteRate, deleteWorkerSle
if remoteCacheSameRegionTimeout != "" {
request.addParamAny("remoteCacheSameRegionTimeout", remoteCacheSameRegionTimeout)
}
if flashReadFlowLimit != "" {
request.addParamAny("flashReadFlowLimit", flashReadFlowLimit)
}
if flashWriteFlowLimit != "" {
request.addParamAny("flashWriteFlowLimit", flashWriteFlowLimit)
}
_, err = api.mc.serveRequest(request)
return
}

29
tool/remotecache-config/cmd/cluster.go
View File

@ -61,6 +61,8 @@ func newClusterSetParasCmd(client *master.MasterClient) *cobra.Command {
optRemoteCacheSameZoneTimeout := ""
optRemoteCacheSameRegionTimeout := ""
optFlashHotKeyMissCount := ""
optFlashReadFlowLimit := ""
optFlashWriteFlowLimit := ""
cmd := &cobra.Command{
Use: CliOpSetCluster,
Short: cmdClusterSetClusterInfoShort,
@ -158,6 +160,28 @@ func newClusterSetParasCmd(client *master.MasterClient) *cobra.Command {
}
}
if optFlashReadFlowLimit != "" {
if tmp, err = strconv.ParseInt(optFlashReadFlowLimit, 10, 64); err != nil {
err = fmt.Errorf("param (%v) failed, should be int", optFlashReadFlowLimit)
return
}
if tmp < 0 {
err = fmt.Errorf("param flashReadFlowLimit(%v) must greater than or equal to 0", optFlashReadFlowLimit)
return
}
}
if optFlashWriteFlowLimit != "" {
if tmp, err = strconv.ParseInt(optFlashWriteFlowLimit, 10, 64); err != nil {
err = fmt.Errorf("param (%v) failed, should be int", optFlashWriteFlowLimit)
return
}
if tmp < 0 {
err = fmt.Errorf("param flashWriteFlowLimit(%v) must greater than or equal to 0", optFlashWriteFlowLimit)
return
}
}
if err = client.AdminAPI().SetClusterParas("", "", "",
"", "", "", "", clientIDKey,
"", "",
@ -165,7 +189,8 @@ func newClusterSetParasCmd(client *master.MasterClient) *cobra.Command {
"", "", "", "", "", "",
"", "", handleTimeout, readDataNodeTimeout,
optRcTTL, optRcReadTimeout, optRemoteCacheMultiRead, optFlashNodeTimeoutCount,
optRemoteCacheSameZoneTimeout, optRemoteCacheSameRegionTimeout, optFlashHotKeyMissCount); err != nil {
optRemoteCacheSameZoneTimeout, optRemoteCacheSameRegionTimeout, optFlashHotKeyMissCount,
optFlashReadFlowLimit, optFlashWriteFlowLimit); err != nil {
return
}
stdout("Cluster parameters has been set successfully. \n")
@ -181,5 +206,7 @@ func newClusterSetParasCmd(client *master.MasterClient) *cobra.Command {
cmd.Flags().StringVar(&optRemoteCacheSameZoneTimeout, CliFlagRemoteCacheSameZoneTimeout, "", "Remote cache same zone timeout microsecond(must > 0)")
cmd.Flags().StringVar(&optRemoteCacheSameRegionTimeout, CliFlagRemoteCacheSameRegionTimeout, "", "Remote cache same region timeout millisecond(must > 0)")
cmd.Flags().StringVar(&optFlashHotKeyMissCount, CliFlagFlashHotKeyMissCount, "", "Flash hot key miss count(must > 0)")
cmd.Flags().StringVar(&optFlashReadFlowLimit, CliFlagFlashReadFlowLimit, "", "Flash read flow limit(must >= 0)")
cmd.Flags().StringVar(&optFlashWriteFlowLimit, CliFlagFlashWriteFlowLimit, "", "Flash write flow limit(must >= 0)")
return cmd
}

2
tool/remotecache-config/cmd/const.go
View File

@ -161,6 +161,8 @@ const (
CliFlagRemoteCacheSameZoneTimeout = "remoteCacheSameZoneTimeout"
CliFlagRemoteCacheSameRegionTimeout = "remoteCacheSameRegionTimeout"
CliFlagFlashHotKeyMissCount = "flashHotKeyMissCount"
CliFlagFlashReadFlowLimit = "flashReadFlowLimit"
CliFlagFlashWriteFlowLimit = "flashWriteFlowLimit"
// CliFlagSetDataPartitionCount = "count" use dp-count instead

2
tool/remotecache-config/cmd/fmt.go
View File

@ -31,6 +31,8 @@ func formatClusterView(cv *proto.ClusterView) string {
sb.WriteString(fmt.Sprintf(" RemoteCacheSameZoneTimeout : %v microsecond\n", cv.RemoteCacheSameZoneTimeout))
sb.WriteString(fmt.Sprintf(" RemoteCacheSameRegionTimeout : %v millisecond\n", cv.RemoteCacheSameRegionTimeout))
sb.WriteString(fmt.Sprintf(" FlashHotKeyMissCount : %v\n", cv.FlashHotKeyMissCount))
sb.WriteString(fmt.Sprintf(" FlashReadFlowLimit : %v\n", cv.FlashReadFlowLimit))
sb.WriteString(fmt.Sprintf(" FlashWriteFlowLimit : %v\n", cv.FlashWriteFlowLimit))
return sb.String()
}