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

feat(master): add setRead/WriteLimits for flashnode

Browse Source 
close:#1000004942

Signed-off-by: chihe <chihe@oppo.com>
This commit is contained in:
chihe 2025-03-20 15:36:41 +08:00 committed by zhumingze1108
parent 487641d0f9
commit ca90d3bac8
11 changed files with 288 additions and 63 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
cmd/common/args.go
View File

@ -31,6 +31,9 @@ func (i *Int) Key(key string) *Argument { return NewArgument(key, &i.V) }
func (i *Int) ID() *Argument { return i.Key("id") }
func (i *Int) ExtentID() *Argument { return i.Key("extentID") }
func (i *Int) Count() *Argument { return i.Key("count") }
func (i *Int) Flow() *Argument { return i.Key("flow") }
func (i *Int) Iocc() *Argument { return i.Key("iocc") }
func (i *Int) Factor() *Argument { return i.Key("factor") }
func (u *Uint) Key(key string) *Argument { return NewArgument(key, &u.V) }
func (u *Uint) ID() *Argument { return u.Key("id") }

4
datanode/server_handler.go
View File

@ -360,8 +360,8 @@ func (s *DataNode) getDiskQos(w http.ResponseWriter, r *http.Request) {
Write util.LimiterStatus `json:"write"`
}{
Path: diskItem.Path,
Read: diskItem.limitRead.Status(),
Write: diskItem.limitWrite.Status(),
Read: diskItem.limitRead.Status(false),
Write: diskItem.limitWrite.Status(false),
}
disks = append(disks, disk)
}

76
flashnode/flashnode_op.go
View File

@ -51,6 +51,10 @@ func (f *FlashNode) handlePacket(conn net.Conn, p *proto.Packet) (err error) {
err = f.opCachePrepare(conn, p)
case proto.OpFlashNodeCacheRead:
err = f.opCacheRead(conn, p)
case proto.OpFlashNodeSetReadIOLimits:
err = f.opSetReadIOLimits(conn, p)
case proto.OpFlashNodeSetWriteIOLimits:
err = f.opSetWriteIOLimits(conn, p)
default:
err = fmt.Errorf("unknown Opcode:%d", p.Opcode)
}
@ -101,7 +105,9 @@ func (f *FlashNode) opFlashNodeHeartbeat(conn net.Conn, p *proto.Packet) (err er
}
resp.Stat = append(resp.Stat, stat)
}
writeStatus := proto.FlashNodeLimiterStatus{Status: f.limitWrite.Status(true), DiskNum: len(f.disks), ReadTimeoutSec: f.handleReadTimeout}
readStatus := proto.FlashNodeLimiterStatus{Status: f.limitRead.Status(true), DiskNum: len(f.disks), ReadTimeoutSec: f.handleReadTimeout}
resp.LimiterStatus = &proto.FlashNodeLimiterStatusInfo{WriteStatus: writeStatus, ReadStatus: readStatus}
reply, err := json.Marshal(resp)
if err != nil {
p.PacketErrorWithBody(proto.OpErr, []byte(err.Error()))
@ -376,3 +382,71 @@ func (f *FlashNode) sendPrepareRequest(addr string, req *proto.CachePrepareReque
}
return nil
}
func (f *FlashNode) opSetReadIOLimits(conn net.Conn, p *proto.Packet) (err error) {
data := p.Data
req := &proto.FlashNodeSetIOLimitsRequest{}
adminTask := &proto.AdminTask{
Request: req,
}
update := false
decode := json.NewDecoder(bytes.NewBuffer(data))
decode.UseNumber()
if err = decode.Decode(adminTask); err == nil {
log.LogDebugf("opSetReadIOLimits req: %v", req)
if req.Iocc != -1 {
f.diskReadIocc = req.Iocc
update = true
}
if req.Flow != -1 {
f.diskReadFlow = req.Flow
update = true
}
if req.Factor != -1 {
f.diskReadIoFactorFlow = req.Factor
update = true
}
if update {
f.limitRead.ResetIOEx(f.diskReadIocc*len(f.disks), f.diskReadIoFactorFlow, f.handleReadTimeout)
f.limitRead.ResetFlow(f.diskReadFlow)
}
} else {
log.LogErrorf("decode FlashNodeSetIOLimitsRequest error: %s", err.Error())
}
p.PacketOkReply()
return
}
func (f *FlashNode) opSetWriteIOLimits(conn net.Conn, p *proto.Packet) (err error) {
data := p.Data
req := &proto.FlashNodeSetIOLimitsRequest{}
adminTask := &proto.AdminTask{
Request: req,
}
update := false
decode := json.NewDecoder(bytes.NewBuffer(data))
decode.UseNumber()
if err = decode.Decode(adminTask); err == nil {
log.LogDebugf("opSetReadIOLimits req: %v", req)
if req.Iocc != -1 {
f.diskWriteIocc = req.Iocc
update = true
}
if req.Flow != -1 {
f.diskWriteFlow = req.Flow
update = true
}
if req.Factor != -1 {
f.diskWriteIoFactorFlow = req.Factor
update = true
}
if update {
f.limitWrite.ResetIOEx(f.diskWriteIocc*len(f.disks), f.diskWriteIoFactorFlow, f.handleReadTimeout)
f.limitWrite.ResetFlow(f.diskWriteFlow)
}
} else {
log.LogErrorf("decode opSetWriteIOLimits error: %s", err.Error())
}
p.PacketOkReply()
return
}

