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(flashnode): add metric info for put block #1000150848

Browse Source 
Signed-off-by: clinx <chenlin1@oppo.com>
This commit is contained in:
clinx 2025-06-20 15:52:32 +08:00 committed by chihe
parent c7aac8a558
commit bcb1482c32
4 changed files with 22 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
remotecache/flashnode/cachengine/block.go
View File

@ -633,8 +633,8 @@ func (cb *CacheBlock) InitForCacheRead(sources []*proto.DataSource, readDataNode
return e
}
offset += size
updateWriteBytesMetric(uint64(size), cb.GetRootPath())
updateWriteCountMetric(cb.GetRootPath())
UpdateWriteBytesMetric(uint64(size), cb.GetRootPath())
UpdateWriteCountMetric(cb.GetRootPath())
return nil
}
logPrefix := func() string {

4
remotecache/flashnode/cachengine/stat.go
View File

@ -11,13 +11,13 @@ type MetricStat struct {
WriteCount uint64
}
func updateWriteBytesMetric(size uint64, d string) {
func UpdateWriteBytesMetric(size uint64, d string) {
if stat, ok := StatMap[d]; ok {
atomic.AddUint64(&stat.WriteBytes, size)
}
}
func updateWriteCountMetric(d string) {
func UpdateWriteCountMetric(d string) {
if stat, ok := StatMap[d]; ok {
atomic.AddUint64(&stat.WriteCount, 1)
}

10
remotecache/flashnode/flashnode_op.go
View File

@ -282,6 +282,10 @@ func (f *FlashNode) opCacheDelete(conn net.Conn, p *proto.Packet) (err error) {
}
func (f *FlashNode) opCachePutBlock(conn net.Conn, p *proto.Packet) (err error) {
bgTime := stat.BeginStat()
defer func() {
stat.EndStat("FlashNode:opCachePutBlock", err, bgTime, 1)
}()
defer func() {
if err != nil {
log.LogWarnf("action[opCachePutBlock] end, logMsg:%s",
@ -330,6 +334,10 @@ func (f *FlashNode) opCachePutBlock(conn net.Conn, p *proto.Packet) (err error)
if e := p.WriteToConn(conn); e != nil {
log.LogErrorf("action[opCachePutBlock] write to conn %v", e)
}
bgTime1 := stat.BeginStat()
defer func() {
stat.EndStat("CachePutBlock:Write", err1, bgTime1, 1)
}()
var (
totalWritten int64
n int
@ -363,6 +371,8 @@ func (f *FlashNode) opCachePutBlock(conn net.Conn, p *proto.Packet) (err error)
Crc: crcBuf[:cachengine.CRCLen],
DataSize: int64(readSize),
})
cachengine.UpdateWriteBytesMetric(proto.PageSize, cb.GetRootPath())
cachengine.UpdateWriteCountMetric(cb.GetRootPath())
err1 = cb.MaybeWriteCompleted(req.BlockLen)
if err1 != nil {
return

9
sdk/remotecache/client.go
View File

@ -434,11 +434,18 @@ func (rc *RemoteCacheClient) Put(ctx context.Context, reqId, key string, r io.Re
}
var conn *net.TCPConn
if conn, err = rc.conns.GetConnect(addr); err != nil {
log.LogWarnf("FlashGroup put: reqId(%v) get connection to curr addr failed, addr(%v) key(%v) err(%v)", reqId, addr, key, err)
moved := fg.moveToUnknownRank(addr, err, rc.FlashNodeTimeoutCount)
log.LogWarnf("FlashGroup put: reqId(%v) get connection to curr addr failed, addr(%v) key(%v) moved(%v) err(%v)", reqId, addr, key, moved, err)
return
}
bgTime := stat.BeginStat()
defer func() {
rc.conns.PutConnect(conn, err != nil)
parts := strings.Split(addr, ":")
if len(parts) > 0 && addr != "" {
stat.EndStat(fmt.Sprintf("flashPutBlock:%v", parts[0]), err, bgTime, 1)
}
stat.EndStat("flashPutBlock", err, bgTime, 1)
}()
req := &proto.PutBlockHead{
UniKey: key,