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(access): report metric of read and write

Browse Source 
. #22762725

Signed-off-by: slasher <shenjie1@oppo.com>
This commit is contained in:
slasher 2024-11-07 10:38:43 +08:00
parent f7859160de
commit bed91e7765
5 changed files with 37 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
blobstore/access/stream/metric.go
View File

@ -15,9 +15,12 @@
package stream
import (
"os"
"github.com/prometheus/client_golang/prometheus"
"github.com/cubefs/cubefs/blobstore/common/proto"
"github.com/cubefs/cubefs/blobstore/common/rpc/auditlog"
)
var unhealthMetric = prometheus.NewCounterVec(
@ -40,11 +43,27 @@ var downloadMetric = prometheus.NewCounterVec(
[]string{"cluster", "way", "reason"},
)
var readwriteMetric *prometheus.HistogramVec
var SteamReportDownload = reportDownload
func init() {
prometheus.MustRegister(unhealthMetric)
prometheus.MustRegister(downloadMetric)
hostname, _ := os.Hostname()
readwriteMetric = prometheus.NewHistogramVec(
prometheus.HistogramOpts{
Namespace: "blobstore",
Subsystem: "access",
Name: "read_write_duration_ms",
Help: "read write duration ms",
Buckets: auditlog.Buckets,
ConstLabels: map[string]string{"host": hostname},
},
[]string{"cluster", "idc", "api"},
)
prometheus.MustRegister(readwriteMetric)
}
func reportUnhealth(cid proto.ClusterID, action, module, host, reason string) {
@ -54,3 +73,8 @@ func reportUnhealth(cid proto.ClusterID, action, module, host, reason string) {
func reportDownload(cid proto.ClusterID, way, reason string) {
downloadMetric.WithLabelValues(cid.ToString(), way, reason).Inc()
}
// upload_read, upload_write, download_read, download_write
func reportReadwrite(cid, idc, api string, ms int64) {
readwriteMetric.WithLabelValues(cid, idc, api).Observe(float64(ms))
}

1
blobstore/access/stream/stream_get.go
View File

@ -142,6 +142,7 @@ func (h *Handler) Get(ctx context.Context, w io.Writer, location proto.Location,
getTime := new(timeReadWrite)
defer func() {
span.AppendRPCTrackLog([]string{getTime.String()})
getTime.Report(clusterID.ToString(), h.IDC, false)
}()
// try to read data shard only,

1
blobstore/access/stream/stream_put.go
View File

@ -102,6 +102,7 @@ func (h *Handler) Put(ctx context.Context,
// release ec buffer which have not takeover
buffer.Release()
span.AppendRPCTrackLog([]string{putTime.String()})
putTime.Report(clusterID.ToString(), h.IDC, true)
}()
// concurrent buffer in per request

1
blobstore/access/stream/stream_putat.go
View File

@ -62,6 +62,7 @@ func (h *Handler) PutAt(ctx context.Context, rc io.Reader,
defer func() {
buffer.Release()
span.AppendRPCTrackLog([]string{putTime.String()})
putTime.Report(clusterID.ToString(), h.IDC, true)
}()
shards, err := encoder.Split(buffer.ECDataBuf)

10
blobstore/access/stream/stream_time.go
View File

@ -38,6 +38,16 @@ func (t *timeReadWrite) IncW(dur time.Duration) {
atomic.AddInt64(&t.w, int64(dur))
}
func (t *timeReadWrite) Report(cid, idc string, upload bool) {
if upload {
reportReadwrite(cid, idc, "upload_read", atomic.LoadInt64(&t.r)/1e6)
reportReadwrite(cid, idc, "upload_write", atomic.LoadInt64(&t.w)/1e6)
} else {
reportReadwrite(cid, idc, "download_read", atomic.LoadInt64(&t.r)/1e6)
reportReadwrite(cid, idc, "download_write", atomic.LoadInt64(&t.w)/1e6)
}
}
// String within milliseconds
func (t *timeReadWrite) String() string {
a := atomic.LoadInt64(&t.a) / 1e6