TrueCloudLab/frostfs-node
20
1
Fork 28
Code Issues 107 Pull requests 4 Projects Releases 2 Packages 4 Activity Actions

[#426] Put prometheus behind pkg/metrics

Browse source 
Signed-off-by: Alex Vanin <alexey@nspcc.ru>
This commit is contained in:
Alex Vanin 2021-03-16 11:14:56 +03:00 • committed by Leonard Lyubich
parent 3c0e47e6fd
commit b8e10571c6
19 changed files with 561 additions and 311 deletions
Download patch file Download diff file Expand all files Collapse all files

226
pkg/services/object/metrics.go
View file

@ -6,198 +6,73 @@ import (
"github.com/nspcc-dev/neofs-api-go/v2/object"
"github.com/nspcc-dev/neofs-node/pkg/services/util"
"github.com/prometheus/client_golang/prometheus"
)
type (
MetricCollector struct {
next ServiceServer
next ServiceServer
metrics MetricRegister
}
getStreamMetric struct {
util.ServerStream
stream GetObjectStream
stream GetObjectStream
metrics MetricRegister
}
putStreamMetric struct {
stream object.PutObjectStreamer
stream object.PutObjectStreamer
metrics MetricRegister
}
MetricRegister interface {
IncGetReqCounter()
IncPutReqCounter()
IncHeadReqCounter()
IncSearchReqCounter()
IncDeleteReqCounter()
IncRangeReqCounter()
IncRangeHashReqCounter()
AddGetReqDuration(time.Duration)
AddPutReqDuration(time.Duration)
AddHeadReqDuration(time.Duration)
AddSearchReqDuration(time.Duration)
AddDeleteReqDuration(time.Duration)
AddRangeReqDuration(time.Duration)
AddRangeHashReqDuration(time.Duration)
AddPutPayload(int)
AddGetPayload(int)
}
)
const (
namespace = "neofs_node"
subsystem = "object"
)
// Request counter metrics.
var (
getCounter = prometheus.NewCounter(prometheus.CounterOpts{
Namespace: namespace,
Subsystem: subsystem,
Name: "get_req_count",
Help: "Number of get request processed",
ConstLabels: nil,
})
putCounter = prometheus.NewCounter(prometheus.CounterOpts{
Namespace: namespace,
Subsystem: subsystem,
Name: "put_req_count",
Help: "Number of put request processed",
})
headCounter = prometheus.NewCounter(prometheus.CounterOpts{
Namespace: namespace,
Subsystem: subsystem,
Name: "head_req_count",
Help: "Number of head request processed",
})
searchCounter = prometheus.NewCounter(prometheus.CounterOpts{
Namespace: namespace,
Subsystem: subsystem,
Name: "search_req_count",
Help: "Number of search request processed",
})
deleteCounter = prometheus.NewCounter(prometheus.CounterOpts{
Namespace: namespace,
Subsystem: subsystem,
Name: "delete_req_count",
Help: "Number of delete request processed",
})
rangeCounter = prometheus.NewCounter(prometheus.CounterOpts{
Namespace: namespace,
Subsystem: subsystem,
Name: "range_req_count",
Help: "Number of range request processed",
})
rangeHashCounter = prometheus.NewCounter(prometheus.CounterOpts{
Namespace: namespace,
Subsystem: subsystem,
Name: "range_hash_req_count",
Help: "Number of range hash request processed",
})
)
// Request duration metrics.
var (
getDuration = prometheus.NewCounter(prometheus.CounterOpts{
Namespace: namespace,
Subsystem: subsystem,
Name: "get_req_duration",
Help: "Accumulated get request process duration",
})
putDuration = prometheus.NewCounter(prometheus.CounterOpts{
Namespace: namespace,
Subsystem: subsystem,
Name: "put_req_duration",
Help: "Accumulated put request process duration",
})
headDuration = prometheus.NewCounter(prometheus.CounterOpts{
Namespace: namespace,
Subsystem: subsystem,
Name: "head_req_duration",
Help: "Accumulated head request process duration",
})
searchDuration = prometheus.NewCounter(prometheus.CounterOpts{
Namespace: namespace,
Subsystem: subsystem,
Name: "search_req_duration",
Help: "Accumulated search request process duration",
})
deleteDuration = prometheus.NewCounter(prometheus.CounterOpts{
Namespace: namespace,
Subsystem: subsystem,
Name: "delete_req_duration",
Help: "Accumulated delete request process duration",
})
rangeDuration = prometheus.NewCounter(prometheus.CounterOpts{
Namespace: namespace,
Subsystem: subsystem,
Name: "range_req_duration",
Help: "Accumulated range request process duration",
})
rangeHashDuration = prometheus.NewCounter(prometheus.CounterOpts{
Namespace: namespace,
Subsystem: subsystem,
Name: "range_hash_req_duration",
Help: "Accumulated range hash request process duration",
})
)
// Object payload metrics.
var (
putPayload = prometheus.NewCounter(prometheus.CounterOpts{
Namespace: namespace,
Subsystem: subsystem,
Name: "put_payload",
Help: "Accumulated payload size at object put method",
})
getPayload = prometheus.NewCounter(prometheus.CounterOpts{
Namespace: namespace,
Subsystem: subsystem,
Name: "get_payload",
Help: "Accumulated payload size at object get method",
})
)
func registerMetrics() {
prometheus.MustRegister(getCounter) // todo: replace with for loop over map
prometheus.MustRegister(putCounter)
prometheus.MustRegister(headCounter)
prometheus.MustRegister(searchCounter)
prometheus.MustRegister(deleteCounter)
prometheus.MustRegister(rangeCounter)
prometheus.MustRegister(rangeHashCounter)
prometheus.MustRegister(getDuration)
prometheus.MustRegister(putDuration)
prometheus.MustRegister(headDuration)
prometheus.MustRegister(searchDuration)
prometheus.MustRegister(deleteDuration)
prometheus.MustRegister(rangeDuration)
prometheus.MustRegister(rangeHashDuration)
prometheus.MustRegister(putPayload)
prometheus.MustRegister(getPayload)
}
func NewMetricCollector(next ServiceServer) *MetricCollector {
registerMetrics()
func NewMetricCollector(next ServiceServer, register MetricRegister) *MetricCollector {
return &MetricCollector{
next: next,
next: next,
metrics: register,
}
}
func (m MetricCollector) Get(req *object.GetRequest, stream GetObjectStream) error {
t := time.Now()
defer func() {
getCounter.Inc()
getDuration.Add(float64(time.Since(t)))
m.metrics.IncGetReqCounter()
m.metrics.AddGetReqDuration(time.Since(t))
}()
return m.next.Get(req, &getStreamMetric{
ServerStream: stream,
stream: stream,
metrics: m.metrics,
})
}
func (m MetricCollector) Put(ctx context.Context) (object.PutObjectStreamer, error) {
t := time.Now()
defer func() {
putCounter.Inc()
putDuration.Add(float64(time.Since(t)))
m.metrics.IncPutReqCounter()
m.metrics.AddPutReqDuration(time.Since(t))
}()
stream, err := m.next.Put(ctx)
@ -205,14 +80,17 @@ func (m MetricCollector) Put(ctx context.Context) (object.PutObjectStreamer, err
return nil, err
}
return &putStreamMetric{stream: stream}, nil
return &putStreamMetric{
stream: stream,
metrics: m.metrics,
}, nil
}
func (m MetricCollector) Head(ctx context.Context, request *object.HeadRequest) (*object.HeadResponse, error) {
t := time.Now()
defer func() {
headCounter.Inc()
headDuration.Add(float64(time.Since(t)))
m.metrics.IncHeadReqCounter()
m.metrics.AddHeadReqDuration(time.Since(t))
}()
return m.next.Head(ctx, request)
@ -221,8 +99,8 @@ func (m MetricCollector) Head(ctx context.Context, request *object.HeadRequest)
func (m MetricCollector) Search(req *object.SearchRequest, stream SearchStream) error {
t := time.Now()
defer func() {
searchCounter.Inc()
searchDuration.Add(float64(time.Since(t)))
m.metrics.IncSearchReqCounter()
m.metrics.AddSearchReqDuration(time.Since(t))
}()
return m.next.Search(req, stream)
@ -231,8 +109,8 @@ func (m MetricCollector) Search(req *object.SearchRequest, stream SearchStream)
func (m MetricCollector) Delete(ctx context.Context, request *object.DeleteRequest) (*object.DeleteResponse, error) {
t := time.Now()
defer func() {
deleteCounter.Inc()
deleteDuration.Add(float64(time.Since(t)))
m.metrics.IncDeleteReqCounter()
m.metrics.AddDeleteReqDuration(time.Since(t))
}()
return m.next.Delete(ctx, request)
@ -241,8 +119,8 @@ func (m MetricCollector) Delete(ctx context.Context, request *object.DeleteReque
func (m MetricCollector) GetRange(req *object.GetRangeRequest, stream GetObjectRangeStream) error {
t := time.Now()
defer func() {
rangeCounter.Inc()
rangeDuration.Add(float64(time.Since(t)))
m.metrics.IncRangeReqCounter()
m.metrics.AddRangeReqDuration(time.Since(t))
}()
return m.next.GetRange(req, stream)
@ -251,8 +129,8 @@ func (m MetricCollector) GetRange(req *object.GetRangeRequest, stream GetObjectR
func (m MetricCollector) GetRangeHash(ctx context.Context, request *object.GetRangeHashRequest) (*object.GetRangeHashResponse, error) {
t := time.Now()
defer func() {
rangeHashCounter.Inc()
rangeHashDuration.Add(float64(time.Since(t)))
m.metrics.IncRangeHashReqCounter()
m.metrics.AddRangeHashReqDuration(time.Since(t))
}()
return m.next.GetRangeHash(ctx, request)
@ -261,8 +139,7 @@ func (m MetricCollector) GetRangeHash(ctx context.Context, request *object.GetRa
func (s getStreamMetric) Send(resp *object.GetResponse) error {
chunk, ok := resp.GetBody().GetObjectPart().(*object.GetObjectPartChunk)
if ok {
ln := len(chunk.GetChunk())
getPayload.Add(float64(ln))
s.metrics.AddGetPayload(len(chunk.GetChunk()))
}
return s.stream.Send(resp)
@ -271,8 +148,7 @@ func (s getStreamMetric) Send(resp *object.GetResponse) error {
func (s putStreamMetric) Send(req *object.PutRequest) error {
chunk, ok := req.GetBody().GetObjectPart().(*object.PutObjectPartChunk)
if ok {
ln := len(chunk.GetChunk())
putPayload.Add(float64(ln))
s.metrics.AddPutPayload(len(chunk.GetChunk()))
}
return s.stream.Send(req)