20
flashnode/serve_http.go
View File

@ -21,7 +21,6 @@ import (
"strconv"
"github.com/cubefs/cubefs/proto"
"github.com/cubefs/cubefs/util"
"github.com/cubefs/cubefs/util/log"
)
@ -172,26 +171,15 @@ func (f *FlashNode) handleSetReadDiskQos(w http.ResponseWriter, r *http.Request)
f.diskReadIoFactorFlow = _defaultDiskReadFactor
}
if updated {
f.limitRead.ResetIO(f.diskReadIocc*len(f.disks), f.diskReadIoFactorFlow)
f.limitRead.ResetIOEx(f.diskReadIocc*len(f.disks), f.diskReadIoFactorFlow, f.handleReadTimeout)
f.limitRead.ResetFlow(f.diskReadFlow)
}
replyOK(w, r, nil)
}
func (f *FlashNode) handleGetDiskQos(w http.ResponseWriter, r *http.Request) {
writeStatus := LimiterStatus{Status: f.limitWrite.Status(), DiskNum: len(f.disks), ReadTimeoutSec: f.handleReadTimeout}
readStatus := LimiterStatus{Status: f.limitRead.Status(), DiskNum: len(f.disks), ReadTimeoutSec: f.handleReadTimeout}
info := LimiterStatusInfo{WriteStatus: writeStatus, ReadStatus: readStatus}
writeStatus := proto.FlashNodeLimiterStatus{Status: f.limitWrite.Status(true), DiskNum: len(f.disks), ReadTimeoutSec: f.handleReadTimeout}
readStatus := proto.FlashNodeLimiterStatus{Status: f.limitRead.Status(true), DiskNum: len(f.disks), ReadTimeoutSec: f.handleReadTimeout}
info := proto.FlashNodeLimiterStatusInfo{WriteStatus: writeStatus, ReadStatus: readStatus}
replyOK(w, r, info)
}
type LimiterStatus struct {
Status util.LimiterStatus
DiskNum int
ReadTimeoutSec int
}
type LimiterStatusInfo struct {
WriteStatus LimiterStatus
ReadStatus LimiterStatus
}

163
master/flash_node.go
View File

