diff --git a/cli/cmd/vol.go b/cli/cmd/vol.go index 3baa9eed0..5dd9ff22b 100644 --- a/cli/cmd/vol.go +++ b/cli/cmd/vol.go @@ -331,7 +331,7 @@ func newVolCreateCmd(client *master.MasterClient) *cobra.Command { cmd.Flags().Int64Var(&optRcTTL, CliFlagRemoteCacheTTL, cmdVolDefaultRemoteCacheTTL, "Remote cache ttl[Unit: s](must >= 10min, default 5day)") cmd.Flags().Int64Var(&optRcReadTimeout, CliFlagRemoteCacheReadTimeout, cmdVolDefaultRemoteCacheReadTimeout, "Remote cache read timeout millisecond(must > 0)") cmd.Flags().Int64Var(&optRemoteCacheMaxFileSizeGB, CliFlagRemoteCacheMaxFileSizeGB, cmdVolDefaultRemoteCacheMaxFileSizeGB, "Remote cache max file size[Unit: GB](must > 0)") - cmd.Flags().StringVar(&optRemoteCacheOnlyForNotSSD, CliFlagRemoteCacheOnlyForNotSSD, "false", "Remote cache only for not ssd(true|false)") + cmd.Flags().StringVar(&optRemoteCacheOnlyForNotSSD, CliFlagRemoteCacheOnlyForNotSSD, "true", "Remote cache only for not ssd(true|false)") cmd.Flags().StringVar(&optRemoteCacheMultiRead, CliFlagRemoteCacheMultiRead, "false", "Remote cache follower read(true|false)") cmd.Flags().Int64Var(&optFlashNodeTimeoutCount, CliFlagFlashNodeTimeoutCount, cmdVolDefaultFlashNodeTimeoutCount, "FlashNode timeout count, flashNode will be removed by client if it's timeout count exceeds this value") cmd.Flags().Int64Var(&optRemoteCacheSameZoneTimeout, CliFlagRemoteCacheSameZoneTimeout, proto.DefaultRemoteCacheSameZoneTimeout, "Remote cache same zone timeout microsecond(must > 0)") diff --git a/remotecache/flashnode/cachengine/engine.go b/remotecache/flashnode/cachengine/engine.go index 74842ed7f..a8233a84d 100644 --- a/remotecache/flashnode/cachengine/engine.go +++ b/remotecache/flashnode/cachengine/engine.go @@ -1046,6 +1046,44 @@ func (c *CacheEngine) GetCacheBytes() map[string]int64 { return result } +func (c *CacheEngine) GetLruUsageRatio() float64 { + var totalLen, totalCapacity int + c.lruCacheMap.Range(func(key, value interface{}) bool { + cacheItem := value.(*lruCacheItem) + totalLen += cacheItem.lruCache.Len() + totalCapacity += cacheItem.config.Capacity + return true + }) + if totalCapacity > 0 { + return float64(totalLen) / float64(totalCapacity) + } + return 0 +} + +func (c *CacheEngine) GetDiskUsageRatio() map[string]float64 { + result := make(map[string]float64) + c.lruCacheMap.Range(func(key, value interface{}) bool { + cacheItem := value.(*lruCacheItem) + if atomic.LoadInt32(&cacheItem.disk.Status) == proto.ReadWrite { + fs := syscall.Statfs_t{} + if err := syscall.Statfs(cacheItem.disk.Path, &fs); err != nil { + log.LogErrorf("get disk(%s) stat err:%v", cacheItem.disk.Path, err) + return true + } + totalSpace := int64(fs.Blocks) * int64(fs.Bsize) + usedSpace := int64(fs.Blocks-fs.Bfree) * int64(fs.Bsize) + if totalSpace > 0 { + usageRatio := float64(usedSpace) / float64(totalSpace) + result[cacheItem.config.Path] = usageRatio + } else { + result[cacheItem.config.Path] = 0 + } + } + return true + }) + return result +} + func (c *CacheEngine) DoInactiveDisk(dataPath string) { if value, ok := c.lruCacheMap.Load(dataPath); ok { cacheItem := value.(*lruCacheItem) diff --git a/remotecache/flashnode/flashnode.go b/remotecache/flashnode/flashnode.go index 5baf22691..eac94782f 100644 --- a/remotecache/flashnode/flashnode.go +++ b/remotecache/flashnode/flashnode.go @@ -18,6 +18,7 @@ import ( "fmt" "net" "os" + "path/filepath" "strconv" "strings" "sync" @@ -368,6 +369,10 @@ func (f *FlashNode) parseConfig(cfg *config.Config) (err error) { continue } } + if os.Getenv(cachengine.EnvDockerTmpfs) == "" && !hasMountsOnLastTwoLevels(path) { + log.LogErrorf("path[%v] is not a mount point, skip it", path) + continue + } totalSpace, err := strconv.ParseInt(arr[1], 10, 64) if err != nil { log.LogErrorf("invalid disk total space for path[%v]. Error: %s", path, err.Error()) @@ -739,3 +744,55 @@ func (f *FlashNode) cleanupStaleWarmupWorkers() { staleCount, len(f.currentWarmUpWorkers)) } } + +// hasMountsOnLastTwoLevels returns true if either the parent directory or the +// given path itself is a mount point. For example, for /home/service/var/data, +// it checks whether /home/service/var OR /home/service/var/data is a mount target. +func hasMountsOnLastTwoLevels(p string) bool { + abs := p + if !filepath.IsAbs(abs) { + var err error + if abs, err = filepath.Abs(p); err != nil { + return false + } + } + abs = filepath.Clean(abs) + parent := filepath.Dir(abs) + + data, err := os.ReadFile("/proc/mounts") + if err != nil { + return false + } + + mounts := make(map[string]struct{}) + for _, line := range strings.Split(string(data), "\n") { + if line == "" { + continue + } + fields := strings.Fields(line) + if len(fields) < 2 { + continue + } + mp := unescapeMountField(fields[1]) + mounts[mp] = struct{}{} + } + + _, okParent := mounts[parent] + _, okSelf := mounts[abs] + return okParent || okSelf +} + +func unescapeMountField(s string) string { + var b strings.Builder + for i := 0; i < len(s); i++ { + if s[i] == '\\' && i+3 < len(s) { + if o, err := strconv.ParseInt(s[i+1:i+4], 8, 0); err == nil { + b.WriteByte(byte(o)) + i += 3 + continue + } + } + b.WriteByte(s[i]) + } + return b.String() +} diff --git a/remotecache/flashnode/metric.go b/remotecache/flashnode/metric.go index cd7731883..f2120b9a4 100644 --- a/remotecache/flashnode/metric.go +++ b/remotecache/flashnode/metric.go @@ -12,30 +12,40 @@ import ( ) const ( - StatPeriod = time.Minute * time.Duration(1) - MetricFlashNodeReadBytes = "flashNodeReadBytes" - MetricFlashNodeReadCount = "flashNodeReadCount" - MetricFlashNodeWriteBytes = "flashNodeWriteBytes" - MetricFlashNodeWriteCount = "flashNodeWriteCount" - MetricFlashNodeHitRate = "flashNodeHitRate" - MetricFlashNodeEvictCount = "flashNodeEvictCount" - MetricFlashNodeCacheBytes = "flashNodeCacheBytes" - MetricFlashNodeHandleReadLatency = "flashNodeHandleReadLatency" - MetricFlashNodeSourceDataLatency = "flashNodeSourceDataLatency" + StatPeriod = time.Minute * time.Duration(1) + MetricFlashNodeReadBytes = "flashNodeReadBytes" + MetricFlashNodeReadCount = "flashNodeReadCount" + MetricFlashNodeWriteBytes = "flashNodeWriteBytes" + MetricFlashNodeWriteCount = "flashNodeWriteCount" + MetricFlashNodeHitRate = "flashNodeHitRate" + MetricFlashNodeEvictCount = "flashNodeEvictCount" + MetricFlashNodeCacheBytes = "flashNodeCacheBytes" + MetricFlashNodeHandleReadLatency = "flashNodeHandleReadLatency" + MetricFlashNodeSourceDataLatency = "flashNodeSourceDataLatency" + MetricFlashNodeHitCacheReadLatency = "flashNodeHitCacheReadLatency" + MetricFlashNodeFlowLimitedCount = "flashNodeFlowLimitedCount" + MetricFlashNodeRunLimitedCount = "flashNodeRunLimitedCount" + MetricFlashNodeLruUsageRatio = "flashNodeLruUsageRatio" + MetricFlashNodeDiskUsageRatio = "flashNodeDiskUsageRatio" ) 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 - MetricCacheBytes *exporter.Gauge - MetricHandleReadLatency *exporter.Gauge - MetricSourceDataLatency *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 + MetricHitCacheReadLatency *exporter.Gauge + MetricFlowLimitedCount *exporter.Gauge + MetricRunLimitedCount *exporter.Gauge + MetricLruUsageRatio *exporter.Gauge + MetricDiskUsageRatio *exporter.Gauge } func (f *FlashNode) registerMetrics(disks []*cachengine.Disk) { @@ -53,6 +63,11 @@ func (f *FlashNode) registerMetrics(disks []*cachengine.Disk) { f.metrics.MetricCacheBytes = exporter.NewGauge(MetricFlashNodeCacheBytes) f.metrics.MetricHandleReadLatency = exporter.NewGauge(MetricFlashNodeHandleReadLatency) f.metrics.MetricSourceDataLatency = exporter.NewGauge(MetricFlashNodeSourceDataLatency) + f.metrics.MetricHitCacheReadLatency = exporter.NewGauge(MetricFlashNodeHitCacheReadLatency) + f.metrics.MetricFlowLimitedCount = exporter.NewGauge(MetricFlashNodeFlowLimitedCount) + f.metrics.MetricRunLimitedCount = exporter.NewGauge(MetricFlashNodeRunLimitedCount) + f.metrics.MetricLruUsageRatio = exporter.NewGauge(MetricFlashNodeLruUsageRatio) + f.metrics.MetricDiskUsageRatio = exporter.NewGauge(MetricFlashNodeDiskUsageRatio) for _, d := range disks { cachengine.StatMap[path.Join(d.Path, cachengine.DefaultCacheDirName)] = new(cachengine.MetricStat) } @@ -90,6 +105,9 @@ func (fm *FlashNodeMetrics) doStat() { fm.setHitRateMetric() fm.setCacheBytesMetric() fm.setLatencyMetric() + fm.setLimitedCountMetric() + fm.setLruUsageRatioMetric() + fm.setDiskUsageRatioMetric() } func (fm *FlashNodeMetrics) setReadBytesMetric() { @@ -144,9 +162,31 @@ func (fm *FlashNodeMetrics) setCacheBytesMetric() { func (fm *FlashNodeMetrics) setLatencyMetric() { handleReadLatency := stat.GetAvgLatencyMs("FlashNode:opCacheRead") sourceDataLatency := stat.GetAvgLatencyMs("MissCacheRead:ReadFromDN") + hitCacheReadLatency := stat.GetAvgLatencyMs("HitCacheRead") 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}) + fm.MetricHitCacheReadLatency.SetWithLabels(float64(hitCacheReadLatency), map[string]string{"cluster": fm.flashNode.clusterID, exporter.FlashNode: fm.flashNode.localAddr}) +} + +func (fm *FlashNodeMetrics) setLimitedCountMetric() { + flowLimitedCount := stat.GetCount("FlashNode:opCacheRead[flow limited]") + runLimitedCount := stat.GetCount("FlashNode:opCacheRead[run limited]") + + fm.MetricFlowLimitedCount.SetWithLabels(float64(flowLimitedCount), map[string]string{"cluster": fm.flashNode.clusterID, exporter.FlashNode: fm.flashNode.localAddr}) + fm.MetricRunLimitedCount.SetWithLabels(float64(runLimitedCount), map[string]string{"cluster": fm.flashNode.clusterID, exporter.FlashNode: fm.flashNode.localAddr}) +} + +func (fm *FlashNodeMetrics) setLruUsageRatioMetric() { + usageRatio := fm.flashNode.cacheEngine.GetLruUsageRatio() + fm.MetricLruUsageRatio.SetWithLabels(usageRatio, map[string]string{"cluster": fm.flashNode.clusterID, exporter.FlashNode: fm.flashNode.localAddr}) +} + +func (fm *FlashNodeMetrics) setDiskUsageRatioMetric() { + diskUsageRatioMap := fm.flashNode.cacheEngine.GetDiskUsageRatio() + for dataPath, usageRatio := range diskUsageRatioMap { + fm.MetricDiskUsageRatio.SetWithLabels(usageRatio, map[string]string{"cluster": fm.flashNode.clusterID, exporter.FlashNode: fm.flashNode.localAddr, exporter.Disk: dataPath}) + } } func (fm *FlashNodeMetrics) updateReadBytesMetric(size uint64, d string) { diff --git a/sdk/remotecache/client.go b/sdk/remotecache/client.go index c92b5b17f..1d05d1f16 100755 --- a/sdk/remotecache/client.go +++ b/sdk/remotecache/client.go @@ -651,7 +651,9 @@ func (rc *RemoteCacheClient) updateHostLatency(hosts []string) { defer fg.hostLock.Unlock() for _, h := range fg.Hosts { if h == host { - fg.hostTimeoutCount[host]++ + if fg.hostTimeoutCount[host] <= rc.FlashNodeTimeoutCount { + fg.hostTimeoutCount[host]++ + } if fg.hostTimeoutCount[host] >= rc.FlashNodeTimeoutCount { rc.hostLatency.Delete(host) } diff --git a/sdk/remotecache/flash_group.go b/sdk/remotecache/flash_group.go index 914456b49..8762d7b4d 100644 --- a/sdk/remotecache/flash_group.go +++ b/sdk/remotecache/flash_group.go @@ -151,8 +151,9 @@ func (fg *FlashGroup) moveToUnknownRank(addr string, err error, timeoutCount int unknowns := fg.rankedHost[UnknownZoneRank] unknowns = append(unknowns, addr) fg.rankedHost[UnknownZoneRank] = unknowns + fg.hostTimeoutCount[addr] = 0 - log.LogWarnf("moveToUnknownRank: fgID(%v) host(%v) timeoutCount(%v) by err %v", fg.ID, addr, fg.hostTimeoutCount[addr], err.Error()) + log.LogWarnf("moveToUnknownRank: fgID(%v) host(%v) timeoutCount reset to 0 by err %v", fg.ID, addr, err.Error()) return moved } diff --git a/util/stat/statistic.go b/util/stat/statistic.go index 97f7504a7..a70f96fcf 100644 --- a/util/stat/statistic.go +++ b/util/stat/statistic.go @@ -480,3 +480,20 @@ func GetAvgLatencyMs(typeName string) float32 { } return float32(avgUs) / 1000 } + +func GetCount(typeName string) uint32 { + if gSt == nil { + return 0 + } + + if gSt.useMutex { + gSt.Lock() + defer gSt.Unlock() + } + + typeInfo := gSt.typeInfoMap[typeName] + if typeInfo == nil { + return 0 + } + return typeInfo.allCount +}