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 metrics MetricCacheBytes,MetricHandleReadLatency,MetricSourceDataLatency

Browse Source 
. #1000092826

Signed-off-by: baihailong <baihailong@oppo.com>
This commit is contained in:
baihailong 2025-05-14 17:58:42 +08:00 committed by zhumingze1108
parent 12f68e47b0
commit 4581cbd979
4 changed files with 77 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
flashnode/cachengine/engine.go
View File

@ -914,6 +914,16 @@ func (c *CacheEngine) GetEvictCount() map[string]int {
return result
}
func (c *CacheEngine) GetCacheBytes() map[string]int64 {
result := make(map[string]int64)
c.lruCacheMap.Range(func(key, value interface{}) bool {
cacheItem := value.(*lruCacheItem)
result[cacheItem.config.Path] = cacheItem.lruCache.GetAllocated()
return true
})
return result
}
func (c *CacheEngine) DoInactiveDisk(dataPath string) {
if value, ok := c.lruCacheMap.Load(dataPath); ok {
cacheItem := value.(*lruCacheItem)

57
flashnode/metric.go
View File

@ -8,27 +8,34 @@ import (
"github.com/cubefs/cubefs/flashnode/cachengine"
"github.com/cubefs/cubefs/util/exporter"
"github.com/cubefs/cubefs/util/log"
"github.com/cubefs/cubefs/util/stat"
)
const (
StatPeriod = time.Minute * time.Duration(1)
MetricFlashNodeReadBytes = "flashNodeReadBytes"
MetricFlashNodeReadCount = "flashNodeReadCount"
MetricFlashNodeWriteBytes = "flashNodeWriteBytes"
MetricFlashNodeWriteCount = "flashNodeWriteCount"
MetricFlashNodeHitRate = "flashNodeHitRate"
MetricFlashNodeEvictCount = "flashNodeEvictCount"
StatPeriod = time.Minute * time.Duration(1)
MetricFlashNodeReadBytes = "flashNodeReadBytes"
MetricFlashNodeReadCount = "flashNodeReadCount"
MetricFlashNodeWriteBytes = "flashNodeWriteBytes"
MetricFlashNodeWriteCount = "flashNodeWriteCount"
MetricFlashNodeHitRate = "flashNodeHitRate"
MetricFlashNodeEvictCount = "flashNodeEvictCount"
MetricFlashNodeCacheBytes = "flashNodeCacheBytes"
MetricFlashNodeHandleReadLatency = "flashNodeHandleReadLatency"
MetricFlashNodeSourceDataLatency = "flashNodeSourceDataLatency"
)
type FlashNodeMetrics struct {
flashNode *FlashNode
stopC chan struct{}
MetricReadBytes *exporter.Gauge
MetricReadCount *exporter.Gauge
MetricWriteBytes *exporter.Gauge
MetricWriteCount *exporter.Gauge
MetricEvictCount *exporter.Gauge
MetricHitRate *exporter.Gauge
flashNode *FlashNode
stopC chan struct{}
MetricReadBytes *exporter.Gauge
MetricReadCount *exporter.Gauge
MetricWriteBytes *exporter.Gauge
MetricWriteCount *exporter.Gauge
MetricEvictCount *exporter.Gauge
MetricHitRate *exporter.Gauge
MetricCacheBytes *exporter.Gauge
MetricHandleReadLatency *exporter.Gauge
MetricSourceDataLatency *exporter.Gauge
}
func (f *FlashNode) registerMetrics(disks []*cachengine.Disk) {
@ -43,6 +50,9 @@ func (f *FlashNode) registerMetrics(disks []*cachengine.Disk) {
f.metrics.MetricWriteCount = exporter.NewGauge(MetricFlashNodeWriteCount)
f.metrics.MetricEvictCount = exporter.NewGauge(MetricFlashNodeEvictCount)
f.metrics.MetricHitRate = exporter.NewGauge(MetricFlashNodeHitRate)
f.metrics.MetricCacheBytes = exporter.NewGauge(MetricFlashNodeCacheBytes)
f.metrics.MetricHandleReadLatency = exporter.NewGauge(MetricFlashNodeHandleReadLatency)
f.metrics.MetricSourceDataLatency = exporter.NewGauge(MetricFlashNodeSourceDataLatency)
for _, d := range disks {
cachengine.StatMap[path.Join(d.Path, cachengine.DefaultCacheDirName)] = new(cachengine.MetricStat)
}
@ -78,6 +88,8 @@ func (fm *FlashNodeMetrics) doStat() {
fm.setWriteCountMetric()
fm.setEvictCountMetric()
fm.setHitRateMetric()
fm.setCacheBytesMetric()
fm.setLatencyMetric()
}
func (fm *FlashNodeMetrics) setReadBytesMetric() {
@ -122,6 +134,21 @@ func (fm *FlashNodeMetrics) setHitRateMetric() {
}
}
func (fm *FlashNodeMetrics) setCacheBytesMetric() {
cacheBytesMap := fm.flashNode.cacheEngine.GetCacheBytes()
for dataPath, bytes := range cacheBytesMap {
fm.MetricCacheBytes.SetWithLabels(float64(bytes), map[string]string{"cluster": fm.flashNode.clusterID, exporter.FlashNode: fm.flashNode.localAddr, exporter.Disk: dataPath})
}
}
func (fm *FlashNodeMetrics) setLatencyMetric() {
handleReadLatency := stat.GetAvgLatencyMs("FlashNode:opCacheRead")
sourceDataLatency := stat.GetAvgLatencyMs("MissCacheRead:ReadFromDN")
fm.MetricHandleReadLatency.SetWithLabels(float64(handleReadLatency), map[string]string{"cluster": fm.flashNode.clusterID, exporter.FlashNode: fm.flashNode.localAddr})
fm.MetricSourceDataLatency.SetWithLabels(float64(sourceDataLatency), map[string]string{"cluster": fm.flashNode.clusterID, exporter.FlashNode: fm.flashNode.localAddr})
}
func (fm *FlashNodeMetrics) updateReadBytesMetric(size uint64, d string) {
if stat, ok := cachengine.StatMap[d]; ok {
atomic.AddUint64(&stat.ReadBytes, size)

4
sdk/data/stream/stream_reader.go
View File

@ -291,7 +291,11 @@ func (s *Streamer) read(data []byte, offset int, size int, storageClass uint32)
cacheReadRequests, err = s.prepareCacheRequests(uint64(offset), uint64(size), data, inodeInfo.Generation)
if err == nil {
var read int
remoteCacheMetric := exporter.NewCounter("readRemoteCache")
remoteCacheMetric.AddWithLabels(1, map[string]string{exporter.Vol: s.client.volumeName})
if read, err = s.readFromRemoteCache(ctx, uint64(offset), uint64(size), cacheReadRequests); err == nil {
remoteCacheHitMetric := exporter.NewCounter("readRemoteCacheHit")
remoteCacheHitMetric.AddWithLabels(1, map[string]string{exporter.Vol: s.client.volumeName})
return read, err
}
}

21
util/stat/statistic.go
View File

@ -442,3 +442,24 @@ func GetProcessMemory(pid int) (Virt, Res uint64, err error) {
}
return
}
func GetAvgLatencyMs(typeName string) float32 {
if gSt == nil {
return 0
}
if gSt.useMutex {
gSt.Lock()
defer gSt.Unlock()
}
typeInfo := gSt.typeInfoMap[typeName]
if typeInfo == nil {
return 0
}
avgUs := int32(0)
if typeInfo.allCount > 0 {
avgUs = int32(typeInfo.allTimeUs / time.Duration(typeInfo.allCount))
}
return float32(avgUs) / 1000
}