@ -53,9 +53,10 @@ type FlashNode struct {
sync.RWMutex
flashNodeValue
DiskStat []*proto.FlashNodeDiskCacheStat
ReportTime time.Time
IsActive bool
DiskStat []*proto.FlashNodeDiskCacheStat
ReportTime time.Time
IsActive bool
LimiterStatus *proto.FlashNodeLimiterStatusInfo
}
func newFlashNode(addr, zoneName, clusterID, version string, isEnable bool) *FlashNode {
@ -99,24 +100,26 @@ func (flashNode *FlashNode) isActiveAndEnable() (ok bool) {
func (flashNode *FlashNode) getFlashNodeViewInfo() (info *proto.FlashNodeViewInfo) {
flashNode.RLock()
info = &proto.FlashNodeViewInfo{
ID: flashNode.ID,
Addr: flashNode.Addr,
ReportTime: flashNode.ReportTime,
IsActive: flashNode.IsActive,
Version: flashNode.Version,
ZoneName: flashNode.ZoneName,
FlashGroupID: flashNode.FlashGroupID,
IsEnable: flashNode.IsEnable,
DiskStat: flashNode.DiskStat,
ID: flashNode.ID,
Addr: flashNode.Addr,
ReportTime: flashNode.ReportTime,
IsActive: flashNode.IsActive,
Version: flashNode.Version,
ZoneName: flashNode.ZoneName,
FlashGroupID: flashNode.FlashGroupID,
IsEnable: flashNode.IsEnable,
DiskStat: flashNode.DiskStat,
LimiterStatus: flashNode.LimiterStatus,
}
flashNode.RUnlock()
return
}
func (flashNode *FlashNode) updateFlashNodeStatHeartbeat(stat []*proto.FlashNodeDiskCacheStat) {
log.LogInfof("updateFlashNodeStatHeartbeat, flashNode:%v, diskStat[%v], time:%v", flashNode.Addr, stat, time.Now().Format("2006-01-02 15:04:05"))
func (flashNode *FlashNode) updateFlashNodeStatHeartbeat(resp *proto.FlashNodeHeartbeatResponse) {
log.LogInfof("updateFlashNodeStatHeartbeat, flashNode:%v, resp[%v], time:%v", flashNode.Addr, resp, time.Now().Format("2006-01-02 15:04:05"))
flashNode.Lock()
flashNode.DiskStat = stat
flashNode.DiskStat = resp.Stat
flashNode.LimiterStatus = resp.LimiterStatus
flashNode.Unlock()
}
@ -144,7 +147,24 @@ func (c *Cluster) syncFlashNodeHeartbeatTasks(tasks []*proto.AdminTask) {
log.LogErrorf("Failed to unmarshal response: %v", err)
continue
}
node.updateFlashNodeStatHeartbeat(resp.Stat)
node.updateFlashNodeStatHeartbeat(resp)
}
}
func (c *Cluster) syncFlashNodeSetIOLimitTasks(tasks []*proto.AdminTask) {
for _, t := range tasks {
if t == nil {
continue
}
node, err := c.peekFlashNode(t.OperatorAddr)
if err != nil {
log.LogWarn(fmt.Sprintf("action[syncFlashNodeHeartbeatTasks],nodeAddr:%v,taskID:%v,err:%v", t.OperatorAddr, t.ID, err.Error()))
continue
}
if _, err = node.TaskManager.syncSendAdminTask(t); err != nil {
log.LogWarn(fmt.Sprintf("action[syncFlashNodeHeartbeatTasks],nodeAddr:%v,taskID:%v,err:%v", t.OperatorAddr, t.ID, err.Error()))
continue
}
}
}
@ -180,6 +200,17 @@ func (flashNode *FlashNode) createHeartbeatTask(masterAddr string, flashNodeHand
return
}
func (flashNode *FlashNode) createSetIOLimitsTask(flow, iocc, factor int, opCode uint8) (task *proto.AdminTask) {
request := &proto.FlashNodeSetIOLimitsRequest{
Flow: flow,
Iocc: iocc,
Factor: factor,
}
task = proto.NewAdminTask(opCode, flashNode.Addr, request)
return
}
func (m *Server) addFlashNode(w http.ResponseWriter, r *http.Request) {
var (
nodeAddr common.String
@ -442,3 +473,103 @@ func argParserNodeAddr(nodeAddr *common.String) *common.Argument {
return unmatchedKey(new(common.String).Addr().Key())
})
}
func (m *Server) setFlashNodeReadIOLimits(w http.ResponseWriter, r *http.Request) {
var (
flow common.Int
iocc common.Int
factor common.Int
readFlow int64
readIocc int64
readFactor int64
err error
)
if err = parseArgs(r, flow.Flow().OmitEmpty().OnEmpty(func() error {
readFlow = -1
return nil
}).OnValue(func() error {
readFlow = flow.V
return nil
}),
iocc.Iocc().OmitEmpty().OnEmpty(func() error {
readIocc = -1
return nil
}).OnValue(func() error {
readIocc = iocc.V
return nil
}),
factor.Factor().OmitEmpty().OnEmpty(func() error {
readFactor = -1
return nil
}).OnValue(func() error {
readFactor = factor.V
return nil
})); err != nil {
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: err.Error()})
return
}
log.LogDebugf("action[setFlashNodeReadIOLimits],flow[%v] iocc[%v] factor [%v]",
readFlow, readIocc, readFactor)
tasks := make([]*proto.AdminTask, 0)
m.cluster.flashNodeTopo.flashNodeMap.Range(func(key, value interface{}) bool {
flashNode := value.(*FlashNode)
if flashNode.isActiveAndEnable() {
task := flashNode.createSetIOLimitsTask(int(readFlow), int(readIocc), int(readFactor), proto.OpFlashNodeSetReadIOLimits)
tasks = append(tasks, task)
}
return true
})
go m.cluster.syncFlashNodeSetIOLimitTasks(tasks)
sendOkReply(w, r, newSuccessHTTPReply("set ReadIOLimits for FlashNode is submit,check it later."))
}
func (m *Server) setFlashNodeWriteIOLimits(w http.ResponseWriter, r *http.Request) {
var (
flow common.Int
iocc common.Int
factor common.Int
writeFlow int64
writeIocc int64
writeFactor int64
err error
)
if err = parseArgs(r, flow.Flow().OmitEmpty().OnEmpty(func() error {
writeFlow = -1
return nil
}).OnValue(func() error {
writeFlow = flow.V
return nil
}),
iocc.Iocc().OmitEmpty().OnEmpty(func() error {
writeIocc = -1
return nil
}).OnValue(func() error {
writeIocc = iocc.V
return nil
}),
factor.Factor().OmitEmpty().OnEmpty(func() error {
writeFactor = -1
return nil
}).OnValue(func() error {
writeFactor = factor.V
return nil
})); err != nil {
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: err.Error()})
return
}
log.LogDebugf("action[setFlashNodeWriteIOLimits],flow[%v] iocc[%v] factor [%v]",
writeFlow, writeIocc, writeFactor)
tasks := make([]*proto.AdminTask, 0)
m.cluster.flashNodeTopo.flashNodeMap.Range(func(key, value interface{}) bool {
flashNode := value.(*FlashNode)
if flashNode.isActiveAndEnable() {
task := flashNode.createSetIOLimitsTask(int(writeFlow), int(writeIocc), int(writeFactor), proto.OpFlashNodeSetWriteIOLimits)
tasks = append(tasks, task)
}
return true
})
go m.cluster.syncFlashNodeSetIOLimitTasks(tasks)
sendOkReply(w, r, newSuccessHTTPReply("set WriteIOLimits for FlashNode is submit,check it later."))
}

2
master/http_server.go
View File

@ -899,6 +899,8 @@ func (m *Server) registerAPIRoutes(router *mux.Router) {
router.NewRoute().Methods(http.MethodGet, http.MethodPost).Path(proto.FlashNodeRemove).HandlerFunc(m.removeFlashNode)
router.NewRoute().Methods(http.MethodGet).Path(proto.FlashNodeGet).HandlerFunc(m.getFlashNode)
router.NewRoute().Methods(http.MethodGet).Path(proto.FlashNodeList).HandlerFunc(m.listFlashNodes)
router.NewRoute().Methods(http.MethodGet, http.MethodPost).Path(proto.FlashNodeSetReadIOLimits).HandlerFunc(m.setFlashNodeReadIOLimits)
router.NewRoute().Methods(http.MethodGet, http.MethodPost).Path(proto.FlashNodeSetWriteIOLimits).HandlerFunc(m.setFlashNodeWriteIOLimits)
// APIs for FlashGroup
router.NewRoute().Methods(http.MethodGet, http.MethodPost).Path(proto.AdminFlashGroupTurn).HandlerFunc(m.turnFlashGroup)

40
proto/admin_proto.go
View File

@ -288,11 +288,13 @@ const (
AdminVolAddAllowedStorageClass = "/vol/addAllowedStorageClass"
// FlashNode API
FlashNodeAdd = "/flashNode/add"
FlashNodeSet = "/flashNode/set"
FlashNodeRemove = "/flashNode/remove"
FlashNodeGet = "/flashNode/get"
FlashNodeList = "/flashNode/list"
FlashNodeAdd = "/flashNode/add"
FlashNodeSet = "/flashNode/set"
FlashNodeRemove = "/flashNode/remove"
FlashNodeGet = "/flashNode/get"
FlashNodeList = "/flashNode/list"
FlashNodeSetReadIOLimits = "/flashNode/setReadIOLimits"
FlashNodeSetWriteIOLimits = "/flashNode/SetWriteIOLimits"
// FlashGroup API
AdminFlashGroupTurn = "/flashGroup/turn"
@ -963,11 +965,23 @@ type FlashNodeDiskCacheStat struct {
// FlashNodeHeartbeatResponse defines the response to the flash node heartbeat.
type FlashNodeHeartbeatResponse struct {
Status uint8
Result string
Version string
ZoneName string
Stat []*FlashNodeDiskCacheStat
Status uint8
Result string
Version string
ZoneName string
Stat []*FlashNodeDiskCacheStat
LimiterStatus *FlashNodeLimiterStatusInfo
}
type FlashNodeLimiterStatus struct {
Status util.LimiterStatus
DiskNum int
ReadTimeoutSec int
}
type FlashNodeLimiterStatusInfo struct {
WriteStatus FlashNodeLimiterStatus
ReadStatus FlashNodeLimiterStatus
}
// DeleteFileRequest defines the request to delete a file.
@ -1663,3 +1677,9 @@ type IsRaftStatusOKRequest struct {
Ready bool
ReplicaNum int
}
type FlashNodeSetIOLimitsRequest struct {
Iocc int
Flow int
Factor int
}

19
proto/distributed_cache.go
View File

@ -230,15 +230,16 @@ type FlashGroupAdminView struct {
}
type FlashNodeViewInfo struct {
ID uint64
Addr string
ReportTime time.Time
IsActive bool
Version string
ZoneName string
FlashGroupID uint64
IsEnable bool
DiskStat []*FlashNodeDiskCacheStat
ID uint64
Addr string
ReportTime time.Time
IsActive bool
Version string
ZoneName string
FlashGroupID uint64
IsEnable bool
DiskStat []*FlashNodeDiskCacheStat
LimiterStatus *FlashNodeLimiterStatusInfo
}
type FlashNodeStat struct {

12
proto/packet.go
View File

@ -287,9 +287,11 @@ const (
OpWriteOpOfProtoVerForbidden uint8 = 0x88
OpMetaForbiddenMigration uint8 = 0x89
// Distributed cache related OP codes.
OpFlashNodeHeartbeat uint8 = 0xDA
OpFlashNodeCachePrepare uint8 = 0xDB
OpFlashNodeCacheRead uint8 = 0xDC
OpFlashNodeHeartbeat uint8 = 0xDA
OpFlashNodeCachePrepare uint8 = 0xDB
OpFlashNodeCacheRead uint8 = 0xDC
OpFlashNodeSetReadIOLimits uint8 = 0xED
OpFlashNodeSetWriteIOLimits uint8 = 0xEE
)
const (
@ -713,6 +715,10 @@ func (p *Packet) GetOpMsg() (m string) {
m = "OpFlashNodeCachePrepare"
case OpFlashNodeCacheRead:
m = "OpFlashNodeCacheRead"
case OpFlashNodeSetReadIOLimits:
m = "OpFlashNodeSetReadIOLimits"
case OpFlashNodeSetWriteIOLimits:
m = "OpFlashNodeSetWriteIOLimits"
default:
m = fmt.Sprintf("op:%v not found", p.Opcode)
}

4
util/limit_io.go
View File

@ -141,12 +141,12 @@ func (l *IoLimiter) TryRunWithContext(ctx context.Context, size int, taskFn func
return true
}
func (l *IoLimiter) Status() (st LimiterStatus) {
func (l *IoLimiter) Status(ignoreUsed bool) (st LimiterStatus) {
st = l.getIO().Status()
limit := l.limit
st.FlowLimit = limit
if limit > 0 {
if limit > 0 && !ignoreUsed {
now := time.Now()
reserve := l.flow.ReserveN(now, l.flow.Burst())
duration := reserve.DelayFrom(now)

8
util/limit_io_test.go
View File

@ -42,7 +42,7 @@ func TestLimitIOBase(t *testing.T) {
{
l := NewIOLimiter(1<<10, 0)
l.Run(10, true, f)
st := l.Status()
st := l.Status(false)
t.Logf("status: %+v", st)
require.Equal(t, 1<<10, st.FlowLimit)
require.True(t, st.FlowUsed > 0)
@ -55,7 +55,7 @@ func TestLimitIOBase(t *testing.T) {
{
done := make(chan struct{})
l := NewIOLimiter(-1, 2)
st := l.Status()
st := l.Status(false)
t.Logf("before status: %+v", st)
for ii := 0; ii < st.IOConcurrency; ii++ {
go func() {
@ -68,7 +68,7 @@ func TestLimitIOBase(t *testing.T) {
}()
}
time.Sleep(100 * time.Millisecond)
t.Logf("after status: %+v", l.Status())
t.Logf("after status: %+v", l.Status(false))
require.False(t, l.TryRun(0, f))
close(done)
q := l.getIO()
@ -86,7 +86,7 @@ func TestLimitIOTimeout(t *testing.T) {
t.Logf("func running!")
}
l := NewIOLimiter(-1, 1)
st := l.Status()
st := l.Status(false)
t.Logf("before status: %+v", st)
q := l.getIO()
rs := q.Run(f, false)